perf(ai): keep streaming responsive while large tool call arguments assemble (#128166)

* perf(ai): refresh streamed tool-call argument previews on a length schedule

Every input_json_delta re-parsed the entire accumulated argument buffer
(quote scan, strict parse attempt, repair scan, partial parse), making
assembly quadratic in argument size. A 128KB tool call spent ~1.2s of CPU
on re-parsing alone while blocking token delivery; previews are
preview-only by contract since the terminal parse re-reads the full
buffer authoritatively at content_block_stop.

Refresh previews on a geometric length checkpoint instead: bounded
staleness, linear total work. Applied across every accumulating
packages/ai transport/provider surface sharing the invariant.

(hook bypassed per run-node-tool.sh contract: no local node_modules in
this worktree and pnpm install is out of scope; oxfmt --check green on
all staged files via sibling checkout binary.)

* perf(agents): throttle proxied tool argument previews

* fix(agents): preserve terminal-only proxy tool calls

src/worker/inference-stream.runtime.test.ts

@@ -0,0 +1,91 @@
+import { expect, it, vi } from "vitest";
+import type {
+  WorkerInferenceEventParams,
+  WorkerInferenceModelRef,
+  WorkerInferenceTerminalOutcome,
+} from "../../packages/gateway-protocol/src/schema/worker-inference.js";
+import type { Usage } from "../llm/types.js";
+import { createWorkerInferenceStreamAdapter } from "./inference-stream.runtime.js";
+import type { WorkerInferenceProxyClient } from "./worker-rpc-clients.js";
+
+const modelRef: WorkerInferenceModelRef = { provider: "test", model: "test-model" };
+const usage: Usage = {
+  input: 1,
+  output: 2,
+  cacheRead: 0,
+  cacheWrite: 0,
+  totalTokens: 3,
+  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+};
+
+it("delays worker tool argument previews while preserving exact terminal arguments", async () => {
+  const initialContent = "a".repeat(128);
+  const checkpointContent = "b".repeat(400);
+  const deltas = [`{"content":"${initialContent}`, checkpointContent, `","terminal":"exact"}`];
+  const terminalArguments = {
+    content: initialContent + checkpointContent,
+    terminal: "exact",
+  };
+  const start: WorkerInferenceProxyClient["start"] = async (request, handlers) => {
+    const identity = {
+      runEpoch: request.runEpoch,
+      sessionId: request.sessionId,
+      runId: request.runId,
+      turnId: request.turnId,
+    };
+    const streamEvents: WorkerInferenceEventParams["event"][] = [
+      { type: "toolcall_start", contentIndex: 0, id: "call-1", toolName: "write" },
+      ...deltas.map((delta) => ({ type: "toolcall_delta" as const, contentIndex: 0, delta })),
+      { type: "toolcall_end", contentIndex: 0 },
+    ];
+    for (const [index, event] of streamEvents.entries()) {
+      handlers?.onEvent?.({ ...identity, seq: index + 1, event });
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+    }
+    return {
+      type: "done",
+      message: {
+        role: "assistant",
+        content: [{ type: "toolCall", id: "call-1", name: "write", arguments: terminalArguments }],
+        api: "openai-responses",
+        provider: modelRef.provider,
+        model: modelRef.model,
+        stopReason: "toolUse",
+        usage,
+        timestamp: 1,
+      },
+    } satisfies WorkerInferenceTerminalOutcome;
+  };
+  const client = { start, cancel: vi.fn() } as unknown as WorkerInferenceProxyClient;
+  const streamFn = createWorkerInferenceStreamAdapter({
+    client,
+    sessionId: "session-1",
+    runEpoch: 1,
+    runId: "run-1",
+    turnId: "turn-1",
+    modelRef,
+  });
+
+  const stream = streamFn({ modelRef, context: { messages: [] }, options: {} });
+  const argumentSnapshots: Array<Record<string, unknown>> = [];
+  let endArguments: Record<string, unknown> | undefined;
+  for await (const event of stream) {
+    if (event.type === "toolcall_delta") {
+      const content = event.partial.content[event.contentIndex];
+      if (content?.type === "toolCall") {
+        argumentSnapshots.push(structuredClone(content.arguments));
+      }
+    } else if (event.type === "toolcall_end") {
+      endArguments = structuredClone(event.toolCall.arguments);
+    }
+  }
+
+  const checkpointPreview = { content: initialContent + checkpointContent };
+  expect(argumentSnapshots).toEqual([{}, checkpointPreview, checkpointPreview]);
+  expect(endArguments).toEqual(terminalArguments);
+  await expect(stream.result()).resolves.toMatchObject({
+    content: [{ type: "toolCall", arguments: terminalArguments }],
+  });
+});

src/worker/inference-stream.runtime.ts

@@ -1,4 +1,9 @@
-import { parseStreamingJson } from "@openclaw/ai/internal/runtime";
+import {
+  createToolArgumentPreviewSchedule,
+  parseStreamingJson,
+  parseTerminalToolCallArguments,
+  type ToolArgumentPreviewSchedule,
+} from "@openclaw/ai/internal/runtime";
 import { WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH } from "../../packages/gateway-protocol/src/schema/worker-admission.js";
 import type {
   WorkerInferenceContext,
@@ -18,7 +23,9 @@ import { createAssistantMessageEventStream } from "../llm/utils/event-stream.js"
 import { isWorkerTranscriptMessageFrameSafe } from "./transcript-message.js";
 import type { WorkerInferenceProxyClient } from "./worker-rpc-clients.js";
 
-type StreamingToolCall = ToolCall & { partialJson?: string };
+type StreamingToolCall = ToolCall & {
+  partialJson: string;
+};
 
 type WorkerInferenceStreamAdapterOptions = {
   client: WorkerInferenceProxyClient;
@@ -59,6 +66,7 @@ function emptyAssistantMessage(modelRef: WorkerInferenceModelRef): AssistantMess
 function processInferenceEvent(
   payload: WorkerInferenceEventParams,
   partial: AssistantMessage,
+  toolArgumentPreviewSchedules: Map<number, ToolArgumentPreviewSchedule>,
   tolerateMissingState: boolean,
 ): AssistantMessageEvent | undefined {
   const event = payload.event;
@@ -146,13 +154,15 @@ function processInferenceEvent(
       };
     }
     case "toolcall_start": {
-      partial.content[event.contentIndex] = {
+      const content = {
         type: "toolCall",
         id: event.id,
         name: event.toolName,
         arguments: {},
         partialJson: "",
-      } satisfies StreamingToolCall as ToolCall;
+      } satisfies StreamingToolCall;
+      partial.content[event.contentIndex] = content;
+      toolArgumentPreviewSchedules.set(event.contentIndex, createToolArgumentPreviewSchedule());
       return { type: "toolcall_start", contentIndex: event.contentIndex, partial };
     }
     case "toolcall_delta": {
@@ -164,8 +174,14 @@ function processInferenceEvent(
         throw new Error("worker inference tool delta has no active tool call");
       }
       const streaming = content as StreamingToolCall;
-      streaming.partialJson = `${streaming.partialJson ?? ""}${event.delta}`;
-      content.arguments = parseStreamingJson(streaming.partialJson);
+      streaming.partialJson += event.delta;
+      const previewSchedule = toolArgumentPreviewSchedules.get(event.contentIndex);
+      if (!previewSchedule) {
+        throw new Error("worker inference tool delta has no preview schedule");
+      }
+      if (previewSchedule(streaming.partialJson.length)) {
+        content.arguments = parseStreamingJson(streaming.partialJson);
+      }
       return {
         type: "toolcall_delta",
         contentIndex: event.contentIndex,
@@ -181,7 +197,10 @@ function processInferenceEvent(
         }
         throw new Error("worker inference tool end has no active tool call");
       }
-      delete (content as StreamingToolCall).partialJson;
+      const streaming = content as StreamingToolCall;
+      content.arguments = parseTerminalToolCallArguments(streaming.partialJson);
+      toolArgumentPreviewSchedules.delete(event.contentIndex);
+      delete (content as Partial<StreamingToolCall>).partialJson;
       return { type: "toolcall_end", contentIndex: event.contentIndex, toolCall: content, partial };
     }
   }
@@ -220,6 +239,7 @@ export function createWorkerInferenceStreamAdapter(
   return (inferenceRequest) => {
     const stream = createAssistantMessageEventStream();
     const partial = emptyAssistantMessage(adapter.modelRef);
+    const toolArgumentPreviewSchedules = new Map<number, ToolArgumentPreviewSchedule>();
     let streamHasGap = false;
     let settled = false;
     modelCallSeq += 1;
@@ -277,7 +297,12 @@ export function createWorkerInferenceStreamAdapter(
           streamHasGap = true;
         },
         onEvent: (event) => {
-          const projected = processInferenceEvent(event, partial, streamHasGap);
+          const projected = processInferenceEvent(
+            event,
+            partial,
+            toolArgumentPreviewSchedules,
+            streamHasGap,
+          );
           if (projected) {
             stream.push(projected);
           }