attribute subagent output to the task that spawned it
the harness stamps every message a subagent produces with parent_tool_use_id. the sidecar wrote it outgoing and nothing ever read it coming back, so a subagent's prose and tool calls were spliced into the main transcript as if the agent you are talking to had produced them — and worse, its deltas were appended to the same text buffer, so two voices were concatenated inside one bubble. both buffering layers (stream-parser's textBuffer and turn-stream's buffer) are now maps keyed by parent, and parentToolUseId rides on ChatEvent, ServerMessage and Message. useChat nests parented output under the Task row that spawned it; ToolActivity draws the trace inside the expanded panel. background tasks get the same treatment from the other end: task:started and task:notification were two unrelated fake assistant bubbles minutes apart, and are now one role:'task' row correlated by taskId that appears pending and resolves in place. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,8 @@ export type Message = {
|
||||
toolInput?: Record<string, unknown>;
|
||||
output?: string;
|
||||
isError?: boolean;
|
||||
/** The Task tool call this came from, when a subagent produced it. See `Parented` below. */
|
||||
parentToolUseId?: string;
|
||||
};
|
||||
|
||||
export type MessageCost = {
|
||||
@@ -82,6 +84,13 @@ export type ClientMessage =
|
||||
cursor: number;
|
||||
};
|
||||
|
||||
// The `Task` tool's id, when this piece of output came from a subagent rather than the agent you are
|
||||
// talking to. The harness stamps every nested message with `parent_tool_use_id`; nothing used to read it,
|
||||
// so a subagent's prose and tool calls were spliced into the main transcript as if the main agent had
|
||||
// produced them — and its deltas were appended to the main agent's text buffer, interleaving two voices
|
||||
// inside one bubble. Attributing them lets the UI file them under the Task row that spawned them.
|
||||
type Parented = { parentToolUseId?: string };
|
||||
|
||||
export type ServerMessage =
|
||||
| {
|
||||
type: 'session:init';
|
||||
@@ -91,26 +100,26 @@ export type ServerMessage =
|
||||
context?: string;
|
||||
contextId?: string;
|
||||
}
|
||||
| {
|
||||
| ({
|
||||
type: 'assistant:text';
|
||||
text: string;
|
||||
}
|
||||
| {
|
||||
} & Parented)
|
||||
| ({
|
||||
type: 'assistant:delta';
|
||||
text: string;
|
||||
}
|
||||
| {
|
||||
} & Parented)
|
||||
| ({
|
||||
type: 'tool:start';
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
toolInput: Record<string, unknown>;
|
||||
}
|
||||
| {
|
||||
} & Parented)
|
||||
| ({
|
||||
type: 'tool:result';
|
||||
toolCallId: string;
|
||||
output: string;
|
||||
isError: boolean;
|
||||
}
|
||||
} & Parented)
|
||||
| {
|
||||
type: 'result';
|
||||
sessionId: string;
|
||||
@@ -166,20 +175,20 @@ export type TurnMessageType =
|
||||
export type TurnMessage = Extract<ServerMessage, { type: TurnMessageType }> & { prevSeq?: number };
|
||||
|
||||
export type ChatEvent =
|
||||
| { type: 'text'; text: string }
|
||||
| { type: 'delta'; text: string }
|
||||
| {
|
||||
| ({ type: 'text'; text: string } & Parented)
|
||||
| ({ type: 'delta'; text: string } & Parented)
|
||||
| ({
|
||||
type: 'tool:start';
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
toolInput: Record<string, unknown>;
|
||||
}
|
||||
| {
|
||||
} & Parented)
|
||||
| ({
|
||||
type: 'tool:result';
|
||||
toolCallId: string;
|
||||
output: string;
|
||||
isError: boolean;
|
||||
}
|
||||
} & Parented)
|
||||
| {
|
||||
type: 'result';
|
||||
cost: MessageCost;
|
||||
|
||||
@@ -164,13 +164,23 @@ export function close(ws: ServerWebSocket<WSData>): void {
|
||||
function foldIntoSession(session: UserSession, msg: TurnMessage, model: string): void {
|
||||
switch (msg.type) {
|
||||
case 'assistant:delta':
|
||||
session.streamBuffer += msg.text;
|
||||
// A subagent's deltas are not the main agent typing; appending them here spliced its sentences into
|
||||
// whatever the agent you are talking to was mid-way through saying.
|
||||
if (!msg.parentToolUseId) session.streamBuffer += msg.text;
|
||||
break;
|
||||
|
||||
case 'assistant:text':
|
||||
session.messages.push({ id: randomUUID(), timestamp: Date.now(), role: 'assistant', text: msg.text, model });
|
||||
session.messages.push({
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'assistant',
|
||||
text: msg.text,
|
||||
model,
|
||||
parentToolUseId: msg.parentToolUseId,
|
||||
});
|
||||
session.meta.messageCount += 1;
|
||||
session.streamBuffer = '';
|
||||
// Only the main agent's own stream feeds the buffer a resume replays as `streamingText`.
|
||||
if (!msg.parentToolUseId) session.streamBuffer = '';
|
||||
break;
|
||||
|
||||
case 'tool:start':
|
||||
@@ -181,6 +191,7 @@ function foldIntoSession(session: UserSession, msg: TurnMessage, model: string):
|
||||
toolCallId: msg.toolCallId,
|
||||
toolName: msg.toolName,
|
||||
toolInput: msg.toolInput,
|
||||
parentToolUseId: msg.parentToolUseId,
|
||||
});
|
||||
session.meta.messageCount += 1;
|
||||
break;
|
||||
@@ -203,7 +214,8 @@ function foldIntoSession(session: UserSession, msg: TurnMessage, model: string):
|
||||
session.meta.cost.totalUSD += msg.cost.totalUSD;
|
||||
session.meta.updatedAt = Date.now();
|
||||
// The turn's cost belongs to the assistant message it paid for (as it did when officer built these).
|
||||
const last = session.messages[session.messages.length - 1];
|
||||
// The turn's cost is the main agent's, so skip past any subagent tail.
|
||||
const last = [...session.messages].reverse().find((m) => !m.parentToolUseId);
|
||||
if (last?.role === 'assistant' && !last.cost) last.cost = msg.cost;
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -3,6 +3,9 @@ import { processLine, createParseState, parseStream } from './stream-parser';
|
||||
import type { ChatEvent } from '../../api/chat/types';
|
||||
import type { StreamParserCallbacks, ParseState } from './stream-parser';
|
||||
|
||||
/** The main agent's accumulated deltas — buffers are keyed by parent_tool_use_id, '' being the agent. */
|
||||
const buffered = (state: ParseState, parent = '') => state.textBuffers.get(parent);
|
||||
|
||||
function makeCallbacks(): { events: ChatEvent[]; sessionIds: string[]; callbacks: StreamParserCallbacks } {
|
||||
const events: ChatEvent[] = [];
|
||||
const sessionIds: string[] = [];
|
||||
@@ -41,7 +44,7 @@ describe('processLine', () => {
|
||||
});
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([{ type: 'delta', text: 'hello' }]);
|
||||
expect(state.textBuffer).toBe('hello');
|
||||
expect(buffered(state)).toBe('hello');
|
||||
});
|
||||
|
||||
test('accumulates text buffer across deltas', () => {
|
||||
@@ -54,13 +57,13 @@ describe('processLine', () => {
|
||||
});
|
||||
processLine(mkDelta('hello '), state, callbacks);
|
||||
processLine(mkDelta('world'), state, callbacks);
|
||||
expect(state.textBuffer).toBe('hello world');
|
||||
expect(buffered(state)).toBe('hello world');
|
||||
expect(events).toHaveLength(2);
|
||||
});
|
||||
|
||||
test('handles assistant text block — clears text buffer', () => {
|
||||
const state = createParseState();
|
||||
state.textBuffer = 'partial';
|
||||
state.textBuffers.set('', 'partial');
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
const line = JSON.stringify({
|
||||
type: 'assistant',
|
||||
@@ -68,12 +71,12 @@ describe('processLine', () => {
|
||||
});
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([{ type: 'text', text: 'full response' }]);
|
||||
expect(state.textBuffer).toBe('');
|
||||
expect(buffered(state)).toBeUndefined();
|
||||
});
|
||||
|
||||
test('handles assistant tool_use block — flushes text buffer first', () => {
|
||||
const state = createParseState();
|
||||
state.textBuffer = 'thinking...';
|
||||
state.textBuffers.set('', 'thinking...');
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
const line = JSON.stringify({
|
||||
type: 'assistant',
|
||||
@@ -86,7 +89,7 @@ describe('processLine', () => {
|
||||
{ type: 'text', text: 'thinking...' },
|
||||
{ type: 'tool:start', toolCallId: 'call_1', toolName: 'sqlite', toolInput: { query: 'SELECT 1' } },
|
||||
]);
|
||||
expect(state.textBuffer).toBe('');
|
||||
expect(buffered(state)).toBeUndefined();
|
||||
});
|
||||
|
||||
test('handles assistant tool_use without prior text buffer', () => {
|
||||
@@ -99,9 +102,7 @@ describe('processLine', () => {
|
||||
},
|
||||
});
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([
|
||||
{ type: 'tool:start', toolCallId: 'call_1', toolName: 'email_db', toolInput: {} },
|
||||
]);
|
||||
expect(events).toEqual([{ type: 'tool:start', toolCallId: 'call_1', toolName: 'email_db', toolInput: {} }]);
|
||||
});
|
||||
|
||||
test('handles user tool_result with string content', () => {
|
||||
@@ -114,9 +115,7 @@ describe('processLine', () => {
|
||||
},
|
||||
});
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([
|
||||
{ type: 'tool:result', toolCallId: 'call_1', output: 'result text', isError: false },
|
||||
]);
|
||||
expect(events).toEqual([{ type: 'tool:result', toolCallId: 'call_1', output: 'result text', isError: false }]);
|
||||
});
|
||||
|
||||
test('handles user tool_result with array content', () => {
|
||||
@@ -140,9 +139,7 @@ describe('processLine', () => {
|
||||
},
|
||||
});
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([
|
||||
{ type: 'tool:result', toolCallId: 'call_2', output: 'line 1\nline 2', isError: true },
|
||||
]);
|
||||
expect(events).toEqual([{ type: 'tool:result', toolCallId: 'call_2', output: 'line 1\nline 2', isError: true }]);
|
||||
});
|
||||
|
||||
test('handles system init — captures session id', () => {
|
||||
@@ -155,7 +152,7 @@ describe('processLine', () => {
|
||||
|
||||
test('handles result — sets gotResult, emits cost', () => {
|
||||
const state = createParseState();
|
||||
state.textBuffer = 'trailing';
|
||||
state.textBuffers.set('', 'trailing');
|
||||
const { events, sessionIds, callbacks } = makeCallbacks();
|
||||
const line = JSON.stringify({
|
||||
type: 'result',
|
||||
@@ -170,13 +167,13 @@ describe('processLine', () => {
|
||||
expect(sessionIds).toEqual(['sess_xyz']);
|
||||
expect(events).toEqual([
|
||||
{ type: 'text', text: 'trailing' },
|
||||
{ type: 'result', cost: { inputTokens: 100, outputTokens: 50, totalUSD: 0.003 } },
|
||||
{ type: 'result', cost: { inputTokens: 100, outputTokens: 50, totalUSD: 0.003 }, claudeSessionId: 'sess_xyz' },
|
||||
]);
|
||||
});
|
||||
|
||||
test('handles error result — flushes buffer and emits error', () => {
|
||||
const state = createParseState();
|
||||
state.textBuffer = 'partial';
|
||||
state.textBuffers.set('', 'partial');
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
const line = JSON.stringify({ type: 'result', is_error: true, result: 'something broke' });
|
||||
processLine(line, state, callbacks);
|
||||
@@ -193,7 +190,87 @@ describe('processLine', () => {
|
||||
const line = JSON.stringify({ type: 'result', is_error: false });
|
||||
processLine(line, state, callbacks);
|
||||
expect(events).toEqual([
|
||||
{ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } },
|
||||
{ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 }, claudeSessionId: undefined },
|
||||
]);
|
||||
});
|
||||
|
||||
// ── Subagent attribution ──
|
||||
//
|
||||
// The harness stamps every message a subagent produces with `parent_tool_use_id` — the id of the Task
|
||||
// tool call that spawned it. Nothing read it, so a subagent's prose arrived as a top-level assistant
|
||||
// message and its tool calls looked like the main agent's. Worse, its deltas landed in the one shared
|
||||
// text buffer, so a flush emitted both voices concatenated into a single message.
|
||||
|
||||
test('stamps subagent output with its parent Task call', () => {
|
||||
const state = createParseState();
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
processLine(
|
||||
JSON.stringify({
|
||||
type: 'assistant',
|
||||
parent_tool_use_id: 'call_task',
|
||||
message: { content: [{ type: 'text', text: 'searched 40 files' }] },
|
||||
}),
|
||||
state,
|
||||
callbacks,
|
||||
);
|
||||
processLine(
|
||||
JSON.stringify({
|
||||
type: 'user',
|
||||
parent_tool_use_id: 'call_task',
|
||||
message: { content: [{ type: 'tool_result', tool_use_id: 'call_grep', content: 'hit', is_error: false }] },
|
||||
}),
|
||||
state,
|
||||
callbacks,
|
||||
);
|
||||
expect(events).toEqual([
|
||||
{ type: 'text', text: 'searched 40 files', parentToolUseId: 'call_task' },
|
||||
{ type: 'tool:result', toolCallId: 'call_grep', output: 'hit', isError: false, parentToolUseId: 'call_task' },
|
||||
]);
|
||||
});
|
||||
|
||||
test('the main agent and a subagent buffer their deltas separately', () => {
|
||||
const state = createParseState();
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
const delta = (text: string, parent?: string) =>
|
||||
JSON.stringify({
|
||||
type: 'stream_event',
|
||||
...(parent ? { parent_tool_use_id: parent } : {}),
|
||||
event: { type: 'content_block_delta', delta: { type: 'text_delta', text } },
|
||||
});
|
||||
|
||||
processLine(delta('I will '), state, callbacks);
|
||||
processLine(delta('reading the ', 'call_task'), state, callbacks);
|
||||
processLine(delta('delegate.'), state, callbacks);
|
||||
processLine(delta('config now.', 'call_task'), state, callbacks);
|
||||
|
||||
expect(buffered(state)).toBe('I will delegate.');
|
||||
expect(buffered(state, 'call_task')).toBe('reading the config now.');
|
||||
|
||||
// And a tool call by the subagent flushes only the subagent's sentence.
|
||||
processLine(
|
||||
JSON.stringify({
|
||||
type: 'assistant',
|
||||
parent_tool_use_id: 'call_task',
|
||||
message: { content: [{ type: 'tool_use', id: 'call_read', name: 'Read', input: {} }] },
|
||||
}),
|
||||
state,
|
||||
callbacks,
|
||||
);
|
||||
expect(events.at(-2)).toEqual({ type: 'text', text: 'reading the config now.', parentToolUseId: 'call_task' });
|
||||
expect(buffered(state)).toBe('I will delegate.');
|
||||
});
|
||||
|
||||
test('a result flushes every buffer, subagents included', () => {
|
||||
const state = createParseState();
|
||||
const { events, callbacks } = makeCallbacks();
|
||||
state.textBuffers.set('', 'main said this');
|
||||
state.textBuffers.set('call_task', 'subagent said this');
|
||||
|
||||
processLine(JSON.stringify({ type: 'result', is_error: false }), state, callbacks);
|
||||
|
||||
expect(events.slice(0, 2)).toEqual([
|
||||
{ type: 'text', text: 'main said this' },
|
||||
{ type: 'text', text: 'subagent said this', parentToolUseId: 'call_task' },
|
||||
]);
|
||||
});
|
||||
|
||||
|
||||
@@ -16,14 +16,30 @@ type StreamParserCallbacks = {
|
||||
};
|
||||
|
||||
type ParseState = {
|
||||
textBuffer: string;
|
||||
/**
|
||||
* Accumulated deltas, keyed by the `parent_tool_use_id` that produced them ('' for the agent you are
|
||||
* talking to). One shared buffer was wrong as soon as a subagent ran: its deltas appended to whatever
|
||||
* the main agent had said so far, and the next flush emitted the two of them concatenated as a single
|
||||
* message from the main agent.
|
||||
*/
|
||||
textBuffers: Map<string, string>;
|
||||
gotResult: boolean;
|
||||
};
|
||||
|
||||
function flushTextBuffer(state: ParseState, onEvent: (event: ChatEvent) => void): void {
|
||||
if (state.textBuffer) {
|
||||
onEvent({ type: 'text', text: state.textBuffer });
|
||||
state.textBuffer = '';
|
||||
/** '' when a message is the main agent's own. Keys the buffer map and stamps every event we emit. */
|
||||
function parentOf(msg: Record<string, unknown>): string {
|
||||
const id = msg.parent_tool_use_id;
|
||||
return typeof id === 'string' ? id : '';
|
||||
}
|
||||
|
||||
/** Only set the field when there IS a parent, so main-agent events stay byte-identical on the wire. */
|
||||
const parented = (parent: string) => (parent ? { parentToolUseId: parent } : {});
|
||||
|
||||
function flushTextBuffer(state: ParseState, parent: string, onEvent: (event: ChatEvent) => void): void {
|
||||
const buffered = state.textBuffers.get(parent);
|
||||
if (buffered) {
|
||||
onEvent({ type: 'text', text: buffered, ...parented(parent) });
|
||||
state.textBuffers.delete(parent);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,8 +48,9 @@ function handleStreamEvent(msg: Record<string, unknown>, state: ParseState, onEv
|
||||
if (event?.type === 'content_block_delta') {
|
||||
const delta = event.delta as Record<string, unknown> | undefined;
|
||||
if (delta?.type === 'text_delta' && typeof delta.text === 'string') {
|
||||
state.textBuffer += delta.text;
|
||||
onEvent({ type: 'delta', text: delta.text });
|
||||
const parent = parentOf(msg);
|
||||
state.textBuffers.set(parent, (state.textBuffers.get(parent) ?? '') + delta.text);
|
||||
onEvent({ type: 'delta', text: delta.text, ...parented(parent) });
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -43,17 +60,20 @@ function handleAssistant(msg: Record<string, unknown>, state: ParseState, onEven
|
||||
const content = message?.content as Array<Record<string, unknown>> | undefined;
|
||||
if (!Array.isArray(content)) return;
|
||||
|
||||
const parent = parentOf(msg);
|
||||
|
||||
for (const block of content) {
|
||||
if (block.type === 'text' && typeof block.text === 'string') {
|
||||
onEvent({ type: 'text', text: block.text });
|
||||
state.textBuffer = '';
|
||||
onEvent({ type: 'text', text: block.text, ...parented(parent) });
|
||||
state.textBuffers.delete(parent);
|
||||
} else if (block.type === 'tool_use') {
|
||||
flushTextBuffer(state, onEvent);
|
||||
flushTextBuffer(state, parent, onEvent);
|
||||
onEvent({
|
||||
type: 'tool:start',
|
||||
toolCallId: (block.id as string) ?? '',
|
||||
toolName: (block.name as string) ?? 'unknown',
|
||||
toolInput: (block.input as Record<string, unknown>) ?? {},
|
||||
...parented(parent),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -64,6 +84,8 @@ function handleUser(msg: Record<string, unknown>, onEvent: (event: ChatEvent) =>
|
||||
const content = message?.content as Array<Record<string, unknown>> | undefined;
|
||||
if (!Array.isArray(content)) return;
|
||||
|
||||
const parent = parentOf(msg);
|
||||
|
||||
for (const block of content) {
|
||||
if (block.type === 'tool_result') {
|
||||
let output = '';
|
||||
@@ -80,28 +102,30 @@ function handleUser(msg: Record<string, unknown>, onEvent: (event: ChatEvent) =>
|
||||
toolCallId: (block.tool_use_id as string) ?? '',
|
||||
output,
|
||||
isError: (block.is_error as boolean) ?? false,
|
||||
...parented(parent),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function handleResult(
|
||||
msg: Record<string, unknown>,
|
||||
state: ParseState,
|
||||
callbacks: StreamParserCallbacks,
|
||||
): void {
|
||||
function handleResult(msg: Record<string, unknown>, state: ParseState, callbacks: StreamParserCallbacks): void {
|
||||
state.gotResult = true;
|
||||
|
||||
const isError = (msg.is_error as boolean) ?? false;
|
||||
const resultText = (msg.result as string) ?? '';
|
||||
|
||||
// A `result` ends the turn, so every buffer is stale — including a subagent's, if it died mid-sentence.
|
||||
function flushAll(): void {
|
||||
for (const parent of [...state.textBuffers.keys()]) flushTextBuffer(state, parent, callbacks.onEvent);
|
||||
}
|
||||
|
||||
if (isError) {
|
||||
flushTextBuffer(state, callbacks.onEvent);
|
||||
flushAll();
|
||||
callbacks.onEvent({ type: 'error', message: resultText || 'Claude Code returned an error' });
|
||||
return;
|
||||
}
|
||||
|
||||
flushTextBuffer(state, callbacks.onEvent);
|
||||
flushAll();
|
||||
|
||||
const usage = msg.usage as Record<string, number> | undefined;
|
||||
const cost: MessageCost = {
|
||||
@@ -180,11 +204,7 @@ export function processMessage(
|
||||
* Process a single NDJSON line from Claude Code's stream output.
|
||||
* Returns false if the line was skipped (empty or malformed), true otherwise.
|
||||
*/
|
||||
export function processLine(
|
||||
line: string,
|
||||
state: ParseState,
|
||||
callbacks: StreamParserCallbacks,
|
||||
): boolean {
|
||||
export function processLine(line: string, state: ParseState, callbacks: StreamParserCallbacks): boolean {
|
||||
if (!line.trim()) return false;
|
||||
|
||||
let msg: Record<string, unknown>;
|
||||
@@ -201,7 +221,7 @@ export function processLine(
|
||||
* Create a fresh parse state for a new stream.
|
||||
*/
|
||||
export function createParseState(): ParseState {
|
||||
return { textBuffer: '', gotResult: false };
|
||||
return { textBuffers: new Map(), gotResult: false };
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -20,39 +20,53 @@ export type TurnStream = {
|
||||
};
|
||||
|
||||
export function createTurnStream(sessionId: string): TurnStream {
|
||||
let buffer = '';
|
||||
// One buffer per speaker, keyed by the Task tool call that owns it ('' for the main agent). A subagent
|
||||
// runs concurrently with the agent that spawned it, so a single buffer interleaved their sentences and
|
||||
// flushed the result as one message attributed to whoever happened to hit the next boundary.
|
||||
const buffers = new Map<string, string>();
|
||||
|
||||
// Emit whatever deltas have accumulated as one complete message. Called at every boundary where the
|
||||
// assistant stops talking, so the transcript holds text blocks rather than a thousand fragments.
|
||||
function flush(): TurnOutput[] {
|
||||
if (!buffer) return [];
|
||||
const text = buffer;
|
||||
buffer = '';
|
||||
return [{ msg: { type: 'assistant:text', text }, durable: true }];
|
||||
/** Only set the field when there IS a parent, so main-agent messages stay byte-identical on the wire. */
|
||||
const parented = (parent: string) => (parent ? { parentToolUseId: parent } : {});
|
||||
|
||||
// Emit whatever deltas have accumulated as one complete message. Called at every boundary where that
|
||||
// speaker stops talking, so the transcript holds text blocks rather than a thousand fragments.
|
||||
function flush(parent: string): TurnOutput[] {
|
||||
const text = buffers.get(parent);
|
||||
if (!text) return [];
|
||||
buffers.delete(parent);
|
||||
return [{ msg: { type: 'assistant:text', text, ...parented(parent) }, durable: true }];
|
||||
}
|
||||
|
||||
/** A turn's end invalidates every buffer, the subagents' included. */
|
||||
function flushAll(): TurnOutput[] {
|
||||
return [...buffers.keys()].flatMap(flush);
|
||||
}
|
||||
|
||||
function push(event: ChatEvent): TurnOutput[] {
|
||||
const parent = ('parentToolUseId' in event ? event.parentToolUseId : undefined) ?? '';
|
||||
|
||||
switch (event.type) {
|
||||
case 'delta':
|
||||
buffer += event.text;
|
||||
return [{ msg: { type: 'assistant:delta', text: event.text }, durable: false }];
|
||||
buffers.set(parent, (buffers.get(parent) ?? '') + event.text);
|
||||
return [{ msg: { type: 'assistant:delta', text: event.text, ...parented(parent) }, durable: false }];
|
||||
|
||||
case 'text': {
|
||||
// An explicit full text block wins over the accumulated deltas that produced it.
|
||||
const text = event.text || buffer;
|
||||
buffer = '';
|
||||
return text ? [{ msg: { type: 'assistant:text', text }, durable: true }] : [];
|
||||
const text = event.text || buffers.get(parent) || '';
|
||||
buffers.delete(parent);
|
||||
return text ? [{ msg: { type: 'assistant:text', text, ...parented(parent) }, durable: true }] : [];
|
||||
}
|
||||
|
||||
case 'tool:start':
|
||||
return [
|
||||
...flush(),
|
||||
...flush(parent),
|
||||
{
|
||||
msg: {
|
||||
type: 'tool:start',
|
||||
toolCallId: event.toolCallId,
|
||||
toolName: event.toolName,
|
||||
toolInput: event.toolInput,
|
||||
...parented(parent),
|
||||
},
|
||||
durable: true,
|
||||
},
|
||||
@@ -61,14 +75,20 @@ export function createTurnStream(sessionId: string): TurnStream {
|
||||
case 'tool:result':
|
||||
return [
|
||||
{
|
||||
msg: { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError },
|
||||
msg: {
|
||||
type: 'tool:result',
|
||||
toolCallId: event.toolCallId,
|
||||
output: event.output,
|
||||
isError: event.isError,
|
||||
...parented(parent),
|
||||
},
|
||||
durable: true,
|
||||
},
|
||||
];
|
||||
|
||||
case 'result':
|
||||
return [
|
||||
...flush(),
|
||||
...flushAll(),
|
||||
{
|
||||
msg: { type: 'result', sessionId, cost: event.cost, claudeSessionId: event.claudeSessionId },
|
||||
durable: true,
|
||||
|
||||
Reference in New Issue
Block a user