diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.test.ts index 1ebd37fc4fb..f5631d9521c 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.test.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.test.ts @@ -91,10 +91,26 @@ it("retains the actual MCP producer's offered schema and rejected shape records" const report={reportedWorkDisposition:"done",summary:"Synthetic complete response",completionClaim:{contractRevision:"1",objectiveSatisfied:true,criteria:[],remainingWork:[]},evidence:[],verification:[]}; const result=await rpc(2,"tools/call",{name:"paperclip_finish",arguments:{...report,continuation:"SYNTHETIC_PRIVATE_INPUT"}}); expect(result.result.isError).toBe(true); - expect(frames).toHaveLength(2); - const [actualOffered,actualRejected]=frames.map(frame=>JSON.parse(frame.params.message)); + const finishFrames=frames.filter(frame=>frame.params.message.includes("finish-input-diagnostic")); + expect(finishFrames).toHaveLength(2); + const [actualOffered,actualRejected]=finishFrames.map(frame=>JSON.parse(frame.params.message)); expect(actualOffered).toEqual({...offered,schemaSha256:createHash("sha256").update(JSON.stringify(schema)).digest("hex")}); expect(actualRejected).toEqual({...actualOffered,stage:"input_rejected",callIdSha256:createHash("sha256").update("2").digest("hex"),validationCategory:"continuation",rootShape:"object",continuationShape:"string"}); expect(JSON.stringify(frames)).not.toContain("SYNTHETIC_PRIVATE_INPUT"); } finally {await bridge.close();} }); + +it("forwards closed tool delivery records after binding and drops any record that could carry tool data", () => { + const frames: unknown[] = [], forward = createOpenCodeFinishDiagnosticForwarder(frame => frames.push(frame)); + const delivery = { schema: "paperclip.opencode.tool-delivery-diagnostic/v1", callIdSha256: "d".repeat(64), + socketRequestIndex: 1, socketIdleMs: null, handlerMs: 512, outcome: "result", written: true }; + forward.diagnostic(JSON.stringify(delivery)); + expect(frames).toEqual([]); + expect(forward.bind("actual-session")).toBe(true); + for (const invalid of [{ ...delivery, result: "private" }, { ...delivery, callIdSha256: "3" }, { ...delivery, outcome: "private" }, { ...delivery, outcome: ["result"] }, + { ...delivery, socketIdleMs: -1 }, { ...delivery, handlerMs: "512" }, { ...delivery, written: "yes" }]) forward.diagnostic(JSON.stringify(invalid)); + expect(frames).toEqual([]); + forward.diagnostic(JSON.stringify({ ...delivery, socketRequestIndex: 2, socketIdleMs: 5270 })); + expect(frames).toEqual([{ method: "warning", params: { threadId: "actual-session", + message: JSON.stringify({ ...delivery, socketRequestIndex: 2, socketIdleMs: 5270 }) } }]); +}); diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.ts b/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.ts index c9790efbc8b..f7d9baa926a 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-finish-diagnostics.ts @@ -17,6 +17,11 @@ export function createOpenCodeFinishDiagnosticForwarder(send: (frame: Warning) = if (message.length > 2048) return; let value: Record; try { value = JSON.parse(message); } catch { return; } + if (value && typeof value === "object" && !Array.isArray(value) + && value.schema === "paperclip.opencode.tool-delivery-diagnostic/v1") { + if (validToolDelivery(value)) emit(value); + return; + } if (!value || typeof value !== "object" || Array.isArray(value) || value.schema !== "paperclip.opencode.finish-input-diagnostic/v1" || typeof value.schemaSha256 !== "string" || !/^[a-f0-9]{64}$/.test(value.schemaSha256)) return; @@ -39,3 +44,16 @@ export function createOpenCodeFinishDiagnosticForwarder(send: (frame: Warning) = }, }; } + +const TOOL_DELIVERY_KEYS = "callIdSha256,handlerMs,outcome,schema,socketIdleMs,socketRequestIndex,written"; +// Delivery records carry only hashes, counters and timings. Reject anything +// else so this channel cannot forward tool inputs or results. +function validToolDelivery(value: Record): boolean { + const count = (entry: unknown) => typeof entry === "number" && Number.isSafeInteger(entry) && entry >= 0; + return Object.keys(value).sort().join(",") === TOOL_DELIVERY_KEYS + && typeof value.callIdSha256 === "string" && /^[a-f0-9]{64}$/.test(value.callIdSha256) + && count(value.socketRequestIndex) && count(value.handlerMs) + && (value.socketIdleMs === null || count(value.socketIdleMs)) + && typeof value.outcome === "string" && ["result", "error", "unanswered"].includes(value.outcome) + && typeof value.written === "boolean"; +} diff --git a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-delivery.test.ts b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-delivery.test.ts new file mode 100644 index 00000000000..74a7742792c --- /dev/null +++ b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-delivery.test.ts @@ -0,0 +1,139 @@ +import { createHash } from "node:crypto"; +import { Agent, request } from "node:http"; +import { afterEach, expect, it } from "vitest"; +import { startOpenCodeMcpBridge, type OpenCodeMcpBridge } from "./mcp-bridge.js"; + +const bridges: OpenCodeMcpBridge[] = []; +const agents: Agent[] = []; +afterEach(async () => { + for (const agent of agents.splice(0)) agent.destroy(); + await Promise.all(bridges.splice(0).map(bridge => bridge.close())); +}); + +type Answer = { status: number; connection: string | undefined; reusedSocket: boolean; body: { id: unknown; result?: { content?: Array<{ text: string }> } } }; + +// A pooling client like Bun's fetch: it reuses an idle keep-alive socket +// whenever the server leaves one open. +function call(bridge: OpenCodeMcpBridge, agent: Agent, id: number, name: string): Promise { + const body = JSON.stringify({ jsonrpc: "2.0", id, method: "tools/call", params: { name, arguments: {} } }); + return new Promise((resolve, reject) => { + const outgoing = request(bridge.url, { + method: "POST", agent, + headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json", "Content-Length": Buffer.byteLength(body) }, + }, response => { + const chunks: Buffer[] = []; + response.on("data", chunk => chunks.push(chunk)); + response.on("end", () => resolve({ + status: response.statusCode ?? 0, + connection: response.headers.connection, + reusedSocket: outgoing.reusedSocket, + body: JSON.parse(Buffer.concat(chunks).toString("utf8")), + })); + response.on("error", reject); + }); + outgoing.on("error", reject); + outgoing.end(body); + }); +} + +const tools = [ + { name: "list_agents", inputSchema: { type: "object" } }, + { name: "list_workflows", inputSchema: { type: "object" } }, + { name: "list_projects", inputSchema: { type: "object" } }, +]; + +it("answers every tool call on its own connection so a pooled socket can never strand a parallel result", async () => { + const messages: string[] = []; + const bridge = await startOpenCodeMcpBridge({ + tools, + handler: async ({ tool }) => { + await new Promise(resolve => setTimeout(resolve, tool === "list_workflows" ? 40 : 10)); + return { tool }; + }, + onDiagnostic: message => { if (message.includes("tool-delivery-diagnostic")) messages.push(message); }, + }); + bridges.push(bridge); + const agent = new Agent({ keepAlive: true, maxSockets: 8 }); + agents.push(agent); + + const first = await call(bridge, agent, 2, "list_agents"); + // Let the client finish handling the first response before the parallel + // pair, as OpenCode does while the model writes its next step. + await new Promise(resolve => setTimeout(resolve, 20)); + const [workflows, projects] = await Promise.all([ + call(bridge, agent, 3, "list_workflows"), + call(bridge, agent, 4, "list_projects"), + ]); + + for (const [answer, id, tool] of [[first, 2, "list_agents"], [workflows, 3, "list_workflows"], [projects, 4, "list_projects"]] as const) { + expect(answer.status).toBe(200); + expect(answer.body.id).toBe(id); + expect(JSON.parse(answer.body.result!.content![0]!.text)).toEqual({ tool }); + expect(answer.connection).toBe("close"); + expect(answer.reusedSocket).toBe(false); + } + + await expect.poll(() => messages.length).toBe(3); + const deliveries = messages.map(message => JSON.parse(message)); + expect(deliveries.map(delivery => delivery.callIdSha256).sort()).toEqual( + ["2", "3", "4"].map(id => createHash("sha256").update(id).digest("hex")).sort(), + ); + for (const delivery of deliveries) { + expect(delivery).toEqual({ + schema: "paperclip.opencode.tool-delivery-diagnostic/v1", + callIdSha256: expect.stringMatching(/^[a-f0-9]{64}$/), + socketRequestIndex: 1, + socketIdleMs: null, + handlerMs: expect.any(Number), + outcome: "result", + written: true, + }); + } + expect(JSON.stringify(messages)).not.toMatch(/list_|"tool"/); +}); + +it("records a call whose client left before its result was written", async () => { + const messages: string[] = []; + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + let started!: () => void; + const handlerStarted = new Promise(resolve => { started = resolve; }); + const bridge = await startOpenCodeMcpBridge({ + tools, + handler: async () => { started(); await held; return { ok: true }; }, + onDiagnostic: message => { if (message.includes("tool-delivery-diagnostic")) messages.push(message); }, + }); + bridges.push(bridge); + const controller = new AbortController(); + const abandoned = fetch(bridge.url, { + method: "POST", signal: controller.signal, + headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", id: 9, method: "tools/call", params: { name: "list_agents", arguments: {} } }), + }).catch(() => undefined); + await handlerStarted; + controller.abort(); + await abandoned; + await expect.poll(() => messages.length).toBe(1); + release(); + expect(JSON.parse(messages[0]!)).toMatchObject({ outcome: "unanswered", written: false, socketRequestIndex: 1 }); +}); + +it("records a rejected call as an answered error", async () => { + const messages: string[] = []; + const bridge = await startOpenCodeMcpBridge({ + tools, + handler: async () => ({ ok: true }), + onDiagnostic: message => { if (message.includes("tool-delivery-diagnostic")) messages.push(message); }, + }); + bridges.push(bridge); + const answer = await (await fetch(bridge.url, { + method: "POST", + headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", id: 11, method: "tools/call", params: { name: "not_declared", arguments: {} } }), + })).json() as { result: { isError: boolean } }; + expect(answer.result.isError).toBe(true); + await expect.poll(() => messages.length).toBe(1); + expect(JSON.parse(messages[0]!)).toMatchObject({ + callIdSha256: createHash("sha256").update("11").digest("hex"), outcome: "error", written: true, + }); +}); diff --git a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-diagnostics.test.ts b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-diagnostics.test.ts index 59844a49dc7..b08fca1a0ef 100644 --- a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-diagnostics.test.ts +++ b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge-diagnostics.test.ts @@ -44,7 +44,7 @@ it("binds closed rejection shapes to the actual offered finish schema without re }); expect(unrelated.result.isError).toBe(true); expect(handler).toHaveBeenCalledTimes(2); - expect(messages.map(message => JSON.parse(message))).toEqual([ + expect(messages.map(message => JSON.parse(message)).filter(message => message.schema === "paperclip.opencode.finish-input-diagnostic/v1")).toEqual([ { schema: "paperclip.opencode.finish-input-diagnostic/v1", stage: "offered", schemaSha256: digest }, ...["string", "array", "number", "boolean"].map((continuationShape, index) => ({ schema: "paperclip.opencode.finish-input-diagnostic/v1", stage: "input_rejected", @@ -63,7 +63,7 @@ it("reports the effective supplied schema and bounds diagnostics without admitti const inputSchema = { type: "object", required: ["continuation"], properties: { continuation: { type: "object" } } }; const bridge = await startOpenCodeMcpBridge({ tools: [{ name: "paperclip_finish", inputSchema }], handler, - onDiagnostic: message => messages.push(message), + onDiagnostic: message => { if (message.includes("finish-input-diagnostic")) messages.push(message); }, }); bridges.push(bridge); for (let index = 0; index < 12; index++) { diff --git a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts index f2428112f2b..2dc6c73cf05 100644 --- a/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts +++ b/packages/paperclip-runner/src/drivers/opencode/mcp-bridge.ts @@ -1,5 +1,6 @@ import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; +import type { Socket } from "node:net"; import { Ajv2020, type ValidateFunction } from "ajv/dist/2020.js"; import { toolInputError } from "../tool-input-error.js"; @@ -57,6 +58,16 @@ const DEFAULT_MAX_BODY_BYTES = 1_048_576; // Native app results/pages are capped at 500 KB before reaching this bridge. // Leave room for their delivery envelope, but never cut a completed JSON value. const MAX_RESULT_BYTES = 1_048_576; +const TOOL_DELIVERY_DIAGNOSTIC_SCHEMA = "paperclip.opencode.tool-delivery-diagnostic/v1"; +const MAX_TOOL_DELIVERY_DIAGNOSTICS = 256; + +type SocketUse = { requests: number; lastResponseAt: number | null }; +type ToolDelivery = { + socketRequestIndex: number; + socketIdleMs: number | null; + receivedAt: number; +}; +type ToolDeliveryOutcome = "result" | "error" | "unanswered"; export function canonicalOpenCodeMcpToolName(name: string): string { const trimmed = name.trim(); @@ -94,10 +105,41 @@ export async function startOpenCodeMcpBridge( }; try { options.onDiagnostic(JSON.stringify(diagnostic)); } catch { /* Optional diagnostics never change admission. */ } }; + let deliveryDiagnostics = 0; + const diagnoseToolDelivery = ( + callId: string, + delivery: ToolDelivery, + outcome: ToolDeliveryOutcome, + written: boolean, + ) => { + if (!options.onDiagnostic || deliveryDiagnostics++ >= MAX_TOOL_DELIVERY_DIAGNOSTICS) return; + const diagnostic = { + schema: TOOL_DELIVERY_DIAGNOSTIC_SCHEMA, + callIdSha256: createHash("sha256").update(callId).digest("hex"), + socketRequestIndex: delivery.socketRequestIndex, + socketIdleMs: delivery.socketIdleMs, + handlerMs: Date.now() - delivery.receivedAt, + outcome, + written, + }; + try { options.onDiagnostic(JSON.stringify(diagnostic)); } catch { /* Optional diagnostics never change delivery. */ } + }; const calls = new Map; waiting: Set }>(); const controllers = new Map(); + const socketUses = new WeakMap(); const server = createServer((request, response) => { + const use = socketUses.get(request.socket) ?? { requests: 0, lastResponseAt: null }; + socketUses.set(request.socket, use); + const receivedAt = Date.now(); + const delivery: ToolDelivery = { + socketRequestIndex: ++use.requests, + socketIdleMs: use.lastResponseAt === null ? null : receivedAt - use.lastResponseAt, + receivedAt, + }; + response.once("finish", () => { use.lastResponseAt = Date.now(); }); void handleRequest(request, response, { + delivery, + diagnoseToolDelivery, secret, tools, admittedTools, @@ -182,6 +224,8 @@ async function handleRequest( request: IncomingMessage, response: ServerResponse, context: { + delivery: ToolDelivery; + diagnoseToolDelivery: (callId: string, delivery: ToolDelivery, outcome: ToolDeliveryOutcome, written: boolean) => void; secret: string; tools: OpenCodeMcpToolDefinition[]; admittedTools: OpenCodeMcpToolDefinition[]; @@ -198,6 +242,11 @@ async function handleRequest( }, ): Promise { setSecurityHeaders(response); + // Never let the MCP client reuse a pooled socket for a later request. A + // native OpenCode run (Bun fetch) once lost a written tools/call result on + // a parallel call and waited about five minutes for Bun's fetch timeout. + // A loopback connect per request costs far less than that failure mode. + response.setHeader("Connection", "close"); if (!authorized(request.headers.authorization, context.secret)) { response.statusCode = 401; response.end(); @@ -265,6 +314,14 @@ async function handleRequest( return; } const params = isRecord(message.params) ? message.params : {}; + const callId = typeof id === "string" || typeof id === "number" ? String(id) : randomBytes(12).toString("hex"); + // Record how each call's answer left the bridge, rejections included, so a + // client that never consumes a written answer can be told apart from a + // bridge that never answered it. + let outcome: ToolDeliveryOutcome = "unanswered"; + response.once("close", () => { + context.diagnoseToolDelivery(callId, context.delivery, outcome, response.writableFinished); + }); const rawName = typeof params.name === "string" ? params.name : ""; // A declared operation may itself start with paperclip_. Preserve that // authority identity before accepting an OpenCode transport prefix alias. @@ -273,10 +330,10 @@ async function handleRequest( ? exactName : canonicalOpenCodeMcpToolName(exactName); if (!context.admittedTools.some((entry) => entry.name === tool)) { + outcome = "error"; writeRpc(response, id, { isError: true, content: [{ type: "text", text: "Unsupported tool." }] }); return; } - const callId = typeof id === "string" || typeof id === "number" ? String(id) : randomBytes(12).toString("hex"); const args = params.arguments ?? {}; const validator = context.validators.get(tool); if (validator === undefined || !validator(args)) { @@ -288,6 +345,7 @@ async function handleRequest( } const error = toolInputError(validator); try { context.onRejectedCall?.({ tool, callId, arguments: args }, error); } catch {} + outcome = "error"; writeRpc(response, id, { isError: true, content: [{ type: "text", text: error }], @@ -298,6 +356,7 @@ async function handleRequest( const existing = context.calls.get(callId); if (existing && existing.fingerprint !== fingerprint) { try { context.onRejectedCall?.({ tool, callId, arguments: args }, "Duplicate call identity conflict."); } catch {} + outcome = "error"; writeRpc(response, id, { isError: true, content: [{ type: "text", text: "Duplicate call identity conflict." }] }); return; } @@ -328,8 +387,11 @@ async function handleRequest( } try { const result = await execution; - writeRpc(response, id, { content: [{ type: "text", text: safeJson(result) }] }); + const text = safeJson(result); + outcome = "result"; + writeRpc(response, id, { content: [{ type: "text", text }] }); } catch (error) { + outcome = "error"; writeRpc(response, id, { isError: true, content: [{ type: "text", text: safeError(error) }] }); } }