diff --git a/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts b/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts index b1e13e59f30..3ebbcab66b7 100644 --- a/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts +++ b/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(); 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 { if (controllerInputClosed) return Promise.reject(new Error("OpenCode controller input is closed")); @@ -492,6 +494,7 @@ function shutdown(exitCode = 0): Promise { ?.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((resolve) => process.stdout.write("", () => resolve())); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts new file mode 100644 index 00000000000..2639eb952f2 --- /dev/null +++ b/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); +}); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-output.ts b/packages/paperclip-runner/src/cli/opencode-proxy-output.ts new file mode 100644 index 00000000000..6aea73ade19 --- /dev/null +++ b/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 | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record : 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 = null; + let bytes = 0; + let timer: ReturnType | 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, + }; +} diff --git a/packages/paperclip-runner/src/live/runnerd-opencode-streaming-tools.test.ts b/packages/paperclip-runner/src/live/runnerd-opencode-streaming-tools.test.ts new file mode 100644 index 00000000000..a78dfd5b66e --- /dev/null +++ b/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 \n#include \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 | 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, +); diff --git a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs b/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs index 3b03d4d52e2..bfcbc7a32e4 100644 --- a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs +++ b/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) => {