Skip to content

Commit e8ec118

Browse files
bloveclaude
andcommitted
fix(langgraph): subagent cards rendered an empty transcript
Every subagent card in cockpit/chat/subagents and examples/chat showed '0 message(s)' despite the child streaming a full response. Three defects stacked, each hiding the next. 1. Attribution never ran. matchSubgraphToSubagent was only called when a child's values.messages[0] was a HUMAN message. Wire capture shows this graph's child starts with the AI reply, so the namespace UUID was never mapped to its call_* id. Now any tools: child claims its tool call on first sight, falling through to the positional rung the ladder already relied on (the delegation tool here takes task_description, not description, so both text rungs were unreachable anyway). 2. Pre-attribution chunks were dropped. addMessageToSubagent silently returned when the namespace had no registered subagent. They are now buffered per namespace and replayed by establish(). 3. Delta chunks replaced instead of accumulating. Children stream AIMessageChunk deltas — ~14 chars each, ~500 per message — and the tracker merged by id with a plain overwrite, keeping only the last tiny delta. So even a correctly attributed card rendered an empty message. Now folds chunk content the way the parent transcript has since #751; snapshots still replace. Verified live against a real model: all three cards go 0 -> 1 message(s), each attributed to its own distinct call_* id, and expanding one renders the child's actual research text. 343/343 lib tests (3 new, each written from wire-captured shapes after an earlier hypothesis-shaped test passed while the UI stayed broken). e2e: chat-subagents, langgraph-subgraphs, deep-agents-subagents, examples/chat 54/54. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 parent d418b38 commit e8ec118

3 files changed

Lines changed: 242 additions & 9 deletions

File tree

libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2817,6 +2817,146 @@ describe('createStreamManagerBridge', () => {
28172817
destroy$.next();
28182818
});
28192819

2820+
it('replays child messages that streamed before the namespace was attributed', async () => {
2821+
const transport = new MockAgentTransport();
2822+
const subjects = makeSubjects();
2823+
const destroy$ = new Subject<void>();
2824+
const bridge = createStreamManagerBridge({
2825+
options: { apiUrl: '', assistantId: 'test', transport, subagentToolNames: ['task'] },
2826+
subjects,
2827+
threadId$: of(null),
2828+
destroy$: destroy$.asObservable(),
2829+
});
2830+
2831+
bridge.submit({});
2832+
// Parent registers the delegation under the TOOL CALL id.
2833+
transport.emit([{
2834+
type: 'messages',
2835+
messages: [{
2836+
id: 'ai-1', type: 'ai', content: '',
2837+
tool_calls: [{ id: 'call_abc', name: 'task', args: { subagent_type: 'researcher', description: 'Research signals' } }],
2838+
}],
2839+
} satisfies StreamEvent]);
2840+
2841+
// The child streams under an INTERNAL UUID namespace, not the tool-call id.
2842+
// This is what LangGraph actually emits — verified on the wire.
2843+
transport.emit([{
2844+
type: 'messages|tools:aa5c61a1-e3ee-ea36' as StreamEvent['type'],
2845+
namespace: ['tools:aa5c61a1-e3ee-ea36'],
2846+
messages: [{ id: 'sub-1', type: 'ai', content: 'early chunk' }],
2847+
messageMetadata: { checkpoint_ns: 'tools:aa5c61a1-e3ee-ea36|model' },
2848+
} satisfies StreamEvent]);
2849+
2850+
// Attribution only arrives later, via a values event carrying the child's
2851+
// first human message, which the description ladder matches on.
2852+
transport.emit([{
2853+
type: 'values|tools:aa5c61a1-e3ee-ea36' as StreamEvent['type'],
2854+
namespace: ['tools:aa5c61a1-e3ee-ea36'],
2855+
data: { messages: [{ type: 'human', content: 'Research signals' }] },
2856+
} as StreamEvent]);
2857+
transport.close();
2858+
2859+
await new Promise(r => setTimeout(r, 10));
2860+
2861+
// The pre-attribution chunk must not be lost — this is what made every
2862+
// subagent card render "0 message(s)".
2863+
expect(subjects.subagents$.value.get('call_abc')?.messages()).toEqual([
2864+
expect.objectContaining({ id: 'sub-1', content: 'early chunk' }),
2865+
]);
2866+
destroy$.next();
2867+
});
2868+
2869+
it('accumulates a child\'s streamed delta chunks instead of keeping only the last', async () => {
2870+
// Child graphs stream AIMessageChunk deltas — ~14 chars each, hundreds per
2871+
// message (measured on the wire). Replacing by id keeps only the final
2872+
// delta, which rendered an attributed card as an empty message.
2873+
const transport = new MockAgentTransport();
2874+
const subjects = makeSubjects();
2875+
const destroy$ = new Subject<void>();
2876+
const bridge = createStreamManagerBridge({
2877+
options: { apiUrl: '', assistantId: 'test', transport, subagentToolNames: ['task'] },
2878+
subjects,
2879+
threadId$: of(null),
2880+
destroy$: destroy$.asObservable(),
2881+
});
2882+
2883+
bridge.submit({});
2884+
transport.emit([{
2885+
type: 'messages',
2886+
messages: [{
2887+
id: 'ai-1', type: 'ai', content: '',
2888+
tool_calls: [{ id: 'call_x', name: 'task', args: { subagent_type: 'research', task_description: 'x' } }],
2889+
}],
2890+
} satisfies StreamEvent]);
2891+
2892+
for (const part of ['LAX ', 'is a ', 'large hub']) {
2893+
transport.emit([{
2894+
type: 'messages|tools:ns-1' as StreamEvent['type'],
2895+
namespace: ['tools:ns-1'],
2896+
messages: [{ id: 'chunk-1', type: 'AIMessageChunk', content: part }],
2897+
messageMetadata: { checkpoint_ns: 'tools:ns-1|model' },
2898+
} satisfies StreamEvent]);
2899+
}
2900+
transport.close();
2901+
await new Promise(r => setTimeout(r, 10));
2902+
2903+
const msgs = subjects.subagents$.value.get('call_x')?.messages() ?? [];
2904+
expect(msgs).toHaveLength(1);
2905+
expect((msgs[0] as unknown as { content: string }).content).toBe('LAX is a large hub');
2906+
destroy$.next();
2907+
});
2908+
2909+
it('attributes a tool child whose values carry no human first message', async () => {
2910+
// The real shape emitted by cockpit/chat/subagents, captured off the wire:
2911+
// the delegation tool takes `task_description` (not `description`), and the
2912+
// child's values.messages[0] is an AI message, never a human one. Both
2913+
// description rungs of the ladder are therefore unreachable, so attribution
2914+
// has to fall back to the unmapped pending/running child.
2915+
const transport = new MockAgentTransport();
2916+
const subjects = makeSubjects();
2917+
const destroy$ = new Subject<void>();
2918+
const bridge = createStreamManagerBridge({
2919+
options: { apiUrl: '', assistantId: 'test', transport, subagentToolNames: ['task'] },
2920+
subjects,
2921+
threadId$: of(null),
2922+
destroy$: destroy$.asObservable(),
2923+
});
2924+
2925+
bridge.submit({});
2926+
transport.emit([{
2927+
type: 'messages',
2928+
messages: [{
2929+
id: 'ai-1', type: 'ai', content: '',
2930+
tool_calls: [{ id: 'call_kd4Q', name: 'task', args: { subagent_type: 'research', task_description: 'Airport details' } }],
2931+
}],
2932+
} satisfies StreamEvent]);
2933+
2934+
// Child streams under an internal UUID namespace.
2935+
transport.emit([{
2936+
type: 'messages|tools:f61899a8-459d' as StreamEvent['type'],
2937+
namespace: ['tools:f61899a8-459d'],
2938+
messages: [{ id: 'sub-1', type: 'ai', content: 'LAX is a large hub' }],
2939+
messageMetadata: { checkpoint_ns: 'tools:f61899a8-459d|model' },
2940+
} satisfies StreamEvent]);
2941+
2942+
// Its values carry an AI first message — no human anywhere.
2943+
transport.emit([{
2944+
type: 'values|tools:f61899a8-459d' as StreamEvent['type'],
2945+
namespace: ['tools:f61899a8-459d'],
2946+
data: { messages: [{ type: 'ai', content: 'LAX is a large hub' }] },
2947+
} as StreamEvent]);
2948+
transport.close();
2949+
2950+
await new Promise(r => setTimeout(r, 10));
2951+
2952+
// This is what the "0 message(s)" card bug looked like: status settled but
2953+
// the child's transcript never arrived.
2954+
expect(subjects.subagents$.value.get('call_kd4Q')?.messages()).toEqual([
2955+
expect.objectContaining({ id: 'sub-1', content: 'LAX is a large hub' }),
2956+
]);
2957+
destroy$.next();
2958+
});
2959+
28202960
it('routes plain-subgraph message tuples to a namespace-keyed child stream, never the transcript', async () => {
28212961
const transport = new MockAgentTransport();
28222962
const subjects = makeSubjects();

libs/langgraph/src/lib/internals/stream-manager.bridge.ts

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -770,6 +770,9 @@ export function createStreamManagerBridge<T, ResolvedBag extends BagTemplate = B
770770
if (child) {
771771
if (child.kind === 'subgraph') {
772772
subagentManager.ensureSubgraphStream(child.key, child.name);
773+
} else {
774+
// Claim the stream for its tool call before its tokens arrive.
775+
subagentManager.ensureToolStreamAttribution(child.key);
773776
}
774777
for (const msg of normalized) {
775778
subagentManager.addMessageToSubagent(child.key, msg);
@@ -986,14 +989,15 @@ export function createStreamManagerBridge<T, ResolvedBag extends BagTemplate = B
986989
if (!child) return;
987990

988991
if (child.kind === 'tool') {
989-
// Attribution ladder applies to tool children only: their namespace id
990-
// may need mapping onto a registered tool call.
992+
// Prefer the precise description match when the child's first message is
993+
// the human task; otherwise claim the stream positionally so a graph that
994+
// doesn't fit that shape still gets attributed.
991995
const messages = values['messages'];
992-
if (Array.isArray(messages) && messages.length > 0) {
993-
const first = messages[0];
994-
if (isRecord(first) && (first['type'] === 'human' || first['type'] === 'user') && typeof first['content'] === 'string') {
995-
subagentManager.matchSubgraphToSubagent(child.key, first['content']);
996-
}
996+
const first = Array.isArray(messages) && messages.length > 0 ? messages[0] : undefined;
997+
if (isRecord(first) && (first['type'] === 'human' || first['type'] === 'user') && typeof first['content'] === 'string') {
998+
subagentManager.matchSubgraphToSubagent(child.key, first['content']);
999+
} else {
1000+
subagentManager.ensureToolStreamAttribution(child.key);
9971001
}
9981002
} else {
9991003
subagentManager.ensureSubgraphStream(child.key, child.name);

libs/langgraph/src/lib/internals/subagent-tracker.ts

Lines changed: 91 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,18 @@ export class SubagentTracker {
5858
private readonly onSubagentChange?: () => void;
5959
private readonly subagents = new Map<string, TrackedSubagent>();
6060
private readonly namespaceToToolCallId = new Map<string, string>();
61+
/**
62+
* Child messages received under a namespace that is not yet attributed to a
63+
* registered subagent. LangGraph's `tools:<id>` namespace carries an internal
64+
* run UUID, not the parent's tool-call id, and the two are only reconciled
65+
* once a `values` event arrives carrying the child's first human message. Any
66+
* chunk streamed before that point would otherwise be dropped — which is what
67+
* made subagent cards render "0 message(s)" despite a full child transcript.
68+
*
69+
* Merged by id like a real transcript, so this stays bounded by the child's
70+
* distinct message count rather than by chunk volume.
71+
*/
72+
private readonly unattributedMessages = new Map<string, BaseMessage[]>();
6173
private readonly pendingMatches = new Map<string, string>();
6274

6375
constructor(options: SubagentTrackerOptions = {}) {
@@ -69,6 +81,7 @@ export class SubagentTracker {
6981
this.subagents.clear();
7082
this.namespaceToToolCallId.clear();
7183
this.pendingMatches.clear();
84+
this.unattributedMessages.clear();
7285
this.onSubagentChange?.();
7386
}
7487

@@ -145,10 +158,13 @@ export class SubagentTracker {
145158
this.namespaceToToolCallId.set(namespaceId, toolCallId);
146159
const subagent = this.subagents.get(toolCallId);
147160
if (subagent) {
161+
const buffered = this.unattributedMessages.get(namespaceId);
148162
this.subagents.set(toolCallId, {
149163
...subagent,
150164
status: subagent.status === 'complete' || subagent.status === 'error' ? subagent.status : 'running',
165+
messages: buffered ? mergeMessages(subagent.messages, buffered) : subagent.messages,
151166
});
167+
this.unattributedMessages.delete(namespaceId);
152168
}
153169
this.onSubagentChange?.();
154170
return toolCallId;
@@ -204,6 +220,27 @@ export class SubagentTracker {
204220
this.onSubagentChange?.();
205221
}
206222

223+
/**
224+
* Attribute a `tools:` child stream to its parent tool call as soon as the
225+
* child is seen, without requiring a description to match on.
226+
*
227+
* The description ladder needs two things this repo's own graphs don't
228+
* reliably provide: a delegation tool that names its argument `description`,
229+
* and a child whose first message is the human task. `cockpit/chat/subagents`
230+
* has neither — it uses `task_description`, and its child's messages begin
231+
* with the AI reply. Attribution therefore never ran, so the child's
232+
* transcript was never claimed and every card rendered "0 message(s)".
233+
*
234+
* Calling the ladder with no description skips both description rungs and
235+
* lands on the positional fallback (first unmapped pending/running tool
236+
* child), which is correct for sequential dispatch and is the same heuristic
237+
* the ladder already relied on in practice.
238+
*/
239+
ensureToolStreamAttribution(namespaceId: string): void {
240+
if (this.namespaceToToolCallId.has(namespaceId)) return;
241+
this.matchSubgraphToSubagent(namespaceId, '');
242+
}
243+
207244
/**
208245
* Register a plain-subgraph child stream on its first namespaced event.
209246
*
@@ -259,7 +296,15 @@ export class SubagentTracker {
259296
addMessageToSubagent(namespaceId: string, message: BaseMessage): void {
260297
const toolCallId = this.resolveToolCallId(namespaceId);
261298
const subagent = this.subagents.get(toolCallId);
262-
if (!subagent) return;
299+
if (!subagent) {
300+
// Not attributed yet — hold it rather than drop it. `establish()` will
301+
// replay the buffer the moment this namespace is matched to a tool call.
302+
this.unattributedMessages.set(
303+
namespaceId,
304+
mergeMessages(this.unattributedMessages.get(namespaceId) ?? [], [message]),
305+
);
306+
return;
307+
}
263308

264309
this.subagents.set(toolCallId, {
265310
...subagent,
@@ -397,14 +442,58 @@ function mergeMessages(existing: BaseMessage[], incoming: BaseMessage[]): BaseMe
397442
const id = getMessageId(msg);
398443
const idx = id ? merged.findIndex(m => getMessageId(m) === id) : -1;
399444
if (idx >= 0) {
400-
merged[idx] = msg;
445+
merged[idx] = accumulateChunk(merged[idx], msg);
401446
} else {
402447
merged.push(msg);
403448
}
404449
}
405450
return merged;
406451
}
407452

453+
/**
454+
* Fold a streamed chunk into the message it belongs to.
455+
*
456+
* A child graph streams `AIMessageChunk`s that are *deltas* — a handful of
457+
* characters each, hundreds per message. Replacing by id (the previous
458+
* behavior) therefore kept only the final delta, so a fully attributed
459+
* subagent still rendered a near-empty message. Snapshots, which carry the
460+
* message-so-far, still replace.
461+
*
462+
* This mirrors the parent transcript's delta handling: append unconditionally
463+
* rather than comparing text, because a prefix-style "dedupe" silently eats
464+
* legitimate tokens that happen to repeat the accumulated prefix.
465+
*/
466+
function accumulateChunk(existing: BaseMessage, incoming: BaseMessage): BaseMessage {
467+
if (!isChunkMessage(incoming)) return incoming;
468+
const previousText = extractText((existing as unknown as Record<string, unknown>)['content']);
469+
const incomingText = extractText((incoming as unknown as Record<string, unknown>)['content']);
470+
if (!incomingText) return existing;
471+
if (!previousText) return incoming;
472+
return { ...(incoming as object), content: previousText + incomingText } as BaseMessage;
473+
}
474+
475+
function isChunkMessage(message: BaseMessage): boolean {
476+
const type = (message as unknown as Record<string, unknown>)['type'];
477+
return typeof type === 'string' && type.endsWith('Chunk');
478+
}
479+
480+
function extractText(content: unknown): string {
481+
if (typeof content === 'string') return content;
482+
if (!Array.isArray(content)) return '';
483+
let out = '';
484+
for (const block of content) {
485+
if (typeof block === 'string') { out += block; continue; }
486+
if (block == null || typeof block !== 'object') continue;
487+
const record = block as Record<string, unknown>;
488+
const blockType = record['type'];
489+
if (blockType === 'text' || blockType === 'output_text' || blockType === undefined) {
490+
const text = record['text'];
491+
if (typeof text === 'string') out += text;
492+
}
493+
}
494+
return out;
495+
}
496+
408497
function getMessageId(message: BaseMessage): string | undefined {
409498
return (message as unknown as { id?: string }).id;
410499
}

0 commit comments

Comments
 (0)