Fail the turn instead of dispatching truncated tool calls (#938)

* Fail the turn instead of dispatching truncated tool calls

* Forward Gemini finishReason to usage and correct truncation docs

Author: TheGreatAxios

src/plugins/path-escape-plugin.ts

@@ -28,14 +28,6 @@ export function pathEscapePlugin(
 ): ToolPlugin {
   return {
     middleware: (next) => async (call, signal) => {
-      if ("_raw" in call.arguments) {
-        return {
-          callId: call.id,
-          content:
-            "Tool call arguments were malformed JSON (likely truncated). Retry with a smaller payload.",
-          isError: true,
-        };
-      }
       let escaped: Record<string, unknown>;
       try {
         escaped = escapeArgs(

vendor/intx-inference/PATCHES.md

@@ -406,6 +406,47 @@ gap rather than carrying indefinitely.
 the cast can be deleted from the vendored file.
 **Re-carry:** clean three-way at `0205b07b`, zero conflicts. Low risk.
 
+## inference-ts-cl-7783-truncated-tool-call
+
+End-of-stream finalization in `harness.ts` (`finalizeStreamTurn`) never
+dispatches a tool call whose arguments do not parse as a normal call. The
+prior code fell back to a `{ _raw: <partial JSON> }` tool_call block, which
+the reactor dispatched — executing a tool with truncated arguments when the
+model was cut off by `max_tokens` (CL-7783: a truncated `Bash`
+`rm -rf /tm…` fragment reached the executor). Now: when `stopReason` is
+`max_tokens` and a tool call is still open, the turn fails with an
+`inference.error` naming the tool and the truncated prefix, advising a
+larger budget or a narrower scope for the model's next attempt; any other
+unparseable-args case fails the same way as invalid JSON. Both errors carry
+category `retryable`, but end-of-stream finalization always runs after the
+attempt has committed visible output, so the harness commitment boundary
+suppresses the mechanical retry — the failure is terminal for the turn, and
+the message is guidance for the next attempt rather than a re-issued retry.
+Two supporting changes: `providers/anthropic.ts` parses
+`stop_reason` out of `MessageDelta` (previously stripped by the schema) and
+surfaces it on `inference.usage`, and `vendor/intx-types`' `InferenceUsageEvent`
+gains the optional `stopReason` field both halves flow through. Guarded by the
+CL-7783 regression suite in `providers/anthropic.test.ts`, which drives the
+exact incident wire sequence and asserts no `tool_call` block reaches the
+reactor. The OpenAI-compatible adapter was audited for the same path: it has
+no adapter-local args fallback (the harness was the only dispatch site) but
+still drops `finish_reason` on both paths, so OpenAI streams get the generic
+invalid-JSON failure rather than the truncation-specific message. The Gemini
+adapter forwards its terminal `finishReason` onto `inference.usage` (same
+spread idiom as Anthropic), but forwards the provider's raw spelling
+(`MAX_TOKENS`), which the harness `max_tokens` comparison does not match —
+so Gemini truncation still lands on the generic invalid-JSON failure until
+the harness normalizes provider spellings.
+
+**Disposition:** Promotion candidate. Safety/correctness fix — prevents
+executing tools with truncated arguments after a `max_tokens` cutoff.
+**Removal path:** Upstream PR (a) surfacing `stop_reason`/`finish_reason` on
+usage events and (b) failing the turn instead of dispatching unparseable tool
+calls at end-of-stream finalization. **Re-carry:** localized to
+`finalizeStreamTurn`, the `MessageDelta` schema, and one optional event
+field; re-applies against upstream `harness.ts`/`anthropic.ts` unless the
+finalization path is reworked.
+
 ---
 
 ## Upstream promotion ledger
@@ -428,6 +469,7 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`).
 | reactor-ts-after-checkpoint-director-only | Gate `afterCheckpoint` on `hasOverride` so auto-commits do not emit it | Alexander Guy <alexander.guy@pm.me> | This ledger (#reactor-ts-after-checkpoint-director-only) | Next vendored sync |
 | sse-ts-max-line-length | Cap the unterminated SSE line buffer (`MAX_LINE_LENGTH`, 16 MiB) | Alexander Guy <alexander.guy@pm.me> | This ledger (#sse-ts-max-line-length) | Next vendored sync |
 | state-ts-deep-freeze-turns-revision | Make `ReactorState.snapshot().turns` a lazy, revision-tracked getter | Alexander Guy <alexander.guy@pm.me> | This ledger (#state-ts-deep-freeze-turns-revision) | Next vendored sync |
+| inference-ts-cl-7783-truncated-tool-call | Surface `stop_reason`/`finish_reason` on usage events; fail the turn instead of dispatching unparseable tool calls at end-of-stream finalization | Alexander Guy <alexander.guy@pm.me> | This ledger (#inference-ts-cl-7783-truncated-tool-call) | Next vendored sync |
 | google-genai-files-ts-body-init-cast | Widen `BodyInit` to accept Node's `Uint8Array` typing so the cast can be removed | Alexander Guy <alexander.guy@pm.me> | This ledger (#google-genai-files-ts-body-init-cast) | Next vendored sync |
 
 Contact basis: identified from the read-only upstream clone

vendor/intx-inference/src/harness.ts

@@ -316,6 +316,8 @@ async function* runSingleAttempt(
   // capture). Appended to the finalized turn after indexed blocks.
   const unindexedSafetyRatings: SafetyRatingBlock[] = [];
   let usageSeen: TokenUsage | null = null;
+  // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call
+  let stopReason: string | undefined;
 
   // Tool call state: keyed by callId (or index for OpenAI).
   type ToolCallState = {
@@ -1058,10 +1060,18 @@ async function* runSingleAttempt(
               // synthesizes its own descriptor cannot drift from the
               // call-start identity the rest of the harness commits to.
               usageSeen = mergeUsage(usageSeen, raw.data.usage);
+              // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call
+              if (raw.data.stopReason !== undefined) {
+                stopReason = raw.data.stopReason;
+              }
               yield {
                 type: "inference.usage",
                 seq: nextSeq(),
-                data: { usage: usageSeen, source: lastCycleSource },
+                data: {
+                  usage: usageSeen,
+                  ...(stopReason === undefined ? {} : { stopReason }),
+                  source: lastCycleSource,
+                },
               };
               break;
             }
@@ -1110,18 +1120,70 @@ async function* runSingleAttempt(
     }
 
     // Finalize any open tool calls that never received an explicit end event.
-    const completedToolCalls: ContentBlock[] = [];
+    // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call:
+    // validate every open call before emitting any of them, and never
+    // dispatch a call whose arguments are incomplete or unparseable. A turn
+    // cut at max_tokens with calls still open is unambiguous truncation;
+    // anything else unparseable is still not a normal call. Both yield an
+    // inference.error (category retryable) naming the call; post-commit the
+    // harness surfaces it terminally rather than mechanically retrying, so
+    // the message guides the model's next attempt.
+    const finalizedToolCalls: {
+      tc: ToolCallState;
+      parsedArgs: Record<string, unknown>;
+    }[] = [];
     for (const tc of openToolCalls.values()) {
-      let parsedArgs: Record<string, unknown>;
+      if (stopReason === "max_tokens") {
+        yield {
+          type: "inference.error",
+          seq: nextSeq(),
+          data: {
+            error: {
+              category: "retryable",
+              message:
+                `Tool call '${tc.name}' (${tc.callId}) was not executed: the provider ended the turn ` +
+                `at max_tokens while its arguments were still streaming (truncated input). ` +
+                `Retry the turn with a larger max_tokens budget or a smaller request so the full tool call fits.`,
+            },
+            partial: snapshotPartial(partial),
+          },
+        };
+        return;
+      }
+      let parsed: unknown;
       try {
         const raw = tc.argsBuffer.trim() === "" ? "{}" : tc.argsBuffer;
-        const parsed = JSON.parse(raw);
-        const validated = ParsedToolArgs(parsed);
-        parsedArgs = validated instanceof type.errors ? {} : validated;
+        parsed = JSON.parse(raw);
       } catch {
-        parsedArgs = { _raw: tc.argsBuffer };
+        const tail =
+          tc.argsBuffer.length > 200
+            ? `…${tc.argsBuffer.slice(-200)}`
+            : tc.argsBuffer;
+        yield {
+          type: "inference.error",
+          seq: nextSeq(),
+          data: {
+            error: {
+              category: "retryable",
+              message:
+                `Tool call '${tc.name}' (${tc.callId}) was not executed: its streamed arguments are not ` +
+                `valid JSON and cannot be dispatched as a normal call. Re-issue the turn; ` +
+                `partial argument text ends with: ${JSON.stringify(tail)}.`,
+            },
+            partial: snapshotPartial(partial),
+          },
+        };
+        return;
       }
+      const validated = ParsedToolArgs(parsed);
+      finalizedToolCalls.push({
+        tc,
+        parsedArgs: validated instanceof type.errors ? {} : validated,
+      });
+    }
 
+    const completedToolCalls: ContentBlock[] = [];
+    for (const { tc, parsedArgs } of finalizedToolCalls) {
       completedToolCalls.push({
         type: "tool_call",
         id: tc.callId,

vendor/intx-inference/src/providers/anthropic.test.ts

@@ -1260,3 +1260,120 @@ describe("createAnthropicAdapter — streaming vs non-streaming parity", () => {
     expect(jdone?.data.usage).toEqual(sdone?.data.usage);
   });
 });
+
+describe("CL-7783 truncated tool_use", () => {
+  // The exact incident wire sequence: a tool_use block opens, one partial
+  // input_json_delta arrives, then message_delta reports stop_reason
+  // max_tokens and the stream stops — no content_block_stop ever closes
+  // the tool block, so its arguments are unparseable by construction.
+  const TRUNCATED_STREAM = sse([
+    {
+      type: "content_block_start",
+      index: 0,
+      content_block: { type: "tool_use", id: "toolu_trunc", name: "Bash" },
+    },
+    {
+      type: "content_block_delta",
+      index: 0,
+      delta: {
+        type: "input_json_delta",
+        partial_json: '{"command":"rm -rf /tm',
+      },
+    },
+    {
+      type: "message_delta",
+      delta: { stop_reason: "max_tokens" },
+      usage: { output_tokens: 12 },
+    },
+    { type: "message_stop" },
+  ]);
+
+  function errorEvents(events: InferenceEvent[]) {
+    return events.filter(
+      (e): e is Extract<InferenceEvent, { type: "inference.error" }> =>
+        e.type === "inference.error",
+    );
+  }
+
+  function usageEvents(events: InferenceEvent[]) {
+    return events.filter(
+      (e): e is Extract<InferenceEvent, { type: "inference.usage" }> =>
+        e.type === "inference.usage",
+    );
+  }
+
+  test("message_delta stop_reason surfaces on the usage event", async () => {
+    const { events } = await driveTurn(TRUNCATED_STREAM, "text/event-stream");
+    const usage = usageEvents(events);
+    expect(usage.length).toBeGreaterThan(0);
+    expect(usage[usage.length - 1]?.data.stopReason).toBe("max_tokens");
+  });
+
+  test("truncated call fails the turn retryably; no tool_call is dispatched", async () => {
+    const { turn, events } = await driveTurn(
+      TRUNCATED_STREAM,
+      "text/event-stream",
+    );
+    expect(turn).toBeUndefined();
+    expect(events.some((e) => e.type === "inference.done")).toBe(false);
+    expect(
+      events.some((e) => e.type === "inference.tool_call.end"),
+    ).toBe(false);
+    const errors = errorEvents(events);
+    expect(errors).toHaveLength(1);
+    expect(errors[0]?.data.error.category).toBe("retryable");
+    expect(errors[0]?.data.error.message).toContain("max_tokens");
+    expect(errors[0]?.data.error.message).toContain("Bash");
+    expect(errors[0]?.data.error.message).toContain("not executed");
+  });
+
+  test("unparseable args with a non-truncation stop reason still never dispatch", async () => {
+    const body = sse([
+      {
+        type: "content_block_start",
+        index: 0,
+        content_block: { type: "tool_use", id: "toolu_bad", name: "Bash" },
+      },
+      {
+        type: "content_block_delta",
+        index: 0,
+        delta: {
+          type: "input_json_delta",
+          partial_json: '{"command":',
+        },
+      },
+      {
+        type: "message_delta",
+        delta: { stop_reason: "end_turn" },
+        usage: { output_tokens: 12 },
+      },
+      { type: "message_stop" },
+    ]);
+    const { turn, events } = await driveTurn(body, "text/event-stream");
+    expect(turn).toBeUndefined();
+    expect(events.some((e) => e.type === "inference.done")).toBe(false);
+    expect(
+      events.some((e) => e.type === "inference.tool_call.end"),
+    ).toBe(false);
+    const errors = errorEvents(events);
+    expect(errors).toHaveLength(1);
+    expect(errors[0]?.data.error.category).toBe("retryable");
+    expect(errors[0]?.data.error.message).toContain("not valid JSON");
+  });
+
+  test("non-streaming message surfaces top-level stop_reason on usage", async () => {
+    const body = JSON.stringify({
+      type: "message",
+      role: "assistant",
+      model: "claude-test",
+      content: [{ type: "text", text: "Done." }],
+      stop_reason: "end_turn",
+      usage: { input_tokens: 5, output_tokens: 3 },
+    });
+    const { events } = await driveTurn(body, "application/json");
+    expect(events.some((e) => e.type === "inference.error")).toBe(false);
+    const usage = usageEvents(events);
+    expect(usage).toHaveLength(1);
+    expect(usage[0]?.data.stopReason).toBe("end_turn");
+  });
+});

vendor/intx-inference/src/providers/anthropic.ts

@@ -554,6 +554,8 @@ const ContentBlockStop = type({
 
 const MessageDelta = type({
   type: "'message_delta'",
+  // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call
+  "delta?": { "stop_reason?": "string" },
   "usage?": { "output_tokens?": "number" },
 });
 
@@ -816,11 +818,17 @@ function parseResponse(
         cacheWrite: 0,
         thinking: 0,
       };
+      // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call
+      const stopReason = event.delta?.stop_reason;
       return [
         {
           type: "inference.usage",
           seq,
-          data: { usage: inferenceUsage, source },
+          data: {
+            usage: inferenceUsage,
+            ...(stopReason === undefined ? {} : { stopReason }),
+            source,
+          },
         },
       ];
     }
@@ -869,6 +877,8 @@ const NonStreamingUsage = type({
 const NonStreamingMessage = type({
   type: "'message'",
   content: "unknown[]",
+  // Locally patched — see vendor/intx-inference/PATCHES.md#inference-ts-cl-7783-truncated-tool-call
+  "stop_reason?": "string",
   usage: NonStreamingUsage,
 });
 
@@ -1064,7 +1074,13 @@ function parseJSONResponse(
   events.push({
     type: "inference.usage",
     seq,
-    data: { usage: toInferenceUsage(message.usage), source },
+    data: {
+      usage: toInferenceUsage(message.usage),
+      ...(message.stop_reason === undefined
+        ? {}
+        : { stopReason: message.stop_reason }),
+      source,
+    },
   });
 
   return events;

vendor/intx-inference/src/providers/google-genai.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, test } from "bun:test";
+import type { LastCycleSource } from "@intx/types/runtime";
+import { createGoogleGenAIAdapter } from "./google-genai";
+
+const TEST_SOURCE: LastCycleSource = {
+  sourceId: "test-google-genai",
+  provider: "google-genai",
+  model: "test-gemini-model",
+};
+
+describe("google-genai adapter — finishReason forwarding (CL-7783)", () => {
+  test("terminal finishReason surfaces on the usage event", () => {
+    const adapter = createGoogleGenAIAdapter(TEST_SOURCE);
+    const events = adapter.parseResponse(
+      JSON.stringify({
+        candidates: [
+          {
+            content: { parts: [{ text: "partial" }], role: "model" },
+            finishReason: "MAX_TOKENS",
+            index: 0,
+          },
+        ],
+        usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 3 },
+      }),
+    );
+    const usage = events.filter((e) => e.type === "inference.usage");
+    expect(usage).toHaveLength(1);
+    expect(usage[0]?.data.stopReason).toBe("MAX_TOKENS");
+  });
+
+  test("non-terminal event without finishReason emits no usage", () => {
+    const adapter = createGoogleGenAIAdapter(TEST_SOURCE);
+    const events = adapter.parseResponse(
+      JSON.stringify({
+        candidates: [
+          {
+            content: { parts: [{ text: "partial" }], role: "model" },
+            index: 0,
+          },
+        ],
+        usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 3 },
+      }),
+    );
+    expect(events.some((e) => e.type === "inference.usage")).toBe(false);
+  });
+});