diff --git a/docs-site/src/content/docs/ko/reference/proxy-formats.md b/docs-site/src/content/docs/ko/reference/proxy-formats.md index 32963d7f7c3..bf854d93543 100644 --- a/docs-site/src/content/docs/ko/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ko/reference/proxy-formats.md @@ -294,6 +294,10 @@ call creation 이후 클라이언트는 다음의 지원되는 모든 inbound 프록시는 업스트림 join URL을 정규화한 뒤, 양방향 텍스트 및 바이너리 프레임을 투명하게 릴레이합니다. 업스트림 인증은 프록시가 소유한 상태로 유지되며, 클라이언트 프로토콜 헤더는 보존됩니다. +sideband 연결을 종료할 때 프록시는 업스트림의 정상 종료를 1초 동안 기다린 뒤, 응답하지 않는 +전송 연결을 강제로 닫습니다. 클라이언트 업그레이드가 실패한 경우에도 실제 업스트림 연결이 +닫힌 것을 확인한 후에만 해당 계정의 활성 통화 용량을 반환합니다. + call creation과 sideband join은 같은 OpenAI 계정으로 이루어져야 하며, 그렇지 않으면 업스트림이 join을 거부합니다(`404`). 두 요청 모두 Codex의 `session-id`와 `thread-id` 헤더를 실어 보냅니다. Pool 모드는 계정 선택을 그 쌍에 묶어 두므로(프로세스 로컬) 프록시에 도착한 join은 통화를 만든 계정을 그대로 쓰고, diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index fe6d7330b18..60609a1947a 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -90,6 +90,10 @@ returns 504, and client cancellation returns 499. Bun does not expose the exact cannot currently be forwarded precisely. A successful connection preserves the initial session frames in order. This handshake policy is separate from the Responses WebSocket transport. +When a sideband connection ends, the proxy allows one second for a graceful upstream +close before terminating an unresponsive transport. It releases the account's active-call +capacity only after that transport has closed, including when the client upgrade fails. + ## Undeclared tool calls On the Responses endpoint, when a request declares a tool catalog, a native passthrough reply diff --git a/scripts/test-layout/layout-clients.json b/scripts/test-layout/layout-clients.json index 34a42e679da..bf99d076825 100644 --- a/scripts/test-layout/layout-clients.json +++ b/scripts/test-layout/layout-clients.json @@ -1,4 +1,5 @@ { + "server-live-close.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/live-sideband.ts b/src/server/index/live-sideband.ts index d0840bdc626..0f022efb6bc 100644 --- a/src/server/index/live-sideband.ts +++ b/src/server/index/live-sideband.ts @@ -184,6 +184,19 @@ function finalizeLiveSideband(ws: ServerWebSocket, upstream?: WebSocket) finally { releaseLiveSidebandAdmission(ws); } } +function forceCloseLiveSidebandUpstream(upstream: WebSocket): void { + // Bun 1.4.4 waits for the peer's closing handshake; another close() while + // CLOSING does not finish it. Stop the transport after the graceful window, + // but leave admission release to the close event or an observed CLOSED state. + try { + const socket = upstream as WebSocket & { terminate?: () => void }; + if (typeof socket.terminate === "function") socket.terminate(); + else upstream.close(1000, "upstream close timeout"); + } catch { + /* retain ownership until transport closure is confirmed */ + } +} + function armLiveSidebandCloseFallback(ws: ServerWebSocket, upstream: WebSocket): void { if (ws.data.liveCloseFallback !== undefined) return; ws.data.liveCloseFallback = setTimeout(() => { @@ -193,13 +206,7 @@ function armLiveSidebandCloseFallback(ws: ServerWebSocket, upstream: Web finalizeLiveSideband(ws, upstream); return; } - // A close frame was already sent below. Retry once, but never surrender - // native-main ownership while the authenticated transport remains live. - try { - upstream.close(1000, "upstream close timeout"); - } catch { - /* upstream is already unusable */ - } + forceCloseLiveSidebandUpstream(upstream); // Some implementations transition synchronously without delivering the // close event. That is still an observed CLOSED transport and is safe to // finalize. CONNECTING/CLOSING peers keep the lease so profile switching @@ -238,17 +245,13 @@ export function closeLiveSidebandBeforeUpgrade( releaseOnce(); return; } - try { - upstream.close(1000, "upstream close timeout"); - } catch { - /* retain ownership until CLOSED is observed */ - } + forceCloseLiveSidebandUpstream(upstream); if ((upstream.readyState as number) === 3) releaseOnce(); }, LIVE_SIDEBAND_CLOSE_FALLBACK_MS); try { upstream.close(code, reason); } catch { - /* the bounded fallback retries without releasing ownership */ + /* the bounded fallback stops the transport before releasing ownership */ } if ((upstream.readyState as number) === 3) releaseOnce(); } @@ -272,12 +275,12 @@ export function closeLiveSideband(ws: ServerWebSocket, code = 1000, reas // The sideband holds a native-main admission lease. Do not release it just // because the downstream left: its authenticated upstream remains live // until the close event arrives or the transport is observed CLOSED. The - // bounded fallback only retries close; it does not release ownership. + // bounded fallback terminates the transport; it cannot release a live lease. armLiveSidebandCloseFallback(ws, upstream); try { upstream.close(code, reason); } catch { - /* the fallback retries close without releasing ownership */ + /* the fallback stops the transport without releasing a live lease */ } } try { diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index 6a10afe729e..a2ae495c10e 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -64,7 +64,7 @@ export interface WsData { liveUpstreamHandoff?: LiveSidebandUpstreamHandoff; /** Once teardown starts, ignore new client frames until the upstream closes. */ liveClosing?: boolean; - /** Schedules one bounded close retry without surrendering native-main ownership. */ + /** Terminates a stalled close without releasing ownership before transport closure. */ liveCloseFallback?: ReturnType; /** Turn/account ownership retained for the complete sideband socket lifetime. */ liveTurnAdmissionLease?: AdmissionLease; diff --git a/structure/runtime.md b/structure/runtime.md index bc6383fb32c..9cc046d63a1 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -430,7 +430,7 @@ Responses route normalization resolves provider summary defaults from the origin `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. -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. +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. `OCX_LIVE_FRAME_LOG` records both frame metadata and sideband lifecycle stages (`upstream-open`, `upstream-failed`, `relay-attached`, `relay-closed`) in one JSONL, content-free in both shapes. The lifecycle half is what separates a join that never reached this proxy from one whose upstream handshake was refused and from a live relay that carried nothing; frame records alone leave all three as an empty file. `tests/server/server-live-realtime-fixtures.test.ts` drives each sideband stage against de-identified Frameless v3 fixtures in `tests/fixtures/realtime-voice-sideband/` so a failure names the stage. diff --git a/tests/codex-integration/codex-stored-main-duplicate.test.ts b/tests/codex-integration/codex-stored-main-duplicate.test.ts index daba4f87eaf..765d27957fb 100644 --- a/tests/codex-integration/codex-stored-main-duplicate.test.ts +++ b/tests/codex-integration/codex-stored-main-duplicate.test.ts @@ -104,6 +104,7 @@ describe("stored Pool duplicate of native main", () => { saveCodexAccountCredential(poolId, { accessToken: jwt("plus", email, "other-user"), refreshToken: "fake-refresh", chatgptAccountId: "shared-account", expiresAt: Date.now() + 3600_000 }); const changed = await runPassiveRead(() => listCodexAuthAccounts(config)); + expect(changed.find(row => row.id === poolId)?.upstreamAccountKey).toMatch(/^uai1_[0-9a-f]{64}$/); expect(changed.find(row => row.id === poolId)?.upstreamAccountKey).not.toBe(main.upstreamAccountKey); expect(changed.find(row => row.id === "__main__")?.upstreamAccountKey).toBe(main.upstreamAccountKey); }); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index ffd401ece74..84e73b24969 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1178,6 +1178,7 @@ "combo-requested-effort.test.ts": "server", "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-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-close.test.ts b/tests/server/server-live-close.test.ts new file mode 100644 index 00000000000..ec6b2612a24 --- /dev/null +++ b/tests/server/server-live-close.test.ts @@ -0,0 +1,188 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import type { ServerWebSocket } from "bun"; +import { createHash } from "node:crypto"; +import { createServer, type AddressInfo, type Socket } from "node:net"; +import { + attachLiveSidebandUpstream, + closeLiveSideband, + closeLiveSidebandBeforeUpgrade, +} from "../../src/server/index/live-sideband"; +import type { WsData } from "../../src/server/ws-bridge"; + +const proxyKeys = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"]; +let savedProxyEnv: Record; +beforeEach(() => { + savedProxyEnv = Object.fromEntries(proxyKeys.map(key => [key, process.env[key]])); + for (const key of proxyKeys) delete process.env[key]; +}); +afterEach(() => { + for (const key of proxyKeys) { + if (savedProxyEnv[key] === undefined) delete process.env[key]; + else process.env[key] = savedProxyEnv[key]; + } +}); + +type Phase = "before upgrade" | "attached"; +type TerminableSocket = WebSocket & { terminate(): void }; + +function closingOwner(phase: Phase, upstream: WebSocket) { + const releasedStates: number[] = []; + let finishes = 0; + const released = Promise.withResolvers(); + const release = () => { releasedStates.push(upstream.readyState); released.resolve(); }; + const data: WsData = { + kind: "live-sideband", liveUpstream: upstream, liveOpened: true, + liveUpstreamHandoff: { failure: () => undefined, take: () => ({ ok: true, frames: [] }) }, + liveTurnAdmissionLease: { release } as WsData["liveTurnAdmissionLease"], + liveFinish: () => { finishes++; }, + }; + const downstream = { data, readyState: WebSocket.OPEN, close() {}, send: () => 1 } as unknown as ServerWebSocket; + if (phase === "attached") attachLiveSidebandUpstream(downstream); + return { + data, releasedStates, finishes: () => finishes, + close() { + if (phase === "attached") closeLiveSideband(downstream, 1001, "client finished"); + else closeLiveSidebandBeforeUpgrade(upstream, release, 1001, "client finished"); + }, + async waitForRelease() { + let timer: ReturnType | undefined; + try { + await Promise.race([ + released.promise, + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("sideband lease remained held")), 3_000); }), + ]); + } finally { clearTimeout(timer); } + }, + }; +} + +/** A real TCP peer that optionally acknowledges Close; Bun.serve always acknowledges it. */ +async function withClosePeer(answerClose: boolean, run: (fixture: { + upstream: TerminableSocket; + closes: Array<{ code: number; reason: string }>; + terminations(): number; + peerClosed: Promise; +}) => Promise) { + const peers = new Set(); + const closes: Array<{ code: number; reason: string }> = []; + const peerClosed = Promise.withResolvers(); + const server = createServer(peer => { + peers.add(peer); + peer.on("error", () => {}); + peer.on("close", () => { peers.delete(peer); peerClosed.resolve(); }); + let handshake = ""; + let opened = false; + let frames = Buffer.alloc(0); + peer.on("data", chunk => { + if (!opened) { + handshake += chunk.toString(); + if (!handshake.includes("\r\n\r\n")) return; + const key = /sec-websocket-key:\s*([^\r\n]+)/i.exec(handshake)?.[1]; + const accept = createHash("sha1").update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest("base64"); + opened = true; + peer.write("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + + `Sec-WebSocket-Accept: ${accept}\r\n\r\n`); + return; + } + frames = Buffer.concat([frames, chunk]); + if (frames.length < 2) return; + const length = frames[1]! & 0x7f; + const masked = (frames[1]! & 0x80) !== 0; + const offset = masked ? 6 : 2; + // The fixture only receives a single, short Close control frame. + if ((frames[0]! & 0x0f) !== 8 || length > 125 || frames.length < offset + length) return; + const payload = Buffer.from(frames.subarray(offset, offset + length)); + if (masked) for (let i = 0; i < payload.length; i++) payload[i] = payload[i]! ^ frames[2 + i % 4]!; + closes.push({ code: payload.readUInt16BE(0), reason: payload.subarray(2).toString() }); + frames = Buffer.alloc(0); + if (answerClose) peer.end(Buffer.concat([Buffer.from([0x88, payload.length]), payload])); + }); + }); + let upstream: TerminableSocket | undefined; + try { + await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); + upstream = new WebSocket(`ws://127.0.0.1:${(server.address() as AddressInfo).port}`) as TerminableSocket; + let terminations = 0; + const terminate = upstream.terminate.bind(upstream); + upstream.terminate = () => { terminations++; terminate(); }; + await new Promise((resolve, reject) => { + upstream!.addEventListener("open", () => resolve(), { once: true }); + upstream!.addEventListener("error", () => reject(new Error("fixture websocket failed to open")), { once: true }); + }); + await run({ upstream, closes, terminations: () => terminations, peerClosed: peerClosed.promise }); + } finally { + upstream?.terminate(); + for (const peer of peers) peer.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } +} + +describe.each(["before upgrade", "attached"] as const)("live sideband close %s", phase => { + // The pinned 1.4.3 runner reports close before the peer responds. Exercise this + // transport regression under the bundled runtime as well as the ordinary suite. + test.skipIf(!Bun.semver.satisfies(Bun.version, ">=1.4.4"))("terminates a silent peer before releasing its lease", async () => { + await withClosePeer(false, async fixture => { + const owner = closingOwner(phase, fixture.upstream); + owner.close(); + expect(owner.releasedStates).toEqual([]); + expect(fixture.upstream.readyState).toBe(WebSocket.CLOSING); + await owner.waitForRelease(); + await fixture.peerClosed; + expect(fixture.terminations()).toBe(1); + expect(fixture.closes).toEqual([{ code: 1001, reason: "client finished" }]); + expect(owner.releasedStates).toEqual([WebSocket.CLOSED]); + if (phase === "attached") { + expect(owner.data.liveUpstream).toBeUndefined(); + expect(owner.data.liveCloseFallback).toBeUndefined(); + expect(owner.finishes()).toBe(1); + } + }); + }); + + test("preserves a graceful close and cancels forced termination", async () => { + await withClosePeer(true, async fixture => { + const owner = closingOwner(phase, fixture.upstream); + owner.close(); + await owner.waitForRelease(); + await fixture.peerClosed; + await Bun.sleep(1_100); + expect(fixture.closes).toEqual([{ code: 1001, reason: "client finished" }]); + expect(fixture.terminations()).toBe(0); + expect(owner.releasedStates).toEqual([WebSocket.CLOSED]); + }); + }); +}); + +test.each(["missing", "throws", "still closing", "closed"] as const)("termination %s preserves actual-close ownership", async mode => { + class Upstream extends EventTarget { + readyState: number = WebSocket.OPEN; + closes = 0; + terminations = 0; + terminate?: () => void; + close() { this.closes++; this.readyState = WebSocket.CLOSING; } + finish() { this.readyState = WebSocket.CLOSED; this.dispatchEvent(new CloseEvent("close", { code: 1000 })); } + } + const cases = (["before upgrade", "attached"] as const).map(phase => { + const upstream = new Upstream(); + if (mode !== "missing") upstream.terminate = () => { + upstream.terminations++; + if (mode === "throws") throw new Error("fixture termination failed"); + if (mode === "closed") upstream.readyState = WebSocket.CLOSED; + }; + const owner = closingOwner(phase, upstream as unknown as WebSocket); + owner.close(); + expect(owner.releasedStates).toEqual([]); + return { upstream, owner }; + }); + try { + await Bun.sleep(1_100); + for (const { upstream, owner } of cases) { + expect(upstream.closes).toBe(mode === "missing" ? 2 : 1); + expect(upstream.terminations).toBe(mode === "missing" ? 0 : 1); + expect(owner.releasedStates).toEqual(mode === "closed" ? [WebSocket.CLOSED] : []); + upstream.finish(); + upstream.finish(); + expect(owner.releasedStates).toEqual([WebSocket.CLOSED]); + } + } finally { for (const { upstream } of cases) upstream.finish(); } +}); diff --git a/tests/server/server-live.test.ts b/tests/server/server-live.test.ts index 28c68c04a1b..cc95a52e006 100644 --- a/tests/server/server-live.test.ts +++ b/tests/server/server-live.test.ts @@ -22,6 +22,7 @@ import { MAX_WS_FRAME_BYTES, openLiveSidebandUpstream, startServer, + type StartServerDeps, } from "../../src/server"; import { activeRegistryMetrics, @@ -32,6 +33,7 @@ import { 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"; import { phaseTimer } from "../helpers/phase-timing"; import { expectSidebandUpgrade, openSidebandClient, redirectSidebandWebSocket, sidebandRelayUpstream } from "../helpers/sideband-relay-probe"; @@ -149,6 +151,15 @@ function forwardConfig(): OcxConfig { } as OcxConfig; } +// Isolate relay/readiness fixtures from host service-manager probes. +// Explicit per-case dependencies override this default. +function startTestServer(port?: number, deps: StartServerDeps = {}): ReturnType { + return startServer(port, { + inspectNativeCodexOwnership: ownedServiceHomeInspection("server-live fixture"), + ...deps, + }); +} + function expectReadyProtocolMetadata(body: Record, managementUrl: string): void { expect(body).toMatchObject({ protocol: 1, @@ -192,7 +203,7 @@ test("POST /v1/live rewrites ChatGPT multipart into backend realtime/calls JSON" config.providers.openai!.baseUrl = "https://chatgpt.com/backend-api/codex///"; saveConfig(config); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody("v=0-offer", { model: "gpt-live", instructions: "hi" }); const response = await fetch(new URL("/v1/live", server.url), { @@ -241,7 +252,7 @@ test.each(["/v1/live", "/backend-api/codex/live"])("POST %s relays to an OpenAI }, } as OcxConfig); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody(); const response = await fetch(new URL(path, server.url), { @@ -272,7 +283,7 @@ test("call-create forwards Frameless protocol headers and keeps auth pool-owned" const upstream = fakeLiveUpstream(captured); saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody("v=0-offer", { model: "gpt-live", instructions: "hi" }); const response = await fetch(new URL("/v1/live", server.url), { @@ -320,7 +331,7 @@ test("call-create without protocol headers does not invent them upstream", async const upstream = fakeLiveUpstream(captured); saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody("v=0-offer", { model: "gpt-live", instructions: "hi" }); const response = await fetch(new URL("/v1/live", server.url), { @@ -353,7 +364,7 @@ test("POST /v1/realtime/calls is accepted and relays like /v1/live", async () => const upstream = fakeLiveUpstream(captured, 201, "/v1/realtime/calls/rtc_codex"); saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody("v=0-codex"); const response = await fetch(new URL("/v1/realtime/calls", server.url), { @@ -384,7 +395,7 @@ test("ChatGPT multipart rewrite allows SDP-only offers without session", async ( const upstream = fakeLiveUpstream(captured); saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody("v=0-sdp-only", null); const response = await fetch(new URL("/v1/live", server.url), { @@ -407,7 +418,7 @@ test("ChatGPT multipart rewrite allows SDP-only offers without session", async ( test("OPTIONS preflight allows ChatGPT-Account-Id for voice clients", async () => { saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const response = await fetch(new URL("/v1/live", server.url), { method: "OPTIONS", @@ -440,7 +451,7 @@ test("POST /v1/live without an OpenAI upstream returns 400", async () => { }, } as OcxConfig); - const server = startServer(0); + const server = startTestServer(0); try { const response = await fetch(new URL("/v1/live", server.url), { method: "POST", @@ -457,7 +468,7 @@ test("POST /v1/live without an OpenAI upstream returns 400", async () => { test("GET /v1/live still hits the unknown-endpoint JSON 404 guard", async () => { saveConfig(forwardConfig()); - const server = startServer(0); + const server = startTestServer(0); try { const response = await fetch(new URL("/v1/live", server.url), { method: "GET", @@ -500,7 +511,7 @@ test("a routed pool account's token overrides the caller bearer on the live rela chatgptAccountId: "acct-pool-a", }); - const server = startServer(0); + const server = startTestServer(0); try { const { body, contentType } = multipartLiveBody(); const response = await fetch(new URL("/v1/live", server.url), { @@ -579,7 +590,7 @@ test("call-create and its sideband join bind to the same pool account (openai/co } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); try { const voiceHeaders = { authorization: `Bearer ${DIRECT_CHATGPT_TOKEN}`, @@ -671,7 +682,7 @@ test("sideband GET /v1/live/{callId} relays the exact frame ceiling bidirectiona // helper because this case is at its size cap and the repo answer is not to compress. const { OriginalWebSocket, restore } = redirectSidebandWebSocket(upstream.port); restoreWebSocket = restore; - const server = live = startServer(0); + const server = live = startTestServer(0); const ws = openSidebandClient(OriginalWebSocket, server.url, "/v1/live/rtc_sideband", DIRECT_CHATGPT_TOKEN); client = ws; // Each leg is its own segment and the timer's ticks read the peer's event count, so a @@ -764,7 +775,7 @@ test("standalone GET /v1/realtime?intent=quicksilver&model= upgrades and relays } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); try { const wsUrl = new URL(`/v1/realtime?intent=quicksilver&model=gpt-realtime-1.5&access_token=dropme`, server.url); wsUrl.protocol = "ws:"; @@ -844,7 +855,7 @@ test("standalone GET /v1/live?model= upgrades and relays bidirectionally", async } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); try { const wsUrl = new URL(`/v1/live?model=gpt-realtime-1.5`, server.url); wsUrl.protocol = "ws:"; @@ -888,7 +899,7 @@ test("standalone realtime upgrade keeps the auth and origin guards", async () => saveConfig(forwardConfig()); process.env.OPENCODEX_API_AUTH_TOKEN = "required-secret"; - const server = startServer(0); + const server = startTestServer(0); try { const upgradeHeaders = { connection: "upgrade", @@ -1211,7 +1222,7 @@ test("sideband relay preserves multibyte UTF-8 frames byte-identically in both d } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); try { const wsUrl = new URL(`/v1/live/rtc_korean`, server.url); wsUrl.protocol = "ws:"; @@ -1315,7 +1326,7 @@ test("sideband frame log preserves delivery without recording damaged or clean t } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); try { const wsUrl = new URL(`/v1/live/rtc_framelog`, server.url); wsUrl.protocol = "ws:"; @@ -1396,7 +1407,7 @@ describe("GET /readyz", () => { for (const { outcome, expectedStatus, expectedHttp } of cases) { const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const pending = await fetch(new URL("/readyz", server.url)); expect(pending.status).toBe(503); @@ -1420,7 +1431,7 @@ describe("GET /readyz", () => { test("fresh gate is pending; /healthz 200 while /readyz 503 with Retry-After", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const base = server.url; // Liveness is immediate and unchanged. @@ -1459,7 +1470,7 @@ describe("GET /readyz", () => { runtimeRole: "hub", hub: { managementPublicOrigin: "https://hub.example.test" }, }); - const server = startServer(0); + const server = startTestServer(0); try { const ready = await fetch(new URL("/readyz", server.url), { headers: { @@ -1484,7 +1495,7 @@ describe("GET /readyz", () => { test("/readyz is 200 with status ready only after gate.markReady()", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const base = server.url; expect(((await (await fetch(new URL("/readyz", base))).json()) as { status: string }).status).toBe("pending"); @@ -1504,7 +1515,7 @@ describe("GET /readyz", () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); gate.markFailed(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const res = await fetch(new URL("/readyz", server.url)); expect(res.status).toBe(503); @@ -1518,7 +1529,7 @@ describe("GET /readyz", () => { test("exact method/path: POST /readyz and GET /readyz/ are NOT matched", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const base = server.url; // POST must not be accepted as readiness (falls through to the JSON 404 guard). @@ -1560,7 +1571,7 @@ describe("GET /readyz", () => { test("/readyz answers without an admission token (unauthenticated, like /healthz)", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { // No Authorization header, no API auth token in env (beforeEach deletes it). const res = await fetch(new URL("/readyz", server.url), { headers: { Authorization: "" } }); @@ -1576,7 +1587,7 @@ describe("GET /readyz", () => { test("ephemeral startServer(0): /readyz reports the actual bound port (not the requested 0)", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { const boundPort = new URL(server.url).port; // The bound port is a real ephemeral port, not the requested 0. @@ -1601,8 +1612,8 @@ describe("per-server readiness gate isolation", () => { saveConfig(forwardConfig()); const gateA: ReadinessGate = createReadinessGate(); const gateB: ReadinessGate = createReadinessGate(); - const serverA = startServer(0, { readinessGate: gateA }); - const serverB = startServer(0, { readinessGate: gateB }); + const serverA = startTestServer(0, { readinessGate: gateA }); + const serverB = startTestServer(0, { readinessGate: gateB }); try { gateA.markReady(); // gateB stays pending. @@ -1624,8 +1635,8 @@ describe("per-server readiness gate isolation", () => { saveConfig(forwardConfig()); const gateA: ReadinessGate = createReadinessGate(); const gateB: ReadinessGate = createReadinessGate(); - const serverA = startServer(0, { readinessGate: gateA }); - const serverB = startServer(0, { readinessGate: gateB }); + const serverA = startTestServer(0, { readinessGate: gateA }); + const serverB = startTestServer(0, { readinessGate: gateB }); try { gateA.markReady(); gateB.markFailed(); @@ -1646,7 +1657,7 @@ describe("per-server readiness gate isolation", () => { test("a server with no gate passed in still serves pending /readyz (fresh private gate)", async () => { saveConfig(forwardConfig()); // No gate supplied → startServer creates a fresh pending private gate. - const server = startServer(0); + const server = startTestServer(0); try { const res = await fetch(new URL("/readyz", server.url)); expect(res.status).toBe(503); @@ -1661,7 +1672,7 @@ describe("per-server readiness gate isolation", () => { const gateA: ReadinessGate = createReadinessGate(); const gateB: ReadinessGate = createReadinessGate(); // Ephemeral start: A binds a real kernel-assigned port (NOT a fixed user port). - const serverA = startServer(0, { readinessGate: gateA }); + const serverA = startTestServer(0, { readinessGate: gateA }); try { gateA.markReady(); const occupiedPort = Number(new URL(serverA.url).port); @@ -1672,7 +1683,7 @@ describe("per-server readiness gate isolation", () => { // Binding the occupied port with a separate gate must throw (EADDRINUSE). // It must never silently share A's listener or reset A's gate. - expect(() => startServer(occupiedPort, { readinessGate: gateB })).toThrow(); + expect(() => startTestServer(occupiedPort, { readinessGate: gateB })).toThrow(); // A is unaffected by the failed bind: still HTTP 200 / ready. const resA = await fetch(new URL("/readyz", serverA.url)); @@ -1702,7 +1713,7 @@ describe("GET /readyz while draining", () => { test("a ready gate reports pending (503) once the listener starts draining", async () => { saveConfig(forwardConfig()); const gate = createReadinessGate(); - const server = startServer(0, { readinessGate: gate }); + const server = startTestServer(0, { readinessGate: gate }); try { gate.markReady(); const base = server.url; @@ -2024,7 +2035,7 @@ test("a failed pre-upgrade handshake retains admission until its CLOSING upstrea saveConfig(forwardConfig()); const upstream = new FakeUpstreamSocket(); upstream.closeMode = "closing"; - const server = startServer(0, { + const server = startTestServer(0, { liveSidebandWebSocketFactory: () => { queueMicrotask(() => upstream.emit("error", {})); return upstream as unknown as WebSocket; @@ -2068,7 +2079,7 @@ test("a failed pre-upgrade handoff retains admission until its CLOSING upstream saveConfig(forwardConfig()); const upstream = new FakeUpstreamSocket(); upstream.closeMode = "closing"; - const server = startServer(0, { + const server = startTestServer(0, { liveSidebandWebSocketFactory: () => { queueMicrotask(() => { upstream.emit("open", {}); @@ -2141,7 +2152,7 @@ test("a sideband join whose upstream handshake fails never opens the client sock } } as typeof WebSocket; - const server = startServer(0); + const server = startTestServer(0); const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; try { const wsUrl = new URL(`/v1/realtime?call_id=rtc_dead_call`, server.url); @@ -2190,7 +2201,7 @@ test("a sideband join whose upstream handshake fails never opens the client sock test("an upstream that opens then closes before relay attachment refuses the client and releases admission", async () => { saveConfig(forwardConfig()); const upstream = new FakeUpstreamSocket(); - const server = startServer(0, { + const server = startTestServer(0, { liveSidebandWebSocketFactory: () => { queueMicrotask(() => { upstream.emit("open", {});