From ef4f935cde73f3442a53274965b2da7589b59870 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C5=A9ng=20Ho=C3=A0ng=20=C4=90=E1=BB=A9c?= Date: Fri, 17 Jul 2026 16:47:41 +0700 Subject: [PATCH 1/2] fix(opencode): stabilize Codex connections --- ...2026-07-17-codex-connection-reliability.md | 364 ++++++++++++++++++ ...-17-codex-connection-reliability-design.md | 202 ++++++++++ .../opencode/src/plugin/openai/codex-http.ts | 216 +++++++++++ packages/opencode/src/plugin/openai/codex.ts | 205 ++++++---- .../opencode/test/plugin/codex-http.test.ts | 319 +++++++++++++++ packages/opencode/test/plugin/codex.test.ts | 260 +++++++++++++ 6 files changed, 1498 insertions(+), 68 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-17-codex-connection-reliability.md create mode 100644 docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md create mode 100644 packages/opencode/src/plugin/openai/codex-http.ts create mode 100644 packages/opencode/test/plugin/codex-http.test.ts diff --git a/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md b/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md new file mode 100644 index 000000000000..a70bfe517938 --- /dev/null +++ b/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md @@ -0,0 +1,364 @@ +# Codex Connection Reliability Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add bounded pre-output retries, Codex-specific 60/360-second timeouts, and one-time OAuth recovery without changing terminal or session semantics. + +**Architecture:** Add a focused HTTP transport at the ChatGPT Codex OAuth fetch boundary. It owns per-attempt header timeout, status/network retry, first-SSE-event inspection, raw-byte replay, and inter-chunk timeout. `codex.ts` continues to own OAuth and WebSocket setup; existing parser/session retry remains authoritative after output begins. + +**Tech Stack:** TypeScript, Bun, Fetch/Streams APIs, OpenAI Responses SSE, `ws`, Bun test. + +--- + +## Files + +- Create `packages/opencode/src/plugin/openai/codex-http.ts` — HTTP timeout/retry and first-event gate. +- Create `packages/opencode/test/plugin/codex-http.test.ts` — isolated transport tests. +- Modify `packages/opencode/src/plugin/openai/codex.ts` — integration, OAuth recovery, WebSocket defaults. +- Modify `packages/opencode/test/plugin/codex.test.ts` — OAuth and integration tests. +- Modify `packages/opencode/test/plugin/openai-ws.test.ts` — 60/360 WebSocket option coverage. +- Create `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` — requested Vietnamese analysis. + +## Invariants + +1. Never retry after exposing an SSE event. +2. Never rotate accounts or endpoints. +3. Never synthesize terminal events or extra `[DONE]` frames. +4. Preserve non-2xx status, body, and headers. +5. Abort stops reads, backoff, refresh, and later attempts. +6. Accepted bytes and tool calls are emitted exactly once. + +--- + +### Task 1: Codex status/network retry transport + +**Files:** +- Create: `packages/opencode/src/plugin/openai/codex-http.ts` +- Create: `packages/opencode/test/plugin/codex-http.test.ts` + +- [ ] **Step 1: Write failing retry-policy tests** + +Create table tests for 502 (3 retries, 3000 ms), 503 (3, 2000 ms), 504 (2, 3000 ms), and network failure using injected `fetch` and `sleep` functions: + +```ts +test.each([ + [502, 3, 3_000], + [503, 3, 2_000], + [504, 2, 3_000], +] as const)("retries %d before exposing a response", async (status, retries, delay) => { + let calls = 0 + const waits: number[] = [] + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => ++calls <= retries ? new Response("retry", { status }) : sse(CREATED), + sleep: async (ms) => void waits.push(ms), + }) + expect(calls).toBe(retries + 1) + expect(waits).toEqual(Array(retries).fill(delay)) + expect(await response.text()).toBe(CREATED) +}) +``` + +- [ ] **Step 2: Run RED test** + +From `packages/opencode`: + +```bash +bun test test/plugin/codex-http.test.ts +``` + +Expected: FAIL because the module does not exist. + +- [ ] **Step 3: Implement retry policy and injectable options** + +Use this public shape: + +```ts +export const CODEX_HEADER_TIMEOUT = 60_000 +export const CODEX_CHUNK_TIMEOUT = 360_000 + +const POLICY = { + 502: { attempts: 3, delay: 3_000 }, + 503: { attempts: 3, delay: 2_000 }, + 504: { attempts: 2, delay: 3_000 }, +} as const + +export interface CodexHTTPOptions { + fetch?: typeof globalThis.fetch + sleep?: (ms: number, signal?: AbortSignal | null) => Promise + headerTimeout?: number + chunkTimeout?: number +} + +export async function fetchCodexHTTP( + input: RequestInfo | URL, + init: RequestInit = {}, + options: CodexHTTPOptions = {}, +): Promise +``` + +Track independent counters for network/connect failures, HTTP 502, HTTP 503, HTTP 504, `server_is_overloaded`, and `service_unavailable_error`. Network/header-timeout failures use the same numeric policy as HTTP 502 but do not consume the HTTP 502 counter. The two SSE codes use the same numeric policy as HTTP 503 but do not consume the HTTP 503 counter or each other's counter. Cancel failed response bodies before retry. `abortableSleep` must reject immediately when `RequestInit.signal` aborts; this signal is the public cancellation contract. + +- [ ] **Step 4: Implement per-attempt 60-second header timeout** + +`fetchAttempt` must combine caller and timeout signals with `AbortSignal.any`, clear its timer in `finally`, and distinguish caller abort from timeout. Do not use a total timeout covering all retries. + +- [ ] **Step 5: Run GREEN test and inspect diff** + +```bash +bun test test/plugin/codex-http.test.ts +git diff -- packages/opencode/src/plugin/openai/codex-http.ts packages/opencode/test/plugin/codex-http.test.ts +``` + +Expected: tests PASS; only the new module and tests appear. Do not commit unless requested. + +--- + +### Task 2: First-SSE-event gate and 360-second chunk timeout + +**Files:** +- Modify: `packages/opencode/src/plugin/openai/codex-http.ts` +- Modify: `packages/opencode/test/plugin/codex-http.test.ts` + +- [ ] **Step 1: Write failing framing and safety tests** + +Add separate cases for: + +- LF and CRLF event boundaries. +- Boundary split across chunks. +- UTF-8 code point split across chunks. +- Exact byte replay once. +- Immediate return after accepting the first event. +- Structured first-event codes `server_is_overloaded` and `service_unavailable_error` each use an independent counter with the 503 numeric policy. +- The same text inside `response.output_text.delta` does not retry. +- Retry exhaustion exposes one final upstream error stream. +- Abort while reading or sleeping starts no later attempt. +- First event larger than 64 KiB fails safely. + +Core assertion: + +```ts +expect(calls).toBe(2) +expect(await response.text()).toBe(successBytes) +expect(exposedOverloadBytes).toBe(0) +``` + +- [ ] **Step 2: Run RED tests** + +```bash +bun test test/plugin/codex-http.test.ts +``` + +- [ ] **Step 3: Implement byte-safe event detection** + +Scan raw bytes for `[10, 10]` and `[13, 10, 13, 10]`; do not search a decoded string. Buffer at most 64 KiB. Parse only `data:` lines from the first complete event and retry only when parsed JSON has `type: "error"` with an approved structured code. + +Use unchanged buffered chunks when rebuilding the accepted stream: + +```ts +for (const chunk of buffered) controller.enqueue(chunk) +while (true) { + const part = await readWithTimeout(reader, chunkTimeout, signal) + if (part.done) break + controller.enqueue(part.value) +} +controller.close() +``` + +- [ ] **Step 4: Add raw-chunk timeout** + +Wrap every `reader.read()` after headers, including the first event, with a fresh 360-second timer. Each received chunk resets the budget. On timeout, cancel the reader and throw `ProviderError.ResponseStreamError("Codex SSE read timed out")`. + +- [ ] **Step 5: Preserve response semantics** + +Return non-2xx responses unchanged after status retry is exhausted. Do not inspect 401/403/429 as SSE. For malformed 2xx non-SSE responses, return a status-200 `Response` whose body is a `ReadableStream` that fails with `ProviderError.ResponseStreamError`; include content type and a bounded body preview in the error. This preserves the declared `Promise` contract while surfacing a typed stream failure. + +- [ ] **Step 6: Run focused suite** + +```bash +bun test test/plugin/codex-http.test.ts +``` + +Expected: all retry, framing, timeout, replay, and abort cases PASS. + +--- + +### Task 3: Install Codex-specific 60/360 defaults + +**Files:** +- Modify: `packages/opencode/src/plugin/openai/codex.ts:101-105,263-266,320-425` +- Modify: `packages/opencode/test/plugin/codex.test.ts` +- Modify: `packages/opencode/test/plugin/openai-ws.test.ts` + +- [ ] **Step 1: Write failing option-plumbing tests** + +Extend plugin options with injectable test values and assert the OAuth loader disables the generic OpenAI timeout wrapper: + +```ts +expect(loaded.headerTimeout).toBe(false) +expect(loaded.chunkTimeout).toBe(false) +``` + +Assert production WebSocket defaults are connect `60_000` and idle/send `360_000`. + +- [ ] **Step 2: Run RED tests** + +```bash +bun test test/plugin/codex.test.ts test/plugin/openai-ws.test.ts +``` + +- [ ] **Step 3: Extend plugin options** + +```ts +interface CodexAuthPluginOptions { + issuer?: string + codexApiEndpoint?: string + experimentalWebSockets?: boolean + httpHeaderTimeout?: number + httpChunkTimeout?: number + websocketConnectTimeout?: number + websocketIdleTimeout?: number +} +``` + +- [ ] **Step 4: Configure WebSocket and HTTP paths** + +Create the pool with Codex defaults: + +```ts +OpenAIWebSocketPool.createWebSocketFetch({ + httpFetch: fetch, + connectTimeout: options.websocketConnectTimeout ?? CODEX_HEADER_TIMEOUT, + idleTimeout: options.websocketIdleTimeout ?? CODEX_CHUNK_TIMEOUT, +}) +``` + +Replace the direct HTTP call with `fetchCodexHTTP` using the HTTP option values. Return `headerTimeout: false` and `chunkTimeout: false` from the OAuth loader so the generic 10-second OpenAI header timer cannot terminate the whole retry sequence. API-key OpenAI behavior stays unchanged. + +- [ ] **Step 5: Run integration tests and typecheck** + +```bash +bun test test/plugin/codex-http.test.ts test/plugin/codex.test.ts test/plugin/openai-ws.test.ts test/provider/header-timeout.test.ts +bun typecheck +``` + +Expected: PASS; typecheck exits 0. + +--- + +### Task 4: One-time 401 refresh and safe token rotation + +**Files:** +- Modify: `packages/opencode/src/plugin/openai/codex.ts:94-99,333-425` +- Modify: `packages/opencode/test/plugin/codex.test.ts:195-290` + +- [ ] **Step 1: Write failing OAuth tests** + +Add tests proving: + +- Missing/empty `refresh_token` preserves `refresh-old`. +- Concurrent 401 responses cause exactly one token refresh. +- Reissued requests use the new access/account ID and byte-identical request bodies. +- A second 401 is returned without a third request. +- Abort after the first 401 prevents refresh/reissue. +- 403 and 429 remain unchanged. + +- [ ] **Step 2: Run RED OAuth tests** + +```bash +bun test test/plugin/codex.test.ts +``` + +- [ ] **Step 3: Make refresh rotation optional** + +Change `TokenResponse.refresh_token` to optional for refresh responses and persist: + +```ts +refresh: tokens.refresh_token?.trim() || currentAuth.refresh +``` + +Initial authorization must still require a usable refresh token. + +- [ ] **Step 4: Refactor one single-flight refresh owner** + +The refresh helper accepts the access token that failed. Before starting refresh, reload auth; if another caller already installed a different access token, reuse it. Otherwise share `refreshPromise`. This prevents concurrent 401 callers from refreshing twice. + +- [ ] **Step 5: Reissue exactly once** + +```ts +const first = await send(currentAccess) +if (first.status !== 401 || init?.signal?.aborted) return first +await first.body?.cancel() +const refreshed = await refresh(currentAccess) +return send(refreshed.access, refreshed.accountId) +``` + +Do not use recursive calls and do not refresh on 403/429. + +- [ ] **Step 6: Run OAuth and transport tests** + +```bash +bun test test/plugin/codex.test.ts test/plugin/codex-http.test.ts +``` + +Expected: all tests PASS; concurrent 401 calls share one refresh. + +--- + +### Task 5: Write Vietnamese 9Router analysis + +**Files:** +- Create: `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` + +- [ ] **Step 1: Write verified analysis** + +Cover request lifecycle, OAuth/client ID, refresh behavior, account fallback, exact timeout comparison, retry budgets, SSE buffering, terminal handling, lack of true resume, adopted mechanisms, rejected gateway-only mechanisms, changed files, and verification evidence. + +- [ ] **Step 2: Scan placeholders and secrets** + +```bash +rg -n "TBD|TODO|Bearer |sk-|refresh[_-]?token" "/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md" +``` + +Expected: no placeholders or real credentials. Field names may be discussed without values. + +--- + +### Task 6: Final verification and review + +**Files:** +- Review every file changed by Tasks 1-5. + +- [ ] **Step 1: Run focused OpenCode tests** + +From `packages/opencode`: + +```bash +bun test test/plugin/codex-http.test.ts test/plugin/codex.test.ts test/plugin/openai-ws.test.ts test/provider/header-timeout.test.ts +bun typecheck +``` + +Expected: all tests PASS and typecheck exits 0. + +- [ ] **Step 2: Inspect complete diff** + +```bash +git status --short +git diff --check +git diff -- packages/opencode/src/plugin/openai packages/opencode/test/plugin packages/opencode/test/provider docs/superpowers +git diff --no-index /dev/null docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md +git diff --no-index /dev/null docs/superpowers/plans/2026-07-17-codex-connection-reliability.md +``` + +Expected: no whitespace errors, secrets, account/endpoint fallback, or terminal-protocol changes. The two `--no-index` commands may exit 1 because they intentionally display new untracked files; inspect their content rather than treating exit 1 as a failure. Inspect `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` separately because it is outside the repository. + +- [ ] **Step 3: Smoke-test connection stability** + +Use the already-working OpenAI OAuth model selection. Run one normal response and one tool call. Confirm both complete once without connection interruption, duplicated text, or duplicated tool execution. Do not change or retest model lowering/service-tier behavior. + +- [ ] **Step 4: Independent review** + +Review for retry after exposed output, timer/body leaks, lost HTTP semantics, duplicate tool events, API-key timeout regressions, and terminal behavior changes. + +- [ ] **Step 5: Report evidence** + +Report changed files, exact tests/typechecks, smoke-test result, and limitations. Do not commit or push unless explicitly requested. diff --git a/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md b/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md new file mode 100644 index 000000000000..ee7ba5c07649 --- /dev/null +++ b/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md @@ -0,0 +1,202 @@ +# Codex Connection Reliability Design + +**Date:** 2026-07-17 + +**Target:** `packages/opencode/src/plugin/openai` in `original-opencode` + +**Reference:** `/home/dunghd/DEVELOPMENT/Research/9router-master` + +## Objective + +Improve ChatGPT Codex OAuth connection reliability without changing OpenCode's existing Responses protocol, terminal-event behavior, session semantics, account selection, or endpoint routing. + +Transparent transport retries are allowed only before any upstream event is exposed to the existing parser/session layer. After an event is exposed, the existing OpenCode stream and session retry behavior remains authoritative. + +## Included behavior + +- HTTP response-header timeout: 60 seconds per attempt. +- HTTP inter-chunk stall timeout: 360 seconds, reset for each raw chunk. +- WebSocket connect timeout: 60 seconds. +- WebSocket idle/send timeout: 360 seconds. +- Pre-output retries for network/connect failures and HTTP 502, 503, and 504. +- Pre-output retry when the first parsed SSE event is a structured Codex overload error. +- One forced OAuth refresh and request reissue after HTTP 401. +- Preserve the existing refresh token when a refresh response omits a replacement. +- Abort-aware timeout and retry backoff. +- Focused tests for timing, SSE framing, cancellation, error preservation, terminal behavior, and tool-call uniqueness. + +## Excluded behavior + +- Account rotation, account cooldown, or quota-based account selection. +- Alternative endpoint or base-URL fallback. +- Model-combo fallback. +- Proxy-style request translation or broad body mutation. +- Synthetic `response.failed`, response IDs, or additional `[DONE]` frames. +- Changes to existing terminal-event handling. +- Transparent retry after any event has been exposed. +- Generic timeout changes for normal OpenAI API-key traffic. + +## Reliability findings from 9Router + +9Router's continuity comes from layered recovery: + +1. It allows 60 seconds for upstream response headers. +2. It allows 360 seconds between raw stream chunks and resets the watchdog on upstream activity. +3. It retries transient network and 502/503/504 failures before returning a response downstream. +4. It detects Codex overload failures hidden in an HTTP 200 SSE response and retries at the beginning of the stream. +5. It supports proactive and reactive credential refresh. +6. It also uses account and endpoint failover, which are gateway responsibilities and are excluded from this OpenCode change. + +9Router does not resume a stream after partial output. OpenCode will preserve the same safety boundary: retry before output, fail normally after output. + +## Existing OpenCode behavior to preserve + +- OpenAI provider HTTP header timeout currently defaults to 10 seconds. +- Codex WebSocket connection timeout currently defaults to 15 seconds. +- Codex WebSocket idle timeout currently defaults to 300 seconds. +- The WebSocket pool already owns session affinity, connection aging, busy-lane HTTP fallback, cancellation, cleanup, and sticky HTTP fallback after repeated socket failures. +- The Responses parser and WebSocket bridge already handle `response.completed`, `response.done`, `response.failed`, `response.incomplete`, and `[DONE]`. +- Session retry already handles retryable failures after transport errors reach the session layer. + +The new timeouts apply only to ChatGPT OAuth/Codex traffic. They must not change ordinary OpenAI API-key behavior. + +## Transport architecture + +The Codex OAuth fetch boundary remains the integration point: + +```text +request + -> OAuth refresh/header construction + -> Codex reliability transport + -> connect/header timeout + -> status/network retry + -> first-SSE-event gate + -> inter-chunk watchdog + -> existing Responses parser + -> existing session processing +``` + +The reliability transport must not own semantic parsing beyond identifying the first complete SSE event and a structured Codex error code. It must return accepted bytes unchanged. + +## Retry policy + +Retry counts are retries after the initial attempt. + +| Failure | Retries | Delay | +|---|---:|---:| +| Network/connect failure | 3 | 3 seconds | +| HTTP 502 | 3 | 3 seconds | +| HTTP 503 | 3 | 2 seconds | +| HTTP 504 | 2 | 3 seconds | +| First SSE event: `server_is_overloaded` | 3 | 2 seconds | +| First SSE event: `service_unavailable_error` | 3 | 2 seconds | +| HTTP 401 | One forced refresh and one reissue | No delay | +| HTTP 403 | No transport retry | Preserve response | +| HTTP 429 | No transport retry | Preserve response and `Retry-After` | + +Each status has an independent retry budget. Backoff must be cancellable; an aborted request must not start another attempt. + +The transport must not retry when: + +- a valid SSE event has already been returned to the caller; +- the overload text occurs in ordinary model output rather than a structured error event; +- the client has aborted; +- the relevant retry budget is exhausted. + +## SSE first-event gate + +The first-event gate must: + +- support LF and CRLF SSE separators; +- support separators split across chunks; +- preserve raw bytes exactly; +- handle UTF-8 code points split across chunks; +- inspect only the first complete SSE event; +- reject a first event larger than 64 KiB as a bounded protocol error; +- recognize retryable codes only in structured error events; +- return immediately after accepting the first event instead of buffering the rest of the stream; +- expose only the final accepted attempt; +- preserve the final upstream error event when retries are exhausted. + +Non-SSE and non-2xx responses retain their original status, body, and headers. A malformed 2xx streaming response is represented as a status-200 `Response` whose body stream fails with a typed stream error. Diagnostics must include its content type and a bounded body preview. Non-2xx responses are never rewritten this way. + +## Timeout policy + +### HTTP + +- Header timeout: 60 seconds for each attempt. +- Inter-chunk timeout: 360 seconds after the response stream begins. +- The inter-chunk timer resets on each raw byte chunk, not translated output. +- Header and chunk timeouts combine with the caller's abort signal. + +### WebSocket + +- Connect timeout: 60 seconds. +- Idle/send timeout: 360 seconds. +- Existing pool retry/fallback and terminal behavior remain unchanged. + +Timeout values are injectable through Codex plugin options for deterministic tests. Production defaults remain 60/360 seconds. + +## OAuth recovery + +The existing loader-local single-flight refresh remains the sole refresh owner. + +- Normal requests refresh when local credentials have expired. +- An HTTP 401 forces one refresh and one request reissue. +- Concurrent callers share one in-flight refresh. +- A second 401 is returned without another refresh. +- A missing, empty, or malformed replacement refresh token must not erase the current valid refresh token. +- The reissued request retains the original method and body while rebuilding authorization/account headers from refreshed credentials. +- Abort before or during refresh prevents reissue. + +HTTP 403 is not automatically refreshed because it can represent a real permission or account-scope failure. + +## Terminal and session behavior + +No terminal-event behavior changes are included. Existing OpenCode handling of completion, failure, incomplete responses, `response.done`, and `[DONE]` remains intact. + +No transport retry occurs after partial output. Such failures continue through the existing stream error and session retry paths. This avoids duplicated text, reasoning, persisted events, and tool execution. + +## Test strategy + +Focused tests must prove: + +1. Network, 502, 503, and 504 retries use the specified budgets and delays. +2. Abort during connect, first-event reading, or backoff prevents later attempts. +3. Codex HTTP and WebSocket defaults are 60/360 seconds through injectable test options. +4. Every raw chunk resets the inter-chunk watchdog. +5. LF, CRLF, split separators, and split UTF-8 are handled correctly. +6. Structured overload first events retry; matching text content does not. +7. An accepted first event is replayed byte-for-byte exactly once. +8. The accepted stream is returned without buffering its remainder. +9. Exhausted retries expose only the final accepted error stream. +10. HTTP 401 refreshes and reissues once; concurrent requests use one refresh. +11. HTTP 403 and 429 preserve status, body, and headers. +12. Omitted refresh-token rotation preserves the previous token. +13. Existing terminal-event tests remain unchanged and pass. +14. Tool calls are emitted and executed only once. + +Run targeted plugin tests and package typecheck from `packages/opencode`, then inspect the final diff. If protocol code is unchanged, the existing protocol suite is a regression check rather than an implementation target. + +## Documentation deliverable + +Write a Vietnamese engineering report under `/home/dunghd/DEVELOPMENT/Research/docs/` containing: + +- 9Router's complete Codex request lifecycle; +- exact 60/360-second timeout behavior; +- status/network/SSE retry layers; +- OAuth refresh and account failover mechanisms; +- stream parsing and terminal handling; +- mechanisms adopted by OpenCode; +- mechanisms deliberately rejected and why; +- changed OpenCode files and verification evidence. + +## Success criteria + +- Slow Codex connection establishment is not aborted at OpenCode's previous 10/15-second thresholds. +- A healthy but temporarily quiet stream may remain idle for up to 360 seconds between raw chunks. +- Transient pre-output network, gateway, and structured overload failures recover transparently within bounded retry budgets. +- Authentication can recover once from server-side token invalidation without losing the refresh token. +- Cancellation remains immediate. +- Existing terminal semantics and session behavior do not change. +- No retry path can duplicate visible output or tool execution. diff --git a/packages/opencode/src/plugin/openai/codex-http.ts b/packages/opencode/src/plugin/openai/codex-http.ts new file mode 100644 index 000000000000..a3b184b1a0d9 --- /dev/null +++ b/packages/opencode/src/plugin/openai/codex-http.ts @@ -0,0 +1,216 @@ +import { ProviderError } from "@/provider/error" + +export const CODEX_HEADER_TIMEOUT = 60_000 +export const CODEX_CHUNK_TIMEOUT = 360_000 +const MAX_EVENT_BYTES = 64 * 1024 +const POLICY = { 502: [3, 3_000], 503: [3, 2_000], 504: [2, 3_000] } as const +const SSE_POLICY = [3, 2_000] as const + +export interface CodexHTTPOptions { + fetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise + sleep?: (ms: number, signal?: AbortSignal | null) => Promise + headerTimeout?: number + chunkTimeout?: number +} + +type Fetcher = (input: RequestInfo | URL, init?: RequestInit) => Promise + +export async function fetchCodexHTTP(input: RequestInfo | URL, init: RequestInit = {}, options: CodexHTTPOptions = {}) { + const fetcher = options.fetch ?? globalThis.fetch + const sleep = options.sleep ?? abortableSleep + const counters = { network: 0, 502: 0, 503: 0, 504: 0, overload: 0, unavailable: 0 } + while (true) { + const result = await fetchAttempt(fetcher, input, init, options.headerTimeout ?? CODEX_HEADER_TIMEOUT) + if (result.kind === "failure") { + if (counters.network === 3) throw result.error + counters.network++ + await sleep(3_000, init.signal ?? undefined) + continue + } + const status = result.response.status + const policy = status === 502 ? POLICY[502] : status === 503 ? POLICY[503] : status === 504 ? POLICY[504] : undefined + const count = status === 502 ? counters[502] : status === 503 ? counters[503] : status === 504 ? counters[504] : 0 + if (policy && count < policy[0]) { + if (status === 502) counters[502]++ + if (status === 503) counters[503]++ + if (status === 504) counters[504]++ + await bestEffortCancel(result.response.body) + await sleep(policy[1], init.signal ?? undefined) + continue + } + if (status < 200 || status >= 300) return result.response + const inspected = await inspect(result.response, options.chunkTimeout ?? CODEX_CHUNK_TIMEOUT, init.signal ?? undefined) + if (inspected.kind === "accepted" || inspected.kind === "malformed") return inspected.response + const key = inspected.code === "server_is_overloaded" ? "overload" : "unavailable" + if (counters[key] === SSE_POLICY[0]) return reconstructed(inspected.response, inspected.buffered, inspected.reader, options.chunkTimeout ?? CODEX_CHUNK_TIMEOUT, init.signal ?? undefined) + counters[key]++ + void bestEffortCancel(inspected.reader) + await sleep(SSE_POLICY[1], init.signal ?? undefined) + } +} + +async function fetchAttempt(fetcher: Fetcher, input: RequestInfo | URL, init: RequestInit, timeout: number) { + const timer = new AbortController() + const signal = init.signal ? AbortSignal.any([init.signal, timer.signal]) : timer.signal + const id = setTimeout(() => timer.abort(new ProviderError.HeaderTimeoutError(timeout)), timeout) + try { + return { kind: "response" as const, response: await fetcher(input, { ...init, signal }) } + } catch (error) { + if (init.signal?.aborted) throw init.signal.reason ?? new DOMException("Aborted", "AbortError") + return { kind: "failure" as const, error: timer.signal.reason instanceof Error ? timer.signal.reason : toError(error) } + } finally { + clearTimeout(id) + } +} + +async function inspect(response: Response, timeout: number, signal?: AbortSignal) { + const contentType = response.headers.get("content-type") ?? "" + if (!contentType.toLowerCase().includes("text/event-stream")) { + let preview = "" + let failure: Error | undefined + try { + if (response.body) preview = await readPreview(response.body, timeout, signal) + } catch (error) { + if (signal?.aborted) throw signal.reason ?? error + failure = toError(error) + } + const headers = new Headers(response.headers) + ;["content-length", "content-encoding", "content-type"].forEach((name) => headers.delete(name)) + const detail = failure ? `preview ${failure.message.toLowerCase()}` : preview + return { kind: "malformed" as const, response: failingResponse(new ProviderError.ResponseStreamError(`Codex response was not SSE (${contentType || "unknown"}): ${detail}`), headers) } + } + if (!response.body) return { kind: "malformed" as const, response: failingResponse(new ProviderError.ResponseStreamError("Codex SSE body was empty"), response.headers) } + const reader = response.body.getReader() + const buffered: Uint8Array[] = [] + let bytes = new Uint8Array() + while (true) { + const part = await readWithTimeout(reader, timeout, signal) + if (part.done) { + void bestEffortCancel(reader) + return { kind: "malformed" as const, response: failingResponse(new ProviderError.ResponseStreamError("Codex SSE ended before its first event"), response.headers) } + } + buffered.push(part.value) + bytes = join([bytes, part.value]) + const end = eventEnd(bytes) + if (end < 0) { + if (bytes.byteLength > MAX_EVENT_BYTES) { + void bestEffortCancel(reader) + return { kind: "malformed" as const, response: failingResponse(new ProviderError.ResponseStreamError("Codex SSE first event exceeded 64 KiB"), response.headers) } + } + continue + } + if (end > MAX_EVENT_BYTES) { + void bestEffortCancel(reader) + return { kind: "malformed" as const, response: failingResponse(new ProviderError.ResponseStreamError("Codex SSE first event exceeded 64 KiB"), response.headers) } + } + const code = firstErrorCode(new TextDecoder().decode(bytes.slice(0, end))) + if (code) return { kind: "retry" as const, response, buffered, reader, code } + return { kind: "accepted" as const, response: new Response(reconstructedStream(buffered, reader, timeout, signal), { status: response.status, headers: response.headers }) } + } +} + +function eventEnd(bytes: Uint8Array) { + for (let i = 1; i < bytes.length; i++) if (bytes[i - 1] === 10 && bytes[i] === 10) return i + 1 + for (let i = 3; i < bytes.length; i++) if (bytes[i - 3] === 13 && bytes[i - 2] === 10 && bytes[i - 1] === 13 && bytes[i] === 10) return i + 1 + return -1 +} + +function firstErrorCode(event: string) { + const data = event.split(/\r?\n/).filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n") + try { + const value: unknown = JSON.parse(data) + if (typeof value !== "object" || value === null || !("type" in value) || value.type !== "error") return undefined + const code = "code" in value ? value.code : "error" in value && typeof value.error === "object" && value.error !== null && "code" in value.error ? value.error.code : undefined + return code === "server_is_overloaded" || code === "service_unavailable_error" ? code : undefined + } catch { + return undefined + } +} + +function reconstructed(response: Response, buffered: Uint8Array[], reader: ReadableStreamDefaultReader, timeout: number, signal?: AbortSignal) { + return new Response(reconstructedStream(buffered, reader, timeout, signal), { status: response.status, headers: response.headers }) +} + +function reconstructedStream(buffered: Uint8Array[], reader: ReadableStreamDefaultReader, timeout: number, signal?: AbortSignal) { + let bufferedPending = true + let finished = false + return new ReadableStream({ + start(controller) { + buffered.forEach((chunk) => controller.enqueue(chunk)) + }, + async pull(controller) { + if (bufferedPending) { + if (controller.desiredSize === 0) return + bufferedPending = false + } + if (finished) return + try { + const part = await readWithTimeout(reader, timeout, signal) + if (part.done) { finished = true; controller.close(); return } + controller.enqueue(part.value) + } catch (error) { + finished = true + controller.error(error) + await bestEffortCancel(reader) + } + }, + cancel() { finished = true; return bestEffortCancel(reader) }, + }) +} + +async function readWithTimeout(reader: ReadableStreamDefaultReader, timeout: number, signal?: AbortSignal) { + if (signal?.aborted) throw signal.reason ?? new DOMException("Aborted", "AbortError") + let timer: ReturnType | undefined + let abort: (() => void) | undefined + const abortPromise = new Promise((_, reject) => { + abort = () => reject(signal?.reason ?? new DOMException("Aborted", "AbortError")) + signal?.addEventListener("abort", abort, { once: true }) + }) + const timeoutPromise = new Promise((_, reject) => { timer = setTimeout(() => reject(new ProviderError.ResponseStreamError("Codex SSE read timed out")), timeout) }) + try { return await Promise.race([reader.read(), abortPromise, timeoutPromise]) } + catch (error) { void bestEffortCancel(reader); throw error } + finally { if (timer) clearTimeout(timer); if (abort && signal) signal.removeEventListener("abort", abort) } +} + +async function readPreview(body: ReadableStream, timeout: number, signal?: AbortSignal) { + const reader = body.getReader() + const chunks: Uint8Array[] = [] + let size = 0 + try { + while (size < 1024) { + const part = await readWithTimeout(reader, timeout, signal) + if (part.done) break + chunks.push(part.value.slice(0, 1024 - size)); size += part.value.byteLength + } + } finally { void bestEffortCancel(reader) } + return new TextDecoder().decode(join(chunks)).slice(0, 1024) +} + +async function bestEffortCancel(value: ReadableStream | ReadableStreamDefaultReader | null | undefined) { + if (!value) return + try { await Promise.race([value.cancel(), Promise.resolve()]) } catch {} +} + +function failingResponse(error: Error, headers: HeadersInit) { + const normalized = new Headers(headers) + ;["content-length", "content-encoding", "content-type"].forEach((name) => normalized.delete(name)) + return new Response(new ReadableStream({ start(controller) { controller.error(error) } }), { status: 200, headers: normalized }) +} +function join(chunks: Uint8Array[]) { const out = new Uint8Array(chunks.reduce((n, chunk) => n + chunk.byteLength, 0)); chunks.reduce((n, chunk) => (out.set(chunk, n), n + chunk.byteLength), 0); return out } +function toError(error: unknown) { return error instanceof Error ? error : new Error(String(error)) } +function abortableSleep(ms: number, signal?: AbortSignal | null) { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason ?? new DOMException("Aborted", "AbortError")) + let id: ReturnType + const abort = () => { + clearTimeout(id) + signal?.removeEventListener("abort", abort) + reject(signal?.reason ?? new DOMException("Aborted", "AbortError")) + } + id = setTimeout(() => { + signal?.removeEventListener("abort", abort) + resolve() + }, ms) + signal?.addEventListener("abort", abort, { once: true }) + }) +} diff --git a/packages/opencode/src/plugin/openai/codex.ts b/packages/opencode/src/plugin/openai/codex.ts index d16b7495654c..7a124407e491 100644 --- a/packages/opencode/src/plugin/openai/codex.ts +++ b/packages/opencode/src/plugin/openai/codex.ts @@ -6,6 +6,7 @@ import { setTimeout as sleep } from "node:timers/promises" import { createServer } from "http" import { OpenAIWebSocketPool } from "./ws-pool" import { OauthCallbackPage } from "@opencode-ai/core/oauth/page" +import { CODEX_CHUNK_TIMEOUT, CODEX_HEADER_TIMEOUT, fetchCodexHTTP } from "./codex-http" const CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann" const ISSUER = "https://auth.openai.com" @@ -14,6 +15,7 @@ const OAUTH_PORT = 1455 const OAUTH_POLLING_SAFETY_MARGIN_MS = 3000 const ALLOWED_MODELS = new Set(["gpt-5.5", "gpt-5.3-codex-spark", "gpt-5.4", "gpt-5.4-mini"]) const DISALLOWED_MODELS = new Set(["gpt-5.5-pro"]) +const MAX_REQUEST_BODY_BYTES = 16 * 1024 * 1024 interface PkceCodes { verifier: string @@ -75,6 +77,17 @@ export function extractAccountId(tokens: TokenResponse): string | undefined { return undefined } +function requireRefreshToken(tokens: TokenResponse) { + if (!tokens.refresh_token) throw new Error("OAuth authorization did not return a refresh token") + return tokens.refresh_token +} + +function requireAccessToken(tokens: TokenResponse) { + const access = tokens.access_token?.trim() + if (!access) throw new Error("OAuth refresh did not return a usable access token") + return access +} + function buildAuthorizeUrl(redirectUri: string, pkce: PkceCodes, state: string): string { const params = new URLSearchParams({ response_type: "code", @@ -94,7 +107,7 @@ function buildAuthorizeUrl(redirectUri: string, pkce: PkceCodes, state: string): interface TokenResponse { id_token: string access_token: string - refresh_token: string + refresh_token?: string expires_in?: number } @@ -102,6 +115,59 @@ interface CodexAuthPluginOptions { issuer?: string codexApiEndpoint?: string experimentalWebSockets?: boolean + httpHeaderTimeout?: number + httpChunkTimeout?: number + websocketConnectTimeout?: number + websocketIdleTimeout?: number + websocketPoolFactory?: (options: { + httpFetch: typeof fetch + connectTimeout?: number + idleTimeout?: number + }) => ((input: RequestInfo | URL, init?: RequestInit) => Promise) & { + close: () => void + remove: (id: string) => void + } + codexHTTPTransport?: CodexHTTPTransport +} + +interface CodexHTTPTransport { + (input: RequestInfo | URL, init: RequestInit | undefined, options: { headerTimeout: number; chunkTimeout: number }): Promise +} + +async function readRequestBody(request: Request) { + if (request.method === "GET" || request.method === "HEAD" || !request.body) return undefined + const reader = request.body.getReader() + const chunks: Uint8Array[] = [] + let size = 0 + try { + while (true) { + const part = await Promise.race([ + reader.read(), + new Promise((_, reject) => { + const abort = () => reject(request.signal.reason ?? new DOMException("Aborted", "AbortError")) + request.signal.addEventListener("abort", abort, { once: true }) + void reader.closed.finally(() => request.signal.removeEventListener("abort", abort)).catch(() => {}) + }), + ]) + if (part.done) return joinBytes(chunks) + size += part.value.byteLength + if (size > MAX_REQUEST_BODY_BYTES) { + void reader.cancel() + throw new Error(`Request body exceeds ${MAX_REQUEST_BODY_BYTES} bytes`) + } + chunks.push(part.value) + } + } catch (error) { + void reader.cancel() + throw error + } +} + + +function joinBytes(chunks: Uint8Array[]) { + const result = new Uint8Array(chunks.reduce((size, chunk) => size + chunk.byteLength, 0)) + chunks.reduce((offset, chunk) => (result.set(chunk, offset), offset + chunk.byteLength), 0) + return result } async function exchangeCodeForTokens(code: string, redirectUri: string, pkce: PkceCodes): Promise { @@ -321,8 +387,17 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug provider: "openai", async loader(getAuth) { const auth = await getAuth() + const codexHTTPTransport = options.codexHTTPTransport ?? ((input, init, transportOptions) => fetchCodexHTTP(input, init, transportOptions)) + const oauthHTTPFetch = ((input: RequestInfo | URL, init?: RequestInit) => codexHTTPTransport(input, init, { + headerTimeout: options.httpHeaderTimeout ?? CODEX_HEADER_TIMEOUT, + chunkTimeout: options.httpChunkTimeout ?? CODEX_CHUNK_TIMEOUT, + })) as typeof fetch const websocketFetch = options.experimentalWebSockets - ? OpenAIWebSocketPool.createWebSocketFetch({ httpFetch: fetch }) + ? (options.websocketPoolFactory ?? OpenAIWebSocketPool.createWebSocketFetch)({ + httpFetch: auth.type === "oauth" ? oauthHTTPFetch : fetch, + connectTimeout: options.websocketConnectTimeout ?? CODEX_HEADER_TIMEOUT, + idleTimeout: options.websocketIdleTimeout ?? CODEX_CHUNK_TIMEOUT, + }) : undefined if (websocketFetch) { websocketFetches.push(websocketFetch) @@ -337,80 +412,57 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug }> | undefined + const awaitRefresh = async (signal: AbortSignal | undefined, failedAccess?: string) => { + const latest = await getAuth() + if (latest.type !== "oauth") throw new Error("OAuth credentials unavailable") + if (failedAccess && latest.access?.trim() && latest.access !== failedAccess) return Object.freeze({ access: latest.access.trim(), accountId: latest.accountId }) + if (!refreshPromise) { + refreshPromise = refreshAccessToken(latest.refresh, issuer).then(async (tokens) => { + const access = requireAccessToken(tokens) + const accountId = extractAccountId(tokens) || latest.accountId + await input.client.auth.set({ path: { id: "openai" }, body: { type: "oauth", refresh: tokens.refresh_token?.trim() || latest.refresh, access, expires: Date.now() + (tokens.expires_in ?? 3600) * 1000, ...(accountId && { accountId }) } }) + return Object.freeze({ access, accountId }) + }).finally(() => { refreshPromise = undefined }) + } + if (!signal) return refreshPromise + return Promise.race([ + refreshPromise, + new Promise((_, reject) => { + const abort = () => reject(signal.reason ?? new DOMException("Aborted", "AbortError")) + signal.addEventListener("abort", abort, { once: true }) + void refreshPromise!.finally(() => signal.removeEventListener("abort", abort)).catch(() => {}) + }), + ]) + } + return { + headerTimeout: false, + chunkTimeout: false, apiKey: OAUTH_DUMMY_KEY, async fetch(requestInput: RequestInfo | URL, init?: RequestInit) { - if (init?.headers) { - if (init.headers instanceof Headers) { - init.headers.delete("authorization") - init.headers.delete("Authorization") - } else if (Array.isArray(init.headers)) { - init.headers = init.headers.filter(([key]) => key.toLowerCase() !== "authorization") - } else { - delete init.headers["authorization"] - delete init.headers["Authorization"] - } - } + const request = new Request(requestInput, init) + const requestBody = await readRequestBody(request) + const snapshot = { url: new URL(request.url), method: request.method, headers: new Headers(request.headers), body: requestBody, signal: request.signal } const currentAuth = await getAuth() if (currentAuth.type !== "oauth") - return websocketFetch ? websocketFetch(requestInput, init) : fetch(requestInput, init) + return websocketFetch ? websocketFetch(snapshot.url, { ...init, method: snapshot.method, headers: snapshot.headers, body: snapshot.body }) : fetch(snapshot.url, { ...init, method: snapshot.method, headers: snapshot.headers, body: snapshot.body }) const authWithAccount = currentAuth as typeof currentAuth & { accountId?: string } - if (!currentAuth.access || currentAuth.expires < Date.now()) { - if (!refreshPromise) { - refreshPromise = refreshAccessToken(currentAuth.refresh, issuer) - .then(async (tokens) => { - const accountId = extractAccountId(tokens) || authWithAccount.accountId - await input.client.auth.set({ - path: { id: "openai" }, - body: { - type: "oauth", - refresh: tokens.refresh_token, - access: tokens.access_token, - expires: Date.now() + (tokens.expires_in ?? 3600) * 1000, - ...(accountId && { accountId }), - }, - }) - return { - access: tokens.access_token, - accountId, - } - }) - .finally(() => { - refreshPromise = undefined - }) - } - - const refreshed = await refreshPromise - currentAuth.access = refreshed.access - authWithAccount.accountId = refreshed.accountId - } + let credentials = Object.freeze({ access: currentAuth.access?.trim() ?? "", accountId: authWithAccount.accountId }) + if (!credentials.access || currentAuth.expires < Date.now()) credentials = await awaitRefresh(snapshot.signal) - const headers = new Headers() - if (init?.headers) { - if (init.headers instanceof Headers) { - init.headers.forEach((value, key) => headers.set(key, value)) - } else if (Array.isArray(init.headers)) { - for (const [key, value] of init.headers) { - if (value !== undefined) headers.set(key, String(value)) - } - } else { - for (const [key, value] of Object.entries(init.headers)) { - if (value !== undefined) headers.set(key, String(value)) - } - } - } - headers.set("authorization", `Bearer ${currentAuth.access}`) - if (authWithAccount.accountId) { - headers.set("ChatGPT-Account-Id", authWithAccount.accountId) + const headers = new Headers(snapshot.headers) + headers.delete("authorization") + headers.delete("chatgpt-account-id") + headers.set("authorization", `Bearer ${credentials.access}`) + if (credentials.accountId) { + headers.set("ChatGPT-Account-Id", credentials.accountId) } const parsed = - requestInput instanceof URL - ? requestInput - : new URL(typeof requestInput === "string" ? requestInput : requestInput.url) + snapshot.url const url = parsed.pathname.includes("/v1/responses") || parsed.pathname.includes("/chat/completions") ? new URL(codexApiEndpoint) @@ -418,11 +470,28 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug const requestInit = { ...init, - body: init?.body, + method: snapshot.method, + body: snapshot.body, headers, } - if (websocketFetch && parsed.pathname.endsWith("/responses")) return websocketFetch(url, requestInit) - return fetch(url, OpenAIWebSocketPool.withoutInternalHeaders(requestInit)) + const sendDirect = (target: URL, directInit: RequestInit) => fetch(target, OpenAIWebSocketPool.withoutInternalHeaders(directInit)) + if (!parsed.pathname.includes("/v1/responses") && !parsed.pathname.includes("/chat/completions")) return sendDirect(snapshot.url, requestInit) + const send = (access: string, accountId: string | undefined) => { + const retryHeaders = new Headers(headers) + retryHeaders.set("authorization", `Bearer ${access}`) + if (accountId) retryHeaders.set("ChatGPT-Account-Id", accountId) + const selected = websocketFetch && parsed.pathname.endsWith("/responses") ? websocketFetch : fetchCodexHTTP + if (selected === websocketFetch) return selected(url, { ...requestInit, body: snapshot.body ? new TextDecoder().decode(snapshot.body) : undefined, headers: retryHeaders, signal: snapshot.signal }) + return selected(url, OpenAIWebSocketPool.withoutInternalHeaders({ ...requestInit, headers: retryHeaders }), { + headerTimeout: options.httpHeaderTimeout ?? CODEX_HEADER_TIMEOUT, + chunkTimeout: options.httpChunkTimeout ?? CODEX_CHUNK_TIMEOUT, + }) + } + const response = await send(credentials.access, credentials.accountId) + if (response.status !== 401 || init?.signal?.aborted) return response + void response.body?.cancel() + const refreshed = await awaitRefresh(snapshot.signal, credentials.access) + return send(refreshed.access, refreshed.accountId) }, } }, @@ -448,7 +517,7 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug const accountId = extractAccountId(tokens) return { type: "success" as const, - refresh: tokens.refresh_token, + refresh: requireRefreshToken(tokens), access: tokens.access_token, expires: Date.now() + (tokens.expires_in ?? 3600) * 1000, accountId, @@ -523,7 +592,7 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug return { type: "success" as const, - refresh: tokens.refresh_token, + refresh: requireRefreshToken(tokens), access: tokens.access_token, expires: Date.now() + (tokens.expires_in ?? 3600) * 1000, accountId: extractAccountId(tokens), diff --git a/packages/opencode/test/plugin/codex-http.test.ts b/packages/opencode/test/plugin/codex-http.test.ts new file mode 100644 index 000000000000..e7080eb0ae63 --- /dev/null +++ b/packages/opencode/test/plugin/codex-http.test.ts @@ -0,0 +1,319 @@ +import { describe, expect, test } from "bun:test" +import { ProviderError } from "../../src/provider/error" +import { CODEX_CHUNK_TIMEOUT, CODEX_HEADER_TIMEOUT, fetchCodexHTTP } from "../../src/plugin/openai/codex-http" + +const CREATED = "data: {\"type\":\"response.created\"}\n\n" + +function sse(value: string, init?: ResponseInit) { + return new Response(value, { headers: { "content-type": "text/event-stream" }, ...init }) +} + +function byteStream(chunks: Uint8Array[], onCancel?: () => void) { + let index = 0 + return new ReadableStream({ + pull(controller) { + const chunk = chunks[index++] + if (chunk) controller.enqueue(chunk) + else controller.close() + }, + cancel() { + onCancel?.() + }, + }) +} + +function responseBytes(bytes: Uint8Array, headers = { "content-type": "text/event-stream" }) { + return new Response(byteStream([bytes]), { headers }) +} + +describe("Codex HTTP transport", () => { + test("uses the independent status retry policies", async () => { + for (const [status, retries, delay] of [ + [502, 3, 3_000], + [503, 3, 2_000], + [504, 2, 3_000], + ] as const) { + let calls = 0 + const waits: number[] = [] + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => ++calls <= retries ? new Response("retry", { status }) : sse(CREATED), + sleep: async (ms) => void waits.push(ms), + headerTimeout: 1, + }) + expect(calls).toBe(retries + 1) + expect(waits).toEqual(Array(retries).fill(delay)) + expect(await response.text()).toBe(CREATED) + } + }) + + test("retries network failures without consuming HTTP counters", async () => { + let calls = 0 + const waits: number[] = [] + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => { + calls++ + if (calls <= 3) throw new Error("connect failed") + return sse(CREATED) + }, + sleep: async (ms) => void waits.push(ms), + }) + expect(calls).toBe(4) + expect(waits).toEqual([3_000, 3_000, 3_000]) + expect(await response.text()).toBe(CREATED) + }) + + test("retries an overloaded first SSE event and replays accepted bytes exactly once", async () => { + const overload = 'data: {"type":"error","code":"server_is_overloaded"}\r\n\r\n' + const accepted = 'data: {"type":"response.created","text":"é"}\r\n\r\n' + const chunks = [new Uint8Array(Buffer.from(accepted.slice(0, 18))), new Uint8Array(Buffer.from(accepted.slice(18)))] + let calls = 0 + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => { + calls++ + if (calls === 1) return sse(overload) + return new Response(new ReadableStream({ + pull(controller) { + const chunk = chunks.shift() + if (chunk) controller.enqueue(chunk) + else controller.close() + }, + }), { headers: { "content-type": "text/event-stream" } }) + }, + sleep: async () => {}, + }) + expect(calls).toBe(2) + expect(new Uint8Array(await response.arrayBuffer())).toEqual(new Uint8Array(Buffer.from(accepted))) + }) + + test("exposes typed errors for malformed status-200 non-SSE bodies", async () => { + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response("not sse", { status: 200, headers: { "content-type": "text/plain" } }), + }) + expect(response.status).toBe(200) + await expect(response.text()).rejects.toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("exports Codex timeout defaults", () => { + expect(CODEX_HEADER_TIMEOUT).toBe(60_000) + expect(CODEX_CHUNK_TIMEOUT).toBe(360_000) + }) + + test("keeps an exhausted SSE error readable exactly once", async () => { + const error = 'data: {"type":"error","code":"server_is_overloaded"}\n\n' + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => sse(error), + sleep: async () => {}, + }) + expect(await response.text()).toBe(error) + }) + + test("allows large later bytes when the first event is small", async () => { + const first = 'data: {"type":"response.created"}\n\n' + const later = "x".repeat(70_000) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => sse(first + later), + }) + expect(await response.text()).toBe(first + later) + }) + + test("does not read beyond the first event until the consumer pulls", async () => { + const first = new TextEncoder().encode('data: {"type":"response.created"}\n\n') + let reads = 0 + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(first) + }, + pull(controller) { + reads++ + controller.enqueue(new TextEncoder().encode("later")) + }, + }, { highWaterMark: 0 }), { headers: { "content-type": "text/event-stream" } }), + }) + expect(reads).toBe(0) + const reader = response.body?.getReader() + expect(await reader?.read()).toEqual({ done: false, value: first }) + expect(reads).toBe(1) + await reader?.cancel() + }) + + test("accepts nested error codes and ignores overload text in output", async () => { + const nested = 'data: {"type":"error","error":{"code":"service_unavailable_error"}}\n\n' + let calls = 0 + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => { + calls++ + return calls === 1 ? sse(nested) : sse('data: {"type":"response.output_text.delta","delta":"server_is_overloaded"}\n\n') + }, + sleep: async () => {}, + }) + expect(calls).toBe(2) + expect(await response.text()).toContain("server_is_overloaded") + }) + + test("turns an empty SSE response into a typed failing status-200 stream", async () => { + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => sse(""), + }) + expect(response.status).toBe(200) + await expect(response.text()).rejects.toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("handles LF separator split across byte chunks", async () => { + const bytes = new TextEncoder().encode(CREATED) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response(byteStream([bytes.slice(0, -1), bytes.slice(-1)]), { headers: { "content-type": "text/event-stream" } }), + }) + expect(new Uint8Array(await response.arrayBuffer())).toEqual(bytes) + }) + + test("handles CRLF separator split across byte chunks", async () => { + const value = 'data: {"type":"response.created"}\r\n\r\n' + const bytes = new TextEncoder().encode(value) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response(byteStream([bytes.slice(0, -1), bytes.slice(-1)]), { headers: { "content-type": "text/event-stream" } }), + }) + expect(new Uint8Array(await response.arrayBuffer())).toEqual(bytes) + }) + + test("preserves UTF-8 bytes split inside a multibyte character", async () => { + const value = 'data: {"type":"response.output_text.delta","delta":"é"}\n\n' + const bytes = new TextEncoder().encode(value) + const position = bytes.indexOf(0xc3) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response(byteStream([bytes.slice(0, position + 1), bytes.slice(position + 1)]), { headers: { "content-type": "text/event-stream" } }), + }) + expect(new Uint8Array(await response.arrayBuffer())).toEqual(bytes) + }) + + test("rejects a truly oversized first event", async () => { + const bytes = new TextEncoder().encode(`data: ${"x".repeat(65 * 1024)}\n\n`) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => responseBytes(bytes) }) + expect(response.status).toBe(200) + await expect(response.text()).rejects.toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("exhausts both SSE codes independently and keeps both final streams readable", async () => { + for (const code of ["server_is_overloaded", "service_unavailable_error"] as const) { + const value = `data: {"type":"error","code":"${code}"}\n\n` + let calls = 0 + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => { calls++; return sse(value) }, sleep: async () => {} }) + expect(calls).toBe(4) + expect(await response.text()).toBe(value) + } + }) + + test("aborting during signal-aware header wait prevents later attempts", async () => { + const controller = new AbortController() + let calls = 0 + const promise = fetchCodexHTTP("https://chatgpt.test/responses", { signal: controller.signal }, { + fetch: async (_input, init) => { + calls++ + await new Promise((resolve, reject) => init?.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true })) + return sse(CREATED) + }, headerTimeout: 100, + }) + setTimeout(() => controller.abort(new Error("cancelled")), 10) + await expect(promise).rejects.toThrow("cancelled") + expect(calls).toBe(1) + }) + + test("aborting during first-event read prevents later attempts", async () => { + const controller = new AbortController() + let calls = 0 + const promise = fetchCodexHTTP("https://chatgpt.test/responses", { signal: controller.signal }, { + fetch: async () => { calls++; return new Response(new ReadableStream({ pull() {} }), { headers: { "content-type": "text/event-stream" } }) }, chunkTimeout: 100, + }) + setTimeout(() => controller.abort(new Error("read cancelled")), 10) + await expect(promise).rejects.toThrow("read cancelled") + expect(calls).toBe(1) + }) + + test("aborting during retry backoff prevents later attempts", async () => { + const controller = new AbortController() + let calls = 0 + const promise = fetchCodexHTTP("https://chatgpt.test/responses", { signal: controller.signal }, { + fetch: async () => { calls++; return new Response("retry", { status: 503 }) }, sleep: (ms, signal) => new Promise((resolve, reject) => signal?.addEventListener("abort", () => reject(signal.reason), { once: true })), + }) + setTimeout(() => controller.abort(new Error("backoff cancelled")), 10) + await expect(promise).rejects.toThrow("backoff cancelled") + expect(calls).toBe(1) + }) + + test("chunk timeout rejects even when reader cancellation never settles", async () => { + const promise = fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(new ReadableStream({ pull() {}, cancel() { return new Promise(() => {}) } }), { headers: { "content-type": "text/event-stream" } }), chunkTimeout: 10 }) + await expect(promise).rejects.toBeInstanceOf(ProviderError.ResponseStreamError) + }) + + test("raw chunk activity resets the chunk timeout", async () => { + const bytes = new TextEncoder().encode(CREATED) + let index = 0 + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(new ReadableStream({ async pull(controller) { if (index < bytes.length) { await new Promise((resolve) => setTimeout(resolve, 8)); controller.enqueue(bytes.slice(index, ++index)) } else controller.close() } }), { headers: { "content-type": "text/event-stream" } }), chunkTimeout: 15 }) + expect(await response.text()).toBe(CREATED) + }) + + test("preserves 401, 403, and 429 responses unchanged", async () => { + for (const status of [401, 403, 429]) { + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(`body-${status}`, { status, headers: { "x-sentinel": "yes" } }) }) + expect(response.status).toBe(status); expect(response.headers.get("x-sentinel")).toBe("yes"); expect(await response.text()).toBe(`body-${status}`) + } + }) + + test("preserves exhausted 502, 503, and 504 responses unchanged", async () => { + for (const status of [502, 503, 504]) { + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(`final-${status}`, { status, headers: { "x-sentinel": "yes" } }), sleep: async () => {} }) + expect(response.status).toBe(status); expect(response.headers.get("x-sentinel")).toBe("yes"); expect(await response.text()).toBe(`final-${status}`) + } + }) + + test("bounds malformed non-SSE preview and attempts cleanup", async () => { + let cancelled = false + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode("z".repeat(3000))) }, cancel() { cancelled = true } }), { status: 200, headers: { "content-type": "application/json", "content-length": "3000" } }), chunkTimeout: 10 }) + expect(response.status).toBe(200); expect(response.headers.get("content-type")).toBeNull(); expect(cancelled).toBe(true) + await expect(response.text()).rejects.toThrow(/application\/json.*z{1024}/) + }) + + test("consumer cancellation cancels upstream and prevents later reads", async () => { + let reads = 0; let cancelled = false + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => new Response(new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode(CREATED)) }, pull() { reads++ }, cancel() { cancelled = true } }), { headers: { "content-type": "text/event-stream" } }) }) + const reader = response.body?.getReader(); await reader?.read(); await reader?.cancel(); expect(cancelled).toBe(true); const before = reads; await Promise.resolve(); expect(reads).toBe(before) + }) + + test("preserves a valid function and tool-call SSE sequence byte-for-byte", async () => { + const value = 'data: {"type":"response.function_call_arguments.delta","delta":"{\\"x\\":1}"}\r\n\r\ndata: {"type":"response.output_item.done","item":{"type":"function_call"}}\r\n\r\n' + const bytes = new TextEncoder().encode(value) + const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { fetch: async () => responseBytes(bytes) }) + expect(new Uint8Array(await response.arrayBuffer())).toEqual(bytes) + }) + + test("removes settled read abort listeners", async () => { + const controller = new AbortController(); let listeners = 0 + const add = controller.signal.addEventListener.bind(controller.signal); const remove = controller.signal.removeEventListener.bind(controller.signal) + controller.signal.addEventListener = ((...args: Parameters) => { listeners++; return add(...args) }) as typeof add + controller.signal.removeEventListener = ((...args: Parameters) => { listeners--; return remove(...args) }) as typeof remove + const response = await fetchCodexHTTP("https://chatgpt.test/responses", { signal: controller.signal }, { fetch: async () => responseBytes(new TextEncoder().encode(CREATED)), chunkTimeout: 20 }) + await response.text(); controller.abort(); expect(listeners).toBe(0) + }) + + test("returns a typed response when malformed preview times out", async () => { + let cancelled = false + const promise = fetchCodexHTTP("https://chatgpt.test/responses", {}, { + fetch: async () => new Response(new ReadableStream({ pull() {}, cancel() { cancelled = true; return new Promise(() => {}) } }), { status: 200, headers: { "content-type": "text/plain" } }), + chunkTimeout: 10, + }) + const response = await promise + expect(response.status).toBe(200) + expect(cancelled).toBe(true) + await expect(response.text()).rejects.toThrow(/text\/plain.*preview.*timed out/i) + }) + + test("caller abort during malformed preview rejects the outer request", async () => { + const controller = new AbortController() + const promise = fetchCodexHTTP("https://chatgpt.test/responses", { signal: controller.signal }, { + fetch: async () => new Response(new ReadableStream({ pull() {} }), { status: 200, headers: { "content-type": "text/plain" } }), + chunkTimeout: 100, + }) + setTimeout(() => controller.abort(new Error("preview cancelled")), 10) + await expect(promise).rejects.toThrow("preview cancelled") + }) +}) diff --git a/packages/opencode/test/plugin/codex.test.ts b/packages/opencode/test/plugin/codex.test.ts index 1381c4ee8adb..81835ae86dfa 100644 --- a/packages/opencode/test/plugin/codex.test.ts +++ b/packages/opencode/test/plugin/codex.test.ts @@ -149,6 +149,266 @@ describe("plugin.codex", () => { await enabled.dispose?.() }) + test("OAuth loader disables generic timeouts and exposes injectable transport defaults", async () => { + const hooks = await CodexAuthPlugin({} as never, { + httpHeaderTimeout: 17, + httpChunkTimeout: 19, + websocketConnectTimeout: 23, + websocketIdleTimeout: 29, + }) + const options = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + expect(options.headerTimeout).toBe(false) + expect(options.chunkTimeout).toBe(false) + await hooks.dispose?.() + }) + + + test("preserves omitted and empty refresh tokens but stores rotated tokens", async () => { + for (const refresh_token of [undefined, "", "refresh-rotated"] as const) { + let auth = { type: "oauth" as const, access: "", refresh: "refresh-old", expires: 0 } + let update: { refresh: string } | undefined + using server = Bun.serve({ + port: 0, + async fetch(request) { + if (new URL(request.url).pathname === "/oauth/token") return Response.json({ access_token: "access-new", ...(refresh_token === undefined ? {} : { refresh_token }), expires_in: 3600 }) + return new Response("ok") + }, + }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: { refresh: string; access: string; expires: number } }) { update = input.body; auth = { ...auth, ...input.body } } } } as never } as never, { issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + await loaded.fetch!(new URL("/responses", server.url)) + expect(update?.refresh).toBe(refresh_token || "refresh-old") + await hooks.dispose?.() + } + }) + + test("preserves whitespace-only refresh tokens and trims rotated refresh tokens", async () => { + for (const refresh_token of [undefined, "", " ", " rotated "] as const) { + let auth = { type: "oauth" as const, access: "", refresh: "refresh-old", expires: 0 } + let update: { refresh: string } | undefined + using server = Bun.serve({ port: 0, async fetch(request) { + if (new URL(request.url).pathname === "/oauth/token") return Response.json({ access_token: "access-new", ...(refresh_token === undefined ? {} : { refresh_token }), expires_in: 3600 }) + return new Response("ok") + } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: { refresh: string; access: string; expires: number } }) { update = input.body; auth = { ...auth, ...input.body } } } } as never } as never, { issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + await loaded.fetch!(new URL("/responses", server.url)) + expect(update?.refresh).toBe(refresh_token?.trim() || "refresh-old") + await hooks.dispose?.() + } + }) + + test("passes production and injectable WebSocket timeouts through the plugin pool factory", async () => { + const calls: Array<{ connectTimeout?: number; idleTimeout?: number }> = [] + const poolFactory = (input: { connectTimeout?: number; idleTimeout?: number }) => { + calls.push(input) + const fetch = Object.assign(async () => new Response("ok"), { close() {}, remove() {} }) + return fetch + } + const defaultHooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketPoolFactory: poolFactory as never }) + const overrideHooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketConnectTimeout: 23, websocketIdleTimeout: 29, websocketPoolFactory: poolFactory as never }) + await defaultHooks.auth!.loader!(async () => ({ type: "api", key: "key" }) as never, {} as never) + await overrideHooks.auth!.loader!(async () => ({ type: "api", key: "key" }) as never, {} as never) + expect(calls.map((call) => ({ connectTimeout: call.connectTimeout, idleTimeout: call.idleTimeout }))).toEqual([{ connectTimeout: 60_000, idleTimeout: 360_000 }, { connectTimeout: 23, idleTimeout: 29 }]) + await defaultHooks.dispose?.() + await overrideHooks.dispose?.() + }) + + test("does not mutate caller headers and replays POST bytes across status retries", async () => { + const callerHeaders = new Headers({ authorization: "caller", "ChatGPT-Account-Id": "caller-account", "x-test": "yes" }) + const bodies: string[] = [] + let calls = 0 + using server = Bun.serve({ port: 0, async fetch(request) { bodies.push(await request.text()); calls++; return new Response("retry", { status: calls === 1 ? 502 : 200, headers: { "content-type": "text/event-stream" } }) } }) + const auth = { type: "oauth" as const, access: "trusted", refresh: "refresh", expires: Date.now() + 60_000, accountId: "trusted-account" } + const hooks = await CodexAuthPlugin({} as never, { codexApiEndpoint: new URL("/responses", server.url).toString() }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const request = new Request("https://api.openai.com/v1/responses", { method: "POST", headers: callerHeaders, body: "payload" }) + await loaded.fetch!(request) + expect(callerHeaders.get("authorization")).toBe("caller") + expect(callerHeaders.get("ChatGPT-Account-Id")).toBe("caller-account") + expect(bodies).toEqual(["payload", "payload"]) + await hooks.dispose?.() + }) + + test("uses direct fetch for unrelated OAuth URLs", async () => { + const hooks = await CodexAuthPlugin({} as never) + const loaded = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + const response = await loaded.fetch!("data:application/json,%7B%22ok%22%3Atrue%7D") + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true }) + }) + + test("rejects refresh responses with an empty access token", async () => { + using server = Bun.serve({ port: 0, fetch: async () => Response.json({ access_token: " ", refresh_token: "refresh" }) }) + const hooks = await CodexAuthPlugin({} as never, { issuer: server.url.origin, codexApiEndpoint: new URL("/v1/responses", server.url).toString() }) + const loaded = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "", refresh: "refresh", expires: 0 }) as never, {} as never) + const result = loaded.fetch!(new URL("/v1/responses", server.url)).catch((error: unknown) => error) + expect(await result).toBeInstanceOf(Error) + }) + + test("recovers a wrapped WebSocket 401 once and replays SSE bytes", async () => { + const sent: Array<{ body: string; authorization: string | null; account: string | null }> = [] + let refreshes = 0 + let auth = { type: "oauth" as const, access: "old", refresh: "refresh", expires: Date.now() + 60_000, accountId: "old-account" } + const sse = 'data: {"type":"response.created"}\n\n' + const factory = () => { + let calls = 0 + return Object.assign(async (_url: URL, init?: RequestInit) => { + calls++ + sent.push({ body: String(init?.body ?? ""), authorization: new Headers(init?.headers).get("authorization"), account: new Headers(init?.headers).get("chatgpt-account-id") }) + return calls === 1 ? new Response("unauthorized", { status: 401 }) : new Response(sse, { status: 200, headers: { "content-type": "text/event-stream" } }) + }, { close() {}, remove() {} }) + } + using server = Bun.serve({ port: 0, fetch: async (request) => { refreshes++; await request.text(); return Response.json({ access_token: "new", refresh_token: "refresh", expires_in: 3600 }) } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: typeof auth }) { auth = { ...auth, ...input.body } } } } as never } as never, { experimentalWebSockets: true, websocketPoolFactory: factory as never, issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const response = await loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: "body" }) + expect(await response.text()).toBe(sse); expect(refreshes).toBe(1); expect(sent).toHaveLength(2); expect(sent[1]).toEqual({ body: "body", authorization: "Bearer new", account: "old-account" }) + await hooks.dispose?.() + }) + + test("returns a second WebSocket 401 unchanged and does not retry again", async () => { + let sends = 0; let refreshes = 0 + let auth = { type: "oauth" as const, access: "old", refresh: "refresh", expires: Date.now() + 60_000 } + const factory = () => Object.assign(async () => { sends++; return new Response("second", { status: 401, headers: { "x-sentinel": "yes" } }) }, { close() {}, remove() {} }) + using server = Bun.serve({ port: 0, fetch: async () => { refreshes++; return Response.json({ access_token: "new", refresh_token: "refresh" }) } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: typeof auth }) { auth = { ...auth, ...input.body } } } } as never } as never, { experimentalWebSockets: true, websocketPoolFactory: factory as never, issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never); const response = await loaded.fetch!("https://api.openai.com/v1/responses") + expect(response.status).toBe(401); expect(response.headers.get("x-sentinel")).toBe("yes"); expect(await response.text()).toBe("second"); expect(sends).toBe(2); expect(refreshes).toBe(1) + await hooks.dispose?.() + }) + + test("WebSocket pool receives reliable OAuth HTTP fallback transport", async () => { + let httpFetch: typeof fetch | undefined + const optionsSeen: { headerTimeout?: number; chunkTimeout?: number }[] = [] + const transport = async (input: RequestInfo | URL, init: RequestInit | undefined, options: { headerTimeout?: number; chunkTimeout?: number }) => { + void input; void init; optionsSeen.push(options) + return new Response('data: {"type":"response.created"}\n\n', { headers: { "content-type": "text/event-stream" } }) + } + const factory = (options: { httpFetch: typeof fetch }) => { httpFetch = options.httpFetch; return Object.assign(async () => new Response('data: {"type":"response.created"}\n\n', { headers: { "content-type": "text/event-stream" } }), { close() {}, remove() {} }) } + const hooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketPoolFactory: factory as never, httpHeaderTimeout: 12, httpChunkTimeout: 13, codexHTTPTransport: transport } as never) + await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + expect(httpFetch).toBeFunction() + const response = await httpFetch!("https://chatgpt.test/responses", { method: "POST", body: "body" }) + expect(response).toBeInstanceOf(Response); expect(await response.text()).toContain("response.created"); expect(optionsSeen).toEqual([{ headerTimeout: 12, chunkTimeout: 13 }]) + await hooks.dispose?.() + + let defaultFetch: typeof fetch | undefined + const defaultHooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketPoolFactory: (options: { httpFetch: typeof fetch }) => { defaultFetch = options.httpFetch; return Object.assign(async () => new Response(), { close() {}, remove() {} }) }, codexHTTPTransport: transport } as never) + await defaultHooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + await defaultFetch!("https://chatgpt.test/responses") + expect(optionsSeen.at(-1)).toEqual({ headerTimeout: 60_000, chunkTimeout: 360_000 }) + await defaultHooks.dispose?.() + }) + + test("stream request bodies replay across 401 and oversized bodies fail before send", async () => { + let calls = 0 + const factory = () => Object.assign(async () => { calls++; return new Response('data: {"type":"response.created"}\n\n', { headers: { "content-type": "text/event-stream" } }) }, { close() {}, remove() {} }) + const hooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketPoolFactory: factory as never }) + const loaded = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + const stream = new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode("body")); controller.close() } }) + const response = await loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: stream }); await response.arrayBuffer(); expect(calls).toBe(1) + await expect(loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: "x".repeat(17 * 1024 * 1024) })).rejects.toThrow(); expect(calls).toBe(1) + await hooks.dispose?.() + }) + + test("Request signal is propagated when init signal is absent", async () => { + const controller = new AbortController(); let seen: AbortSignal | undefined + const factory = () => Object.assign(async (_url: URL, init?: RequestInit) => { seen = init?.signal ?? undefined; return new Promise(() => {}) }, { close() {}, remove() {} }) + const hooks = await CodexAuthPlugin({} as never, { experimentalWebSockets: true, websocketPoolFactory: factory as never }) + const loaded = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + const promise = loaded.fetch!(new Request("https://api.openai.com/v1/responses", { signal: controller.signal })); await waitFor(() => seen !== undefined); controller.abort(); expect(seen).toBeDefined(); await hooks.dispose?.(); void promise.catch(() => {}) + }) + + test("shares blocked reactive refresh while an aborted waiter does not reissue", async () => { + let refreshRequests = 0 + let releaseRefresh!: () => void + const refreshReady = new Promise((resolve) => { releaseRefresh = resolve }) + let auth = { type: "oauth" as const, access: "old", refresh: "refresh", expires: Date.now() + 60_000, accountId: "account" } + let apiCalls = 0 + using server = Bun.serve({ port: 0, async fetch(request) { + if (new URL(request.url).pathname === "/oauth/token") { refreshRequests++; await refreshReady; return Response.json({ access_token: "new", refresh_token: "refresh", expires_in: 3600 }) } + apiCalls++; await request.text(); return new Response(apiCalls <= 2 ? "unauthorized" : 'data: {"type":"response.created"}\n\n', { status: apiCalls <= 2 ? 401 : 200, headers: apiCalls > 2 ? { "content-type": "text/event-stream" } : undefined }) + } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: typeof auth }) { auth = { ...auth, ...input.body } } } } as never } as never, { issuer: server.url.origin, codexApiEndpoint: new URL("/responses", server.url).toString() }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const a = new AbortController(); const first = loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", signal: a.signal, body: "a" }); const second = loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: "b" }) + await waitFor(() => refreshRequests === 1); a.abort(new Error("caller aborted")); await expect(first).rejects.toThrow("caller aborted"); releaseRefresh(); const response = await second; expect(await response.text()).toContain("response.created"); expect(refreshRequests).toBe(1); expect(apiCalls).toBe(3) + await hooks.dispose?.() + }) + + test("rejects an oversized stream before consuming its remainder", async () => { + let pulls = 0; let transportCalls = 0 + const hooks = await CodexAuthPlugin({} as never, { codexHTTPTransport: async () => { transportCalls++; return new Response('data: {"type":"response.created"}\n\n', { headers: { "content-type": "text/event-stream" } }) } } as never) + const loaded = await hooks.auth!.loader!(async () => ({ type: "oauth", access: "access", refresh: "refresh", expires: Date.now() + 60_000 }) as never, {} as never) + const stream = new ReadableStream({ pull(controller) { pulls++; controller.enqueue(new Uint8Array(17 * 1024 * 1024)); controller.enqueue(new Uint8Array(1)); controller.close() } }) + await expect(loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: stream })).rejects.toThrow(/16.*bytes/); expect(pulls).toBe(1); expect(transportCalls).toBe(0); await hooks.dispose?.() + }) + + test("refreshes once after a 401 and reissues the identical request", async () => { + let auth = { type: "oauth" as const, access: "access-old", refresh: "refresh-old", expires: Date.now() + 60_000, accountId: "account-old" } + let refreshes = 0 + const requests: Array<{ authorization: string | null; account: string | null; body: string }> = [] + using server = Bun.serve({ port: 0, async fetch(request) { + if (new URL(request.url).pathname === "/oauth/token") { refreshes++; return Response.json({ access_token: "access-new", refresh_token: "refresh-new", expires_in: 3600 }) } + requests.push({ authorization: request.headers.get("authorization"), account: request.headers.get("chatgpt-account-id"), body: await request.text() }) + return new Response("response", { status: requests.length === 1 ? 401 : 200 }) + } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: typeof auth }) { auth = { ...auth, ...input.body } } } } as never } as never, { issuer: server.url.origin, codexApiEndpoint: new URL("/responses", server.url).toString() }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const body = "request-body" + const response = await loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body }) + expect(response.status).toBe(200); expect(refreshes).toBe(1); expect(requests.map((request) => request.body)).toEqual([body, body]); expect(requests[1]).toEqual({ authorization: "Bearer access-new", account: "account-old", body }) + await hooks.dispose?.() + }) + + test("shares one refresh for concurrent 401 requests and stops after the second 401", async () => { + for (const finalStatus of [200, 401] as const) { + let auth = { type: "oauth" as const, access: "access-old", refresh: "refresh-old", expires: Date.now() + 60_000, accountId: "account" } + let refreshes = 0; let requests = 0 + using server = Bun.serve({ port: 0, async fetch(request) { + if (new URL(request.url).pathname === "/oauth/token") { refreshes++; return Response.json({ access_token: "access-new", refresh_token: "refresh-new", expires_in: 3600 }) } + await request.text(); requests++; return new Response("body", { status: requests <= 2 ? 401 : finalStatus }) + } }) + const hooks = await CodexAuthPlugin({ client: { auth: { async set(input: { body: typeof auth }) { auth = { ...auth, ...input.body } } } } as never } as never, { issuer: server.url.origin, codexApiEndpoint: new URL("/responses", server.url).toString() }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const responses = await Promise.all([loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: "a" }), loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", body: "b" })]) + expect(refreshes).toBe(1); expect(requests).toBe(finalStatus === 401 ? 4 : 4); expect(responses.every((response) => response.status === finalStatus)).toBe(true) + await hooks.dispose?.() + } + }) + + test("does not refresh 403 or 429 responses", async () => { + for (const status of [403, 429]) { + let refreshes = 0 + using server = Bun.serve({ port: 0, async fetch(request) { if (new URL(request.url).pathname === "/oauth/token") { refreshes++; return Response.json({ access_token: "new", refresh_token: "new" }) }; await request.text(); return new Response(`body-${status}`, { status, headers: { "x-sentinel": "yes" } }) } }) + const auth = { type: "oauth" as const, access: "access", refresh: "refresh", expires: Date.now() + 60_000 } + const hooks = await CodexAuthPlugin({} as never, { codexApiEndpoint: new URL("/responses", server.url).toString(), issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + const response = await loaded.fetch!("https://api.openai.com/v1/responses") + expect(response.status).toBe(status); expect(response.headers.get("x-sentinel")).toBe("yes"); expect(await response.text()).toBe(`body-${status}`); expect(refreshes).toBe(0) + await hooks.dispose?.() + } + }) + + test("abort after the first 401 prevents the refreshed reissue", async () => { + const controller = new AbortController() + let calls = 0 + const auth = { type: "oauth" as const, access: "access", refresh: "refresh", expires: Date.now() + 60_000 } + using server = Bun.serve({ port: 0, async fetch(request) { + calls++ + if (new URL(request.url).pathname === "/oauth/token") return Response.json({ access_token: "new", refresh_token: "new", expires_in: 3600 }) + await request.text() + controller.abort(new Error("cancel after 401")) + return new Response("unauthorized", { status: 401 }) + } }) + const hooks = await CodexAuthPlugin({} as never, { codexApiEndpoint: new URL("/responses", server.url).toString(), issuer: server.url.origin }) + const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never) + await expect(loaded.fetch!("https://api.openai.com/v1/responses", { method: "POST", signal: controller.signal, body: "body" })).rejects.toThrow("cancel after 401") + expect(calls).toBe(1) + await hooks.dispose?.() + }) + test("filters unsupported modes and uses Codex context limits for OAuth GPT models", async () => { const hooks = await CodexAuthPlugin({} as never) const limit = { context: 1_050_000, input: 922_000, output: 128_000 } From 37db86d77a569811d7379a7605db9fc23f748eb1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C5=A9ng=20Ho=C3=A0ng=20=C4=90=E1=BB=A9c?= Date: Sat, 18 Jul 2026 08:59:06 +0700 Subject: [PATCH 2/2] chore: remove internal planning docs --- ...2026-07-17-codex-connection-reliability.md | 364 ------------------ ...-17-codex-connection-reliability-design.md | 202 ---------- 2 files changed, 566 deletions(-) delete mode 100644 docs/superpowers/plans/2026-07-17-codex-connection-reliability.md delete mode 100644 docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md diff --git a/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md b/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md deleted file mode 100644 index a70bfe517938..000000000000 --- a/docs/superpowers/plans/2026-07-17-codex-connection-reliability.md +++ /dev/null @@ -1,364 +0,0 @@ -# Codex Connection Reliability Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Add bounded pre-output retries, Codex-specific 60/360-second timeouts, and one-time OAuth recovery without changing terminal or session semantics. - -**Architecture:** Add a focused HTTP transport at the ChatGPT Codex OAuth fetch boundary. It owns per-attempt header timeout, status/network retry, first-SSE-event inspection, raw-byte replay, and inter-chunk timeout. `codex.ts` continues to own OAuth and WebSocket setup; existing parser/session retry remains authoritative after output begins. - -**Tech Stack:** TypeScript, Bun, Fetch/Streams APIs, OpenAI Responses SSE, `ws`, Bun test. - ---- - -## Files - -- Create `packages/opencode/src/plugin/openai/codex-http.ts` — HTTP timeout/retry and first-event gate. -- Create `packages/opencode/test/plugin/codex-http.test.ts` — isolated transport tests. -- Modify `packages/opencode/src/plugin/openai/codex.ts` — integration, OAuth recovery, WebSocket defaults. -- Modify `packages/opencode/test/plugin/codex.test.ts` — OAuth and integration tests. -- Modify `packages/opencode/test/plugin/openai-ws.test.ts` — 60/360 WebSocket option coverage. -- Create `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` — requested Vietnamese analysis. - -## Invariants - -1. Never retry after exposing an SSE event. -2. Never rotate accounts or endpoints. -3. Never synthesize terminal events or extra `[DONE]` frames. -4. Preserve non-2xx status, body, and headers. -5. Abort stops reads, backoff, refresh, and later attempts. -6. Accepted bytes and tool calls are emitted exactly once. - ---- - -### Task 1: Codex status/network retry transport - -**Files:** -- Create: `packages/opencode/src/plugin/openai/codex-http.ts` -- Create: `packages/opencode/test/plugin/codex-http.test.ts` - -- [ ] **Step 1: Write failing retry-policy tests** - -Create table tests for 502 (3 retries, 3000 ms), 503 (3, 2000 ms), 504 (2, 3000 ms), and network failure using injected `fetch` and `sleep` functions: - -```ts -test.each([ - [502, 3, 3_000], - [503, 3, 2_000], - [504, 2, 3_000], -] as const)("retries %d before exposing a response", async (status, retries, delay) => { - let calls = 0 - const waits: number[] = [] - const response = await fetchCodexHTTP("https://chatgpt.test/responses", {}, { - fetch: async () => ++calls <= retries ? new Response("retry", { status }) : sse(CREATED), - sleep: async (ms) => void waits.push(ms), - }) - expect(calls).toBe(retries + 1) - expect(waits).toEqual(Array(retries).fill(delay)) - expect(await response.text()).toBe(CREATED) -}) -``` - -- [ ] **Step 2: Run RED test** - -From `packages/opencode`: - -```bash -bun test test/plugin/codex-http.test.ts -``` - -Expected: FAIL because the module does not exist. - -- [ ] **Step 3: Implement retry policy and injectable options** - -Use this public shape: - -```ts -export const CODEX_HEADER_TIMEOUT = 60_000 -export const CODEX_CHUNK_TIMEOUT = 360_000 - -const POLICY = { - 502: { attempts: 3, delay: 3_000 }, - 503: { attempts: 3, delay: 2_000 }, - 504: { attempts: 2, delay: 3_000 }, -} as const - -export interface CodexHTTPOptions { - fetch?: typeof globalThis.fetch - sleep?: (ms: number, signal?: AbortSignal | null) => Promise - headerTimeout?: number - chunkTimeout?: number -} - -export async function fetchCodexHTTP( - input: RequestInfo | URL, - init: RequestInit = {}, - options: CodexHTTPOptions = {}, -): Promise -``` - -Track independent counters for network/connect failures, HTTP 502, HTTP 503, HTTP 504, `server_is_overloaded`, and `service_unavailable_error`. Network/header-timeout failures use the same numeric policy as HTTP 502 but do not consume the HTTP 502 counter. The two SSE codes use the same numeric policy as HTTP 503 but do not consume the HTTP 503 counter or each other's counter. Cancel failed response bodies before retry. `abortableSleep` must reject immediately when `RequestInit.signal` aborts; this signal is the public cancellation contract. - -- [ ] **Step 4: Implement per-attempt 60-second header timeout** - -`fetchAttempt` must combine caller and timeout signals with `AbortSignal.any`, clear its timer in `finally`, and distinguish caller abort from timeout. Do not use a total timeout covering all retries. - -- [ ] **Step 5: Run GREEN test and inspect diff** - -```bash -bun test test/plugin/codex-http.test.ts -git diff -- packages/opencode/src/plugin/openai/codex-http.ts packages/opencode/test/plugin/codex-http.test.ts -``` - -Expected: tests PASS; only the new module and tests appear. Do not commit unless requested. - ---- - -### Task 2: First-SSE-event gate and 360-second chunk timeout - -**Files:** -- Modify: `packages/opencode/src/plugin/openai/codex-http.ts` -- Modify: `packages/opencode/test/plugin/codex-http.test.ts` - -- [ ] **Step 1: Write failing framing and safety tests** - -Add separate cases for: - -- LF and CRLF event boundaries. -- Boundary split across chunks. -- UTF-8 code point split across chunks. -- Exact byte replay once. -- Immediate return after accepting the first event. -- Structured first-event codes `server_is_overloaded` and `service_unavailable_error` each use an independent counter with the 503 numeric policy. -- The same text inside `response.output_text.delta` does not retry. -- Retry exhaustion exposes one final upstream error stream. -- Abort while reading or sleeping starts no later attempt. -- First event larger than 64 KiB fails safely. - -Core assertion: - -```ts -expect(calls).toBe(2) -expect(await response.text()).toBe(successBytes) -expect(exposedOverloadBytes).toBe(0) -``` - -- [ ] **Step 2: Run RED tests** - -```bash -bun test test/plugin/codex-http.test.ts -``` - -- [ ] **Step 3: Implement byte-safe event detection** - -Scan raw bytes for `[10, 10]` and `[13, 10, 13, 10]`; do not search a decoded string. Buffer at most 64 KiB. Parse only `data:` lines from the first complete event and retry only when parsed JSON has `type: "error"` with an approved structured code. - -Use unchanged buffered chunks when rebuilding the accepted stream: - -```ts -for (const chunk of buffered) controller.enqueue(chunk) -while (true) { - const part = await readWithTimeout(reader, chunkTimeout, signal) - if (part.done) break - controller.enqueue(part.value) -} -controller.close() -``` - -- [ ] **Step 4: Add raw-chunk timeout** - -Wrap every `reader.read()` after headers, including the first event, with a fresh 360-second timer. Each received chunk resets the budget. On timeout, cancel the reader and throw `ProviderError.ResponseStreamError("Codex SSE read timed out")`. - -- [ ] **Step 5: Preserve response semantics** - -Return non-2xx responses unchanged after status retry is exhausted. Do not inspect 401/403/429 as SSE. For malformed 2xx non-SSE responses, return a status-200 `Response` whose body is a `ReadableStream` that fails with `ProviderError.ResponseStreamError`; include content type and a bounded body preview in the error. This preserves the declared `Promise` contract while surfacing a typed stream failure. - -- [ ] **Step 6: Run focused suite** - -```bash -bun test test/plugin/codex-http.test.ts -``` - -Expected: all retry, framing, timeout, replay, and abort cases PASS. - ---- - -### Task 3: Install Codex-specific 60/360 defaults - -**Files:** -- Modify: `packages/opencode/src/plugin/openai/codex.ts:101-105,263-266,320-425` -- Modify: `packages/opencode/test/plugin/codex.test.ts` -- Modify: `packages/opencode/test/plugin/openai-ws.test.ts` - -- [ ] **Step 1: Write failing option-plumbing tests** - -Extend plugin options with injectable test values and assert the OAuth loader disables the generic OpenAI timeout wrapper: - -```ts -expect(loaded.headerTimeout).toBe(false) -expect(loaded.chunkTimeout).toBe(false) -``` - -Assert production WebSocket defaults are connect `60_000` and idle/send `360_000`. - -- [ ] **Step 2: Run RED tests** - -```bash -bun test test/plugin/codex.test.ts test/plugin/openai-ws.test.ts -``` - -- [ ] **Step 3: Extend plugin options** - -```ts -interface CodexAuthPluginOptions { - issuer?: string - codexApiEndpoint?: string - experimentalWebSockets?: boolean - httpHeaderTimeout?: number - httpChunkTimeout?: number - websocketConnectTimeout?: number - websocketIdleTimeout?: number -} -``` - -- [ ] **Step 4: Configure WebSocket and HTTP paths** - -Create the pool with Codex defaults: - -```ts -OpenAIWebSocketPool.createWebSocketFetch({ - httpFetch: fetch, - connectTimeout: options.websocketConnectTimeout ?? CODEX_HEADER_TIMEOUT, - idleTimeout: options.websocketIdleTimeout ?? CODEX_CHUNK_TIMEOUT, -}) -``` - -Replace the direct HTTP call with `fetchCodexHTTP` using the HTTP option values. Return `headerTimeout: false` and `chunkTimeout: false` from the OAuth loader so the generic 10-second OpenAI header timer cannot terminate the whole retry sequence. API-key OpenAI behavior stays unchanged. - -- [ ] **Step 5: Run integration tests and typecheck** - -```bash -bun test test/plugin/codex-http.test.ts test/plugin/codex.test.ts test/plugin/openai-ws.test.ts test/provider/header-timeout.test.ts -bun typecheck -``` - -Expected: PASS; typecheck exits 0. - ---- - -### Task 4: One-time 401 refresh and safe token rotation - -**Files:** -- Modify: `packages/opencode/src/plugin/openai/codex.ts:94-99,333-425` -- Modify: `packages/opencode/test/plugin/codex.test.ts:195-290` - -- [ ] **Step 1: Write failing OAuth tests** - -Add tests proving: - -- Missing/empty `refresh_token` preserves `refresh-old`. -- Concurrent 401 responses cause exactly one token refresh. -- Reissued requests use the new access/account ID and byte-identical request bodies. -- A second 401 is returned without a third request. -- Abort after the first 401 prevents refresh/reissue. -- 403 and 429 remain unchanged. - -- [ ] **Step 2: Run RED OAuth tests** - -```bash -bun test test/plugin/codex.test.ts -``` - -- [ ] **Step 3: Make refresh rotation optional** - -Change `TokenResponse.refresh_token` to optional for refresh responses and persist: - -```ts -refresh: tokens.refresh_token?.trim() || currentAuth.refresh -``` - -Initial authorization must still require a usable refresh token. - -- [ ] **Step 4: Refactor one single-flight refresh owner** - -The refresh helper accepts the access token that failed. Before starting refresh, reload auth; if another caller already installed a different access token, reuse it. Otherwise share `refreshPromise`. This prevents concurrent 401 callers from refreshing twice. - -- [ ] **Step 5: Reissue exactly once** - -```ts -const first = await send(currentAccess) -if (first.status !== 401 || init?.signal?.aborted) return first -await first.body?.cancel() -const refreshed = await refresh(currentAccess) -return send(refreshed.access, refreshed.accountId) -``` - -Do not use recursive calls and do not refresh on 403/429. - -- [ ] **Step 6: Run OAuth and transport tests** - -```bash -bun test test/plugin/codex.test.ts test/plugin/codex-http.test.ts -``` - -Expected: all tests PASS; concurrent 401 calls share one refresh. - ---- - -### Task 5: Write Vietnamese 9Router analysis - -**Files:** -- Create: `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` - -- [ ] **Step 1: Write verified analysis** - -Cover request lifecycle, OAuth/client ID, refresh behavior, account fallback, exact timeout comparison, retry budgets, SSE buffering, terminal handling, lack of true resume, adopted mechanisms, rejected gateway-only mechanisms, changed files, and verification evidence. - -- [ ] **Step 2: Scan placeholders and secrets** - -```bash -rg -n "TBD|TODO|Bearer |sk-|refresh[_-]?token" "/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md" -``` - -Expected: no placeholders or real credentials. Field names may be discussed without values. - ---- - -### Task 6: Final verification and review - -**Files:** -- Review every file changed by Tasks 1-5. - -- [ ] **Step 1: Run focused OpenCode tests** - -From `packages/opencode`: - -```bash -bun test test/plugin/codex-http.test.ts test/plugin/codex.test.ts test/plugin/openai-ws.test.ts test/provider/header-timeout.test.ts -bun typecheck -``` - -Expected: all tests PASS and typecheck exits 0. - -- [ ] **Step 2: Inspect complete diff** - -```bash -git status --short -git diff --check -git diff -- packages/opencode/src/plugin/openai packages/opencode/test/plugin packages/opencode/test/provider docs/superpowers -git diff --no-index /dev/null docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md -git diff --no-index /dev/null docs/superpowers/plans/2026-07-17-codex-connection-reliability.md -``` - -Expected: no whitespace errors, secrets, account/endpoint fallback, or terminal-protocol changes. The two `--no-index` commands may exit 1 because they intentionally display new untracked files; inspect their content rather than treating exit 1 as a failure. Inspect `/home/dunghd/DEVELOPMENT/Research/docs/9router-codex-connection-stability.md` separately because it is outside the repository. - -- [ ] **Step 3: Smoke-test connection stability** - -Use the already-working OpenAI OAuth model selection. Run one normal response and one tool call. Confirm both complete once without connection interruption, duplicated text, or duplicated tool execution. Do not change or retest model lowering/service-tier behavior. - -- [ ] **Step 4: Independent review** - -Review for retry after exposed output, timer/body leaks, lost HTTP semantics, duplicate tool events, API-key timeout regressions, and terminal behavior changes. - -- [ ] **Step 5: Report evidence** - -Report changed files, exact tests/typechecks, smoke-test result, and limitations. Do not commit or push unless explicitly requested. diff --git a/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md b/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md deleted file mode 100644 index ee7ba5c07649..000000000000 --- a/docs/superpowers/specs/2026-07-17-codex-connection-reliability-design.md +++ /dev/null @@ -1,202 +0,0 @@ -# Codex Connection Reliability Design - -**Date:** 2026-07-17 - -**Target:** `packages/opencode/src/plugin/openai` in `original-opencode` - -**Reference:** `/home/dunghd/DEVELOPMENT/Research/9router-master` - -## Objective - -Improve ChatGPT Codex OAuth connection reliability without changing OpenCode's existing Responses protocol, terminal-event behavior, session semantics, account selection, or endpoint routing. - -Transparent transport retries are allowed only before any upstream event is exposed to the existing parser/session layer. After an event is exposed, the existing OpenCode stream and session retry behavior remains authoritative. - -## Included behavior - -- HTTP response-header timeout: 60 seconds per attempt. -- HTTP inter-chunk stall timeout: 360 seconds, reset for each raw chunk. -- WebSocket connect timeout: 60 seconds. -- WebSocket idle/send timeout: 360 seconds. -- Pre-output retries for network/connect failures and HTTP 502, 503, and 504. -- Pre-output retry when the first parsed SSE event is a structured Codex overload error. -- One forced OAuth refresh and request reissue after HTTP 401. -- Preserve the existing refresh token when a refresh response omits a replacement. -- Abort-aware timeout and retry backoff. -- Focused tests for timing, SSE framing, cancellation, error preservation, terminal behavior, and tool-call uniqueness. - -## Excluded behavior - -- Account rotation, account cooldown, or quota-based account selection. -- Alternative endpoint or base-URL fallback. -- Model-combo fallback. -- Proxy-style request translation or broad body mutation. -- Synthetic `response.failed`, response IDs, or additional `[DONE]` frames. -- Changes to existing terminal-event handling. -- Transparent retry after any event has been exposed. -- Generic timeout changes for normal OpenAI API-key traffic. - -## Reliability findings from 9Router - -9Router's continuity comes from layered recovery: - -1. It allows 60 seconds for upstream response headers. -2. It allows 360 seconds between raw stream chunks and resets the watchdog on upstream activity. -3. It retries transient network and 502/503/504 failures before returning a response downstream. -4. It detects Codex overload failures hidden in an HTTP 200 SSE response and retries at the beginning of the stream. -5. It supports proactive and reactive credential refresh. -6. It also uses account and endpoint failover, which are gateway responsibilities and are excluded from this OpenCode change. - -9Router does not resume a stream after partial output. OpenCode will preserve the same safety boundary: retry before output, fail normally after output. - -## Existing OpenCode behavior to preserve - -- OpenAI provider HTTP header timeout currently defaults to 10 seconds. -- Codex WebSocket connection timeout currently defaults to 15 seconds. -- Codex WebSocket idle timeout currently defaults to 300 seconds. -- The WebSocket pool already owns session affinity, connection aging, busy-lane HTTP fallback, cancellation, cleanup, and sticky HTTP fallback after repeated socket failures. -- The Responses parser and WebSocket bridge already handle `response.completed`, `response.done`, `response.failed`, `response.incomplete`, and `[DONE]`. -- Session retry already handles retryable failures after transport errors reach the session layer. - -The new timeouts apply only to ChatGPT OAuth/Codex traffic. They must not change ordinary OpenAI API-key behavior. - -## Transport architecture - -The Codex OAuth fetch boundary remains the integration point: - -```text -request - -> OAuth refresh/header construction - -> Codex reliability transport - -> connect/header timeout - -> status/network retry - -> first-SSE-event gate - -> inter-chunk watchdog - -> existing Responses parser - -> existing session processing -``` - -The reliability transport must not own semantic parsing beyond identifying the first complete SSE event and a structured Codex error code. It must return accepted bytes unchanged. - -## Retry policy - -Retry counts are retries after the initial attempt. - -| Failure | Retries | Delay | -|---|---:|---:| -| Network/connect failure | 3 | 3 seconds | -| HTTP 502 | 3 | 3 seconds | -| HTTP 503 | 3 | 2 seconds | -| HTTP 504 | 2 | 3 seconds | -| First SSE event: `server_is_overloaded` | 3 | 2 seconds | -| First SSE event: `service_unavailable_error` | 3 | 2 seconds | -| HTTP 401 | One forced refresh and one reissue | No delay | -| HTTP 403 | No transport retry | Preserve response | -| HTTP 429 | No transport retry | Preserve response and `Retry-After` | - -Each status has an independent retry budget. Backoff must be cancellable; an aborted request must not start another attempt. - -The transport must not retry when: - -- a valid SSE event has already been returned to the caller; -- the overload text occurs in ordinary model output rather than a structured error event; -- the client has aborted; -- the relevant retry budget is exhausted. - -## SSE first-event gate - -The first-event gate must: - -- support LF and CRLF SSE separators; -- support separators split across chunks; -- preserve raw bytes exactly; -- handle UTF-8 code points split across chunks; -- inspect only the first complete SSE event; -- reject a first event larger than 64 KiB as a bounded protocol error; -- recognize retryable codes only in structured error events; -- return immediately after accepting the first event instead of buffering the rest of the stream; -- expose only the final accepted attempt; -- preserve the final upstream error event when retries are exhausted. - -Non-SSE and non-2xx responses retain their original status, body, and headers. A malformed 2xx streaming response is represented as a status-200 `Response` whose body stream fails with a typed stream error. Diagnostics must include its content type and a bounded body preview. Non-2xx responses are never rewritten this way. - -## Timeout policy - -### HTTP - -- Header timeout: 60 seconds for each attempt. -- Inter-chunk timeout: 360 seconds after the response stream begins. -- The inter-chunk timer resets on each raw byte chunk, not translated output. -- Header and chunk timeouts combine with the caller's abort signal. - -### WebSocket - -- Connect timeout: 60 seconds. -- Idle/send timeout: 360 seconds. -- Existing pool retry/fallback and terminal behavior remain unchanged. - -Timeout values are injectable through Codex plugin options for deterministic tests. Production defaults remain 60/360 seconds. - -## OAuth recovery - -The existing loader-local single-flight refresh remains the sole refresh owner. - -- Normal requests refresh when local credentials have expired. -- An HTTP 401 forces one refresh and one request reissue. -- Concurrent callers share one in-flight refresh. -- A second 401 is returned without another refresh. -- A missing, empty, or malformed replacement refresh token must not erase the current valid refresh token. -- The reissued request retains the original method and body while rebuilding authorization/account headers from refreshed credentials. -- Abort before or during refresh prevents reissue. - -HTTP 403 is not automatically refreshed because it can represent a real permission or account-scope failure. - -## Terminal and session behavior - -No terminal-event behavior changes are included. Existing OpenCode handling of completion, failure, incomplete responses, `response.done`, and `[DONE]` remains intact. - -No transport retry occurs after partial output. Such failures continue through the existing stream error and session retry paths. This avoids duplicated text, reasoning, persisted events, and tool execution. - -## Test strategy - -Focused tests must prove: - -1. Network, 502, 503, and 504 retries use the specified budgets and delays. -2. Abort during connect, first-event reading, or backoff prevents later attempts. -3. Codex HTTP and WebSocket defaults are 60/360 seconds through injectable test options. -4. Every raw chunk resets the inter-chunk watchdog. -5. LF, CRLF, split separators, and split UTF-8 are handled correctly. -6. Structured overload first events retry; matching text content does not. -7. An accepted first event is replayed byte-for-byte exactly once. -8. The accepted stream is returned without buffering its remainder. -9. Exhausted retries expose only the final accepted error stream. -10. HTTP 401 refreshes and reissues once; concurrent requests use one refresh. -11. HTTP 403 and 429 preserve status, body, and headers. -12. Omitted refresh-token rotation preserves the previous token. -13. Existing terminal-event tests remain unchanged and pass. -14. Tool calls are emitted and executed only once. - -Run targeted plugin tests and package typecheck from `packages/opencode`, then inspect the final diff. If protocol code is unchanged, the existing protocol suite is a regression check rather than an implementation target. - -## Documentation deliverable - -Write a Vietnamese engineering report under `/home/dunghd/DEVELOPMENT/Research/docs/` containing: - -- 9Router's complete Codex request lifecycle; -- exact 60/360-second timeout behavior; -- status/network/SSE retry layers; -- OAuth refresh and account failover mechanisms; -- stream parsing and terminal handling; -- mechanisms adopted by OpenCode; -- mechanisms deliberately rejected and why; -- changed OpenCode files and verification evidence. - -## Success criteria - -- Slow Codex connection establishment is not aborted at OpenCode's previous 10/15-second thresholds. -- A healthy but temporarily quiet stream may remain idle for up to 360 seconds between raw chunks. -- Transient pre-output network, gateway, and structured overload failures recover transparently within bounded retry budgets. -- Authentication can recover once from server-side token invalidation without losing the refresh token. -- Cancellation remains immediate. -- Existing terminal semantics and session behavior do not change. -- No retry path can duplicate visible output or tool execution.