diff --git a/scripts/test-layout/layout-clients.json b/scripts/test-layout/layout-clients.json index 9bf6607a9ab..5420987b6e6 100644 --- a/scripts/test-layout/layout-clients.json +++ b/scripts/test-layout/layout-clients.json @@ -2,6 +2,7 @@ "agent-clients.test.ts": "clients", "workbuddy-selector-recovery.test.ts": "clients", "server-live-close.test.ts": "server", + "server-live-cancel.test.ts": "server", "link-diagnostics.test.ts": "clients", "link-tunnel-history.test.ts": "clients", "link-readiness.test.ts": "clients", diff --git a/src/server/index/serve-options.ts b/src/server/index/serve-options.ts index 781e73c7441..8a89b16b610 100644 --- a/src/server/index/serve-options.ts +++ b/src/server/index/serve-options.ts @@ -1767,7 +1767,7 @@ export function createServeOptions(ctx: ServeOptionsContext) { if (audioController) registerTurn(audioController, turnAdmissionLease); const acquisition = audioController ? clearableDeadline(120_000, AbortSignal.any([req.signal, audioController.signal])) : undefined; - const detachClientAbort = turnAdmissionLease.bindAbortSignal(req.signal); + const detachClientAbort = turnAdmissionLease.bindAbortSignal(req.signal); // detached at the dial; then close/CLOSED owns release const releaseAcquisition = () => { detachClientAbort(); acquisition?.clear(); @@ -1820,6 +1820,7 @@ export function createServeOptions(ctx: ServeOptionsContext) { finally { finalizeLiveRequest(499, { closeReason: "client_cancel" }); } return withCors(formatErrorResponse(499, "client_closed_request", "Audio connection canceled"), req, policy); } + detachClientAbort(); const upstreamHandshake = await openLiveSidebandUpstream( resolved.upstreamWsUrl, resolved.headers, @@ -1888,7 +1889,6 @@ export function createServeOptions(ctx: ServeOptionsContext) { return withCors(formatErrorResponse(502, "upstream_error", "Audio WebSocket upgrade failed"), req, policy); } if (upgraded) { - detachClientAbort(); acquisition?.clear(); finalizeLiveRequest(101); return undefined as unknown as Response; diff --git a/structure/runtime.md b/structure/runtime.md index 35e50832d4e..3075c4720ec 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -428,7 +428,7 @@ Responses route normalization resolves provider summary defaults from the origin ## Live sideband handshake -`src/server/index/serve-options.ts` establishes the authorized upstream live sideband before accepting the client WebSocket upgrade, and `src/server/index/live-sideband.ts` implements the bounded upstream dial. `openLiveSidebandUpstream` bounds the handshake to ten seconds and retains at most 32 frames and 1 MiB of preamble within the frame limit. `src/server/ws-bridge.ts` defines the runtime handoff carrying captured frames or terminal state. Failed handshakes return 502/504 and client cancellation returns 499; exact upstream 404/410 status is unavailable from Bun's client WebSocket. Admission ownership lasts until upstream close/CLOSED, including failed upgrades and failed attachment. The ordinary Responses WebSocket exchange remains separate. +`src/server/index/serve-options.ts` establishes the authorized upstream live sideband before accepting the client WebSocket upgrade, and `src/server/index/live-sideband.ts` implements the bounded upstream dial. `openLiveSidebandUpstream` bounds the handshake to ten seconds and retains at most 32 frames and 1 MiB of preamble within the frame limit. `src/server/ws-bridge.ts` defines the runtime handoff carrying captured frames or terminal state. Failed handshakes return 502/504 and client cancellation returns 499; exact upstream 404/410 status is unavailable from Bun's client WebSocket. Admission ownership lasts until upstream close/CLOSED, including failed upgrades, failed attachment and a client that disconnects while the join is pending. Before the upstream dial, client cancellation retires the turn at once; from the dial on, only close/CLOSED releases it (`tests/server/server-live-cancel.test.ts`). The ordinary Responses WebSocket exchange remains separate. The relay is transparent in both directions, and that includes the close: a downstream client's close code and reason are carried to the upstream through `clientCloseForUpstream`, which only substitutes 1000 for a code no endpoint may send and truncates the reason to the 123-byte control-frame limit. This matters to the caller, because a Frameless v3 client reads upstream 1000 as the session completing and any other code as a transport loss to reconnect. Both pre-upgrade and attached teardown allow one second for graceful closure, then call Bun's `terminate()`. Admission releases only after close/CLOSED; missing or failing termination cannot release a live lease, and a missing method receives one close retry. `tests/server/server-live-close.test.ts` covers silent real TCP peers on Bun 1.4.4, graceful peers and unconfirmed closure. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index cead7ca3353..e2dafe00d0b 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1185,6 +1185,7 @@ "server-kiro-device-login.test.ts": "server", "server-kiro-oauth-401-replay.test.ts": "server", "server-kiro-refusal-e2e.test.ts": "server", "server-live-existing-call.test.ts": "server", "server-live-close.test.ts": "server", + "server-live-cancel.test.ts": "server", "server-live-frame-log.test.ts": "server", "server-live-realtime-fixtures.test.ts": "server", "server-live.test.ts": "server", "server-loopback-host-gate.test.ts": "server", "server-management-auth.test.ts": "server", diff --git a/tests/server/server-live-cancel.test.ts b/tests/server/server-live-cancel.test.ts new file mode 100644 index 00000000000..ffc3503e528 --- /dev/null +++ b/tests/server/server-live-cancel.test.ts @@ -0,0 +1,212 @@ +/** + * Client cancellation of a live-sideband join: the admission lease belongs to the upstream + * transport, so a caller that disconnects while the upstream handshake is pending must not + * release it before that transport has actually closed. Before any transport exists, the same + * cancellation must retire the turn at once. + */ +import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; +import { existsSync, mkdirSync } from "node:fs"; +import { connect } from "node:net"; +import { join } from "node:path"; +import * as accountStore from "../../src/codex/account-store"; +import { saveConfig } from "../../src/config"; +import { startServer } from "../../src/server"; +import { activeRegistryMetrics } from "../../src/server/lifecycle"; +import type { OcxConfig } from "../../src/types"; +import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; +import { ownedServiceHomeInspection } from "../helpers/owned-service-home-inspection"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +const previousApiToken = process.env.OPENCODEX_API_AUTH_TOKEN; +const previousOpencodexHome = process.env.OPENCODEX_HOME; +const TEST_DIR = join(import.meta.dir, ".tmp-server-live-cancel-test"); +const DIRECT_CHATGPT_TOKEN = fakeChatGptJwt({ chatgpt_account_id: "acct-123" }); +let isolatedCodexHome: IsolatedCodexHome | null = null; + +beforeEach(() => { + if (existsSync(TEST_DIR)) removeTreeWithRetry(TEST_DIR); + mkdirSync(TEST_DIR, { recursive: true }); + process.env.OPENCODEX_HOME = TEST_DIR; + delete process.env.OPENCODEX_API_AUTH_TOKEN; + isolatedCodexHome = installIsolatedCodexHome("ocx-server-live-cancel-codex-"); +}); + +afterEach(() => { + if (previousApiToken === undefined) delete process.env.OPENCODEX_API_AUTH_TOKEN; + else process.env.OPENCODEX_API_AUTH_TOKEN = previousApiToken; + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + isolatedCodexHome?.restore(); + isolatedCodexHome = null; + if (existsSync(TEST_DIR)) removeTreeWithRetry(TEST_DIR); +}); + +/** A sideband upstream that stays CONNECTING and only reaches CLOSED when the test says so. */ +class PendingUpstreamSocket { + private readonly listeners = new Map void>>(); + readyState: number = WebSocket.CONNECTING; + closeCalls = 0; + private readonly closeRequested = Promise.withResolvers(); + readonly closeRequestedPromise = this.closeRequested.promise; + + addEventListener(type: string, listener: (event: { code?: number; reason?: string }) => void): void { + const bucket = this.listeners.get(type) ?? []; + bucket.push(listener); + this.listeners.set(type, bucket); + } + + removeEventListener(type: string, listener: (event: { code?: number; reason?: string }) => void): void { + this.listeners.set(type, (this.listeners.get(type) ?? []).filter(entry => entry !== listener)); + } + + close(): void { + this.closeCalls += 1; + this.readyState = WebSocket.CLOSING; + this.closeRequested.resolve(); + } + + finishClose(): void { + this.readyState = WebSocket.CLOSED; + for (const listener of this.listeners.get("close") ?? []) listener({ code: 1006, reason: "closed after cancellation" }); + } +} + +function forwardConfig(): OcxConfig { + return { + port: 0, + defaultProvider: "openai", + openaiProviderTierVersion: 2, + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + } as OcxConfig; +} + +function withDeadline(promise: Promise, message: string, ms = 5_000): Promise { + let timer: ReturnType | undefined; + return Promise.race([ + promise, + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(message)), ms); }), + ]).finally(() => clearTimeout(timer)); +} + +test("a client that cancels a pending sideband join keeps admission until the upstream closes", async () => { + saveConfig(forwardConfig()); + const upstream = new PendingUpstreamSocket(); + const dialed = Promise.withResolvers(); + const server = startServer(0, { + inspectNativeCodexOwnership: ownedServiceHomeInspection("server-live-cancel fixture"), + liveSidebandWebSocketFactory: () => { + dialed.resolve(); + return upstream as unknown as WebSocket; + }, + }); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + const client = connect(Number(server.port), "127.0.0.1"); + client.on("error", () => {}); + try { + await withDeadline(new Promise(resolve => client.once("connect", () => resolve())), "client never connected"); + client.write([ + "GET /v1/realtime?call_id=rtc_cancel_pending_join HTTP/1.1", + `Host: 127.0.0.1:${server.port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Version: 13", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + `Authorization: Bearer ${DIRECT_CHATGPT_TOKEN}`, + "ChatGPT-Account-Id: acct-123", + "OpenAI-Alpha: quicksilver=v2", + "X-Session-Id: rts_cancel_pending_join", + "", "", + ].join("\r\n")); + await withDeadline(dialed.promise, "sideband join never dialed the upstream"); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore + 1); + + // The caller goes away while the upstream handshake is still pending. + client.destroy(); + await withDeadline(upstream.closeRequestedPromise, "cancellation never asked the upstream to close"); + // Give every abort listener and pending continuation a turn to run. + await Bun.sleep(0); + await Bun.sleep(0); + expect(upstream.readyState).toBe(WebSocket.CLOSING); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore + 1); + + upstream.finishClose(); + await Bun.sleep(0); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + upstream.finishClose(); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + } finally { + client.destroy(); + upstream.finishClose(); + await server.stop(true); + } +}, { timeout: 15_000 }); + +test("a client that cancels during credential resolution releases admission before any dial", async () => { + saveConfig({ + port: 0, + defaultProvider: "openai", + openaiProviderTierVersion: 2, + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "pool", + }, + }, + codexAccounts: [{ id: "pool-cancel", email: "cancel@example.test", isMain: false, chatgptAccountId: "acct-cancel" }], + activeCodexAccountId: "pool-cancel", + codexMainAccountHardLock: false, + } as OcxConfig); + accountStore.saveCodexAccountCredential("pool-cancel", { + accessToken: "fixture-expired-access", refreshToken: "fixture-refresh", expiresAt: 0, chatgptAccountId: "acct-cancel", + }); + // Hold credential preparation open, as a slow token refresh would, without a caller signal. + const entered = Promise.withResolvers(); + const held = Promise.withResolvers(); + const tokenSpy = spyOn(accountStore, "getValidCodexToken").mockImplementation(async () => { + entered.resolve(); + return held.promise; + }); + let dials = 0; + const server = startServer(0, { + inspectNativeCodexOwnership: ownedServiceHomeInspection("server-live-cancel fixture"), + liveSidebandWebSocketFactory: () => { dials += 1; throw new Error("a cancelled join must not dial"); }, + }); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + const client = connect(Number(server.port), "127.0.0.1"); + client.on("error", () => {}); + try { + await withDeadline(new Promise(resolve => client.once("connect", () => resolve())), "client never connected"); + client.write([ + "GET /v1/realtime?model=gpt-live HTTP/1.1", + `Host: 127.0.0.1:${server.port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Version: 13", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "", "", + ].join("\r\n")); + await withDeadline(entered.promise, "credential resolution never started"); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore + 1); + + client.destroy(); + const deadline = Date.now() + 5_000; + while (activeRegistryMetrics().activeTurns.active !== activeTurnsBefore && Date.now() < deadline) await Bun.sleep(5); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + expect(dials).toBe(0); + } finally { + held.reject(new Error("fixture released")); + client.destroy(); + await server.stop(true); + tokenSpy.mockRestore(); + } +}, { timeout: 15_000 });