diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts index 06800153d..b8d66eaed 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts @@ -2906,6 +2906,52 @@ describe('createStreamManagerBridge', () => { destroy$.next(); }); + it('never cross-wires concurrent children when arrival order != dispatch order', async () => { + const transport = new MockAgentTransport(); + const subjects = makeSubjects(); + const destroy$ = new Subject(); + const bridge = createStreamManagerBridge({ + options: { apiUrl: '', assistantId: 'test', transport, subagentToolNames: ['task'] }, + subjects, threadId$: of(null), destroy$: destroy$.asObservable(), + }); + bridge.submit({}); + // Parent dispatches TWO children in one AI message: alpha then beta. + transport.emit([{ + type: 'messages', + messages: [{ + id: 'ai-1', type: 'ai', content: '', + tool_calls: [ + { id: 'call_ALPHA', name: 'task', args: { subagent_type: 'alpha', task_description: 'a' } }, + { id: 'call_BETA', name: 'task', args: { subagent_type: 'beta', task_description: 'b' } }, + ], + }], + } satisfies StreamEvent]); + // BETA's child streams FIRST (parallel fan-out; arrival order != dispatch order). + transport.emit([{ + type: 'messages|tools:ns-BETA' as StreamEvent['type'], namespace: ['tools:ns-BETA'], + messages: [{ id: 'm-beta', type: 'AIMessageChunk', content: 'beta output' }], + messageMetadata: { checkpoint_ns: 'tools:ns-BETA' }, + } satisfies StreamEvent]); + transport.emit([{ + type: 'messages|tools:ns-ALPHA' as StreamEvent['type'], namespace: ['tools:ns-ALPHA'], + messages: [{ id: 'm-alpha', type: 'AIMessageChunk', content: 'alpha output' }], + messageMetadata: { checkpoint_ns: 'tools:ns-ALPHA' }, + } satisfies StreamEvent]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + const alpha = subjects.subagents$.value.get('call_ALPHA'); + const beta = subjects.subagents$.value.get('call_BETA'); + // Two children are outstanding at once and the namespaces carry no link to + // the tool-call ids, so neither stream can be attributed honestly. The old + // fallback claimed the first unmapped call and handed alpha's card beta's + // output. Refusing to guess is the correct outcome: an empty card, never a + // confidently wrong one. + const txt = (x: unknown) => (x as { content?: string } | undefined)?.content; + expect(txt(alpha?.messages()[0])).not.toBe('beta output'); + expect(txt(beta?.messages()[0])).not.toBe('alpha output'); + destroy$.next(); + }); + it('attributes a tool child whose values carry no human first message', async () => { // The real shape emitted by cockpit/chat/subagents, captured off the wire: // the delegation tool takes `task_description` (not `description`), and the diff --git a/libs/langgraph/src/lib/internals/subagent-tracker.ts b/libs/langgraph/src/lib/internals/subagent-tracker.ts index c9f9b8ec0..1825d2f52 100644 --- a/libs/langgraph/src/lib/internals/subagent-tracker.ts +++ b/libs/langgraph/src/lib/internals/subagent-tracker.ts @@ -186,13 +186,31 @@ export class SubagentTracker { } } - // Last-resort fallback — tool children only. A subgraph child is keyed by - // its own namespace and must never absorb an unrelated child's events. - for (const [toolCallId, subagent] of this.subagents) { - if (subagent.kind !== 'tool') continue; - if (!mapped.has(toolCallId) && (subagent.status === 'pending' || subagent.status === 'running')) { - return establish(toolCallId); - } + // Last-resort fallback — tool children only, and only when there is + // nothing to guess between. + // + // LangGraph's `tools:` namespace is a checkpoint id assigned + // independently of the parent's `call_*` tool-call id; the two are not + // linked anywhere on the wire (verified against a live run). So when a + // delegation tool carries no matchable description, position is the only + // signal left. + // + // That is sound with exactly one outstanding child — the shape every graph + // in this repo produces, since each dispatches one tool call per assistant + // turn. With several outstanding at once (parallel fan-out) arrival order + // is NOT dispatch order, and claiming the first unmapped call cross-wires + // the children: one card renders another's output. Leaving the stream + // unattributed keeps its messages buffered instead, so an empty card is + // the worst case rather than a confidently wrong one. It can still resolve + // later: as siblings complete, the candidate set shrinks back to one. + const candidates = [...this.subagents].filter( + ([toolCallId, subagent]) => + subagent.kind === 'tool' && + !mapped.has(toolCallId) && + (subagent.status === 'pending' || subagent.status === 'running'), + ); + if (candidates.length === 1) { + return establish(candidates[0][0]); } if (description) {