fix: handle cancelled SSE streams

src/proxy.ts

@@ -1364,10 +1364,11 @@ function createBridgeStreamResponse(
   const completionId = `chatcmpl-${crypto.randomUUID().replace(/-/g, "").slice(0, 28)}`;
   const created = Math.floor(Date.now() / 1000);
 
+  let closed = false;
+  let cancelStream: (() => void) | undefined;
   const stream = new ReadableStream({
     start(controller) {
       const encoder = new TextEncoder();
-      let closed = false;
       const sendSSE = (data: object) => {
         if (closed) return;
         controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
@@ -1376,11 +1377,27 @@ function createBridgeStreamResponse(
         if (closed) return;
         controller.enqueue(encoder.encode("data: [DONE]\n\n"));
       };
+      const stopBridge = () => {
+        if (activeBridges.get(bridgeKey)?.bridge === bridge) {
+          activeBridges.delete(bridgeKey);
+        }
+        clearInterval(heartbeatTimer);
+        // end() only half-closes the request; killing also stops the response.
+        try { bridge.proc.kill(); } catch {}
+      };
       const closeController = () => {
         if (closed) return;
         closed = true;
         controller.close();
       };
+      // A paused bridge closes its controller on purpose and stays registered
+      // in activeBridges for tool-result continuation. Tearing that down would
+      // break the round-trip OpenCode is about to make.
+      cancelStream = () => {
+        if (closed) return;
+        closed = true;
+        stopBridge();
+      };
 
       const makeChunk = (
         delta: Record<string, unknown>,
@@ -1503,10 +1520,8 @@ function createBridgeStreamResponse(
             sendSSE(makeChunk({}, "stop"));
             sendSSE(makeUsageChunk());
             sendDone();
+            stopBridge();
             closeController();
-            activeBridges.delete(bridgeKey);
-            clearInterval(heartbeatTimer);
-            bridge.end();
           }
         },
       );
@@ -1542,6 +1557,9 @@ function createBridgeStreamResponse(
         }
       });
     },
+    cancel() {
+      cancelStream?.();
+    },
   });
 
   return new Response(stream, { headers: SSE_HEADERS });

test/smoke.ts

@@ -3,7 +3,9 @@ import http2 from "node:http2";
 import type { AddressInfo } from "node:net";
 import { create, toBinary } from "@bufbuild/protobuf";
 import {
+  AgentServerMessageSchema,
   GetUsableModelsResponseSchema,
+  InteractionUpdateSchema,
   ModelDetailsSchema,
 } from "../src/proto/agent_pb";
 
@@ -25,10 +27,12 @@ interface TestCursorBackend {
   refreshUrl: string;
   setDiscoveryMode: (mode: DiscoveryMode) => void;
   setDiscoveredModels: (models: Array<{ id: string; name: string; reasoning?: boolean }>) => void;
+  setHoldRunStream: (hold: boolean) => void;
   resetObservations: () => void;
   getDiscoveryAuthHeaders: () => string[];
   getDiscoveryRequestBodies: () => Uint8Array[];
   getRefreshAuthHeaders: () => string[];
+  waitForRunStreamClose: () => Promise<void>;
   close: () => Promise<void>;
 }
 
@@ -54,6 +58,27 @@ function assertArrayEqual(
   }
 }
 
+async function withTimeout<T>(
+  promise: Promise<T>,
+  message: string,
+  timeoutMs = 10_000,
+): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, reject) => {
+        timer = setTimeout(
+          () => reject(new Error(`Timed out after ${timeoutMs}ms waiting for: ${message}`)),
+          timeoutMs,
+        );
+      }),
+    ]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
 function makeJwt(expiresAtSeconds: number): string {
   const header = btoa(JSON.stringify({ alg: "HS256", typ: "JWT" }));
   const payload = btoa(JSON.stringify({ exp: expiresAtSeconds }));
@@ -68,13 +93,33 @@ function frameConnectUnaryMessage(payload: Uint8Array): Buffer {
   return frame;
 }
 
+/** A Connect-framed text delta, enough to make the proxy emit its first SSE chunk. */
+function frameRunTextDelta(text: string): Buffer {
+  return frameConnectUnaryMessage(
+    toBinary(
+      AgentServerMessageSchema,
+      create(AgentServerMessageSchema, {
+        message: {
+          case: "interactionUpdate",
+          value: create(InteractionUpdateSchema, {
+            message: { case: "textDelta", value: { text } },
+          }),
+        },
+      }),
+    ),
+  );
+}
+
 async function createTestCursorBackend(): Promise<TestCursorBackend> {
   let discoveryMode: DiscoveryMode = "success";
   let discoveredModels: Array<{ id: string; name: string; reasoning?: boolean }> = [
     { id: "composer-2", name: "Composer 2", reasoning: true },
   ];
   const discoveryAuthHeaders: string[] = [];
   const discoveryRequestBodies: Uint8Array[] = [];
+  let runStreamClosed = Promise.withResolvers<void>();
+  const heldRunStreams = new Set<http2.ServerHttp2Stream>();
+  let holdRunStream = false;
   const refreshAuthHeaders: string[] = [];
 
   const refreshServer = http.createServer((req, res) => {
@@ -109,11 +154,24 @@ async function createTestCursorBackend(): Promise<TestCursorBackend> {
     const path = String(headers[":path"] ?? "");
     const authHeader = String(headers.authorization ?? "");
     if (path === "/agent.v1.AgentService/Run") {
+      const closed = runStreamClosed;
       stream.respond({
         ":status": 200,
         "content-type": "application/connect+proto",
       });
-      stream.end();
+      stream.on("close", () => {
+        heldRunStreams.delete(stream);
+        closed.resolve();
+      });
+      if (holdRunStream) {
+        // Cancellation can only be observed on a stream still running when the
+        // client goes away. Bun also withholds SSE response headers until the
+        // first body byte, so emit one to unblock the caller's fetch.
+        heldRunStreams.add(stream);
+        stream.write(frameRunTextDelta("streaming"));
+      } else {
+        stream.end();
+      }
       return;
     }
 
@@ -180,10 +238,20 @@ async function createTestCursorBackend(): Promise<TestCursorBackend> {
     setDiscoveredModels(models) {
       discoveredModels = models;
     },
+    setHoldRunStream(hold) {
+      holdRunStream = hold;
+      if (!hold) {
+        for (const stream of heldRunStreams) stream.close();
+      }
+    },
     resetObservations() {
       discoveryAuthHeaders.length = 0;
       discoveryRequestBodies.length = 0;
       refreshAuthHeaders.length = 0;
+      runStreamClosed = Promise.withResolvers<void>();
+    },
+    waitForRunStreamClose() {
+      return runStreamClosed.promise;
     },
     getDiscoveryAuthHeaders() {
       return [...discoveryAuthHeaders];
@@ -276,6 +344,49 @@ async function testProxyStartStop(modules: TestModules) {
   console.log("[test] Proxy stop OK");
 }
 
+async function testStreamCancellationStopsBridge(
+  modules: TestModules,
+  backend: TestCursorBackend,
+) {
+  console.log("[test] Testing SSE cancellation teardown...");
+  backend.resetObservations();
+  backend.setHoldRunStream(true);
+  const controller = new AbortController();
+  try {
+    const port = await modules.startProxy(async () => "test-token");
+    const fetchTimeout = setTimeout(() => controller.abort(), 10_000);
+    let res: Response;
+    try {
+      res = await fetch(`http://localhost:${port}/v1/chat/completions`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          model: "composer-2",
+          stream: true,
+          messages: [{ role: "user", content: "hello" }],
+        }),
+        signal: controller.signal,
+      });
+    } finally {
+      clearTimeout(fetchTimeout);
+    }
+    assertEqual(res.status, 200, "Expected a streaming response");
+
+    controller.abort();
+    // The bridge subprocess holds the only handle on this Run stream, so its
+    // closure proves the subprocess was killed.
+    await withTimeout(
+      backend.waitForRunStreamClose(),
+      "the Cursor Run stream to close after the client disconnects",
+    );
+  } finally {
+    controller.abort();
+    modules.stopProxy();
+    backend.setHoldRunStream(false);
+  }
+  console.log("[test] SSE cancellation teardown OK");
+}
+
 async function testAuthParams(modules: TestModules) {
   console.log("[test] Generating auth params...");
   const params = await modules.generateCursorAuthParams();
@@ -533,6 +644,7 @@ async function main() {
 
   try {
     await testProxyStartStop(modules);
+    await testStreamCancellationStopsBridge(modules, backend);
     await testAuthParams(modules);
     await testTokenExpiry(modules);
     await testPluginShape(modules);