diff --git a/devlog/_plan/260911_l6_streaming_tools/000_packet.md b/devlog/_plan/260911_l6_streaming_tools/000_packet.md new file mode 100644 index 0000000000..45a0a023b4 --- /dev/null +++ b/devlog/_plan/260911_l6_streaming_tools/000_packet.md @@ -0,0 +1,83 @@ +# Dispatch packet — L6 (revision 5) + +Round unit: `devlog/_plan/260911_lane_dispatch_round` on `dev`. Base freeze: `origin/dev` `6d3ad12e3` (2.51.0). +Five audit rounds shaped this packet. The last one was a seven-lane feasibility check that asked whether each stack is implementable inside its owned paths; three lanes came back with gaps, and the fixes are folded here. `010_lane_partition.md` is the authoritative ownership list; `130_wp4_feasibility.md` records why each path was granted. + + +## Shared frame + +**Repository.** Your worktree is named in your packet, already checked out on your lane branch, cut +from `origin/dev` `6d3ad12e3` (2.51.0). Work only there. Do not add, move, or remove a worktree. + +**Loop.** Run `$codexclaw:cxc-loop` as HOTL for your lane: one work-phase per issue, in order. Your +goal ends when your last PR is green and reported, not when the code looks right. + +**Subagents.** Unlimited `xai/grok-4.6` subagents, read-only, spawned with `spawn_agent` +(`model: "xai/grok-4.6"`). Use them to reproduce, to read the call sites you are about to change, to +find a second caller of a helper you are touching, and to review your staged diff adversarially +before you push. A finding enters your work only with an exact `path:line` anchor. Subagents never +write, commit, push, or call a mutating `gh`. Treat a `fail` verdict the way this round did: fold it +in and re-audit. This packet is at revision 3 because two audit rounds rejected revisions 1 and 2. + +**MUST NOT.** + +- No local product suite: no `bun test`, no `bun run test`, no `bun run test:changed`, no + `bun run typecheck`, no `bun run build:gui`, no `bun install`. Report them as `NOT RUN`. +- No merge, no release, no force-push to a shared branch, no direct push to `dev`. +- No path outside your owned list, including paths a carried PR happens to touch. Dropping a hunk + from a carried PR is expected; report what you dropped. +- No locale key in `gui/src/i18n/*`. If you need one, stop and report. +- No security write-up in `devlog/`; scratch space only, per `AGENTS.md`. + +**MUST.** + +- Prefix every mutating git command with `git -c core.hooksPath=/dev/null`. This repository's hooks + can start a GUI install, typecheck, and build, which the no-local-suite rule forbids. +- Push with `--no-verify`. +- Write the focused regression test `AGENTS.md` requires for a behaviour change, in the domain + directory beside the existing tests for that subsystem, and register it in both + `scripts/test-layout/layout.json` `explicit` and `tests/fixtures/test-layout-expected.json`. You + will not run it; hosted CI will. Those two maps are append-only and other lanes are adding to them + too; the orchestrator resolves the conflicts at merge, so do not skip the entry. +- Fill every section of `.github/PULL_REQUEST_TEMPLATE.md` and put `Closes #` in the body. In + **Verification**, state that the local suite, typecheck, and build were `NOT RUN` by operator + instruction and that hosted CI on the exact pushed head is the proof. +- When you carry another author's PR, add a `Co-authored-by` trailer in a branch commit. Resolve the + address with `gh api users/ --jq '.id'` and use `+@users.noreply.github.com`. +- Keep a devlog unit under `devlog/_plan/260911_l_/`. + +**Stacking.** First PR targets `dev`; the second targets the first PR's head branch, the third the +second. Retarget a child to `dev` after its parent lands. No native GitHub stacks. + +**Decisions already made for you.** Both audit rounds found items where the issue left a real choice +open. Those calls are recorded in your packet in bold. Implement the recorded decision; if you think +it is wrong, report the reason and stop. + +**Stop conditions.** Stop and report when the fix needs a path you do not own, when it needs a policy +no issue has fixed, when a locale key is unavoidable, or when hosted CI fails for a reason outside +your diff. + +**Report format.** Per PR: number, exact head SHA, CI run id and conclusion, the issue it closes, the +co-authors credited, the hunks you dropped from a carried PR, and any decision you made. Say +`NOT RUN` for local checks. + +**Decision boundary.** You do not merge, do not close another author's PR, and do not rank your lane +against another. When your last PR is green, report and stop. + +## L6 — streaming and vendor tool leakage + +Worktree `~/.codex/worktrees/260911-l6/opencodex`, branch `codex/260911-l6-streaming-tools`. + +Owned: `src/server/responses/codex-ws-exchange.ts`, `src/server/responses/codex-ws-wire.ts`, +directory `src/adapters/qoder/`. + +1. **#4191 — a long Codex thread fails only through the proxy** (WS 1006 / prelude timeout) while the + bypass works immediately. The timeout is `codex-ws-exchange.ts:214`. Reproduce first: establish + what length and timing trigger it and where the prelude budget goes. **Decision: the only in-scope + fix is to classify and report the timeout honestly, including the close code and the cause.** A + configurable budget, an SSE fallback, or a size preflight comes back as a report, not a patch. +2. **#4190 — vendor CLI agent scaffolding leaks into routed output** for `qoder`. **Decision: fix it + inside `src/adapters/qoder/` by sanitizing the vendor scaffolding out of routed output, failing + closed when the shape is unrecognized.** If the fix needs `src/adapters/coding-agent/protocol.ts`, + which no lane owns, stop and report. + diff --git a/devlog/_plan/260911_l6_streaming_tools/010_4191_ws_failure_classification.md b/devlog/_plan/260911_l6_streaming_tools/010_4191_ws_failure_classification.md new file mode 100644 index 0000000000..f141f3ca56 --- /dev/null +++ b/devlog/_plan/260911_l6_streaming_tools/010_4191_ws_failure_classification.md @@ -0,0 +1,102 @@ +# 010 — #4191: classify the WebSocket failure instead of restating it + +Unit: `devlog/_plan/260911_l6_streaming_tools`. Lane L6, work-phase 1. +Issue: [#4191](https://github.com/lidge-jun/opencodex/issues/4191). + +## What the report actually contains + +A long Codex thread failed for hours through the proxy and recovered instantly when the +proxy was bypassed — same account, model, machine, repository and thread. Two messages +appeared, alternating: + + codex websocket closed before a Responses terminal event (close 1006 Connection ended) + codex websocket response prelude timed out + +Neither sentence distinguishes the cases it is true of. The reporter compensated by running +an A/B toggle by hand, which is why the only usable evidence in a very careful report is +"it works when OpenCodex is off". That is the defect this work-phase addresses. + +## Where the prelude budget goes + +Read `codex-ws-exchange.ts` at the frozen base: `preludeTimer` is armed once in `onOpen` +immediately after `ws.send()` succeeds, and the only thing that clears it is +`commitResponse()`. `commitResponse()` runs from `onMessage` only on the **non-control** +branch — a frame the metadata channel did not claim. `codex.rate_limits` and +`codex.response.metadata` are claimed by `CodexWsMetadata.consume()` and set +`controlFrame = true`, so they never reach it. + +So the 90 s budget is measured from send to the first *Responses* event, and upstream +liveness does not extend it. A socket that answers in 40 ms with a quota frame, then keeps +sending quota updates while the backend works through a very large replayed thread, still +dies at exactly 90 s — and reports the same sentence as a socket that was never answered at +all. That is the mechanism the report's second error most likely describes, and before this +change nothing in the message could tell the two apart. + +The size story is next to it. `codexWsCreateFrameExceedsLimit` preflights the create frame +at 16 MiB − 64 KiB and routes an oversized turn to HTTP SSE, so the very largest threads are +already safe. The band immediately below the limit is not: it still dials the socket, and a +long full-replay thread sits in that band. Whether this reporter's thread was there is not +knowable from the report, which is precisely why the create-frame byte count belongs in the +message. + +## What was implemented + +A content-free stage record, `CodexWsFailureStage` in `codex-ws-wire.ts`, carrying the +create-frame byte count, whether the send completed, the number of upstream frames, how many +of those the metadata channel claimed, how many Responses events were relayed downstream, +and two durations — send to first frame, and send to failure. `classifyCodexWsFailure` +reduces that to one of four causes: `before-send`, `no-upstream-frame`, +`no-response-event`, `after-response-started`. `codexWsFailureDetail` renders it as a +suffix appended after the existing message, including after the close-code tail, so +`(close 1006 Connection ended)` remains one contiguous substring for every existing reader. + +`codex-ws-exchange.ts` keeps the counters and stamps `sentAt` after a successful send. The +frame is measured lazily, only when a failure message is being built, so the happy path never +pays for sizing a multi-megabyte string. The three paths that end an already-open exchange +without a terminal event now carry the detail: the prelude timeout, the close-before-terminal +message, and the transport error. The size- and queue-limit failures already name their own +precise cause and were left alone. + +The reporter asked for five diagnostics. Four are now in the message the client receives: +serialized frame size, whether the send completed, elapsed time to the first upstream event +and to the failure, and whether any downstream Responses bytes were emitted. The fifth — +preserving the upstream close code — was already there and is unchanged. + +## Deliberately not implemented + +The issue and its two maintainer comments leave three things open. All three come back as a +report, per the lane packet. + +**Automatic HTTP/SSE fallback after an open socket dies.** `failStream` treats a completed +send as possibly executing upstream, and settles as a body failure rather than a resendable +rejection. The maintainer comment on the issue is explicit that `responseCommitted === false` +and zero downstream bytes are *not* proof the upstream did not accept or execute the frame, +so a resend gated on either can duplicate a turn. The classification added here must not be +read as a fallback-eligibility signal; the type comment says so at the definition, because +`no-upstream-frame` is exactly the value a future reader would be tempted to misuse. + +**A configurable or longer prelude budget.** #3976 asked for this and 90 s is already three +times the original 30 s. Raising a fixed constant without a reproduction is guesswork, and +the finding above suggests the real question is different: whether upstream liveness on the +control channel should extend the budget at all, or whether a thread that produces only quota +frames for 90 s should be refused earlier and more clearly. Both are policy, and no issue has +fixed that policy. + +**A size preflight below the current ceiling.** The band under 16 MiB − 64 KiB is a real gap, +but narrowing WS eligibility by predicted frame size or expected time-to-first-token changes +which turns take which transport for every user, not only failing ones. It needs its own +issue with measurements, and the byte count now in the failure message is what would supply +them. + +## Verification + +Focused regression test at `tests/responses/ws-failure-stage.test.ts`, registered in +`scripts/test-layout/layout.json` and `tests/fixtures/test-layout-expected.json`. It covers +the classifier's four stages, the renderer's exact output, the contiguity of the close-code +tail that existing assertions depend on, and four end-to-end cases through the fake socket: +an unanswered close, a quota-only close, a close after relayed events, and the prelude +timeout under fake timers. + +Local suite, typecheck and build: NOT RUN, by operator instruction for this dispatch round. +Hosted CI on the pushed head is the evidence. + diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index b5ea45c4a3..cee1514e55 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1304,6 +1304,7 @@ "winsw-stop-hardening.test.ts": "windows", "winsw.test.ts": "service", "ws-endpoint.test.ts": "responses", + "ws-failure-stage.test.ts": "responses", "ws-upstream-reuse.test.ts": "responses", "ws-upstream.test.ts": "responses", "xai-client.test.ts": "images", diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 31813756ea..64a8e5aba4 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -5,7 +5,8 @@ import { CODEX_RESPONSES_HTTP_URL, type PreparedCodexWsRequest } from "./codex-w import { CodexWsCorrelation } from "./codex-ws-correlation"; import type { CodexWsSession } from "./codex-ws-session"; import { UPGRADE_DEADLINE_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, MAX_CODEX_WS_FRAME_BYTES, - MAX_CODEX_WS_QUEUE_BYTES, markCodexWsResponse, normalizeResponsesWsRelayEvent, closedBeforeTerminalMessage } from "./codex-ws-wire"; + MAX_CODEX_WS_QUEUE_BYTES, markCodexWsResponse, normalizeResponsesWsRelayEvent, closedBeforeTerminalMessage, + codexWsFailureDetail, type CodexWsFailureStage } from "./codex-ws-wire"; interface ExchangeOptions { session: CodexWsSession; @@ -94,6 +95,14 @@ export function codexWsExchange(options: ExchangeOptions): Promise { let received = false; let responseCommitted = false; let terminal = false; + // #4191: the counters behind the failure classification. A user whose long + // thread died here could not tell an unanswered socket from one that carried + // only quota frames, because both arrived as the same one-line message. + let upstreamFrames = 0; + let controlFrames = 0; + let relayedEvents = 0; + let sentAt: number | null = null; + let firstFrameAt: number | null = null; let controller: ReadableStreamDefaultController | null = null; const encoder = new TextEncoder(); const metadata = url === CODEX_RESPONSES_HTTP_URL ? new CodexWsMetadata(onQuota) : null; @@ -123,6 +132,21 @@ export function codexWsExchange(options: ExchangeOptions): Promise { ws.removeEventListener("error", onError); }; + /** + * Snapshot the stage for a failure message. Measuring the frame is deferred + * to here so the happy path never pays for it: a full-replay thread's frame + * runs to megabytes, and this is the only place its size is worth knowing. + */ + const failureStage = (): CodexWsFailureStage => ({ + requestBytes: Buffer.byteLength(frameText, "utf8"), + sent, + upstreamFrames, + controlFrames, + relayedEvents, + firstFrameMs: sentAt !== null && firstFrameAt !== null ? Math.max(0, firstFrameAt - sentAt) : null, + elapsedMs: sentAt !== null ? Math.max(0, Date.now() - sentAt) : null, + }); + const commitResponse = () => { if (responseCommitted) return; responseCommitted = true; @@ -192,6 +216,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { sent = true; try { ws.send(frameText); + sentAt = Date.now(); } catch { if (received || responseCommitted) { if (terminal) session.dispose(); @@ -211,13 +236,18 @@ export function codexWsExchange(options: ExchangeOptions): Promise { } if (!metadata) commitResponse(); else if (!responseCommitted && !terminal) { - preludeTimer = setTimeout(() => failStream("codex websocket response prelude timed out"), CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); + preludeTimer = setTimeout( + () => failStream(`codex websocket response prelude timed out${codexWsFailureDetail(failureStage())}`), + CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, + ); } }; const onMessage = (event: MessageEvent) => { if (!controller || terminal) return; received = true; + upstreamFrames += 1; + if (firstFrameAt === null) firstFrameAt = Date.now(); const text = typeof event.data === "string" ? event.data : ""; if (!text) return; // UTF-8 byte length is always at least the JS string length. Reject this @@ -243,6 +273,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { if (sanitized !== null) { relayText = sanitized; controlFrame = true; + controlFrames += 1; } } catch (error) { failStream(error); @@ -296,6 +327,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { failStream("codex websocket response stream closed while enqueueing"); return; } + if (!controlFrame) relayedEvents += 1; if (type === "response.completed" || type === "response.failed" || type === "response.incomplete" || type === "error") { const completedId = correlation?.completed(normalized.payload) ?? null; terminal = true; @@ -316,7 +348,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { resolve(sseFallback(url, init)); return; } - if (sent && !terminal) failStream(closedBeforeTerminalMessage(event)); + if (sent && !terminal) failStream(closedBeforeTerminalMessage(event, failureStage())); }; const onError = () => { @@ -327,7 +359,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { cleanup(); session.dispose(); resolve(sseFallback(url, init)); - } else failStream("codex websocket transport error"); + } else failStream(`codex websocket transport error${codexWsFailureDetail(failureStage())}`); }; detachOwner = session.bindOwner(reason => cancelExchange(reason)); ws.addEventListener("open", onOpen); diff --git a/src/server/responses/codex-ws-wire.ts b/src/server/responses/codex-ws-wire.ts index 271e7280bb..770ce21b54 100644 --- a/src/server/responses/codex-ws-wire.ts +++ b/src/server/responses/codex-ws-wire.ts @@ -50,6 +50,72 @@ export function markCodexWsResponse(response: Response, observed: boolean): void const CLOSED_BEFORE_TERMINAL = "codex websocket closed before a Responses terminal event"; +/** + * Content-free stage record for an exchange that ended without a Responses + * terminal event (#4191). + * + * The field report that drove this could not be told apart from a network + * outage, because every such failure reached the user as one of two bare + * sentences. Both are true of a socket that was never answered, a socket that + * carried only quota control frames, and a socket that died mid-response — + * three different upstream stories with three different owners. These counters + * are the smallest set that separates them, and every one of them is a size, a + * count, or a duration: no request body, no header, no account identifier, and + * no conversation text can reach a message built from this record. + */ +export type CodexWsFailureStage = { + /** UTF-8 size of the `response.create` frame this exchange dialled with. */ + requestBytes: number; + /** True once `ws.send()` returned, so the turn may be executing upstream. */ + sent: boolean; + /** Frames the socket delivered, of any kind, including ones that did not parse. */ + upstreamFrames: number; + /** Frames the metadata channel claimed (quota, response metadata). */ + controlFrames: number; + /** Responses events actually written to the downstream SSE body. */ + relayedEvents: number; + /** Milliseconds from send to the first upstream frame; null when none arrived. */ + firstFrameMs: number | null; + /** Milliseconds from send to this failure; null when the failure predates the send. */ + elapsedMs: number | null; +}; + +/** + * Which upstream story the counters tell. Ordered by how much the upstream had + * committed to, because that is what decides who owns the failure — and, for a + * future maintainer reading #4191, it is deliberately NOT a fallback-eligibility + * signal. `no-upstream-frame` does not mean the frame was not accepted; the + * no-replay-after-send contract in `codex-ws-exchange.ts` stands regardless of + * what this classifier says. + */ +export type CodexWsFailureCause = + | "before-send" + | "no-upstream-frame" + | "no-response-event" + | "after-response-started"; + +export function classifyCodexWsFailure(stage: CodexWsFailureStage): CodexWsFailureCause { + if (!stage.sent) return "before-send"; + if (stage.relayedEvents > 0) return "after-response-started"; + if (stage.upstreamFrames === 0) return "no-upstream-frame"; + return "no-response-event"; +} + +/** + * Render the stage as a suffix appended to an existing failure message. + * + * It is a suffix, not an interpolation, on purpose: the close-code tail these + * messages already carry is matched as a contiguous substring by the callers + * and tests that read it, so nothing may be inserted ahead of it. + */ +export function codexWsFailureDetail(stage: CodexWsFailureStage): string { + const duration = (value: number | null): string => (value === null ? "n/a" : `${value}ms`); + return ` [cause=${classifyCodexWsFailure(stage)} request=${stage.requestBytes}B` + + ` sent=${stage.sent ? "yes" : "no"} frames=${stage.upstreamFrames}` + + ` control=${stage.controlFrames} relayed=${stage.relayedEvents}` + + ` first-frame=${duration(stage.firstFrameMs)} elapsed=${duration(stage.elapsedMs)}]`; +} + export type ResponsesWsRelayEvent = { type: string; text: string; @@ -111,18 +177,23 @@ export function normalizeResponsesWsRelayEvent(text: string): ResponsesWsRelayEv * inspector, so `/api/logs` keeps neither this message nor a specific code — * only `streamAborted`. Machine-readable typing would mean changing the error * taxonomy, which is deliberately out of scope for this transport fix. + * + * When a stage is supplied its detail is appended last, after the close-code + * tail, so the code and reason stay one contiguous substring. */ -export function closedBeforeTerminalMessage(event: unknown): string { +export function closedBeforeTerminalMessage(event: unknown, stage?: CodexWsFailureStage): string { const detail = event as { code?: unknown; reason?: unknown } | null | undefined; const code = typeof detail?.code === "number" ? detail.code : null; const reason = typeof detail?.reason === "string" ? detail.reason.trim() : ""; - if (code === null) return CLOSED_BEFORE_TERMINAL; + const stageDetail = stage ? codexWsFailureDetail(stage) : ""; + if (code === null) return `${CLOSED_BEFORE_TERMINAL}${stageDetail}`; const suffix = reason ? ` ${code} ${reason}` : ` ${code}`; if (code === WS_CLOSE_MESSAGE_TOO_BIG) { return `codex websocket rejected the request frame as too large (close${suffix});` - + ` requests at or above ${MAX_CODEX_WS_CREATE_FRAME_BYTES} bytes must use the HTTP SSE transport`; + + ` requests at or above ${MAX_CODEX_WS_CREATE_FRAME_BYTES} bytes must use the HTTP SSE transport` + + stageDetail; } - return `${CLOSED_BEFORE_TERMINAL} (close${suffix})`; + return `${CLOSED_BEFORE_TERMINAL} (close${suffix})${stageDetail}`; } /** diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d772205061..486a6de544 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1139,6 +1139,7 @@ "winsw-stop-hardening.test.ts": "windows", "winsw.test.ts": "service", "ws-endpoint.test.ts": "responses", + "ws-failure-stage.test.ts": "responses", "ws-upstream-reuse.test.ts": "responses", "ws-upstream.test.ts": "responses", "xai-client.test.ts": "images", diff --git a/tests/responses/ws-failure-stage.test.ts b/tests/responses/ws-failure-stage.test.ts new file mode 100644 index 0000000000..af4db2136a --- /dev/null +++ b/tests/responses/ws-failure-stage.test.ts @@ -0,0 +1,230 @@ +import { afterEach, beforeEach, describe, expect, jest, test } from "bun:test"; +import { + classifyCodexWsFailure, + closedBeforeTerminalMessage, + codexWsFailureDetail, + type CodexWsFailureStage, +} from "../../src/server/responses/codex-ws-wire"; +import { + codexWsUpstreamFetch, + CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, +} from "../../src/server/responses/ws-upstream"; + +/** + * #4191: a long Codex thread died only through the proxy, and every variant of + * that death arrived as the same one-line message. The reporter could not tell + * an unanswered socket from one that had already started replying, so the only + * usable evidence in the whole report was an A/B toggle. These cases hold the + * transport to naming the stage it failed at. + */ + +const CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"; +const BOUNDED_WS_RUNTIME = "1.4.0"; + +type Listener = (event: unknown) => void; + +/** Minimal scriptable stand-in for Bun's WebSocket, mirroring `ws-upstream.test.ts`. */ +class FakeWebSocket { + static instances: FakeWebSocket[] = []; + static script: (ws: FakeWebSocket) => void = () => {}; + url: string; + sent: string[] = []; + 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); + } + + send(data: string) { + this.sent.push(data); + } + + close() {} +} + +const RealWebSocket = globalThis.WebSocket; + +function installFake(script: (ws: FakeWebSocket) => void) { + FakeWebSocket.script = script; + globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket; +} + +function streamingInit(): RequestInit { + return { + method: "POST", + headers: { "content-type": "application/json", authorization: "Bearer test" }, + body: JSON.stringify({ model: "gpt-5.5", stream: true }), + }; +} + +function noFallback(): Promise { + throw new Error("fallback must not run after open"); +} + +function stage(overrides: Partial = {}): CodexWsFailureStage { + return { + requestBytes: 812, + sent: true, + upstreamFrames: 0, + controlFrames: 0, + relayedEvents: 0, + firstFrameMs: null, + elapsedMs: 90_003, + ...overrides, + }; +} + +async function failureMessage(script: (ws: FakeWebSocket) => void): Promise { + installFake(script); + const response = await codexWsUpstreamFetch( + CODEX_URL, + streamingInit(), + noFallback as unknown as typeof fetch, + BOUNDED_WS_RUNTIME, + ); + try { + await response.text(); + } catch (error) { + return error instanceof Error ? error.message : String(error); + } + throw new Error("expected the relayed body to fail"); +} + +beforeEach(() => { + FakeWebSocket.instances = []; + FakeWebSocket.script = () => {}; +}); + +afterEach(() => { + globalThis.WebSocket = RealWebSocket; + FakeWebSocket.instances = []; + FakeWebSocket.script = () => {}; +}); + +describe("codex WS failure classification", () => { + test("separates the four stages a dead exchange can be in", () => { + expect(classifyCodexWsFailure(stage({ sent: false, elapsedMs: null }))).toBe("before-send"); + expect(classifyCodexWsFailure(stage())).toBe("no-upstream-frame"); + // Quota control frames prove the upstream answered; they are not a response. + expect(classifyCodexWsFailure(stage({ upstreamFrames: 3, controlFrames: 3 }))).toBe("no-response-event"); + expect(classifyCodexWsFailure(stage({ upstreamFrames: 9, controlFrames: 2, relayedEvents: 7 }))) + .toBe("after-response-started"); + }); + + test("a frame that never left outranks the counters behind it", () => { + // The send is what makes a turn possibly live upstream, so it is read first. + expect(classifyCodexWsFailure(stage({ sent: false, upstreamFrames: 4, relayedEvents: 2 }))) + .toBe("before-send"); + }); + + test("renders every field, with n/a for the durations that do not exist yet", () => { + expect(codexWsFailureDetail(stage({ upstreamFrames: 2, controlFrames: 2, firstFrameMs: 41 }))).toBe( + " [cause=no-response-event request=812B sent=yes frames=2 control=2 relayed=0" + + " first-frame=41ms elapsed=90003ms]", + ); + expect(codexWsFailureDetail(stage({ sent: false, elapsedMs: null }))).toBe( + " [cause=before-send request=812B sent=no frames=0 control=0 relayed=0" + + " first-frame=n/a elapsed=n/a]", + ); + }); +}); + +describe("closedBeforeTerminalMessage", () => { + test("keeps the close code contiguous and appends the stage last", () => { + const message = closedBeforeTerminalMessage({ code: 1006, reason: "Connection ended" }, stage()); + // The close tail is read as one substring by existing callers and tests. + expect(message).toContain("closed before a Responses terminal event (close 1006 Connection ended)"); + expect(message.endsWith(codexWsFailureDetail(stage()))).toBe(true); + }); + + test("leaves the oversized-frame guidance intact ahead of the stage", () => { + const message = closedBeforeTerminalMessage({ code: 1009, reason: "Message Too Big" }, stage()); + expect(message).toMatch(/rejected the request frame as too large \(close 1009 Message Too Big\)/); + expect(message).toContain("must use the HTTP SSE transport [cause="); + }); + + test("omits the stage entirely when none is supplied", () => { + expect(closedBeforeTerminalMessage({ code: 1006 })) + .toBe("codex websocket closed before a Responses terminal event (close 1006)"); + expect(closedBeforeTerminalMessage(null)) + .toBe("codex websocket closed before a Responses terminal event"); + }); +}); + +describe("codexWsUpstreamFetch failure reporting", () => { + test("names an unanswered socket, and measures the frame it actually sent", async () => { + const message = await failureMessage(ws => { + ws.emit("open", {}); + ws.emit("close", { code: 1006, reason: "Connection ended" }); + }); + const sentBytes = Buffer.byteLength(FakeWebSocket.instances[0]!.sent[0]!, "utf8"); + expect(message).toContain("closed before a Responses terminal event (close 1006 Connection ended)"); + expect(message).toContain(`[cause=no-upstream-frame request=${sentBytes}B sent=yes frames=0`); + expect(message).toContain("control=0 relayed=0 first-frame=n/a"); + }); + + test("distinguishes a socket that answered with quota but never started a response", async () => { + const message = await failureMessage(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify({ + type: "codex.rate_limits", rate_limits: { primary: { used_percent: 10, window_minutes: 10080 } }, + }) }); + ws.emit("close", { code: 1006 }); + }); + expect(message).toContain("cause=no-response-event"); + expect(message).toContain("frames=1 control=1 relayed=0"); + expect(message).toMatch(/first-frame=\d+ms/); + }); + + test("distinguishes a drop that landed after the response was already flowing", async () => { + const message = await failureMessage(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify({ type: "response.created", response: { id: "r1" } }) }); + ws.emit("message", { data: JSON.stringify({ + type: "response.output_text.delta", delta: "hi", item_id: "m1", output_index: 0, content_index: 0, + }) }); + ws.emit("close", { code: 1006 }); + }); + expect(message).toContain("cause=after-response-started"); + expect(message).toContain("frames=2 control=0 relayed=2"); + }); + + test("the prelude timeout says which stage ran out of budget", async () => { + jest.useFakeTimers(); + const opened = Promise.withResolvers(); + try { + installFake(ws => { ws.emit("open", {}); opened.resolve(); }); + const pending = codexWsUpstreamFetch( + CODEX_URL, + streamingInit(), + noFallback as unknown as typeof fetch, + BOUNDED_WS_RUNTIME, + ); + await opened.promise; + jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); + const response = await pending; + await expect(response.text()).rejects.toThrow( + /prelude timed out \[cause=no-upstream-frame request=\d+B sent=yes frames=0 control=0 relayed=0/, + ); + } finally { + jest.useRealTimers(); + } + }); +}); +