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 82a2964d9..06800153d 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts @@ -2817,6 +2817,146 @@ describe('createStreamManagerBridge', () => { destroy$.next(); }); + it('replays child messages that streamed before the namespace was attributed', 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 registers the delegation under the TOOL CALL id. + transport.emit([{ + type: 'messages', + messages: [{ + id: 'ai-1', type: 'ai', content: '', + tool_calls: [{ id: 'call_abc', name: 'task', args: { subagent_type: 'researcher', description: 'Research signals' } }], + }], + } satisfies StreamEvent]); + + // The child streams under an INTERNAL UUID namespace, not the tool-call id. + // This is what LangGraph actually emits — verified on the wire. + transport.emit([{ + type: 'messages|tools:aa5c61a1-e3ee-ea36' as StreamEvent['type'], + namespace: ['tools:aa5c61a1-e3ee-ea36'], + messages: [{ id: 'sub-1', type: 'ai', content: 'early chunk' }], + messageMetadata: { checkpoint_ns: 'tools:aa5c61a1-e3ee-ea36|model' }, + } satisfies StreamEvent]); + + // Attribution only arrives later, via a values event carrying the child's + // first human message, which the description ladder matches on. + transport.emit([{ + type: 'values|tools:aa5c61a1-e3ee-ea36' as StreamEvent['type'], + namespace: ['tools:aa5c61a1-e3ee-ea36'], + data: { messages: [{ type: 'human', content: 'Research signals' }] }, + } as StreamEvent]); + transport.close(); + + await new Promise(r => setTimeout(r, 10)); + + // The pre-attribution chunk must not be lost — this is what made every + // subagent card render "0 message(s)". + expect(subjects.subagents$.value.get('call_abc')?.messages()).toEqual([ + expect.objectContaining({ id: 'sub-1', content: 'early chunk' }), + ]); + destroy$.next(); + }); + + it('accumulates a child\'s streamed delta chunks instead of keeping only the last', async () => { + // Child graphs stream AIMessageChunk deltas — ~14 chars each, hundreds per + // message (measured on the wire). Replacing by id keeps only the final + // delta, which rendered an attributed card as an empty message. + 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({}); + transport.emit([{ + type: 'messages', + messages: [{ + id: 'ai-1', type: 'ai', content: '', + tool_calls: [{ id: 'call_x', name: 'task', args: { subagent_type: 'research', task_description: 'x' } }], + }], + } satisfies StreamEvent]); + + for (const part of ['LAX ', 'is a ', 'large hub']) { + transport.emit([{ + type: 'messages|tools:ns-1' as StreamEvent['type'], + namespace: ['tools:ns-1'], + messages: [{ id: 'chunk-1', type: 'AIMessageChunk', content: part }], + messageMetadata: { checkpoint_ns: 'tools:ns-1|model' }, + } satisfies StreamEvent]); + } + transport.close(); + await new Promise(r => setTimeout(r, 10)); + + const msgs = subjects.subagents$.value.get('call_x')?.messages() ?? []; + expect(msgs).toHaveLength(1); + expect((msgs[0] as unknown as { content: string }).content).toBe('LAX is a large hub'); + 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 + // child's values.messages[0] is an AI message, never a human one. Both + // description rungs of the ladder are therefore unreachable, so attribution + // has to fall back to the unmapped pending/running child. + 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({}); + transport.emit([{ + type: 'messages', + messages: [{ + id: 'ai-1', type: 'ai', content: '', + tool_calls: [{ id: 'call_kd4Q', name: 'task', args: { subagent_type: 'research', task_description: 'Airport details' } }], + }], + } satisfies StreamEvent]); + + // Child streams under an internal UUID namespace. + transport.emit([{ + type: 'messages|tools:f61899a8-459d' as StreamEvent['type'], + namespace: ['tools:f61899a8-459d'], + messages: [{ id: 'sub-1', type: 'ai', content: 'LAX is a large hub' }], + messageMetadata: { checkpoint_ns: 'tools:f61899a8-459d|model' }, + } satisfies StreamEvent]); + + // Its values carry an AI first message — no human anywhere. + transport.emit([{ + type: 'values|tools:f61899a8-459d' as StreamEvent['type'], + namespace: ['tools:f61899a8-459d'], + data: { messages: [{ type: 'ai', content: 'LAX is a large hub' }] }, + } as StreamEvent]); + transport.close(); + + await new Promise(r => setTimeout(r, 10)); + + // This is what the "0 message(s)" card bug looked like: status settled but + // the child's transcript never arrived. + expect(subjects.subagents$.value.get('call_kd4Q')?.messages()).toEqual([ + expect.objectContaining({ id: 'sub-1', content: 'LAX is a large hub' }), + ]); + destroy$.next(); + }); + it('routes plain-subgraph message tuples to a namespace-keyed child stream, never the transcript', async () => { const transport = new MockAgentTransport(); const subjects = makeSubjects(); diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts index caab64fae..446e14e9f 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts @@ -770,6 +770,9 @@ export function createStreamManagerBridge 0) { - const first = messages[0]; - if (isRecord(first) && (first['type'] === 'human' || first['type'] === 'user') && typeof first['content'] === 'string') { - subagentManager.matchSubgraphToSubagent(child.key, first['content']); - } + const first = Array.isArray(messages) && messages.length > 0 ? messages[0] : undefined; + if (isRecord(first) && (first['type'] === 'human' || first['type'] === 'user') && typeof first['content'] === 'string') { + subagentManager.matchSubgraphToSubagent(child.key, first['content']); + } else { + subagentManager.ensureToolStreamAttribution(child.key); } } else { subagentManager.ensureSubgraphStream(child.key, child.name); diff --git a/libs/langgraph/src/lib/internals/subagent-tracker.ts b/libs/langgraph/src/lib/internals/subagent-tracker.ts index 895a86bf7..c9f9b8ec0 100644 --- a/libs/langgraph/src/lib/internals/subagent-tracker.ts +++ b/libs/langgraph/src/lib/internals/subagent-tracker.ts @@ -58,6 +58,18 @@ export class SubagentTracker { private readonly onSubagentChange?: () => void; private readonly subagents = new Map(); private readonly namespaceToToolCallId = new Map(); + /** + * Child messages received under a namespace that is not yet attributed to a + * registered subagent. LangGraph's `tools:` namespace carries an internal + * run UUID, not the parent's tool-call id, and the two are only reconciled + * once a `values` event arrives carrying the child's first human message. Any + * chunk streamed before that point would otherwise be dropped — which is what + * made subagent cards render "0 message(s)" despite a full child transcript. + * + * Merged by id like a real transcript, so this stays bounded by the child's + * distinct message count rather than by chunk volume. + */ + private readonly unattributedMessages = new Map(); private readonly pendingMatches = new Map(); constructor(options: SubagentTrackerOptions = {}) { @@ -69,6 +81,7 @@ export class SubagentTracker { this.subagents.clear(); this.namespaceToToolCallId.clear(); this.pendingMatches.clear(); + this.unattributedMessages.clear(); this.onSubagentChange?.(); } @@ -145,10 +158,13 @@ export class SubagentTracker { this.namespaceToToolCallId.set(namespaceId, toolCallId); const subagent = this.subagents.get(toolCallId); if (subagent) { + const buffered = this.unattributedMessages.get(namespaceId); this.subagents.set(toolCallId, { ...subagent, status: subagent.status === 'complete' || subagent.status === 'error' ? subagent.status : 'running', + messages: buffered ? mergeMessages(subagent.messages, buffered) : subagent.messages, }); + this.unattributedMessages.delete(namespaceId); } this.onSubagentChange?.(); return toolCallId; @@ -204,6 +220,27 @@ export class SubagentTracker { this.onSubagentChange?.(); } + /** + * Attribute a `tools:` child stream to its parent tool call as soon as the + * child is seen, without requiring a description to match on. + * + * The description ladder needs two things this repo's own graphs don't + * reliably provide: a delegation tool that names its argument `description`, + * and a child whose first message is the human task. `cockpit/chat/subagents` + * has neither — it uses `task_description`, and its child's messages begin + * with the AI reply. Attribution therefore never ran, so the child's + * transcript was never claimed and every card rendered "0 message(s)". + * + * Calling the ladder with no description skips both description rungs and + * lands on the positional fallback (first unmapped pending/running tool + * child), which is correct for sequential dispatch and is the same heuristic + * the ladder already relied on in practice. + */ + ensureToolStreamAttribution(namespaceId: string): void { + if (this.namespaceToToolCallId.has(namespaceId)) return; + this.matchSubgraphToSubagent(namespaceId, ''); + } + /** * Register a plain-subgraph child stream on its first namespaced event. * @@ -259,7 +296,15 @@ export class SubagentTracker { addMessageToSubagent(namespaceId: string, message: BaseMessage): void { const toolCallId = this.resolveToolCallId(namespaceId); const subagent = this.subagents.get(toolCallId); - if (!subagent) return; + if (!subagent) { + // Not attributed yet — hold it rather than drop it. `establish()` will + // replay the buffer the moment this namespace is matched to a tool call. + this.unattributedMessages.set( + namespaceId, + mergeMessages(this.unattributedMessages.get(namespaceId) ?? [], [message]), + ); + return; + } this.subagents.set(toolCallId, { ...subagent, @@ -397,7 +442,7 @@ function mergeMessages(existing: BaseMessage[], incoming: BaseMessage[]): BaseMe const id = getMessageId(msg); const idx = id ? merged.findIndex(m => getMessageId(m) === id) : -1; if (idx >= 0) { - merged[idx] = msg; + merged[idx] = accumulateChunk(merged[idx], msg); } else { merged.push(msg); } @@ -405,6 +450,50 @@ function mergeMessages(existing: BaseMessage[], incoming: BaseMessage[]): BaseMe return merged; } +/** + * Fold a streamed chunk into the message it belongs to. + * + * A child graph streams `AIMessageChunk`s that are *deltas* — a handful of + * characters each, hundreds per message. Replacing by id (the previous + * behavior) therefore kept only the final delta, so a fully attributed + * subagent still rendered a near-empty message. Snapshots, which carry the + * message-so-far, still replace. + * + * This mirrors the parent transcript's delta handling: append unconditionally + * rather than comparing text, because a prefix-style "dedupe" silently eats + * legitimate tokens that happen to repeat the accumulated prefix. + */ +function accumulateChunk(existing: BaseMessage, incoming: BaseMessage): BaseMessage { + if (!isChunkMessage(incoming)) return incoming; + const previousText = extractText((existing as unknown as Record)['content']); + const incomingText = extractText((incoming as unknown as Record)['content']); + if (!incomingText) return existing; + if (!previousText) return incoming; + return { ...(incoming as object), content: previousText + incomingText } as BaseMessage; +} + +function isChunkMessage(message: BaseMessage): boolean { + const type = (message as unknown as Record)['type']; + return typeof type === 'string' && type.endsWith('Chunk'); +} + +function extractText(content: unknown): string { + if (typeof content === 'string') return content; + if (!Array.isArray(content)) return ''; + let out = ''; + for (const block of content) { + if (typeof block === 'string') { out += block; continue; } + if (block == null || typeof block !== 'object') continue; + const record = block as Record; + const blockType = record['type']; + if (blockType === 'text' || blockType === 'output_text' || blockType === undefined) { + const text = record['text']; + if (typeof text === 'string') out += text; + } + } + return out; +} + function getMessageId(message: BaseMessage): string | undefined { return (message as unknown as { id?: string }).id; }