fix(runner): prevent OpenCode tool starvation behind streamed text

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts

@@ -20,6 +20,7 @@ import {
   shouldForwardOpenCodeProxyItem,
 } from "./opencode-proxy-events.js";
 import { enqueueOpenCodeProxyInput, openCodeProxyCompletionFeedback } from "./opencode-proxy-input.js";
+import { createOpenCodeProxyOutput } from "./opencode-proxy-output.js";
 import {
   assertOpenCodeProxyCollaborationMode,
   openCodeProxyCollaborationModes,
@@ -58,9 +59,10 @@ const announcedTurnIds = new Set<string>();
 const launchBinding = trustedOpenCodeLaunchBinding(process.argv.slice(2));
 const usageLedger = new OpenCodeProxyUsageLedger();
 
-function send(value: unknown): void {
+const output = createOpenCodeProxyOutput(value => {
   process.stdout.write(`${JSON.stringify(value)}\n`);
-}
+});
+const send = output.send;
 
 function requestController(method: string, params: unknown, retainEnvelope = false): Promise<unknown> {
   if (controllerInputClosed) return Promise.reject(new Error("OpenCode controller input is closed"));
@@ -492,6 +494,7 @@ function shutdown(exitCode = 0): Promise<void> {
       ?.close({ reason: "proxy_shutdown", force: true })
       .catch(() => {});
     await eventPump?.catch(() => {});
+    output.flush();
     // A failed turn can be the last frame before a fatal exit. Pipe writes
     // are asynchronous on some platforms, so let it reach runnerd first.
     await new Promise<void>((resolve) => process.stdout.write("", () => resolve()));

packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts

@@ -0,0 +1,59 @@
+import { afterEach, expect, it, vi } from "vitest";
+import { createOpenCodeProxyOutput } from "./opencode-proxy-output.js";
+
+afterEach(() => vi.useRealTimers());
+
+function delta(text: string, itemId = "answer", turnId = "turn") {
+  return { method: "item/agentMessage/delta", params: {
+    threadId: "thread", turnId, itemId, kind: "agentMessage", text, delta: text,
+    item: { id: itemId, type: "agentMessage", text },
+  } };
+}
+
+it("retains every streamed byte and flushes before a tool request or terminal", () => {
+  vi.useFakeTimers();
+  const write = vi.fn();
+  const output = createOpenCodeProxyOutput(write);
+  for (const text of ["A ", "流", "ed answer."]) output.send(delta(text));
+  expect(write).not.toHaveBeenCalled();
+  const tool = { id: "tool-1", method: "item/tool/call", params: { tool: "paperclip_finish" } };
+  output.send(tool);
+  expect(write.mock.calls.map(([value]) => value)).toEqual([delta("A 流ed answer."), tool]);
+  output.send(delta("Final."));
+  const terminal = { method: "turn/completed", params: { turnId: "turn" } };
+  output.send(terminal);
+  expect(write.mock.calls.slice(2).map(([value]) => value)).toEqual([delta("Final."), terminal]);
+  vi.runAllTimers();
+  expect(write).toHaveBeenCalledTimes(4);
+});
+
+it("keeps item, turn, channel and reasoning boundaries in order", () => {
+  vi.useFakeTimers();
+  const write = vi.fn();
+  const output = createOpenCodeProxyOutput(write);
+  const frames = [delta("First"), delta("Second", "other"), delta("Next", "other", "next-turn"),
+    { method: "item/delta", params: { kind: "reasoning", text: "Private reasoning" } },
+    { ...delta("Progress"), params: { ...delta("Progress").params, channel: "progress" } },
+    { ...delta("Final"), params: { ...delta("Final").params, channel: "final" } }];
+  frames.forEach(frame => output.send(frame));
+  output.flush();
+  expect(write.mock.calls.map(([value]) => value)).toEqual(frames);
+});
+
+it("bounds streaming latency, UTF-8 buffer size and shutdown retention", () => {
+  vi.useFakeTimers();
+  const write = vi.fn();
+  const output = createOpenCodeProxyOutput(write);
+  output.send(delta("Visible"));
+  vi.advanceTimersByTime(50);
+  expect(write).toHaveBeenLastCalledWith(delta("Visible"));
+  const chunk = "流".repeat(900);
+  output.send(delta(chunk));
+  output.send(delta(chunk));
+  expect(write).toHaveBeenCalledTimes(2);
+  output.flush();
+  expect(write.mock.calls.slice(1).map(([value]) => value.params.delta).join("")).toBe(chunk + chunk);
+  expect(write).toHaveBeenCalledTimes(3);
+  vi.runAllTimers();
+  expect(write).toHaveBeenCalledTimes(3);
+});

packages/paperclip-runner/src/cli/opencode-proxy-output.ts

@@ -0,0 +1,64 @@
+// Runner normalizes each text delta with a 4,000-character limit. A byte cap
+// also stays inside that limit for Unicode without truncating a merged delta.
+const MAX_TEXT_BYTES = 4_000;
+const FLUSH_INTERVAL_MS = 50;
+
+function record(value: unknown): Record<string, unknown> | null {
+  return value !== null && typeof value === "object" && !Array.isArray(value)
+    ? value as Record<string, unknown> : null;
+}
+
+function streamedText(value: unknown) {
+  const frame = record(value);
+  const params = record(frame?.params);
+  const item = record(params?.item);
+  if (frame?.method !== "item/agentMessage/delta" || frame.id !== undefined || !params || !item
+    || ![params.threadId, params.turnId, params.itemId].every(id => typeof id === "string" && id.length > 0)
+    || !["agentMessage", "text"].includes(String(params.kind))
+    || item.type !== "agentMessage" || typeof params.delta !== "string"
+    || params.text !== params.delta || item.text !== params.delta) return null;
+  // Compare every non-text field so batching never changes item, turn,
+  // channel, or any other metadata boundary.
+  const key = JSON.stringify({ ...frame, params: { ...params, text: null, delta: null, item: { ...item, text: null } } });
+  return { frame, params, item, key, text: params.delta };
+}
+
+/** Avoid making an MCP call wait behind a durable write for every text token. */
+export function createOpenCodeProxyOutput(write: (frame: unknown) => void) {
+  let pending: ReturnType<typeof streamedText> = null;
+  let bytes = 0;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const flush = () => {
+    if (timer) clearTimeout(timer);
+    timer = undefined;
+    const current = pending;
+    pending = null;
+    bytes = 0;
+    if (current) write({ ...current.frame, params: { ...current.params,
+      text: current.text, delta: current.text, item: { ...current.item, text: current.text } } });
+  };
+  return {
+    send(frame: unknown): void {
+      const text = streamedText(frame);
+      if (!text) {
+        flush();
+        write(frame);
+        return;
+      }
+      const size = Buffer.byteLength(text.text);
+      if (pending && (pending.key !== text.key || bytes + size > MAX_TEXT_BYTES)) flush();
+      if (size > MAX_TEXT_BYTES) {
+        write(frame);
+        return;
+      }
+      if (pending) pending.text += text.text;
+      else pending = text;
+      bytes += size;
+      if (!timer) {
+        timer = setTimeout(flush, FLUSH_INTERVAL_MS);
+        timer.unref();
+      }
+    },
+    flush,
+  };
+}

packages/paperclip-runner/src/live/runnerd-opencode-streaming-tools.test.ts

@@ -0,0 +1,82 @@
+import { createHash } from "node:crypto";
+import { execFileSync } from "node:child_process";
+import { existsSync, readFileSync } from "node:fs";
+import { chmod, cp, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { expect, it } from "vitest";
+import { codexSemanticToolSpecs, dynamicToolResponse } from "../drivers/codex/codex-driver-values.js";
+import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js";
+
+const runner = defaultCapabilityRunnerdBinary();
+const digest = (path: string) => `sha256:${createHash("sha256").update(readFileSync(path)).digest("hex")}`;
+
+it.skipIf(process.platform === "win32" || !existsSync(runner)).each(["discovery", "completion"])(
+  "acknowledges %s before its MCP deadline after a burst of streamed text",
+  async tool => {
+    const root = await mkdtemp(join(tmpdir(), "runnerd-opencode-streaming-"));
+    try {
+      const workspace = join(root, "workspace");
+      const executable = join(root, "fake-opencode");
+      await mkdir(workspace);
+      const providerNode = process.platform === "linux" ? join(root, "node") : process.execPath;
+      if (process.platform === "linux") {
+        await cp(process.execPath, providerNode, { dereference: true });
+        await chmod(providerNode, 0o500);
+      }
+      const fixture = resolve("test/fixtures/fake-opencode-server.mjs");
+      execFileSync("cc", ["-x", "c", "-o", executable, "-"], {
+        input: `#include <unistd.h>\n#include <stdlib.h>\nint main(int argc, char **argv) { char **args = calloc(argc + 2, sizeof(char *)); args[0] = ${JSON.stringify(providerNode)}; args[1] = ${JSON.stringify(fixture)}; for (int i = 1; i < argc; i++) args[i + 1] = argv[i]; execv(args[0], args); return 127; }`,
+      });
+      await chmod(executable, 0o755);
+      const proxy = join(root, "opencode-app-server-proxy.cjs");
+      const proxyBytes = execFileSync(process.execPath, ["--input-type=module", "-e", `
+        import { bundleVerifiedProviderEntrypoints } from "./scripts/build-verified-provider-entrypoints.mjs";
+        const entries = await bundleVerifiedProviderEntrypoints({ write: false });
+        process.stdout.write(entries.find(({ entrypoint }) => entrypoint.name === "opencode-app-server-proxy").verifiedResult.outputFiles[0].contents);
+      `], { maxBuffer: 16 * 1024 * 1024 });
+      await writeFile(proxy, proxyBytes, { mode: 0o755 });
+      const seen: string[] = [];
+      const text: string[] = [];
+      const { transport } = createRunnerdCodexTransport({
+        provider: "opencode", runnerBinary: runner, stateDirectory: join(root, "state"), sourceCodexHome: "",
+        providerNodeCommand: providerNode, providerNodeCommandSha256: digest(providerNode),
+        opencodeCommand: executable, opencodeCommandSha256: digest(executable),
+        opencodeProxyPath: proxy, opencodeProxySha256: digest(proxy),
+        environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "synthetic-fixture-key" },
+      });
+      transport.setServerRequestHandler(async request => {
+        seen.push(String(request.params.tool));
+        return dynamicToolResponse(request.params.tool === "paperclip_finish" ? "Accepted completion." : { tools: [] });
+      });
+      let completed: Promise<void> | undefined;
+      try {
+        await transport.request("thread/start", {
+          cwd: workspace, model: "openrouter/deepseek/deepseek-v4.1-flash-1017",
+          completionContract: { revision: "streaming-tools-v1", criterionIds: ["objective"] },
+          dynamicTools: [{ name: "paperclip_search_assigned_tools", description: "Search assigned synthetic tools.", inputSchema: { type: "object" } }, ...codexSemanticToolSpecs()],
+        });
+        completed = (async () => {
+          for await (const event of transport.notifications()) {
+            if (event.method === "item/agentMessage/delta") text.push(String(event.params.delta));
+            if (event.method === "turn/completed") return;
+          }
+        })();
+        await transport.request("turn/start", { input: [{ type: "text", text: `stream-backlog-${tool}` }] });
+        await expect.poll(async () => {
+          try { return JSON.parse(await readFile(join(workspace, "fake-stream-backlog-receipt.json"), "utf8")); }
+          catch { return null; }
+        }, { timeout: 15_000 }).toMatchObject({ acknowledged: true });
+        await completed;
+        expect(seen).toEqual(tool === "discovery" ? ["paperclip_search_assigned_tools", "paperclip_finish"] : ["paperclip_finish"]);
+        expect(text.join("")).toBe("Streaming progress. ".repeat(800));
+      } finally {
+        await transport.close();
+        await completed;
+      }
+    } finally {
+      await rm(root, { recursive: true, force: true });
+    }
+  },
+  90_000,
+);

packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs

@@ -170,8 +170,9 @@ function parsedPromptText(promptBody) {
   }
 }
 
-async function mcpRequest(method, params) {
+async function mcpRequest(method, params, signal) {
   const response = await fetch(mcp.url, {
+    signal,
     method: "POST",
     headers: {
       ...mcp.headers,
@@ -215,7 +216,7 @@ async function callFirstPaperclipTool() {
   );
 }
 
-async function callTerminalTool(promptBody) {
+async function callTerminalTool(promptBody, signal) {
   const prompt = parsedPromptText(promptBody);
   const blocked = String(prompt.message ?? "").includes("block-result");
   const monitored = String(prompt.message ?? "").includes("monitor-default");
@@ -273,7 +274,7 @@ async function callTerminalTool(promptBody) {
   const call = () => mcpRequest("tools/call", {
     name: blocked ? "paperclip_block" : "paperclip_finish",
     arguments: result,
-  });
+  }, signal);
   const first = await call();
   if (String(prompt.message ?? prompt.task?.prompt ?? "").includes("completion-feedback")) {
     const outcomes = [first];
@@ -477,8 +478,31 @@ const server = createServer(async (request, response) => {
         }
       }
       setTimeout(async () => {
-        await callFirstPaperclipTool();
         const parsedPrompt = parsedPromptText(promptPayload);
+        const backlogTool = String(parsedPrompt.message ?? "").match(/stream-backlog-(discovery|completion)/)?.[1];
+        if (backlogTool) {
+          const messageID = "message-stream-backlog";
+          const partID = "part-stream-backlog";
+          emit({ type: "message.updated", properties: { sessionID: session.id, info: { id: messageID, sessionID: session.id, role: "assistant" } } });
+          emit({ type: "message.part.updated", properties: { sessionID: session.id, part: { id: partID, messageID, sessionID: session.id, type: "text", text: "" } } });
+          for (let i = 0; i < 800; i++) emit({ type: "message.part.delta", properties: {
+            sessionID: session.id, messageID, partID, field: "text", delta: "Streaming progress. ",
+          } });
+          let receipt;
+          try {
+            const result = backlogTool === "completion"
+              ? await callTerminalTool(promptPayload, AbortSignal.timeout(5000))
+              : await mcpRequest("tools/call", { name: "paperclip_search_assigned_tools", arguments: {} }, AbortSignal.timeout(5000));
+            receipt = { acknowledged: result.result?.isError !== true && !!result.result, result };
+          } catch (error) {
+            receipt = { acknowledged: false, timedOut: error.name === "TimeoutError" };
+          }
+          await writeFile(join(process.cwd(), "fake-stream-backlog-receipt.json"), JSON.stringify(receipt));
+          if (backlogTool === "discovery" && receipt.acknowledged) await callTerminalTool(promptPayload);
+          emit({ type: "session.idle", properties: { sessionID: session.id } });
+          return;
+        }
+        await callFirstPaperclipTool();
         if (String(parsedPrompt.message ?? "").includes("managed-github-operation")) {
           const result = await new Promise(resolve => {
             execFile("gh", ["api", "user"], { timeout: 5000 }, (error, stdout, stderr) => {