fix(anthropic): keep tool argument deltas incremental and fail malformed ones

The previous commit fixed the stream/non-stream inconsistency by buffering
`input_json_delta` fragments until block close. That was the wrong repair
and the review caught it: `src/bridge.ts:628` maps every adapter delta to
a client-visible `response.function_call_arguments.delta` frame, and
`tests/responses-stream-tool-events.test.ts:30` pins that split fragments
stay split. Withholding them left a started function call showing empty
arguments until the block ended -- a protocol regression traded for a
consistency fix.

Fragments flow again as they arrive. The assembled copy is still kept, but
only to check at `content_block_stop` whether the payload parses. When it
does not, the turn errors instead of emitting `tool_call_end`: the
fragments are already downstream and cannot be repaired the way
`toolUseArguments` repairs the non-stream path, and a client that received
`not json` must not then be told the call completed normally.

`streamedToolArgumentsParse` treats an empty buffer as valid, because a
no-argument tool call sends no `input_json_delta` at all -- checking for
`"{}"` instead would have failed those and also swallowed a legitimate
empty-object argument.

Four tests, each checked by ablation rather than assumed:

- malformed fragments error the turn and emit no `tool_call_end` or `done`
  (disabling the guard: 16 pass / 1 fail)
- valid fragments arrive as two separate deltas, not one joined string --
  this is the contract the previous commit broke (removing incremental
  forwarding: 15 pass / 2 fail)
- two `tool_use` blocks in one message keep separate buffers, so the
  second cannot inherit the first's JSON
- a block with no fragments is not treated as malformed

Suites: anthropic-stream-hardening, responses-stream-tool-events,
anthropic-adapter, anthropic-compatible-stream, anthropic-hardening,
anthropic-tail-guard, claude-messages-endpoint -- 64 pass, 0 fail.

Co-authored-by: Wibias <37517432+Wibias@users.noreply.github.com>

Author: lidge-jun

src/adapters/anthropic.ts

@@ -284,6 +284,23 @@ function toolUseArguments(input: unknown): string {
   return JSON.stringify(input ?? {});
 }
 
+/**
+ * Whether the 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";
 }
@@ -835,18 +852,28 @@ 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") {
-                  // Buffered rather than forwarded per-delta so the assembled arguments can be
-                  // validated at content_block_stop, the same bar the non-stream path applies in
-                  // toolUseArguments(). A tool call cannot run before its arguments are complete,
-                  // so holding the fragments costs nothing and keeps the two paths from
-                  // disagreeing about what a malformed payload becomes.
+                  // 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") {
-                  yield { type: "tool_call_delta", arguments: toolUseArguments(currentToolCallJson) };
+                  // 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 -- a client that
+                  // received `not json` must not be told the call completed normally.
+                  if (!streamedToolArgumentsParse(currentToolCallJson)) {
+                    yield {
+                      type: "error",
+                      message: "Anthropic stream sent malformed tool_use arguments (invalid JSON)",
+                    };
+                    return;
+                  }
                   yield { type: "tool_call_end" };
                   currentToolCallId = "";
                   currentToolCallJson = "";

tests/anthropic-stream-hardening.test.ts

@@ -104,11 +104,11 @@ describe("anthropic non-stream tool_use input", () => {
     });
   });
 
-  test("streamed tool arguments are validated at block close, matching the non-stream path", async () => {
-    // The streaming path used to forward each partial_json fragment verbatim, so a relay emitting
-    // malformed JSON produced arguments:"not json" while the SAME payload through parseResponse
-    // produced "{}". Two paths, two answers for one defect. Fragments are buffered and validated
-    // once the block closes, so both now degrade identically.
+  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.
     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',
@@ -122,14 +122,15 @@ describe("anthropic non-stream tool_use input", () => {
       '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: "{}" },
-    ]);
+    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("streamed tool arguments are reassembled intact when the JSON is valid", async () => {
-    // The buffering must not corrupt the normal case: fragments split mid-token still arrive as
-    // one well-formed argument string.
+  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.
+    // Both fragments, in order, and the call ends normally.
     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',
@@ -144,8 +145,55 @@ describe("anthropic non-stream tool_use input", () => {
     ].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":"Paris"}' },
+      { 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 () => {
+    // The validation buffer is per block; a leak would make the second block inherit the first
+    // block's JSON and could fail a well-formed call.
+    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 () => {
+    // A no-argument tool call sends no input_json_delta at all. An empty buffer must stay valid.
+    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");
   });
 
   test("EOF after message_delta.stop_reason settles instead of erroring", async () => {