From 7fd85d3b67235f33a1772b59e7778d9f9d99e82f Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 11:55:44 +0900 Subject: [PATCH 01/10] docs(devlog): roadmap for the Codex WS commit-boundary realignment --- .../260911_ws_commit_boundary/000_plan.md | 52 +++++++++++ .../010_journey_evaluation.md | 50 +++++++++++ .../020_design_record.md | 90 +++++++++++++++++++ 3 files changed, 192 insertions(+) create mode 100644 devlog/_plan/260911_ws_commit_boundary/000_plan.md create mode 100644 devlog/_plan/260911_ws_commit_boundary/010_journey_evaluation.md create mode 100644 devlog/_plan/260911_ws_commit_boundary/020_design_record.md diff --git a/devlog/_plan/260911_ws_commit_boundary/000_plan.md b/devlog/_plan/260911_ws_commit_boundary/000_plan.md new file mode 100644 index 0000000000..95f7274034 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/000_plan.md @@ -0,0 +1,52 @@ +# WS commit boundary — 260911 + +Base: `origin/dev` `babb76449f` (fetched 2026-09-11 KST). Branch `codex/260911-ws-commit-boundary`, +worktree `/Users/jun/.codex/worktrees/260911-wsc/opencodex`. + +## Why this unit exists + +#4191 reports a long Codex thread that fails only while routed through OpenCodex, as either +`codex websocket closed before a Responses terminal event (close 1006 Connection ended)` or +`codex websocket response prelude timed out`, and works immediately when the proxy is bypassed. +#4083 raised the fixed prelude deadline from 30 s to 90 s for slow multi-image starts; #3976 asked +for the number to be configurable; #2471 fixed the 16 MiB create-frame ceiling. + +The lane dispatch round (`260911_lane_dispatch_round`) added the #4191 failure-stage counters so +a user can tell an unanswered socket from one that carried only quota frames. That was +diagnosis. This unit is the fix to the boundary the diagnosis exposed, after an external +semantic review (`010_journey_evaluation.md`) overturned the first framing. + +## Scope + +- `src/server/responses/codex-ws-exchange.ts` — settle post-send, pre-response failures as an + honest HTTP status; replace the fixed prelude timer with silence-based liveness; cancel the + upstream turn on a pre-commit client abort. +- `src/server/responses/codex-ws-wire.ts` — liveness constants and the non-replayable body shape. +- `src/lib/upstream-retry.ts` — a non-replayable marker that `fetchWithTransientRetry` honours. +- `tests/responses/ws-upstream.test.ts`, `tests/lib/upstream-retry.test.ts` — oracle updates and + new cases. + +Out of scope, recorded in `020_design_record.md`: resume-by-id after 1006 (Codex does not request +background responses, so the vendor resume surface does not apply), the opt-in provider path +without a metadata channel (it commits at send today and keeps doing so), the create-frame size +predicate, and `src/server/responses/core.ts`. + +## Rules for this unit + +- No local product suite: no `bun test`, `bun run test`, `test:changed`, `typecheck`, + `build:gui`, or `bun install` in this worktree. Every verification line reads NOT RUN until + remote CI on the final head says otherwise. +- Push with `--no-verify` and `core.hooksPath=/dev/null`. +- xai/grok-4.6 subagents are read-only verifiers of the diff; aside/web research is free. +- One work-phase is one PABCD cycle: wp1 this roadmap, wp2 honest status + marker, wp3 liveness + and abort propagation, wp4 PR, review, CI. + +## Work phases + +| wp | unit | doc | exit | +|---|---|---|---| +| wp1 | roadmap | 000, 010, 020 | docs committed on the branch | +| wp2 | honest post-send status | 030 | code + tests committed, NOT RUN | +| wp3 | liveness + abort | 040 | code + tests committed, NOT RUN | +| wp4 | PR + review + CI | 050 | final-head CI green, review dispositioned | + diff --git a/devlog/_plan/260911_ws_commit_boundary/010_journey_evaluation.md b/devlog/_plan/260911_ws_commit_boundary/010_journey_evaluation.md new file mode 100644 index 0000000000..a4dd9c48f8 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/010_journey_evaluation.md @@ -0,0 +1,50 @@ +# Journey evaluation — how the framing changed + +## What was done before this unit + +1. Lane dispatch round: seven file-disjoint lanes from `6d3ad12e3`, each a worktree and a + Codex thread, merged serially on final-head green CI (#4217 … #4248). One of those lanes landed + the #4191 failure-stage counters in `codex-ws-wire.ts`: request bytes, sent, frames, control + frames, relayed events, first-frame and elapsed durations. The counters are content-free by + construction and only classify; they were never a fallback signal. +2. Structure question from the owner: `codex -> http -> opencodex -> ws -> openai` — is the + asymmetry itself the bug? Source reading said no: WS is chosen only for streaming POSTs on a + bounded-relay Bun, the create frame is measured before dialling, and the one reversible point is + the send. First framing: the reversible window is too narrow and judged by size alone; widen the + HTTP path below the ceiling and scale the prelude budget by frame size. +3. Semantic review by anthropic/claude-fable-5-1. Three corrections were accepted after source + confirmation: + - The no-resend-after-send rule is not a defect. RFC 9110 §9.2.2 forbids an intermediary from + automatically repeating a non-idempotent request; the user agent owns that decision. Offering + "allow fallback after send" as an option was the wrong question. + - The broken contract is the status code. `commitResponse` builds `new Response(stream, + { status: 200 })` before any upstream frame, and `failStream` commits that 200 on the failure + path (`if (sent) commitResponse()`) precisely so the pre-stream wrapper cannot resend. The proxy + therefore converts "no response" into "a response that failed", removes the status the client + would use for its own retry policy, and neuters the client's first-byte timeout with chunked + headers. Direct-to-vendor Codex survives the same at-most-once lane through its own retry; the + proxy is stricter than the party whose money is at stake and pays for it with a hard failure. + - The 90 s prelude is the wrong kind of quantity: it folds "dead" and "slow" into one number. + Dead is a liveness question with a native answer (ping/pong); slow already has an owner (the + client deadline). A fixed proxy deadline in series always inherits the tighter bound. + +## What the evaluation keeps and drops + +Kept: every existing oracle (no HTTP fallback after send, one `response.create` per exchange, +refused-create 4xx projection, correlation before conversion, bounded queue). Kept: the 90 s +number, but demoted from "time to first response event" to "unanswered silence with no pong", +which is unreachable on a socket whose peer answers pings. + +Dropped: post-send HTTP fallback (never acceptable), size-scaled prelude budgets (treats the +symptom), resume-by-id after 1006 (Codex sends `stream: true` without `background: true`; the +vendor resume endpoint requires a background response, so there is nothing to resume for this +client; recorded as a follow-up for callers that do opt in). + +## What this unit does not claim + +It does not claim the Codex backend answers WebSocket pings; the exchange feature-detects +`ws.ping` and degrades to the previous 90 s behaviour when no pong ever arrives. It does not run +any local suite. Whether the honest 504 improves the #4191 user's experience is a live question +that only a field report can answer; what this unit guarantees is that the proxy stops hiding the +signal that user's client needs. + diff --git a/devlog/_plan/260911_ws_commit_boundary/020_design_record.md b/devlog/_plan/260911_ws_commit_boundary/020_design_record.md new file mode 100644 index 0000000000..2f7b963c69 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/020_design_record.md @@ -0,0 +1,90 @@ +# Design record — commit boundary, liveness, abort + +## Invariants that stay + +- I1 No HTTP SSE fallback once `ws.send()` has returned (`sent === true`). +- I2 One `response.create` frame per exchange; no proxy-internal resend after send. +- I3 A refused create (`type: error`, no `stream_id`, 4xx status) before any response event is + projected as that 4xx with the metadata snapshot (#3740); correlation runs first. +- I4 After the first `response.*` or `error` event has been relayed, every later failure is a + body error on the already-committed 200 (the relay synthesizes `response.failed`). + +## New invariant + +- I5 The client commit never precedes the upstream acknowledgment. Before the first + `response.*`/`error` event the exchange holds no client Response. A failure in that window + settles as a JSON error with an honest gateway status, marked non-replayable. + +## Diff-level plan + +### `src/lib/upstream-retry.ts` + +Add a `WeakSet` with `markResponseNonReplayable(res)` and +`isNonReplayableResponse(res)`. In `fetchWithTransientRetry` the loop guard becomes +`if (res.ok || !isTransientUpstreamStatus(res.status) || isNonReplayableResponse(res)) return res;`. +Rationale in the doc comment: the origin may already be executing the request (RFC 9110 §9.2.2), +so a gateway status from a post-send transport is returned to the caller for its own policy. + +### `src/server/responses/codex-ws-wire.ts` + +- `CODEX_WS_LIVENESS_PING_INTERVAL_MS = 15_000`. +- `CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS` keeps its value (90 000) and gains a new meaning in its + comment: the longest inbound silence (no message frame, no pong) tolerated before the first + response event. +- `codexWsPreResponseFailure(status, message, prelude: Headers): Response` — builds + `{ error: { type: "upstream_error", code, message } }` with `content-type: application/json`, + `cache-control: no-store`, the metadata snapshot headers, and calls + `markResponseNonReplayable`. `code` is `upstream_timeout` for 504 and + `upstream_closed_before_response` for 502. +- `CodexWsFailureStage` gains `pings` and `pongs`; `codexWsFailureDetail` appends + ` pings=N pongs=N` inside the bracket, after `elapsed`. `tests/responses/ws-failure-stage.test.ts` + is updated in the same commit. + +### `src/server/responses/codex-ws-exchange.ts` + +- `failStream(error, status: 502 | 504 = 502)`: when `sent && !responseCommitted`, resolve + `codexWsPreResponseFailure(status, message, metadata.snapshot())` instead of committing a 200, + close the controller, dispose the session. When committed, unchanged. +- `cancelExchange(reason)` when `sent && !responseCommitted`: mark terminal, cleanup, dispose the + session (this closes the socket, which is the upstream cancel), `reject(reason)`. The caller's + own abort is never retried by the wrappers (`isConnectionResetError` excludes AbortError and the + retry loops check `abortSignal.aborted`). +- Liveness replaces the single `preludeTimer`: + - `armSilence()` (re)starts a `CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS` timer whose expiry calls + `failStream("codex websocket response prelude timed out" + detail, 504)`. + - `onMessage` and `onPong` call `armSilence()` while `!responseCommitted`. + - After send, when `typeof ws.ping === "function"`, a repeating + `CODEX_WS_LIVENESS_PING_INTERVAL_MS` timer calls `ws.ping()` until commit or terminal; a + throwing `ping()` stops the pinger only. + - `cleanup()` clears both timers and removes the `pong` listener; `commitResponse()` clears + them too. +- The non-metadata path (`if (!metadata) commitResponse()`) is unchanged. + +### Tests (`tests/responses/ws-upstream.test.ts`) + +Updated oracles: prelude overflow → 502 JSON, not a WS-marked stream; first-response deadline +through `fetchWithTransientRetry` → 504, one send, zero HTTP; foreign-stream identity mismatch → +502; close 1006 / 1009 before any response event → 502 carrying the same messages; abort after send +before commit → the pending fetch rejects with the caller reason and the socket is closed. + +New cases: a pong resets the silence clock past 90 s and the response still completes with one +send; a socket exposing `ping` is pinged every 15 s of prelude and stops after commit; a socket +without `ping` is never pinged and keeps the 90 s bound; `fetchWithTransientRetry` returns a +non-replayable 504 without a second call (`tests/lib/upstream-retry.test.ts`). + +## Risks and their answers + +- Client behaviour on 504: Codex retries stream requests on 5xx with backoff, which is the same + policy it applies on the direct path; the proxy no longer substitutes its own. +- Pool recovery on 5xx: `shouldRetryCodexPoolAccountQuota` rotates only on body-confirmed quota + evidence; the new body carries none. Opaque-blob recovery excludes 5xx other than 502 with an + encrypted-output body, which this is not. +- Backend pong support unknown: feature-detected and degrades to the current bound. +- Request log: the failure is now a 504/502 row instead of a 200 with `streamAborted`; this is + the intended diagnostic change. + +## Verification + +NOT RUN locally by rule. Remote CI on the final head is the only executable proof; the read-only +grok-4.6 review of the diff is the second pair of eyes. + From c3c1ea673161a5e8676f7417cb1ecdd4989a15e6 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:16:56 +0900 Subject: [PATCH 02/10] docs(devlog): audit round 1 dispositions for the WS commit boundary --- .../260911_ws_commit_boundary/000_plan.md | 8 +++-- .../020_design_record.md | 7 ++++- .../025_audit_round1.md | 29 +++++++++++++++++++ 3 files changed, 41 insertions(+), 3 deletions(-) create mode 100644 devlog/_plan/260911_ws_commit_boundary/025_audit_round1.md diff --git a/devlog/_plan/260911_ws_commit_boundary/000_plan.md b/devlog/_plan/260911_ws_commit_boundary/000_plan.md index 95f7274034..2194f3c901 100644 --- a/devlog/_plan/260911_ws_commit_boundary/000_plan.md +++ b/devlog/_plan/260911_ws_commit_boundary/000_plan.md @@ -22,14 +22,18 @@ semantic review (`010_journey_evaluation.md`) overturned the first framing. honest HTTP status; replace the fixed prelude timer with silence-based liveness; cancel the upstream turn on a pre-commit client abort. - `src/server/responses/codex-ws-wire.ts` — liveness constants and the non-replayable body shape. -- `src/lib/upstream-retry.ts` — a non-replayable marker that `fetchWithTransientRetry` honours. +- `src/lib/upstream-retry.ts` — a non-replayable marker that `fetchWithTransientRetry` honours, and the + structured error codes the other resend paths stop on. +- `src/server/responses/core.ts` — two early returns on the marker (pool quota rotation, opaque-blob + recovery); `src/combos/failover.ts` — structured-code stop. See 025. +- `docs-site/src/content/docs/reference/configuration/server.md` — the prelude paragraph. - `tests/responses/ws-upstream.test.ts`, `tests/lib/upstream-retry.test.ts` — oracle updates and new cases. Out of scope, recorded in `020_design_record.md`: resume-by-id after 1006 (Codex does not request background responses, so the vendor resume surface does not apply), the opt-in provider path without a metadata channel (it commits at send today and keeps doing so), the create-frame size -predicate, and `src/server/responses/core.ts`. +predicate, and any core.ts change beyond the two marker guards named in 025. ## Rules for this unit diff --git a/devlog/_plan/260911_ws_commit_boundary/020_design_record.md b/devlog/_plan/260911_ws_commit_boundary/020_design_record.md index 2f7b963c69..36dbd8b533 100644 --- a/devlog/_plan/260911_ws_commit_boundary/020_design_record.md +++ b/devlog/_plan/260911_ws_commit_boundary/020_design_record.md @@ -11,7 +11,8 @@ ## New invariant -- I5 The client commit never precedes the upstream acknowledgment. Before the first +- I5 (exchanges with a metadata channel, i.e. the canonical Codex backend) The client commit never + precedes the upstream acknowledgment. Before the first `response.*`/`error` event the exchange holds no client Response. A failure in that window settles as a JSON error with an honest gateway status, marked non-replayable. @@ -72,6 +73,10 @@ send; a socket exposing `ping` is pinged every 15 s of prelude and stops after c without `ping` is never pinged and keeps the 90 s bound; `fetchWithTransientRetry` returns a non-replayable 504 without a second call (`tests/lib/upstream-retry.test.ts`). +## Audit amendments + +See `025_audit_round1.md`; its deltas override this file where they differ. + ## Risks and their answers - Client behaviour on 504: Codex retries stream requests on 5xx with backoff, which is the same diff --git a/devlog/_plan/260911_ws_commit_boundary/025_audit_round1.md b/devlog/_plan/260911_ws_commit_boundary/025_audit_round1.md new file mode 100644 index 0000000000..69ff1642b9 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/025_audit_round1.md @@ -0,0 +1,29 @@ +# Audit round 1 — xai/grok-4.6 (read-only), dispositions + +Verdict received: FAIL as written. Every finding below is dispositioned; the design record is +amended in place and the scope in 000 is widened to match. + +| # | severity | finding | disposition | +|---|---|---|---| +| 1 | blocker | A marker honoured only by `fetchWithTransientRetry` leaves the Codex pool quota rotation (`shouldRetryCodexPoolAccountQuota`, core.ts:1120) and the combo 5xx hop (core.ts:3004 → `comboFailureDecision`) free to send again after `ws.send()`. | ACCEPTED. core.ts and src/combos/failover.ts enter scope minimally: (a) `shouldRetryCodexPoolAccountQuota` and `opaqueBlobRejectionBodyForRecovery` return early on `isNonReplayableResponse`; (b) the JSON body carries a structured `error.code` (`upstream_no_response`, `upstream_closed_before_response`) and `comboFailureDecision` returns `stop` for those codes, the same mechanism `origin_rejected` already uses. The code set lives in `src/lib/upstream-retry.ts` so combos need no server import. | +| 2 | major | Resetting the 90 s clock on quota/control frames removes the cap for a quota-only socket; it then runs to `connectTimeoutMs` (default 200 s) and settles as a `TimeoutError` 502 from `transportFailureResponse`, not the 504 the record promises. | ACCEPTED as a named behaviour change, with the status fixed. A socket that keeps sending frames or pongs is alive; the record now says so and names the quota-only case explicitly: it waits up to the operator's `connectTimeoutMs`, then the composite signal aborts with `TimeoutError`, and `cancelExchange` maps a pre-commit `TimeoutError` to the same non-replayable 504 instead of rejecting. Only a caller abort (AbortError) rejects. | +| 3 | major | Oracle list is short: metadata budget overflow rows (794), cumulative prelude bound (807), pre-response oversized frame (1064), and the `failureMessage()` helper cases in ws-failure-stage (171, 182, 208) all leave the 200 body-error shape. Foreign-stream 502 conflicts with the in-source note that a reused socket's foreign error must not become an HTTP refusal. | ACCEPTED. All listed tests are updated in wp2. The foreign-stream note was about a 4xx conversion that could authorize account replay; a non-replayable 502 authorizes nothing, and the test now asserts status 502, one send, zero fallback. The source comment is reworded to say that. | +| 4 | major | `failStream` rewrite could skip `cleanup()` and leak the pinger, silence timer, pong listener, or double-settle via `onClose`. | ACCEPTED. Order fixed in the record: `terminal = true; cleanup();` then settle, then `session.dispose()`. `cleanup()` and `commitResponse()` both clear the liveness timers and detach `pong`. | +| 5 | minor | I5 is stated globally while the no-metadata path commits at send. | ACCEPTED. I5 is scoped to exchanges with a metadata channel (the canonical Codex backend). | +| 6 | minor | `connectTimeoutMs` < 90 s makes a post-send abort a 502 connect timeout, not a 504. | ACCEPTED via finding 2: any pre-commit `TimeoutError` becomes the non-replayable 504. | +| 7 | minor | `docs-site/` paragraph on the fixed 90-second prelude deadline (reference/configuration/server.md:38-47) becomes wrong. | ACCEPTED. The paragraph is rewritten in wp3 to describe silence-based liveness and the honest status. | +| 8 | nit | Exact `codexWsFailureDetail` pin, `stage()` fixture defaults, fake-timer stepping for pong tests, feature-detect `ping` not pong. | ACCEPTED. `stage()` defaults `pings: 0, pongs: 0`; pinned strings updated; pong tests step the clock. | + +Not accepted: none. + +## Amended plan deltas (authoritative over 020 where they differ) + +- Scope adds `src/server/responses/core.ts` (two early returns), `src/combos/failover.ts` (one + structured-code stop), `docs-site/src/content/docs/reference/configuration/server.md` (one + paragraph), `tests/responses/ws-failure-stage.test.ts`, `tests/combos/*` only if an existing + decision table needs the new row. +- `cancelExchange(reason)` pre-commit: `reason?.name === "TimeoutError"` → non-replayable 504 + with `upstream_no_response`; anything else → `reject(reason)`. +- Liveness semantics: silence = no inbound message frame and no pong. Any inbound frame resets. + Quota-only sockets are alive and wait for the client or `connectTimeoutMs`. + From 0309c809b033d658fc1ddc5053d8841d5065fdad Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:20:15 +0900 Subject: [PATCH 03/10] docs(devlog): wp2 diff plan for the honest post-send status --- .../260911_ws_commit_boundary/030_wp2_plan.md | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 devlog/_plan/260911_ws_commit_boundary/030_wp2_plan.md diff --git a/devlog/_plan/260911_ws_commit_boundary/030_wp2_plan.md b/devlog/_plan/260911_ws_commit_boundary/030_wp2_plan.md new file mode 100644 index 0000000000..29176614f0 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/030_wp2_plan.md @@ -0,0 +1,36 @@ +# wp2 — honest post-send status and the non-replayable marker + +Previous D (wp1): roadmap locked at c3c1ea6731; direction unchanged — the fix is the commit +boundary, not the transport choice. 025 deltas are authoritative over 020. + +## Files and exact changes + +### src/lib/upstream-retry.ts +- Add a WeakSet with markResponseNonReplayable(res) and isNonReplayableResponse(res). +- Add NON_REPLAYABLE_UPSTREAM_CODES = {"upstream_no_response", "upstream_closed_before_response"} and isNonReplayableUpstreamCode(code). +- fetchWithTransientRetry loop guard: return res when isNonReplayableResponse(res). + +### src/combos/failover.ts +- comboFailureDecision: after the 499/origin_rejected checks, return "stop" when isNonReplayableUpstreamCode(options?.code). + +### src/server/responses/core.ts +- shouldRetryCodexPoolAccountQuota: first line returns false on isNonReplayableResponse(response). +- opaqueBlobRejectionBodyForRecovery: same early return undefined. + +### src/server/responses/codex-ws-wire.ts +- codexWsPreResponseFailure(status: 502 | 504, message, prelude: Headers): Response — JSON body { error: { type: "upstream_error", code, message } }, code by status (504 upstream_no_response, 502 upstream_closed_before_response), headers = prelude snapshot + content-type application/json + cache-control no-store, marked non-replayable. + +### src/server/responses/codex-ws-exchange.ts +- failStream(error, status = 502): when sent && !responseCommitted && metadata: terminal = true; cleanup(); resolve(codexWsPreResponseFailure(status, message, metadata.snapshot())); close the unused controller; session.dispose(). Otherwise the existing body-error path. (The non-metadata path commits at send.) +- cancelExchange(reason) when sent && !responseCommitted && metadata: TimeoutError -> failStream(reason, 504); otherwise terminal = true; cleanup(); session.dispose(); reject(reason). +- The prelude timer expiry calls failStream(..., 504); liveness itself is wp3. +- Reword the foreign-stream comment: a pre-response failure settles as a non-replayable 502; the 4xx projection stays reserved for a genuine refused create. + +### Tests +- ws-upstream.test.ts: update 794/807 (metadata overflow -> 502 JSON, isCodexWsUpstreamResponse false), 873 foreign -> 502 + one send, 1064 oversized pre-response -> 502, 1180 abort after open -> the fetch rejects with the caller reason and the socket is closed, 1218 -> 502, 1262 -> 504 with sends === 1, 1533/1547 -> 502 with the same messages in error.message. +- ws-failure-stage.test.ts: failureMessage() returns error.message from a 5xx JSON body, else the thrown body error. +- New: upstream-transient-retry.test.ts — a marked 504 returns after one send; an unmarked 504 still retries. combos test — comboFailureDecision(504, "Provider error 504", { code: "upstream_no_response" }) is stop. + +## Verification +NOT RUN locally (owner rule). Remote CI on the final head in wp4. + From 42988a169383a0ffebe53005811aa513135dec03 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:33:28 +0900 Subject: [PATCH 04/10] fix(codex-ws): settle a post-send, pre-response failure as an honest gateway status The Codex WebSocket relay committed a 200 SSE Response the moment a failure landed after ws.send(), on the reasoning that a 5xx could make the pre-stream retry wrapper resend the frame. That was right about the resend and wrong about the status: it turned "no response" into "a response that failed", removed the code a user agent needs for its own retry policy, and neutered the client's first-byte timeout with chunked headers (#4191, #4083). Before the first response.*/error event the exchange now resolves a JSON 504 (prelude silence, or the proxy's own connect deadline) or 502 (close, transport error, prelude overflow, foreign stream) carrying the stage detail and the metadata snapshot. The response is marked non-replayable: fetchWithTransientRetry returns it without a second send, the Codex pool quota rotation and opaque-blob recovery ignore it, and comboFailureDecision stops on its structured error code. A caller abort in that window rejects with the caller's reason and disposes the socket, cancelling the turn. Behaviour after the response has started is unchanged. Local suite: NOT RUN by owner rule; remote CI on the final head is the gate. --- src/combos/failover.ts | 5 ++ src/lib/upstream-retry.ts | 37 ++++++++++- src/server/responses/codex-ws-exchange.ts | 47 +++++++++++--- src/server/responses/codex-ws-wire.ts | 34 ++++++++++ src/server/responses/core.ts | 7 +- .../upstream-transient-retry.test.ts | 42 +++++++++++- tests/responses/ws-failure-stage.test.ts | 18 +++++- tests/responses/ws-upstream.test.ts | 64 +++++++++++++------ ...uter-combo-failover-classification.test.ts | 10 +++ 9 files changed, 230 insertions(+), 34 deletions(-) diff --git a/src/combos/failover.ts b/src/combos/failover.ts index 3868bc4b0c..e592298d88 100644 --- a/src/combos/failover.ts +++ b/src/combos/failover.ts @@ -1,5 +1,6 @@ import { parseResetCooldownMs } from "../codex/routing"; import { classifyError, isCyberPolicyCode } from "../lib/errors"; +import { isNonReplayableUpstreamCode } from "../lib/upstream-retry"; import type { OcxComboTarget } from "../types"; import { targetKey } from "./types"; import { @@ -413,6 +414,10 @@ export function comboFailureDecision( ): ComboFailureDecision { if (status === 499) return "stop"; if (message.toLowerCase().includes("origin_rejected")) return "stop"; + // The origin may already be executing this turn (the Codex WebSocket relay sent the create + // frame and never saw a response event). Hopping would send the same request to a second + // target while the first may still be generating; the honest status goes to the client. + if (isNonReplayableUpstreamCode(options?.code)) return "stop"; // Cyber policy is a hard non-retryable refusal — honor structured code even when // classificationText was truncated before the JSON code field. if (isCyberPolicyCode(options?.code)) return "stop"; diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 74302af51f..afdbbcf3ae 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -16,6 +16,39 @@ */ import { clearableDeadline } from "./abort"; +/** + * Responses the origin may already be executing. RFC 9110 §9.2.2 forbids an intermediary + * from automatically repeating a non-idempotent request; when a post-send transport (the + * Codex WebSocket relay) settles a gateway status because the origin never acknowledged the + * turn, the request body must not be sent again by this process — not by the transient-5xx + * layer below, not by a pool account rotation, not by a combo hop. The status is returned + * to the caller so the user agent can apply its own retry policy, exactly as it does on the + * direct path. The WeakSet is the in-process marker; the structured error codes are the + * marker that survives body re-wrapping (combo failure consumption re-parses the JSON). + */ +const nonReplayableResponses = new WeakSet(); + +export function markResponseNonReplayable(response: Response): void { + nonReplayableResponses.add(response); +} + +export function isNonReplayableResponse(response: Response): boolean { + return nonReplayableResponses.has(response); +} + +/** Origin never produced a response event; the turn may still be executing. */ +export const UPSTREAM_NO_RESPONSE_CODE = "upstream_no_response"; +/** Transport closed after the send, before any response event. */ +export const UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE = "upstream_closed_before_response"; +const NON_REPLAYABLE_UPSTREAM_CODES: ReadonlySet = new Set([ + UPSTREAM_NO_RESPONSE_CODE, + UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE, +]); + +export function isNonReplayableUpstreamCode(code: unknown): boolean { + return typeof code === "string" && NON_REPLAYABLE_UPSTREAM_CODES.has(code); +} + // 1 initial + 2 retries: the pool may hold more than one stale socket. const RESET_RETRY_MAX_ATTEMPTS = 3; const RESET_RETRY_BASE_DELAY_MS = 150; @@ -395,7 +428,9 @@ export async function fetchWithTransientRetry( let attemptStart = Date.now(); let res = await fetchWithResetRetry(countedFetch, { ...opts, attempts: remaining() }); for (let attempt = 0; sent < budget; attempt++) { - if (res.ok || !isTransientUpstreamStatus(res.status)) return res; + // A non-replayable gateway status was settled after the request body had already left + // for the origin; retrying it here is the automatic resend the marker exists to forbid. + if (res.ok || !isTransientUpstreamStatus(res.status) || isNonReplayableResponse(res)) return res; // Checked before cancelResponseBodyBestEffort so an already-aborted caller never receives // a response whose body we just cancelled. if (opts.abortSignal?.aborted) return res; diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 64a8e5aba4..7eee36b420 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -6,7 +6,7 @@ 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, - codexWsFailureDetail, type CodexWsFailureStage } from "./codex-ws-wire"; + codexWsFailureDetail, codexWsPreResponseFailure, type CodexWsFailureStage } from "./codex-ws-wire"; interface ExchangeOptions { session: CodexWsSession; @@ -148,7 +148,8 @@ export function codexWsExchange(options: ExchangeOptions): Promise { }); const commitResponse = () => { - if (responseCommitted) return; + // A pre-response settlement (gateway status) has already resolved this exchange. + if (responseCommitted || terminal) return; responseCommitted = true; clearTimeout(preludeTimer); const responseHeaders = metadata?.snapshot() ?? new Headers(); @@ -159,11 +160,26 @@ export function codexWsExchange(options: ExchangeOptions): Promise { resolve(response); }; - const failStream = (error: unknown) => { + const failStream = (error: unknown, status: 502 | 504 = 502) => { if (terminal) return; terminal = true; - // A frame may already be executing upstream. Settle as a body failure, - // never a fetch rejection/5xx that the pre-stream wrapper could resend. + if (sent && !responseCommitted && metadata) { + // Nothing has been promised to the client yet, so the honest answer is a gateway + // status, not a 200 whose body then fails. The frame may already be executing + // upstream: the response is marked non-replayable so no layer of this process sends + // it again, and the client applies its own retry policy as it would on the direct + // path. Same settle order as a refused create: snapshot, detach, close, dispose. + const prelude = metadata.snapshot(); + cleanup(); + try { controller?.close(); } catch { /* unused stream already closed */ } + session.dispose(); + const message = error instanceof Error ? error.message : String(error); + resolve(codexWsPreResponseFailure(status, message, prelude)); + return; + } + // A response is already flowing (or this transport has no metadata channel and + // committed at send). Settle as a body failure, never a fetch rejection/5xx that the + // pre-stream wrapper could resend. if (sent) commitResponse(); cleanup(); try { controller?.error(typeof error === "string" ? new Error(error) : error); } catch { /* stream already done */ } @@ -188,6 +204,20 @@ export function codexWsExchange(options: ExchangeOptions): Promise { reject(reason); return; } + if (!responseCommitted && metadata) { + // Sent, unacknowledged. The proxy's own connect deadline is an origin-silence + // verdict and settles like one; a caller abort is the caller's decision, so the + // exchange rejects with that reason and disposing the socket cancels the turn. + if ((reason as { name?: unknown } | null)?.name === "TimeoutError") { + failStream(`codex websocket response did not start before the connect deadline${codexWsFailureDetail(failureStage())}`, 504); + return; + } + terminal = true; + cleanup(); + session.dispose(); + reject(reason); + return; + } failStream(reason); }; const onAbort = () => cancelExchange(signal?.reason ?? new DOMException("The operation was aborted.", "AbortError")); @@ -237,7 +267,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { if (!metadata) commitResponse(); else if (!responseCommitted && !terminal) { preludeTimer = setTimeout( - () => failStream(`codex websocket response prelude timed out${codexWsFailureDetail(failureStage())}`), + () => failStream(`codex websocket response prelude timed out${codexWsFailureDetail(failureStage())}`, 504), CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, ); } @@ -288,8 +318,9 @@ export function codexWsExchange(options: ExchangeOptions): Promise { if (!controlFrame && !type.startsWith("response.") && type !== "error") return; if (!controlFrame) { try { correlation?.accept(normalized.payload); } catch (error) { failStream(error); return; } - // Correlation must run first: a reused socket's foreign-stream error - // must not become an HTTP refusal that could authorize account replay. + // Correlation must run first: a reused socket's foreign-stream error settles as a + // non-replayable 502 above, never as the refused-create 4xx projection below, which + // is the one status family that could authorize an account replay. if (metadata && sent && !responseCommitted && type === "error") { let rejection: Response | null; try { rejection = wrappedRejectionResponse(normalized.payload, metadata.snapshot()); } diff --git a/src/server/responses/codex-ws-wire.ts b/src/server/responses/codex-ws-wire.ts index 770ce21b54..64b732b22e 100644 --- a/src/server/responses/codex-ws-wire.ts +++ b/src/server/responses/codex-ws-wire.ts @@ -1,4 +1,9 @@ import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; +import { + markResponseNonReplayable, + UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE, + UPSTREAM_NO_RESPONSE_CODE, +} from "../../lib/upstream-retry"; // If the 101 never arrives (network black hole), give SSE a chance well before // the caller's connect timeout (default 200s) would fire. export const UPGRADE_DEADLINE_MS = 10_000; @@ -48,6 +53,35 @@ export function markCodexWsResponse(response: Response, observed: boolean): void if (observed) quotaObservedResponses.add(response); } +/** + * The honest settlement for an exchange that sent its create frame and never saw a + * response event. + * + * Before this existed the relay committed a 200 SSE Response and errored its body, on the + * reasoning that a 5xx could make the pre-stream retry wrapper resend the frame. That + * reasoning was right about the resend and wrong about the status: it turned "no response" + * into "a response that failed", removed the code a user agent uses for its own retry + * policy, and neutered the client's first-byte timeout with chunked headers. The client + * on the direct path receives a gateway status in this situation and retries under its own + * policy; this response restores that equivalence. The resend is forbidden by the + * non-replayable marker instead (see upstream-retry.ts), and the structured code lets the + * combo failover reach the same verdict after it re-parses the body. + * + * 504 is the origin's silence (nothing at all, or nothing but liveness, for the tolerated + * window); 502 is a transport that closed or misbehaved after the send. Both carry the + * content-free stage detail in the message and the metadata snapshot in the headers, the + * same way a refused create does, so quota captured during the prelude is not lost. + */ +export function codexWsPreResponseFailure(status: 502 | 504, message: string, prelude: Headers): Response { + const headers = new Headers(prelude); + headers.set("content-type", "application/json"); + headers.set("cache-control", "no-store"); + const code = status === 504 ? UPSTREAM_NO_RESPONSE_CODE : UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE; + const response = new Response(JSON.stringify({ error: { type: "upstream_error", code, message } }), { status, headers }); + markResponseNonReplayable(response); + return response; +} + const CLOSED_BEFORE_TERMINAL = "codex websocket closed before a Responses terminal event"; /** diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 425621814c..f24b1b79a5 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -210,6 +210,7 @@ import { applyUpstreamRecoveryInit, fetchWithResetRetry, fetchWithTransientRetry, + isNonReplayableResponse, prepareSameTarget429Wait, } from "../../lib/upstream-retry"; import { @@ -827,7 +828,8 @@ async function opaqueBlobRejectionBodyForRecovery( signal: AbortSignal, ): Promise { if ( - response.status < 400 + isNonReplayableResponse(response) + || response.status < 400 || (response.status >= 500 && response.status !== 502) || adapterName !== "openai-responses" || alreadyAttempted @@ -1121,6 +1123,9 @@ export async function shouldRetryCodexPoolAccountQuota( response: Response, signal?: AbortSignal, ): Promise { + // A post-send WebSocket gateway status must not become a second account's send; the + // body carries no quota evidence either, but the marker is the contract, not the prose. + if (isNonReplayableResponse(response)) return false; if (response.status === 402 || response.status === 429) return true; if (response.status < 500 || response.status >= 600) return false; try { diff --git a/tests/providers/upstream-transient-retry.test.ts b/tests/providers/upstream-transient-retry.test.ts index 58199ac55a..77a224eb36 100644 --- a/tests/providers/upstream-transient-retry.test.ts +++ b/tests/providers/upstream-transient-retry.test.ts @@ -1,5 +1,11 @@ import { describe, expect, test } from "bun:test"; -import { fetchWithTransientRetry, isTransientUpstreamStatus } from "../../src/lib/upstream-retry"; +import { + fetchWithTransientRetry, + isNonReplayableResponse, + isNonReplayableUpstreamCode, + isTransientUpstreamStatus, + markResponseNonReplayable, +} from "../../src/lib/upstream-retry"; import { transientRetryPolicyFor } from "../../src/providers/key-failover"; import type { OcxProviderConfig } from "../../src/types"; @@ -50,6 +56,40 @@ describe("transientRetryPolicyFor", () => { }); describe("fetchWithTransientRetry", () => { + test("a non-replayable gateway status is returned after one send, body intact", async () => { + // The Codex WebSocket relay settles a 504 when the origin never acknowledged a frame it + // already sent. 504 is transient by status, but the origin may be executing that turn: + // the marker, not the status, decides that this layer must not send the body again. + let sends = 0; + const res = await fetchWithTransientRetry(async () => { + sends += 1; + const response = bodyResponse(504); + markResponseNonReplayable(response); + return response; + }, { attempts: 3, slowAttemptMs: 60_000 }); + expect(sends).toBe(1); + expect(res.status).toBe(504); + expect(isNonReplayableResponse(res)).toBe(true); + expect((res as Response & { __wasCancelled: () => boolean }).__wasCancelled()).toBe(false); + }); + + test("an unmarked 504 keeps the transient retry", async () => { + let sends = 0; + const res = await fetchWithTransientRetry(async () => { + sends += 1; + return bodyResponse(504); + }, { attempts: 2, slowAttemptMs: 60_000 }); + expect(sends).toBe(2); + expect(res.status).toBe(504); + }); + + test("the structured codes name exactly the two post-send verdicts", () => { + expect(isNonReplayableUpstreamCode("upstream_no_response")).toBe(true); + expect(isNonReplayableUpstreamCode("upstream_closed_before_response")).toBe(true); + expect(isNonReplayableUpstreamCode("upstream_error")).toBe(false); + expect(isNonReplayableUpstreamCode(undefined)).toBe(false); + }); + test("attempts is one total-send budget, not a per-layer multiplier", async () => { // The two layers used to multiply: attempts:3 meant 3 transient rounds each independently // retrying 3 connection resets, so a single call could emit up to 9 upstream sends. All diff --git a/tests/responses/ws-failure-stage.test.ts b/tests/responses/ws-failure-stage.test.ts index af4db2136a..d578d801d4 100644 --- a/tests/responses/ws-failure-stage.test.ts +++ b/tests/responses/ws-failure-stage.test.ts @@ -98,6 +98,20 @@ async function failureMessage(script: (ws: FakeWebSocket) => void): Promise { + if (response.status >= 500) { + const body = await response.json() as { error?: { message?: unknown } }; + if (typeof body.error?.message !== "string") throw new Error("expected a gateway failure body"); + return body.error.message; + } try { await response.text(); } catch (error) { @@ -219,7 +233,8 @@ describe("codexWsUpstreamFetch failure reporting", () => { await opened.promise; jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); const response = await pending; - await expect(response.text()).rejects.toThrow( + expect(response.status).toBe(504); + expect(await failureMessageOf(response)).toMatch( /prelude timed out \[cause=no-upstream-frame request=\d+B sent=yes frames=0 control=0 relayed=0/, ); } finally { @@ -227,4 +242,3 @@ describe("codexWsUpstreamFetch failure reporting", () => { } }); }); - diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index 9e4f689208..6afb9e312c 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -3,7 +3,7 @@ import { providerFetch } from "../../src/server/responses/fetch-helpers"; import { handleResponses } from "../../src/server/responses"; import { isEagerRelaySseResponse } from "../../src/server/relay"; import { isWin32EagerRewrite } from "../../src/lib/bun-stream-caps"; -import { fetchWithTransientRetry } from "../../src/lib/upstream-retry"; +import { fetchWithTransientRetry, isNonReplayableResponse } from "../../src/lib/upstream-retry"; import { codexWsExchange } from "../../src/server/responses/codex-ws-exchange"; import { CodexWsSession } from "../../src/server/responses/codex-ws-session"; import { prepareCodexWsRequest } from "../../src/server/responses/codex-ws-request"; @@ -798,9 +798,14 @@ describe("codexWsUpstreamFetch", () => { for (const [name, value] of Object.entries(headers)) expect(response.headers.get(name)).toBe(value); expect(await response.json()).toEqual({ error: refusal.error }); } else { - expect(response.status).toBe(200); - expect(isCodexWsUpstreamResponse(response)).toBe(true); - await expect(response.text()).rejects.toThrow("metadata"); + // The overflow lands before any response event: an honest 502, never a 200 whose + // body then fails, and never the HTTP fallback (the frame was sent). + expect(response.status).toBe(502); + expect(isCodexWsUpstreamResponse(response)).toBe(false); + expect(response.headers.get("content-type")).toBe("application/json"); + const failure = ((await response.json()) as { error: { code: string; message: string } }).error; + expect(failure.code).toBe("upstream_closed_before_response"); + expect(failure.message).toContain("metadata"); } }); @@ -808,8 +813,8 @@ describe("codexWsUpstreamFetch", () => { const response = await receive({ ...refusal, headers: boundedHeaders(5, "x".repeat(4096)) }, [ { type: "codex.response.metadata", headers: boundedHeaders(4, "y".repeat(4096)) }, ]); - expect(response.status).toBe(200); - await expect(response.text()).rejects.toThrow("metadata"); + expect(response.status).toBe(502); + expect(((await response.json()) as { error: { message: string } }).error.message).toContain("metadata"); }); test.each([ @@ -896,8 +901,11 @@ describe("codexWsUpstreamFetch", () => { expect(session.reserve()).toBe(true); const response = await codexWsExchange(options); if (foreign) { - expect(response.status).toBe(200); - await expect(response.text()).rejects.toThrow("identity mismatch"); + // A foreign stream before any response event is a transport that misbehaved after + // the send: non-replayable 502, and never the 4xx refusal projection. + expect(response.status).toBe(502); + expect(isCodexWsUpstreamResponse(response)).toBe(false); + expect(((await response.json()) as { error: { message: string } }).error.message).toContain("identity mismatch"); } else { expect(response.status).toBe(429); expect(isCodexWsUpstreamResponse(response)).toBe(false); @@ -1070,7 +1078,10 @@ describe("codexWsUpstreamFetch", () => { throw new Error("fallback must not run after open"); }) as unknown as typeof fetch); - await expect(response.text()).rejects.toThrow("frame exceeds the response size limit"); + // No response event preceded the oversized frame, so the exchange never owed a stream. + expect(response.status).toBe(502); + expect(((await response.json()) as { error: { message: string } }).error.message) + .toContain("frame exceeds the response size limit"); expect(FakeWebSocket.instances[0].closed).toBe(true); }); @@ -1189,9 +1200,9 @@ describe("codexWsUpstreamFetch", () => { await opened.promise; controller.abort(new Error("turn cancelled")); - const response = await pending; - - await expect(response.text()).rejects.toThrow("turn cancelled"); + // Sent and unacknowledged: the caller's abort is the caller's decision, so the fetch + // itself rejects with that reason and closing the socket cancels the upstream turn. + await expect(pending).rejects.toThrow("turn cancelled"); expect(FakeWebSocket.instances[0].closed).toBe(true); }); @@ -1215,7 +1226,7 @@ describe("codexWsUpstreamFetch", () => { await response.text(); }); - test("post-send prelude overflow settles as an errored body without HTTP fallback", async () => { + test("post-send prelude overflow settles as a non-replayable 502 without HTTP fallback", async () => { installFake(ws => { ws.emit("open", {}); ws.emit("message", { data: JSON.stringify({ type: "codex.response.metadata", headers: { "x-models-etag": "x".repeat(CODEX_WS_METADATA_MAX_BYTES) } }) }); @@ -1225,9 +1236,10 @@ describe("codexWsUpstreamFetch", () => { resends++; return new Response("unexpected resend"); }) as typeof fetch); - expect(response.status).toBe(200); - expect(isCodexWsUpstreamResponse(response)).toBe(true); - await expect(response.text()).rejects.toThrow("metadata"); + expect(response.status).toBe(502); + expect(isCodexWsUpstreamResponse(response)).toBe(false); + expect(isNonReplayableResponse(response)).toBe(true); + expect(((await response.json()) as { error: { message: string } }).error.message).toContain("metadata"); expect(resends).toBe(0); expect(FakeWebSocket.instances[0].sent).toHaveLength(1); }); @@ -1259,7 +1271,7 @@ describe("codexWsUpstreamFetch", () => { } }); - test("the first-response deadline settles a sent request through the outer retry wrapper without resending", async () => { + test("the first-response deadline settles a sent request as a 504 the outer retry wrapper does not resend", async () => { const { fetchWithTransientRetry } = await import("../../src/lib/upstream-retry"); jest.useFakeTimers(); const opened = Promise.withResolvers(); @@ -1277,8 +1289,14 @@ describe("codexWsUpstreamFetch", () => { await opened.promise; jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); const response = await pending; - expect(response.status).toBe(200); - await expect(response.text()).rejects.toThrow("prelude timed out"); + // 504 is a transient status for the wrapper; the non-replayable marker is what stops + // the second send, and the status is what lets the client apply its own policy. + expect(response.status).toBe(504); + expect(isNonReplayableResponse(response)).toBe(true); + const failure = ((await response.json()) as { error: { code: string; message: string } }).error; + expect(failure.code).toBe("upstream_no_response"); + expect(failure.message).toContain("prelude timed out"); + expect(failure.message).toContain("cause=no-upstream-frame"); expect(sends).toBe(1); expect(http).toBe(0); } finally { @@ -1539,7 +1557,8 @@ describe("oversized Codex create frames", () => { throw new Error("fallback must not run after open"); }) as unknown as typeof fetch); - await expect(response.text()).rejects.toThrow( + expect(response.status).toBe(502); + expect(((await response.json()) as { error: { message: string } }).error.message).toMatch( /rejected the request frame as too large \(close 1009 Message Too Big\)/, ); }); @@ -1553,7 +1572,10 @@ describe("oversized Codex create frames", () => { throw new Error("fallback must not run after open"); }) as unknown as typeof fetch); - await expect(response.text()).rejects.toThrow("closed before a Responses terminal event (close 1006)"); + expect(response.status).toBe(502); + const failure = ((await response.json()) as { error: { code: string; message: string } }).error; + expect(failure.code).toBe("upstream_closed_before_response"); + expect(failure.message).toContain("closed before a Responses terminal event (close 1006)"); }); test("dials the configured provider's own wss URL for an opt-in upstream", async () => { diff --git a/tests/routing/router-combo-failover-classification.test.ts b/tests/routing/router-combo-failover-classification.test.ts index 742f411bfb..93b58170a6 100644 --- a/tests/routing/router-combo-failover-classification.test.ts +++ b/tests/routing/router-combo-failover-classification.test.ts @@ -129,6 +129,16 @@ describe("combo failure hop/stop verdicts", () => { test("INVARIANT: a structured model lifecycle 410 still hops", () => { expect(comboFailureDecision(410, "model retired", { code: "model_end_of_life" })).toBe("hop"); }); + + test("a post-send gateway status from the Codex WebSocket relay never hops", () => { + // The relay sent the create frame and the origin never acknowledged it (504) or the + // transport closed first (502). The turn may still be executing at the first target, so + // a second target must not receive the same request; the client decides the retry. + expect(comboFailureDecision(504, "Provider error 504", { code: "upstream_no_response" })).toBe("stop"); + expect(comboFailureDecision(502, "Provider error 502", { code: "upstream_closed_before_response" })).toBe("stop"); + // The same statuses without the structured code keep the ordinary transient hop. + expect(comboFailureDecision(504, "Provider error 504")).toBe("hop"); + }); }); describe("cooled targets are not selectable", () => { From d79eac5c179ae96ab2aea4f13e8cf2837eac5ff3 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:35:16 +0900 Subject: [PATCH 05/10] docs(devlog): wp3 diff plan for prelude liveness --- .../260911_ws_commit_boundary/040_wp3_plan.md | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 devlog/_plan/260911_ws_commit_boundary/040_wp3_plan.md diff --git a/devlog/_plan/260911_ws_commit_boundary/040_wp3_plan.md b/devlog/_plan/260911_ws_commit_boundary/040_wp3_plan.md new file mode 100644 index 0000000000..8034ba2721 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/040_wp3_plan.md @@ -0,0 +1,30 @@ +# wp3 — liveness replaces the fixed prelude deadline + +Previous D (wp2): honest 502/504 with the non-replayable marker landed at 42988a1693; draft PR #4256 opened so remote CI runs on that head. Direction unchanged. + +## Files and exact changes + +### src/server/responses/codex-ws-wire.ts +- CODEX_WS_LIVENESS_PING_INTERVAL_MS = 15_000. +- CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS keeps 90_000; its comment now defines it as the longest inbound silence (no message frame, no pong) tolerated before the first response event. +- CodexWsFailureStage gains pings and pongs (numbers); codexWsFailureDetail appends " pings=N pongs=N" after elapsed, inside the bracket. + +### src/server/responses/codex-ws-exchange.ts +- Counters pings, pongs. Timers silenceTimer (replaces preludeTimer) and pingTimer. +- armSilence(): clearTimeout(silenceTimer); if (responseCommitted || terminal) return; silenceTimer = setTimeout(() => failStream("codex websocket response prelude timed out" + detail, 504), CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS). +- schedulePing(): only when typeof ws.ping === "function"; pingTimer = setTimeout(() => { if (responseCommitted || terminal) return; try { ws.ping(); pings += 1; } catch { return; } schedulePing(); }, CODEX_WS_LIVENESS_PING_INTERVAL_MS). +- onPong(): pongs += 1; if (!responseCommitted) armSilence(). Listener added with the others, removed in cleanup(). +- After a successful send on the metadata path: armSilence(); schedulePing(). onMessage calls armSilence() while !responseCommitted (after the terminal guard). +- commitResponse() and cleanup() clear both timers; cleanup() removes the pong listener. +- The non-metadata path is untouched (commits at send; no liveness). + +### docs-site/src/content/docs/reference/configuration/server.md +- Replace the "fixed 90-second response-prelude deadline" paragraph: silence-based liveness, ping every 15 s, any inbound frame or pong resets, 90 s of nothing settles a non-replayable 504, closes/transport errors before the first response event settle 502, connectTimeoutMs remains the outer bound and a pre-response connect timeout is the same 504; no HTTP resend either way. + +### Tests +- tests/responses/ws-upstream.test.ts, new describe "prelude liveness": (a) a socket whose ping() emits pong stays alive across 7 x 15 s steps (105 s > 90 s) and the response still completes with one send and zero HTTP; (b) a socket whose ping() never pongs settles 504 at 90 s with pongs=0 in the message; (c) after response.created the pinger stops (no further ping calls across 60 s). The existing first-response-deadline test already covers a socket without ping(). +- tests/responses/ws-failure-stage.test.ts: stage() defaults pings: 0, pongs: 0; the two exact toBe strings gain " pings=0 pongs=0". + +## Verification +NOT RUN locally (owner rule). Remote CI on the final head in wp4. + From 075b9f39f5db19cd4f70853501d7aecf04d95a6c Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:49:38 +0900 Subject: [PATCH 06/10] feat(codex-ws): replace the fixed prelude deadline with silence-based liveness The 90 s response-prelude timer folded "the origin is dead" and "the origin is slow" into one number the proxy owned. Dead is a liveness question and WebSocket answers it natively; slow already has owners: the client's own deadline and the operator's connectTimeoutMs. While waiting for the first response event the exchange now pings every 15 s on sockets that expose ping(), any inbound frame or pong resets the 90 s silence clock, and only silence settles the 504. A peer that never pongs keeps exactly the previous bound; a peer that does can never trip it while alive. The failure stage names the pings sent and pongs received so a field report can tell an unanswered peer from a slow one (#4191, #4083). Local suite: NOT RUN by owner rule; remote CI on the final head is the gate. --- .../docs/reference/configuration/server.md | 22 ++-- src/server/responses/codex-ws-exchange.ts | 55 ++++++++-- src/server/responses/codex-ws-wire.ts | 18 +++- src/server/responses/ws-upstream.ts | 2 +- tests/responses/ws-failure-stage.test.ts | 8 +- tests/responses/ws-upstream.test.ts | 101 ++++++++++++++++++ 6 files changed, 184 insertions(+), 22 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 6a3533f10c..b5c6158090 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -35,16 +35,18 @@ runs helper features around provider requests. | `visionSidecar?` | `OcxVisionSidecarConfig` | on when usable | Image-description sidecar options. | | `images?` | `OcxImagesConfig` | automatic OpenAI selection | Standalone Images relay options for Codex `image_gen`. | -The canonical ChatGPT upstream WebSocket has a fixed 90-second response-prelude deadline, -measured after sending the create frame. Quota and response-metadata control frames do not -reset it; the first non-control Responses event ends it. This is not a total generation -deadline, and neither `connectTimeoutMs` nor `stallTimeoutSec` retunes the 90 seconds -themselves. That constant is only the WebSocket-specific upper bound: the exchange runs under -the signal `connectTimeoutMs` (default 200s) aborts, and that abort cancels an already-sent -create before the prelude timer can fire. A `connectTimeoutMs` below 90 seconds therefore -ends the wait earlier, so the deadline a request actually gets is the shorter of the two. If -either expires after sending, the stream fails without an HTTP resend, avoiding duplicate -inference. +While the canonical ChatGPT upstream WebSocket waits for the first Responses event after +sending the create frame, it watches for liveness rather than a fixed deadline. The proxy pings +the socket every 15 seconds; any inbound frame — quota, response metadata, or a pong — resets a +90-second silence clock, and only 90 seconds with nothing at all settles the request as an +HTTP 504 with an `upstream_no_response` error. A slow but alive origin therefore waits for the +client's own deadline or for `connectTimeoutMs` (default 200s), whichever comes first; a +connect timeout that fires after the create frame was sent settles as the same 504. A socket +that closes or errors before the first Responses event settles as an HTTP 502 with +`upstream_closed_before_response`. These statuses are never retried inside the proxy — the +frame may already be executing upstream, so the client applies its own retry policy exactly as +it would when connected to the backend directly. Once the response has started, a later drop +surfaces inside the stream as before. `stallTimeoutSec` is unrelated to this window. `noProxy` accepts either a comma-separated string or an array. Both forms add entries without replacing an inherited `NO_PROXY`: diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 7eee36b420..f682e3d73d 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -4,7 +4,7 @@ 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_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, MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, markCodexWsResponse, normalizeResponsesWsRelayEvent, closedBeforeTerminalMessage, codexWsFailureDetail, codexWsPreResponseFailure, type CodexWsFailureStage } from "./codex-ws-wire"; @@ -101,6 +101,8 @@ export function codexWsExchange(options: ExchangeOptions): Promise { let upstreamFrames = 0; let controlFrames = 0; let relayedEvents = 0; + let pings = 0; + let pongs = 0; let sentAt: number | null = null; let firstFrameAt: number | null = null; let controller: ReadableStreamDefaultController | null = null; @@ -108,7 +110,11 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const metadata = url === CODEX_RESPONSES_HTTP_URL ? new CodexWsMetadata(onQuota) : null; const correlation = session.retainable ? new CodexWsCorrelation(session.reused, id => session.hasCompleted(id)) : null; let detachOwner = () => {}; - let preludeTimer: ReturnType | undefined; + // 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. + let silenceTimer: ReturnType | undefined; + let pingTimer: ReturnType | undefined; const stream = new ReadableStream({ start(c) { controller = c; }, cancel() { @@ -121,7 +127,8 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const cleanup = () => { clearTimeout(upgradeTimer); - clearTimeout(preludeTimer); + clearTimeout(silenceTimer); + clearTimeout(pingTimer); signal?.removeEventListener("abort", onAbort); metadata?.finish(); correlation?.finish(); @@ -130,6 +137,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { ws.removeEventListener("message", onMessage); ws.removeEventListener("close", onClose); ws.removeEventListener("error", onError); + ws.removeEventListener("pong", onPong); }; /** @@ -145,13 +153,16 @@ export function codexWsExchange(options: ExchangeOptions): Promise { relayedEvents, firstFrameMs: sentAt !== null && firstFrameAt !== null ? Math.max(0, firstFrameAt - sentAt) : null, elapsedMs: sentAt !== null ? Math.max(0, Date.now() - sentAt) : null, + pings, + pongs, }); const commitResponse = () => { // A pre-response settlement (gateway status) has already resolved this exchange. if (responseCommitted || terminal) return; responseCommitted = true; - clearTimeout(preludeTimer); + clearTimeout(silenceTimer); + clearTimeout(pingTimer); const responseHeaders = metadata?.snapshot() ?? new Headers(); responseHeaders.set("content-type", "text/event-stream; charset=utf-8"); const response = new Response(stream, { status: 200, headers: responseHeaders }); @@ -186,6 +197,33 @@ export function codexWsExchange(options: ExchangeOptions): Promise { session.dispose(); }; + /** (Re)start the silence bound; every inbound frame or pong is proof of life. */ + const armSilence = () => { + clearTimeout(silenceTimer); + if (responseCommitted || terminal) return; + silenceTimer = setTimeout( + () => failStream(`codex websocket response prelude timed out${codexWsFailureDetail(failureStage())}`, 504), + CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, + ); + }; + /** Ping on a fixed interval until the response starts; a socket without ping() is never pinged. */ + const schedulePing = () => { + const socket = ws as WebSocket & { ping?: (data?: string) => void }; + if (typeof socket.ping !== "function" || responseCommitted || terminal) return; + pingTimer = setTimeout(() => { + if (responseCommitted || terminal) return; + try { socket.ping(); } catch { return; } + pings += 1; + // ping() may close the socket synchronously and settle the exchange; re-check. + if (!terminal) schedulePing(); + }, CODEX_WS_LIVENESS_PING_INTERVAL_MS); + }; + const onPong = () => { + if (terminal) return; + pongs += 1; + if (!responseCommitted) armSilence(); + }; + const upgradeTimer = setTimeout(() => { if (opened || settledPreOpen) return; settledPreOpen = true; @@ -238,6 +276,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { ws.removeEventListener("message", onMessage); ws.removeEventListener("close", onClose); ws.removeEventListener("error", onError); + ws.removeEventListener("pong", onPong); session.dispose(); reject(error); return; @@ -266,16 +305,15 @@ export function codexWsExchange(options: ExchangeOptions): Promise { } if (!metadata) commitResponse(); else if (!responseCommitted && !terminal) { - preludeTimer = setTimeout( - () => failStream(`codex websocket response prelude timed out${codexWsFailureDetail(failureStage())}`, 504), - CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, - ); + armSilence(); + schedulePing(); } }; const onMessage = (event: MessageEvent) => { if (!controller || terminal) return; received = true; + if (!responseCommitted) armSilence(); upstreamFrames += 1; if (firstFrameAt === null) firstFrameAt = Date.now(); const text = typeof event.data === "string" ? event.data : ""; @@ -397,6 +435,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { ws.addEventListener("message", onMessage); ws.addEventListener("close", onClose); ws.addEventListener("error", onError); + ws.addEventListener("pong", onPong); if (signal?.aborted) onAbort(); else if (session.opened) onOpen(); }); diff --git a/src/server/responses/codex-ws-wire.ts b/src/server/responses/codex-ws-wire.ts index 64b732b22e..6dc3a45c3b 100644 --- a/src/server/responses/codex-ws-wire.ts +++ b/src/server/responses/codex-ws-wire.ts @@ -7,6 +7,17 @@ import { // If the 101 never arrives (network black hole), give SSE a chance well before // the caller's connect timeout (default 200s) would fire. export const UPGRADE_DEADLINE_MS = 10_000; +// Liveness while the exchange waits for its first response event. The proxy cannot know +// how long the origin legitimately needs before `response.created` (a multi-image replay +// spends that time in prefill; #4083 measured 30 s), and it is not the party that owns +// the slowness policy — the client holds its own deadline and the operator holds +// `connectTimeoutMs`. What the proxy can decide is whether the peer is still there, and +// WebSocket has a native answer: ping. While waiting, the exchange pings every +// CODEX_WS_LIVENESS_PING_INTERVAL_MS on sockets that expose `ping()`; any inbound frame +// or pong resets the silence clock, and only CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS of +// nothing at all settles the exchange as an origin-silence 504. A peer that never pongs +// 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; // Keep the push-based WS transport inside the same memory envelope as the // bounded SSE relays that consume this response. Unlike fetch response bodies, @@ -112,6 +123,10 @@ export type CodexWsFailureStage = { firstFrameMs: 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. */ + pings: number; + /** Pongs the peer answered with; zero on a peer that never answers pings. */ + pongs: number; }; /** @@ -147,7 +162,8 @@ export function codexWsFailureDetail(stage: CodexWsFailureStage): string { 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)}]`; + + ` first-frame=${duration(stage.firstFrameMs)} elapsed=${duration(stage.elapsedMs)}` + + ` pings=${stage.pings} pongs=${stage.pongs}]`; } export type ResponsesWsRelayEvent = { diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index 87b3767d2b..b6799267d6 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -20,7 +20,7 @@ import { codexWsExchange } from "./codex-ws-exchange"; import { CodexWsSession } from "./codex-ws-session"; import { codexWsPool, codexWsReuseIdentity } from "./codex-ws-pool"; import { codexWsCreateFrameExceedsLimit } from "./codex-ws-wire"; -export { CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, +export { CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, MAX_CODEX_WS_CREATE_FRAME_BYTES, CODEX_WS_CREATE_FRAME_LIMIT_BYTES, codexWsCreateFrameExceedsLimit, isCodexWsQuotaObservedResponse, isCodexWsUpstreamResponse } from "./codex-ws-wire"; export const MIN_BOUNDED_CODEX_WS_BUN_VERSION = "1.4.0"; diff --git a/tests/responses/ws-failure-stage.test.ts b/tests/responses/ws-failure-stage.test.ts index d578d801d4..04c12c6357 100644 --- a/tests/responses/ws-failure-stage.test.ts +++ b/tests/responses/ws-failure-stage.test.ts @@ -86,6 +86,8 @@ function stage(overrides: Partial = {}): CodexWsFailureStag relayedEvents: 0, firstFrameMs: null, elapsedMs: 90_003, + pings: 0, + pongs: 0, ...overrides, }; } @@ -150,12 +152,14 @@ describe("codex WS failure classification", () => { 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]", + + " first-frame=41ms elapsed=90003ms pings=0 pongs=0]", ); 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]", + + " first-frame=n/a elapsed=n/a pings=0 pongs=0]", ); + // A peer that answered pings but never started a response is named as such. + expect(codexWsFailureDetail(stage({ upstreamFrames: 0, pings: 6, pongs: 6 }))).toContain(" pings=6 pongs=6]"); }); }); diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index 6afb9e312c..9151fba103 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -20,6 +20,7 @@ import { MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, + CODEX_WS_LIVENESS_PING_INTERVAL_MS, shouldUseCodexWsUpstream as rawShouldUseCodexWsUpstream, } from "../../src/server/responses/ws-upstream"; import type { OcxProviderConfig } from "../../src/types"; @@ -1304,6 +1305,106 @@ describe("codexWsUpstreamFetch", () => { } }); + describe("prelude liveness", () => { + // The 90 s bound is a silence bound, not a deadline: a peer that answers pings is alive, + // and how long an alive origin may take before response.created belongs to the client's + // own deadline and the operator's connectTimeoutMs, not to a fixed number in the proxy. + const noResend = (counter: { http: number }) => (async () => { + counter.http++; + return new Response("must not resend"); + }) as typeof fetch; + + test("a peer that answers pings stays alive past the silence bound and still completes with one send", async () => { + jest.useFakeTimers(); + const opened = Promise.withResolvers(); + let pingsSent = 0; + installFake(ws => { + Object.assign(ws, { ping: () => { pingsSent++; ws.emit("pong", {}); } }); + ws.emit("open", {}); + opened.resolve(); + }); + const counter = { http: 0 }; + try { + 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. + for (let step = 0; step < 7; step++) jest.advanceTimersByTime(CODEX_WS_LIVENESS_PING_INTERVAL_MS); + expect(pingsSent).toBe(7); + 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; + expect(response.status).toBe(200); + 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. + jest.advanceTimersByTime(CODEX_WS_LIVENESS_PING_INTERVAL_MS * 4); + expect(pingsSent).toBe(7); + expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true); + } finally { + jest.useRealTimers(); + } + }); + + test("a peer that never pongs keeps the previous 90 s bound and names the unanswered pings", async () => { + jest.useFakeTimers(); + const opened = Promise.withResolvers(); + let pingsSent = 0; + installFake(ws => { + Object.assign(ws, { ping: () => { pingsSent++; } }); + ws.emit("open", {}); + opened.resolve(); + }); + const counter = { http: 0 }; + try { + const pending = codexWsUpstreamFetch(CODEX_URL, streamingInit(), noResend(counter)); + await opened.promise; + jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); + const response = await pending; + expect(response.status).toBe(504); + const failure = ((await response.json()) as { error: { code: string; message: string } }).error; + expect(failure.code).toBe("upstream_no_response"); + expect(failure.message).toContain("prelude timed out"); + expect(failure.message).toMatch(/pings=[56] pongs=0\]/); + expect(pingsSent).toBeGreaterThanOrEqual(5); + expect(FakeWebSocket.instances[0].sent).toHaveLength(1); + expect(counter.http).toBe(0); + expect(FakeWebSocket.instances[0].closed).toBe(true); + } finally { + jest.useRealTimers(); + } + }); + + test("a control frame is proof of life too: quota keeps the exchange waiting", async () => { + jest.useFakeTimers(); + const opened = Promise.withResolvers(); + installFake(ws => { ws.emit("open", {}); opened.resolve(); }); + const counter = { http: 0 }; + try { + const pending = codexWsUpstreamFetch(CODEX_URL, streamingInit(), noResend(counter)); + await opened.promise; + const ws = FakeWebSocket.instances[0]; + // No ping() on this socket. Quota at 60 s and 120 s resets the silence clock each time. + jest.advanceTimersByTime(60_000); + ws.emit("message", { data: JSON.stringify({ type: "codex.rate_limits", rate_limits: { primary: { used_percent: 10, window_minutes: 10080 } } }) }); + jest.advanceTimersByTime(60_000); + ws.emit("message", { data: JSON.stringify({ type: "codex.rate_limits", rate_limits: { primary: { used_percent: 11, window_minutes: 10080 } } }) }); + jest.advanceTimersByTime(60_000); + 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; + expect(response.status).toBe(200); + expect(response.headers.get("x-codex-primary-used-percent")).toBe("11"); + expect(await response.text()).toContain("event: response.completed"); + expect(ws.sent).toHaveLength(1); + expect(counter.http).toBe(0); + } finally { + jest.useRealTimers(); + } + }); + }); + test("malformed native WS metadata still normalizes the real HTTP fallback routing hint", async () => { let fallbackInit: RequestInit | undefined; const body = JSON.stringify({ model: "gpt-6-astra", service_tier: "priority", stream: true, client_metadata: [] }); From 9e9dd5315f501723ac9f858b52f73b1e100cb568 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 12:50:16 +0900 Subject: [PATCH 07/10] docs(devlog): wp4 plan for review and final-head CI --- .../_plan/260911_ws_commit_boundary/045_wp4_plan.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 devlog/_plan/260911_ws_commit_boundary/045_wp4_plan.md diff --git a/devlog/_plan/260911_ws_commit_boundary/045_wp4_plan.md b/devlog/_plan/260911_ws_commit_boundary/045_wp4_plan.md new file mode 100644 index 0000000000..02ec4a0393 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/045_wp4_plan.md @@ -0,0 +1,13 @@ +# wp4 — PR, read-only diff review, final-head CI + +Previous D (wp3): liveness landed at 075b9f39f5 and is pushed to draft PR #4256. Direction unchanged. + +## Steps +- xai/grok-4.6 read-only review of the full diff origin/dev...HEAD (t2); findings fixed or dispositioned in 050_review.md. +- Mark PR #4256 ready for review with the final description (t1). +- Every newest-per-workflow pull_request run on the final head SHA success (t3); record run ids in 060_ci_evidence.md. +- Merge is not part of this goal (owner said "PR까지"); report readiness. + +## Verification +NOT RUN locally. Remote CI only. + From a00ef49af70ef4f8bd5b4f6cc58c6b42d1f110bd Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 13:03:53 +0900 Subject: [PATCH 08/10] fix(codex-ws): capture ping() before the liveness timer closure (TS2722) --- src/server/responses/codex-ws-exchange.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index f682e3d73d..2119063ce4 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -208,11 +208,11 @@ export function codexWsExchange(options: ExchangeOptions): Promise { }; /** Ping on a fixed interval until the response starts; a socket without ping() is never pinged. */ const schedulePing = () => { - const socket = ws as WebSocket & { ping?: (data?: string) => void }; - if (typeof socket.ping !== "function" || responseCommitted || terminal) return; + const ping = (ws as WebSocket & { ping?: (data?: string) => void }).ping; + if (typeof ping !== "function" || responseCommitted || terminal) return; pingTimer = setTimeout(() => { if (responseCommitted || terminal) return; - try { socket.ping(); } catch { return; } + try { ping.call(ws); } catch { return; } pings += 1; // ping() may close the socket synchronously and settle the exchange; re-check. if (!terminal) schedulePing(); From 76527004445393267b9ad718dc09dd93d6a31f87 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 13:12:33 +0900 Subject: [PATCH 09/10] fix(codex-ws): keep commitResponse for the body-error path; step the never-pongs oracle Review finding: guarding commitResponse on terminal left an exchange without a metadata channel unsettled when failStream ran after send, because that path relies on commitResponse to hand the client its 200 before erroring the body. The pre-response JSON settle now claims the commit slot itself instead. The never-pongs test steps the fake clock per ping interval. --- .../260911_ws_commit_boundary/050_review.md | 27 +++++++++++++++++++ src/server/responses/codex-ws-exchange.ts | 5 ++-- tests/responses/ws-upstream.test.ts | 5 +++- 3 files changed, 34 insertions(+), 3 deletions(-) create mode 100644 devlog/_plan/260911_ws_commit_boundary/050_review.md diff --git a/devlog/_plan/260911_ws_commit_boundary/050_review.md b/devlog/_plan/260911_ws_commit_boundary/050_review.md new file mode 100644 index 0000000000..6ac5ddea26 --- /dev/null +++ b/devlog/_plan/260911_ws_commit_boundary/050_review.md @@ -0,0 +1,27 @@ +# wp4 review — xai/grok-4.6 read-only diff review (agent 01a08e96) + +Verdict received: FAIL with two majors and one minor. + +| # | severity | finding | disposition | +|---|---|---|---| +| 1 | major | commitResponse bailing on terminal leaves the non-metadata body-error path unsettled (failStream sets terminal first, then calls commitResponse, then controller.error with no resolve). | ACCEPTED, real bug. commitResponse guards only responseCommitted again; the JSON settle path sets responseCommitted = true before resolving so a second 200 can never be committed. | +| 2 | major | bun-types WebSocketEventMap lists only close/error/message/open, so a client pong event may never reach onPong; ping-alive would be harness-only. | REBUTTED with a runtime probe. Bun 1.4.0 (the minimum version the bounded relay gate accepts) was probed on 2026-09-11 with a local Bun.serve websocket and a client new WebSocket: ws.ping is a function, ws.pong is a function, and addEventListener("pong") fired with the ping payload (seen: open, pong:x, message:ack). The type map is incomplete; the runtime dispatches the event. The exchange still feature-detects ping() and degrades to the message-only 90 s bound where no pong arrives, which is exactly the never-pongs oracle. Probe script kept below. | +| 3 | minor | The never-pongs oracle used one 90 s jump and relied on recursive fake-timer scheduling. | ACCEPTED. The test now steps 15 s at a time like its sibling. | + +Findings 4-6 were confirmations (harness oracles, TypeScript after a00ef49af7, privacy of the JSON body). + +## Probe (not product code, run once in /tmp) + +```js +const srv = Bun.serve({ port: 0, fetch(req, s){ if (s.upgrade(req)) return; return new Response("no"); }, + websocket: { open(ws){}, message(ws,m){ if (m==="hi") ws.send("ack"); }, ping(ws,data){ }, pong(ws,data){ } } }); +const ws = new WebSocket("ws://127.0.0.1:"+srv.port); +const seen = []; +for (const ev of ["open","message","close","error","ping","pong"]) ws.addEventListener(ev, e => seen.push(ev + (e.data!==undefined? ":"+String(e.data):""))); +await new Promise(r => ws.addEventListener("open", r, {once:true})); +ws.ping("x"); ws.send("hi"); await new Promise(r => setTimeout(r, 400)); +console.log(JSON.stringify({ bun: Bun.version, ping: typeof ws.ping, pong: typeof ws.pong, seen })); +``` + +Output: {"bun":"1.4.0","ping":"function","pong":"function","seen":["open","pong:x","message:ack"]} + diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 2119063ce4..f65a7ea538 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -158,8 +158,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { }); const commitResponse = () => { - // A pre-response settlement (gateway status) has already resolved this exchange. - if (responseCommitted || terminal) return; + if (responseCommitted) return; responseCommitted = true; clearTimeout(silenceTimer); clearTimeout(pingTimer); @@ -181,6 +180,8 @@ export function codexWsExchange(options: ExchangeOptions): Promise { // it again, and the client applies its own retry policy as it would on the direct // path. Same settle order as a refused create: snapshot, detach, close, dispose. const prelude = metadata.snapshot(); + // Claim the commit slot so no later path can resolve a second, 200 Response. + responseCommitted = true; cleanup(); try { controller?.close(); } catch { /* unused stream already closed */ } session.dispose(); diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index 9151fba103..80c6291780 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -1360,7 +1360,10 @@ describe("codexWsUpstreamFetch", () => { try { const pending = codexWsUpstreamFetch(CODEX_URL, streamingInit(), noResend(counter)); await opened.promise; - jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS); + // Step the clock so each chained ping timer is scheduled and fired in turn. + for (let elapsed = 0; elapsed < CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS; elapsed += CODEX_WS_LIVENESS_PING_INTERVAL_MS) { + jest.advanceTimersByTime(CODEX_WS_LIVENESS_PING_INTERVAL_MS); + } const response = await pending; expect(response.status).toBe(504); const failure = ((await response.json()) as { error: { code: string; message: string } }).error; From 361200e79b9d48ab7d6e685cfbe10b963f91e0aa Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 11 Sep 2026 13:27:34 +0900 Subject: [PATCH 10/10] docs(codex-ws): qualify ping by socket capability; explain the TimeoutError discriminator --- .../src/content/docs/reference/configuration/server.md | 7 ++++--- src/server/responses/codex-ws-exchange.ts | 5 +++++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index b5c6158090..3b61d51739 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -36,9 +36,10 @@ runs helper features around provider requests. | `images?` | `OcxImagesConfig` | automatic OpenAI selection | Standalone Images relay options for Codex `image_gen`. | While the canonical ChatGPT upstream WebSocket waits for the first Responses event after -sending the create frame, it watches for liveness rather than a fixed deadline. The proxy pings -the socket every 15 seconds; any inbound frame — quota, response metadata, or a pong — resets a -90-second silence clock, and only 90 seconds with nothing at all settles the request as an +sending the create frame, it watches for liveness rather than a fixed deadline. When the socket +supports protocol pings, the proxy pings it every 15 seconds. Any inbound frame — quota, +response metadata, or a pong — resets a 90-second silence clock, so a socket without ping +support still stays alive on its own frames, and only 90 seconds with nothing at all settles the request as an HTTP 504 with an `upstream_no_response` error. A slow but alive origin therefore waits for the client's own deadline or for `connectTimeoutMs` (default 200s), whichever comes first; a connect timeout that fires after the create frame was sent settles as the same 504. A socket diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index f65a7ea538..698eb93340 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -247,6 +247,11 @@ export function codexWsExchange(options: ExchangeOptions): Promise { // Sent, unacknowledged. The proxy's own connect deadline is an origin-silence // verdict and settles like one; a caller abort is the caller's decision, so the // exchange rejects with that reason and disposing the socket cancels the turn. + // Both arrive on the same composite signal (fetchWithHeaderTimeout joins the + // caller's controller with its own), so the reason is the only discriminator: the + // deadline aborts with a TimeoutError DOMException, and every caller abort in this + // process (upstream.abort() in core.ts) carries the default AbortError. A future + // proxy-side deadline that aborts with TimeoutError would still be honestly a 504. if ((reason as { name?: unknown } | null)?.name === "TimeoutError") { failStream(`codex websocket response did not start before the connect deadline${codexWsFailureDetail(failureStage())}`, 504); return;