ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 11:25:50 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(providers): prevent malformed streamed tool calls (#126391)

Browse Source 
Co-authored-by: 曾令彪 0668001395 <zeng.lingbiao@xydigit.com>
This commit is contained in:
Peter Steinberger
2026-08-19 11:30:09 -07:00
committed by GitHub
parent 0a8f2c1845
commit 8a9e21d3bc
21 changed files with 986 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
+4 -4
View File
@@ -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
extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts
+1
View File
@@ -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,
extensions/amazon-bedrock/stream.runtime.test.ts
+127 -4
View File
@@ -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");
}
},
);
extensions/amazon-bedrock/stream.runtime.ts
+51 -11
View File
@@ -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<Block, "partialJson">;
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<number, Uint8Array[]>();
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<number, Uint8Array[]>,
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
extensions/ollama/src/stream-runtime.test.ts
+152 -24
View File
@@ -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<T>(
}
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}',
extensions/ollama/src/stream.runtime.ts
+91 -59
View File
@@ -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<StopReason, "aborted" | "error">;
errorMessage: string;
timestamp?: number;
}): AssistantMessage & {
stopReason: Extract<StopReason, "aborted" | "error">;
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<string, unknown> {
model: string;
created_at: string;
message: {
@@ -589,10 +573,6 @@ function ensureArgsObject(value: unknown): Record<string, unknown> {
return parseJsonObjectPreservingUnsafeIntegers(value) ?? {};
}
function normalizeOllamaToolCallArguments(value: unknown): Record<string, unknown> {
return ensureArgsObject(value);
}
function inferOllamaSchemaType(schema: Record<string, unknown>): 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 {
packages/ai/src/providers/anthropic.test.ts
+130
View File
@@ -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([
{
packages/ai/src/providers/anthropic.ts
+30 -13
View File
@@ -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<number, number>();
const sealedToolCalls: Array<{
block: Extract<Block, { type: "toolCall" }>;
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.
packages/ai/src/providers/mistral.bounded-stream.test.ts
+12
View File
@@ -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) => {
packages/ai/src/providers/mistral.ts
+32 -61
View File
@@ -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" ||
packages/ai/src/providers/openai-completions-tool-calls.ts
+29 -25
View File
@@ -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<TBlock extends object>(
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<string, unknown>;
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) {
packages/ai/src/providers/openai-completions.legacy-function-call.test.ts
+16
View File
@@ -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" }),
packages/ai/src/providers/tool-result-text.test.ts
+9 -5
View File
@@ -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();
});
packages/ai/src/providers/tool-result-text.ts
+58 -25
View File
@@ -133,42 +133,37 @@ export function isImageWithMediaPayload<T>(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}`;
}
packages/ai/src/transports.ts
+4 -1
View File
@@ -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";
packages/ai/src/transports/anthropic-transport-stream.test.ts
+147 -1
View File
@@ -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([
packages/ai/src/transports/anthropic-transport-stream.ts
+26 -7
View File
@@ -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<number, number>();
const sealedToolCalls: Array<{
block: Extract<TransportContentBlock, { type: "toolCall" }>;
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);
packages/ai/src/transports/openai-responses-stream-terminal-internal.ts
+3 -5
View File
@@ -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 };
}
packages/ai/src/transports/transport-stream-shared.test.ts
+29
View File
@@ -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");
},
);
});
packages/ai/src/transports/transport-stream-shared.ts
+29
View File
@@ -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<Usage["contextUsage"]>;
@@ -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<string, unknown> {
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<T extends { arguments: Record<string, unknown> }>(
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<Record<string, string> | undefined>
): Record<string, string> | undefined {
src/plugin-sdk/provider-transport-runtime.ts
+6
View File
@@ -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";