diff --git a/apps/desktop/src/main/__tests__/materialize-turns.test.ts b/apps/desktop/src/main/__tests__/materialize-turns.test.ts index 048da46b2d..527f1aaca9 100644 --- a/apps/desktop/src/main/__tests__/materialize-turns.test.ts +++ b/apps/desktop/src/main/__tests__/materialize-turns.test.ts @@ -171,6 +171,41 @@ describe('materializeTurns', () => { assert.equal(turns[0]?.assistantThinking, 'first I considered...'); }); + it('concatenates per-step assistant messages into one answer with the first step id and last ts', () => { + // A multi-step turn persists one AssistantMessage per model step; the turn + // view-model joins their text (and thinking) in order, anchors on the first + // step id, and measures durationMs to the final step. + const turns = materializeTurns([ + userMsg('t1', 100, 'q'), + toolCallMsg('t1', 101, 'c1'), + toolResultMsg('t1', 102, 'c1'), + { + type: 'assistant', + id: 'step-1', + turnId: 't1', + ts: 103, + text: 'first, I check the file', + modelId: 'claude-sonnet-4-5', + thinking: { text: 'reasoning one' }, + } as StoredMessage, + { + type: 'assistant', + id: 'step-2', + turnId: 't1', + ts: 205, + text: 'here is the answer', + modelId: 'claude-sonnet-4-5', + thinking: { text: 'reasoning two' }, + } as StoredMessage, + ]); + assert.equal(turns.length, 1); + assert.equal(turns[0]?.assistant?.id, 'step-1'); + assert.equal(turns[0]?.assistant?.text, 'first, I check the file\n\nhere is the answer'); + assert.equal(turns[0]?.assistant?.ts, 205); + assert.equal(turns[0]?.assistantThinking, 'reasoning one\n\nreasoning two'); + assert.equal(turns[0]?.durationMs, 105); + }); + it('leaves durationMs undefined when assistant message is missing (in-progress turn)', () => { const turns = materializeTurns([userMsg('t1', 100, 'q')]); assert.equal(turns[0]?.durationMs, undefined); diff --git a/apps/desktop/src/renderer/app-shell-session-events.ts b/apps/desktop/src/renderer/app-shell-session-events.ts index 83d9a0d82b..7975a0cc3d 100644 --- a/apps/desktop/src/renderer/app-shell-session-events.ts +++ b/apps/desktop/src/renderer/app-shell-session-events.ts @@ -295,9 +295,18 @@ export function createAppShellSessionEventHandlers(options: { // racing a committed-message refresh. setStreamingBySession((current) => { const prevSlot = current[sessionId]; - const prevText = prevSlot?.text ?? ''; + // Per-step guard: each model step streams under its own messageId. When + // the id changes (a new step began, or a prior step's slot is still + // draining), start this step's bubble fresh instead of appending onto + // the previous step's text — otherwise the next step's answer would + // flicker duplicated onto the last one. + const sameMessage = + prevSlot === undefined + || prevSlot.messageId === undefined + || prevSlot.messageId === event.messageId; + const prevText = sameMessage ? (prevSlot?.text ?? '') : ''; const applied = applyAssistantDelta(prevText, event.text); - const nextTruncated = (prevSlot?.truncated ?? false) || applied.truncated; + const nextTruncated = (sameMessage ? (prevSlot?.truncated ?? false) : false) || applied.truncated; // Avoid a re-render when nothing materially changed (e.g. // a non-string `event.text` defensively dropped by the // helper, no truncated change). diff --git a/apps/desktop/src/renderer/conversation-markdown.ts b/apps/desktop/src/renderer/conversation-markdown.ts index b1da15081f..6a26f7f3ff 100644 --- a/apps/desktop/src/renderer/conversation-markdown.ts +++ b/apps/desktop/src/renderer/conversation-markdown.ts @@ -43,7 +43,11 @@ export function renderConversationMarkdown(sessionName: string, messages: Stored for (const tid of turnOrder) { const turnMessages = byTurn.get(tid) ?? []; const user = turnMessages.find((m) => m.type === 'user'); - const assistant = turnMessages.find((m) => m.type === 'assistant'); + // A turn holds one assistant message per model step; join their text in step + // order so the export carries the whole answer, not just the first step. + const assistantText = turnMessages + .flatMap((m) => (m.type === 'assistant' && m.text.length > 0 ? [m.text] : [])) + .join('\n\n'); const toolCalls = turnMessages.filter((m) => m.type === 'tool_call'); if (user) { @@ -67,13 +71,13 @@ export function renderConversationMarkdown(sessionName: string, messages: Stored lines.push(''); } - if (assistant) { + if (assistantText.length > 0) { lines.push('## Maka'); lines.push(''); // Defensive: backend redacts at write-time, but the export landing // in the user's clipboard is a high-risk surface — paste destinations // are external. Second-layer redaction is cheap insurance. - lines.push(redactSecrets((assistant as { text: string }).text)); + lines.push(redactSecrets(assistantText)); lines.push(''); } } diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index 513f7947f5..3c256e4829 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -102,6 +102,14 @@ export interface ToolStartEvent extends BaseEvent { args: unknown; displayName?: string; intent?: string; + /** + * Id of the assistant step this tool call belongs to (equals the step's + * AssistantMessage id / the step's text+thinking messageId). Lets model + * replay group a step's reasoning + text + tool calls into one provider + * assistant message. Absent on legacy events; consumers treat a missing + * stepId as un-pairable (degraded, per-turn) history. + */ + stepId?: string; } export type ToolOutputStream = typeof TOOL_OUTPUT_STREAMS[number]; diff --git a/packages/core/src/runtime-event.ts b/packages/core/src/runtime-event.ts index 25c9a5dd6e..8eb6f188f8 100644 --- a/packages/core/src/runtime-event.ts +++ b/packages/core/src/runtime-event.ts @@ -241,6 +241,14 @@ export interface RuntimeEventRefs { toolCallId?: string; providerEventId?: string; artifactId?: string; + /** + * Assistant step id for a function_call event: the id of the step's + * text/thinking messages (their `providerEventId`). Model replay pairs a + * step's signed thinking with its tool calls by this id. Absent on legacy + * (per-turn) events; a missing stepId marks history that cannot be paired + * and is replayed with the older degraded semantics. + */ + stepId?: string; } // ============================================================================ diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 78da75be17..87de6336c8 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -2053,6 +2053,62 @@ describe('AiSdkBackend model history', () => { ); }); + test('provider error mid-step still persists the streamed partial text (partialOutputRetained)', async () => { + // Codex P1: the non-abort error exit (provider failure / watchdog timeout) + // must flush the in-flight step's partial accumulators just like the abort + // exit does — the user already saw the streamed text, so it belongs in the + // ledger. The gate releases only after the backend has emitted the partial + // text_delta, so consumption-before-error is deterministic. + const gate = makeGate(); + const model = new MockLanguageModelV3({ + doStream: { + stream: new ReadableStream({ + async start(controller) { + controller.enqueue({ type: 'stream-start', warnings: [] }); + controller.enqueue({ type: 'text-start', id: 'text-1' }); + controller.enqueue({ type: 'text-delta', id: 'text-1', delta: 'partial answer' }); + await gate.promise; + controller.error(new Error('provider exploded mid-step')); + }, + }), + }, + }); + const assistants: AssistantMessage[] = []; + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message) => { + if (message.type === 'assistant') assistants.push(message); + }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + if (event.type === 'text_delta' && event.text === 'partial answer') gate.release(); + } + + // The streamed partial persists as this step's AssistantMessage. + assert.equal(assistants.length, 1); + assert.equal(assistants[0]!.text, 'partial answer'); + // And the turn still closes as an error, not a false success. + assert.equal(events.some((event) => event.type === 'error'), true); + const completes = events.filter((event) => event.type === 'complete'); + assert.equal(completes.length > 0, true); + assert.equal( + completes.every((event) => (event as { stopReason?: string }).stopReason === 'error'), + true, + ); + }); + test('writes host history compact block and replays the host summary in the same request', async () => { const model = completionModel(); const events: SessionEvent[] = []; @@ -2456,7 +2512,11 @@ describe('AiSdkBackend model history', () => { assert.equal(prompt.includes('large fold source fact 0'), false); }); - test('uses StoredMessage projection when RuntimeEvent tool results are unmatched', async () => { + test('keeps RuntimeEvent replay when a tool result is unmatched (orphan dropped, rest replayed)', async () => { + // `unmatched_tool_result` is a non-blocking diagnostic: the materializer + // drops the orphan itself (a standalone tool message is an Anthropic 400), + // so the ledger stays on RuntimeEvent replay instead of falling back to + // StoredMessage projection. const model = completionModel(); const backend = new AiSdkBackend({ sessionId: 'session-1', @@ -2491,9 +2551,9 @@ describe('AiSdkBackend model history', () => { ], })); + // RuntimeEvent replay (not the StoredMessage projection), orphan gone. assert.deepEqual(compactPrompt(model), [ - { role: 'user', content: [{ type: 'text', text: 'projection user' }] }, - { role: 'assistant', content: [{ type: 'text', text: 'projection assistant' }] }, + { role: 'user', content: [{ type: 'text', text: 'runtime user' }] }, { role: 'user', content: [{ type: 'text', text: 'current user' }] }, ]); }); @@ -5691,14 +5751,168 @@ describe('AiSdkBackend thinking persistence', () => { assert.match(prompt, /sig-replay/); }); - test('signed thinking from a tool-calling turn is NOT replayed as a stray reasoning block', async () => { - // Reproduce the ledger a signed Anthropic tool turn produces. The backend - // accumulates the turn's reasoning and emits ONE thinking_complete AFTER the - // tool events, so the ledger order is tool_start → tool_result → - // thinking_complete → text_complete. If that thinking re-entered - // provider-native replay it would materialize as an assistant reasoning - // message positioned AFTER the tool result — Anthropic 400. The replay must - // send the tool call/result but drop the thinking. + test('signed thinking from a per-step tool-calling turn IS replayed, merged with its tool call', async () => { + // Per-step ledger: the tool_start carries the step id (stepId === the + // step's message id 'm1'), so the step's signed reasoning + text + tool call + // regroup into ONE assistant message on replay (reasoning leads, then text, + // then the tool call, then the tool result) — the Anthropic-valid shape. + const ctx = { + sessionId: 'session-1', + invocationId: 'inv-1', + runId: 'run-prev', + turnId: 'turn-prev', + now: () => 7, + newId: idGenerator(), + } as unknown as InvocationContext; + const memory = createSessionEventMapMemory(); + const priorEvents: SessionEvent[] = [ + { type: 'tool_start', id: 'e1', turnId: 'turn-prev', ts: 1, toolUseId: 'tool-1', toolName: 'Read', args: { path: 'package.json' }, stepId: 'm1' }, + { type: 'tool_result', id: 'e2', turnId: 'turn-prev', ts: 2, toolUseId: 'tool-1', isError: false, content: { kind: 'text', text: 'file contents' } }, + { type: 'thinking_complete', id: 'e3', turnId: 'turn-prev', ts: 3, messageId: 'm1', text: 'reasoning about the tool result', signature: 'sig-tool' }, + { type: 'text_complete', id: 'e4', turnId: 'turn-prev', ts: 4, messageId: 'm1', text: 'the answer' }, + ]; + const runtimeContext = priorEvents.map((event) => mapSessionEventToRuntimeEvent(event, ctx, memory)); + + const secondModel = completionModel(); + const secondBackend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => secondModel, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + }); + + await drain(secondBackend.send({ turnId: 'turn-current', text: 'follow up', context: [], runtimeContext })); + + const prompt = JSON.stringify(compactPrompt(secondModel)); + // Reasoning (with signature), text, and the tool call all reach the request. + assert.match(prompt, /"type":"reasoning"/); + assert.match(prompt, /sig-tool/); + assert.match(prompt, /reasoning about the tool result/); + assert.match(prompt, /"toolName":"Read"|"toolCallId":"tool-1"/); + // Reasoning leads the tool call inside the assistant message (Anthropic order). + assert.ok(prompt.indexOf('reasoning about the tool result') < prompt.indexOf('tool-1')); + }); + + test('thinking-only tool step (no text) replays reasoning + tool call in one assistant message without an empty text block', async () => { + // Anthropic interleaved thinking's most common step shape: the step reasons, + // calls a tool, and produces NO closing text — the backend still flushes the + // step's AssistantMessage (text: '') so the signed block persists, and emits + // text_complete with empty text. On replay the step must merge into ONE + // assistant message [reasoning, tool-call] with NO empty text part between + // them (emitStep skips text.length === 0; an empty text block is provider + // noise and this locks that skip path). + const ctx = { + sessionId: 'session-1', + invocationId: 'inv-1', + runId: 'run-prev', + turnId: 'turn-prev', + now: () => 7, + newId: idGenerator(), + } as unknown as InvocationContext; + const memory = createSessionEventMapMemory(); + const priorEvents: SessionEvent[] = [ + { type: 'tool_start', id: 'e1', turnId: 'turn-prev', ts: 1, toolUseId: 'tool-1', toolName: 'Read', args: { path: 'package.json' }, stepId: 'm1' }, + { type: 'tool_result', id: 'e2', turnId: 'turn-prev', ts: 2, toolUseId: 'tool-1', isError: false, content: { kind: 'text', text: 'file contents' } }, + { type: 'thinking_complete', id: 'e3', turnId: 'turn-prev', ts: 3, messageId: 'm1', text: 'plan the read', signature: 'sig-interleaved' }, + { type: 'text_complete', id: 'e4', turnId: 'turn-prev', ts: 4, messageId: 'm1', text: '' }, + ]; + const runtimeContext = priorEvents.map((event) => mapSessionEventToRuntimeEvent(event, ctx, memory)); + + const secondModel = completionModel(); + const secondBackend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => secondModel, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + }); + + await drain(secondBackend.send({ turnId: 'turn-current', text: 'follow up', context: [], runtimeContext })); + + const prompt = compactPrompt(secondModel) as Array<{ role: string; content: unknown }>; + const assistantMessages = prompt.filter((message) => message.role === 'assistant'); + assert.equal(assistantMessages.length, 1, 'reasoning and tool call must merge into one assistant message'); + const parts = assistantMessages[0]!.content as Array<{ type: string; text?: string }>; + // Reasoning leads the tool call; no text part at all (not even an empty one). + assert.deepEqual(parts.map((part) => part.type), ['reasoning', 'tool-call']); + assert.equal(parts[0]!.text, 'plan the read'); + const promptJson = JSON.stringify(prompt); + assert.match(promptJson, /sig-interleaved/); + assert.match(promptJson, /"toolCallId":"tool-1"/); + }); + + test('an orphan tool_result does not degrade replay: dropped, while paired history replays provider-native', async () => { + // Codex P2: `unmatched_tool_result` must not be a blocking diagnostic — the + // materializer intentionally drops the orphan (a standalone tool message is + // an Anthropic 400), so one orphan must not push the whole ledger back to + // stored-message projection. Paired call/result and the step's signed + // reasoning must all still reach the provider request. + const ctx = { + sessionId: 'session-1', + invocationId: 'inv-1', + runId: 'run-prev', + turnId: 'turn-prev', + now: () => 7, + newId: idGenerator(), + } as unknown as InvocationContext; + const memory = createSessionEventMapMemory(); + const priorEvents: SessionEvent[] = [ + // Orphan: result with no prior tool_start (its call was sliced away). + { type: 'tool_result', id: 'e0', turnId: 'turn-prev', ts: 1, toolUseId: 'tool-orphan', isError: false, content: { kind: 'text', text: 'orphan payload' } }, + // Paired per-step tool call + result + signed reasoning + text. + { type: 'tool_start', id: 'e1', turnId: 'turn-prev', ts: 2, toolUseId: 'tool-1', toolName: 'Read', args: { path: 'package.json' }, stepId: 'm1' }, + { type: 'tool_result', id: 'e2', turnId: 'turn-prev', ts: 3, toolUseId: 'tool-1', isError: false, content: { kind: 'text', text: 'file contents' } }, + { type: 'thinking_complete', id: 'e3', turnId: 'turn-prev', ts: 4, messageId: 'm1', text: 'plan the read', signature: 'sig-paired' }, + { type: 'text_complete', id: 'e4', turnId: 'turn-prev', ts: 5, messageId: 'm1', text: 'the answer' }, + ]; + const runtimeContext = priorEvents.map((event) => mapSessionEventToRuntimeEvent(event, ctx, memory)); + + const secondModel = completionModel(); + const secondBackend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => secondModel, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + }); + + await drain(secondBackend.send({ turnId: 'turn-current', text: 'follow up', context: [], runtimeContext })); + + const prompt = compactPrompt(secondModel) as Array<{ role: string; content: unknown }>; + const promptJson = JSON.stringify(prompt); + // Provider-native replay happened: reasoning + signature + paired tool pair. + assert.match(promptJson, /"type":"reasoning"/); + assert.match(promptJson, /sig-paired/); + assert.match(promptJson, /"toolCallId":"tool-1"/); + assert.match(promptJson, /file contents/); + // The orphan result is dropped — no tool message for it anywhere. + assert.doesNotMatch(promptJson, /tool-orphan/); + assert.doesNotMatch(promptJson, /orphan payload/); + }); + + test('signed thinking from a legacy (unpaired) tool turn is NOT replayed as a stray reasoning block', async () => { + // Legacy per-turn ledger: the tool_start carries NO step id, so its + // end-of-turn reasoning cannot be paired to a tool-use assistant message and + // is still dropped from replay (no worse than before; avoids Anthropic 400). const ctx = { sessionId: 'session-1', invocationId: 'inv-1', @@ -5834,6 +6048,162 @@ describe('AiSdkBackend thinking persistence', () => { assert.match(prompt, /"type":"reasoning"/); assert.match(prompt, /sig-omitted/); }); + + test('grace notice never reuses a taken step id when the stream ends without a trailing finish-step', async () => { + // ChatGPT P2: on the catch-all path (no trailing finish-step) the last + // step's id is already taken — by the thinking-only AssistantMessage the + // catch-all flush just wrote, and by the tool step's tool_start.stepId. The + // grace notice must mint its own id or the ledger gets a duplicate message + // id / replay adopts the grace text as the tool step's closer. + // + // streamText always synthesizes trailing step boundaries, so drive the + // backend through a patched startStream: step 1 runs a real tool via the + // wrapped execute (genuine tool_start.stepId), step 2 is thinking-only and + // the stream ends abruptly with no finish-step / finish. + const appended: StoredMessage[] = []; + const events: SessionEvent[] = []; + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message) => { appended.push(message); }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => completionModel(), + tools: [testTool('Read', z.object({ path: z.string() }))], + newId: idGenerator(), + now: monotonicClock(), + }); + type FakeStreamInput = { + tools: Record Promise }>; + abortSignal: AbortSignal; + }; + (backend as unknown as { + modelAdapter: { startStream: (input: FakeStreamInput) => Promise }; + }).modelAdapter.startStream = async (input: FakeStreamInput) => ({ + fullStream: (async function* () { + // Step 1 (pure tool): execute mid-step, then close the step. + await input.tools['Read']!.execute({ path: 'a.md' }, { toolCallId: 'tool-1', abortSignal: input.abortSignal }); + yield { type: 'finish-step', finishReason: { unified: 'tool-calls', raw: 'tool_calls' } }; + // Step 2 (thinking-only): signed reasoning, then the stream ends with + // NO trailing finish-step and NO finish chunk. + yield { type: 'reasoning-delta', delta: 'final thoughts' }; + yield { type: 'reasoning-delta', delta: '', providerMetadata: { anthropic: { signature: 'sig-last' } } }; + })(), + usage: Promise.resolve(undefined), + totalUsage: Promise.resolve(undefined), + finishReason: Promise.resolve('tool-calls'), + }); + + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + const assistants = appended.filter((m): m is AssistantMessage => m.type === 'assistant'); + // Catch-all flush persisted the thinking-only step; grace added its own row. + assert.equal(assistants.length, 2); + const thinkingOnly = assistants.find((m) => m.thinking?.signature === 'sig-last'); + const grace = assistants.find((m) => m.text.includes('步工具调用上限')); + assert.ok(thinkingOnly, 'thinking-only last step must persist'); + assert.ok(grace, 'grace notice must persist'); + // The grace id collides with nothing: not the last step's assistant row... + assert.notEqual(grace.id, thinkingOnly.id); + // ...and not any tool step's stepId. + const toolStepIds = events + .filter((event): event is Extract => event.type === 'tool_start') + .map((event) => event.stepId); + assert.equal(toolStepIds.length, 1); + assert.equal(toolStepIds.includes(grace.id), false); + // No duplicate message ids anywhere in the ledger. + const ids = appended.map((m) => (m as { id: string }).id); + assert.equal(new Set(ids).size, ids.length, `duplicate ledger ids: ${ids.join(', ')}`); + }); + + test('flushes one AssistantMessage per step, each with its own thinking + signature, and stamps tool_start.stepId', async () => { + // Two-step tool turn: step 1 reasons + calls a tool; step 2 reasons + answers. + // Each step must persist its own AssistantMessage with its own signature, and + // the step-1 tool_start must carry the step-1 assistant id. + let streamCalls = 0; + const model = new MockLanguageModelV3({ + doStream: async () => { + streamCalls += 1; + const chunks: LanguageModelV3StreamPart[] = streamCalls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { type: 'reasoning-start', id: 'r1' }, + { type: 'reasoning-delta', id: 'r1', delta: 'think one' }, + { type: 'reasoning-delta', id: 'r1', delta: '', providerMetadata: { anthropic: { signature: 'sig-step-1' } } }, + { type: 'reasoning-end', id: 'r1' }, + { type: 'text-start', id: 't1' }, + { type: 'text-delta', id: 't1', delta: 'calling the tool' }, + { type: 'text-end', id: 't1' }, + { type: 'tool-call', toolCallId: 'tool-1', toolName: 'Read', input: JSON.stringify({ path: 'a.md' }) }, + { type: 'finish', finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, usage: { inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 }, outputTokens: { total: 1, text: 1, reasoning: 0 } } }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'reasoning-start', id: 'r2' }, + { type: 'reasoning-delta', id: 'r2', delta: 'think two' }, + { type: 'reasoning-delta', id: 'r2', delta: '', providerMetadata: { anthropic: { signature: 'sig-step-2' } } }, + { type: 'reasoning-end', id: 'r2' }, + { type: 'text-start', id: 't2' }, + { type: 'text-delta', id: 't2', delta: 'final answer' }, + { type: 'text-end', id: 't2' }, + { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' }, usage: { inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 }, outputTokens: { total: 1, text: 1, reasoning: 0 } } }, + ]; + return { + stream: simulateReadableStream({ chunks, initialDelayInMs: null, chunkDelayInMs: null }), + }; + }, + }); + + const assistants: AssistantMessage[] = []; + const events: SessionEvent[] = []; + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (m) => { if (m.type === 'assistant') assistants.push(m); }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [testTool('Read', z.object({ path: z.string() }))], + newId: idGenerator(), + now: monotonicClock(), + }); + + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + // Two assistant rows with distinct ids and correctly paired signatures. + assert.equal(assistants.length, 2); + assert.equal(assistants[0]!.text, 'calling the tool'); + assert.equal(assistants[0]!.thinking?.text, 'think one'); + assert.equal(assistants[0]!.thinking?.signature, 'sig-step-1'); + assert.equal(assistants[1]!.text, 'final answer'); + assert.equal(assistants[1]!.thinking?.text, 'think two'); + assert.equal(assistants[1]!.thinking?.signature, 'sig-step-2'); + assert.notEqual(assistants[0]!.id, assistants[1]!.id); + + // The tool_start of step 1 carries the step-1 assistant id. + const toolStart = events.find( + (event): event is Extract => event.type === 'tool_start', + ); + assert.ok(toolStart, 'expected a tool_start event'); + assert.equal(toolStart.stepId, assistants[0]!.id); + + // Each step emits its own thinking_complete/text_complete pointing at its row. + const textCompletes = events.filter( + (event): event is Extract => event.type === 'text_complete', + ); + assert.deepEqual( + textCompletes.map((event) => [event.messageId, event.text]), + [[assistants[0]!.id, 'calling the tool'], [assistants[1]!.id, 'final answer']], + ); + }); }); async function runArchiveGatedReplay(input: { diff --git a/packages/runtime/src/__tests__/model-adapter.test.ts b/packages/runtime/src/__tests__/model-adapter.test.ts index de77ceae94..ae58fdf62e 100644 --- a/packages/runtime/src/__tests__/model-adapter.test.ts +++ b/packages/runtime/src/__tests__/model-adapter.test.ts @@ -76,6 +76,47 @@ describe('ModelAdapter stream and error normalization', () => { assert.equal(error?.message, 'Rate limit exceeded'); }); + test('treats AI SDK v6 step boundaries (start-step / finish-step) as no-ops', () => { + const events: SessionEvent[] = []; + const queue = new AsyncEventQueue(); + const adapter = newAdapter(); + const callbacks = { + textCalls: 0, + thinkingCalls: 0, + signatureCalls: 0, + onText() { this.textCalls += 1; }, + onTextComplete() {}, + onThinking() { this.thinkingCalls += 1; }, + onThinkingSignature() { this.signatureCalls += 1; }, + }; + const push = queue.push.bind(queue); + queue.push = (event: SessionEvent) => { + events.push(event); + push(event); + }; + + // The backend owns step accounting (count + per-step AssistantMessage flush + // + messageId rotation), so the adapter must not emit events or touch the + // text/thinking callbacks for step-boundary chunks. + const chunks: AiSdkStreamChunk[] = [ + { type: 'start-step' } as AiSdkStreamChunk, + { type: 'text-delta', text: 'one' }, + { type: 'finish-step', finishReason: { unified: 'tool-calls', raw: 'tool_calls' } } as AiSdkStreamChunk, + { type: 'start-step' } as AiSdkStreamChunk, + { type: 'text-delta', text: 'two' }, + { type: 'finish-step', finishReason: { unified: 'stop', raw: 'stop' } } as AiSdkStreamChunk, + ]; + for (const chunk of chunks) { + adapter.handleStreamChunk(chunk, 'turn-1', 'assistant-1', queue, callbacks); + } + + // Only the two text deltas produce events / callbacks; boundaries are inert. + assert.deepEqual(events.map((event) => event.type), ['text_delta', 'text_delta']); + assert.equal(callbacks.textCalls, 2); + assert.equal(callbacks.thinkingCalls, 0); + assert.equal(callbacks.signatureCalls, 0); + }); + test('captures the Anthropic reasoning signature without emitting an empty thinking delta', () => { const events: SessionEvent[] = []; const queue = new AsyncEventQueue(); diff --git a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts index acd8e637d2..c40f7d3588 100644 --- a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts +++ b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts @@ -398,7 +398,9 @@ describe('projectRuntimeEventsToStoredMessages', () => { expect(out.diagnostics.map((diag) => diag.code)).toEqual(['partial_skipped']); }); - test('model thinking attaches to same-turn assistant text without breaking compatibility', () => { + test('model thinking attaches to the assistant text row that shares its step message id', () => { + // Real emission and backfill give a step's thinking and text the same message + // id (providerEventId / storedMessageId), so the projection pairs by id. const out = projectRuntimeEventsToStoredMessages([ ev({ id: 'evt-thinking', @@ -406,6 +408,7 @@ describe('projectRuntimeEventsToStoredMessages', () => { role: 'model', author: 'agent', content: { kind: 'thinking', text: 'private reasoning', signature: 'sig-1' }, + refs: { storedMessageId: 'legacy-assistant' }, }), ev({ id: 'evt-assistant', @@ -431,6 +434,70 @@ describe('projectRuntimeEventsToStoredMessages', () => { expect(compareRuntimeReadModelMessages(out.messages, legacy).compatible).toBe(true); }); + test('per-step thinking pairs each step assistant row by its own message id', () => { + // Two steps in one turn, each with its own signed thinking. The ledger order + // per step is thinking → text (finish-step flush), and each step's thinking + // carries its step message id, so it must attach to its own assistant row — + // not the last row of the turn. + const out = projectRuntimeEventsToStoredMessages([ + ev({ + id: 'evt-think-1', + ts: ts + 1, + role: 'model', + author: 'agent', + content: { kind: 'thinking', text: 'reasoning one', signature: 'sig-1' }, + refs: { providerEventId: 'step-1' }, + }), + ev({ + id: 'evt-text-1', + ts: ts + 2, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'answer one' }, + refs: { providerEventId: 'step-1' }, + }), + ev({ + id: 'evt-think-2', + ts: ts + 3, + role: 'model', + author: 'agent', + content: { kind: 'thinking', text: 'reasoning two', signature: 'sig-2' }, + refs: { providerEventId: 'step-2' }, + }), + ev({ + id: 'evt-text-2', + ts: ts + 4, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'answer two' }, + refs: { providerEventId: 'step-2' }, + }), + ], { runHeaders: [header] }); + + const assistants = out.messages.filter((message) => message.type === 'assistant'); + expect(assistants).toEqual([ + { + type: 'assistant', + id: 'step-1', + turnId, + ts: ts + 2, + text: 'answer one', + modelId: 'claude-sonnet-4-5', + thinking: { text: 'reasoning one', signature: 'sig-1' }, + }, + { + type: 'assistant', + id: 'step-2', + turnId, + ts: ts + 4, + text: 'answer two', + modelId: 'claude-sonnet-4-5', + thinking: { text: 'reasoning two', signature: 'sig-2' }, + }, + ]); + expect(out.diagnostics).toEqual([]); + }); + test('unsupported and incomplete events are diagnostic-only', () => { const out = projectRuntimeEventsToStoredMessages([ ev({ diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 4f5431f715..ed3f961269 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -537,6 +537,12 @@ export class AiSdkBackend implements AgentBackend { private currentWatchdog: StreamWatchdog | null = null; private currentRunTrace: RunTrace | null = null; private priorRequestShape: RequestShapeDiagnostic | undefined; + /** + * Id of the assistant step currently streaming. Read by ToolRuntime via + * `getCurrentStepId` so each tool call's `tool_start` carries the step it + * belongs to. Rotated at every step boundary in `send()`; null between turns. + */ + private currentStepMessageId: string | null = null; constructor(input: AiSdkBackendInput) { this.input = input; @@ -570,6 +576,7 @@ export class AiSdkBackend implements AgentBackend { now: this.now, getPermissionPauseTarget: () => this.currentWatchdog, getCurrentRunId: () => this.currentRunId ?? undefined, + getCurrentStepId: () => this.currentStepMessageId ?? undefined, spawnChildAgent: input.spawnChildAgent, listChildAgents: input.listChildAgents, readChildAgentOutput: input.readChildAgentOutput, @@ -677,11 +684,74 @@ export class AiSdkBackend implements AgentBackend { const queue = new AsyncEventQueue(); this.currentQueue = queue; - const assistantMessageId = this.newId(); - let assistantText = ''; - let thinkingText = ''; - let thinkingSignature: string | undefined; + // One AssistantMessage is flushed per AI SDK step (not per turn), so the + // ledger records the text↔tool timeline at step granularity and each step's + // Anthropic thinking signature stays paired with its own thinking text. The + // turn's first step reuses this id; every later step rotates to a fresh one + // at its step boundary (see the fullStream loop below). + this.currentStepMessageId = this.newId(); + let stepText = ''; + let stepThinking = ''; + let stepSignature: string | undefined; + // Whether any step flushed non-empty text this turn — drives the step-cap + // grace notice below (a turn whose every step was tool-only gets the notice). + let turnHadAnyText = false; const startedAt = this.now(); + + // Flush the current step's AssistantMessage (text + thinking) and the paired + // terminal thinking/text events, then clear the per-step accumulators. + // Persist when the step produced text OR reasoning — a thinking-only step + // (Anthropic's signed/omitted reasoning has empty text) still round-trips its + // signed block; a pure-tool step (no text, no thinking) writes nothing, so + // tool-only steps leave no placeholder assistant row. thinking_complete + // precedes text_complete so the read-model attaches this step's reasoning to + // this step's assistant row. Hoisted to send() scope so both the streaming + // path and the abort/error handler can flush a partial step. + const flushStep = async (): Promise => { + const hasThinking = stepThinking.length > 0 || stepSignature !== undefined; + if (stepText.length === 0 && !hasThinking) return; + const stepId = this.currentStepMessageId ?? this.newId(); + const msg: AssistantMessage = { + type: 'assistant', + id: stepId, + turnId, + ts: this.now(), + text: stepText, + modelId: this.input.modelId, + ...(hasThinking + ? { + thinking: { + text: stepThinking, + ...(stepSignature !== undefined ? { signature: stepSignature } : {}), + }, + } + : {}), + }; + await this.input.appendMessage(msg); + if (hasThinking) { + queue.push({ + type: 'thinking_complete', + id: this.newId(), + turnId, + ts: this.now(), + messageId: stepId, + text: stepThinking, + ...(stepSignature !== undefined ? { signature: stepSignature } : {}), + } satisfies ThinkingCompleteEvent); + } + queue.push({ + type: 'text_complete', + id: this.newId(), + turnId, + ts: this.now(), + messageId: stepId, + text: stepText, + } satisfies TextCompleteEvent); + if (stepText.length > 0) turnHadAnyText = true; + stepText = ''; + stepThinking = ''; + stepSignature = undefined; + }; let tokenUsage: NormalizedAiSdkUsage | undefined; let tokenUsageCostUsd: number | undefined; let streamStatus: LlmCallRecord['status'] = 'success'; @@ -936,18 +1006,35 @@ export class AiSdkBackend implements AgentBackend { for await (const chunk of result.fullStream) { if (this.aborted) break; watchdog.markActivity(); - if (chunk.type === 'step-finish') { + // Step boundary, version-tolerant: AI SDK v6 delimits steps with + // `start-step` / `finish-step`, older releases said `step-finish`. + // Missing the boundary would silently degrade back to one message per + // turn, so match both names. A duplicate boundary is harmless: the + // second flush no-ops (accumulators already cleared) and one extra id + // rotation just discards an unused id. + const isStepFinishChunk = chunk.type === 'finish-step' || chunk.type === 'step-finish'; + if (isStepFinishChunk) { runtimeSteps += 1; } - if (chunk.type === 'finish' || chunk.type === 'step-finish') { + if (chunk.type === 'finish' || isStepFinishChunk) { rawFinishReason = rawFinishReasonString(chunk.finishReason) ?? rawFinishReason; } - this.modelAdapter.handleStreamChunk(chunk, turnId, assistantMessageId, queue, { - onText: (t) => { assistantText += t; }, - onTextComplete: (t) => { assistantText = t; }, - onThinking: (t) => { thinkingText += t; }, - onThinkingSignature: (sig) => { thinkingSignature = sig; }, + this.modelAdapter.handleStreamChunk(chunk, turnId, this.currentStepMessageId!, queue, { + onText: (t) => { stepText += t; }, + onTextComplete: (t) => { stepText = t; }, + onThinking: (t) => { stepThinking += t; }, + onThinkingSignature: (sig) => { stepSignature = sig; }, }); + // The step's text/thinking deltas are all in (the fullStream is + // drained in order), so flush this step's AssistantMessage and rotate + // to a fresh id for the next step. The step's tool calls (appended + // mid-step via execute()) already carry the pre-rotation id via + // `getCurrentStepId`, so replay can regroup them with this step's + // reasoning even though they land before this row in the ledger. + if (isStepFinishChunk) { + await flushStep(); + this.currentStepMessageId = this.newId(); + } } // If the stream loop exited because stop() flipped this.aborted while a @@ -958,6 +1045,10 @@ export class AiSdkBackend implements AgentBackend { throw Object.assign(new Error('aborted'), { name: 'AbortError' }); } + // Catch-all: flush any residual step content if the provider closed the + // stream without a trailing `finish-step` for the last step. + await flushStep(); + // Same-turn deferred load: prepareStep expanded the provider tool set on // later steps, so refine the durable cost record + prefix baseline against // the final active set — otherwise this turn under-reports the loaded @@ -982,74 +1073,42 @@ export class AiSdkBackend implements AgentBackend { if (finishReasonForGrace === 'tool-calls' && runtimeSteps < this.maxSteps) { runtimeSteps = this.maxSteps; } + // Step-cap grace notice: when the loop tripped `stepCountIs(maxSteps)` + // mid-tool-loop and no step ever produced closing text, append a final + // assistant message (its own step id) so the UI has a closing line and + // the user can send "继续" for a fresh turn. if ( !this.aborted - && assistantText.length === 0 + && !turnHadAnyText && finishReasonForGrace === 'tool-calls' ) { - assistantText = + // Always a fresh id. When the stream closed without a trailing + // finish-step, `currentStepMessageId` is already taken: the catch-all + // flush just used it for a thinking-only last step's AssistantMessage + // (reuse would duplicate a ledger id), and a pure-tool last step's + // tool_starts carry it as stepId (replay would adopt the grace text + // as that step's closer). A rotated-but-unused id is discardable. + const graceId = this.newId(); + const graceText = `⚠️ 已达到本轮 ${this.maxSteps} 步工具调用上限。\n\n` + '上一步工具调用已落盘；如果还需要继续，请发一条新消息让对话进入下一回合（可以直接输入「继续」）。'; - } - - // Persist the assistant turn if it produced text OR reasoning. A turn - // that ends with only thinking (no final text) must still persist its - // reasoning; gating solely on text would drop it. The AssistantMessage - // carries an empty `text` in that case, and we still emit text_complete - // (empty) so the RuntimeEvent read-model has a same-turn assistant row - // to attach the thinking to — otherwise projectThinking has nothing to - // hang the reasoning on and discards it. - // - // A signature is thinking even with no text: Anthropic's omitted / - // redacted thinking returns a signed reasoning block whose text is - // empty. Dropping it would lose the signed block for provider-native - // replay (reasoning continuity). Persist `text: ''` + signature so the - // block round-trips faithfully; the render layer is responsible for not - // surfacing an empty thinking entry. - const hasThinking = thinkingText.length > 0 || thinkingSignature !== undefined; - if (assistantText.length > 0 || hasThinking) { - const msg: AssistantMessage = { + await this.input.appendMessage({ type: 'assistant', - id: assistantMessageId, + id: graceId, turnId, ts: this.now(), - text: assistantText, + text: graceText, modelId: this.input.modelId, - ...(hasThinking - ? { - thinking: { - text: thinkingText, - ...(thinkingSignature !== undefined ? { signature: thinkingSignature } : {}), - }, - } - : {}), - }; - await this.input.appendMessage(msg); - // Emit the terminal thinking event before text_complete so the - // RuntimeEvent stream carries a non-partial `thinking` message. Only - // partial `thinking_delta`s were emitted during streaming, so without - // this the read-model projection (and materialized session) drops all - // reasoning. The read-model attaches this to the same-turn assistant - // text row emitted just below. - if (hasThinking) { - queue.push({ - type: 'thinking_complete', - id: this.newId(), - turnId, - ts: this.now(), - messageId: assistantMessageId, - text: thinkingText, - ...(thinkingSignature !== undefined ? { signature: thinkingSignature } : {}), - } satisfies ThinkingCompleteEvent); - } + }); queue.push({ type: 'text_complete', id: this.newId(), turnId, ts: this.now(), - messageId: assistantMessageId, - text: assistantText, + messageId: graceId, + text: graceText, } satisfies TextCompleteEvent); + turnHadAnyText = true; } // Final usage event. AI SDK `usage` is the last step only; `totalUsage` @@ -1161,6 +1220,12 @@ export class AiSdkBackend implements AgentBackend { } catch (err) { streamStatus = this.aborted ? 'aborted' : 'error'; streamErrorClass = this.modelAdapter.classifyError(watchdogTimeoutError ?? err); + // Flush the in-flight step's partial text/thinking before the terminal + // abort/error events. Earlier steps already flushed at their + // `finish-step`; this keeps their and this step's streamed-out output on + // BOTH exits — user stop and provider error / watchdog timeout — so + // partialOutputRetained reflects what the user actually saw. + await flushStep().catch(() => {}); if (this.aborted) { queue.push({ type: 'abort', @@ -2168,27 +2233,48 @@ export class AiSdkBackend implements AgentBackend { return true; } + /** + * Materialize a replay plan into provider messages, grouping each assistant + * step's reasoning + text + tool calls into ONE assistant message (Anthropic + * requires the signed thinking block to lead the tool-use assistant message). + * + * The ledger lands a step's parts as: tool_call(s), tool_result(s), thinking, + * text (the per-step AssistantMessage flushes at `finish-step`, after the + * step's tool events). Model text carries the step id and closes the step: it + * emits `[reasoning, text, tool-call…]` then the tool results. Steps with no + * text closer — a thinking + tool step (its empty text closer is skipped from + * the plan as `empty_text_skipped`) or a pure-tool step — flush grouped by + * stepId, claiming any parked reasoning for that step. Legacy per-turn items + * (no step id) keep the older shape: tool calls form a tool-only assistant, + * text/thinking become standalone messages. + */ private async materializeRuntimeReplayPlan(plan: RuntimeEventModelReplayPlan): Promise { + type ToolCallItem = Extract; + type ToolResultItem = Extract; + type ThinkingItem = Extract; const out: ModelMessage[] = []; - let toolBlock: { - calls: Extract[]; - results: Map>; - pending: Set; - } | undefined; - const flushToolBlock = () => { - if (!toolBlock) return; - out.push({ - role: 'assistant', - content: toolBlock.calls.map((item) => ({ - type: 'tool-call', - toolCallId: item.toolCallId, - toolName: item.toolName, - input: item.input, - })), - }); - for (const call of toolBlock.calls) { - const result = toolBlock.results.get(call.toolCallId); + let bufferedCalls: ToolCallItem[] = []; + const results = new Map(); + const reasoningByStep = new Map(); + + const reasoningPart = (item: ThinkingItem) => ({ + type: 'reasoning' as const, + text: item.text, + providerOptions: { anthropic: { signature: item.signature } }, + }); + // Tool results are emitted only when their tool_call claims them here. A + // result whose call never appears in the plan (sliced-away call, corrupt + // ledger) is INTENTIONALLY dropped at the end: a standalone tool message + // with no preceding tool_use in an assistant message is an Anthropic 400. + // The old item-by-item materializer emitted such orphans; do not "fix" this + // back — the plan flags them as `unmatched_tool_result` (a non-blocking + // diagnostic precisely so this drop path is reachable; see + // hasBlockingReplayDiagnostics). + const pushToolResults = (calls: readonly ToolCallItem[]) => { + for (const call of calls) { + const result = results.get(call.toolCallId); if (!result) continue; + results.delete(call.toolCallId); out.push({ role: 'tool', content: [{ @@ -2199,26 +2285,95 @@ export class AiSdkBackend implements AgentBackend { }], }); } - toolBlock = undefined; + }; + // Emit one assistant message for a step: reasoning (if any), text (if any), + // then the step's tool calls, followed by those calls' tool results. + const emitStep = (reasoning: ThinkingItem | undefined, text: string, calls: readonly ToolCallItem[]) => { + const content: unknown[] = []; + if (reasoning) content.push(reasoningPart(reasoning)); + if (text.length > 0) content.push({ type: 'text', text }); + for (const call of calls) { + content.push({ type: 'tool-call', toolCallId: call.toolCallId, toolName: call.toolName, input: call.input }); + } + if (content.length > 0) out.push({ role: 'assistant', content } as ModelMessage); + pushToolResults(calls); + }; + // Emit tool calls no assistant text closed: a thinking + tool step with no + // text (its empty closer is skipped from the plan), a pure-tool step, or a + // legacy per-turn tool block. Group consecutive calls by stepId so each step + // stays one assistant message, and claim the step's parked reasoning by + // stepId — this is how the common Anthropic interleaved-thinking step shape + // (reasoning + tool call, no text) gets its reasoning merged ahead of its + // calls. Calls without a stepId group together (legacy shape, no reasoning). + const emitGroupedCalls = (calls: readonly ToolCallItem[]) => { + let group: ToolCallItem[] = []; + const emitGroup = () => { + if (group.length === 0) return; + const stepId = group[0]!.stepId; + const reasoning = stepId !== undefined ? reasoningByStep.get(stepId) : undefined; + if (stepId !== undefined) reasoningByStep.delete(stepId); + emitStep(reasoning, '', group); + group = []; + }; + for (const call of calls) { + if (group.length > 0 && group[0]!.stepId !== call.stepId) emitGroup(); + group.push(call); + } + emitGroup(); + }; + const flushLooseCalls = () => { + if (bufferedCalls.length === 0) return; + const calls = bufferedCalls; + bufferedCalls = []; + emitGroupedCalls(calls); }; for (const item of plan.items) { - if (item.kind === 'tool_call') { - toolBlock ??= { calls: [], results: new Map(), pending: new Set() }; - toolBlock.calls.push(item); - toolBlock.pending.add(item.toolCallId); - continue; - } - if (item.kind === 'tool_result' && toolBlock?.pending.has(item.toolCallId)) { - toolBlock.results.set(item.toolCallId, item); - toolBlock.pending.delete(item.toolCallId); - if (toolBlock.pending.size === 0) flushToolBlock(); - continue; + switch (item.kind) { + case 'tool_call': + bufferedCalls.push(item); + break; + case 'tool_result': + results.set(item.toolCallId, item); + break; + case 'thinking': + if (item.stepId !== undefined) { + reasoningByStep.set(item.stepId, item); + } else { + // Legacy standalone reasoning (pure-reasoning turn): emit on its own. + flushLooseCalls(); + out.push({ role: 'assistant', content: [reasoningPart(item)] } as ModelMessage); + } + break; + case 'text': + if (item.role !== 'assistant') { + flushLooseCalls(); + out.push(await this.materializeRuntimeReplayItem(item)); + break; + } + if (item.stepId !== undefined) { + const stepId = item.stepId; + const thisCalls = bufferedCalls.filter((call) => call.stepId === stepId); + const otherCalls = bufferedCalls.filter((call) => call.stepId !== stepId); + bufferedCalls = []; + // Earlier steps' unclosed calls flush first (with their own parked + // reasoning, if any) so step order is preserved. + if (otherCalls.length > 0) emitGroupedCalls(otherCalls); + emitStep(reasoningByStep.get(stepId), item.content, thisCalls); + reasoningByStep.delete(stepId); + } else { + // Legacy per-turn assistant text: standalone after any tool block. + flushLooseCalls(); + out.push({ role: 'assistant', content: item.content }); + } + break; } - flushToolBlock(); - out.push(await this.materializeRuntimeReplayItem(item)); } - flushToolBlock(); + flushLooseCalls(); + // Any reasoning whose closing text never arrived (defensive): emit standalone. + for (const reasoning of reasoningByStep.values()) { + out.push({ role: 'assistant', content: [reasoningPart(reasoning)] } as ModelMessage); + } return out; } @@ -2359,6 +2514,7 @@ export class AiSdkBackend implements AgentBackend { this.currentTurnId = null; this.currentRunId = null; this.currentRunTrace = null; + this.currentStepMessageId = null; this.toolRuntime.resetTurnState(); this.aborted = false; } @@ -2423,10 +2579,13 @@ function stableStringifyForSignature(value: unknown): string { } function hasBlockingReplayDiagnostics(plan: RuntimeEventModelReplayPlan): boolean { + // `unmatched_tool_result` is deliberately NOT blocking: the materializer + // drops an orphan tool result (its call sliced away or the ledger corrupt) + // on its own — see pushToolResults — so one orphan must not degrade the + // whole ledger to stored-message projection. return plan.diagnostics.some((diagnostic) => diagnostic.code === 'unsupported_role' || diagnostic.code === 'unsupported_content' || - diagnostic.code === 'unmatched_tool_result' || diagnostic.code === 'tool_id_mismatch' ); } diff --git a/packages/runtime/src/ai-sdk-flow.ts b/packages/runtime/src/ai-sdk-flow.ts index dacf13487b..b972fe7389 100644 --- a/packages/runtime/src/ai-sdk-flow.ts +++ b/packages/runtime/src/ai-sdk-flow.ts @@ -195,7 +195,10 @@ export function mapSessionEventToRuntimeEvent( name: event.toolName, args: event.args, }, - refs: { toolCallId: event.toolUseId }, + refs: { + toolCallId: event.toolUseId, + ...(event.stepId !== undefined ? { stepId: event.stepId } : {}), + }, }; if (event.displayName !== undefined || event.intent !== undefined) { const stateDelta: Record = {}; diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index 457da127c4..da68d80892 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -249,6 +249,13 @@ export class ModelAdapter { } case 'reasoning-start': break; + // Step boundaries (AI SDK v6 emits `start-step` / `finish-step`; older + // `step-finish` kept for compatibility) and the terminal `finish` carry no + // text/thinking to stream. The backend owns step accounting: it counts and + // flushes one AssistantMessage per step and rotates the messageId at each + // `finish-step`. Handling them here would double-count, so they are no-ops. + case 'start-step': + case 'finish-step': case 'step-finish': case 'finish': break; diff --git a/packages/runtime/src/model-history.ts b/packages/runtime/src/model-history.ts index d96427fb52..8971a39a43 100644 --- a/packages/runtime/src/model-history.ts +++ b/packages/runtime/src/model-history.ts @@ -76,6 +76,7 @@ export type RuntimeEventReplayDiagnosticCode = | 'unsupported_content' | 'system_runtime_fact_diagnostic_only' | 'terminal_fact_diagnostic_only' + | 'empty_text_skipped' | 'unsigned_thinking_skipped' | 'signed_thinking_in_tool_turn_skipped' | 'unmatched_tool_result' @@ -102,6 +103,8 @@ export type RuntimeEventModelReplayItem = content: string; /** Original attachments (if any) so replay can render image parts. */ attachments?: AttachmentRef[]; + /** Assistant step id (model-role text only); groups a step's parts. */ + stepId?: string; eventId: string; ts: number; } @@ -109,6 +112,8 @@ export type RuntimeEventModelReplayItem = kind: 'thinking'; text: string; signature?: string; + /** Assistant step id; pairs this reasoning with its step's tool calls. */ + stepId?: string; eventId: string; ts: number; } @@ -117,6 +122,8 @@ export type RuntimeEventModelReplayItem = toolCallId: string; toolName: string; input: unknown; + /** Assistant step id (from tool_start); groups the call with its step. */ + stepId?: string; eventId: string; ts: number; } @@ -271,20 +278,37 @@ export function buildRuntimeEventModelReplayPlan( const callsById = new Map(); const semanticKinds = new Set(); - // Turns that call tools cannot also replay their thinking provider-native. - // The backend accumulates a turn's reasoning into a single end-of-turn - // `thinking_complete`, emitted AFTER the turn's tool_call/tool_result events, - // so the thinking lands last in ledger order. Materialization can only render - // a thinking item as a standalone assistant reasoning message; placed after - // the tool result it (a) drops the leading thinking block Anthropic requires - // on the tool-use assistant message and (b) leaves an orphan thinking block — - // Anthropic rejects both (400). Pure-reasoning turns (no tools) are safe and - // still replay. Pre-scan so the decision is independent of event order, and - // union any full-ledger tool-turn ids the caller supplies — `events` may be a - // budget/search slice that kept a tool turn's thinking but dropped its tool - // events (see BuildRuntimeEventModelReplayPlanOptions.toolActivityTurnIds). - const turnsWithToolActivity = new Set(options.toolActivityTurnIds ?? []); - for (const id of collectToolActivityTurnIds(events)) turnsWithToolActivity.add(id); + // Signed thinking in a tool turn is only replayable when the turn's tool calls + // carry a step id (RuntimeEventRefs.stepId, stamped from tool_start): the + // materializer then merges the step's reasoning + tool calls into one provider + // assistant message. Legacy per-turn history has no step id — its single + // end-of-turn reasoning lands after the tool events and cannot be reattached + // to the tool-use assistant message (Anthropic 400), so it is still skipped. + // + // Classify each tool turn: paired (all its function_call events carry a + // stepId) vs unpaired (any lacks one). Union caller-supplied whole-ledger + // tool-turn ids that this (possibly sliced) `events` view cannot confirm as + // paired, so a sliced-away tool turn degrades safely to the legacy skip. + // + // The judgment is deliberately TURN-granular, not per-step: one turn is + // written by one backend build, so old (no stepId) and new (stepId) tool + // calls cannot mix within a turn — per-step classification would add + // complexity for a state that cannot exist. And pairedToolTurnIds is not + // dead state: it is what lets a caller-supplied tool-turn id (from the FULL + // ledger) stay replayable when this sliced view can prove the turn's calls + // are step-paired — without it every sliced tool turn would degrade. + const pairedToolTurnIds = new Set(); + const unpairedToolTurnIds = new Set(); + for (const event of events) { + if (isPartialRuntimeEvent(event)) continue; + if (event.content?.kind === 'function_call' && event.turnId) { + if (event.refs?.stepId) pairedToolTurnIds.add(event.turnId); + else unpairedToolTurnIds.add(event.turnId); + } + } + for (const id of options.toolActivityTurnIds ?? []) { + if (!pairedToolTurnIds.has(id)) unpairedToolTurnIds.add(id); + } for (const event of events) { if (isPartialRuntimeEvent(event)) { @@ -314,6 +338,22 @@ export function buildRuntimeEventModelReplayPlan( } if (!runtimeEventHasModelVisibleContent(event)) { + // A model-role empty text event is the step closer of a thinking-only / + // tool-only step (the backend emits text_complete with '' so the + // read-model gets an assistant row for the step's reasoning). It carries + // nothing to replay but is NOT unsupported history — flagging it + // unsupported_content would block provider-native replay of the whole + // ledger (hasBlockingReplayDiagnostics). Skip it benignly; the + // materializer pairs the step's parked reasoning with its tool calls by + // stepId at flush time, so no text closer is needed. + if (event.content.kind === 'text' && event.role === 'model') { + diagnostics.push(diagnostic( + event, + 'empty_text_skipped', + 'empty model text RuntimeEvent (thinking/tool-only step closer) skipped for model replay', + )); + continue; + } diagnostics.push(diagnostic( event, 'unsupported_content', @@ -347,6 +387,11 @@ export function buildRuntimeEventModelReplayPlan( role, content: formatTextWithAttachmentRefs(event.content), ...(event.content.attachments ? { attachments: event.content.attachments } : {}), + // Model text carries its step id (the message id) so the materializer + // can close a step and group its reasoning + tool calls. + ...(role === 'assistant' && event.refs?.providerEventId + ? { stepId: event.refs.providerEventId } + : {}), eventId: event.id, ts: event.ts, }); @@ -374,14 +419,17 @@ export function buildRuntimeEventModelReplayPlan( )); continue; } - if (event.turnId && turnsWithToolActivity.has(event.turnId)) { - // Signed, but its turn also calls tools — unreplayable in position - // (see turnsWithToolActivity above). Keep it in the read-model for the - // UI; skip it from replay items without downgrading the whole history. + if (event.turnId && unpairedToolTurnIds.has(event.turnId)) { + // Signed, but its turn has tool calls with no step id to pair against + // (legacy per-turn history) — the end-of-turn reasoning cannot be + // reattached to the tool-use assistant message. Keep it in the + // read-model for the UI; skip it from replay without downgrading the + // whole history. Per-step history (paired tool calls) is not skipped: + // the materializer merges each step's reasoning with its tool calls. diagnostics.push(diagnostic( event, 'signed_thinking_in_tool_turn_skipped', - 'signed thinking RuntimeEvent skipped for model replay: its turn also calls tools, and end-of-turn thinking cannot be reattached to the tool-use assistant message', + 'signed thinking RuntimeEvent skipped for model replay: its turn calls tools with no step id to pair the reasoning to a tool-use assistant message', )); continue; } @@ -390,6 +438,7 @@ export function buildRuntimeEventModelReplayPlan( kind: 'thinking', text: event.content.text, signature: event.content.signature, + ...(event.refs?.providerEventId ? { stepId: event.refs.providerEventId } : {}), eventId: event.id, ts: event.ts, }); @@ -409,6 +458,7 @@ export function buildRuntimeEventModelReplayPlan( toolCallId: event.content.id, toolName: event.content.name, input: event.content.args, + ...(event.refs?.stepId ? { stepId: event.refs.stepId } : {}), eventId: event.id, ts: event.ts, }); diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts index e01801ec98..b179401982 100644 --- a/packages/runtime/src/runtime-event-read-model.ts +++ b/packages/runtime/src/runtime-event-read-model.ts @@ -78,11 +78,18 @@ interface ProjectionState { toolName: string; hint?: string; }>; - thinkingByTurn: Map; + /** + * Thinking awaiting its assistant text row, keyed by the step message id + * (function of the event's providerEventId / storedMessageId — the same id the + * step's assistant row gets). Per-step turns have several entries per turn, so + * keying by message id (not turn) attaches each step's reasoning to its own row. + */ + thinkingByMessageId: Map; } interface PendingThinking { event: RuntimeEvent; + messageId: string; text: string; signature?: string; } @@ -96,7 +103,7 @@ export function projectRuntimeEventsToStoredMessages( diagnostics: [], toolNameByUseId: new Map(), permissionRequestById: new Map(), - thinkingByTurn: new Map(), + thinkingByMessageId: new Map(), }; const messages: StoredMessage[] = []; @@ -163,8 +170,8 @@ export function projectRuntimeEventsToStoredMessages( } } - for (const pending of state.thinkingByTurn.values()) { - diagnostic(state, pending.event, 'unsupported_event', 'thinking content has no same-turn assistant text row'); + for (const pending of state.thinkingByMessageId.values()) { + diagnostic(state, pending.event, 'unsupported_event', 'thinking content has no assistant text row with a matching message id'); } return { messages, diagnostics: state.diagnostics }; @@ -382,15 +389,16 @@ function projectText( diagnostic(state, event, 'incomplete_event', 'model text RuntimeEvent requires AgentRunHeader.modelId'); return false; } + const assistantId = stableMessageId(event, state, 'assistant'); messages.push({ type: 'assistant', - id: stableMessageId(event, state, 'assistant'), + id: assistantId, turnId: event.turnId, ts: event.ts, text: event.content.text, modelId: header.modelId, }); - attachPendingThinking(event, state, messages); + attachPendingThinking(event, state, messages, assistantId); return true; } @@ -420,13 +428,18 @@ function projectThinking( messages: StoredMessage[], ): boolean { if (event.content?.kind !== 'thinking') return false; + const messageId = thinkingMessageId(event); const pending: PendingThinking = { event, + messageId, text: event.content.text, ...(event.content.signature !== undefined ? { signature: event.content.signature } : {}), }; + // The step's assistant text row lands after its thinking in ledger order, so + // attach eagerly if it already exists (older ordering), else park by message id + // for projectText's attachPendingThinking to claim. if (attachThinkingToAssistant(event, pending, messages)) return true; - state.thinkingByTurn.set(thinkingKey(event), pending); + state.thinkingByMessageId.set(messageId, pending); return true; } @@ -650,12 +663,12 @@ function attachPendingThinking( event: RuntimeEvent, state: ProjectionState, messages: StoredMessage[], + assistantMessageId: string, ): void { - const key = thinkingKey(event); - const pending = state.thinkingByTurn.get(key); + const pending = state.thinkingByMessageId.get(assistantMessageId); if (!pending) return; if (attachThinkingToAssistant(event, pending, messages)) { - state.thinkingByTurn.delete(key); + state.thinkingByMessageId.delete(assistantMessageId); } } @@ -664,9 +677,12 @@ function attachThinkingToAssistant( pending: PendingThinking, messages: StoredMessage[], ): boolean { + // Attach to the assistant row whose id equals the thinking's step message id + // (per-step pairing). Scans from the tail so the newest matching row wins. for (let index = messages.length - 1; index >= 0; index -= 1) { const message = messages[index]!; if (message.type !== 'assistant' || message.turnId !== event.turnId) continue; + if (message.id !== pending.messageId) continue; message.thinking = { text: pending.text, ...(pending.signature !== undefined ? { signature: pending.signature } : {}), @@ -676,8 +692,10 @@ function attachThinkingToAssistant( return false; } -function thinkingKey(event: RuntimeEvent): string { - return `${event.runId}:${event.turnId}`; +function thinkingMessageId(event: RuntimeEvent): string { + return event.refs?.providerEventId + ?? event.refs?.storedMessageId + ?? event.id; } function abortSourceFromRuntime(event: RuntimeEvent, header: AgentRunHeader): string | undefined { diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index 8f2b67ff00..4539e18339 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -112,6 +112,13 @@ export interface ToolRuntimeInput { now: () => number; getPermissionPauseTarget: () => { pause(): void; resume(): void } | null; getCurrentRunId?: () => string | undefined; + /** + * Id of the assistant step currently streaming, stamped onto each tool call's + * `tool_start` event so model replay can group a step's reasoning + tool calls + * into one provider assistant message. Undefined leaves the step unpaired + * (legacy per-turn behavior). + */ + getCurrentStepId?: () => string | undefined; spawnChildAgent?: (input: { parentRunId: string; spec: AgentSpec; @@ -248,6 +255,7 @@ export class ToolRuntime { args, }; await this.input.appendMessage(callMsg); + const stepId = this.input.getCurrentStepId?.(); const startEv: ToolStartEvent = { type: 'tool_start', id: this.input.newId(), @@ -258,6 +266,7 @@ export class ToolRuntime { args, ...(tool.displayName ? { displayName: tool.displayName } : {}), ...(toolIntent ? { intent: toolIntent } : {}), + ...(stepId !== undefined ? { stepId } : {}), }; queue.push(startEv); trace?.emit('tool', 'tool_started', 'Tool execution started', { diff --git a/packages/ui/src/materialize.ts b/packages/ui/src/materialize.ts index 05284cd64a..f07eef281a 100644 --- a/packages/ui/src/materialize.ts +++ b/packages/ui/src/materialize.ts @@ -277,18 +277,33 @@ export function materializeTurns( ...(message.attachments && message.attachments.length > 0 ? { attachments: message.attachments } : {}), }; } else if (message.type === 'assistant') { - turn.assistant = { id: message.id, role: 'assistant', text: message.text, ts: message.ts }; + // A turn now holds one AssistantMessage per model step. Concatenate their + // text (and thinking) in step order so the turn reads as one answer; keep + // the first step's id as the stable anchor, and advance ts to the latest + // step so durationMs measures to the turn's final assistant message. + const priorText = turn.assistant?.text ?? ''; + const mergedText = message.text.length > 0 + ? (priorText.length > 0 ? `${priorText}\n\n${message.text}` : message.text) + : priorText; + turn.assistant = { + id: turn.assistant?.id ?? message.id, + role: 'assistant', + text: mergedText, + ts: message.ts, + }; turn.modelId = message.modelId; if (message.thinking?.text) { - turn.assistantThinking = message.thinking.text; + turn.assistantThinking = turn.assistantThinking + ? `${turn.assistantThinking}\n\n${message.thinking.text}` + : message.thinking.text; } // Time-to-answer measured from the earliest message in this turn (usually - // the user's send) to the assistant message ts. Tool runs are inside - // this window, so the same metric captures both LLM latency and tool - // wall-time. We only compute this once the assistant message lands, so - // a streaming turn stays at undefined ("进行中" per kenji's PR82 - // review) instead of ticking up against the current clock and forcing - // visible re-renders. + // the user's send) to the turn's final assistant message ts. Tool runs are + // inside this window, so the same metric captures both LLM latency and tool + // wall-time. We only compute this once an assistant message lands, so a + // streaming turn stays at undefined ("进行中" per kenji's PR82 review) + // instead of ticking up against the current clock and forcing visible + // re-renders. Recomputed as each step lands, so it ends at the last step. if (message.ts !== undefined && message.ts >= turn.startedAt) { turn.durationMs = message.ts - turn.startedAt; }