From ed5ec3ebdcf2ed8d8ddae9d0eb4c00a9dc6846e9 Mon Sep 17 00:00:00 2001 From: roboomp Date: Wed, 15 Jul 2026 15:46:34 +0000 Subject: [PATCH] fix(ai): handled cursor tls session errors Attached the Cursor HTTP/2 session error path to the provider stream completion before starting the request. Added a subprocess regression that resets the TLS handshake and verifies the turn surfaces a stream error without terminating the process. Fixes #5593 --- packages/ai/CHANGELOG.md | 1 + packages/ai/src/providers/cursor.ts | 99 +++++++++---------- .../ai/test/cursor-transport-error.test.ts | 24 +++++ packages/ai/test/fixtures/cursor-tls-reset.ts | 39 ++++++++ 4 files changed, 112 insertions(+), 51 deletions(-) create mode 100644 packages/ai/test/cursor-transport-error.test.ts create mode 100644 packages/ai/test/fixtures/cursor-tls-reset.ts diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 54982aad0..abdfa99cc 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixed +- Fixed Cursor TLS connection resets escaping the provider stream as process-fatal uncaught exceptions, allowing the active turn to fail or retry without terminating the session. ([#5593](https://github.com/can1357/oh-my-pi/issues/5593)) - Parsed Ollama NDJSON response bytes directly instead of decoding and buffering every network chunk as text. ([#5542](https://github.com/can1357/oh-my-pi/issues/5542)) - Fixed Amazon Bedrock stream error handling for non-`Error` values that `JSON.stringify` cannot serialize ([#5539](https://github.com/can1357/oh-my-pi/issues/5539)). diff --git a/packages/ai/src/providers/cursor.ts b/packages/ai/src/providers/cursor.ts index 93ea77378..e259d62b1 100644 --- a/packages/ai/src/providers/cursor.ts +++ b/packages/ai/src/providers/cursor.ts @@ -351,6 +351,8 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( let h2Request: http2.ClientHttp2Stream | null = null; let heartbeatTimer: NodeJS.Timeout | null = null; let debugResponseLogPromise: Promise | undefined; + const h2Completion = Promise.withResolvers(); + let resolveH2: (() => void) | undefined = h2Completion.resolve; try { const apiKey = options?.apiKey; @@ -406,6 +408,7 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( } else { h2Client = http2.connect(baseUrl); } + h2Client.on("error", h2Completion.reject); h2Request = h2Client.request(requestHeaders); @@ -449,8 +452,6 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( conversationStateCache.set(conversationId, checkpoint); }; - let resolveH2: (() => void) | undefined; - h2Request.on("response", headers => { debugResponseLogPromise = debugSession?.openResponseLog( `HTTP/2 ${headers[":status"] ?? ""}`.trim(), @@ -517,8 +518,6 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( } }); - h2Request.write(frameConnectMessage(requestBytes)); - const sendHeartbeat = () => { if (!h2Request || h2Request.closed) { return; @@ -530,58 +529,56 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( h2Request.write(frameConnectMessage(heartbeatBytes)); }; - heartbeatTimer = setInterval(sendHeartbeat, 5000); + const closeDebugLog = async (): Promise => { + const log = await debugResponseLogPromise; + await log?.close(); + }; - await new Promise((resolve, reject) => { - resolveH2 = resolve; - - const closeDebugLog = async (): Promise => { - const log = await debugResponseLogPromise; - await log?.close(); - }; - - h2Request!.on("trailers", trailers => { - const status = trailers["grpc-status"]; - const msg = trailers["grpc-message"]; - if (status && status !== "0") { - void closeDebugLog().finally(() => { - reject( - new AIError.ProviderResponseError( - `gRPC error ${status}: ${decodeURIComponent(String(msg || ""))}`, - { kind: "envelope" }, - ), - ); - }); - } - }); - - h2Request!.on("end", () => { - resolveH2 = undefined; - void closeDebugLog() - .then(() => { - if (endStreamError) { - reject(endStreamError); - return; - } - resolve(); - }) - .catch(reject); - }); - - h2Request!.on("error", error => { - void closeDebugLog().finally(() => reject(error)); - }); - - if (options?.signal) { - options.signal.addEventListener("abort", () => { - h2Request?.close(); - void closeDebugLog().finally(() => { - reject(new AIError.AbortError()); - }); + h2Request.on("trailers", trailers => { + const status = trailers["grpc-status"]; + const msg = trailers["grpc-message"]; + if (status && status !== "0") { + void closeDebugLog().finally(() => { + h2Completion.reject( + new AIError.ProviderResponseError( + `gRPC error ${status}: ${decodeURIComponent(String(msg || ""))}`, + { kind: "envelope" }, + ), + ); }); } }); + h2Request.on("end", () => { + resolveH2 = undefined; + void closeDebugLog() + .then(() => { + if (endStreamError) { + h2Completion.reject(endStreamError); + return; + } + h2Completion.resolve(); + }) + .catch(h2Completion.reject); + }); + + h2Request.on("error", error => { + void closeDebugLog().finally(() => h2Completion.reject(error)); + }); + + if (options?.signal) { + options.signal.addEventListener("abort", () => { + h2Request?.close(); + void closeDebugLog().finally(() => { + h2Completion.reject(new AIError.AbortError()); + }); + }); + } + + h2Request.write(frameConnectMessage(requestBytes)); + heartbeatTimer = setInterval(sendHeartbeat, 5000); + await h2Completion.promise; + endCurrentTextBlock(output, stream, state); endCurrentThinkingBlock(output, stream, state); if (state.currentToolCall) { diff --git a/packages/ai/test/cursor-transport-error.test.ts b/packages/ai/test/cursor-transport-error.test.ts new file mode 100644 index 000000000..de9dfaebd --- /dev/null +++ b/packages/ai/test/cursor-transport-error.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "bun:test"; +import * as path from "node:path"; + +describe("Cursor transport errors", () => { + it("surfaces a TLS handshake reset as a stream error", async () => { + const child = Bun.spawn([process.execPath, path.join(import.meta.dir, "fixtures/cursor-tls-reset.ts")], { + cwd: path.resolve(import.meta.dir, "../../.."), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([ + new Response(child.stdout).text(), + new Response(child.stderr).text(), + child.exited, + ]); + + expect(exitCode).toBe(0); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + eventTypes: ["start", "error"], + stopReason: "error", + }); + }); +}); diff --git a/packages/ai/test/fixtures/cursor-tls-reset.ts b/packages/ai/test/fixtures/cursor-tls-reset.ts new file mode 100644 index 000000000..5213b160d --- /dev/null +++ b/packages/ai/test/fixtures/cursor-tls-reset.ts @@ -0,0 +1,39 @@ +import * as net from "node:net"; +import { streamCursor } from "@oh-my-pi/pi-ai/providers/cursor"; +import type { Context, Model } from "@oh-my-pi/pi-ai/types"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; + +const server = net.createServer(socket => socket.resetAndDestroy()); +const listening = Promise.withResolvers(); +server.once("error", listening.reject); +server.listen(0, "127.0.0.1", listening.resolve); +await listening.promise; + +const address = server.address(); +if (!address || typeof address === "string") throw new Error("TCP server did not bind"); + +const model: Model<"cursor-agent"> = buildModel({ + id: "cursor-reset-fixture", + name: "Cursor reset fixture", + api: "cursor-agent", + provider: "cursor", + baseUrl: `https://127.0.0.1:${address.port}`, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 1, + maxTokens: 1, +}); +const context: Context = { + messages: [{ role: "user", content: "trigger TLS reset", timestamp: Date.now() }], +}; + +try { + const stream = streamCursor(model, context, { apiKey: "test-token" }); + const eventTypes: string[] = []; + for await (const event of stream) eventTypes.push(event.type); + const result = await stream.result(); + process.stdout.write(`${JSON.stringify({ eventTypes, stopReason: result.stopReason })}\n`); +} finally { + server.close(); +}