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 dc21cb13fcd..9bdab7c2477 100644 --- a/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts +++ b/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts @@ -20,7 +20,8 @@ import { shouldForwardOpenCodeProxyItem, } from "./opencode-proxy-events.js"; import { enqueueOpenCodeProxyInput, openCodeProxyCompletionFeedback } from "./opencode-proxy-input.js"; -import { createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; +import { createBackpressuredFrameWriter, createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; +import { createOpenCodeProxyControllerRequests } from "./opencode-proxy-controller.js"; import { assertOpenCodeProxyCollaborationMode, openCodeProxyCollaborationModes, @@ -42,12 +43,6 @@ type RpcMessage = { }; const input = createInterface({ input: process.stdin, crlfDelay: Infinity }); -const pending = new Map< - string, - { resolve(value: unknown): void; reject(error: Error): void; retainEnvelope: boolean } ->(); -let controllerInputClosed = false; -let nextServerRequestId = 1; let driver: OpenCodeServerDriver | null = null; let session: HarnessSession | null = null; let eventPump: Promise | null = null; @@ -59,19 +54,15 @@ const announcedTurnIds = new Set(); const launchBinding = trustedOpenCodeLaunchBinding(process.argv.slice(2)); const usageLedger = new OpenCodeProxyUsageLedger(); -const output = createOpenCodeProxyOutput(value => { - process.stdout.write(`${JSON.stringify(value)}\n`); -}); +// runnerd reads this stream only after its controller acknowledges earlier +// events. Keep that backlog in a queue we control so an abandoned tool call +// can still be withdrawn before runnerd dispatches it. +const stdoutFrames = createBackpressuredFrameWriter(process.stdout); +const output = createOpenCodeProxyOutput((value, options) => stdoutFrames.write(value, options)); 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")); - const id = `opencode-${nextServerRequestId++}`; - send({ id, method, params }); - return new Promise((resolveValue, reject) => - pending.set(id, { resolve: resolveValue, reject, retainEnvelope }), - ); -} +const controller = createOpenCodeProxyControllerRequests(send); +const requestController = controller.request; function record(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -137,7 +128,7 @@ async function open( // Runner owns these calls until a durable result, cancellation, or lease // expiry. A shorter MCP deadline would leave an unknown effect pending. durableToolRequests: true, - completionFeedback: openCodeProxyCompletionFeedback((method, params) => requestController(method, params, true)), + completionFeedback: openCodeProxyCompletionFeedback((method, params, signal) => requestController(method, params, true, signal)), dynamicToolHandler: async (call) => requestController("item/tool/call", { threadId: session?.ids().driverSessionId ?? "opening", @@ -145,7 +136,7 @@ async function open( callId: call.callId, tool: call.tool, arguments: call.arguments, - }), + }, false, call.signal), onDiagnostic: (message) => process.stderr.write(`[opencode] ${message}\n`), // runnerd gives this proxy its own process group. Keep `opencode serve` in // that same group so runnerd's TERM/KILL fallback cannot orphan it. @@ -334,19 +325,7 @@ async function pumpEvents(opened: HarnessSession): Promise { async function handle(message: RpcMessage): Promise { if (message.method === undefined && message.id !== undefined) { - const waiter = pending.get(String(message.id)); - if (!waiter) return; - pending.delete(String(message.id)); - if (message.error !== undefined) - waiter.reject(new Error(JSON.stringify(message.error))); - else { - const contentItems = record(message.result).contentItems; - waiter.resolve( - !waiter.retainEnvelope && Array.isArray(contentItems) - ? (contentItems[0] ?? message.result) - : message.result, - ); - } + controller.respond({ id: message.id, result: message.result, error: message.error }); return; } if (!message.method || message.id === undefined) return; @@ -490,14 +469,13 @@ function failProxy(error: unknown): void { function shutdown(exitCode = 0): Promise { shutdownPromise ??= (async () => { - for (const waiter of pending.values()) - waiter.reject(new Error("OpenCode proxy is shutting down")); - pending.clear(); + controller.rejectAll(new Error("OpenCode proxy is shutting down")); await session ?.close({ reason: "proxy_shutdown", force: true }) .catch(() => {}); await eventPump?.catch(() => {}); output.flush(); + stdoutFrames.drain(); // 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())); @@ -506,10 +484,7 @@ function shutdown(exitCode = 0): Promise { } input.on("close", () => { - controllerInputClosed = true; - for (const waiter of pending.values()) - waiter.reject(new Error("OpenCode controller input closed before its response")); - pending.clear(); + controller.closeInput(new Error("OpenCode controller input closed before its response")); // Preserve command order and the bootstrap drain. A queued command cannot // create a new unanswered controller request after EOF. void pendingInput.then(() => shutdown()); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-abandoned-tool-call.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-abandoned-tool-call.test.ts new file mode 100644 index 00000000000..79343e79720 --- /dev/null +++ b/packages/paperclip-runner/src/cli/opencode-proxy-abandoned-tool-call.test.ts @@ -0,0 +1,194 @@ +import { EventEmitter } from "node:events"; +import { chmod, copyFile, mkdtemp, readdir, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; + +import { startOpenCodeMcpBridge, type OpenCodeMcpBridge, type OpenCodeMcpCall } from "../drivers/opencode/mcp-bridge.js"; +import { OpenCodeServerDriver } from "../drivers/opencode/opencode-server-driver.js"; +import { createOpenCodeProxyControllerRequests } from "./opencode-proxy-controller.js"; +import { openCodeProxyCompletionFeedback } from "./opencode-proxy-input.js"; +import { createBackpressuredFrameWriter, createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; + +/** A stdout whose reader (runnerd) is waiting on controller acknowledgements. */ +class StalledStdout extends EventEmitter { + readonly lines: string[] = []; + accepting = false; + write(chunk: string): boolean { + this.lines.push(chunk); + return this.accepting; + } + resume(): void { + this.accepting = true; + this.emit("drain"); + } + frames(): Array> { + return this.lines.map(line => JSON.parse(line) as Record); + } +} + +const bridges: OpenCodeMcpBridge[] = []; +afterEach(async () => { + await Promise.all(bridges.splice(0).map(bridge => bridge.close())); +}); + +/** + * Compose the real MCP bridge with the proxy's frame writer the way + * opencode-app-server-proxy does: a tool call becomes an item/tool/call frame + * that waits behind earlier event frames until runnerd reads stdout. + */ +async function harness() { + const stdout = new StalledStdout(); + const writer = createBackpressuredFrameWriter(stdout); + const outcomes: string[] = []; + let next = 1; + const queued = new Set(); + const handler = (call: OpenCodeMcpCall) => new Promise((_resolve, reject) => { + const id = `opencode-${next++}`; + queued.add(call.callId); + writer.write({ id, method: "item/tool/call", params: { callId: call.callId, tool: call.tool } }, { + signal: call.signal, + onAbandoned: () => { + outcomes.push(`abandoned:${call.callId}`); + reject(new Error("abandoned before dispatch")); + }, + }); + }); + const bridge = await startOpenCodeMcpBridge({ + tools: [{ name: "call_api", inputSchema: { type: "object" } }], + timeoutMs: null, + handler, + }); + bridges.push(bridge); + const post = (body: Record, signal?: AbortSignal) => fetch(bridge.url, { + method: "POST", + headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", ...body }), + signal, + }); + // OpenCode events that runnerd has not yet read sit ahead of the call. + writer.write({ method: "item/started", params: { itemId: "backlog" } }); + expect(stdout.lines).toHaveLength(1); + writer.write({ method: "item/completed", params: { itemId: "backlog" } }); + return { stdout, writer, outcomes, queued, post }; +} + +const mutatingCall = (id: string) => ({ + id, + method: "tools/call", + params: { name: "call_api", arguments: { method: "POST", path: "/tool-gateway/gateways/g/tokens" } }, +}); + +async function until(condition: () => boolean): Promise { + for (let attempt = 0; attempt < 200 && !condition(); attempt += 1) { + await new Promise(resolve => setTimeout(resolve, 5)); + } + expect(condition()).toBe(true); +} + +describe("OpenCode proxy abandoned tool calls", () => { + it("never hands runnerd a mutating call whose MCP client timed out and cancelled it", async () => { + const { stdout, outcomes, queued, post } = await harness(); + const response = post(mutatingCall("rotate-token")); + await until(() => queued.has("rotate-token")); + // The MCP SDK sends this when its request timeout fires. + const cancel = await post({ + method: "notifications/cancelled", + params: { requestId: "rotate-token", reason: "McpError: MCP error -32001: Request timed out" }, + }); + expect(cancel.status).toBe(202); + const delivered = await (await response).json() as { result: { isError: boolean } }; + expect(delivered.result.isError).toBe(true); + expect(outcomes).toEqual(["abandoned:rotate-token"]); + + // runnerd finally drains the backlog: it sees the events, never the call. + stdout.resume(); + expect(stdout.frames().map(frame => frame.method)).toEqual(["item/started", "item/completed"]); + }); + + it("never hands runnerd a call whose MCP client dropped its connection", async () => { + const { stdout, outcomes, queued, post } = await harness(); + const client = new AbortController(); + const response = post(mutatingCall("dropped"), client.signal); + await until(() => queued.has("dropped")); + client.abort(); + await expect(response).rejects.toThrow(); + await until(() => outcomes.length > 0); + expect(outcomes).toEqual(["abandoned:dropped"]); + stdout.resume(); + expect(stdout.frames().some(frame => frame.method === "item/tool/call")).toBe(false); + }); + + it("still dispatches a waiting call that its client has not abandoned", async () => { + const { stdout, outcomes, queued, post } = await harness(); + void post(mutatingCall("live")).catch(() => undefined); + await until(() => queued.has("live")); + stdout.resume(); + expect(stdout.frames().map(frame => frame.method)).toEqual(["item/started", "item/completed", "item/tool/call"]); + expect(outcomes).toEqual([]); + }); +}); + +describe("OpenCode proxy abandoned completion (driver + proxy wiring)", () => { + const cleanup: string[] = []; + afterEach(async () => { + await Promise.all(cleanup.splice(0).map(path => rm(path, { recursive: true, force: true }))); + }); + + it("withdraws a queued paperclip_finish feedback frame when its MCP client cancels", async () => { + const fixtureRoot = await mkdtemp(join(tmpdir(), "paperclip-opencode-abandon-fixture-")); + const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-abandon-")); + const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-abandon-workspace-")); + cleanup.push(fixtureRoot, root, workspace); + const fixture = join(fixtureRoot, "fake-opencode-server.mjs"); + await copyFile(resolve("test/fixtures/fake-opencode-server.mjs"), fixture); + await chmod(fixture, 0o755); + + // The proxy's real stdout path: delta batching, backpressured writer and + // controller requests, over a stdout that runnerd is not reading yet. + const stdout = new StalledStdout(); + const writer = createBackpressuredFrameWriter(stdout); + const output = createOpenCodeProxyOutput((frame, options) => writer.write(frame, options)); + const controller = createOpenCodeProxyControllerRequests(output.send); + output.send({ method: "item/started", params: { itemId: "backlog" } }); + expect(stdout.lines).toHaveLength(1); + + let queuedCallId: string | undefined; + const driver = new OpenCodeServerDriver({ + model: "openrouter/deepseek/deepseek-v4-flash-0731", runtimeDirectory: root, command: fixture, + environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" }, + completionFeedback: openCodeProxyCompletionFeedback((method, params, signal) => { + queuedCallId = String(params.callId); + return controller.request(method, params, true, signal); + }), + }); + const session = await driver.openSession({ runId: "abandon-finish", normalizedSessionId: "abandon-finish", workingDirectory: workspace }); + try { + const proposed: unknown[] = []; + void (async () => { + for await (const event of session.events()) if (event.eventType === "run.result.proposed") proposed.push(event); + })().catch(() => undefined); + await session.startTurn({ message: { role: "user", text: "finish" } }); + await until(() => queuedCallId !== undefined && writer.queuedFrames > 0); + + // The MCP client times out and cancels the call it is waiting on. + const sessionDir = (await readdir(root, { withFileTypes: true })).find(entry => entry.isDirectory())!.name; + const config = JSON.parse(await readFile(join(root, sessionDir, "config/opencode/opencode.json"), "utf8")); + const cancel = await fetch(config.mcp.paperclip.url, { + method: "POST", + headers: { ...config.mcp.paperclip.headers, "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: Number(queuedCallId), reason: "Request timed out" } }), + }); + expect(cancel.status).toBe(202); + await until(() => writer.queuedFrames === 0); + + stdout.resume(); + const completionFrames = stdout.frames().filter(frame => frame.method === "item/tool/call"); + expect(completionFrames).toEqual([]); + expect(proposed).toEqual([]); + expect((await session.snapshot()).semanticResult ?? null).toBeNull(); + } finally { + await session.close({ reason: "test complete" }); + } + }, 30_000); +}); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-completion.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-completion.test.ts index b76d5946f63..32d8692021a 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-completion.test.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-completion.test.ts @@ -10,7 +10,17 @@ describe("OpenCode controller completion feedback", () => { const feedback = "Include [Saved document](/PAP/issues/PAP-1#document-plan) in your final response."; const request = vi.fn().mockResolvedValue({ success: true, contentItems: [{ type: "inputText", text: feedback }] }); await expect(openCodeProxyCompletionFeedback(request)(result, { ...call, tool })).resolves.toBe(feedback); - expect(request).toHaveBeenCalledExactlyOnceWith("item/tool/call", { ...call, tool, arguments: result }); + expect(request).toHaveBeenCalledExactlyOnceWith("item/tool/call", { ...call, tool, arguments: result }, undefined); + }); + + it("withdraws an abandoned completion through its signal without sending the signal as data", async () => { + const client = new AbortController(); + const request = vi.fn((_method: string, _params: Record, signal?: AbortSignal) => + new Promise((_resolve, reject) => signal?.addEventListener("abort", () => reject(new Error("abandoned before dispatch"))))); + const pending = openCodeProxyCompletionFeedback(request)(result, { ...call, signal: client.signal }); + client.abort(); + await expect(pending).rejects.toThrow("abandoned before dispatch"); + expect(request).toHaveBeenCalledExactlyOnceWith("item/tool/call", { ...call, arguments: result }, client.signal); }); it("preserves the controller rejection for same-turn correction", async () => { diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-controller.ts b/packages/paperclip-runner/src/cli/opencode-proxy-controller.ts new file mode 100644 index 00000000000..2e732b83aef --- /dev/null +++ b/packages/paperclip-runner/src/cli/opencode-proxy-controller.ts @@ -0,0 +1,72 @@ +import type { OpenCodeProxyFrameOptions } from "./opencode-proxy-output.js"; + +type Waiter = { resolve(value: unknown): void; reject(error: Error): void; retainEnvelope: boolean }; + +export function abandonedToolCallError(): Error { + return new Error( + "Paperclip tool call was abandoned by the MCP client before it was dispatched; it did not run.", + ); +} + +function record(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : {}; +} + +/** + * Requests the proxy sends to its runnerd controller, and their responses. + * + * A request with a signal stays withdrawable while its frame is still queued + * in this process. Once written, it belongs to runnerd and its response is + * awaited even if the MCP client leaves. + */ +export function createOpenCodeProxyControllerRequests( + send: (frame: unknown, options?: OpenCodeProxyFrameOptions) => void, +) { + const pending = new Map(); + let inputClosed = false; + let nextId = 1; + const rejectAll = (error: Error) => { + for (const waiter of pending.values()) waiter.reject(error); + pending.clear(); + }; + return { + request(method: string, params: unknown, retainEnvelope = false, signal?: AbortSignal): Promise { + if (inputClosed) return Promise.reject(new Error("OpenCode controller input is closed")); + if (signal?.aborted) return Promise.reject(abandonedToolCallError()); + const id = `opencode-${nextId++}`; + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject, retainEnvelope }); + send({ id, method, params }, signal ? { + signal, + onAbandoned: () => { + if (pending.delete(id)) reject(abandonedToolCallError()); + }, + } : undefined); + }); + }, + /** Settles the waiter for a controller response; false when none waits. */ + respond(message: { id: string | number; result?: unknown; error?: unknown }): boolean { + const waiter = pending.get(String(message.id)); + if (!waiter) return false; + pending.delete(String(message.id)); + if (message.error !== undefined) { + waiter.reject(new Error(JSON.stringify(message.error))); + } else { + const contentItems = record(message.result).contentItems; + waiter.resolve( + !waiter.retainEnvelope && Array.isArray(contentItems) + ? (contentItems[0] ?? message.result) + : message.result, + ); + } + return true; + }, + rejectAll, + closeInput(error: Error): void { + inputClosed = true; + rejectAll(error); + }, + }; +} diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-input.ts b/packages/paperclip-runner/src/cli/opencode-proxy-input.ts index 25bd17161d7..bc234f7986f 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-input.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-input.ts @@ -4,10 +4,11 @@ import type { OpenCodeCompletionFeedback } from "../drivers/opencode/opencode-se /** Preserve the exact call binding and the controller's acceptance/rejection. */ export function openCodeProxyCompletionFeedback( - request: (method: string, params: Record) => Promise, + request: (method: string, params: Record, signal?: AbortSignal) => Promise, ): OpenCodeCompletionFeedback { - return async (result, call) => { - const response = await request("item/tool/call", { ...call, arguments: result }); + return async (result, { signal, ...call }) => { + // The signal withdraws a still-queued request; it is never sent as data. + const response = await request("item/tool/call", { ...call, arguments: result }, signal); if (!response || typeof response !== "object" || Array.isArray(response)) throw new Error("Completion controller returned an invalid response"); const value = response as Record; diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts index ef39061f45e..2ad96e61eaa 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-output.test.ts @@ -1,5 +1,5 @@ -import { afterEach, expect, it, vi } from "vitest"; -import { createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createBackpressuredFrameWriter, createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; afterEach(() => vi.useRealTimers()); @@ -71,3 +71,66 @@ it("flushes preceding text before an oversized original delta without retaining output.flush(); expect(write).toHaveBeenCalledTimes(2); }); + +describe("backpressured frame writer", () => { + function sink() { + const lines: string[] = []; + let drain: (() => void) | undefined; + let accepting = false; + return { + lines, + write(chunk: string) { lines.push(chunk); return accepting; }, + once(_event: "drain", listener: () => void) { drain = listener; }, + resume() { accepting = true; const listener = drain; drain = undefined; listener?.(); }, + }; + } + + it("holds frames past the stream's buffer and delivers them in order", () => { + const stream = sink(); + const writer = createBackpressuredFrameWriter(stream); + for (const n of [1, 2, 3]) writer.write({ n }); + expect(stream.lines).toEqual(['{"n":1}\n']); + expect(writer.queuedFrames).toBe(2); + stream.resume(); + expect(stream.lines).toEqual(['{"n":1}\n', '{"n":2}\n', '{"n":3}\n']); + }); + + it("drops an aborted queued frame once and keeps later frames flowing", () => { + const stream = sink(); + const writer = createBackpressuredFrameWriter(stream); + const call = new AbortController(); + const abandoned = vi.fn(); + writer.write({ n: 1 }); + writer.write({ n: 2, call: true }, { signal: call.signal, onAbandoned: abandoned }); + writer.write({ n: 3 }); + call.abort(); + call.abort(); + expect(abandoned).toHaveBeenCalledOnce(); + stream.resume(); + expect(stream.lines).toEqual(['{"n":1}\n', '{"n":3}\n']); + }); + + it("does not withdraw a frame already handed to the stream", () => { + const stream = sink(); + const writer = createBackpressuredFrameWriter(stream); + const call = new AbortController(); + const abandoned = vi.fn(); + writer.write({ n: 1 }, { signal: call.signal, onAbandoned: abandoned }); + call.abort(); + expect(abandoned).not.toHaveBeenCalled(); + expect(stream.lines).toEqual(['{"n":1}\n']); + }); + + it("refuses an already-aborted frame and drains live frames on shutdown", () => { + const stream = sink(); + const writer = createBackpressuredFrameWriter(stream); + const abandoned = vi.fn(); + writer.write({ n: 1 }); + writer.write({ n: 2 }, { signal: AbortSignal.abort(), onAbandoned: abandoned }); + writer.write({ n: 3 }); + expect(abandoned).toHaveBeenCalledOnce(); + writer.drain(); + expect(stream.lines).toEqual(['{"n":1}\n', '{"n":3}\n']); + expect(writer.queuedFrames).toBe(0); + }); +}); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-output.ts b/packages/paperclip-runner/src/cli/opencode-proxy-output.ts index 6aea73ade19..06e15296d90 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-output.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-output.ts @@ -23,8 +23,18 @@ function streamedText(value: unknown) { return { frame, params, item, key, text: params.delta }; } +export interface OpenCodeProxyFrameOptions { + /** + * Abandons the frame while it is still queued in this process. Only a + * request whose sender can tolerate never being delivered may pass one. + */ + signal?: AbortSignal; + /** Called once when the frame is dropped before reaching the stream. */ + onAbandoned?: () => void; +} + /** Avoid making an MCP call wait behind a durable write for every text token. */ -export function createOpenCodeProxyOutput(write: (frame: unknown) => void) { +export function createOpenCodeProxyOutput(write: (frame: unknown, options?: OpenCodeProxyFrameOptions) => void) { let pending: ReturnType = null; let bytes = 0; let timer: ReturnType | undefined; @@ -38,11 +48,11 @@ export function createOpenCodeProxyOutput(write: (frame: unknown) => void) { text: current.text, delta: current.text, item: { ...current.item, text: current.text } } }); }; return { - send(frame: unknown): void { + send(frame: unknown, options?: OpenCodeProxyFrameOptions): void { const text = streamedText(frame); if (!text) { flush(); - write(frame); + write(frame, options); return; } const size = Buffer.byteLength(text.text); @@ -62,3 +72,84 @@ export function createOpenCodeProxyOutput(write: (frame: unknown) => void) { flush, }; } + +export interface OpenCodeProxyFrameSink { + write(chunk: string): boolean; + once(event: "drain", listener: () => void): unknown; +} + +type QueuedFrame = { line: string; options?: OpenCodeProxyFrameOptions; onAbort?: () => void }; + +/** + * Writes newline-delimited frames to a stream without handing it more than its + * own buffer accepts. + * + * runnerd reads this stream only as fast as its controller acknowledges earlier + * events, so a backlog can build here while OpenCode keeps working. Node would + * otherwise buffer that backlog itself, and every frame in it would eventually + * be delivered. Holding the backlog in this queue instead lets a request whose + * MCP client has already given up (for example a timed-out tool call) be + * dropped before runnerd can dispatch it. Order of delivered frames is kept. + */ +export function createBackpressuredFrameWriter(sink: OpenCodeProxyFrameSink) { + const queue: QueuedFrame[] = []; + let waitingForDrain = false; + const release = (entry: QueuedFrame) => { + if (entry.onAbort) entry.options?.signal?.removeEventListener("abort", entry.onAbort); + entry.onAbort = undefined; + }; + const abandon = (entry: QueuedFrame) => { + release(entry); + try { entry.options?.onAbandoned?.(); } catch { /* A sender callback never blocks later frames. */ } + }; + const writeEntry = (entry: QueuedFrame): boolean => { + release(entry); + return sink.write(entry.line); + }; + const pump = () => { + while (!waitingForDrain && queue.length > 0) { + const entry = queue.shift()!; + if (entry.options?.signal?.aborted) { + abandon(entry); + continue; + } + if (!writeEntry(entry)) { + waitingForDrain = true; + sink.once("drain", () => { + waitingForDrain = false; + pump(); + }); + } + } + }; + return { + write(frame: unknown, options?: OpenCodeProxyFrameOptions): void { + if (options?.signal?.aborted) { + try { options.onAbandoned?.(); } catch { /* See abandon(). */ } + return; + } + const entry: QueuedFrame = { line: `${JSON.stringify(frame)}\n`, options }; + if (options?.signal) { + entry.onAbort = () => { + const index = queue.indexOf(entry); + if (index < 0) return; + queue.splice(index, 1); + abandon(entry); + }; + options.signal.addEventListener("abort", entry.onAbort, { once: true }); + } + queue.push(entry); + pump(); + }, + /** Hands every remaining live frame to the stream, ignoring backpressure. */ + drain(): void { + for (const entry of queue.splice(0)) { + if (entry.options?.signal?.aborted) abandon(entry); + else writeEntry(entry); + } + }, + get queuedFrames(): number { + return queue.length; + }, + }; +} diff --git a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.test.ts b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.test.ts index 3511eb73cd7..12de1ae1b0e 100644 --- a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.test.ts +++ b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.test.ts @@ -84,6 +84,42 @@ describe("OpenCode MCP bridge", () => { expect(handler).toHaveBeenCalledOnce(); }); + it("abandons a call on disconnect only after every waiting connection is gone", async () => { + let started!: () => void; + const begun = new Promise(resolve => { started = resolve; }); + const aborted = vi.fn(); + const handler = vi.fn(({ signal }) => new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { aborted(); reject(new Error("abandoned")); }, { once: true }); + started(); + })); + const bridge = await startOpenCodeMcpBridge({ + tools: [{ name: "documents.write", inputSchema: { type: "object" } }], timeoutMs: null, handler, + }); + bridges.push(bridge); + const body = JSON.stringify({ jsonrpc: "2.0", id: "retried", method: "tools/call", params: { name: "documents.write", arguments: {} } }); + const post = (signal: AbortSignal) => fetch(bridge.url, { + method: "POST", body, signal, + headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json" }, + }); + const first = new AbortController(); + const retry = new AbortController(); + const firstResponse = post(first.signal).catch(() => "dropped"); + await begun; + const retryResponse = post(retry.signal).catch(() => "dropped"); + await new Promise(resolve => setTimeout(resolve, 50)); + first.abort(); + expect(await firstResponse).toBe("dropped"); + await new Promise(resolve => setTimeout(resolve, 50)); + expect(aborted).not.toHaveBeenCalled(); + retry.abort(); + expect(await retryResponse).toBe("dropped"); + for (let attempt = 0; attempt < 100 && !aborted.mock.calls.length; attempt += 1) { + await new Promise(resolve => setTimeout(resolve, 5)); + } + expect(aborted).toHaveBeenCalledOnce(); + expect(handler).toHaveBeenCalledOnce(); + }); + it("binds to loopback, requires authentication, and exposes only its closed catalog", async () => { const bridge = await startOpenCodeMcpBridge({ secret: "session-secret", diff --git a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts index 81e854d2a83..55d0fddb396 100644 --- a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts +++ b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts @@ -93,7 +93,7 @@ export async function startOpenCodeMcpBridge( }; try { options.onDiagnostic(JSON.stringify(diagnostic)); } catch { /* Optional diagnostics never change admission. */ } }; - const calls = new Map }>(); + const calls = new Map; waiting: Set }>(); const controllers = new Map(); const server = createServer((request, response) => { void handleRequest(request, response, { @@ -185,7 +185,7 @@ async function handleRequest( tools: OpenCodeMcpToolDefinition[]; admittedTools: OpenCodeMcpToolDefinition[]; validators: Map; - calls: Map }>; + calls: Map; waiting: Set }>; controllers: Map; handler: OpenCodeMcpBridgeOptions["handler"]; onRejectedCall: OpenCodeMcpBridgeOptions["onRejectedCall"]; @@ -309,10 +309,23 @@ async function handleRequest( ); if (!existing) { if (context.calls.size >= 2_048) context.calls.delete(context.calls.keys().next().value!); - context.calls.set(callId, { fingerprint, promise: execution }); + context.calls.set(callId, { fingerprint, promise: execution, waiting: new Set() }); context.controllers.set(callId, controller!); void execution.finally(() => context.controllers.delete(callId)).catch(() => undefined); } + // A client that drops its connection has abandoned the call just like one + // that sends notifications/cancelled. Abort once no connection still waits + // on this call identity, so a call still waiting to be handed to the + // controller is never dispatched after every client gave up. A live retry + // of the same call keeps it alive. + const waiting = (existing ?? context.calls.get(callId))?.waiting; + if (waiting && context.controllers.has(callId)) { + waiting.add(response); + response.once("close", () => { + waiting.delete(response); + if (!response.writableFinished && waiting.size === 0) context.controllers.get(callId)?.abort(); + }); + } try { const result = await execution; writeRpc(response, id, { content: [{ type: "text", text: safeJson(result) }] }); diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts index 4b2f24b79e5..76f9fd97e1d 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts @@ -172,6 +172,12 @@ type DynamicToolHandler = (call: { threadId: string; turnId: string; arguments: unknown; + /** + * Aborts when the MCP client abandons the call (cancel notification or a + * dropped connection). A handler that has not yet handed the call to its + * controller must not dispatch it after this fires. + */ + signal?: AbortSignal; }) => Promise; export type OpenCodeCompletionFeedback = (result: PrpStructuredRunResult, call: { @@ -179,6 +185,8 @@ export type OpenCodeCompletionFeedback = (result: PrpStructuredRunResult, call: callId: string; threadId: string; turnId: string; + /** Aborts when the MCP client abandons the completion call. */ + signal?: AbortSignal; }) => Promise; export interface OpenCodeServerDriverOptions { @@ -1149,6 +1157,7 @@ class OpenCodeHarnessSession implements HarnessSession { tool: string; callId: string; arguments: unknown; + signal?: AbortSignal; }): Promise { this.#pendingToolCalls += 1; this.#clearEmptyResponseAckTimer(); @@ -1156,7 +1165,7 @@ class OpenCodeHarnessSession implements HarnessSession { finally { this.#pendingToolCalls -= 1; } } - async #dispatchToolCall(call: { tool: string; callId: string; arguments: unknown }): Promise { + async #dispatchToolCall(call: { tool: string; callId: string; arguments: unknown; signal?: AbortSignal }): Promise { // The authenticated MCP bridge has resolved this against its closed // declared catalog. A second prefix strip changes legitimate operation IDs. const tool = call.tool; @@ -1186,6 +1195,7 @@ class OpenCodeHarnessSession implements HarnessSession { this.#dynamicToolHandler ? () => this.#dynamicToolHandler!({ tool: "get_task_context", callId: `${call.callId}:continuation`, threadId: this.#providerSessionId, turnId, arguments: {}, + ...(call.signal ? { signal: call.signal } : {}), }) : undefined); if (this.#closed || this.#activeTurnId !== turnId) throw new Error("The turn ended while resolving completion"); @@ -1223,6 +1233,7 @@ class OpenCodeHarnessSession implements HarnessSession { const feedback = this.#completionFeedback ? await this.#completionFeedback(validation.result, { tool, callId: call.callId, threadId: this.#providerSessionId, turnId, + ...(call.signal ? { signal: call.signal } : {}), }) : "Semantic completion accepted."; if (typeof feedback !== "string" || !feedback.trim()) @@ -1279,6 +1290,7 @@ class OpenCodeHarnessSession implements HarnessSession { threadId: this.#providerSessionId, turnId, arguments: call.arguments, + ...(call.signal ? { signal: call.signal } : {}), }); this.#emit( "item.completed", @@ -2449,6 +2461,7 @@ async function startRuntime(input: { tool: string; callId: string; arguments: unknown; + signal?: AbortSignal; }) => Promise; }): Promise { const taskEnvironment = configuredOpenCodeEnvironment(input.options.environment);