From 1bce6fac9dcc599cdd52fda0cb979cceb224fd54 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 00:22:04 +0900 Subject: [PATCH 1/2] docs(devlog): seed the L6 dispatch packet --- .../260911_l6_streaming_tools/000_packet.md | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 devlog/_plan/260911_l6_streaming_tools/000_packet.md 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. + From 030316c5dbffe81f31dc12340829b42de22e2bbc Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 07:49:40 +0900 Subject: [PATCH 2/2] fix(responses): classify Codex WS failures instead of restating them A long Codex thread that dies on the WebSocket transport reports one of two sentences, and neither distinguishes the cases it is true of: a socket that was never answered, a socket that carried only quota control frames, and a socket that died after the response was already flowing all produce the same text. The reporter of #4191 had to establish that much by toggling the proxy by hand. Add a content-free stage record to the exchange: create-frame byte count, whether the send completed, upstream frame count, how many of those the metadata channel claimed, Responses events relayed downstream, and the durations from send to first frame and to the failure. classifyCodexWsFailure reduces it to before-send, no-upstream-frame, no-response-event, or after-response-started, and the detail is appended after the existing message so the close-code tail stays one contiguous substring for every reader that matches on it. The frame is measured only when a failure message is being built, so the happy path never pays for sizing a multi-megabyte string. The classification is not a fallback-eligibility signal, and the no-replay-after- send contract is unchanged: a completed send may be executing upstream whatever the counters say. Refs #4191 --- .../010_4191_ws_failure_classification.md | 102 ++++++++ scripts/test-layout/layout.json | 1 + src/server/responses/codex-ws-exchange.ts | 40 ++- src/server/responses/codex-ws-wire.ts | 79 +++++- tests/fixtures/test-layout-expected.json | 1 + tests/responses/ws-failure-stage.test.ts | 230 ++++++++++++++++++ 6 files changed, 445 insertions(+), 8 deletions(-) create mode 100644 devlog/_plan/260911_l6_streaming_tools/010_4191_ws_failure_classification.md create mode 100644 tests/responses/ws-failure-stage.test.ts 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(); + } + }); +}); +