diff --git a/src/adapters/cursor/h2-pool.ts b/src/adapters/cursor/h2-pool.ts index 078b5e6440..36e94e7566 100644 --- a/src/adapters/cursor/h2-pool.ts +++ b/src/adapters/cursor/h2-pool.ts @@ -1,4 +1,5 @@ import http2 from "node:http2"; +import { registerOptionalShutdownHook } from "../../lib/optional-shutdown-hooks"; const DEFAULT_MAX_SESSIONS = 8; const SESSION_CLOSE_TIMEOUT_MS = 2_000; @@ -10,9 +11,12 @@ interface PoolEntry { } /** - * HTTP/2 connection pool for Cursor Connect unary/stream calls. - * Sessions are keyed by origin (scheme+host+port) and reused across - * GetUsableModels / Run requests to avoid fresh TCP+TLS per call. + * HTTP/2 connection pool for Cursor Connect DISCOVERY calls (GetUsableModels). + * Sessions are keyed by origin (scheme+host+port) and reused to avoid fresh + * TCP+TLS per call. The Run path deliberately dials its own session: Run + * streams are long-lived bidi whose lifecycle/EOF semantics are owned by + * live-transport (see devlog 260822_senpi_cursor_transfer/190 — Run-path + * pooling is a separate, deliberate unit if ever taken). */ export class CursorH2SessionPool { private readonly entries = new Map(); @@ -20,11 +24,23 @@ export class CursorH2SessionPool { constructor(private readonly maxSessions = DEFAULT_MAX_SESSIONS) {} + /** + * Lazily registered on first use so a process that never talks to Cursor registers + * nothing (optional-subsystem doctrine). The seam is synchronous and best-effort; + * shutdown() is fire-and-forget there because lifecycle's drainAndShutdown runs + * under its own absolute deadline. + */ + private armShutdownHook: (() => void) | undefined = () => { + this.armShutdownHook = undefined; + registerOptionalShutdownHook("cursor-h2-pool", () => { void this.shutdown(); }); + }; + request( url: string, headers: http2.OutgoingHttpHeaders, ): http2.ClientHttp2Stream { if (this.closed) throw new Error("Cursor H2 session pool is closed"); + this.armShutdownHook?.(); const origin = new URL(url).origin; const entry = this.usableEntry(origin) ?? this.createEntry(origin); try { diff --git a/src/oauth/cursor.ts b/src/oauth/cursor.ts index d7607cef83..d30bc33b87 100644 --- a/src/oauth/cursor.ts +++ b/src/oauth/cursor.ts @@ -101,6 +101,19 @@ function sleep(ms: number, signal?: AbortSignal): Promise { }); } +/** Terminal poll statuses (T07, senpi PR #905): the login is denied/expired — retrying cannot succeed. */ +const POLL_TERMINAL_STATUSES = new Set([400, 401, 403, 410]); + +export class CursorAuthTerminalError extends Error { + readonly status: number; + + constructor(status: number) { + super(`Cursor login rejected by the auth server (HTTP ${status}); start a new login`); + this.name = "CursorAuthTerminalError"; + this.status = status; + } +} + /** * Poll cursor.com for login completion. 404 = still pending (back off), 200 = tokens. * `baseDelayMs` is injectable so tests can avoid the real 1s cadence; production uses the default. @@ -135,9 +148,17 @@ export async function pollCursorAuth( return { accessToken: data.accessToken, refreshToken: data.refreshToken }; } + // T07: a terminal auth status means the login attempt itself is dead (denied, + // expired, revoked). Fail on the FIRST such response instead of burning the + // 3-strike retry budget and masking the reason behind a generic error. + if (POLL_TERMINAL_STATUSES.has(response.status)) { + throw new CursorAuthTerminalError(response.status); + } + throw new Error(`Cursor auth poll failed: ${response.status}`); } catch (err) { if (signal?.aborted) throw err instanceof Error ? err : new Error("Cursor login cancelled"); + if (err instanceof CursorAuthTerminalError) throw err; consecutiveErrors++; if (consecutiveErrors >= 3) { throw new Error("Too many consecutive errors during Cursor auth polling"); diff --git a/tests/cursor-h2-pool-shutdown.test.ts b/tests/cursor-h2-pool-shutdown.test.ts new file mode 100644 index 0000000000..f832206f80 --- /dev/null +++ b/tests/cursor-h2-pool-shutdown.test.ts @@ -0,0 +1,62 @@ +import http2 from "node:http2"; +import { afterEach, describe, expect, test } from "bun:test"; +import { CursorH2SessionPool } from "../src/adapters/cursor/h2-pool"; +import { + resetOptionalShutdownHooksForTests, + runOptionalShutdownHooks, +} from "../src/lib/optional-shutdown-hooks"; + +afterEach(() => { + resetOptionalShutdownHooksForTests(); +}); + +async function withH2Server(run: (baseUrl: string) => Promise): Promise { + const server = http2.createServer(); + server.on("stream", stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200 }); + // hold the stream open; shutdown must not depend on server cooperation + }); + await new Promise((resolve, reject) => { + const onError = (error: Error) => reject(error); + server.once("error", onError); + server.listen(0, "127.0.0.1", () => { + server.off("error", onError); + resolve(); + }); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("HTTP/2 fixture did not bind a TCP port"); + try { + return await run(`http://127.0.0.1:${address.port}`); + } finally { + await new Promise(resolve => server.close(() => resolve())); + } +} + +describe("CursorH2SessionPool shutdown hook (devlog 120b)", () => { + test("first request() registers a shutdown hook that closes pooled sessions", async () => { + await withH2Server(async baseUrl => { + const pool = new CursorH2SessionPool(); + const stream = pool.request(baseUrl, { ":method": "POST", ":path": "/x" }); + expect(pool.size).toBe(1); + runOptionalShutdownHooks(); + // shutdown() is fire-and-forget in the sync seam; give it a beat to settle. + await new Promise(resolve => setTimeout(resolve, 100)); + expect(pool.size).toBe(0); + expect(() => pool.request(baseUrl, { ":method": "POST", ":path": "/x" })).toThrow(/closed/); + stream.destroy(); + }); + }); + + test("running the hooks twice is safe (idempotent shutdown)", async () => { + await withH2Server(async baseUrl => { + const pool = new CursorH2SessionPool(); + pool.request(baseUrl, { ":method": "POST", ":path": "/x" }).destroy(); + runOptionalShutdownHooks(); + runOptionalShutdownHooks(); + await new Promise(resolve => setTimeout(resolve, 50)); + expect(pool.size).toBe(0); + }); + }); +}); diff --git a/tests/cursor-oauth.test.ts b/tests/cursor-oauth.test.ts index 19abe77e4f..c4f3b2651b 100644 --- a/tests/cursor-oauth.test.ts +++ b/tests/cursor-oauth.test.ts @@ -57,6 +57,32 @@ describe("Cursor OAuth core flow", () => { await expect(pollCursorAuth("uuid", "ver", ctrl.signal, 1)).rejects.toThrow(/cancel/i); }); + test("pollCursorAuth fails on the FIRST terminal status without retrying (T07)", async () => { + for (const status of [400, 401, 403, 410]) { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + return new Response("", { status }); + }) as typeof fetch; + const err = await pollCursorAuth("uuid", "ver", undefined, 1).catch((e: unknown) => e as Error); + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toContain(String(status)); + expect((err as Error).message).toMatch(/new login/i); + expect(calls).toBe(1); + } + }); + + test("pollCursorAuth keeps the 3-strike retry for server errors (500)", async () => { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + return new Response("", { status: 500 }); + }) as typeof fetch; + const err = await pollCursorAuth("uuid", "ver", undefined, 1).catch((e: unknown) => e as Error); + expect((err as Error).message).toMatch(/consecutive errors/i); + expect(calls).toBe(3); + }); + test("refreshCursorToken posts the refresh token as a Bearer and returns new creds", async () => { let seenAuth = ""; globalThis.fetch = (async (_url: string | URL, init?: RequestInit) => {