diff --git a/docs/api-reference/veryfront/agent.md b/docs/api-reference/veryfront/agent.md index 83a3b10a1b..e39a96eb3e 100644 --- a/docs/api-reference/veryfront/agent.md +++ b/docs/api-reference/veryfront/agent.md @@ -1040,7 +1040,7 @@ Input delivered to a hosted agent-service detached execution callback. | Name | Description | Source | | ------------------------------------ | ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ | -| `AgentRuntime` | Implement agent runtime. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/index.ts#L840) | +| `AgentRuntime` | Implement agent runtime. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/index.ts#L844) | | `AgentRuntimeMessageConversionError` | Error shape for agent runtime message conversion. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/message-adapter.ts#L138) | | `AgentServiceAuthError` | Error shape for hosted service auth. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/auth.ts#L14) | | `AppendConversationRunEventsError` | Error shape for append conversation run events. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable-append-errors.ts#L4) | diff --git a/src/agent/runtime/chat-stream-handler.ts b/src/agent/runtime/chat-stream-handler.ts index 14e9ceedc2..d78d94e484 100644 --- a/src/agent/runtime/chat-stream-handler.ts +++ b/src/agent/runtime/chat-stream-handler.ts @@ -88,6 +88,51 @@ export interface StreamingToolResult { preliminary?: boolean; } +/** + * Flush a tool call's buffered `tool-input-start` and input deltas to the + * client. + * + * The start event is withheld until the call commits — `tool-input-end`, + * `tool-input-available` or `tool-call`. `tool-call` rebuilds the entry from + * its own `toolName` and announces from there, so a name that supersedes the + * one seen at `tool-input-start` is the one the client is given, and the + * superseded name never reaches the wire. + * + * Idempotent via `inputAnnounced`, which is also what gates the terminal + * `tool-output-error`. A call whose stream ended before any commit event was + * never announced, so it must be announced here before its failure can + * render. Such a call has no superseding name to wait for: the event that + * would carry one never arrived, and `inputAvailable` stays false, which is + * what makes that terminal path reachable at all. + */ +export function announceStreamedToolCallInput( + controller: ReadableStreamDefaultController, + encoder: TextEncoder, + toolCall: StreamingToolCall, +): void { + if (toolCall.inputAnnounced === true) { + return; + } + + const dynamic = toolCall.dynamic ?? isDynamicTool(toolCall.name); + sendSSE(controller, encoder, { + type: "tool-input-start", + toolCallId: toolCall.id, + toolName: toolCall.name, + ...(dynamic ? { dynamic: true } : {}), + }); + + for (const delta of toolCall.inputDeltas ?? []) { + sendSSE(controller, encoder, { + type: "tool-input-delta", + toolCallId: toolCall.id, + inputTextDelta: delta, + }); + } + + toolCall.inputAnnounced = true; +} + export interface StreamingReasoningPart { id: string; text: string; @@ -772,27 +817,7 @@ export function processStreamInternal( }; const announceToolInputStart = (toolCall: StreamingToolCall) => { - if (toolCall.inputAnnounced === true) { - return; - } - - const dynamic = toolCall.dynamic ?? isDynamicTool(toolCall.name); - sendSSE(controller, encoder, { - type: "tool-input-start", - toolCallId: toolCall.id, - toolName: toolCall.name, - ...(dynamic ? { dynamic: true } : {}), - }); - - for (const delta of toolCall.inputDeltas ?? []) { - sendSSE(controller, encoder, { - type: "tool-input-delta", - toolCallId: toolCall.id, - inputTextDelta: delta, - }); - } - - toolCall.inputAnnounced = true; + announceStreamedToolCallInput(controller, encoder, toolCall); }; const ensureToolLifecycle = (part: { diff --git a/src/agent/runtime/index.ts b/src/agent/runtime/index.ts index fc5c1e9b2e..fb2daf80b2 100644 --- a/src/agent/runtime/index.ts +++ b/src/agent/runtime/index.ts @@ -47,6 +47,7 @@ import { } from "./mcp-server-tool-sources.ts"; import { runWithRuntimeRemoteToolSources } from "./remote-tool-source-context.ts"; import { + announceStreamedToolCallInput, createStreamState, processStream, type StreamingToolCall, @@ -104,7 +105,10 @@ import { prepareAgentRuntimeStep, withIntegrationToolDiscoveryStatus, } from "./agent-runtime-step.ts"; -import { buildStreamedAssistantMessage } from "./streamed-assistant-message.ts"; +import { + buildStreamedAssistantMessage, + isPersistedReasoningPart, +} from "./streamed-assistant-message.ts"; import { type DeferredToolSummary, flattenSystemInstructions, @@ -2256,14 +2260,41 @@ export class AgentRuntime { const streamedToolCalls = Array.from(state.toolCalls.values()); const finalToolResults = collectFinalStreamToolResults(state); + // Recovery replays the whole step, so it also re-emits this step's + // reasoning — duplicating it in the live stream and in history, with a + // signature that no longer matches the replayed content. Reasoning that + // was persisted is reasoning the client already saw, so fail closed. + // This is a stopgap: reasoning is default-on across the hosted catalog, + // which makes recovery inert on most hosted paths. See #3736 for the + // reconciliation protocol that would let it run again. + const hasExposedReasoning = state.reasoningParts.some(isPersistedReasoningPart); const canRecoverInterruptedLocalToolBatch = !recoveredInterruptedLocalToolBatch && - step + 1 < maxSteps; + step + 1 < maxSteps && + !hasExposedReasoning; const shouldContinue = shouldContinueAfterStreamStep(state, { recoverInterruptedToolCalls: canRecoverInterruptedLocalToolBatch, }); const shouldRecoverInterruptedLocalToolBatch = canRecoverInterruptedLocalToolBatch && shouldContinue && streamedToolCalls.some(isInterruptedClientToolCall); + // Exactly `shouldRecoverInterruptedLocalToolBatch` with the reasoning + // gate lifted: the batch this step would have replayed had it not + // already exposed reasoning. Re-asking is what separates "recovery was + // declined" from "this step merely carried reasoning"; + // `shouldContinueAfterStreamStep` only reads state, so asking twice has + // no side effects, and the cheap conditions short-circuit ahead of it. + const declinedRecoveryForExposedReasoning = hasExposedReasoning && + !recoveredInterruptedLocalToolBatch && + step + 1 < maxSteps && + streamedToolCalls.some(isInterruptedClientToolCall) && + shouldContinueAfterStreamStep(state, { recoverInterruptedToolCalls: true }); + if (declinedRecoveryForExposedReasoning) { + logger.warn("Declined interrupted local tool batch recovery after exposed reasoning", { + step, + toolName: streamedToolCalls.find(isInterruptedClientToolCall)?.name, + reasoningPartCount: state.reasoningParts.filter(isPersistedReasoningPart).length, + }); + } const assistantMessage = buildStreamedAssistantMessage({ ...state, accumulatedText: recoveryPresentationText, @@ -2361,7 +2392,7 @@ export class AgentRuntime { const recordIncompleteLocalToolError = async ( toolCall: StreamingToolCall, - options: { includeInResponse?: boolean } = {}, + options: { includeInResponse?: boolean; announceInput?: boolean } = {}, ): Promise => { if ( toolCall.providerExecuted === true || @@ -2370,6 +2401,22 @@ export class AgentRuntime { ) { return false; } + if (options.announceInput === true) { + // An interrupted call never reached `tool-input-end`, so its + // `tool-input-start` is still buffered and `inputAnnounced` is false + // — which would suppress the `tool-output-error` below. On the + // declined-recovery path that leaves the client with a reasoning + // block and then nothing at all. + // + // The name is safe to publish here. `tool-call` is what can supersede + // a name, and it also sets `inputAvailable`, which fails the guard + // above — so reaching this line means no such event arrived and the + // buffered name is the only one this call will ever have. It is the + // same name recorded below and in the persisted assistant message, + // so the card matches a reload. Announcing is idempotent, so a call + // surfaced upstream is not reported twice. + announceStreamedToolCallInput(controller, encoder, toolCall); + } const incompleteToolCall: ToolCall = { id: toolCall.id, name: toolCall.name, @@ -2398,7 +2445,9 @@ export class AgentRuntime { await persistToolResult(toolResult); } for (const toolCall of streamedToolCalls) { - await recordIncompleteLocalToolError(toolCall); + await recordIncompleteLocalToolError(toolCall, { + announceInput: declinedRecoveryForExposedReasoning, + }); } sendSSE(controller, encoder, { type: "step-end" }); break; diff --git a/src/agent/runtime/refresh.test.ts b/src/agent/runtime/refresh.test.ts index 672b012909..6d7a8efd26 100644 --- a/src/agent/runtime/refresh.test.ts +++ b/src/agent/runtime/refresh.test.ts @@ -3030,6 +3030,427 @@ describe("agent runtime refresh hooks", () => { ); }); + it("does not retry a truncated non-placeholder tool call after exposing reasoning", async () => { + let finishedResponse: AgentResponse | undefined; + let callCount = 0; + const studioSuggestions = tool({ + id: "studio_suggestions", + description: "Capture Studio suggestions", + inputSchema: defineSchema((v) => v.object({}))(), + execute: async () => ({ suggestions: [] }), + }); + const model: ModelRuntime = { + provider: "anthropic", + modelId: "claude-sonnet-4-6", + async doGenerate() { + return { + content: [{ type: "text", text: "unused" }], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream() { + callCount++; + return { + stream: createRuntimeStream([ + { type: "reasoning-start", id: "reasoning-before-interruption" }, + { + type: "reasoning-delta", + id: "reasoning-before-interruption", + delta: "Check hidden state.", + }, + { type: "reasoning-end", id: "reasoning-before-interruption" }, + { + type: "tool-input-start", + id: "toolu_interrupted_after_reasoning", + toolName: "studio_suggestions", + }, + { + type: "tool-input-delta", + id: "toolu_interrupted_after_reasoning", + delta: '{"suggestions":[', + }, + { type: "finish", finishReason: "tool-calls" }, + ]), + }; + }, + }; + + const assistant = eagerAgent({ + model: "anthropic/claude-sonnet-4-6", + system: "Reasoning recovery replay regression test", + tools: { studio_suggestions: studioSuggestions }, + maxSteps: 3, + resolveModelTransport: async () => ({ model }), + }); + + const body = await (await assistant.stream({ + input: "Check before acting", + onFinish: (result) => { + finishedResponse = result; + }, + })).toDataStreamResponse().text(); + + assertEquals(callCount, 1); + assertEquals(body.match(/Check hidden state\./g)?.length ?? 0, 1); + assertExists(finishedResponse); + assertEquals( + finishedResponse.messages + .filter((message) => message.role === "assistant") + .flatMap((message) => message.parts) + .filter((part) => part.type === "reasoning") + .flatMap((part) => "text" in part && typeof part.text === "string" ? [part.text] : []), + ["Check hidden state."], + ); + + // Terminal state. Declining recovery ends the run here, so the truncated + // call has to reach the client as a failed tool card: its tool-input-start + // was buffered awaiting a commit that never came, and without flushing it + // the stream would stop after the reasoning block with nothing rendered. + assertEquals(body.match(/"type":"message-finish"/g)?.length ?? 0, 1); + assertEquals(body.includes('"finishReason":"tool-calls"'), true); + assertEquals(body.match(/"type":"tool-input-start"/g)?.length ?? 0, 1); + assertEquals(body.match(/"type":"tool-input-available"/g)?.length ?? 0, 0); + assertEquals(body.match(/"type":"tool-output-error"/g)?.length ?? 0, 1); + // Both terminal error events key off `inputAnnounced`, and only statement + // order keeps the incomplete-tool loop from firing before the announce. + // Exactly one failure event must reach the client, never two. + assertEquals(body.match(/"type":"tool-input-error"/g)?.length ?? 0, 0); + assertEquals( + body.includes( + 'Stream terminated before tool-call event fired for \\"studio_suggestions\\"', + ), + true, + ); + assertEquals( + finishedResponse.toolCalls.map((toolCall) => [toolCall.name, toolCall.status]), + [["studio_suggestions", "error"]], + ); + }); + + it("announces a truncated delegated remote tool under its namespaced name", async () => { + // The announce on the declined-recovery path is only reachable when + // `tool-call` never fired — that event sets `inputAvailable: true`, which + // makes `recordIncompleteLocalToolError` return at its guard. So no later + // event can supersede the name, and the name on the card must equal both + // the one the provider streamed and the one written to history. + let finishedResponse: AgentResponse | undefined; + let callCount = 0; + const gmailSource: RemoteToolSource = { + id: "gmail", + listTools: () => + Promise.resolve([{ + name: "gmail__list_emails", + description: "List Gmail messages", + parameters: { type: "object", properties: {} }, + }]), + executeTool: () => Promise.resolve({ messages: [] }), + }; + const model: ModelRuntime = { + provider: "anthropic", + modelId: "claude-sonnet-4-6", + async doGenerate() { + return { + content: [{ type: "text", text: "unused" }], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream() { + callCount++; + if (callCount > 1) { + return { + stream: createRuntimeStream([ + { type: "text-delta", text: "Unexpected recovery." }, + { type: "finish", finishReason: "stop" }, + ]), + }; + } + + return { + stream: createRuntimeStream([ + { type: "reasoning-start", id: "reasoning-before-remote" }, + { + type: "reasoning-delta", + id: "reasoning-before-remote", + delta: "Check the inbox first.", + }, + { type: "reasoning-end", id: "reasoning-before-remote" }, + { + type: "tool-input-start", + id: "toolu_remote_truncated", + toolName: "gmail__list_emails", + }, + { + type: "tool-input-delta", + id: "toolu_remote_truncated", + delta: '{"query":"is:unread', + }, + { type: "finish", finishReason: "tool-calls" }, + ]), + }; + }, + }; + + const assistant = eagerAgent( + { + model: "anthropic/claude-sonnet-4-6", + system: "Remote tool truncation regression test", + tools: { gmail__list_emails: true }, + __vfRemoteToolSources: [gmailSource], + __vfAllowedRemoteTools: ["gmail__list_emails"], + maxSteps: 3, + resolveModelTransport: async () => ({ model }), + } as AgentConfig & RuntimeRemoteToolConfig, + ); + + const body = await (await assistant.stream({ + input: "Summarize my inbox", + onFinish: (result) => { + finishedResponse = result; + }, + })).toDataStreamResponse().text(); + + assertEquals(callCount, 1); + assertEquals(body.includes("Unexpected recovery."), false); + assertExists(finishedResponse); + + // The card is announced under the exact namespaced name the provider sent. + // No bare `list_emails`, and no placeholder for an unresolved namespace. + assertEquals(body.match(/"type":"tool-input-start"/g)?.length ?? 0, 1); + assertEquals( + body.match( + /"type":"tool-input-start","toolCallId":"toolu_remote_truncated","toolName":"gmail__list_emails"/g, + ) + ?.length ?? 0, + 1, + ); + assertEquals(body.match(/"toolName":"list_emails"/g)?.length ?? 0, 0); + assertEquals(body.match(/"type":"tool-output-error"/g)?.length ?? 0, 1); + assertEquals(body.match(/"type":"tool-input-error"/g)?.length ?? 0, 0); + assertEquals(body.match(/"type":"tool-input-available"/g)?.length ?? 0, 0); + + // The name on the wire matches the name persisted to history, so a reload + // renders the same tool as the live stream did. + const persistedToolNames = finishedResponse.messages + .flatMap((message) => message.parts) + .flatMap((part) => + "toolCallId" in part && part.toolCallId === "toolu_remote_truncated" && + "toolName" in part && typeof part.toolName === "string" + ? [part.toolName] + : [] + ); + assertEquals(persistedToolNames.every((name) => name === "gmail__list_emails"), true); + assertEquals(persistedToolNames.length > 0, true); + assertEquals( + finishedResponse.toolCalls.map((toolCall) => [toolCall.name, toolCall.status]), + [["gmail__list_emails", "error"]], + ); + }); + + it("announces the finalized name when tool-call supersedes the buffered one", async () => { + // The buffered-name concern in the abstract: a `tool-call` carrying a + // different name than its `tool-input-start`. It is handled where the + // rename happens — `tool-call` announces the final name itself — and it + // also sets `inputAvailable: true`, so the terminal announce on the + // declined-recovery path can never see a superseded name. + let callCount = 0; + const renamed = tool({ + id: "summarize_inbox", + description: "Summarize the inbox", + inputSchema: defineSchema((v) => v.object({}))(), + execute: async () => ({ messages: [] }), + }); + const model: ModelRuntime = { + provider: "anthropic", + modelId: "claude-sonnet-4-6", + async doGenerate() { + return { + content: [{ type: "text", text: "unused" }], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream() { + callCount++; + if (callCount > 1) { + return { + stream: createRuntimeStream([ + { type: "text-delta", text: "Done." }, + { type: "finish", finishReason: "stop" }, + ]), + }; + } + + return { + stream: createRuntimeStream([ + { type: "reasoning-start", id: "reasoning-before-rename" }, + { + type: "reasoning-delta", + id: "reasoning-before-rename", + delta: "Check the inbox.", + }, + { type: "reasoning-end", id: "reasoning-before-rename" }, + { + type: "tool-input-start", + id: "toolu_renamed", + toolName: "list_emails", + }, + { type: "tool-input-delta", id: "toolu_renamed", delta: "{}" }, + { + type: "tool-call", + toolCallId: "toolu_renamed", + toolName: "summarize_inbox", + input: "{}", + }, + { type: "finish", finishReason: "tool-calls" }, + ]), + }; + }, + }; + + const assistant = eagerAgent({ + model: "anthropic/claude-sonnet-4-6", + system: "Tool rename regression test", + tools: { summarize_inbox: renamed }, + maxSteps: 3, + resolveModelTransport: async () => ({ model }), + }); + + const body = await (await assistant.stream({ input: "Summarize my inbox" })) + .toDataStreamResponse().text(); + + // Announced once, under the finalized name, by the `tool-call` branch. + assertEquals(body.match(/"type":"tool-input-start"/g)?.length ?? 0, 1); + assertEquals( + body.match( + /"type":"tool-input-start","toolCallId":"toolu_renamed","toolName":"summarize_inbox"/g, + ) + ?.length ?? 0, + 1, + ); + assertEquals(body.match(/"toolName":"list_emails"/g)?.length ?? 0, 0); + // Finalized, so it is not an incomplete call: the terminal announce path + // is not reached and no failure is fabricated for it. + assertEquals(body.match(/"type":"tool-output-error"/g)?.length ?? 0, 0); + assertEquals(body.match(/"type":"tool-input-error"/g)?.length ?? 0, 0); + }); + + it("preserves a bare placeholder tool call when reasoning blocks recovery", async () => { + let finishedResponse: AgentResponse | undefined; + let callCount = 0; + const studioSuggestions = tool({ + id: "studio_suggestions", + description: "Capture Studio suggestions", + inputSchema: defineSchema((v) => v.object({}))(), + execute: async () => ({ suggestions: [] }), + }); + const model: ModelRuntime = { + provider: "anthropic", + modelId: "claude-sonnet-4-6", + async doGenerate() { + return { + content: [{ type: "text", text: "unused" }], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream() { + callCount++; + if (callCount > 1) { + return { + stream: createRuntimeStream([ + { type: "text-delta", text: "Unexpected recovery." }, + { type: "finish", finishReason: "stop" }, + ]), + }; + } + + return { + stream: createRuntimeStream([ + { type: "reasoning-start", id: "reasoning-before-placeholder" }, + { + type: "reasoning-delta", + id: "reasoning-before-placeholder", + delta: "Plan the suggestions.", + }, + { type: "reasoning-end", id: "reasoning-before-placeholder" }, + { type: "text-delta", text: "Created the Outlook assistant." }, + { + type: "tool-input-start", + id: "toolu_placeholder_after_reasoning", + toolName: "studio_suggestions", + }, + { type: "tool-input-delta", id: "toolu_placeholder_after_reasoning", delta: "{}" }, + { + type: "finish", + finishReason: "tool-calls", + usage: { inputTokens: 1, outputTokens: 1 }, + }, + ]), + }; + }, + }; + + const assistant = eagerAgent({ + model: "anthropic/claude-sonnet-4-6", + system: "Reasoning recovery placeholder regression test", + tools: { studio_suggestions: studioSuggestions }, + maxSteps: 3, + resolveModelTransport: async () => ({ model }), + }); + + const body = await (await assistant.stream({ + input: "Create an Outlook assistant", + onFinish: (result) => { + finishedResponse = result; + }, + })).toDataStreamResponse().text(); + + assertEquals(callCount, 1); + assertEquals(body.includes("Unexpected recovery."), false); + assertExists(finishedResponse); + + // A bare `{}` placeholder is normally dropped from the assistant message + // when substantive text accompanies it. Terminalizing the step passes + // `preserveRecoverablePlaceholderToolCalls`, so the call and its error are + // kept in history and go back to the model on the next turn — the same + // outcome as maxSteps exhaustion. + const assistantParts = finishedResponse.messages + .filter((message) => message.role === "assistant") + .flatMap((message) => message.parts); + assertEquals( + assistantParts.filter((part) => part.type === "reasoning") + .flatMap((part) => "text" in part && typeof part.text === "string" ? [part.text] : []), + ["Plan the suggestions."], + ); + assertEquals( + assistantParts.flatMap((part) => part.type === "text" && "text" in part ? [part.text] : []), + ["Created the Outlook assistant."], + ); + assertExists( + assistantParts.find((part) => + "toolCallId" in part && part.toolCallId === "toolu_placeholder_after_reasoning" + ), + ); + assertEquals( + finishedResponse.toolCalls.map((toolCall) => [toolCall.name, toolCall.status]), + [["studio_suggestions", "error"]], + ); + assertEquals( + finishedResponse.messages + .flatMap((message) => message.parts) + .filter((part) => + part.type === "tool-result" && "toolCallId" in part && + part.toolCallId === "toolu_placeholder_after_reasoning" + ).length, + 1, + ); + assertEquals(body.match(/"type":"tool-input-start"/g)?.length ?? 0, 1); + assertEquals(body.match(/"type":"tool-output-error"/g)?.length ?? 0, 1); + assertEquals(body.match(/"type":"tool-input-error"/g)?.length ?? 0, 0); + }); + it("streams provider events before recovery replay text begins", async () => { let finishedResponse: AgentResponse | undefined; let callCount = 0; diff --git a/src/agent/runtime/streamed-assistant-message.test.ts b/src/agent/runtime/streamed-assistant-message.test.ts index f33d8ea652..cddbaf9da8 100644 --- a/src/agent/runtime/streamed-assistant-message.test.ts +++ b/src/agent/runtime/streamed-assistant-message.test.ts @@ -2,7 +2,10 @@ import "#veryfront/schemas/_test-setup.ts"; import { assertEquals } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; import type { ChatStreamState } from "./chat-stream-handler.ts"; -import { buildStreamedAssistantMessage } from "./streamed-assistant-message.ts"; +import { + buildStreamedAssistantMessage, + isPersistedReasoningPart, +} from "./streamed-assistant-message.ts"; describe("agent/streamed-assistant-message", () => { it("builds an assistant message from completed stream state", () => { @@ -72,6 +75,44 @@ describe("agent/streamed-assistant-message", () => { }); }); + it("treats an empty signature or redacted payload as absent", () => { + // `reasoning-end` assigns these on `typeof … === "string"`, so "" reaches + // the builder, and the SSE emission for the same part already drops it. + // `isPersistedReasoningPart` gates the interrupted-batch replay decision, + // so widening it here would both persist an empty reasoning part and make + // recovery fail closed on a step that exposed nothing. + const state: ChatStreamState = { + accumulatedText: "Final answer", + reasoningParts: [ + { id: "reasoning_blank_signature", text: "", signature: "" }, + { id: "reasoning_blank_redacted", text: "", redactedData: "" }, + { id: "reasoning_blank_both", text: "", signature: "", redactedData: "" }, + { id: "reasoning_kept", text: "kept", signature: "" }, + ], + finishReason: "stop", + toolCalls: new Map(), + suppressedToolCalls: [], + toolResults: [], + usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, + }; + + const message = buildStreamedAssistantMessage(state, { + id: "msg_blank", + timestamp: 7, + }); + + assertEquals(message.parts, [ + { type: "reasoning", text: "kept" }, + { type: "text", text: "Final answer" }, + ]); + assertEquals(isPersistedReasoningPart({ id: "a", text: "", signature: "" }), false); + assertEquals(isPersistedReasoningPart({ id: "b", text: "", redactedData: "" }), false); + assertEquals(isPersistedReasoningPart({ id: "c", text: "" }), false); + assertEquals(isPersistedReasoningPart({ id: "d", text: "", signature: "sig" }), true); + assertEquals(isPersistedReasoningPart({ id: "e", text: "", redactedData: "r" }), true); + assertEquals(isPersistedReasoningPart({ id: "f", text: " " }), true); + }); + it("omits recoverable placeholder tool parts when assistant text exists", () => { const state: ChatStreamState = { accumulatedText: "Created the Outlook assistant.", diff --git a/src/agent/runtime/streamed-assistant-message.ts b/src/agent/runtime/streamed-assistant-message.ts index 7963595e54..bd00967c22 100644 --- a/src/agent/runtime/streamed-assistant-message.ts +++ b/src/agent/runtime/streamed-assistant-message.ts @@ -1,5 +1,5 @@ import type { Message, MessagePart } from "../types.ts"; -import type { ChatStreamState } from "./chat-stream-handler.ts"; +import type { ChatStreamState, StreamingReasoningPart } from "./chat-stream-handler.ts"; import { materializeStreamedToolCall, shouldOmitRecoverablePlaceholderToolCall, @@ -10,6 +10,29 @@ export interface StreamedAssistantMessageIdentity { timestamp: number; } +/** + * Whether this reasoning part is kept in the assistant message, and so has + * already been exposed to the client and written to history. A part carrying + * no text, no signature and no redacted data is an id-only shell from a + * `reasoning-start` that never received deltas; it is dropped. + * + * Empty counts as absent for every field. `reasoning-end` accepts a signature + * or redacted payload on `typeof … === "string"`, so `""` is reachable, and + * the matching SSE emission already treats it as absent. Keeping the same rule + * here is what stops the persisted message from disagreeing with the wire. + * + * This is the single definition of "persisted reasoning". The runtime reuses + * it to decide whether an interrupted local tool batch may be replayed — a + * replay re-emits the step's reasoning, which duplicates whatever this + * predicate kept. Adding a field to `StreamingReasoningPart` therefore has to + * change this one function, not two copies of it. + */ +export function isPersistedReasoningPart(part: StreamingReasoningPart): boolean { + return part.text.length > 0 || + (part.signature?.length ?? 0) > 0 || + (part.redactedData?.length ?? 0) > 0; +} + export function buildStreamedAssistantMessage( state: Pick, identity: StreamedAssistantMessageIdentity, @@ -18,11 +41,7 @@ export function buildStreamedAssistantMessage( const parts: MessagePart[] = []; for (const reasoningPart of state.reasoningParts) { - if ( - reasoningPart.text.length === 0 && - !reasoningPart.signature && - !reasoningPart.redactedData - ) { + if (!isPersistedReasoningPart(reasoningPart)) { continue; } parts.push({