diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index e213e8deb20e..41f489f9a50c 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -22,7 +22,7 @@ extensions/amazon-bedrock-mantle/register.sync.runtime.ts 2 extensions/amazon-bedrock/config-compat.ts 6 extensions/amazon-bedrock/embedding-provider.ts 1 extensions/amazon-bedrock/register.sync.runtime.ts 19 -extensions/amazon-bedrock/stream.runtime.ts 14 +extensions/amazon-bedrock/stream.runtime.ts 13 extensions/anthropic-vertex/region.ts 1 extensions/anthropic-vertex/stream-runtime.ts 8 extensions/anthropic/cli-backend.ts 2 @@ -925,7 +925,7 @@ extensions/ollama/src/provider-models.ts 1 extensions/ollama/src/setup-pull.ts 1 extensions/ollama/src/setup.runtime.ts 1 extensions/ollama/src/stream-compat.ts 1 -extensions/ollama/src/stream.runtime.ts 10 +extensions/ollama/src/stream.runtime.ts 7 extensions/onepassword/index.ts 2 extensions/onepassword/src/op-client.ts 2 extensions/onepassword/src/tool.ts 1 @@ -1522,11 +1522,11 @@ packages/ai/src/providers/anthropic-server-fallback.ts 3 packages/ai/src/providers/anthropic-thinking-replay.ts 2 packages/ai/src/providers/anthropic-tool-projection.ts 2 packages/ai/src/providers/anthropic-usage.ts 3 -packages/ai/src/providers/anthropic.ts 13 +packages/ai/src/providers/anthropic.ts 12 packages/ai/src/providers/azure-openai-responses.ts 1 packages/ai/src/providers/clean-for-gemini.ts 15 packages/ai/src/providers/google-shared.ts 5 -packages/ai/src/providers/mistral.ts 7 +packages/ai/src/providers/mistral.ts 6 packages/ai/src/providers/openai-chatgpt-responses-protocol.ts 1 packages/ai/src/providers/openai-chatgpt-responses.ts 27 packages/ai/src/providers/openai-completions-tool-calls.ts 5 diff --git a/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts b/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts index ff58b5026a9a..cef472c1a9a0 100644 --- a/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts +++ b/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts @@ -76,6 +76,7 @@ describe("Bedrock provider-owned stream lifecycle", () => { delta: { toolUse: { input: '{"query":"ready"}' } }, }, }, + { contentBlockStop: { contentBlockIndex: 0 } }, ], endEvent: "toolcall_end", stopReason: BedrockStopReason.TOOL_USE, diff --git a/extensions/amazon-bedrock/stream.runtime.test.ts b/extensions/amazon-bedrock/stream.runtime.test.ts index 11e5c0c84455..ebf0bfd6cbcc 100644 --- a/extensions/amazon-bedrock/stream.runtime.test.ts +++ b/extensions/amazon-bedrock/stream.runtime.test.ts @@ -392,6 +392,123 @@ describe("Bedrock profile endpoint resolution", () => { }); describe("Bedrock stop reasons", () => { + it("rejects malformed terminal tool JSON before completing any sibling call", async () => { + vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ + $metadata: { httpStatusCode: 200 }, + stream: streamEvents([ + { messageStart: { role: ConversationRole.ASSISTANT } }, + { + contentBlockStart: { + contentBlockIndex: 0, + start: { toolUse: { toolUseId: "call_valid", name: "read" } }, + }, + }, + { + contentBlockDelta: { + contentBlockIndex: 0, + delta: { toolUse: { input: '{"path":"README.md"}' } }, + }, + }, + { contentBlockStop: { contentBlockIndex: 0 } }, + { + contentBlockStart: { + contentBlockIndex: 1, + start: { toolUse: { toolUseId: "call_invalid", name: "read" } }, + }, + }, + { + contentBlockDelta: { + contentBlockIndex: 1, + delta: { toolUse: { input: '{"path":"SECRET.md"' } }, + }, + }, + { contentBlockStop: { contentBlockIndex: 1 } }, + { messageStop: { stopReason: BedrockStopReason.TOOL_USE } }, + ]), + } as never); + const stream = streamBedrockForTest(bedrockModel({}), { + messages: [{ role: "user", content: "read", timestamp: 0 }], + } as never); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe("Provider completed tool call with malformed JSON arguments"); + expect(result.errorMessage).not.toContain("SECRET.md"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + }); + + it("rejects an active tool call that never receives contentBlockStop", async () => { + vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ + $metadata: { httpStatusCode: 200 }, + stream: streamEvents([ + { messageStart: { role: ConversationRole.ASSISTANT } }, + { + contentBlockStart: { + contentBlockIndex: 0, + start: { toolUse: { toolUseId: "call_unsealed", name: "read" } }, + }, + }, + { + contentBlockDelta: { + contentBlockIndex: 0, + delta: { toolUse: { input: '{"path":"README.md"' } }, + }, + }, + { messageStop: { stopReason: BedrockStopReason.TOOL_USE } }, + ]), + } as never); + const stream = streamBedrockForTest(bedrockModel({}), { + messages: [{ role: "user", content: "read", timestamp: 0 }], + } as never); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(eventTypes.at(-1)).toBe("error"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + expect(result.content.some((block) => block.type === "toolCall")).toBe(false); + }); + + it("uses a complete tool input seeded at block start", async () => { + vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ + $metadata: { httpStatusCode: 200 }, + stream: streamEvents([ + { messageStart: { role: ConversationRole.ASSISTANT } }, + { + contentBlockStart: { + contentBlockIndex: 0, + start: { + toolUse: { + toolUseId: "call_seeded", + name: "read", + input: { path: "README.md" }, + }, + }, + }, + }, + { contentBlockStop: { contentBlockIndex: 0 } }, + { messageStop: { stopReason: BedrockStopReason.TOOL_USE } }, + ]), + } as never); + + const result = await streamBedrockForTest(bedrockModel({}), { + messages: [{ role: "user", content: "read", timestamp: 0 }], + } as never).result(); + + expect(result.content).toContainEqual( + expect.objectContaining({ type: "toolCall", arguments: { path: "README.md" } }), + ); + }); + it.each([ { name: "text", @@ -400,6 +517,7 @@ describe("Bedrock stop reasons", () => { { contentBlockStop: { contentBlockIndex: 0 } }, ], contentType: "text", + retainsPartial: true, }, { name: "tool call", @@ -419,10 +537,11 @@ describe("Bedrock stop reasons", () => { { contentBlockStop: { contentBlockIndex: 0 } }, ], contentType: "toolCall", + retainsPartial: false, }, ])( "reports truncated $name streams without a terminal messageStop", - async ({ events, contentType }) => { + async ({ events, contentType, retainsPartial }) => { vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ $metadata: { httpStatusCode: 200 }, stream: streamEvents([{ messageStart: { role: ConversationRole.ASSISTANT } }, ...events]), @@ -441,9 +560,13 @@ describe("Bedrock stop reasons", () => { expect(eventTypes).not.toContain("done"); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("Bedrock stream ended before messageStop"); - expect(result.content).toEqual([expect.objectContaining({ type: contentType })]); - expect(result.content[0]).not.toHaveProperty("index"); - expect(result.content[0]).not.toHaveProperty("partialJson"); + expect(result.content).toEqual( + retainsPartial ? [expect.objectContaining({ type: contentType })] : [], + ); + if (retainsPartial) { + expect(result.content[0]).not.toHaveProperty("index"); + expect(result.content[0]).not.toHaveProperty("partialJson"); + } }, ); diff --git a/extensions/amazon-bedrock/stream.runtime.ts b/extensions/amazon-bedrock/stream.runtime.ts index 9a95c7ec513b..fa8cd325460e 100644 --- a/extensions/amazon-bedrock/stream.runtime.ts +++ b/extensions/amazon-bedrock/stream.runtime.ts @@ -70,13 +70,23 @@ import { createDeferredEventBuffer, notifyLlmRequestActivity, } from "openclaw/plugin-sdk/provider-stream-shared"; -import { describeToolResultMediaPlaceholder } from "openclaw/plugin-sdk/provider-transport-runtime"; -import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + describeToolResultMediaPlaceholder, + finalizeTerminalToolCallArguments, +} from "openclaw/plugin-sdk/provider-transport-runtime"; +import { isRecord, normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; import { supportsBedrockPromptCaching, type BedrockOptions } from "./bedrock-options.js"; import { supportsBedrockNativeMaxEffort } from "./thinking-policy.js"; -type Block = (TextContent | ThinkingContent | ToolCall) & { index?: number; partialJson?: string }; +type Block = (TextContent | ThinkingContent | ToolCall) & { + index?: number; + partialJson?: string; +}; type BedrockEventSink = { push(event: AssistantMessageEvent): void }; +type PendingBedrockToolCall = { + block: ToolCall & Pick; + contentIndex: number; +}; function usesClaudeFable5BedrockContract(model: Model<"bedrock-converse-stream">): boolean { return resolveClaudeFable5ModelIdentity(model) !== undefined; @@ -160,6 +170,7 @@ const streamBedrock: StreamFunction<"bedrock-converse-stream", BedrockOptions> = }; const blocks = output.content as Block[]; + const pendingToolCallEnds: PendingBedrockToolCall[] = []; const redactedReasoningChunks = new Map(); const fable5 = usesClaudeFable5BedrockContract(model); // Claude classifiers may refuse after partial output. Hold every event until @@ -315,6 +326,7 @@ const streamBedrock: StreamFunction<"bedrock-converse-stream", BedrockOptions> = output, eventSink, redactedReasoningChunks, + pendingToolCallEnds, ); } else if (item.messageStop) { sawMessageStop = true; @@ -359,20 +371,23 @@ const streamBedrock: StreamFunction<"bedrock-converse-stream", BedrockOptions> = // Some valid provider streams omit contentBlockStop; never persist their scratch state. for (const block of blocks) { - if (block.index !== undefined) { + if (block.index !== undefined && block.type !== "toolCall") { handleContentBlockStop( { contentBlockIndex: block.index }, blocks, output, eventSink, redactedReasoningChunks, + pendingToolCallEnds, ); } } + flushPendingBedrockToolCalls(pendingToolCallEnds, blocks, output, eventSink); refusalBuffer?.flush(); stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { + output.content = output.content.filter((block) => block.type !== "toolCall"); for (const block of output.content) { delete (block as Block).index; // partialJson is only a streaming scratch buffer; never persist it. @@ -523,11 +538,12 @@ function handleContentBlockStart( const start = event.start; if (start?.toolUse) { + const startArguments = isRecord(start.toolUse) ? start.toolUse.input : undefined; const block: Block = { type: "toolCall", id: start.toolUse.toolUseId || "", name: start.toolUse.name || "", - arguments: {}, + arguments: isRecord(startArguments) ? startArguments : {}, partialJson: "", index, }; @@ -653,19 +669,20 @@ function handleContentBlockStop( output: AssistantMessage, stream: BedrockEventSink, redactedReasoningChunks: Map, + pendingToolCallEnds: PendingBedrockToolCall[], ): void { const index = blocks.findIndex((b) => b.index === event.contentBlockIndex); const block = blocks[index]; if (!block) { return; } - delete block.index; - switch (block.type) { case "text": + delete block.index; stream.push({ type: "text_end", contentIndex: index, content: block.text, partial: output }); break; case "thinking": + delete block.index; if (block.redacted) { const chunks = redactedReasoningChunks.get(event.contentBlockIndex!); if (chunks) { @@ -688,14 +705,37 @@ function handleContentBlockStop( }); break; case "toolCall": - // Finalize in-place and strip the scratch buffer so replay only - // carries parsed arguments. - delete (block as Block).partialJson; - stream.push({ type: "toolcall_end", contentIndex: index, toolCall: block, partial: output }); + delete block.index; + pendingToolCallEnds.push({ block, contentIndex: index }); break; } } +function flushPendingBedrockToolCalls( + pending: PendingBedrockToolCall[], + blocks: Block[], + output: AssistantMessage, + stream: BedrockEventSink, +): void { + if (blocks.some((block) => block.type === "toolCall" && block.index !== undefined)) { + throw new Error("Provider completed stream with an incomplete tool call"); + } + finalizeTerminalToolCallArguments( + pending.map(({ block }) => block), + (block) => + block.partialJson && block.partialJson.length > 0 ? block.partialJson : block.arguments, + ); + for (const toolCall of pending) { + delete toolCall.block.partialJson; + stream.push({ + type: "toolcall_end", + contentIndex: toolCall.contentIndex, + toolCall: toolCall.block, + partial: output, + }); + } +} + function resolveClaudeProfileNameModelId(modelName?: string): string | undefined { const normalized = modelName diff --git a/extensions/ollama/src/stream-runtime.test.ts b/extensions/ollama/src/stream-runtime.test.ts index 27d70ab05fbc..32101c26a944 100644 --- a/extensions/ollama/src/stream-runtime.test.ts +++ b/extensions/ollama/src/stream-runtime.test.ts @@ -789,6 +789,50 @@ describe("convertToOllamaMessages", () => { expect(result).toEqual([{ role: "tool", content: "file contents here", tool_name: "read" }]); }); + it("preserves structured, image, error, and call identity in tool results", () => { + const result = convertToOllamaMessages([ + { + role: "toolResult", + toolCallId: "call_inspect", + toolName: "inspect", + isError: true, + content: [ + { type: "text", text: "inspection failed" }, + { type: "json", value: { retry: false } }, + { type: "image", mimeType: "image/png", data: "aW1hZ2U=" }, + { type: "audio", mimeType: "audio/wav", data: "YXVkaW8=" }, + ], + }, + ]); + + expect(result).toEqual([ + { + role: "tool", + content: + '[tool error] inspection failed\n{"type":"json","value":{"retry":false}}\n[unsupported tool-result audio omitted]', + images: ["aW1hZ2U="], + tool_call_id: "call_inspect", + tool_name: "inspect", + }, + ]); + expect(result[0]?.content).not.toContain("YXVkaW8="); + + expect( + convertToOllamaMessages([ + { + role: "toolResult", + toolCallId: "call_empty_error", + toolName: "inspect", + isError: true, + content: [], + }, + ])[0], + ).toMatchObject({ + content: "[tool error] (no tool output)", + tool_call_id: "call_empty_error", + }); + }); + it("omits tool_name when not provided in toolResult", () => { const messages = [{ role: "toolResult", content: "output" }]; const result = convertToOllamaMessages(messages); @@ -1092,15 +1136,13 @@ describe("buildAssistantMessage", () => { }); }); - it("falls back to empty arguments for malformed stringified tool call arguments", () => { + it("rejects malformed stringified tool call arguments", () => { const response = createToolCallResponse([ { function: { name: "bash", arguments: '{"command":"ls"' } }, ]); - const result = buildAssistantMessage(response, modelInfo); - expectToolCallContent(requireEntry(result.content, 0, "Ollama tool-call content"), { - name: "bash", - arguments: {}, - }); + expect(() => buildAssistantMessage(response, modelInfo)).toThrow( + "Provider completed tool call with malformed JSON arguments", + ); }); it("sets all costs to zero for local models", () => { @@ -1240,25 +1282,29 @@ describe("parseNdjsonStream", () => { await expectNoParsedChunks(reader); }); - it("does not log a dangling surrogate for a malformed complete line", async () => { + it("rejects malformed complete lines without exposing their bytes", async () => { const prefix = "x".repeat(119); const reader = mockNdjsonReader([`${prefix}😀tail`]); - await expectNoParsedChunks(reader); - - expect(ollamaStreamWarnMock).toHaveBeenCalledExactlyOnceWith( - `Skipping malformed NDJSON line: ${prefix}`, + await expect(expectNoParsedChunks(reader)).rejects.toThrow( + "OpenClaw transport error: malformed_streaming_fragment", ); + expect(ollamaStreamWarnMock).not.toHaveBeenCalled(); }); - it("does not log a dangling surrogate for malformed trailing data", async () => { + it("rejects malformed trailing data without exposing its bytes", async () => { const prefix = "x".repeat(119); const reader = mockNdjsonReader([`${prefix}😀tail`], { trailingNewline: false }); - await expectNoParsedChunks(reader); + await expect(expectNoParsedChunks(reader)).rejects.toThrow( + "OpenClaw transport error: malformed_streaming_fragment", + ); + expect(ollamaStreamWarnMock).not.toHaveBeenCalled(); + }); - expect(ollamaStreamWarnMock).toHaveBeenCalledExactlyOnceWith( - `Skipping malformed trailing data: ${prefix}`, + it.each(["null", "[]", "42"])("rejects non-object NDJSON records: %s", async (record) => { + await expect(expectNoParsedChunks(mockNdjsonReader([record]))).rejects.toThrow( + "OpenClaw transport error: malformed_streaming_fragment", ); }); @@ -1412,20 +1458,16 @@ describe("parseNdjsonStream", () => { await source.cancelPending; }); - it("skips malformed NDJSON and unlocks after a valid terminal record", async () => { + it("rejects malformed NDJSON before a later valid terminal record", async () => { const stream = createClosedNdjsonStream([ "not-json", '{"model":"m","created_at":"t","message":{"role":"assistant","content":"done"},"done":true}', ]); - const chunks = []; - for await (const chunk of parseNdjsonStream(stream.getReader())) { - chunks.push(chunk); - } - - expect(chunks).toHaveLength(1); - expect(chunks[0]?.done).toBe(true); - expect(ollamaStreamWarnMock).toHaveBeenCalledWith("Skipping malformed NDJSON line: not-json"); + await expect(expectNoParsedChunks(stream.getReader())).rejects.toThrow( + "OpenClaw transport error: malformed_streaming_fragment", + ); + expect(ollamaStreamWarnMock).not.toHaveBeenCalled(); expect(stream.locked).toBe(false); }); }); @@ -1618,6 +1660,23 @@ async function nextEventWithin( } describe("createOllamaStreamFn streaming events", () => { + it("preserves coded connection failures through shared error projection", async () => { + fetchWithSsrFGuardMock.mockRejectedValue( + Object.assign(new Error("connect failed"), { code: "ECONNREFUSED" }), + ); + + const events = await collectStreamEvents( + await createOllamaTestStream({ baseUrl: "http://ollama-host:11434" }), + ); + + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + type: "error", + reason: "error", + error: { errorMessage: "connect failed", errorCode: "ECONNREFUSED" }, + }); + }); + it("reports the successful HTTP response before streaming events", async () => { const timeline: string[] = []; const onResponse = vi.fn((response, callbackModel) => { @@ -1662,6 +1721,9 @@ describe("createOllamaStreamFn streaming events", () => { it("reports failed HTTP responses before the stream error", async () => { const timeline: string[] = []; + let terminalError: + | { errorMessage?: string; errorCode?: string; errorBody?: string } + | undefined; const onResponse = vi.fn(() => { timeline.push("response"); }); @@ -1679,6 +1741,9 @@ describe("createOllamaStreamFn streaming events", () => { }); for await (const event of stream) { timeline.push(event.type); + if (event.type === "error") { + terminalError = event.error; + } } expect(onResponse).toHaveBeenCalledWith( @@ -1686,6 +1751,11 @@ describe("createOllamaStreamFn streaming events", () => { expect.objectContaining({ id: "qwen3:32b" }), ); expect(timeline).toEqual(["response", "error"]); + expect(terminalError).toMatchObject({ + errorMessage: "429 rate limited", + errorCode: "429", + errorBody: "rate limited", + }); }); it("does not wait for unread response cancellation when the response hook fails", async () => { @@ -1887,6 +1957,64 @@ describe("createOllamaStreamFn streaming events", () => { } }); + it("fails the full stream for malformed terminal tool arguments", async () => { + const events = await collectMockedOllamaEvents([ + '{"model":"m","created_at":"t","message":{"role":"assistant","content":"","tool_calls":[{"id":"call_valid","function":{"name":"read","arguments":{"path":"README.md"}}},{"id":"call_invalid","function":{"name":"bash","arguments":"{\\"command\\":\\"ls\\""}}]},"done":false}', + '{"model":"m","created_at":"t","message":{"role":"assistant","content":""},"done":true}', + ]); + + expect(events.map((event) => event.type)).toEqual(["error"]); + expect(events[0]).toMatchObject({ + type: "error", + error: { errorMessage: "Provider completed tool call with malformed JSON arguments" }, + }); + }); + + it("fails on malformed NDJSON without accepting a later terminal record", async () => { + const events = await collectMockedOllamaEvents([ + "not-json", + '{"model":"m","created_at":"t","message":{"role":"assistant","content":"done"},"done":true}', + ]); + + expect(events.map((event) => event.type)).toEqual(["error"]); + expect(events[0]).toMatchObject({ + type: "error", + error: { errorMessage: "OpenClaw transport error: malformed_streaming_fragment" }, + }); + }); + + it.each([ + ["malformed", "not-json"], + [ + "additional", + '{"model":"m","created_at":"t","message":{"role":"assistant","content":"extra"},"done":false}', + ], + ])("fails when %s data trails a terminal Ollama record", async (_label, trailing) => { + const events = await collectMockedOllamaEvents([ + '{"model":"m","created_at":"t","message":{"role":"assistant","content":"done"},"done":true}', + trailing, + ]); + + const eventTypes = events.map((event) => event.type); + expect(eventTypes.at(-1)).toBe("error"); + expect(eventTypes).not.toContain("text_end"); + expect(eventTypes).not.toContain("done"); + }); + + it("projects official streamed Ollama error records with status metadata", async () => { + const events = await collectMockedOllamaEvents(['{"error":"model failed","status":503}']); + + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + type: "error", + error: { + errorMessage: "503: model failed", + errorCode: "503", + errorBody: '{"error":"model failed","status":503}', + }, + }); + }); + it("estimates usage when the final Ollama chunk omits counters", async () => { const events = await collectMockedOllamaEvents([ '{"model":"m","created_at":"t","message":{"role":"assistant","content":"Estimated answer"},"done":false}', diff --git a/extensions/ollama/src/stream.runtime.ts b/extensions/ollama/src/stream.runtime.ts index 2db5ebe1410a..702d853d0b33 100644 --- a/extensions/ollama/src/stream.runtime.ts +++ b/extensions/ollama/src/stream.runtime.ts @@ -1,7 +1,6 @@ // Ollama stream runtime implements native transport behavior. import { randomUUID } from "node:crypto"; import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; -import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { parseJsonObjectPreservingUnsafeIntegers, parseJsonPreservingUnsafeIntegers, @@ -20,14 +19,22 @@ import type { ProviderRuntimeModel } from "openclaw/plugin-sdk/plugin-entry"; import { isNonSecretApiKeyMarker } from "openclaw/plugin-sdk/provider-auth"; import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http"; import { createPlainTextToolCallCompatWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; -import { createSubsystemLogger } from "openclaw/plugin-sdk/runtime-env"; +import { + describeUnsupportedToolResultMedia, + extractToolResultText, + failTransportStream, + formatToolResultText, + isImageWithMediaPayload, + MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, + parseTerminalToolCallArguments, +} from "openclaw/plugin-sdk/provider-transport-runtime"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; import { isRecord, normalizeOptionalString, readStringValue, } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { estimateStringChars, truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import { estimateStringChars } from "openclaw/plugin-sdk/text-utility-runtime"; import { OLLAMA_CLOUD_BASE_URL, OLLAMA_DEFAULT_BASE_URL } from "./defaults.js"; import { normalizeOllamaWireModelId } from "./model-id.js"; import { buildOllamaBaseUrlSsrFPolicy, isOllamaCloudModel } from "./provider-models.js"; @@ -53,8 +60,6 @@ export { wrapOllamaCompatNumCtx, } from "./stream-compat.js"; -const log = createSubsystemLogger("ollama-stream"); - export const OLLAMA_NATIVE_BASE_URL = OLLAMA_DEFAULT_BASE_URL; const OLLAMA_STREAM_COOPERATIVE_YIELD_INTERVAL_MS = 12; @@ -414,28 +419,6 @@ function buildStreamAssistantMessage(params: { }; } -function buildStreamErrorAssistantMessage(params: { - model: StreamModelDescriptor; - stopReason: Extract; - errorMessage: string; - timestamp?: number; -}): AssistantMessage & { - stopReason: Extract; - errorMessage: string; -} { - return { - ...buildStreamAssistantMessage({ - model: params.model, - content: [], - stopReason: params.stopReason, - usage: buildUsageWithNoCost({}), - timestamp: params.timestamp, - }), - stopReason: params.stopReason, - errorMessage: params.errorMessage, - }; -} - interface OllamaChatRequest { model: string; messages: OllamaChatMessage[]; @@ -452,6 +435,7 @@ interface OllamaChatMessage { images?: string[]; tool_calls?: OllamaToolCall[]; tool_name?: string; + tool_call_id?: string; } interface OllamaTool { @@ -471,7 +455,7 @@ interface OllamaToolCall { }; } -interface OllamaChatResponse { +interface OllamaChatResponse extends Record { model: string; created_at: string; message: { @@ -589,10 +573,6 @@ function ensureArgsObject(value: unknown): Record { return parseJsonObjectPreservingUnsafeIntegers(value) ?? {}; } -function normalizeOllamaToolCallArguments(value: unknown): Record { - return ensureArgsObject(value); -} - function inferOllamaSchemaType(schema: Record): string | undefined { if (schema.properties && isRecord(schema.properties)) { return "object"; @@ -774,8 +754,16 @@ function normalizeOllamaToolCallName( return trimmed.replace(/^(?:functions?|tools?)[./]+/iu, "").trim(); } +type OllamaInputMessage = { + role: string; + content: unknown; + toolName?: unknown; + toolCallId?: unknown; + isError?: unknown; +}; + export function convertToOllamaMessages( - messages: Array<{ role: string; content: unknown }>, + messages: OllamaInputMessage[], system?: string, options: OllamaToolCallNameOptions = {}, ): OllamaChatMessage[] { @@ -809,14 +797,28 @@ export function convertToOllamaMessages( } if (msg.role === "tool" || msg.role === "toolResult") { - const text = extractTextContent(msg.content); - const toolName = - typeof (msg as { toolName?: unknown }).toolName === "string" - ? (msg as { toolName?: string }).toolName - : undefined; + const content = Array.isArray(msg.content) + ? msg.content + : [{ type: "text", text: typeof msg.content === "string" ? msg.content : "" }]; + const text = extractToolResultText(content, { includeStructured: true }); + const images = content.filter(isImageWithMediaPayload).map((part) => part.data); + const omittedMediaPlaceholder = describeUnsupportedToolResultMedia(content, { + images: true, + audio: false, + }); + const mediaPlaceholder = images.length > 0 ? "(see attached image)" : undefined; + const toolName = typeof msg.toolName === "string" ? msg.toolName : undefined; + const toolCallId = typeof msg.toolCallId === "string" ? msg.toolCallId : undefined; result.push({ role: "tool", - content: text, + content: formatToolResultText({ + text, + mediaPlaceholder, + omittedMediaPlaceholder, + isError: msg.isError === true, + }), + ...(images.length > 0 ? { images } : {}), + ...(toolCallId ? { tool_call_id: toolCallId } : {}), ...(toolName ? { tool_name: toolName } : {}), }); } @@ -879,7 +881,7 @@ export function buildAssistantMessage( type: "toolCall", id: readOllamaToolCallId(toolCall.id) ?? `ollama_call_${randomUUID()}`, name: normalizeOllamaToolCallName(toolCall.function.name, options), - arguments: normalizeOllamaToolCallArguments(toolCall.function.arguments), + arguments: parseTerminalToolCallArguments(toolCall.function.arguments), }); } } @@ -917,20 +919,12 @@ export async function* parseNdjsonStream( if (!trimmed) { continue; } - try { - yield parseJsonPreservingUnsafeIntegers(trimmed) as OllamaChatResponse; - } catch { - log.warn(`Skipping malformed NDJSON line: ${truncateUtf16Safe(trimmed, 120)}`); - } + yield parseOllamaNdjsonRecord(trimmed); } } if (buffer.trim()) { - try { - yield parseJsonPreservingUnsafeIntegers(buffer.trim()) as OllamaChatResponse; - } catch { - log.warn(`Skipping malformed trailing data: ${truncateUtf16Safe(buffer.trim(), 120)}`); - } + yield parseOllamaNdjsonRecord(buffer.trim()); } } finally { // Start cancellation best-effort; do not await it — a pending cancel @@ -940,6 +934,36 @@ export async function* parseNdjsonStream( } } +function parseOllamaNdjsonRecord(value: string): OllamaChatResponse { + let parsed: unknown; + try { + parsed = parseJsonPreservingUnsafeIntegers(value); + } catch { + throw new Error(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE); + } + if (!isRecord(parsed)) { + throw new Error(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE); + } + if (typeof parsed.error === "string") { + const status = + typeof parsed.status === "number" && Number.isFinite(parsed.status) + ? parsed.status + : undefined; + throw Object.assign( + new Error(status === undefined ? parsed.error : `${status}: ${parsed.error}`), + { + ...(status === undefined ? {} : { status }), + body: parsed, + }, + ); + } + if (!isRecord(parsed.message) || typeof parsed.done !== "boolean") { + throw new Error(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE); + } + // SAFETY: Required Ollama chat-record fields are validated above; optional fields remain inert. + return parsed as OllamaChatResponse; +} + function resolveOllamaChatUrl(baseUrl: string): string { const trimmed = baseUrl.trim().replace(/\/+$/, ""); const normalizedBase = trimmed.replace(/\/v1$/i, ""); @@ -1080,7 +1104,10 @@ function createRawOllamaStreamFn( response, OLLAMA_STREAM_ERROR_BODY_LIMIT_BYTES, ).catch(() => "unknown error"); - throw new Error(`${response.status} ${errorText}`); + throw Object.assign(new Error(`${response.status} ${errorText}`), { + status: response.status, + body: errorText, + }); } if (!response.body) { throw new Error("Ollama API returned empty response body"); @@ -1224,8 +1251,11 @@ function createRawOllamaStreamFn( for await (const chunk of parseNdjsonStream(reader)) { throwIfOllamaStreamAborted(options?.signal); - // Keep guarded timeouts tied to stream progress so slow remote - // inference is not aborted while Ollama is still emitting tokens. + if (finalResponse) { + throw new Error(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE); + } + // Keep guarded timeouts tied to inference progress. Once done arrives, + // trailing validation stays on the existing bounded request deadline. refreshTimeout?.(); const thinkingDelta = chunk.message?.thinking ?? chunk.message?.reasoning; if (thinkingDelta && shouldEmitThinking) { @@ -1279,7 +1309,7 @@ function createRawOllamaStreamFn( if (chunk.done) { pendingFinalVisibleContent = resolveVisibleContent(true); finalResponse = chunk; - break; + continue; } await cooperativeScheduler.afterEvent(); } @@ -1377,13 +1407,15 @@ function createRawOllamaStreamFn( } } catch (err) { const stopReason = options?.signal?.aborted ? "aborted" : "error"; - stream.push({ - type: "error", - reason: stopReason, - error: buildStreamErrorAssistantMessage({ + failTransportStream({ + stream, + signal: options?.signal, + error: err, + output: buildStreamAssistantMessage({ model, + content: [], stopReason, - errorMessage: formatErrorMessage(err), + usage: buildUsageWithNoCost({}), }), }); } finally { diff --git a/packages/ai/src/providers/anthropic.test.ts b/packages/ai/src/providers/anthropic.test.ts index 1afc2dc4638c..c1b149aef033 100644 --- a/packages/ai/src/providers/anthropic.test.ts +++ b/packages/ai/src/providers/anthropic.test.ts @@ -1827,6 +1827,136 @@ describe("Anthropic provider", () => { ]); }); + it("rejects a malformed later tool before any sibling becomes executable", async () => { + const client = createAnthropicSseClient([ + { + type: "message_start", + message: { id: "msg_tools", usage: { input_tokens: 1, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "call_valid", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":"README.md"}' }, + }, + { type: "content_block_stop", index: 0 }, + { + type: "content_block_start", + index: 1, + content_block: { type: "tool_use", id: "call_invalid", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 1, + delta: { type: "input_json_delta", partial_json: '{"path":"SECRET.md"' }, + }, + { type: "content_block_stop", index: 1 }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 1, output_tokens: 2 }, + }, + { type: "message_stop" }, + ]); + const stream = streamAnthropic( + makeAnthropicModel(), + { messages: [{ role: "user", content: "read", timestamp: 0 }] }, + { apiKey: "sk-ant-provider", client: client as never }, + ); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe("Provider completed tool call with malformed JSON arguments"); + expect(result.errorMessage).not.toContain("SECRET.md"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + }); + + it("rejects an active tool call that never receives content_block_stop", async () => { + const client = createAnthropicSseClient([ + { + type: "message_start", + message: { id: "msg_unsealed", usage: { input_tokens: 1, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "call_unsealed", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":"README.md"' }, + }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + { type: "message_stop" }, + ]); + const stream = streamAnthropic( + makeAnthropicModel(), + { messages: [{ role: "user", content: "read", timestamp: 0 }] }, + { apiKey: "sk-ant-provider", client: client as never }, + ); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(eventTypes.at(-1)).toBe("error"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + expect(result.content.some((block) => block.type === "toolCall")).toBe(false); + }); + + it("uses a complete tool input seeded at block start when no deltas arrive", async () => { + const client = createAnthropicSseClient([ + { + type: "message_start", + message: { id: "msg_seeded", usage: { input_tokens: 1, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { + type: "tool_use", + id: "call_seeded", + name: "read", + input: { path: "README.md" }, + }, + }, + { type: "content_block_stop", index: 0 }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + { type: "message_stop" }, + ]); + + const result = await streamAnthropic( + makeAnthropicModel(), + { messages: [{ role: "user", content: "read", timestamp: 0 }] }, + { apiKey: "sk-ant-provider", client: client as never }, + ).result(); + + expect(result.content).toContainEqual( + expect.objectContaining({ type: "toolCall", arguments: { path: "README.md" } }), + ); + }); + it("discards buffered Fable output when the stream fails before terminal status", async () => { const client = createAnthropicSseClient([ { diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index e21a997c62bc..2d7f81835804 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -29,7 +29,10 @@ import { type AnthropicCompactionBlock, } from "../transports/anthropic-compaction-replay.js"; import { applyAnthropicCacheControlToMessages } from "../transports/anthropic-payload-policy.js"; -import { transportAbortError } from "../transports/transport-stream-shared.js"; +import { + finalizeTerminalToolCallArguments, + transportAbortError, +} from "../transports/transport-stream-shared.js"; import { MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE } from "../transports/transport-utils.js"; import type { AnthropicMessagesCompat, @@ -427,11 +430,15 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp model, ); - type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson: string })) & { + type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson?: string })) & { index: number; }; const blocks = output.content as Block[]; const blockIndexes = new Map(); + const sealedToolCalls: Array<{ + block: Extract; + contentIndex: number; + }> = []; const compactionCapture = createCompactionCapture(output, model, requestOptions); for await (const event of iterateAnthropicEvents(response, refusalBuffer !== undefined)) { @@ -465,6 +472,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp // reference them, so rebuild the deferred timeline from the // surviving text prefix the fallback model continued from. refusalBuffer?.discard(); + sealedToolCalls.length = 0; blockIndexes.clear(); applyAnthropicFallbackBoundary({ output, @@ -598,7 +606,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp const index = blockIndexes.get(event.index); const block = index === undefined ? undefined : blocks[index]; if (index !== undefined && block?.type === "toolCall") { - block.partialJson += event.delta.partial_json; + block.partialJson = (block.partialJson ?? "") + event.delta.partial_json; block.arguments = parseStreamingJson(block.partialJson); eventSink.push({ type: "toolcall_delta", @@ -639,16 +647,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp partial: output, }); } else if (block.type === "toolCall") { - block.arguments = parseStreamingJson(block.partialJson); - // Finalize in-place and strip the scratch buffer so replay only - // carries parsed arguments. - delete (block as { partialJson?: string }).partialJson; - eventSink.push({ - type: "toolcall_end", - contentIndex: index, - toolCall: block, - partial: output, - }); + sealedToolCalls.push({ block, contentIndex: index }); } } } else if (event.type === "message_delta") { @@ -675,11 +674,29 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error(output.errorMessage ?? "An unknown error occurred"); } + if ([...blockIndexes.values()].some((index) => blocks[index]?.type === "toolCall")) { + throw new Error("Provider completed stream with an incomplete tool call"); + } + finalizeTerminalToolCallArguments( + sealedToolCalls.map(({ block }) => block), + (block) => + block.partialJson && block.partialJson.length > 0 ? block.partialJson : block.arguments, + ); + for (const sealed of sealedToolCalls) { + delete sealed.block.partialJson; + eventSink.push({ + type: "toolcall_end", + contentIndex: sealed.contentIndex, + toolCall: sealed.block, + partial: output, + }); + } refusalBuffer?.flush(); stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { + output.content = output.content.filter((block) => block.type !== "toolCall"); for (const block of output.content) { delete (block as { index?: number }).index; // partialJson is only a streaming scratch buffer; never persist it. diff --git a/packages/ai/src/providers/mistral.bounded-stream.test.ts b/packages/ai/src/providers/mistral.bounded-stream.test.ts index 543820c667bb..a2b0f3ecf202 100644 --- a/packages/ai/src/providers/mistral.bounded-stream.test.ts +++ b/packages/ai/src/providers/mistral.bounded-stream.test.ts @@ -323,6 +323,18 @@ describe("Mistral terminal ownership through the installed SDK and real HTTP/SSE expect(events).toContain("toolcall_end"); }); + it("preserves unsafe integers in provider-confirmed tool arguments", async () => { + const { result, events } = await streamMistralTerminalFixture({ + finishReason: "tool_calls", + done: true, + toolArguments: ['{"target":9223372036854775807}'], + }); + expect(result.content).toContainEqual( + expect.objectContaining({ type: "toolCall", arguments: { target: "9223372036854775807" } }), + ); + expect(events).toContain("toolcall_end"); + }); + it.each(["null", "[]", "42", '"dangerous"'] as const)( "rejects a provider-confirmed non-object JSON argument: %s", async (argumentsJson) => { diff --git a/packages/ai/src/providers/mistral.ts b/packages/ai/src/providers/mistral.ts index a9e28c9764de..88cafcf3f4f5 100644 --- a/packages/ai/src/providers/mistral.ts +++ b/packages/ai/src/providers/mistral.ts @@ -8,12 +8,14 @@ import type { ContentChunk, FunctionTool, } from "@mistralai/mistralai/models/components"; -import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { getEnvApiKey } from "../env-api-keys.js"; import { getAiTransportHost } from "../host.js"; import { calculateCost, clampThinkingLevel } from "../model-utils.js"; import { transformProviderMessages as transformMessages } from "../provider-transcript-transform.js"; -import { transportAbortError } from "../transports/transport-stream-shared.js"; +import { + finalizeTerminalToolCallArguments, + transportAbortError, +} from "../transports/transport-stream-shared.js"; import type { AssistantMessage, Context, @@ -40,6 +42,7 @@ import { buildBaseOptions, clampMaxTokensToModel } from "./simple-options.js"; import { describeToolResultMediaPlaceholder, extractToolResultText, + formatToolResultText, isImageWithMediaPayload, } from "./tool-result-text.js"; @@ -756,31 +759,22 @@ async function consumeChatStream( } return; } - try { - for (const index of toolBlockIdentities.keys()) { - const rawArguments = (blocks[index] as ToolCall & { partialArgs?: string }).partialArgs ?? ""; - if (!isRecord(JSON.parse(rawArguments))) { - throw new Error("Mistral tool-call arguments must be a JSON object"); - } - } - } catch { - throw new Error("Mistral completed tool call has invalid JSON arguments"); - } - for (const index of toolBlockIdentities.keys()) { - const block = output.content.at(index); - if (block?.type !== "toolCall") { - continue; - } - const toolBlock = block as ToolCall & { partialArgs?: string }; + const completedToolCalls = [...toolBlockIdentities.keys()].flatMap((contentIndex) => { + const block = blocks[contentIndex]; + return block?.type === "toolCall" + ? [{ block: block as ToolCall & { partialArgs?: string }, contentIndex }] + : []; + }); + finalizeTerminalToolCallArguments( + completedToolCalls.map(({ block }) => block), + (block) => block.partialArgs ?? "", + "Mistral completed tool call has invalid JSON arguments", + ); + for (const { block, contentIndex } of completedToolCalls) { // Finalize in-place and strip the scratch buffer so replay only // carries parsed arguments. - delete toolBlock.partialArgs; - stream.push({ - type: "toolcall_end", - contentIndex: index, - toolCall: toolBlock, - partial: output, - }); + delete block.partialArgs; + stream.push({ type: "toolcall_end", contentIndex, toolCall: block, partial: output }); } } @@ -901,13 +895,20 @@ function toChatMessages( const textResult = extractToolResultText(msg.content); const mediaPlaceholder = describeToolResultMediaPlaceholder(msg.content); const hasImages = msg.content.some(isImageWithMediaPayload); - const toolText = buildToolResultText( - textResult, + const omittedMediaPlaceholder = + hasImages && !supportsImages + ? textResult.trim() + ? "[tool image omitted: model does not support images]" + : mediaPlaceholder === "(see attached media)" + ? "(media omitted: model does not support images)" + : "(image omitted: model does not support images)" + : undefined; + const toolText = formatToolResultText({ + text: textResult, mediaPlaceholder, - hasImages, - supportsImages, - msg.isError, - ); + omittedMediaPlaceholder, + isError: msg.isError, + }); toolContent.push({ type: "text", text: toolText }); for (const part of msg.content) { if (!supportsImages) { @@ -932,36 +933,6 @@ function toChatMessages( return result; } -function buildToolResultText( - text: string, - mediaPlaceholder: string | undefined, - hasImages: boolean, - supportsImages: boolean, - isError: boolean, -): string { - const trimmed = text.trim(); - const errorPrefix = isError ? "[tool error] " : ""; - - if (trimmed.length > 0) { - const imageSuffix = - hasImages && !supportsImages ? "\n[tool image omitted: model does not support images]" : ""; - return `${errorPrefix}${trimmed}${imageSuffix}`; - } - - if (mediaPlaceholder) { - if (!hasImages || supportsImages) { - return `${errorPrefix}${mediaPlaceholder}`; - } - const omitted = - mediaPlaceholder === "(see attached media)" - ? "(media omitted: model does not support images)" - : "(image omitted: model does not support images)"; - return `${errorPrefix}${omitted}`; - } - - return isError ? "[tool error] (no tool output)" : "(no tool output)"; -} - function usesReasoningEffort(model: Model<"mistral-conversations">): boolean { return ( model.id === "mistral-small-2603" || diff --git a/packages/ai/src/providers/openai-completions-tool-calls.ts b/packages/ai/src/providers/openai-completions-tool-calls.ts index 83a1add53d10..6a8aa4772da1 100644 --- a/packages/ai/src/providers/openai-completions-tool-calls.ts +++ b/packages/ai/src/providers/openai-completions-tool-calls.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import type { ChatCompletionChunk } from "openai/resources/chat/completions.js"; import { measureUtf8AppendBytes } from "../transports/openai-transport-shared.js"; +import { finalizeTerminalToolCallArguments } from "../transports/transport-stream-shared.js"; type ChatCompletionToolCallDelta = ChatCompletionChunk.Choice.Delta.ToolCall; const MAX_BUFFERED_TOOL_CALL_ARGUMENT_BYTES = 256_000; @@ -218,31 +219,34 @@ export function finalizeOpenAICompletionsToolCalls( return; } - for (const block of output.content) { - if (!isToolCall(block)) { - continue; - } - const toolCall = block as { name?: unknown; arguments?: unknown; partialArgs?: unknown }; - let completeArguments: unknown; - try { - completeArguments = - typeof toolCall.partialArgs === "string" && toolCall.partialArgs.trim().length > 0 - ? (JSON.parse(toolCall.partialArgs) as unknown) - : undefined; - } catch { - completeArguments = undefined; - } - if ( - typeof toolCall.name !== "string" || - toolCall.name.trim().length === 0 || - !isRecord(completeArguments) - ) { - output.stopReason = "error"; - output.errorMessage = "Provider returned an incomplete or malformed tool call"; - output.content = output.content.filter((candidate) => !isToolCall(candidate)); - return; - } - toolCall.arguments = completeArguments; + type FinalToolCall = TBlock & { + name?: unknown; + arguments: Record; + partialArgs?: unknown; + }; + const toolCalls = output.content.filter(isToolCall) as FinalToolCall[]; + const rejectToolCalls = () => { + output.stopReason = "error"; + output.errorMessage = "Provider returned an incomplete or malformed tool call"; + output.content = output.content.filter((candidate) => !isToolCall(candidate)); + }; + if ( + toolCalls.some( + (call) => + typeof call.name !== "string" || + call.name.trim().length === 0 || + typeof call.partialArgs !== "string" || + call.partialArgs.trim().length === 0, + ) + ) { + rejectToolCalls(); + return; + } + try { + finalizeTerminalToolCallArguments(toolCalls, (call) => call.partialArgs); + } catch { + rejectToolCalls(); + return; } for (let contentIndex = 0; contentIndex < output.content.length; contentIndex += 1) { diff --git a/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts b/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts index c8deab2e4c46..85a2a94f458e 100644 --- a/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts +++ b/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts @@ -460,6 +460,22 @@ describe.each([ ]); }); + it("preserves unsafe integers in confirmed tool arguments", async () => { + const { result } = await collectFixture( + confirmedModernCallChunks('{"target":9223372036854775807}', { + id: "call_unsafe_integer", + }), + ); + + expect(result.content).toContainEqual( + expect.objectContaining({ + type: "toolCall", + id: "call_unsafe_integer", + arguments: { target: "9223372036854775807" }, + }), + ); + }); + it("preserves confirmed tool completion before following text blocks close", async () => { const { eventTypes, result } = await collectFixture([ modernCallChunk('{"query":"cats"}', { id: "call_before_text" }), diff --git a/packages/ai/src/providers/tool-result-text.test.ts b/packages/ai/src/providers/tool-result-text.test.ts index 44d6de1810a3..17d17a7bde4e 100644 --- a/packages/ai/src/providers/tool-result-text.test.ts +++ b/packages/ai/src/providers/tool-result-text.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import { describeToolResultMediaPlaceholder, + describeUnsupportedToolResultMedia, extractToolResultText, hasMediaPayload, isImageWithMediaPayload, @@ -145,12 +146,15 @@ describe("describeToolResultMediaPlaceholder", () => { }); it("does not advertise payload-less media husks", () => { + const husks = [ + { type: "image", mimeType: "image/png", data: "" }, + { type: "image", path: "/tmp/image.png" }, + { type: "audio", mimeType: "audio/mpeg" }, + { type: "text", text: "ordinary text", mimeType: "image/png" }, + ]; + expect(describeToolResultMediaPlaceholder(husks)).toBeUndefined(); expect( - describeToolResultMediaPlaceholder([ - { type: "image", mimeType: "image/png", data: "" }, - { type: "image", path: "/tmp/image.png" }, - { type: "audio", mimeType: "audio/mpeg" }, - ]), + describeUnsupportedToolResultMedia(husks, { images: true, audio: false }), ).toBeUndefined(); }); diff --git a/packages/ai/src/providers/tool-result-text.ts b/packages/ai/src/providers/tool-result-text.ts index 2a7d01d19f6d..dc396d42b05f 100644 --- a/packages/ai/src/providers/tool-result-text.ts +++ b/packages/ai/src/providers/tool-result-text.ts @@ -133,42 +133,37 @@ export function isImageWithMediaPayload(block: T): block is T & { type: "imag return isRecord(block) && block.type === "image" && hasMediaPayload(block); } -export function describeToolResultMediaPlaceholder(blocks: readonly unknown[]): string | undefined { +function classifyToolResultMedia(blocks: readonly unknown[]): { + hasImage: boolean; + hasAudio: boolean; +} { let hasImage = false; let hasAudio = false; - for (const block of blocks) { - if (!hasMediaPayload(block)) { + if (!hasMediaPayload(block) || block.type === "text") { continue; } - const record = block; - const type = typeof record.type === "string" ? record.type : undefined; - const mimeType = readMimeType(record); - - if ( - (type && IMAGE_TOOL_RESULT_TYPES.has(type)) || - mimeType?.toLowerCase().startsWith("image/") - ) { - hasImage = true; - } - if ( - (type && AUDIO_TOOL_RESULT_TYPES.has(type)) || - mimeType?.toLowerCase().startsWith("audio/") - ) { - hasAudio = true; - } + const type = typeof block.type === "string" ? block.type : undefined; + const mimeType = readMimeType(block)?.toLowerCase(); + hasImage ||= Boolean( + (type && IMAGE_TOOL_RESULT_TYPES.has(type)) || mimeType?.startsWith("image/"), + ); + hasAudio ||= Boolean( + (type && AUDIO_TOOL_RESULT_TYPES.has(type)) || mimeType?.startsWith("audio/"), + ); } + return { hasImage, hasAudio }; +} +export function describeToolResultMediaPlaceholder(blocks: readonly unknown[]): string | undefined { + const { hasImage, hasAudio } = classifyToolResultMedia(blocks); if (hasImage && hasAudio) { return "(see attached media)"; } if (hasAudio) { return "(see attached audio)"; } - if (hasImage) { - return "(see attached image)"; - } - return undefined; + return hasImage ? "(see attached image)" : undefined; } export function extractToolResultBlockText(block: unknown): string | undefined { @@ -187,7 +182,10 @@ export function extractToolResultBlockText(block: unknown): string | undefined { return structured ? sanitizeSurrogates(truncateProviderToolText(structured)) : undefined; } -export function extractToolResultText(blocks: readonly unknown[]): string { +export function extractToolResultText( + blocks: readonly unknown[], + options?: { includeStructured?: boolean }, +): string { const explicitTexts: string[] = []; const structuredTexts: string[] = []; for (const block of blocks) { @@ -203,7 +201,42 @@ export function extractToolResultText(blocks: readonly unknown[]): string { } } if (explicitTexts.length > 0) { - return sanitizeSurrogates(explicitTexts.join("\n")); + const text = ( + options?.includeStructured ? [...explicitTexts, ...structuredTexts] : explicitTexts + ).join("\n"); + return sanitizeSurrogates(options?.includeStructured ? truncateProviderToolText(text) : text); } return sanitizeSurrogates(truncateProviderToolText(structuredTexts.join("\n"))); } + +type ToolResultMediaSupport = { images: boolean; audio: boolean }; + +/** Describe media that cannot be represented on the target provider wire. */ +export function describeUnsupportedToolResultMedia( + blocks: readonly unknown[], + support: ToolResultMediaSupport, +): string | undefined { + const { hasImage, hasAudio } = classifyToolResultMedia(blocks); + const omittedImage = hasImage && !support.images; + const omittedAudio = hasAudio && !support.audio; + if (omittedImage && omittedAudio) { + return "[unsupported tool-result media omitted]"; + } + if (omittedAudio) { + return "[unsupported tool-result audio omitted]"; + } + return omittedImage ? "[unsupported tool-result image omitted]" : undefined; +} + +export function formatToolResultText(params: { + text: string; + mediaPlaceholder?: string; + omittedMediaPlaceholder?: string; + isError: boolean; +}): string { + const trimmed = params.text.trim(); + const body = trimmed + ? `${trimmed}${params.omittedMediaPlaceholder ? `\n${params.omittedMediaPlaceholder}` : ""}` + : (params.omittedMediaPlaceholder ?? params.mediaPlaceholder ?? "(no tool output)"); + return `${params.isError ? "[tool error] " : ""}${body}`; +} diff --git a/packages/ai/src/transports.ts b/packages/ai/src/transports.ts index bac30c22a6fd..665bd0d2d26d 100644 --- a/packages/ai/src/transports.ts +++ b/packages/ai/src/transports.ts @@ -23,4 +23,7 @@ export * from "./transports/provider-transport-stream.js"; export * from "./transports/responses-image-payload-sanitizer.js"; export * from "./transports/simple-completion-transport.js"; export * from "./transports/transport-stream-shared.js"; -export { isCodeModeModelVisibleToolName } from "./transports/transport-utils.js"; +export { + isCodeModeModelVisibleToolName, + MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, +} from "./transports/transport-utils.js"; diff --git a/packages/ai/src/transports/anthropic-transport-stream.test.ts b/packages/ai/src/transports/anthropic-transport-stream.test.ts index f4940e358f37..1195e3610809 100644 --- a/packages/ai/src/transports/anthropic-transport-stream.test.ts +++ b/packages/ai/src/transports/anthropic-transport-stream.test.ts @@ -764,7 +764,10 @@ describe("anthropic transport stream", () => { ); const result = await runTransportStream( - makeAnthropicTransportModel(), + makeAnthropicTransportModel({ + provider: "pioneer", + baseUrl: "https://bedrock-compatible.example/v1", + }), { messages: [{ role: "user", content: "run date" }], } as AnthropicStreamContext, @@ -1993,6 +1996,7 @@ describe("anthropic transport stream", () => { delta: { stop_reason: "tool_use" }, usage: { input_tokens: 10, output_tokens: 5 }, }, + { type: "message_stop" }, ]), ); @@ -2018,6 +2022,148 @@ describe("anthropic transport stream", () => { }); }); + it("rejects malformed terminal tool JSON before completing any sibling call", async () => { + guardedFetchMock.mockResolvedValueOnce( + createSseResponse([ + { + type: "message_start", + message: { id: "msg_malformed_tools", usage: { input_tokens: 2, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "call_valid", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":"README.md"}' }, + }, + { type: "content_block_stop", index: 0 }, + { + type: "content_block_start", + index: 1, + content_block: { type: "tool_use", id: "call_invalid", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 1, + delta: { type: "input_json_delta", partial_json: '{"path":"SECRET.md"' }, + }, + { type: "content_block_stop", index: 1 }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 2, output_tokens: 2 }, + }, + { type: "message_stop" }, + ]), + ); + const streamFn = createAnthropicMessagesTransportStreamFn(); + const stream = await Promise.resolve( + streamFn( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "read" }] } as AnthropicStreamContext, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, + ), + ); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe("Provider completed tool call with malformed JSON arguments"); + expect(result.errorMessage).not.toContain("SECRET.md"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + }); + + it("rejects an active tool call that never receives content_block_stop", async () => { + guardedFetchMock.mockResolvedValueOnce( + createSseResponse([ + { + type: "message_start", + message: { id: "msg_unsealed", usage: { input_tokens: 2, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "call_unsealed", name: "read", input: {} }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":"README.md"' }, + }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 2, output_tokens: 1 }, + }, + { type: "message_stop" }, + ]), + ); + const streamFn = createAnthropicMessagesTransportStreamFn(); + const stream = await Promise.resolve( + streamFn( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "read" }] } as AnthropicStreamContext, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, + ), + ); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(eventTypes.at(-1)).toBe("error"); + expect(eventTypes).not.toContain("toolcall_end"); + expect(eventTypes).not.toContain("done"); + expect(result.content.some((block) => block.type === "toolCall")).toBe(false); + }); + + it("uses seeded Anthropic tool input when no argument deltas arrive", async () => { + guardedFetchMock.mockResolvedValueOnce( + createSseResponse([ + { + type: "message_start", + message: { id: "msg_seeded_tool", usage: { input_tokens: 2, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { + type: "tool_use", + id: "call_seeded", + name: "read", + input: { path: "README.md" }, + }, + }, + { type: "content_block_stop", index: 0 }, + { + type: "message_delta", + delta: { stop_reason: "tool_use" }, + usage: { input_tokens: 2, output_tokens: 1 }, + }, + { type: "message_stop" }, + ]), + ); + + const result = await runTransportStream( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "read" }] } as AnthropicStreamContext, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, + ); + + expect(result.content).toContainEqual( + expect.objectContaining({ type: "toolCall", arguments: { path: "README.md" } }), + ); + }); + it("preserves Anthropic OAuth identity and tool-name remapping with transport overrides", async () => { guardedFetchMock.mockResolvedValueOnce( createSseResponse([ diff --git a/packages/ai/src/transports/anthropic-transport-stream.ts b/packages/ai/src/transports/anthropic-transport-stream.ts index bba288c39ae6..9b939d1a384a 100644 --- a/packages/ai/src/transports/anthropic-transport-stream.ts +++ b/packages/ai/src/transports/anthropic-transport-stream.ts @@ -105,6 +105,7 @@ import { createEmptyTransportUsage, createWritableTransportEventStream, failTransportStream, + finalizeTerminalToolCallArguments, finalizeTransportStream, mergeTransportHeaders, sanitizeNonEmptyTransportPayloadText, @@ -1197,6 +1198,10 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { ); const blocks = output.content; const blockIndexes = new Map(); + const sealedToolCalls: Array<{ + block: Extract; + contentIndex: number; + }> = []; const compactionCapture = createCompactionCapture(output, model, transportOptions); // Signature deltas are opaque and only complete at content_block_stop. // Keep partial bytes out of output so interrupted streams cannot poison replay. @@ -1369,6 +1374,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { // events reference them, so rebuild the deferred timeline from // the surviving text prefix the fallback model continued from. refusalBuffer?.discard(); + sealedToolCalls.length = 0; pendingTextEnds.length = 0; blockIndexes.clear(); pendingThinkingSignatures.clear(); @@ -1679,13 +1685,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { continue; } if (block.type === "toolCall") { - delete block.partialJson; - eventSink.push({ - type: "toolcall_end", - contentIndex: index, - toolCall: block, - partial: output, - }); + sealedToolCalls.push({ block, contentIndex: index }); finishReasoningContentSidecars(event.index); } continue; @@ -1727,6 +1727,23 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error(output.errorMessage ?? "An unknown error occurred"); } + if ([...blockIndexes.values()].some((index) => blocks[index]?.type === "toolCall")) { + throw new Error("Provider completed stream with an incomplete tool call"); + } + finalizeTerminalToolCallArguments( + sealedToolCalls.map(({ block }) => block), + (block) => + block.partialJson && block.partialJson.length > 0 ? block.partialJson : block.arguments, + ); + for (const sealed of sealedToolCalls) { + delete sealed.block.partialJson; + eventSink.push({ + type: "toolcall_end", + contentIndex: sealed.contentIndex, + toolCall: sealed.block, + partial: output, + }); + } refusalBuffer?.flush(); // Backstop: streaming tags commentary at the tool-boundary above, but // replay/non-streaming assembly may reach here with tool calls untagged. @@ -1745,6 +1762,8 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { if (refusalBuffer) { refusalBuffer.discard(); output.content = []; + } else { + output.content = output.content.filter((block) => block.type !== "toolCall"); } if (usedCompactionReplay && isAnthropicReplayRejection(error)) { suppressAnthropicCompaction(output, model, options); diff --git a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts index 2db9cd56b53a..8302db68dec0 100644 --- a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts @@ -23,7 +23,6 @@ import type { ToolCall, Usage, } from "../types.js"; -import { parseJsonObjectPreservingUnsafeIntegers } from "./json-unsafe-integers.js"; import { captureOpenAIResponsesCompaction } from "./openai-responses-compaction-replay.js"; import { OPENAI_RESPONSES_COMPACTION_REPLAY_TYPE, @@ -31,6 +30,7 @@ import { type OpenAIResponsesReasoningReplayMetadata, } from "./openai-responses-contracts.js"; import { encodeTextSignatureV1 } from "./openai-responses-replay-internal.js"; +import { parseTerminalToolCallArguments } from "./transport-stream-shared.js"; export type ResponsesEventSink = { push(event: AssistantMessageEvent): void }; export type TextBlockReference = { @@ -97,12 +97,10 @@ export function resolveCompletedResponsesToolCall( if (!name) { throw new Error("Responses stream completed tool call without a function name"); } - const argumentsValue = parseJsonObjectPreservingUnsafeIntegers( + const argumentsValue = parseTerminalToolCallArguments( streamed?.arguments ?? item.arguments, + "Responses stream completed tool call with invalid JSON arguments", ); - if (!argumentsValue) { - throw new Error("Responses stream completed tool call with invalid JSON arguments"); - } return { name, arguments: argumentsValue }; } diff --git a/packages/ai/src/transports/transport-stream-shared.test.ts b/packages/ai/src/transports/transport-stream-shared.test.ts new file mode 100644 index 000000000000..409d94357755 --- /dev/null +++ b/packages/ai/src/transports/transport-stream-shared.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { parseTerminalToolCallArguments } from "./transport-stream-shared.js"; + +const MALFORMED_TOOL_CALL_TERMINAL_ERROR_MESSAGE = + "Provider completed tool call with malformed JSON arguments"; + +describe("parseTerminalToolCallArguments", () => { + it("preserves unsafe integer literals in complete object arguments", () => { + expect(parseTerminalToolCallArguments('{"target":9223372036854775807,"safe":42}')).toEqual({ + target: "9223372036854775807", + safe: 42, + }); + expect(parseTerminalToolCallArguments({})).toEqual({}); + }); + + it.each(["", " ", '{"secret":"do-not-echo"', "[]", "null", null])( + "rejects non-object or malformed terminal input %# without exposing it", + (value) => { + let thrown: unknown; + try { + parseTerminalToolCallArguments(value); + } catch (error) { + thrown = error; + } + expect(thrown).toMatchObject({ message: MALFORMED_TOOL_CALL_TERMINAL_ERROR_MESSAGE }); + expect(String(thrown)).not.toContain("do-not-echo"); + }, + ); +}); diff --git a/packages/ai/src/transports/transport-stream-shared.ts b/packages/ai/src/transports/transport-stream-shared.ts index f14ec518bc9a..87c80d9c6b6e 100644 --- a/packages/ai/src/transports/transport-stream-shared.ts +++ b/packages/ai/src/transports/transport-stream-shared.ts @@ -8,6 +8,7 @@ import { asNonArrayRecord, asOptionalRecord } from "@openclaw/normalization-core import { createAssistantMessageEventStream } from "../utils/event-stream.js"; import { projectProviderError, type ProviderErrorProjection } from "../utils/provider-error.js"; import { sanitizeSurrogates } from "../utils/sanitize-unicode.js"; +import { parseJsonObjectPreservingUnsafeIntegers } from "./json-unsafe-integers.js"; type ContextUsage = NonNullable; @@ -27,6 +28,8 @@ export type WritableTransportStream = Pick< >; const EMPTY_TOOL_RESULT_TEXT = "(no output)"; +const MALFORMED_TOOL_CALL_TERMINAL_ERROR_MESSAGE = + "Provider completed tool call with malformed JSON arguments"; export function sanitizeTransportPayloadText(text: string): string { if (typeof text !== "string") { return ""; @@ -58,6 +61,32 @@ export function coerceTransportToolCallArguments(argumentsValue: unknown): Recor return {}; } +/** Admit only complete object-shaped terminal tool arguments; partial parsing is preview-only. */ +export function parseTerminalToolCallArguments( + value: unknown, + errorMessage = MALFORMED_TOOL_CALL_TERMINAL_ERROR_MESSAGE, +): Record { + const parsed = parseJsonObjectPreservingUnsafeIntegers(value); + if (!parsed) { + throw new Error(errorMessage); + } + return parsed; +} + +/** Validate a complete sibling set before mutating any call into executable state. */ +export function finalizeTerminalToolCallArguments }>( + calls: readonly T[], + readArguments: (call: T) => unknown, + errorMessage?: string, +): void { + const validated = calls.map( + (call) => [call, parseTerminalToolCallArguments(readArguments(call), errorMessage)] as const, + ); + for (const [call, argumentsValue] of validated) { + call.arguments = argumentsValue; + } +} + export function mergeTransportHeaders( ...headerSources: Array | undefined> ): Record | undefined { diff --git a/src/plugin-sdk/provider-transport-runtime.ts b/src/plugin-sdk/provider-transport-runtime.ts index cf4d1a58656a..407967b34f61 100644 --- a/src/plugin-sdk/provider-transport-runtime.ts +++ b/src/plugin-sdk/provider-transport-runtime.ts @@ -10,15 +10,21 @@ export { export { transformTransportMessages } from "../agents/transport-message-transform.js"; export { describeToolResultMediaPlaceholder, + describeUnsupportedToolResultMedia, extractToolResultText, + formatToolResultText, + isImageWithMediaPayload, } from "@openclaw/ai/internal/shared"; export { coerceTransportToolCallArguments, createEmptyTransportUsage, createWritableTransportEventStream, failTransportStream, + finalizeTerminalToolCallArguments, finalizeTransportStream, + MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, mergeTransportHeaders, + parseTerminalToolCallArguments, sanitizeTransportPayloadText, type WritableTransportStream, } from "@openclaw/ai/transports";