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

@@ -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)).
 

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<RequestDebugResponseLog | undefined> | undefined;
+		const h2Completion = Promise.withResolvers<void>();
+		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<void> => {
+				const log = await debugResponseLogPromise;
+				await log?.close();
+			};
 
-			await new Promise<void>((resolve, reject) => {
-				resolveH2 = resolve;
-
-				const closeDebugLog = async (): Promise<void> => {
-					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) {

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",
+		});
+	});
+});

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<void>();
+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();
+}