From b57668092f8ff9344c7c32f3a476e6b03f9f7d37 Mon Sep 17 00:00:00 2001 From: Koji Wakayama Date: Mon, 6 Apr 2026 22:23:28 +0200 Subject: [PATCH] fix(runtime): persist provider-executed streamed tool results --- deno.json | 2 +- src/agent/runtime/ai-stream-handler.test.ts | 43 ++++++++++++++++ src/agent/runtime/ai-stream-handler.ts | 46 +++++++++++++++++ src/agent/runtime/index.ts | 55 ++++++++++++++++++++- src/utils/version-constant.ts | 2 +- 5 files changed, 144 insertions(+), 4 deletions(-) diff --git a/deno.json b/deno.json index 71813010bd..8a93cea1e6 100644 --- a/deno.json +++ b/deno.json @@ -1,6 +1,6 @@ { "name": "veryfront", - "version": "0.1.138", + "version": "0.1.139", "license": "Apache-2.0", "nodeModulesDir": "auto", "exclude": [ diff --git a/src/agent/runtime/ai-stream-handler.test.ts b/src/agent/runtime/ai-stream-handler.test.ts index b5018511d3..582b51539a 100644 --- a/src/agent/runtime/ai-stream-handler.test.ts +++ b/src/agent/runtime/ai-stream-handler.test.ts @@ -46,6 +46,7 @@ describe("ai-stream-handler", () => { assertEquals(state.accumulatedText, ""); assertEquals(state.finishReason, null); assertEquals(state.toolCalls.size, 0); + assertEquals(state.toolResults.length, 0); assertEquals(state.usage, { promptTokens: 0, completionTokens: 0, totalTokens: 0 }); }); }); @@ -278,6 +279,11 @@ describe("ai-stream-handler", () => { await processStream(result, state, controller, encoder, "t", undefined); + assertEquals(state.toolResults, [{ + toolCallId: "tc-web", + toolName: "web_search", + output: { results: [{ title: "AI" }] }, + }]); assertEquals(events[0], { type: "tool-output-available", toolCallId: "tc-web", @@ -303,6 +309,11 @@ describe("ai-stream-handler", () => { await processStream(result, state, controller, encoder, "t", undefined); + assertEquals(state.toolResults, [{ + toolCallId: "tc-web", + toolName: "web_search", + error: { error: "Search failed" }, + }]); assertEquals(events[0], { type: "tool-output-error", toolCallId: "tc-web", @@ -328,6 +339,12 @@ describe("ai-stream-handler", () => { await processStream(result, state, controller, encoder, "t", undefined); + assertEquals(state.toolResults, [{ + toolCallId: "tc-provider-error", + toolName: "web_search", + error: "Expected object, received string", + providerExecuted: true, + }]); assertEquals(events[0], { type: "tool-output-error", toolCallId: "tc-provider-error", @@ -336,6 +353,32 @@ describe("ai-stream-handler", () => { }); }); + it("uses Error.message for streamed tool-error SSE events", async () => { + const { events, controller, encoder } = createSSECollector(); + const state = createStreamState(); + + const result = createMockResult([ + { + type: "tool-error", + toolCallId: "tc-provider-error-object", + toolName: "web_search", + input: { query: "Veryfront" }, + error: new Error("Provider timeout"), + providerExecuted: true, + }, + { type: "finish", finishReason: "error", totalUsage: null }, + ]); + + await processStream(result, state, controller, encoder, "t", undefined); + + assertEquals(events[0], { + type: "tool-output-error", + toolCallId: "tc-provider-error-object", + errorText: "Provider timeout", + providerExecuted: true, + }); + }); + it("ignores tool-input-delta for unknown tool call IDs", async () => { const { events, controller, encoder } = createSSECollector(); const state = createStreamState(); diff --git a/src/agent/runtime/ai-stream-handler.ts b/src/agent/runtime/ai-stream-handler.ts index 1a7156efeb..8c6257cf5e 100644 --- a/src/agent/runtime/ai-stream-handler.ts +++ b/src/agent/runtime/ai-stream-handler.ts @@ -24,10 +24,21 @@ export interface StreamingToolCall { dynamic?: boolean; } +export interface StreamingToolResult { + toolCallId: string; + toolName: string; + output?: unknown; + error?: unknown; + providerExecuted?: boolean; + dynamic?: boolean; + preliminary?: boolean; +} + export interface AIStreamState { accumulatedText: string; finishReason: string | null; toolCalls: Map; + toolResults: StreamingToolResult[]; usage: { promptTokens: number; completionTokens: number; totalTokens: number }; } @@ -93,6 +104,10 @@ function stringifyToolError(output: unknown): string { return output; } + if (output instanceof Error && typeof output.message === "string" && output.message.length > 0) { + return output.message; + } + try { return JSON.stringify(output); } catch { @@ -105,6 +120,7 @@ export function createStreamState(): AIStreamState { accumulatedText: "", finishReason: null, toolCalls: new Map(), + toolResults: [], usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, }; } @@ -213,6 +229,15 @@ export function processStream( case "tool-result": { const isError = "isError" in part && part.isError === true; if (isError) { + state.toolResults.push({ + toolCallId: part.toolCallId, + toolName: part.toolName, + error: "output" in part ? part.output : undefined, + ...("providerExecuted" in part && part.providerExecuted !== undefined + ? { providerExecuted: part.providerExecuted } + : {}), + ...("dynamic" in part && part.dynamic ? { dynamic: true } : {}), + }); sendSSE(controller, encoder, { type: "tool-output-error", toolCallId: part.toolCallId, @@ -225,6 +250,18 @@ export function processStream( break; } + state.toolResults.push({ + toolCallId: part.toolCallId, + toolName: part.toolName, + output: part.output, + ...("providerExecuted" in part && part.providerExecuted !== undefined + ? { providerExecuted: part.providerExecuted } + : {}), + ...("dynamic" in part && part.dynamic ? { dynamic: true } : {}), + ...("preliminary" in part && part.preliminary !== undefined + ? { preliminary: part.preliminary } + : {}), + }); sendSSE(controller, encoder, { type: "tool-output-available", toolCallId: part.toolCallId, @@ -241,6 +278,15 @@ export function processStream( } case "tool-error": { + state.toolResults.push({ + toolCallId: part.toolCallId, + toolName: part.toolName, + error: part.error, + ...("providerExecuted" in part && part.providerExecuted !== undefined + ? { providerExecuted: part.providerExecuted } + : {}), + ...("dynamic" in part && part.dynamic ? { dynamic: true } : {}), + }); sendSSE(controller, encoder, { type: "tool-output-error", toolCallId: part.toolCallId, diff --git a/src/agent/runtime/index.ts b/src/agent/runtime/index.ts index 4668274e42..4728ad500b 100644 --- a/src/agent/runtime/index.ts +++ b/src/agent/runtime/index.ts @@ -107,6 +107,22 @@ function isAbortError(error: unknown, abortSignal?: AbortSignal): boolean { return error instanceof DOMException && error.name === "AbortError"; } +function stringifyToolError(error: unknown): string { + if (typeof error === "string" && error.length > 0) { + return error; + } + + if (error instanceof Error && typeof error.message === "string" && error.message.length > 0) { + return error.message; + } + + try { + return JSON.stringify(error); + } catch { + return String(error); + } +} + function getSkillActivationRequiredError(toolName: string): string { return `Tool "${toolName}" cannot run before load-skill succeeds in the same step. ` + `Call "${LOAD_SKILL_TOOL_ID}" first to establish the active skill context.`; @@ -686,6 +702,7 @@ export class AgentRuntime { // Request-scoped skill policy (not class-level mutable state) let activeSkillPolicy: string[] | undefined; let finalFinishReason: string | undefined; + let latestAssistantText = ""; const allowedRemoteToolNames = getRuntimeAllowedRemoteTools(this.config); for (let step = 0; step < maxSteps; step++) { @@ -749,9 +766,32 @@ export class AgentRuntime { parts: streamParts, timestamp: Date.now(), }; + latestAssistantText = getTextFromParts(assistantMessage.parts); currentMessages.push(assistantMessage); await this.memory.add(assistantMessage); + for (const tr of state.toolResults) { + if (tr.preliminary) { + continue; + } + + const toolResultMessage: Message = { + id: `tool_${tr.toolCallId}`, + role: "tool", + parts: [ + { + type: "tool-result", + toolCallId: tr.toolCallId, + toolName: tr.toolName, + result: tr.error === undefined ? tr.output : { error: stringifyToolError(tr.error) }, + }, + ], + timestamp: Date.now(), + }; + currentMessages.push(toolResultMessage); + await this.memory.add(toolResultMessage); + } + if (state.finishReason !== "tool-calls" || !state.toolCalls.size) { sendSSE(controller, encoder, { type: "step-end" }); break; @@ -769,6 +809,18 @@ export class AgentRuntime { const toolCall: ToolCall = { id: tc.id, name: tc.name, args, status: "pending" }; if (tc.providerExecuted === true) { + const matchingResult = state.toolResults.find((result) => + result.toolCallId === tc.id && result.preliminary !== true + ); + + if (matchingResult) { + toolCall.status = matchingResult.error === undefined ? "completed" : "error"; + toolCall.result = matchingResult.output; + toolCall.error = matchingResult.error === undefined + ? undefined + : stringifyToolError(matchingResult.error); + toolCalls.push(toolCall); + } continue; } @@ -881,9 +933,8 @@ export class AgentRuntime { this.status = "thinking"; } - const lastMessage = currentMessages[currentMessages.length - 1]; return { - text: lastMessage ? getTextFromParts(lastMessage.parts) : "", + text: latestAssistantText, messages: currentMessages, toolCalls, status: "completed", diff --git a/src/utils/version-constant.ts b/src/utils/version-constant.ts index 15de36a46a..fdb0f1a53e 100644 --- a/src/utils/version-constant.ts +++ b/src/utils/version-constant.ts @@ -1,3 +1,3 @@ // Keep in sync with deno.json version. // scripts/release.ts updates this constant during releases. -export const VERSION = "0.1.138"; +export const VERSION = "0.1.139";