fix(live): bound WebSocket close without releasing live ownership

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은 통화를 만든 계정을 그대로 쓰고,

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

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",

src/server/index/live-sideband.ts

@@ -184,6 +184,19 @@ function finalizeLiveSideband(ws: ServerWebSocket<WsData>, 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<WsData>, upstream: WebSocket): void {
   if (ws.data.liveCloseFallback !== undefined) return;
   ws.data.liveCloseFallback = setTimeout(() => {
@@ -193,13 +206,7 @@ function armLiveSidebandCloseFallback(ws: ServerWebSocket<WsData>, 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<WsData>, 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 {

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<typeof setTimeout>;
   /** Turn/account ownership retained for the complete sideband socket lifetime. */
   liveTurnAdmissionLease?: AdmissionLease;

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.
 

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);
   });

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",

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<string, string | undefined>;
+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<void>();
+  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<WsData>;
+  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<typeof setTimeout> | undefined;
+      try {
+        await Promise.race([
+          released.promise,
+          new Promise<never>((_, 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<void>;
+}) => Promise<void>) {
+  const peers = new Set<Socket>();
+  const closes: Array<{ code: number; reason: string }> = [];
+  const peerClosed = Promise.withResolvers<void>();
+  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<void>((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<void>((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<void>(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(); }
+});