Harden Cursor request and bridge liveness

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

src/proxy.ts

@@ -330,7 +330,7 @@ interface SpawnBridgeOptions {
 function spawnBridge(options: SpawnBridgeOptions): {
   proc: ReturnType<typeof Bun.spawn>;
   /** 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<string, unknown>, 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,