From ca34602167e6570486defaae901753696d9edd91 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 15:42:12 +0800 Subject: [PATCH 01/12] fix(runtime): flush one AssistantMessage per AI SDK step Emit a per-step AssistantMessage (text + thinking + signature) at each finish-step boundary instead of one accumulated message at turn end, so the ledger records the text-tool timeline at step granularity and each step's Anthropic thinking signature stays paired with its own reasoning text. Each step rotates a fresh assistant messageId; tool calls stamp their step id onto tool_start (new ToolStartEvent.stepId -> RuntimeEventRefs.stepId) so model replay can regroup a step's reasoning with its tool calls. Pure-tool steps write no placeholder; the step-cap grace notice and abort partials flush per step. Also count steps on the v6 finish-step chunk (the old step-finish check was dead) and document start-step/finish-step as no-ops in the adapter. --- packages/core/src/events.ts | 8 ++ packages/core/src/runtime-event.ts | 8 ++ packages/runtime/src/ai-sdk-backend.ts | 179 ++++++++++++++++--------- packages/runtime/src/ai-sdk-flow.ts | 5 +- packages/runtime/src/model-adapter.ts | 7 + packages/runtime/src/tool-runtime.ts | 9 ++ 6 files changed, 152 insertions(+), 64 deletions(-) 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/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 4f5431f715..47cb390706 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,30 @@ export class AiSdkBackend implements AgentBackend { for await (const chunk of result.fullStream) { if (this.aborted) break; watchdog.markActivity(); - if (chunk.type === 'step-finish') { + // AI SDK v6 delimits steps with `start-step` / `finish-step`; the step + // count and finish reason ride the terminal `finish-step` / `finish`. + if (chunk.type === 'finish-step') { runtimeSteps += 1; } - if (chunk.type === 'finish' || chunk.type === 'step-finish') { + if (chunk.type === 'finish' || chunk.type === 'finish-step') { 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; }, }); + // Step boundary: 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 (chunk.type === 'finish-step') { + await flushStep(); + this.currentStepMessageId = this.newId(); + } } // If the stream loop exited because stop() flipped this.aborted while a @@ -958,6 +1040,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 +1068,36 @@ 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 = + const graceId = this.currentStepMessageId ?? 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` @@ -1162,6 +1210,10 @@ export class AiSdkBackend implements AgentBackend { streamStatus = this.aborted ? 'aborted' : 'error'; streamErrorClass = this.modelAdapter.classifyError(watchdogTimeoutError ?? err); if (this.aborted) { + // Flush the in-flight step's partial text/thinking before the abort + // signal. Earlier steps already flushed at their `finish-step`; this + // keeps their and this step's partial output (partialOutputRetained). + await flushStep().catch(() => {}); queue.push({ type: 'abort', id: this.newId(), @@ -2359,6 +2411,7 @@ export class AiSdkBackend implements AgentBackend { this.currentTurnId = null; this.currentRunId = null; this.currentRunTrace = null; + this.currentStepMessageId = null; this.toolRuntime.resetTurnState(); this.aborted = false; } 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/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', { From c645a73314834d55961a3c51883f37426c2e6b0b Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 15:53:27 +0800 Subject: [PATCH 02/12] fix(runtime): merge per-step reasoning + tool calls in model replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Narrow the tool-turn signed-thinking replay skip to legacy per-turn history (tool calls with no step id); per-step turns whose tool calls carry refs.stepId now regroup into one provider assistant message (reasoning leads, then text, then tool calls, then tool results) — the Anthropic-valid shape, so signed thinking from tool-calling turns replays instead of being dropped. Read-model projectThinking pairs each step's reasoning to its own assistant row by step message id instead of attaching to the last same-turn assistant row. Tests: reverse the tool-turn thinking case to the per-step (merged) shape, keep a legacy (unpaired) suppression regression, add a multi-step backend integration test and a two-assistant read-model pairing test. --- .../src/__tests__/ai-sdk-backend.test.ts | 146 +++++++++++++++++- .../runtime-event-read-model.test.ts | 69 ++++++++- packages/runtime/src/ai-sdk-backend.ts | 123 +++++++++++---- packages/runtime/src/model-history.ts | 63 +++++--- .../runtime/src/runtime-event-read-model.ts | 42 +++-- 5 files changed, 370 insertions(+), 73 deletions(-) diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 78da75be17..8ede617674 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -5691,14 +5691,59 @@ 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('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 +5879,91 @@ describe('AiSdkBackend thinking persistence', () => { assert.match(prompt, /"type":"reasoning"/); assert.match(prompt, /sig-omitted/); }); + + 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__/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 47cb390706..572ab43d97 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -2220,27 +2220,38 @@ 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. Tool calls of a + * preceding pure-tool step (no text closer) flush first as a tool-only + * assistant. 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 } }, + }); + 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: [{ @@ -2251,26 +2262,72 @@ 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); + }; + // Flush buffered tool calls that no assistant text closed (legacy per-turn + // tool block or a pure-tool step) as a tool-only assistant + their results. + const flushLooseCalls = () => { + if (bufferedCalls.length === 0) return; + const calls = bufferedCalls; + bufferedCalls = []; + emitStep(undefined, '', 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 = []; + if (otherCalls.length > 0) emitStep(undefined, '', 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; } diff --git a/packages/runtime/src/model-history.ts b/packages/runtime/src/model-history.ts index d96427fb52..d70f363d9d 100644 --- a/packages/runtime/src/model-history.ts +++ b/packages/runtime/src/model-history.ts @@ -102,6 +102,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 +111,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 +121,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 +277,29 @@ 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. + 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)) { @@ -347,6 +362,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 +394,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 +413,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 +433,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 { From 2a61b6f0c5a11049a300946026be6e6949772300 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 15:58:22 +0800 Subject: [PATCH 03/12] fix(ui,desktop): render multi-step assistant messages as one answer materialize concatenates a turn's per-step AssistantMessages (text and thinking) in step order, keeping the first step id as the anchor and advancing durationMs to the turn's final assistant message. The desktop Markdown export joins all step texts. The renderer text_delta slot resets on a messageId change so a new step's answer starts a fresh bubble instead of appending onto the draining prior step. --- .../src/renderer/app-shell-session-events.ts | 13 ++++++-- .../src/renderer/conversation-markdown.ts | 12 +++++-- packages/ui/src/materialize.ts | 31 ++++++++++++++----- 3 files changed, 43 insertions(+), 13 deletions(-) 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..f84207764a 100644 --- a/apps/desktop/src/renderer/conversation-markdown.ts +++ b/apps/desktop/src/renderer/conversation-markdown.ts @@ -43,7 +43,13 @@ 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 + .filter((m) => m.type === 'assistant') + .map((m) => (m as { text: string }).text) + .filter((text) => text.length > 0) + .join('\n\n'); const toolCalls = turnMessages.filter((m) => m.type === 'tool_call'); if (user) { @@ -67,13 +73,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/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; } From 4758fb54332e806930e6001d0813716c36b63941 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:01:24 +0800 Subject: [PATCH 04/12] test(runtime): cover AI SDK v6 step-boundary no-op handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Assert handleStreamChunk emits no events and leaves the text/thinking callbacks untouched for start-step / finish-step chunks — the backend owns step accounting. --- .../src/__tests__/model-adapter.test.ts | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) 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(); From 00d43c1233dad1d614621915d01940af14a9a3b7 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:02:57 +0800 Subject: [PATCH 05/12] test(desktop): lock multi-step assistant concatenation in materializeTurns --- .../main/__tests__/materialize-turns.test.ts | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) 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); From 1ac82a0033ea3ae97284c148c5cd6366a95ec17e Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:16:53 +0800 Subject: [PATCH 06/12] fix(runtime): thinking-only step closer no longer blocks provider-native replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The empty text_complete a thinking/tool-only step emits (so the read-model gets an assistant row for the step's reasoning) was classified unsupported_content in the replay plan — a blocking diagnostic that degraded the WHOLE ledger to stored-message projection. With interleaved thinking this is the most common step shape, so any such step poisoned replay for the rest of the session. Skip empty model text benignly (new empty_text_skipped diagnostic, non-blocking) and pair a step's parked reasoning with its tool calls by stepId at flush time in the materializer, so the no-text step still replays as one assistant message [reasoning, tool-call…] with no empty text block. Locked by a new replay test for the thinking + tool, no-text per-step shape. --- .../src/__tests__/ai-sdk-backend.test.ts | 54 +++++++++++++++++++ packages/runtime/src/ai-sdk-backend.ts | 41 +++++++++++--- packages/runtime/src/model-history.ts | 17 ++++++ 3 files changed, 104 insertions(+), 8 deletions(-) diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 8ede617674..8df10f8136 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -5740,6 +5740,60 @@ describe('AiSdkBackend thinking persistence', () => { 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('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 diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 572ab43d97..ee33529f77 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -2228,10 +2228,12 @@ export class AiSdkBackend implements AgentBackend { * 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. Tool calls of a - * preceding pure-tool step (no text closer) flush first as a tool-only - * assistant. Legacy per-turn items (no step id) keep the older shape: tool - * calls form a tool-only assistant, text/thinking become standalone messages. + * 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; @@ -2275,13 +2277,34 @@ export class AiSdkBackend implements AgentBackend { if (content.length > 0) out.push({ role: 'assistant', content } as ModelMessage); pushToolResults(calls); }; - // Flush buffered tool calls that no assistant text closed (legacy per-turn - // tool block or a pure-tool step) as a tool-only assistant + their results. + // 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 = []; - emitStep(undefined, '', calls); + emitGroupedCalls(calls); }; for (const item of plan.items) { @@ -2312,7 +2335,9 @@ export class AiSdkBackend implements AgentBackend { const thisCalls = bufferedCalls.filter((call) => call.stepId === stepId); const otherCalls = bufferedCalls.filter((call) => call.stepId !== stepId); bufferedCalls = []; - if (otherCalls.length > 0) emitStep(undefined, '', otherCalls); + // 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 { diff --git a/packages/runtime/src/model-history.ts b/packages/runtime/src/model-history.ts index d70f363d9d..2f2edc9234 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' @@ -329,6 +330,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', From 0dbdf53e245311cb1d9ef13ba99c58372317b2b3 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:17:09 +0800 Subject: [PATCH 07/12] docs(runtime): note intentional drop of unclaimed tool results in replay materializer An orphan tool_result (call sliced away or ledger corrupt) is dropped on purpose: a standalone tool message with no preceding tool_use is an Anthropic 400. The plan flags these as unmatched_tool_result; do not restore the old item-by-item orphan emission. --- packages/runtime/src/ai-sdk-backend.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index ee33529f77..a7a3c0b24a 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -2249,6 +2249,12 @@ export class AiSdkBackend implements AgentBackend { 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 already flags them as `unmatched_tool_result`. const pushToolResults = (calls: readonly ToolCallItem[]) => { for (const call of calls) { const result = results.get(call.toolCallId); From e82d1e2c5579c6a0513d6272c66808a124865aac Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:37:51 +0800 Subject: [PATCH 08/12] fix(runtime): flush the in-flight step's partial output on provider error The catch block only flushed the current step's streamed text/thinking on the abort (user_stop) exit; a provider failure or watchdog timeout dropped it. Hoist the flush ahead of both branches so partialOutputRetained reflects what the user actually saw on every terminal exit. Regression: mock stream emits a text delta then errors; the step's AssistantMessage persists with the streamed partial and the turn still closes as an error. --- .../src/__tests__/ai-sdk-backend.test.ts | 56 +++++++++++++++++++ packages/runtime/src/ai-sdk-backend.ts | 10 ++-- 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 8df10f8136..1861838b2a 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[] = []; diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index a7a3c0b24a..f4231fdadb 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -1209,11 +1209,13 @@ 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) { - // Flush the in-flight step's partial text/thinking before the abort - // signal. Earlier steps already flushed at their `finish-step`; this - // keeps their and this step's partial output (partialOutputRetained). - await flushStep().catch(() => {}); queue.push({ type: 'abort', id: this.newId(), From a15e882f365e0c55eaf19c7672c49ce7c8763ff3 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:40:09 +0800 Subject: [PATCH 09/12] fix(runtime): stop treating unmatched_tool_result as a replay-blocking diagnostic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The materializer already drops an orphan tool result on its own (a standalone tool message with no preceding tool_use is an Anthropic 400), but listing unmatched_tool_result in hasBlockingReplayDiagnostics meant one orphan degraded the ENTIRE ledger to stored-message projection — the drop path was unreachable. Unblock it: paired calls/results and signed reasoning replay provider-native, the orphan is dropped. Rewrote the test that locked the old fallback and added an end-to-end orphan + paired-history case. --- .../src/__tests__/ai-sdk-backend.test.ts | 65 ++++++++++++++++++- packages/runtime/src/ai-sdk-backend.ts | 9 ++- 2 files changed, 69 insertions(+), 5 deletions(-) diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 1861838b2a..72673c4f6e 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -2512,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', @@ -2547,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' }] }, ]); }); @@ -5850,6 +5854,61 @@ describe('AiSdkBackend thinking persistence', () => { 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 diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index f4231fdadb..543f475066 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -2256,7 +2256,9 @@ export class AiSdkBackend implements AgentBackend { // 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 already flags them as `unmatched_tool_result`. + // 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); @@ -2566,10 +2568,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' ); } From 2578703bb65ecaa3397ed3cb6e6d5171facfcbfa Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:40:48 +0800 Subject: [PATCH 10/12] fix(runtime): match both finish-step and step-finish at the step boundary If a dependency version emits the older step-finish name, matching only finish-step would silently degrade back to one AssistantMessage per turn. Duplicate boundaries are safe: the second flush no-ops on cleared accumulators and an extra id rotation only discards an unused id. --- packages/runtime/src/ai-sdk-backend.ts | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 543f475066..e37cf94c3b 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -1006,12 +1006,17 @@ export class AiSdkBackend implements AgentBackend { for await (const chunk of result.fullStream) { if (this.aborted) break; watchdog.markActivity(); - // AI SDK v6 delimits steps with `start-step` / `finish-step`; the step - // count and finish reason ride the terminal `finish-step` / `finish`. - if (chunk.type === 'finish-step') { + // 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 === 'finish-step') { + if (chunk.type === 'finish' || isStepFinishChunk) { rawFinishReason = rawFinishReasonString(chunk.finishReason) ?? rawFinishReason; } this.modelAdapter.handleStreamChunk(chunk, turnId, this.currentStepMessageId!, queue, { @@ -1020,13 +1025,13 @@ export class AiSdkBackend implements AgentBackend { onThinking: (t) => { stepThinking += t; }, onThinkingSignature: (sig) => { stepSignature = sig; }, }); - // Step boundary: 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 + // 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 (chunk.type === 'finish-step') { + if (isStepFinishChunk) { await flushStep(); this.currentStepMessageId = this.newId(); } From 5a6f1b85432ce2ebfe7b92cfe72cc80ddebd13a3 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 16:41:57 +0800 Subject: [PATCH 11/12] docs(runtime,desktop): clarify turn-granular pairing intent; narrow markdown cast model-history: document why paired/unpaired tool-turn classification is turn-granular (one turn = one backend build; formats cannot mix within a turn) and that pairedToolTurnIds exists for sliced-view safe degradation, not dead state. conversation-markdown: replace the bare text cast with discriminated union narrowing via flatMap; no behavior change. --- apps/desktop/src/renderer/conversation-markdown.ts | 4 +--- packages/runtime/src/model-history.ts | 8 ++++++++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/apps/desktop/src/renderer/conversation-markdown.ts b/apps/desktop/src/renderer/conversation-markdown.ts index f84207764a..6a26f7f3ff 100644 --- a/apps/desktop/src/renderer/conversation-markdown.ts +++ b/apps/desktop/src/renderer/conversation-markdown.ts @@ -46,9 +46,7 @@ export function renderConversationMarkdown(sessionName: string, messages: Stored // 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 - .filter((m) => m.type === 'assistant') - .map((m) => (m as { text: string }).text) - .filter((text) => text.length > 0) + .flatMap((m) => (m.type === 'assistant' && m.text.length > 0 ? [m.text] : [])) .join('\n\n'); const toolCalls = turnMessages.filter((m) => m.type === 'tool_call'); diff --git a/packages/runtime/src/model-history.ts b/packages/runtime/src/model-history.ts index 2f2edc9234..8971a39a43 100644 --- a/packages/runtime/src/model-history.ts +++ b/packages/runtime/src/model-history.ts @@ -289,6 +289,14 @@ export function buildRuntimeEventModelReplayPlan( // 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) { From 9d661ab17728c9db2efe9be886a009156c67f478 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Wed, 8 Jul 2026 17:14:31 +0800 Subject: [PATCH 12/12] fix(runtime): mint a fresh id for the step-cap grace notice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On the catch-all path (stream closed without a trailing finish-step) the current step id is already taken: the catch-all flush just used it for a thinking-only last step's AssistantMessage (reuse duplicated 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). Always newId() — a rotated-but-unused id is discardable by design. Regression drives the backend through a patched startStream (streamText always synthesizes trailing boundaries): real tool execute in step 1, thinking-only step 2, abrupt close; asserts the grace id collides with no assistant row and no tool_start.stepId, and the ledger has no duplicate ids. Verified red against the old code. --- .../src/__tests__/ai-sdk-backend.test.ts | 71 +++++++++++++++++++ packages/runtime/src/ai-sdk-backend.ts | 8 ++- 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 72673c4f6e..87de6336c8 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -6049,6 +6049,77 @@ describe('AiSdkBackend thinking persistence', () => { 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 diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index e37cf94c3b..ed3f961269 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -1082,7 +1082,13 @@ export class AiSdkBackend implements AgentBackend { && !turnHadAnyText && finishReasonForGrace === 'tool-calls' ) { - const graceId = this.currentStepMessageId ?? this.newId(); + // 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` + '上一步工具调用已落盘；如果还需要继续，请发一条新消息让对话进入下一回合（可以直接输入「继续」）。';