From 82270fbe69a59793e9a898810f50dd692b2028e8 Mon Sep 17 00:00:00 2001 From: Noam Kushinsky Date: Mon, 3 Aug 2026 15:02:52 +0300 Subject: [PATCH 1/2] fix: handle cancelled SSE streams --- src/proxy.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/proxy.ts b/src/proxy.ts index 93c44ec..7a871ab 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -1364,10 +1364,10 @@ function createBridgeStreamResponse( const completionId = `chatcmpl-${crypto.randomUUID().replace(/-/g, "").slice(0, 28)}`; const created = Math.floor(Date.now() / 1000); + let closed = false; 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`)); @@ -1542,6 +1542,9 @@ function createBridgeStreamResponse( } }); }, + cancel() { + closed = true; + }, }); return new Response(stream, { headers: SSE_HEADERS }); From 67cd2f491d50a7fb512d9d27c66c41d24bd6f97d Mon Sep 17 00:00:00 2001 From: Noam Kushinsky Date: Wed, 19 Aug 2026 00:28:44 +0300 Subject: [PATCH 2/2] fix: tear down the bridge when an SSE stream is cancelled Muting the controller stopped the crash but left the bridge fully alive: heartbeats kept writing every 5s and bridge.onData kept running, so a disconnected client's exec would still register the session in activeBridges with no follow-up ever coming to resume it. Killing is what actually stops an in-flight run. bridge.end() only half-closes the request side; the bridge keeps reading the response half and the subprocess survives. cancel() is a no-op on an already-closed stream so it cannot tear down a bridge paused for tool-result continuation. --- src/proxy.ts | 23 ++++++++-- test/smoke.ts | 114 +++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 132 insertions(+), 5 deletions(-) diff --git a/src/proxy.ts b/src/proxy.ts index 7a871ab..be8e75c 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -1365,6 +1365,7 @@ function createBridgeStreamResponse( const created = Math.floor(Date.now() / 1000); let closed = false; + let cancelStream: (() => void) | undefined; const stream = new ReadableStream({ start(controller) { const encoder = new TextEncoder(); @@ -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, @@ -1503,10 +1520,8 @@ function createBridgeStreamResponse( sendSSE(makeChunk({}, "stop")); sendSSE(makeUsageChunk()); sendDone(); + stopBridge(); closeController(); - activeBridges.delete(bridgeKey); - clearInterval(heartbeatTimer); - bridge.end(); } }, ); @@ -1543,7 +1558,7 @@ function createBridgeStreamResponse( }); }, cancel() { - closed = true; + cancelStream?.(); }, }); diff --git a/test/smoke.ts b/test/smoke.ts index 53d948a..0384bc0 100644 --- a/test/smoke.ts +++ b/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; close: () => Promise; } @@ -54,6 +58,27 @@ function assertArrayEqual( } } +async function withTimeout( + promise: Promise, + message: string, + timeoutMs = 10_000, +): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, 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,6 +93,23 @@ 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 { let discoveryMode: DiscoveryMode = "success"; let discoveredModels: Array<{ id: string; name: string; reasoning?: boolean }> = [ @@ -75,6 +117,9 @@ async function createTestCursorBackend(): Promise { ]; const discoveryAuthHeaders: string[] = []; const discoveryRequestBodies: Uint8Array[] = []; + let runStreamClosed = Promise.withResolvers(); + const heldRunStreams = new Set(); + let holdRunStream = false; const refreshAuthHeaders: string[] = []; const refreshServer = http.createServer((req, res) => { @@ -109,11 +154,24 @@ async function createTestCursorBackend(): Promise { 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 { 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(); + }, + 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);