diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index 3a5654300..aacdd054b 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -284,6 +284,23 @@ function toolUseArguments(input: unknown): string { return JSON.stringify(input ?? {}); } +/** + * Whether arguments assembled from a stream's `input_json_delta` fragments are usable. + * A tool block that sent no fragments at all is fine — that is a no-argument call. Anything + * else has to parse, because unlike the non-stream path the fragments have already been + * forwarded to the client and cannot be repaired after the fact. + */ +function streamedToolArgumentsParse(assembled: string): boolean { + const trimmed = assembled.trim(); + if (!trimmed) return true; + try { + JSON.parse(trimmed); + return true; + } catch { + return false; + } +} + function anthropicKeyUsesBearer(provider: OcxProviderConfig): boolean { return provider.apiKeyTransport === "bearer"; } @@ -765,6 +782,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti let currentBlockType = ""; let currentToolCallId = ""; let currentToolCallName = ""; + let currentToolCallJson = ""; let pendingUsage: Record | undefined; let pendingStopReason: string | undefined; let emittedDone = false; @@ -808,6 +826,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti if (block.type === "tool_use") { currentToolCallId = block.id ?? synthesizeToolUseId(); currentToolCallName = toolNames.fromWire(block.name ?? ""); + currentToolCallJson = ""; yield { type: "tool_call_start", id: currentToolCallId, name: currentToolCallName }; } if (block.type === "redacted_thinking" && typeof block.data === "string") { @@ -833,14 +852,32 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti // so a stray signature on a non-thinking block can never be captured. yield { type: "thinking_signature", signature: delta.signature }; } else if (delta.type === "input_json_delta" && typeof delta.partial_json === "string" && currentBlockType === "tool_use") { + // Forwarded immediately: the bridge maps each delta to a client-visible + // response.function_call_arguments.delta frame, so withholding fragments until + // block close would leave a started call showing empty arguments. A copy is kept + // to validate the assembled payload at content_block_stop. + currentToolCallJson += delta.partial_json; yield { type: "tool_call_delta", arguments: delta.partial_json }; } break; } case "content_block_stop": { if (currentBlockType === "tool_use") { + // The non-stream path repairs an unparseable payload in toolUseArguments(); the + // stream cannot, because the fragments are already downstream. Fail the turn + // instead of ending a tool call whose arguments will not parse — the bridge's + // terminal-error path cancels the open call (status incomplete) rather than + // completing it before response.failed (#765). + if (!streamedToolArgumentsParse(currentToolCallJson)) { + yield { + type: "error", + message: "Anthropic stream sent malformed tool_use arguments (invalid JSON)", + }; + return; + } yield { type: "tool_call_end" }; currentToolCallId = ""; + currentToolCallJson = ""; } currentBlockType = ""; break; diff --git a/src/bridge.ts b/src/bridge.ts index 6bfad291d..3a8eb0baa 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -63,6 +63,23 @@ function responseError(status: number, type: string, message: string): OcxErrorP return classifyError(status, type, message); } +/** + * Whether assembled function-call arguments are usable JSON. + * An empty buffer is valid (no-arg tools send no deltas). Non-empty must parse — + * once fragments have been streamed to the client they cannot be repaired the way + * non-stream adapters degrade a bad payload to `{}`. + */ +function toolCallArgumentsUsable(args: string): boolean { + const trimmed = args.trim(); + if (!trimmed) return true; + try { + JSON.parse(trimmed); + return true; + } catch { + return false; + } +} + function adapterFailureFromEvent(event: Extract): { httpStatus: number; error: OcxErrorPayload } { if (event.status === undefined && event.errorType === undefined && event.code === undefined) { return adapterFailureFromMessage(event.message); @@ -417,6 +434,39 @@ export function bridgeToResponsesSSE( currentToolCall = null; }; + // Terminal-error / incomplete path for an open tool call (#765 remainder). + // Closing via closeCurrentToolCall() would emit function_call_arguments.done and + // status:"completed" BEFORE response.failed — the client still sees an issued call. + // Cancel instead: no *.done argument frames, status:"incomplete" (same pattern as an + // in-flight web_search_call closing as "failed"). Args still serialize as "{}" when + // empty so echoed items cannot poison the next turn with JSON.parse(""). + const failCurrentToolCall = () => { + if (!currentToolCall) return; + const argsStr = currentToolCall.args || "{}"; + const item = currentToolCall.toolSearch + ? { + type: "tool_search_call", id: currentToolCall.itemId, + call_id: currentToolCall.callId, execution: "client", + arguments: parseArgsObj(currentToolCall.args), status: "incomplete", + } + : currentToolCall.freeform + ? { + type: "custom_tool_call", id: currentToolCall.itemId, + call_id: currentToolCall.callId, name: currentToolCall.name, + input: freeformInput(currentToolCall.args), status: "incomplete", + } + : { + type: "function_call", id: currentToolCall.itemId, + call_id: currentToolCall.callId, name: currentToolCall.name, + arguments: argsStr, status: "incomplete", + ...(currentToolCall.namespace ? { namespace: currentToolCall.namespace } : {}), + }; + emit("response.output_item.done", { output_index: currentToolCall.outputIndex, item }); + finishedItems.push(item as OutputItem); + outputIndex++; + currentToolCall = null; + }; + // Finalize an open web-search cell. `status` is "completed" on a normal end, or "failed" when // the stream terminates (error/incomplete) while a search was still in flight, so Codex never // leaves a "Searching the web" spinner spinning forever. @@ -653,6 +703,32 @@ export function bridgeToResponsesSSE( break; } case "tool_call_end": { + // Fragments already streamed cannot be repaired. Refuse to complete a function call + // whose assembled arguments do not parse — cancel the item and fail the turn so the + // client never sees status:"completed" for unusable args (#765 stream remainder). + if ( + currentToolCall + && !currentToolCall.freeform + && !currentToolCall.toolSearch + && !toolCallArgumentsUsable(currentToolCall.args) + ) { + failCurrentToolCall(); + const failure = responseError( + 502, + "upstream_error", + "upstream stream produced malformed tool call arguments", + ); + emit("response.failed", { + response: { + ...responseSnapshot("failed", finishedItems), + error: failure, + last_error: failure, + }, + }); + reportTerminal("failed"); + terminalEvent = true; + break; + } closeCurrentToolCall(); break; } @@ -749,7 +825,7 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - if (currentToolCall) closeCurrentToolCall(); + if (currentToolCall) failCurrentToolCall(); if (currentWebSearch) closeCurrentWebSearch("failed", []); flushHiddenReasoningEnvelope(); options?.onUsage?.(event.usage); @@ -773,7 +849,7 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - if (currentToolCall) closeCurrentToolCall(); + if (currentToolCall) failCurrentToolCall(); if (currentWebSearch) closeCurrentWebSearch("failed", []); const failure = adapterFailureFromEvent(event); if (event.usage) options?.onUsage?.(event.usage); @@ -803,6 +879,7 @@ export function bridgeToResponsesSSE( } catch (err) { if (!terminated) { flushHiddenRawReasoning(); + if (currentToolCall) failCurrentToolCall(); if (currentWebSearch) closeCurrentWebSearch("failed", []); emit("response.failed", { response: { @@ -832,7 +909,7 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - if (currentToolCall) closeCurrentToolCall(); + if (currentToolCall) failCurrentToolCall(); if (currentWebSearch) closeCurrentWebSearch("failed", []); options?.onUsage?.(undefined); emit("response.incomplete", { @@ -872,7 +949,7 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - if (currentToolCall) closeCurrentToolCall(); + if (currentToolCall) failCurrentToolCall(); if (currentWebSearch) closeCurrentWebSearch("failed", []); emit("response.incomplete", { response: { @@ -1021,7 +1098,7 @@ export function buildResponseJSON( }); currentRawReasoning = ""; }; - const flushToolCall = () => { + const flushToolCall = (status: "completed" | "incomplete" = "completed") => { if (!currentToolCallId) return; const mapped = options?.toolNsMap?.get(currentToolCallName); const realName = mapped?.name ?? currentToolCallName; @@ -1032,19 +1109,19 @@ export function buildResponseJSON( output.push({ type: "tool_search_call", id: `tsc_${uuid()}`, call_id: currentToolCallId, execution: "client", - arguments: parseArgsObj(currentToolCallArgs), status: "completed", + arguments: parseArgsObj(currentToolCallArgs), status, }); } else if (freeform) { output.push({ type: "custom_tool_call", id: `ctc_${uuid()}`, call_id: currentToolCallId, name: realName, - input: freeformInput(currentToolCallArgs), status: "completed", + input: freeformInput(currentToolCallArgs), status, }); } else { output.push({ type: "function_call", id: `fc_${uuid()}`, call_id: currentToolCallId, name: realName, - arguments: currentToolCallArgs || "{}", status: "completed", + arguments: currentToolCallArgs || "{}", status, ...(ns ? { namespace: ns } : {}), }); } @@ -1110,6 +1187,23 @@ export function buildResponseJSON( currentToolCallArgs += e.arguments; break; case "tool_call_end": + if (!toolCallArgumentsUsable(currentToolCallArgs) && currentToolCallId) { + // Mirror the streaming path: refuse to complete unusable arguments. + const mapped = options?.toolNsMap?.get(currentToolCallName); + const realName = mapped?.name ?? currentToolCallName; + const toolSearch = options?.toolSearchToolNames?.has(realName) ?? false; + const freeform = !toolSearch && (options?.freeformToolNames?.has(realName) ?? false); + if (!freeform && !toolSearch) { + flushToolCall("incomplete"); + errorEvent = { + type: "error", + message: "upstream stream produced malformed tool call arguments", + status: 502, + errorType: "upstream_error", + }; + break; + } + } flushToolCall(); break; case "web_search_call_begin": @@ -1155,7 +1249,8 @@ export function buildResponseJSON( flushText(cleanDone && !errorEvent && !incompleteEvent ? "final_answer" : undefined); flushSummaryReasoning(); flushRawReasoning(); - flushToolCall(); + // Open tool call on a failed/incomplete turn must not land as status:"completed". + if (currentToolCallId) flushToolCall(errorEvent || incompleteEvent ? "incomplete" : "completed"); // A truncated turn must never be installed as replacement history: emit the // compaction item only when the turn actually completed (#422). if ( diff --git a/structure/01_runtime.md b/structure/01_runtime.md index aaa21b715..07e733b76 100644 --- a/structure/01_runtime.md +++ b/structure/01_runtime.md @@ -63,6 +63,9 @@ The bridge enforces a heartbeat stall deadline. It defaults to 300 seconds sampl invariant; sidecars keep their own clocks. On expiry the stream is closed and the upstream request cancelled. If the adapter generator ends without an explicit done/error event, the response is marked `incomplete` rather than `completed` so Codex can distinguish a clean finish from a truncated stream. +On `error` / incomplete / stall / EOF — and when assembled non-freeform tool arguments fail to parse — +an open tool call is cancelled as `status: "incomplete"` without `function_call_arguments.done`, so +the client never sees a completed call ahead of `response.failed` / `response.incomplete`. The server exposes `POST /api/stop` which restores native Codex config, stops any installed service (to prevent respawn), and exits the process. The GUI sidebar stop button calls this endpoint. diff --git a/tests/anthropic-stream-hardening.test.ts b/tests/anthropic-stream-hardening.test.ts index 71e42cddb..afc454e98 100644 --- a/tests/anthropic-stream-hardening.test.ts +++ b/tests/anthropic-stream-hardening.test.ts @@ -122,3 +122,91 @@ describe("anthropic non-stream tool_use input", () => { expect(events.some(e => e.type === "error")).toBe(false); }); }); + +describe("anthropic streamed tool argument validation", () => { + test("malformed streamed tool arguments fail the turn instead of completing the call", async () => { + // The stream forwards fragments as they arrive (the bridge turns each into a client-visible + // frame), so a malformed payload cannot be repaired the way the non-stream path repairs it. + // It must not end the call normally either: a client that already received `not json` would + // be told the tool call completed. The turn errors instead; the bridge cancels the open call. + const response = new Response([ + "event: content_block_start\n", + 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_9","name":"get_weather"}}\n\n', + "event: content_block_delta\n", + 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"not json"}}\n\n', + "event: content_block_stop\n", + 'data: {"type":"content_block_stop"}\n\n', + "event: message_stop\n", + 'data: {"type":"message_stop"}\n\n', + ].join("")); + const events = await collect(createAnthropicAdapter(provider).parseStream(response)); + expect(events.at(-1)?.type).toBe("error"); + expect(events.some(e => e.type === "tool_call_end")).toBe(false); + expect(events.some(e => e.type === "done")).toBe(false); + }); + + test("valid streamed tool arguments still arrive as incremental deltas", async () => { + // The bridge maps each adapter delta to response.function_call_arguments.delta, so the + // fragments must keep flowing as they arrive rather than being withheld until block close. + const response = new Response([ + "event: content_block_start\n", + 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_10","name":"get_weather"}}\n\n', + "event: content_block_delta\n", + 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"city\\":"}}\n\n', + "event: content_block_delta\n", + 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"\\"Paris\\"}"}}\n\n', + "event: content_block_stop\n", + 'data: {"type":"content_block_stop"}\n\n', + "event: message_stop\n", + 'data: {"type":"message_stop"}\n\n', + ].join("")); + const events = await collect(createAnthropicAdapter(provider).parseStream(response)); + expect(events.filter(e => e.type === "tool_call_delta")).toMatchObject([ + { type: "tool_call_delta", arguments: '{"city":' }, + { type: "tool_call_delta", arguments: '"Paris"}' }, + ]); + expect(events.some(e => e.type === "tool_call_end")).toBe(true); + expect(events.at(-1)?.type).toBe("done"); + }); + + test("two tool_use blocks in one message keep separate argument buffers", async () => { + const response = new Response([ + "event: content_block_start\n", + 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_a","name":"one"}}\n\n', + "event: content_block_delta\n", + 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"one\\":1}"}}\n\n', + "event: content_block_stop\n", + 'data: {"type":"content_block_stop"}\n\n', + "event: content_block_start\n", + 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_b","name":"two"}}\n\n', + "event: content_block_delta\n", + 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"two\\":2}"}}\n\n', + "event: content_block_stop\n", + 'data: {"type":"content_block_stop"}\n\n', + "event: message_stop\n", + 'data: {"type":"message_stop"}\n\n', + ].join("")); + const events = await collect(createAnthropicAdapter(provider).parseStream(response)); + expect(events.filter(e => e.type === "tool_call_delta")).toMatchObject([ + { type: "tool_call_delta", arguments: '{"one":1}' }, + { type: "tool_call_delta", arguments: '{"two":2}' }, + ]); + expect(events.filter(e => e.type === "tool_call_end")).toHaveLength(2); + expect(events.at(-1)?.type).toBe("done"); + }); + + test("a tool_use block with no argument fragments is not treated as malformed", async () => { + const response = new Response([ + "event: content_block_start\n", + 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_c","name":"now"}}\n\n', + "event: content_block_stop\n", + 'data: {"type":"content_block_stop"}\n\n', + "event: message_stop\n", + 'data: {"type":"message_stop"}\n\n', + ].join("")); + const events = await collect(createAnthropicAdapter(provider).parseStream(response)); + expect(events.some(e => e.type === "error")).toBe(false); + expect(events.some(e => e.type === "tool_call_end")).toBe(true); + expect(events.at(-1)?.type).toBe("done"); + }); +}); diff --git a/tests/responses-stream-tool-events.test.ts b/tests/responses-stream-tool-events.test.ts index dcbcb80b3..5fe63dd3b 100644 --- a/tests/responses-stream-tool-events.test.ts +++ b/tests/responses-stream-tool-events.test.ts @@ -51,4 +51,47 @@ describe("Responses streaming tool event contract", () => { status: "completed", }); }); + + test("terminal error cancels an open tool call instead of completing it", async () => { + // #765 remainder: adapter error with an open tool call must not emit + // function_call_arguments.done / status:"completed" before response.failed. + const frames = await collectSse(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "call_bad", name: "get_weather" }, + { type: "tool_call_delta", arguments: "not json" }, + { type: "error", message: "Anthropic stream sent malformed tool_use arguments (invalid JSON)" }, + ]), "routed/model")); + + expect(frames.some(frame => frame.event === "response.function_call_arguments.done")).toBe(false); + const itemDone = frames.filter(frame => frame.event === "response.output_item.done") + .map(frame => frame.data.item as Record) + .find(item => item?.type === "function_call"); + expect(itemDone).toMatchObject({ + type: "function_call", + call_id: "call_bad", + status: "incomplete", + }); + const failed = frames.find(frame => frame.event === "response.failed"); + expect(failed).toBeTruthy(); + const failedOutput = (failed?.data.response as Record).output as Record[]; + expect(failedOutput.some(item => item.type === "function_call" && item.status === "completed")).toBe(false); + expect(frames.some(frame => frame.event === "response.completed")).toBe(false); + }); + + test("malformed assembled arguments at tool_call_end fail the turn without completing", async () => { + const frames = await collectSse(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "call_1", name: "get_weather" }, + { type: "tool_call_delta", arguments: "not json" }, + { type: "tool_call_end" }, + { type: "done" }, + ]), "routed/model")); + + expect(frames.some(frame => frame.event === "response.function_call_arguments.done")).toBe(false); + expect(frames.some(frame => frame.event === "response.completed")).toBe(false); + const failed = frames.find(frame => frame.event === "response.failed"); + expect(failed).toBeTruthy(); + const itemDone = frames.filter(frame => frame.event === "response.output_item.done") + .map(frame => frame.data.item as Record) + .find(item => item?.type === "function_call"); + expect(itemDone).toMatchObject({ status: "incomplete" }); + }); });