From 7991031727f3d8981080a80c42bebce9d26dcc50 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:39:12 +0900 Subject: [PATCH 1/3] fix(live): keep sideband admission until the upstream closes after client cancellation Client cancellation no longer releases the turn admission lease directly. It still stops acquisition through req.signal, and every exit releases the lease when the upstream transport closes, including a join cancelled while its upstream handshake is pending. --- scripts/test-layout/layout-clients.json | 1 + src/server/index/serve-options.ts | 6 +- structure/runtime.md | 2 +- tests/fixtures/test-layout-expected.json | 1 + tests/server/server-live-cancel.test.ts | 148 +++++++++++++++++++++++ 5 files changed, 154 insertions(+), 4 deletions(-) create mode 100644 tests/server/server-live-cancel.test.ts 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 c680d02e2cb..e0e173e2ba9 100644 --- a/src/server/index/serve-options.ts +++ b/src/server/index/serve-options.ts @@ -1766,9 +1766,10 @@ 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); + // Client cancellation reaches acquisition through req.signal (relay resolution, the + // upstream dial). It must not release the lease: a pending join's upstream transport + // still owns the admission, and each exit below releases it on close/CLOSED. const releaseAcquisition = () => { - detachClientAbort(); acquisition?.clear(); if (audioController) unregisterTurn(audioController); else turnAdmissionLease.release(); @@ -1887,7 +1888,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 ebeac303a54..1455ed11f1b 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; client cancellation stops acquisition but never releases the lease itself (`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 c7858ea3657..bb0e1f6ced3 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1183,6 +1183,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..528d6cc1ee8 --- /dev/null +++ b/tests/server/server-live-cancel.test.ts @@ -0,0 +1,148 @@ +/** + * 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. + */ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { existsSync, mkdirSync } from "node:fs"; +import { connect } from "node:net"; +import { join } from "node:path"; +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 }); From 47a4900bfd52fd1bf914acd546baf4dd054d81ed Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 27 Oct 2026 01:04:53 +0900 Subject: [PATCH 2/3] fix(live): retire a cancelled sideband join promptly until the upstream dial Keep the request cancellation binding while relay resolution runs (no transport exists yet) and detach it just before the upstream dial, from which point close/CLOSED owns the release. Adds a regression for cancellation during credential resolution. --- src/server/index/serve-options.ts | 7 +-- structure/runtime.md | 2 +- tests/server/server-live-cancel.test.ts | 68 ++++++++++++++++++++++++- 3 files changed, 71 insertions(+), 6 deletions(-) diff --git a/src/server/index/serve-options.ts b/src/server/index/serve-options.ts index e0e173e2ba9..5b53777590b 100644 --- a/src/server/index/serve-options.ts +++ b/src/server/index/serve-options.ts @@ -1766,10 +1766,10 @@ export function createServeOptions(ctx: ServeOptionsContext) { if (audioController) registerTurn(audioController, turnAdmissionLease); const acquisition = audioController ? clearableDeadline(120_000, AbortSignal.any([req.signal, audioController.signal])) : undefined; - // Client cancellation reaches acquisition through req.signal (relay resolution, the - // upstream dial). It must not release the lease: a pending join's upstream transport - // still owns the admission, and each exit below releases it on close/CLOSED. + // Cancellation retires the turn until the dial; then the transport owns it until close/CLOSED. + const detachClientAbort = turnAdmissionLease.bindAbortSignal(req.signal); const releaseAcquisition = () => { + detachClientAbort(); acquisition?.clear(); if (audioController) unregisterTurn(audioController); else turnAdmissionLease.release(); @@ -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, diff --git a/structure/runtime.md b/structure/runtime.md index 1455ed11f1b..e126b553e5b 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, failed attachment and a client that disconnects while the join is pending; client cancellation stops acquisition but never releases the lease itself (`tests/server/server-live-cancel.test.ts`). 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/server/server-live-cancel.test.ts b/tests/server/server-live-cancel.test.ts index 528d6cc1ee8..ffc3503e528 100644 --- a/tests/server/server-live-cancel.test.ts +++ b/tests/server/server-live-cancel.test.ts @@ -1,12 +1,14 @@ /** * 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. + * 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, test } from "bun:test"; +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"; @@ -146,3 +148,65 @@ test("a client that cancels a pending sideband join keeps admission until the up 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 }); From 96d25ab3e38e1c8e64b8d06fe26533fec3b8b207 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:08:40 +0900 Subject: [PATCH 3/3] style(live): keep serve-options under the file-size limit after merging dev --- src/server/index/serve-options.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/server/index/serve-options.ts b/src/server/index/serve-options.ts index 7855956a5a6..8a89b16b610 100644 --- a/src/server/index/serve-options.ts +++ b/src/server/index/serve-options.ts @@ -1767,8 +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; - // Cancellation retires the turn until the dial; then the transport owns it until close/CLOSED. - 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();