Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>();
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
Expand Down
32 changes: 25 additions & 7 deletions libs/langgraph/src/lib/internals/subagent-tracker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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:<uuid>` 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) {
Expand Down
Loading