From b745b5912a6bc46c550ae634a5cce73641fa562c Mon Sep 17 00:00:00 2001 From: Fred Amartey <43480311+FredAmartey@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:15:20 -0400 Subject: [PATCH] fix(responses): notice a Codex WebSocket that dies after the response started The exchange stopped pinging once the first Responses event committed the 200, so a path that died without a FIN or RST kept the committed stream silent until codex-rs's 300 s stream_idle_timeout. Keep the pinger for the whole exchange, send one probe with the create frame, and end a committed stream whose peer answered a ping and then sent no frame and no pong for two intervals. --- .../docs/reference/codex-retry-diagnostics.md | 33 +++ scripts/test-layout/layout.json | 2 +- src/server/responses/codex-ws-exchange.ts | 48 +++- src/server/responses/codex-ws-wire.ts | 7 +- structure/transports/responses-wire-shapes.md | 2 +- tests/fixtures/test-layout-expected.json | 2 +- tests/responses/ws-committed-liveness.test.ts | 230 ++++++++++++++++++ tests/responses/ws-upstream.test.ts | 8 +- 8 files changed, 315 insertions(+), 17 deletions(-) create mode 100644 tests/responses/ws-committed-liveness.test.ts diff --git a/docs-site/src/content/docs/reference/codex-retry-diagnostics.md b/docs-site/src/content/docs/reference/codex-retry-diagnostics.md index d2e2ee8c2e3..a4eb1ec9722 100644 --- a/docs-site/src/content/docs/reference/codex-retry-diagnostics.md +++ b/docs-site/src/content/docs/reference/codex-retry-diagnostics.md @@ -48,6 +48,39 @@ that the account pool routed to the main account, the earliest known time anothe pool account recovers, whichever comes first (never sooner than one minute). Resume the Goal after that time; routing then picks an account from fresh state. +## Triage: `idle timeout waiting for SSE` + +Codex ends a stream that has sent no event for its `stream_idle_timeout_ms` +(300 seconds by default) and retries the turn. The report alone does not say +why the stream went quiet. On the native ChatGPT upstream WebSocket, one +possible cause is a path that failed silently after the response started: a VPN +or tunnel restart, a Wi-Fi rejoin or an expired NAT mapping can drop it without +a FIN or RST reaching the proxy. + +On the canonical ChatGPT Codex endpoint +(`https://chatgpt.com/backend-api/codex/responses`, the exchange with the +metadata channel), the proxy pings the upstream for the whole exchange. Once the +response has started, a peer that has answered at least one ping on that +exchange and then delivers no frame and no pong for two ping intervals ends the +stream with a retryable failure. Codex then reconnects 30 to 45 seconds after +the last frame or pong instead of after 300 seconds, and the reconnect row names +the cause: + +```text +Reconnecting... 1/5 (stream disconnected before completion: Upstream stream terminated unexpectedly: codex websocket stopped answering after the response started [cause=after-response-started ...]) +``` + +When a ping reaches a path that a restarted tunnel resets, the same row reports +`codex websocket closed before a Responses terminal event (close 1006)` +instead, within about one ping interval. An `idle timeout waiting for SSE` that +still arrives after the full timeout comes from a peer that kept answering pings +while sending no events, a peer that never answered a ping, or an exchange +without the metadata channel, such as the OpenAI API Responses WebSocket, which +sends no pings. All three keep the client's idle deadline on purpose. +`ocx logs get --json` shows the attempt's `codexWsStage` with its +`pings` and `pongs` counts, which tell the three apart. Raising +`stream_idle_timeout_ms` only makes them wait longer. + ## Why the first reconnect row can be absent The public Codex `rust-v0.158.0-alpha.2.1` source gates the ordinary stream-retry diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 2fdded5a374..f3a4bcc5fd8 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1953,7 +1953,7 @@ "ws-capacity-reclamation.test.ts": "responses", "ws-capacity-tool-ownership.test.ts": "responses", "ws-endpoint.test.ts": "responses", - "ws-failure-stage.test.ts": "responses", + "ws-failure-stage.test.ts": "responses", "ws-committed-liveness.test.ts": "responses", "ws-native-injection.test.ts": "responses", "ws-native-result-continuations.test.ts": "responses", "ws-native-steering.test.ts": "responses", diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 3e09e434c79..11e0c8b5422 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -8,7 +8,8 @@ import { CodexWsMetadata, type CodexWsQuotaObserver } from "./codex-ws-metadata" import { CODEX_RESPONSES_HTTP_URL, type PreparedCodexWsRequest } from "./codex-ws-request"; import { CodexWsCorrelation } from "./codex-ws-correlation"; import type { CodexWsSession } from "./codex-ws-session"; -import { UPGRADE_DEADLINE_MS, CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, MAX_CODEX_WS_FRAME_BYTES, +import { UPGRADE_DEADLINE_MS, CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, + CODEX_WS_COMMITTED_SILENT_INTERVALS, MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, markCodexWsResponse, markCodexWsPreludeProjection, normalizeResponsesWsRelayEvent, closedBeforeTerminalMessage, codexWsCreateFrameExceedsLimit, codexWsFailureDetail, codexWsPreResponseFailure, markCodexWsStage, codexWsOcxVersion, markCodexWsSocketDeath, codexWsWrapperStatus, isCodexWsTerminalEvent, type CodexWsFailureStage, type CodexWsStageRecord, @@ -155,8 +156,11 @@ export function codexWsExchange(options: ExchangeOptions): Promise { // Liveness while waiting for the first response event (metadata path only): the // silence timer is re-armed by every inbound frame or pong; the pinger runs on a fixed // interval so a peer that answers pings can never trip the silence bound while alive. + // After the commit, each ping tick checks whether any frame or pong arrived since the last. let silenceTimer: ReturnType | undefined; let pingTimer: ReturnType | undefined; + let inboundAtLastTick = 0; + let silentIntervals = 0; // The resolved 200, retained so a later body failure can replace its // success-shaped stage record with the failure-shaped one (#4191). let committedResponse: Response | null = null; @@ -260,6 +264,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { markCodexWsStage(response, stageRecord(null)); committedResponse = response; resolve(response); + startCommittedLiveness(); }; /** @@ -377,18 +382,40 @@ export function codexWsExchange(options: ExchangeOptions): Promise { CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, ); }; - /** Ping on a fixed interval until the response starts; a socket without ping() is never pinged. */ + const pingable = () => typeof (ws as WebSocket & { ping?: unknown }).ping === "function"; + /** False when ping() threw. It may also close the socket and settle the exchange. */ + const sendPing = (): boolean => { + if (!pingable()) return false; + try { (ws as WebSocket & { ping: (data?: string) => void }).ping(); } catch { return false; } + pings += 1; + return true; + }; + /** Ping on a fixed interval for the life of the exchange; a socket without ping() is never pinged. */ const schedulePing = () => { - const ping = (ws as WebSocket & { ping?: (data?: string) => void }).ping; - if (typeof ping !== "function" || responseCommitted || terminal) return; + if (!pingable() || terminal) return; pingTimer = setTimeout(() => { - if (responseCommitted || terminal) return; - try { ping.call(ws); } catch { return; } - pings += 1; - // ping() may close the socket synchronously and settle the exchange; re-check. + if (terminal) return; + if (responseCommitted && pongs > 0) { + const inbound = upstreamFrames + pongs; + silentIntervals = inbound === inboundAtLastTick ? silentIntervals + 1 : 0; + inboundAtLastTick = inbound; + if (silentIntervals >= CODEX_WS_COMMITTED_SILENT_INTERVALS) { + failStream(`codex websocket stopped answering after the response started${codexWsFailureDetail(failureStage())}`); + return; + } + } + sendPing(); if (!terminal) schedulePing(); }, CODEX_WS_LIVENESS_PING_INTERVAL_MS); }; + /** Only a peer that has ponged is judged by its silence, so probe again if none has yet. */ + const startCommittedLiveness = () => { + if (!metadata || terminal || !pingable()) return; + inboundAtLastTick = upstreamFrames + pongs; + silentIntervals = 0; + if (pongs === 0) sendPing(); + if (!terminal) schedulePing(); + }; const onPong = () => { if (terminal) return; pongs += 1; @@ -559,7 +586,10 @@ export function codexWsExchange(options: ExchangeOptions): Promise { if (!metadata) commitResponse(); else if (!responseCommitted && !terminal) { armSilence(); - schedulePing(); + // Probe now, not one interval in: the path can die seconds after the send, before a + // held commit, and the committed check needs a pong from before that. + sendPing(); + if (!terminal) schedulePing(); } }; diff --git a/src/server/responses/codex-ws-wire.ts b/src/server/responses/codex-ws-wire.ts index 4327d7662ee..77065d2af28 100644 --- a/src/server/responses/codex-ws-wire.ts +++ b/src/server/responses/codex-ws-wire.ts @@ -25,6 +25,11 @@ export const UPGRADE_DEADLINE_MS = 10_000; // keeps exactly the previous 90 s bound; a peer that does can never trip it while alive. export const CODEX_WS_LIVENESS_PING_INTERVAL_MS = 15_000; export const CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS = 90_000; +// The pinger keeps running after the commit: a tunnel restart or Wi-Fi rejoin can drop the path +// without a FIN or RST, and codex-rs would wait out its 300 s idle timeout. A peer that has ponged +// and then sends no frame and no pong for this many intervals is gone, and the stream fails so the +// client retries. One missed pong is not a verdict. +export const CODEX_WS_COMMITTED_SILENT_INTERVALS = 2; // Keep the push-based WS transport inside the same memory envelope as the // bounded SSE relays that consume this response. Unlike fetch response bodies, // a WebSocket cannot be paused when a ReadableStream applies backpressure, so @@ -404,7 +409,7 @@ export type CodexWsFailureStage = { firstResponseMs: number | null; /** Milliseconds from send to this failure; null when the failure predates the send. */ elapsedMs: number | null; - /** Liveness pings the exchange sent while waiting for the first response event. */ + /** Liveness pings the exchange sent, before and after the first response event. */ pings: number; /** Pongs the peer answered with; zero on a peer that never answers pings. */ pongs: number; diff --git a/structure/transports/responses-wire-shapes.md b/structure/transports/responses-wire-shapes.md index 1578413634c..4bbe49be3e4 100644 --- a/structure/transports/responses-wire-shapes.md +++ b/structure/transports/responses-wire-shapes.md @@ -480,7 +480,7 @@ A standalone no-response exchange has a 90-second prelude deadline in addition to the upgrade deadline. That prelude deadline is a ceiling, not a floor: the exchange runs under the caller's abort signal, so a `connectTimeoutMs` shorter than 90 seconds cancels -an already-sent create before the prelude timer fires. +an already-sent create before the prelude timer fires. After the first Responses event an exchange with the metadata channel keeps pinging on the same interval, and sends one ping at the commit unless an earlier one was answered; a peer that has answered a ping on that exchange and then delivers no frame and no pong for two consecutive intervals is treated as a dead socket, so the committed stream fails as a body error that the relay ends with its `upstream_reset` `response.failed` tail. This check never cuts a peer that keeps answering, however long the model is silent; that peer, like one that never answered a ping, still meets the client's own idle deadline (`tests/responses/ws-committed-liveness.test.ts`). These are transport-fidelity guarantees, not a provider-billing guarantee. Every exchange also leaves a content-free stage record (`CodexWsStageRecord`, #4191): create-frame bytes (measured on failure only — the committed-success record keeps it null so the happy path never byte-counts a megabyte replay frame), send completion, numeric close code, elapsed, first-frame and first-response-event durations, frame counters, liveness ping/pong counts, pool reuse, the OCX/Bun versions, and (#6740) `terminalEvent` — the normalized `response.completed` / `response.failed` / `response.incomplete` / `error` frame that ended a plain exchange, observed before any synthetic tail — with `terminalErrorStatus`, the wrapper status of a terminal `error` frame. Both are null when no terminal frame arrived, on native-control exchanges (which can end on an acknowledgement), and on rows written before they existed; malformed persisted values read as null without dropping the stage. The exchange pins the record on the resolved Response (`markCodexWsStage`, the same marker seam as `markCodexWsResponse`); `handleResponses` adopts it onto the serving attempt, and usage.jsonl persists it per attempt behind a drop-guard normalizer, so hand-edited rows cannot inject strings into the DTO. Later snapshots update the same response-local record in place, so an attempt holding the committed reference observes final success or failure counters. Each exchange supplies a complete fresh snapshot; separate responses keep distinct records. On eager-relay cancel-drain expiry, upstream cancellation finalizes the transport snapshot before the cancellation hook writes the usage row; an actual terminal observed within the drain still wins over cancellation. The record never carries conversation text, headers, close-reason text, or account identifiers, and it is not a fallback-eligibility signal: nothing it says permits a resend. The one replacement an operator can grant after a socket dies is the resend gate's decision (see [ambiguous-resend gate](responses-failover.md#ambiguous-resend-gate)). diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d41fccb67a1..a16a622fe6d 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1340,7 +1340,7 @@ "ws-capacity-reclamation.test.ts": "responses", "ws-capacity-tool-ownership.test.ts": "responses", "ws-endpoint.test.ts": "responses", - "ws-failure-stage.test.ts": "responses", + "ws-failure-stage.test.ts": "responses", "ws-committed-liveness.test.ts": "responses", "ws-native-injection.test.ts": "responses", "ws-native-result-continuations.test.ts": "responses", "ws-native-steering.test.ts": "responses", diff --git a/tests/responses/ws-committed-liveness.test.ts b/tests/responses/ws-committed-liveness.test.ts new file mode 100644 index 00000000000..3b80460a6a3 --- /dev/null +++ b/tests/responses/ws-committed-liveness.test.ts @@ -0,0 +1,230 @@ +import { afterEach, beforeEach, describe, expect, jest, test } from "bun:test"; +import { CODEX_WS_LIVENESS_PING_INTERVAL_MS, codexWsUpstreamFetch } from "../../src/server/responses/ws-upstream"; +import { codexWsPool } from "../../src/server/responses/codex-ws-pool"; +import { BOUNDED_WS_RUNTIME, streamingInit } from "../helpers/ws-upstream-fixtures"; + +/** + * Pinging used to stop at the commit, so a path that died after `response.created` left the stream + * silent until codex-rs's 300 s idle timeout. A peer that ponged and then goes silent is now gone; + * one that keeps answering, or never answered, keeps the stream as before. + */ + +const CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"; +const INTERVAL = CODEX_WS_LIVENESS_PING_INTERVAL_MS; + +type Listener = (event: unknown) => void; + +/** Minimal scriptable stand-in for Bun's WebSocket, mirroring `ws-pre-output-refusal.test.ts`. */ +class FakeWebSocket { + static instances: FakeWebSocket[] = []; + static script: (ws: FakeWebSocket) => void = () => {}; + url: string; + sent: string[] = []; + pings = 0; + answersPings = true; + pingThrows = false; + closed = false; + listeners = new Map(); + + constructor(url: string) { + this.url = url; + FakeWebSocket.instances.push(this); + queueMicrotask(() => FakeWebSocket.script(this)); + } + + addEventListener(type: string, listener: Listener) { + const list = this.listeners.get(type) ?? []; + list.push(listener); + this.listeners.set(type, list); + } + + removeEventListener(type: string, listener: Listener) { + this.listeners.set(type, (this.listeners.get(type) ?? []).filter(value => value !== listener)); + } + + emit(type: string, event: unknown = {}) { + for (const listener of this.listeners.get(type) ?? []) listener(event); + } + + frame(payload: unknown) { + this.emit("message", { data: JSON.stringify(payload) }); + } + + ping() { + if (this.pingThrows) throw new Error("WebSocket is not open"); + this.pings += 1; + if (this.answersPings) this.emit("pong", {}); + } + + send(data: string) { + this.sent.push(data); + } + + close() { this.closed = true; } +} + +const RealWebSocket = globalThis.WebSocket; + +beforeEach(() => { + FakeWebSocket.instances = []; + FakeWebSocket.script = () => {}; + jest.useFakeTimers(); +}); + +afterEach(() => { + jest.useRealTimers(); + globalThis.WebSocket = RealWebSocket; + codexWsPool.dispose(); +}); + +const noFallback = (async () => { + throw new Error("fallback must not run after open"); +}) as unknown as typeof fetch; + +const RATE_LIMITS = { type: "codex.rate_limits", rate_limits: { primary: { used_percent: 10, window_minutes: 10080 } } }; +const CREATED = { type: "response.created", response: { id: "resp_1", status: "in_progress", output: [] } }; +const IN_PROGRESS = { type: "response.in_progress", response: { id: "resp_1", status: "in_progress", output: [] } }; +const SUMMARY_DELTA = { type: "response.reasoning_summary_text.delta", item_id: "rs_1", summary_index: 0, delta: "…" }; +const COMPLETED = { type: "response.completed", response: { id: "resp_1", status: "completed", output: [] } }; + +const flush = async () => { + for (let turn = 0; turn < 20; turn++) await Promise.resolve(); +}; + +/** + * Quota, an empty preamble, then nothing: the shape that hit the 300 s timeout. `holdMs` turns on + * the Pool pre-output hold; `diesInPreamble` kills the path before that hold commits. + */ +async function committedExchange(answersPings: boolean, { holdMs = 0, diesInPreamble = false } = {}) { + globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket; + FakeWebSocket.script = ws => { + ws.answersPings = answersPings; + ws.emit("open", {}); + ws.frame(RATE_LIMITS); + ws.frame(CREATED); + ws.frame(IN_PROGRESS); + if (diesInPreamble) ws.answersPings = false; + }; + const hold = holdMs > 0 ? { maxMs: holdMs, until: Number.MAX_SAFE_INTEGER } : undefined; + const pending = codexWsUpstreamFetch( + CODEX_URL, streamingInit(), noFallback, BOUNDED_WS_RUNTIME, + undefined, undefined, undefined, undefined, undefined, hold, + ); + if (hold) { + await flush(); + jest.advanceTimersByTime(holdMs); + } + const response = await pending; + expect(response.status).toBe(200); + const ws = FakeWebSocket.instances[0]!; + // A reader, not response.text(): its reads settle on microtasks, so fake timers can step them. + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let outcome: { text: string } | { error: string } | undefined; + void (async () => { + let text = ""; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + outcome = { text }; + } catch (error) { + outcome = { error: error instanceof Error ? error.message : String(error) }; + } + })(); + /** One ping interval per step, so each chained timer is scheduled and fired. */ + const advance = async (intervals: number, each?: () => void) => { + for (let step = 0; step < intervals; step++) { + jest.advanceTimersByTime(INTERVAL); + each?.(); + await flush(); + } + await flush(); + }; + return { ws, advance, outcome: () => outcome }; +} + +describe("codex ws committed-stream liveness", () => { + test("a socket that stops answering after the response started is settled long before the client's idle timeout", async () => { + const { ws, advance, outcome } = await committedExchange(true); + // The path dies without a close. + ws.answersPings = false; + await advance(1); + // One unanswered ping is not a verdict. + expect(outcome()).toBeUndefined(); + await advance(2); + const settled = outcome(); + expect(settled).toBeDefined(); + if (!settled || !("error" in settled)) throw new Error(`expected a body failure, got ${JSON.stringify(settled)}`); + expect(settled.error).toContain("stopped answering after the response started"); + expect(settled.error).toContain("cause=after-response-started"); + expect(settled.error).toMatch(/pongs=1\]/); + expect(ws.closed).toBe(true); + expect(ws.sent).toHaveLength(1); + }); + + test("a socket whose ping() starts throwing is still judged by its silence", async () => { + const { ws, advance, outcome } = await committedExchange(true); + ws.pingThrows = true; + await advance(3); + const settled = outcome(); + if (!settled || !("error" in settled)) throw new Error(`expected a body failure, got ${JSON.stringify(settled)}`); + expect(settled.error).toContain("stopped answering after the response started"); + }); + + test("a peer that keeps answering pings is never cut, however long the model stays silent", async () => { + const { ws, advance, outcome } = await committedExchange(true); + // 300 s without a message frame, every ping answered. + await advance(20); + expect(outcome()).toBeUndefined(); + ws.frame(COMPLETED); + await advance(0); + const settled = outcome(); + if (!settled || !("text" in settled)) throw new Error(`expected a completed body, got ${JSON.stringify(settled)}`); + expect(settled.text).toContain("event: response.completed"); + // The pinger ends with the exchange. + const pings = ws.pings; + await advance(4); + expect(ws.pings).toBe(pings); + expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true); + }); + + test("frames are proof of life on their own: a streaming peer that stopped answering pings is not cut", async () => { + const { ws, advance, outcome } = await committedExchange(true); + ws.answersPings = false; + await advance(10, () => ws.frame(SUMMARY_DELTA)); + expect(outcome()).toBeUndefined(); + ws.frame(COMPLETED); + await advance(0); + const settled = outcome(); + if (!settled || !("text" in settled)) throw new Error(`expected a completed body, got ${JSON.stringify(settled)}`); + expect(settled.text).toContain("event: response.completed"); + }); + + test("a peer that has never answered a ping keeps the committed stream exactly as before", async () => { + const { ws, advance, outcome } = await committedExchange(false); + await advance(20); + // No pong ever came back, so silence proves nothing. + expect(outcome()).toBeUndefined(); + expect(ws.closed).toBe(false); + ws.frame(COMPLETED); + await advance(0); + const settled = outcome(); + if (!settled || !("text" in settled)) throw new Error(`expected a completed body, got ${JSON.stringify(settled)}`); + expect(settled.text).toContain("event: response.completed"); + }); + + test("a path that dies while the pre-output hold keeps the 200 back is still caught after the commit", async () => { + // The path dies seconds after the send, before a 10 s hold commits. + const { ws, advance, outcome } = await committedExchange(true, { holdMs: 10_000, diesInPreamble: true }); + await advance(1); + expect(outcome()).toBeUndefined(); + await advance(1); + const settled = outcome(); + if (!settled || !("error" in settled)) throw new Error(`expected a body failure, got ${JSON.stringify(settled)}`); + expect(settled.error).toContain("stopped answering after the response started"); + expect(ws.closed).toBe(true); + }); +}); diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index 8079c64309f..9d587853a5f 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -1558,9 +1558,9 @@ describe("codexWsUpstreamFetch", () => { const pending = codexWsUpstreamFetch(CODEX_URL, streamingInit(), noResend(counter)); await opened.promise; const ws = FakeWebSocket.instances[0]; - // Seven steps: 105 s of no message frames, well past the 90 s bound, every step ponged. + // Seven steps plus the send probe: 105 s of no message frames, past the 90 s bound. for (let step = 0; step < 7; step++) jest.advanceTimersByTime(CODEX_WS_LIVENESS_PING_INTERVAL_MS); - expect(pingsSent).toBe(7); + expect(pingsSent).toBe(8); ws.emit("message", { data: JSON.stringify({ type: "response.created", response: { id: "r1" } }) }); ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: { id: "r1" } }) }); const response = await pending; @@ -1568,9 +1568,9 @@ describe("codexWsUpstreamFetch", () => { expect(await response.text()).toContain("event: response.completed"); expect(ws.sent).toHaveLength(1); expect(counter.http).toBe(0); - // The pinger stops once the response has started. + // The pinger stops once the exchange settles. jest.advanceTimersByTime(CODEX_WS_LIVENESS_PING_INTERVAL_MS * 4); - expect(pingsSent).toBe(7); + expect(pingsSent).toBe(8); expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true); } finally { jest.useRealTimers();