fix(anthropic): validate streamed tool arguments the same way as non-stream

The review of the batch caught a real inconsistency I introduced. Making
`toolUseArguments` degrade an unparseable string to `{}` fixed the
non-stream path only: the streaming path forwarded every `partial_json`
fragment verbatim, so one malformed relay payload became `arguments:"not
json"` through `parseStream` and `"{}"` through `parseResponse`. Two
answers to the same defect, and the streaming one is what a live client
actually hits.

Fragments are now buffered per tool block and validated once at
`content_block_stop`, through the same helper. A tool call cannot execute
before its arguments are complete, so buffering costs nothing.

Three tests, each verified by ablation rather than assumed:

- malformed streamed fragments degrade to `{}` (reverting the buffering
  fails it)
- valid fragments split mid-token reassemble intact, so buffering does
  not corrupt the normal case (also fails when reverted)
- EOF carrying `message_delta.stop_reason` but no `message_stop` settles
  with that stop reason instead of erroring

That third one closes the second review finding: the two existing EOF
tests assert the pre-existing error behavior and never send a stop
reason, so they stayed green with the new fallback reverted -- they were
not testing the change they shipped with. Disabling the fallback now
fails this test (14 pass / 1 fail) and restoring it returns 15/15.

Suites: anthropic-stream-hardening, anthropic-adapter, anthropic-hardening,
anthropic-compatible-stream, anthropic-tail-guard, anthropic-empty-content,
anthropic-thinking-signature, anthropic-reasoning, anthropic-tool-schema,
claude-messages-endpoint, anthropic-tool-reorder -- 115 pass, 0 fail.

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

Author: lidge-jun

src/adapters/anthropic.ts

@@ -765,6 +765,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
       let currentBlockType = "";
       let currentToolCallId = "";
       let currentToolCallName = "";
+      let currentToolCallJson = "";
       let pendingUsage: Record<string, number> | undefined;
       let pendingStopReason: string | undefined;
       let emittedDone = false;
@@ -808,6 +809,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 +835,21 @@ 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") {
-                  yield { type: "tool_call_delta", arguments: delta.partial_json };
+                  // 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.
+                  currentToolCallJson += delta.partial_json;
                 }
                 break;
               }
               case "content_block_stop": {
                 if (currentBlockType === "tool_use") {
+                  yield { type: "tool_call_delta", arguments: toolUseArguments(currentToolCallJson) };
                   yield { type: "tool_call_end" };
                   currentToolCallId = "";
+                  currentToolCallJson = "";
                 }
                 currentBlockType = "";
                 break;

tests/anthropic-stream-hardening.test.ts

@@ -103,4 +103,65 @@ describe("anthropic non-stream tool_use input", () => {
       arguments: "{}",
     });
   });
+
+  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.
+    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 "}}\n\n',
+      "event: content_block_delta\n",
+      'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"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.filter(e => e.type === "tool_call_delta")).toMatchObject([
+      { type: "tool_call_delta", arguments: "{}" },
+    ]);
+  });
+
+  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.
+    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":"Paris"}' },
+    ]);
+  });
+
+  test("EOF after message_delta.stop_reason settles instead of erroring", async () => {
+    // The blocker the other EOF tests could not see: they assert the pre-existing error path, so
+    // reverting the stop-reason fallback leaves them green. This drives the fallback itself -- a
+    // stream that reported why it stopped but never sent message_stop.
+    const response = new Response([
+      "event: content_block_start\n",
+      'data: {"type":"content_block_start","content_block":{"type":"text","text":""}}\n\n',
+      "event: content_block_delta\n",
+      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}}\n\n',
+      "event: message_delta\n",
+      'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n',
+    ].join(""));
+    const events = await collect(createAnthropicAdapter(provider).parseStream(response));
+    expect(events.at(-1)).toMatchObject({ type: "done", stopReason: "end_turn" });
+    expect(events.some(e => e.type === "error")).toBe(false);
+  });
 });