From 505cb43176b9d16e8d2fc9b2ed69d13a31b902de Mon Sep 17 00:00:00 2001 From: hkobew Date: Tue, 10 Mar 2026 11:30:38 -0400 Subject: [PATCH 1/3] feat: add streamDeliveryResources schema, CLI flags, and validation for memory record streaming --- docs/memory.md | 60 +++++ .../commands/add/__tests__/validate.test.ts | 65 ++++++ src/cli/commands/add/types.ts | 4 + src/cli/commands/add/validate.ts | 31 +++ src/cli/primitives/MemoryPrimitive.tsx | 208 +++++++++++++----- .../__tests__/agentcore-project.test.ts | 96 ++++++++ src/schema/schemas/agentcore-project.ts | 29 +++ 7 files changed, 436 insertions(+), 57 deletions(-) diff --git a/docs/memory.md b/docs/memory.md index 0a1195b8b..7841a6772 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -222,6 +222,66 @@ Memory events expire after a configurable duration (7-365 days, default 30): } ``` +## Memory Record Streaming + +Memory record streaming delivers real-time events when memory records are created, updated, or deleted. Events are +pushed to a delivery target in your account, enabling event-driven architectures without polling. + +### Enabling Streaming + +Via CLI flags: + +```bash +agentcore add memory \ + --name MyMemory \ + --strategies SEMANTIC \ + --data-stream-arn arn:aws:kinesis:us-west-2:123456789012:stream/my-stream \ + --stream-content-level FULL_CONTENT +``` + +For advanced configurations (e.g. multiple delivery targets), pass the full JSON: + +```bash +agentcore add memory \ + --name MyMemory \ + --strategies SEMANTIC \ + --stream-delivery-resources '{"resources":[{"kinesis":{"dataStreamArn":"arn:aws:kinesis:us-west-2:123456789012:stream/my-stream","contentConfigurations":[{"type":"MEMORY_RECORDS","level":"FULL_CONTENT"}]}}]}' +``` + +### Configuration + +```json +{ + "type": "AgentCoreMemory", + "name": "MyMemory", + "eventExpiryDuration": 30, + "strategies": [{ "type": "SEMANTIC" }], + "streamDeliveryResources": { + "resources": [ + { + "kinesis": { + "dataStreamArn": "arn:aws:kinesis:us-west-2:123456789012:stream/my-stream", + "contentConfigurations": [{ "type": "MEMORY_RECORDS", "level": "FULL_CONTENT" }] + } + } + ] + } +} +``` + +### Content Level + +| Level | Description | +| --------------- | ---------------------------------------------------------- | +| `FULL_CONTENT` | Events include memory record text and all metadata | +| `METADATA_ONLY` | Events include only metadata (IDs, timestamps, namespaces) | + +The CDK construct automatically grants the memory execution role permission to publish to the configured delivery +target. + +For more details, see the +[Memory Record Streaming documentation](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/memory-record-streaming.html). + ## Using Memory in Code The memory ID is available via environment variable: diff --git a/src/cli/commands/add/__tests__/validate.test.ts b/src/cli/commands/add/__tests__/validate.test.ts index 6c73adb3e..4963a73b0 100644 --- a/src/cli/commands/add/__tests__/validate.test.ts +++ b/src/cli/commands/add/__tests__/validate.test.ts @@ -994,6 +994,71 @@ describe('validate', () => { valid: true, }); }); + + // Streaming validation + it('accepts valid streaming options', () => { + expect( + validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + contentLevel: 'FULL_CONTENT', + }) + ).toEqual({ valid: true }); + }); + + it('accepts dataStreamArn without contentLevel (defaults to FULL_CONTENT)', () => { + expect( + validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + }) + ).toEqual({ valid: true }); + }); + + it('rejects contentLevel without dataStreamArn', () => { + const result = validateAddMemoryOptions({ ...validMemoryOptions, contentLevel: 'FULL_CONTENT' }); + expect(result.valid).toBe(false); + expect(result.error).toContain('--data-stream-arn is required'); + }); + + it('rejects invalid contentLevel', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + contentLevel: 'INVALID', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('Invalid content level'); + }); + + it('rejects invalid deliveryType', () => { + const result = validateAddMemoryOptions({ ...validMemoryOptions, deliveryType: 'sqs' }); + expect(result.valid).toBe(false); + expect(result.error).toContain('Invalid delivery type'); + }); + + it('accepts valid deliveryType', () => { + expect(validateAddMemoryOptions({ ...validMemoryOptions, deliveryType: 'kinesis' })).toEqual({ valid: true }); + }); + + it('rejects dataStreamArn not starting with arn:', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'not-an-arn', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('valid ARN'); + }); + + it('rejects combining streamDeliveryResources with flat flags', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + streamDeliveryResources: '{"resources":[]}', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('cannot be combined'); + }); }); describe('validateAddCredentialOptions', () => { diff --git a/src/cli/commands/add/types.ts b/src/cli/commands/add/types.ts index 51beeb111..4a1922bba 100644 --- a/src/cli/commands/add/types.ts +++ b/src/cli/commands/add/types.ts @@ -109,6 +109,10 @@ export interface AddMemoryOptions { name?: string; strategies?: string; expiry?: number; + deliveryType?: string; + dataStreamArn?: string; + contentLevel?: string; + streamDeliveryResources?: string; json?: boolean; } diff --git a/src/cli/commands/add/validate.ts b/src/cli/commands/add/validate.ts index 44a010168..7426518b6 100644 --- a/src/cli/commands/add/validate.ts +++ b/src/cli/commands/add/validate.ts @@ -35,6 +35,8 @@ export interface ValidationResult { // Constants const MEMORY_OPTIONS = ['none', 'shortTerm', 'longAndShortTerm'] as const; const VALID_STRATEGIES = ['SEMANTIC', 'SUMMARIZATION', 'USER_PREFERENCE', 'EPISODIC']; +const VALID_STREAM_CONTENT_LEVELS = ['FULL_CONTENT', 'METADATA_ONLY']; +const VALID_DELIVERY_TYPES = ['kinesis']; /** * Validate that a credential name exists in the project spec. @@ -677,6 +679,35 @@ export function validateAddMemoryOptions(options: AddMemoryOptions): ValidationR } } + if (options.streamDeliveryResources && (options.dataStreamArn || options.contentLevel)) { + return { + valid: false, + error: '--stream-delivery-resources cannot be combined with --data-stream-arn or --stream-content-level', + }; + } + + if (options.contentLevel && !options.dataStreamArn) { + return { valid: false, error: '--data-stream-arn is required when --stream-content-level is set' }; + } + + if (options.dataStreamArn && !options.dataStreamArn.startsWith('arn:')) { + return { valid: false, error: '--data-stream-arn must be a valid ARN (starts with arn:)' }; + } + + if (options.deliveryType && !VALID_DELIVERY_TYPES.includes(options.deliveryType)) { + return { + valid: false, + error: `Invalid delivery type. Must be one of: ${VALID_DELIVERY_TYPES.join(', ')}`, + }; + } + + if (options.contentLevel && !VALID_STREAM_CONTENT_LEVELS.includes(options.contentLevel)) { + return { + valid: false, + error: `Invalid content level. Must be one of: ${VALID_STREAM_CONTENT_LEVELS.join(', ')}`, + }; + } + return { valid: true }; } diff --git a/src/cli/primitives/MemoryPrimitive.tsx b/src/cli/primitives/MemoryPrimitive.tsx index 8220e20d6..3cee74a2b 100644 --- a/src/cli/primitives/MemoryPrimitive.tsx +++ b/src/cli/primitives/MemoryPrimitive.tsx @@ -1,6 +1,18 @@ import { findConfigRoot } from '../../lib'; -import type { Memory, MemoryStrategy, MemoryStrategyType } from '../../schema'; -import { DEFAULT_EPISODIC_REFLECTION_NAMESPACES, DEFAULT_STRATEGY_NAMESPACES, MemorySchema } from '../../schema'; +import type { + Memory, + MemoryStrategy, + MemoryStrategyType, + StreamContentLevel, + StreamDeliveryResources, +} from '../../schema'; +import { + DEFAULT_EPISODIC_REFLECTION_NAMESPACES, + DEFAULT_STRATEGY_NAMESPACES, + MemorySchema, + StreamContentLevelSchema, + StreamDeliveryResourcesSchema, +} from '../../schema'; import { validateAddMemoryOptions } from '../commands/add/validate'; import { getErrorMessage } from '../errors'; import type { RemovalPreview, RemovalResult, SchemaChange } from '../operations/remove/types'; @@ -16,6 +28,12 @@ export interface AddMemoryOptions { name: string; strategies?: string; expiry?: number; + deliveryType?: string; + // Flat flags for the simple single-stream case + dataStreamArn?: string; + contentLevel?: string; + // Raw JSON for advanced/multi-target configurations. Takes precedence over flat flags. + streamDeliveryResources?: string; } /** @@ -42,10 +60,21 @@ export class MemoryPrimitive extends BasePrimitive ({ type: type as MemoryStrategyType })) : []; + const streamDeliveryResources = options.streamDeliveryResources + ? this.parseStreamDeliveryResources(options.streamDeliveryResources) + : options.dataStreamArn + ? this.buildStreamDeliveryResources({ + deliveryType: options.deliveryType ?? 'kinesis', + dataStreamArn: options.dataStreamArn, + contentLevel: StreamContentLevelSchema.parse(options.contentLevel ?? 'FULL_CONTENT'), + }) + : undefined; + const memory = await this.createMemory({ name: options.name, eventExpiryDuration: options.expiry ?? DEFAULT_EVENT_EXPIRY, strategies, + streamDeliveryResources, }); return { success: true, memoryName: memory.name }; @@ -129,73 +158,102 @@ export class MemoryPrimitive extends BasePrimitive', 'Event expiry duration in days (default: 30) [non-interactive]') + .option('--delivery-type ', 'Delivery target type (default: kinesis) [non-interactive]') + .option('--data-stream-arn ', 'Kinesis data stream ARN for memory record streaming [non-interactive]') + .option( + '--stream-content-level ', + 'Stream content level: FULL_CONTENT or METADATA_ONLY (default: FULL_CONTENT) [non-interactive]' + ) + .option( + '--stream-delivery-resources ', + 'Stream delivery config as JSON string (advanced, overrides flat flags) [non-interactive]' + ) .option('--json', 'Output as JSON [non-interactive]') - .action(async (cliOptions: { name?: string; strategies?: string; expiry?: string; json?: boolean }) => { - try { - if (!findConfigRoot()) { - console.error('No agentcore project found. Run `agentcore create` first.'); - process.exit(1); - } + .action( + async (cliOptions: { + name?: string; + strategies?: string; + expiry?: string; + deliveryType?: string; + dataStreamArn?: string; + streamContentLevel?: string; + streamDeliveryResources?: string; + json?: boolean; + }) => { + try { + if (!findConfigRoot()) { + console.error('No agentcore project found. Run `agentcore create` first.'); + process.exit(1); + } + + if (cliOptions.name || cliOptions.json) { + // CLI mode + const expiry = cliOptions.expiry ? parseInt(cliOptions.expiry, 10) : undefined; + const validation = validateAddMemoryOptions({ + name: cliOptions.name, + strategies: cliOptions.strategies, + expiry, + deliveryType: cliOptions.deliveryType, + dataStreamArn: cliOptions.dataStreamArn, + contentLevel: cliOptions.streamContentLevel, + streamDeliveryResources: cliOptions.streamDeliveryResources, + }); - if (cliOptions.name || cliOptions.json) { - // CLI mode - const expiry = cliOptions.expiry ? parseInt(cliOptions.expiry, 10) : undefined; - const validation = validateAddMemoryOptions({ - name: cliOptions.name, - strategies: cliOptions.strategies, - expiry, - }); + if (!validation.valid) { + if (cliOptions.json) { + console.log(JSON.stringify({ success: false, error: validation.error })); + } else { + console.error(validation.error); + } + process.exit(1); + } + + const result = await this.add({ + name: cliOptions.name!, + strategies: cliOptions.strategies, + expiry, + deliveryType: cliOptions.deliveryType, + dataStreamArn: cliOptions.dataStreamArn, + contentLevel: cliOptions.streamContentLevel, + streamDeliveryResources: cliOptions.streamDeliveryResources, + }); - if (!validation.valid) { if (cliOptions.json) { - console.log(JSON.stringify({ success: false, error: validation.error })); + console.log(JSON.stringify(result)); + } else if (result.success) { + console.log(`Added memory '${result.memoryName}'`); } else { - console.error(validation.error); + console.error(result.error); } - process.exit(1); + process.exit(result.success ? 0 : 1); + } else { + // TUI fallback — dynamic imports to avoid pulling ink (async) into registry + const [{ render }, { default: React }, { AddFlow }] = await Promise.all([ + import('ink'), + import('react'), + import('../tui/screens/add/AddFlow'), + ]); + const { clear, unmount } = render( + React.createElement(AddFlow, { + isInteractive: false, + onExit: () => { + clear(); + unmount(); + process.exit(0); + }, + }) + ); } - - const result = await this.add({ - name: cliOptions.name!, - strategies: cliOptions.strategies, - expiry, - }); - + } catch (error) { if (cliOptions.json) { - console.log(JSON.stringify(result)); - } else if (result.success) { - console.log(`Added memory '${result.memoryName}'`); + console.log(JSON.stringify({ success: false, error: getErrorMessage(error) })); } else { - console.error(result.error); + console.error(getErrorMessage(error)); } - process.exit(result.success ? 0 : 1); - } else { - // TUI fallback — dynamic imports to avoid pulling ink (async) into registry - const [{ render }, { default: React }, { AddFlow }] = await Promise.all([ - import('ink'), - import('react'), - import('../tui/screens/add/AddFlow'), - ]); - const { clear, unmount } = render( - React.createElement(AddFlow, { - isInteractive: false, - onExit: () => { - clear(); - unmount(); - process.exit(0); - }, - }) - ); - } - } catch (error) { - if (cliOptions.json) { - console.log(JSON.stringify({ success: false, error: getErrorMessage(error) })); - } else { - console.error(getErrorMessage(error)); + process.exit(1); } - process.exit(1); } - }); + ); this.registerRemoveSubcommand(removeCmd); } @@ -211,6 +269,7 @@ export class MemoryPrimitive extends BasePrimitive { const project = await this.readProjectSpec(); @@ -231,6 +290,7 @@ export class MemoryPrimitive extends BasePrimitive { }); expect(result.success).toBe(true); }); + + it('accepts memory with streamDeliveryResources', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'StreamMemory', + eventExpiryDuration: 30, + strategies: [{ type: 'SEMANTIC' }], + streamDeliveryResources: { + resources: [ + { + kinesis: { + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + contentConfigurations: [{ type: 'MEMORY_RECORDS', level: 'FULL_CONTENT' }], + }, + }, + ], + }, + }); + expect(result.success).toBe(true); + }); + + it('accepts memory without streamDeliveryResources', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'NoStream', + eventExpiryDuration: 30, + strategies: [], + }); + expect(result.success).toBe(true); + expect(result.data?.streamDeliveryResources).toBeUndefined(); + }); + + it('rejects streamDeliveryResources with empty resources array', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'Test', + eventExpiryDuration: 30, + strategies: [], + streamDeliveryResources: { resources: [] }, + }); + expect(result.success).toBe(false); + }); + + it('rejects streamDeliveryResources with empty contentConfigurations', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'Test', + eventExpiryDuration: 30, + strategies: [], + streamDeliveryResources: { + resources: [ + { + kinesis: { dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', contentConfigurations: [] }, + }, + ], + }, + }); + expect(result.success).toBe(false); + }); + + it('rejects streamDeliveryResources with empty dataStreamArn', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'Test', + eventExpiryDuration: 30, + strategies: [], + streamDeliveryResources: { + resources: [ + { + kinesis: { dataStreamArn: '', contentConfigurations: [{ type: 'MEMORY_RECORDS', level: 'FULL_CONTENT' }] }, + }, + ], + }, + }); + expect(result.success).toBe(false); + }); + + it('rejects invalid content level in streamDeliveryResources', () => { + const result = MemorySchema.safeParse({ + type: 'AgentCoreMemory', + name: 'Test', + eventExpiryDuration: 30, + strategies: [], + streamDeliveryResources: { + resources: [ + { + kinesis: { + dataStreamArn: 'arn:test', + contentConfigurations: [{ type: 'MEMORY_RECORDS', level: 'INVALID' }], + }, + }, + ], + }, + }); + expect(result.success).toBe(false); + }); }); describe('CredentialNameSchema', () => { diff --git a/src/schema/schemas/agentcore-project.ts b/src/schema/schemas/agentcore-project.ts index 89fb1a82f..8d0f35f1c 100644 --- a/src/schema/schemas/agentcore-project.ts +++ b/src/schema/schemas/agentcore-project.ts @@ -96,6 +96,34 @@ export const MemoryNameSchema = z 'Must begin with a letter and contain only alphanumeric characters and underscores (max 48 chars)' ); +export const StreamContentLevelSchema = z.enum(['FULL_CONTENT', 'METADATA_ONLY']); +export type StreamContentLevel = z.infer; + +// TODO: kinesis is currently the only supported delivery type. When additional types +// (e.g. S3, EventBridge) are added, this should become a discriminated union. +// Non-kinesis resources will produce a Zod error about the missing kinesis field. +export const StreamDeliveryResourcesSchema = z.object({ + resources: z + .array( + z.object({ + kinesis: z.object({ + dataStreamArn: z.string().min(1), + contentConfigurations: z + .array( + z.object({ + type: z.literal('MEMORY_RECORDS'), + level: StreamContentLevelSchema, + }) + ) + .min(1), + }), + }) + ) + .min(1), +}); + +export type StreamDeliveryResources = z.infer; + export const MemorySchema = z.object({ name: MemoryNameSchema, eventExpiryDuration: z.number().int().min(7).max(365), @@ -113,6 +141,7 @@ export const MemorySchema = z.object({ tags: TagsSchema.optional(), encryptionKeyArn: z.string().optional(), executionRoleArn: z.string().optional(), + streamDeliveryResources: StreamDeliveryResourcesSchema.optional(), }); export type Memory = z.infer; From 8202106d8dc1015edf9e11ca98bda93ec87331fc Mon Sep 17 00:00:00 2001 From: Harrison Weinstock Date: Mon, 6 Apr 2026 20:24:02 +0000 Subject: [PATCH 2/3] fix: address review feedback on streaming support - Replace unsafe `as MemoryStrategyType` casts with Zod parse - Include deliveryType in streamDeliveryResources conflict check - Reject deliveryType without dataStreamArn - Validate streamDeliveryResources JSON eagerly in validator - Include Zod error details in parseStreamDeliveryResources - Extract DEFAULT_DELIVERY_TYPE constant - Tighten createMemory strategies param type --- .../commands/add/__tests__/validate.test.ts | 52 ++++++++++++++++++- src/cli/commands/add/validate.ts | 32 ++++++++++-- src/cli/primitives/MemoryPrimitive.tsx | 51 +++++++++--------- 3 files changed, 103 insertions(+), 32 deletions(-) diff --git a/src/cli/commands/add/__tests__/validate.test.ts b/src/cli/commands/add/__tests__/validate.test.ts index 4963a73b0..2b23baa9a 100644 --- a/src/cli/commands/add/__tests__/validate.test.ts +++ b/src/cli/commands/add/__tests__/validate.test.ts @@ -1032,13 +1032,23 @@ describe('validate', () => { }); it('rejects invalid deliveryType', () => { - const result = validateAddMemoryOptions({ ...validMemoryOptions, deliveryType: 'sqs' }); + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + deliveryType: 'sqs', + }); expect(result.valid).toBe(false); expect(result.error).toContain('Invalid delivery type'); }); it('accepts valid deliveryType', () => { - expect(validateAddMemoryOptions({ ...validMemoryOptions, deliveryType: 'kinesis' })).toEqual({ valid: true }); + expect( + validateAddMemoryOptions({ + ...validMemoryOptions, + dataStreamArn: 'arn:aws:kinesis:us-west-2:123456789012:stream/test', + deliveryType: 'kinesis', + }) + ).toEqual({ valid: true }); }); it('rejects dataStreamArn not starting with arn:', () => { @@ -1059,6 +1069,44 @@ describe('validate', () => { expect(result.valid).toBe(false); expect(result.error).toContain('cannot be combined'); }); + + it('rejects combining streamDeliveryResources with deliveryType', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + deliveryType: 'kinesis', + streamDeliveryResources: + '{"resources":[{"kinesis":{"dataStreamArn":"arn:aws:kinesis:us-west-2:123456789012:stream/test","contentConfigurations":[{"type":"MEMORY_RECORDS","level":"FULL_CONTENT"}]}}]}', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('cannot be combined'); + }); + + it('rejects deliveryType without dataStreamArn', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + deliveryType: 'kinesis', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('--data-stream-arn is required'); + }); + + it('rejects invalid streamDeliveryResources JSON', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + streamDeliveryResources: 'not json', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('Invalid JSON'); + }); + + it('rejects streamDeliveryResources that fails schema validation', () => { + const result = validateAddMemoryOptions({ + ...validMemoryOptions, + streamDeliveryResources: '{"resources":[]}', + }); + expect(result.valid).toBe(false); + expect(result.error).toContain('does not match the expected schema'); + }); }); describe('validateAddCredentialOptions', () => { diff --git a/src/cli/commands/add/validate.ts b/src/cli/commands/add/validate.ts index 7426518b6..e23057398 100644 --- a/src/cli/commands/add/validate.ts +++ b/src/cli/commands/add/validate.ts @@ -8,6 +8,7 @@ import { ProtocolModeSchema, RuntimeAuthorizerTypeSchema, SDKFrameworkSchema, + StreamDeliveryResourcesSchema, TARGET_TYPE_AUTH_CONFIG, TargetLanguageSchema, getSupportedFrameworksForProtocol, @@ -36,7 +37,8 @@ export interface ValidationResult { const MEMORY_OPTIONS = ['none', 'shortTerm', 'longAndShortTerm'] as const; const VALID_STRATEGIES = ['SEMANTIC', 'SUMMARIZATION', 'USER_PREFERENCE', 'EPISODIC']; const VALID_STREAM_CONTENT_LEVELS = ['FULL_CONTENT', 'METADATA_ONLY']; -const VALID_DELIVERY_TYPES = ['kinesis']; +const VALID_DELIVERY_TYPES = ['kinesis'] as const; +export const DEFAULT_DELIVERY_TYPE = 'kinesis'; /** * Validate that a credential name exists in the project spec. @@ -679,10 +681,11 @@ export function validateAddMemoryOptions(options: AddMemoryOptions): ValidationR } } - if (options.streamDeliveryResources && (options.dataStreamArn || options.contentLevel)) { + if (options.streamDeliveryResources && (options.dataStreamArn || options.contentLevel || options.deliveryType)) { return { valid: false, - error: '--stream-delivery-resources cannot be combined with --data-stream-arn or --stream-content-level', + error: + '--stream-delivery-resources cannot be combined with --data-stream-arn, --stream-content-level, or --delivery-type', }; } @@ -690,11 +693,18 @@ export function validateAddMemoryOptions(options: AddMemoryOptions): ValidationR return { valid: false, error: '--data-stream-arn is required when --stream-content-level is set' }; } + if (options.deliveryType && !options.dataStreamArn) { + return { valid: false, error: '--data-stream-arn is required when --delivery-type is set' }; + } + if (options.dataStreamArn && !options.dataStreamArn.startsWith('arn:')) { return { valid: false, error: '--data-stream-arn must be a valid ARN (starts with arn:)' }; } - if (options.deliveryType && !VALID_DELIVERY_TYPES.includes(options.deliveryType)) { + if ( + options.deliveryType && + !VALID_DELIVERY_TYPES.includes(options.deliveryType as (typeof VALID_DELIVERY_TYPES)[number]) + ) { return { valid: false, error: `Invalid delivery type. Must be one of: ${VALID_DELIVERY_TYPES.join(', ')}`, @@ -708,6 +718,20 @@ export function validateAddMemoryOptions(options: AddMemoryOptions): ValidationR }; } + if (options.streamDeliveryResources) { + try { + StreamDeliveryResourcesSchema.parse(JSON.parse(options.streamDeliveryResources)); + } catch (e) { + return { + valid: false, + error: + e instanceof SyntaxError + ? 'Invalid JSON in --stream-delivery-resources' + : 'Invalid --stream-delivery-resources: does not match the expected schema', + }; + } + } + return { valid: true }; } diff --git a/src/cli/primitives/MemoryPrimitive.tsx b/src/cli/primitives/MemoryPrimitive.tsx index 3cee74a2b..c808bbac3 100644 --- a/src/cli/primitives/MemoryPrimitive.tsx +++ b/src/cli/primitives/MemoryPrimitive.tsx @@ -10,16 +10,18 @@ import { DEFAULT_EPISODIC_REFLECTION_NAMESPACES, DEFAULT_STRATEGY_NAMESPACES, MemorySchema, + MemoryStrategyTypeSchema, StreamContentLevelSchema, StreamDeliveryResourcesSchema, } from '../../schema'; -import { validateAddMemoryOptions } from '../commands/add/validate'; +import { DEFAULT_DELIVERY_TYPE, validateAddMemoryOptions } from '../commands/add/validate'; import { getErrorMessage } from '../errors'; import type { RemovalPreview, RemovalResult, SchemaChange } from '../operations/remove/types'; import { DEFAULT_EVENT_EXPIRY } from '../tui/screens/memory/types'; import { BasePrimitive } from './BasePrimitive'; import type { AddResult, AddScreenComponent, RemovableResource } from './types'; import type { Command } from '@commander-js/extra-typings'; +import { z } from 'zod'; /** * Options for adding a memory resource. @@ -57,14 +59,14 @@ export class MemoryPrimitive extends BasePrimitive s.trim()) .filter(Boolean) - .map(type => ({ type: type as MemoryStrategyType })) + .map(type => ({ type: MemoryStrategyTypeSchema.parse(type) })) : []; const streamDeliveryResources = options.streamDeliveryResources ? this.parseStreamDeliveryResources(options.streamDeliveryResources) : options.dataStreamArn ? this.buildStreamDeliveryResources({ - deliveryType: options.deliveryType ?? 'kinesis', + deliveryType: options.deliveryType ?? DEFAULT_DELIVERY_TYPE, dataStreamArn: options.dataStreamArn, contentLevel: StreamContentLevelSchema.parse(options.contentLevel ?? 'FULL_CONTENT'), }) @@ -268,7 +270,7 @@ export class MemoryPrimitive extends BasePrimitive { const project = await this.readProjectSpec(); @@ -277,12 +279,11 @@ export class MemoryPrimitive extends BasePrimitive { - const strategyType = s.type as MemoryStrategyType; - const defaultNamespaces = DEFAULT_STRATEGY_NAMESPACES[strategyType]; + const defaultNamespaces = DEFAULT_STRATEGY_NAMESPACES[s.type]; return { - type: strategyType, + type: s.type, ...(defaultNamespaces && { namespaces: defaultNamespaces }), - ...(strategyType === 'EPISODIC' && { reflectionNamespaces: DEFAULT_EPISODIC_REFLECTION_NAMESPACES }), + ...(s.type === 'EPISODIC' && { reflectionNamespaces: DEFAULT_EPISODIC_REFLECTION_NAMESPACES }), }; }); @@ -304,32 +305,30 @@ export class MemoryPrimitive extends BasePrimitive i.message).join(', ')}` : ''; + throw new Error(`Stream delivery config does not match the expected schema${detail}`); } } } From 98d1212551d94aea0c9bf423f9bbc697bc6769cc Mon Sep 17 00:00:00 2001 From: Harrison Weinstock Date: Mon, 6 Apr 2026 21:28:26 +0000 Subject: [PATCH 3/3] fix: update create-memory test for strict strategy validation CUSTOM is not a valid MemoryStrategyType. The previous test relied on an unsafe `as` cast to pass an invalid strategy through. Now that we use Zod parse, invalid strategies are correctly rejected. --- .../operations/memory/__tests__/create-memory.test.ts | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/cli/operations/memory/__tests__/create-memory.test.ts b/src/cli/operations/memory/__tests__/create-memory.test.ts index 097f88689..a0b8077c4 100644 --- a/src/cli/operations/memory/__tests__/create-memory.test.ts +++ b/src/cli/operations/memory/__tests__/create-memory.test.ts @@ -76,20 +76,18 @@ describe('add', () => { expect(addedMemory.strategies[0]!.namespaces).toEqual(['/users/{actorId}/facts']); }); - it('creates memory with strategy without default namespaces', async () => { + it('rejects invalid strategy type', async () => { const project = makeProject([]); mockReadProjectSpec.mockResolvedValue(project); - mockWriteProjectSpec.mockResolvedValue(undefined); - await primitive.add({ + const result = await primitive.add({ name: 'NewMem', strategies: 'CUSTOM', expiry: 30, }); - const writtenSpec = mockWriteProjectSpec.mock.calls[0]![0]; - const addedMemory = writtenSpec.memories.find((m: { name: string }) => m.name === 'NewMem'); - expect(addedMemory.strategies[0]!.namespaces).toBeUndefined(); + expect(result).toEqual(expect.objectContaining({ success: false, error: expect.any(String) })); + expect(mockWriteProjectSpec).not.toHaveBeenCalled(); }); it('returns error on duplicate memory name', async () => {