fix(bb-agent): reject empty channelId in stream()

.changeset/fix-bb-agent-empty-channel-id.md

@@ -0,0 +1,7 @@
+---
+"@aws-blocks/bb-agent": patch
+---
+
+fix(bb-agent): treat empty channelId as unset in stream()
+
+An empty `channelId` now falls back to `conversationId` or a random UUID, preventing all streams from sharing the same channel. Empty strings are treated as unset rather than used literally.

packages/bb-agent/README.md

@@ -47,7 +47,7 @@ const agent = new Agent(scope, id, config)
 | `getPendingInterrupts(conversationId)` | `Promise<Array<...>>` | Get unanswered interrupts (for reload support). |
 | `getChannel(channelId)` | `Promise<RealtimeChannel>` | Get a Realtime channel for subscribing to chunks. |
 
-`stream()` submits the message to AsyncJob and returns immediately — no API Gateway timeout risk. The agent runs asynchronously and publishes chunks to Realtime.
+`stream()` submits the message to AsyncJob and returns immediately — no API Gateway timeout risk. The agent runs asynchronously and publishes chunks to Realtime. The channel ID is resolved as `options.channelId || options.conversationId || crypto.randomUUID()` — empty strings are treated as unset and fall through to the next value.
 
 **Important: Subscribe before sending.** The agent starts emitting chunks immediately after `stream()` is called. If you subscribe to the channel after calling `stream()`, early chunks may be dropped. Always subscribe first, await `established`, then send:
 

packages/bb-agent/src/agent.ts

@@ -9,14 +9,14 @@
 import { FileBucket } from '@aws-blocks/bb-file-bucket';
 import { Logger } from '@aws-blocks/bb-logger';
 import type { ChildLogger } from '@aws-blocks/bb-logger';
 import { Agent as StrandsAgent, tool, SessionManager, ModelStreamUpdateEvent, AfterToolCallEvent, BeforeToolCallEvent, AgentResultEvent, InterruptEvent } from '@strands-agents/sdk';
 import type { AgentResult } from '@strands-agents/sdk';
 import { InterruptResponseContent } from '@strands-agents/sdk';
 import { z } from 'zod';
 import { createStrandsModel, checkModelHealth } from './model-factory.js';
 import type { SnapshotStorage } from '@strands-agents/sdk';
 import { messageSchema, conversationSchema, agentStreamChunkSchema } from './schemas.js';
 import type { AgentConfig, AgentStreamChunk, AgentStreamResult, StreamOptions, Message, Conversation, TokenUsage, ConversationManagerConfig, ModelConfig, JSONValue, InterruptResponse, DefaultToolContext, AgentTool, ToolDefinition } from './types.js';
 import { AgentErrors, blocksAgentError, InterruptError } from './errors.js';
 import { SlidingWindowConversationManager, SummarizingConversationManager } from '@strands-agents/sdk';
 import { BB_NAME, BB_VERSION } from './version.js';
@@ -67,7 +67,7 @@
  * @see https://strandsagents.com/docs/user-guide/concepts/agents/conversation-management/
  */
 function createConversationManager(config?: ConversationManagerConfig) {
 	if (!config || !config.strategy || config.strategy === 'sliding-window') {
 		const windowSize = config && 'windowSize' in config ? config.windowSize : undefined;
 		return new SlidingWindowConversationManager({ windowSize });
 	}
@@ -313,13 +313,13 @@
 
 		const strandsTools = toolDefs.map(t => tool({
 			// `name` is always set by resolveTools() from the Record key.
 			name: t.name!,
 			description: t.description,
 			inputSchema: t.parameters,
 			callback: (input, context) => t.handler({
 				input,
 				context: readContext(context),
 				interrupt: (params) => context!.interrupt(params),
 			}),
 		}));
 
@@ -403,7 +403,7 @@
 	 */
 	async stream(message: string, options?: StreamOptions<TContext>): Promise<AgentStreamResult> {
 		const conversationId = options?.conversationId;
-		const channelId = options?.channelId ?? conversationId ?? crypto.randomUUID();
+		const channelId = options?.channelId || conversationId || crypto.randomUUID();
 		if (!options?.userId && !this.config.inferenceOnly) throw blocksAgentError(AgentErrors.PersistenceRequired, 'userId is required when persistence is enabled. Pass it via options.userId.');
 		const userId = options?.userId ?? 'anonymous';
 		const context = this.resolveContext(options?.context);

packages/bb-agent/src/index.test.ts

@@ -115,6 +115,26 @@
 	});
 });
 
+// ── stream() empty channelId fallback ────────────────────────────────────────
+
+describe('stream() empty channelId fallback', () => {
+	test('empty channelId is treated as unset', async () => {
+		const scope = new Scope('test-empty-ch');
+		const agent = new Agent(scope, 'ec', { systemPrompt: 'test', model: { deployed: { provider: 'canned' }, local: { provider: 'canned' } } });
+		const result = await agent.stream('hello', { userId: 'test-user', channelId: '' });
+		assert.notStrictEqual(result.channelId, '');
+		assert.ok(result.channelId.length > 0);
+	});
+
+	test('empty conversationId is treated as unset', async () => {
+		const scope = new Scope('test-empty-conv');
+		const agent = new Agent(scope, 'ev', { systemPrompt: 'test', model: { deployed: { provider: 'canned' }, local: { provider: 'canned' } } });
+		const result = await agent.stream('hello', { userId: 'test-user', conversationId: '' });
+		assert.notStrictEqual(result.channelId, '');
+		assert.ok(result.channelId.length > 0);
+	});
+});
+
 // ── tool factory enforcement (compile-time) ──────────────────────────────────
 
 describe('tool factory enforcement', () => {
@@ -727,7 +747,7 @@
 
 		await chat.sendMessage('hello');
 		// Simulate error chunk from server
 		chunkHandler!({ type: 'error', error: 'model throttled' });
 
 		assert.strictEqual(errorReceived, 'model throttled');
 		assert.strictEqual(loadingStates.at(-1), false, 'loading should be false after error');
@@ -753,11 +773,11 @@
 		});
 
 		await chat.sendMessage('hello');
 		chunkHandler!({ type: 'interrupt', interrupts: [{ id: 'int-1', name: 'approve:deleteRecords', reason: { tool: 'deleteRecords' } }] });
 
 		assert.ok(interruptsReceived, 'onInterrupt should be called');
 		assert.strictEqual(interruptsReceived!.length, 1);
 		assert.strictEqual(interruptsReceived![0].name, 'approve:deleteRecords');
 		assert.strictEqual(loadingStates.at(-1), false, 'loading should be false after interrupt');
 	});
 

packages/bb-agent/src/types.ts

@@ -251,7 +251,7 @@ export interface ToolCallRecord {
 
 export interface StreamOptions<TContext = DefaultToolContext> {
 	conversationId?: string;
-	/** Channel ID for Realtime delivery. Defaults to conversationId or a random UUID. */
+	/** Channel ID for Realtime delivery. Defaults to conversationId or a random UUID. Empty strings are treated as unset. */
 	channelId?: string;
 	/** User ID for conversation scoping. Defaults to 'anonymous'. */
 	userId?: string;