diff --git a/src/h2-bridge.mjs b/src/h2-bridge.mjs index 5b48d3d..956ed78 100644 --- a/src/h2-bridge.mjs +++ b/src/h2-bridge.mjs @@ -17,6 +17,7 @@ * is written as the request body and the stream is ended immediately. * After config, subsequent stdin messages are raw bytes to write to the H2 stream. * H2 response data is written to stdout using the same length-prefixed framing. + * Control records use kind 1 and report the session/stream writability state. */ import http2 from "node:http2"; import crypto from "node:crypto"; @@ -28,6 +29,7 @@ const DEFAULT_SESSION_TIMEOUT_MS = 120_000; const MAX_PING_INTERVAL_MS = 300_000; const MAX_PING_TIMEOUT_MS = 60_000; const MAX_SESSION_TIMEOUT_MS = 600_000; +const MAX_FAILURE_EXIT_DELAY_MS = 1_000; function boundedDuration(name, fallback, maximum) { const value = Number.parseInt(process.env[name] ?? "", 10); @@ -43,8 +45,8 @@ function writeMessage(kind, data) { process.stdout.write(payload); } -function writeTransportStatus(transport) { - writeMessage(1, Buffer.from(JSON.stringify({ transport }))); +function writeTransportStatus(session, stream, writable) { + writeMessage(1, Buffer.from(JSON.stringify({ session, stream, writable }))); } // --- Buffered stdin reader --- @@ -122,6 +124,7 @@ function emitDiagnostic(stage, details = {}) { try { process.stderr.write(`${JSON.stringify(event)}\n`); } catch {} } emitDiagnostic("bridge_start"); +writeTransportStatus("connecting", "opening", false); const client = http2.connect(url || "https://api2.cursor.sh"); const pingIntervalMs = boundedDuration( @@ -161,7 +164,7 @@ function sendPing() { client.on("connect", () => { emitDiagnostic("h2_connect"); - writeTransportStatus("writable"); + writeTransportStatus("open", "open", true); // PING validates the HTTP/2/TCP path, not model output. A 15s cadence and // 10s response window leave normal multi-minute model thinking untouched. pingInterval = setInterval(sendPing, pingIntervalMs); @@ -186,11 +189,18 @@ let failed = false; function failBridge(errorName) { if (failed || cleanExit) return; failed = true; - writeTransportStatus("failed"); + writeTransportStatus("failed", "failed", false); clearLivenessTimers(); emitDiagnostic("terminal", { errorName, exitCode: 1 }); try { client.destroy(); } catch {} - process.exit(1); + // The normal path exits immediately. Tests can retain this short, bounded + // window to exercise the parent-side failed-transport resume rejection + // while the child process itself is still alive. + const delay = boundedDuration( + "CURSOR_BRIDGE_FAILURE_EXIT_DELAY_MS", 0, MAX_FAILURE_EXIT_DELAY_MS, + ); + if (delay > 0) setTimeout(() => process.exit(1), delay); + else process.exit(1); } client.on("error", () => failBridge("Http2SessionError")); @@ -248,9 +258,10 @@ h2Stream.on("end", () => { clearLivenessTimers(); if (!unary && !sawConnectEndStream) { failBridge("ConnectProtocolError"); + return; } cleanExit = true; - writeTransportStatus("closed"); + writeTransportStatus("closed", "closed", false); emitDiagnostic("terminal", { exitCode: 0 }); client.close(); // Give stdout time to flush diff --git a/src/proxy.ts b/src/proxy.ts index 2b27c13..494bf5e 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -330,7 +330,7 @@ interface SpawnBridgeOptions { function spawnBridge(options: SpawnBridgeOptions): { proc: ReturnType; /** Returns false when the child has already exited or its stdin rejects the frame. */ - write: (data: Uint8Array) => boolean; + write: (data: Uint8Array, requireWritable?: boolean) => boolean; end: () => void; /** Stop the child when this request has reached a non-resumable terminal state. */ terminate: () => void; @@ -340,6 +340,8 @@ function spawnBridge(options: SpawnBridgeOptions): { get alive(): boolean; /** True only after the child reports a live, writable HTTP/2 transport. */ get writable(): boolean; + /** Last child-reported transport state for lifecycle decisions and diagnostics. */ + get transportState(): "connecting" | "writable" | "closed" | "failed"; } { const proc = Bun.spawn(["node", BRIDGE_PATH], { stdin: "pipe", @@ -414,7 +416,7 @@ function spawnBridge(options: SpawnBridgeOptions): { // Track exit state so late onClose registrations fire immediately. let exited = false; let exitCode = 1; - let writable = false; + let transportState: "connecting" | "writable" | "closed" | "failed" = "connecting"; (async () => { const reader = proc.stdout.getReader(); @@ -436,10 +438,21 @@ function spawnBridge(options: SpawnBridgeOptions): { // a retained bridge can safely accept a tool-result resume. if (payload[0] === 1) { try { - const status = JSON.parse(payload.subarray(1).toString("utf8")); - writable = status.transport === "writable"; + const status = JSON.parse(payload.subarray(1).toString("utf8")) as { + session?: unknown; + stream?: unknown; + writable?: unknown; + }; + transportState = status.session === "open" && status.stream === "open" && + status.writable === true + ? "writable" + : status.session === "failed" || status.stream === "failed" + ? "failed" + : status.session === "closed" || status.stream === "closed" + ? "closed" + : "connecting"; } catch { - writable = false; + transportState = "failed"; } } else if (payload[0] === 0) { cbs.data?.(Buffer.from(payload.subarray(1))); @@ -459,9 +472,10 @@ function spawnBridge(options: SpawnBridgeOptions): { return { proc, get alive() { return !exited; }, - get writable() { return !exited && writable; }, - write(data) { - if (exited) return false; + get writable() { return !exited && transportState === "writable"; }, + get transportState() { return transportState; }, + write(data, requireWritable = false) { + if (exited || (requireWritable && transportState !== "writable")) return false; try { proc.stdin.write(lpEncode(data)); return true; @@ -698,7 +712,18 @@ async function handleChatCompletion( return handleToolResultResume(activeBridge, toolResults, userText, modelId, convKey); } - // Bridge died (timeout, server disconnect, etc.). + if (activeBridge.bridge.alive) { + // The subprocess can outlive its HTTP/2 session briefly. Its stdin is + // still writable at the OS level, but Cursor can no longer receive a + // resume frame. Fail this resume explicitly instead of silently queuing + // bytes and waiting for the semantic watchdog. + clearInterval(activeBridge.heartbeatTimer); + activeBridge.bridge.terminate(); + return createUnavailableResumeResponse(modelId); + } + + // A fully exited bridge has no live server-side state; retain the existing + // fresh-request fallback, which rebuilds from checkpoint and tool history. // Clean up and fall through to start a fresh bridge. clearInterval(activeBridge.heartbeatTimer); activeBridge.bridge.end(); @@ -2016,6 +2041,11 @@ function handleToolResultResume( ): Response { const { bridge, heartbeatTimer, blobStore, mcpTools, cloudRule, pendingExecs } = active; emitLifecycleDiagnostic("proxy", "tool_resume", { toolCount: toolResults.length }); + if (!bridge.writable) { + clearInterval(heartbeatTimer); + bridge.terminate(); + return createUnavailableResumeResponse(modelId); + } // Answer each pending exec with a matching tool result: redirected native // execs get their typed native result frame, MCP execs get an mcpResult. @@ -2034,10 +2064,16 @@ function handleToolResultResume( } if (result && exec.native) { - const sent = sendNativeExecResult(exec, exec.native, text, (bytes) => - bridge.write(frameConnectMessage(bytes)), - ); - if (sent) continue; + try { + const sent = sendNativeExecResult(exec, exec.native, text, (bytes) => + bridge.write(frameConnectMessage(bytes), true), + ); + if (sent) continue; + } catch { + clearInterval(heartbeatTimer); + bridge.terminate(); + return createUnavailableResumeResponse(modelId); + } } const mcpResult = result @@ -2077,9 +2113,14 @@ function handleToolResultResume( message: { case: "execClientMessage", value: execClientMessage }, }); - bridge.write( + if (!bridge.write( frameConnectMessage(toBinary(AgentClientMessageSchema, clientMessage)), - ); + true, + )) { + clearInterval(heartbeatTimer); + bridge.terminate(); + return createUnavailableResumeResponse(modelId); + } } try { @@ -2095,6 +2136,26 @@ function handleToolResultResume( } } +/** A bounded SSE terminal for a paused bridge whose transport has failed. */ +function createUnavailableResumeResponse(modelId: string): Response { + const completionId = `chatcmpl-${crypto.randomUUID().replace(/-/g, "").slice(0, 28)}`; + const created = Math.floor(Date.now() / 1000); + const makeChunk = (delta: Record, finishReason: string | null) => ({ + id: completionId, + object: "chat.completion.chunk", + created, + model: modelId, + choices: [{ index: 0, delta, finish_reason: finishReason }], + }); + const error = "Cursor bridge transport is unavailable for tool resume"; + const body = [ + `data: ${JSON.stringify(makeChunk({ content: `\n[Error: ${error}]` }, null))}\n\n`, + `data: ${JSON.stringify(makeChunk({}, "stop"))}\n\n`, + "data: [DONE]\n\n", + ].join(""); + return new Response(body, { headers: SSE_HEADERS }); +} + async function handleNonStreamingResponse( payload: CursorRequestPayload, accessToken: string, diff --git a/test/smoke.ts b/test/smoke.ts index 0b08cbd..7debd34 100644 --- a/test/smoke.ts +++ b/test/smoke.ts @@ -6,16 +6,20 @@ import { AgentServerMessageSchema, ConversationStateStructureSchema, ExecServerControlMessageSchema, + ExecServerMessageSchema, GetBlobArgsSchema, GetUsableModelsResponseSchema, HeartbeatUpdateSchema, InteractionQuerySchema, InteractionUpdateSchema, KvServerMessageSchema, + McpArgsSchema, ModelDetailsSchema, TextDeltaUpdateSchema, SetupVmEnvironmentArgsSchema, ThinkingDetailsSchema, + ThinkingDeltaUpdateSchema, + TokenDeltaUpdateSchema, TurnEndedUpdateSchema, } from "../src/proto/agent_pb"; import { @@ -477,6 +481,128 @@ async function testNonStreamingProtocolErrorsAreStructured() { console.log("[test] Non-streaming protocol errors return structured responses"); } +async function testBridgeInvalidEndStreamHasOneTerminalState() { + console.log("[test] Testing delayed bridge protocol-failure terminal ownership..."); + const fixture = await createFakeCursorServer({ deadlineMs: 1_000 }); + fixture.setScenario("missing-end-stream"); + const lpEncode = (data: Uint8Array) => { + const frame = Buffer.alloc(4 + data.length); + frame.writeUInt32BE(data.length, 0); + frame.set(data, 4); + return frame; + }; + try { + const bridge = Bun.spawn({ + cmd: ["node", "src/h2-bridge.mjs"], + cwd: process.cwd(), + env: { + ...process.env, + CURSOR_BRIDGE_FAILURE_EXIT_DELAY_MS: "50", + }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + const config = new TextEncoder().encode(JSON.stringify({ + accessToken: "test-token", + url: fixture.apiUrl, + path: "/agent.v1.AgentService/Run", + unary: false, + })); + bridge.stdin.write(lpEncode(config)); + bridge.stdin.write(lpEncode(new Uint8Array([0]))); + bridge.stdin.end(); + + const [exitCode, stdout] = await Promise.all([ + bridge.exited, + new Response(bridge.stdout).arrayBuffer(), + ]); + assertEqual(exitCode, 1, "A missing Connect end-stream must retain its failure exit code when delayed"); + + const controls: Array<{ session?: string; stream?: string; writable?: boolean }> = []; + const bytes = Buffer.from(stdout); + for (let offset = 0; offset + 4 <= bytes.length;) { + const length = bytes.readUInt32BE(offset); + offset += 4; + assert(offset + length <= bytes.length, "Bridge stdout framing must remain complete after protocol failure"); + const payload = bytes.subarray(offset, offset + length); + offset += length; + if (payload[0] === 1) controls.push(JSON.parse(payload.subarray(1).toString("utf8"))); + } + assertEqual( + controls.filter((status) => status.session === "failed" && status.stream === "failed").length, + 1, + "A delayed Connect protocol failure must publish exactly one failed transport state", + ); + assert( + !controls.some((status) => status.session === "closed" || status.stream === "closed"), + "A delayed Connect protocol failure must not publish a contradictory closed transport state", + ); + } finally { + await fixture.close(); + } + console.log("[test] Delayed Connect protocol failure remains failed exactly once without closed status"); +} + +async function testNonStreamingCollectionFailureContract() { + console.log("[test] Testing non-streaming truncation, stalls, and bridge failures..."); + const promptSentinel = "non-streaming-secret-prompt-sentinel"; + const forbidden = [ + "secret-access-token", + "Bearer secret-access-token", + "session=secret-cookie", + promptSentinel, + ]; + const scenarios: Array<{ scenario: CursorScenario; expected: string; environment?: Record }> = [ + { scenario: "missing-end-stream", expected: "Bridge exited with code 1 before Connect end-stream" }, + { scenario: "transport-close-without-terminal", expected: "Bridge exited with code" }, + { + scenario: "heartbeat-trickle-kept-open", + expected: "response_stall_timeout", + environment: { + CURSOR_PROXY_FIRST_PROGRESS_TIMEOUT_MS: "25", + CURSOR_PROXY_SEMANTIC_PROGRESS_TIMEOUT_MS: "25", + }, + }, + ]; + + for (const { scenario, expected, environment = {} } of scenarios) { + const fixture = await createFakeCursorServer({ deadlineMs: 1_000 }); + fixture.setScenario(scenario); + try { + const result = await runScenarioSubprocess(fixture.apiUrl, ` + const { startProxy, stopProxy } = await import("./src/proxy.ts"); + const port = await startProxy(async () => "secret-access-token"); + const response = await fetch("http://localhost:" + port + "/v1/chat/completions", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer secret-access-token", + Cookie: "session=secret-cookie", + }, + body: JSON.stringify({ model: "test", stream: false, messages: [{ role: "user", content: "${promptSentinel}" }] }), + }); + const body = await response.json(); + stopProxy(); + console.log(JSON.stringify({ status: response.status, body })); + process.exit(0); + `, environment); + assertEqual(result.status, scenario === "heartbeat-trickle-kept-open" ? 504 : 500, + `${scenario} must return an explicit non-streaming error status`); + const serialized = JSON.stringify(result.body); + assert(serialized.includes(expected), `${scenario} must identify its terminal failure`); + assert(!serialized.includes('"finish_reason":"stop"'), + `${scenario} must not report clean completion without a completion signal`); + for (const secret of forbidden) { + assert(!serialized.includes(secret), `${scenario} error payload must redact ${secret}`); + } + } finally { + await fixture.close(); + } + } + console.log("[test] Non-streaming failures are bounded, explicit, and sanitized"); +} + async function runFakeScenario(apiUrl: string, scenario: CursorScenario): Promise { const session = http2.connect(apiUrl); const stream = session.request({ @@ -826,6 +952,30 @@ async function testServerMessageClassification(modules: TestModules) { }) } }); assertEqual(state.semanticProgressCount, 1, "Text output must count as semantic progress"); + dispatch({ message: { case: "interactionUpdate", value: create(InteractionUpdateSchema, { + message: { case: "thinkingDelta", value: create(ThinkingDeltaUpdateSchema, { text: "reasoning" }) }, + }) } }); + dispatch({ message: { case: "interactionUpdate", value: create(InteractionUpdateSchema, { + message: { case: "tokenDelta", value: create(TokenDeltaUpdateSchema, { tokens: 1 }) }, + }) } }); + dispatch({ message: { case: "interactionUpdate", value: create(InteractionUpdateSchema, { + message: { case: "tokenDelta", value: create(TokenDeltaUpdateSchema, { tokens: 0 }) }, + }) } }); + dispatch({ message: { case: "execServerMessage", value: create(ExecServerMessageSchema, { + id: 1, + execId: "semantic-progress-exec", + message: { + case: "mcpArgs", + value: create(McpArgsSchema, { + name: "fixture-tool", + toolName: "fixture-tool", + toolCallId: "semantic-progress-call", + args: { input: new TextEncoder().encode("{}") }, + }), + }, + }) } }); + assertEqual(state.semanticProgressCount, 4, "Thinking, positive token deltas, and exec transitions must count as semantic progress"); + let unsupportedError: Error | undefined; const safeQueryCases = [ "webSearchRequestQuery", @@ -844,6 +994,7 @@ async function testServerMessageClassification(modules: TestModules) { assert(!unsupportedError, `${queryCase} must be safely refused rather than terminal`); assert(responseBytes && responseBytes.length > 5, `${queryCase} must emit a typed interaction response`); } + assertEqual(state.semanticProgressCount, 4, "Interaction queries must not count as semantic progress"); dispatch({ message: { case: "interactionQuery", value: create(InteractionQuerySchema, { id: 1, query: { case: "setupVmEnvironmentArgs", value: create(SetupVmEnvironmentArgsSchema, {}) }, @@ -888,11 +1039,12 @@ async function testServerMessageClassification(modules: TestModules) { async function runScenarioSubprocess( apiUrl: string, script: string, + environment: Record = {}, ): Promise> { const child = Bun.spawn({ cmd: [process.execPath, "-e", script], cwd: process.cwd(), - env: { ...process.env, CURSOR_API_URL: apiUrl }, + env: { ...process.env, CURSOR_API_URL: apiUrl, ...environment }, stdout: "pipe", stderr: "pipe", }); @@ -909,6 +1061,7 @@ async function observeToolResume( apiUrl: string, scenario: "tool-pause-resume" | "tool-pause-transport-death", collision = false, + environment: Record = {}, ): Promise> { return runScenarioSubprocess(apiUrl, ` import { startProxy, stopProxy } from "./src/proxy.ts"; @@ -933,15 +1086,17 @@ async function observeToolResume( second = await boundedText(await request([{ role: "user", content: opening }])); secondCall = JSON.parse(second.text.match(/data: (.+)\\n\\n/)?.[1] ?? "{}").choices?.[0]?.delta?.tool_calls?.[0]?.id; } + const resumeStartedAt = performance.now(); const resumed = await boundedText(await request([ { role: "user", content: opening }, { role: "assistant", content: null, tool_calls: [{ id: firstCall, type: "function", function: { name: "fixture-tool", arguments: "{}" } }] }, { role: "tool", tool_call_id: firstCall, content: "result-for-" + firstCall }, ])); + const resumeElapsedMs = performance.now() - resumeStartedAt; stopProxy(); - console.log(JSON.stringify({ first, firstCall, second, secondCall, resumed })); + console.log(JSON.stringify({ first, firstCall, second, secondCall, resumed, resumeElapsedMs })); process.exit(0); - `); + `, environment); } async function observeStalledRefresh(): Promise> { @@ -1019,16 +1174,25 @@ async function testResumeRefreshAndNonStreaming(modules: TestModules) { console.log("[test] Identical-opening conversations retain isolated paused bridges"); fixture.setScenario("tool-pause-transport-death"); - const droppedWrite = await observeToolResume(fixture.apiUrl, "tool-pause-transport-death"); + const droppedWrite = await observeToolResume( + fixture.apiUrl, + "tool-pause-transport-death", + false, + { CURSOR_BRIDGE_FAILURE_EXIT_DELAY_MS: "100" }, + ); + const elapsedMs = Number(droppedWrite.resumeElapsedMs); + const droppedWriteBody = String((droppedWrite.resumed as { text?: string }).text ?? ""); assert( - String((droppedWrite.resumed as { text?: string }).text ?? "").includes('"finish_reason":"tool_calls"'), - "A closed transport must not accept a dropped resume write as a live bridge", + !Boolean((droppedWrite.resumed as { timedOut?: boolean }).timedOut) && elapsedMs < 150, + "A process-alive, transport-dead resume must terminate before the bounded observation window", ); assert( - !String((droppedWrite.resumed as { text?: string }).text ?? "").includes("resumed-correctly"), - "A dead bridge must reject its pending result rather than accepting a stale write", + droppedWriteBody.includes("[Error: Cursor bridge transport is unavailable for tool resume]"), + "A process-alive dead transport must report the sanitized resume error", ); - console.log("[test] Closed-stream bridge rejects stale writes and resumes through a fresh request"); + assertEqual(doneCount(droppedWriteBody), 1, "A process-alive dead transport resume must emit exactly one [DONE]"); + assert(!droppedWriteBody.includes("resumed-correctly"), "A dead transport must never accept a stale resume write"); + console.log(`[test] Process-alive dead transport resume errors explicitly once in ${Math.round(elapsedMs)}ms`); fixture.setScenario("tool-pause-clean-exit"); const port = await modules.startProxy(async () => "test-token"); @@ -1460,6 +1624,8 @@ async function main() { await testPluginShape(modules); await testArrayContentParsing(modules); await testNonStreamingProtocolErrorsAreStructured(); + await testBridgeInvalidEndStreamHasOneTerminalState(); + await testNonStreamingCollectionFailureContract(); await testFakeCursorScenarioHarness(); await testStreamingTermination(); await testClientAbortTeardown();