diff --git a/devlog/_plan/260829_bugpr_lane_h_residual_issues/150_issue_2887_pool_401_refresh.md b/devlog/_plan/260829_bugpr_lane_h_residual_issues/150_issue_2887_pool_401_refresh.md new file mode 100644 index 0000000000..66d9a7ab69 --- /dev/null +++ b/devlog/_plan/260829_bugpr_lane_h_residual_issues/150_issue_2887_pool_401_refresh.md @@ -0,0 +1,252 @@ +# 150 — issue #2887: an ordinary stored Codex pool account is quarantined on its first Responses 401 + +## What the reporter saw + +A stored Codex pool credential with a time-valid access token and a usable refresh +token receives one pre-stream `401` from Responses and is immediately marked +`needsReauth` with its affinity cleared. The refresh endpoint is never called. + +## The path, from source + +Ordinary stored credentials and native `__main__` are deliberately different auth +context variants: + +- ordinary stored: `kind: "pool"`, carrying the stored record's credential + `generation` — `src/codex/auth-context.ts:645-659` +- native main: `kind: "main-pool"`, with no stored-record generation — + `src/codex/auth-context.ts:607-642` + +There are exactly three kinds — `main`, `pool`, `main-pool` — and every configured +non-main stored account is `pool`, including exact selectors and accounts serving +Daybreak models. There is no separate reserve or WHAM variant, so `pool` is the whole +blast radius. + +A time-valid ordinary token never refreshes: `getValidCodexToken()` returns as soon +as `expiresAt > now + 60s` (`src/codex/account-store.ts:402-410`). That is correct on +the happy path and is the reason the `401` arrives holding a token the store still +considers good. + +The recovery that should follow is gated on the wrong discriminant. The pre-stream +`401` refresh-and-replay loop admits only `main-pool` +(`src/server/responses/core.ts:3815-3823`), and its helper independently rejects any +other context (`:1747-1760`). The generic OAuth replay cannot pick it up either — that +branch is limited to xAI, GitHub Copilot, and Kiro (`:3020-3024`). `/v1/responses/compact` +carries the identical gate (`src/server/responses/compact.ts:679-686`). + +So the `401` falls through to terminal handling, is classified `credential` +(`src/codex/routing.ts:349-367`), and that branch marks reauth and removes every +affinity entry for the account (`:2146-2157`). + +`/v1/chat/completions` bridges into `handleResponses` +(`src/server/chat-completions.ts:242`), so fixing core covers it too. + +One correction to the report's wording: `needsReauth` is a process-local `Set` +(`src/codex/account-runtime-state.ts:3-10`), not a persisted credential-store field. +The account recovers on restart. That makes the defect less severe than "stored account +corrupted" and no less real — inside a running proxy the account is out of rotation and +its affinity is gone. + +## What gets built + +The machinery exists; ordinary pool has more of it than main does. Refresh-grant keyed +in-process flights (`src/codex/account-store.ts:288-297`, `:413-439`), a cross-process +file lock with locked re-read (`:448-475`), and generation-CAS persistence (`:521-530`). +The missing piece is an entrypoint that bypasses the freshness shortcut for exactly one +rejected generation. + +### 1. A fenced forced-refresh entrypoint — `src/codex/account-store.ts` + +Takes `accountId`, the rejected credential generation, the rejected access token, and the +caller's abort signal. + +The fence is not a single check at the entrance. Flights are keyed by **refresh-grant +fingerprint**, not by account or generation (`:413`), so a forced caller can join a flight +started by an ordinary refresh, or by another account sharing the grant. The generation +must be re-checked in three places: at entry, in the joined-flight branch before the CAS +write (`:419-439`), and again after the file-lock re-read (`:448-455`). If the stored +generation is no longer the rejected one, someone already replaced the credential — return +what is stored and perform no refresh and no bump. + +`findFreshCredentialForGrant()` (`:375-387`) needs one extra condition. It can return +another alias's still-fresh copy of the **same** access token that just got rejected, which +would bump the generation and replay with the identical bearer — a guaranteed second `401` +dressed up as recovery. The rejected token is therefore an explicit input, and a candidate +equal to it does not satisfy a forced refresh. + +### 2. Dispatch on both endpoints — `core.ts`, `compact.ts` + +Widen the existing `401` branch to `pool` and route it to the new entrypoint. One +request-local replay guard; a second `401` falls through to terminal handling. The replay +reuses the same account — alternate-account selection stays out of the first replay or +fixed-account and pin semantics change. + +Core has a hole compact does not: when the main refresh fails it returns the `401` +response immediately without recording an outcome (`core.ts:3830`), whereas compact records +it (`compact.ts:694`). With no second upstream `401` there is nothing to quarantine on, so +an account whose grant is genuinely dead stays selectable and every request repeats the +same doomed refresh. Core must record a **terminal** refresh failure. + +### 3. Terminal versus retryable refresh failure + +The first draft of this plan asserted `:244-275` already classifies transient refresh +errors. That is wrong: those lines define generation-conflict, lock-timeout, busy, and +stale errors only. A raw network failure or timeout is untyped, and a token-endpoint 5xx +becomes `TokenRefreshError("unknown")` (`:496-506`). Treating "unknown" as terminal +rebuilds this exact bug behind a new door — an upstream blip would quarantine a healthy +account. + +Only `revoked` and `expired` are terminal. Everything else — `unknown`, network +failure, abort, `CodexCredentialRefreshBusyError`, `CodexCredentialRefreshStaleError`, +`CodexCredentialRefreshLockTimeoutError`, `CodexCredentialGenerationConflictError` — is +transient: surface an error to the client, quarantine nothing. + +### 4. Fence the quarantine — `src/codex/routing.ts` + +Add a credential-generation field to `CodexUpstreamOutcomeMeta` and require +`isCodexAccountGenerationLive()` before the `credential` branch quarantines or clears +affinity. This must be a **new** field: the existing `writerGeneration` (`:232`) is the +config-store generation, an unrelated counter. + +The field is optional and absent means historical behavior, so the sidecar recorders that +also report raw pool status (`src/providers/openai-sidecar.ts:133`, `src/server/search.ts:165`, +`src/server/images.ts:514`, `src/server/live.ts:657`) keep working exactly as today. Their +lack of a fence is pre-existing and is recorded as residual below, not silently adopted. + +### 5. Hand the affinity generation forward — `src/codex/routing.ts` + +The first draft claimed affinity survives. It provably does not. An affinity entry stores +the generation it was bound under (`:963-981`) and `isThreadAffinityGenerationLive()` +demands exact equality (`:921-923`). A successful forced refresh CAS-writes generation +`G+1`, so the entry the replay just "preserved" is dead on the very next request, which +deletes it at `:1849-1851`. Not quarantining is not the same as keeping affinity. + +The fix is an explicit same-lineage handoff, and the codebase already has the exact test +for "same lineage": a refresh-owned bump preserves `replacedAt` (`account-store.ts:213`) +while an external replacement stamps a fresh one (`:142`). +`settleCodexQuotaRecoveryProbe()` uses precisely that distinction to accept a `+1` +transition (`routing.ts:564-576`). The affinity handoff advances an entry from `G` to +`G+1` under the same conditions: the account matches, the transition is exactly `+1`, and +`replacedAt` is unchanged. + +## Verification + +Endpoint coverage goes beside the existing main-pool cases in +`tests/responses-native-main-refresh.test.ts:135-161`, whose fixture has no ordinary pool +accounts at all (`:17-31`) — which is why this shipped. + +The assertion is the wrong behavior, not a value comparison. A first ordinary-pool `401` +today produces one upstream send, zero token-endpoint calls, a `401` at the client, +`needsReauth` set, and affinity removed. + +Named mutations, each of which must turn a specific test red: + +1. Restore `authCtx.kind === "main-pool"` on either endpoint → that endpoint's ordinary-pool + case fails with the signature above. +2. Drop the rejected-token condition from the same-grant reuse path → the replay sends the + identical bearer and the test sees two `401`s instead of a `200`. +3. Classify `TokenRefreshError("unknown")` as terminal → the transient-failure case + quarantines a healthy account. +4. Delete the affinity handoff → the **next** request after a successful replay finds a dead + entry and re-selects, which is why the test must issue a second request rather than + asserting on the entry at replay time. +5. Remove the generation fence from the `credential` branch → a stale `401` carrying a + superseded generation quarantines the replacement. + +Store-level concurrency goes near `tests/codex-account-store.test.ts:343-424`: a forced +caller joining an **ordinary** flight for the same grant (not merely two forced callers), +a same-grant alias holding the rejected token, and two concurrent forced refreshes +collapsing to one token call and one generation increment. + +## Residual, carried knowingly + +The sidecar recorders in `openai-sidecar.ts`, `search.ts`, `images.ts`, and `live.ts` +record pool `401`s without a credential-generation fence. That is pre-existing behavior and +unchanged by this work, but a forced refresh makes generation bumps more frequent, so the +window in which a stale sidecar `401` can quarantine a freshly refreshed credential gets +wider. Threading the fence through four more call sites is a separate mechanical change and +does not belong in the same work-phase as the behavioral fix. + +Mid-stream SSE `401`s (`core.ts:1304`, `:4155`) are in scope for the fence but never for +replay: once the stream is committed, a transparent retry would duplicate output the client +has already seen. + + +## Post-implementation review: six defects, all in the fix + +An independent source review of the landed commit returned FAIL and reproduced each +finding with its own probe. Five were in code written for this fix; two of those +existed in `getValidCodexToken` before it and the forced path made them reachable. + +**A joined flight could copy a sibling account's credential.** Flights are keyed by +refresh grant and shared by every account holding it. If the owner's own credential is +externally replaced while it waits for the file lock, the grant-mismatch branch returns +that replacement — and a joiner, checking only its own current grant, would CAS-write +another account's access *and* refresh tokens onto itself. Flight results now carry +`resolvedGrantFingerprint`, tagged with the grant the flight was **opened** for rather +than the rotated one it produced. Tagging the rotated grant instead broke the existing +`same refresh grant joins a live flight` test, which is what caught the distinction. + +**The lineage check was tautological.** Both callers read `replacedAt` after the refresh +and passed it to a function that re-read the same record, so the comparison could not +fail — an external replacement passed it and inherited the rejected credential's +affinity, the exact case the handoff claimed to refuse. Lineage is now proven by the +call that performed the CAS: the store reports `selfRefreshed`, and the handoff only +runs when that is true. The `replacedAt` parameter is gone. + +**A same-bearer refresh neither recovered nor quarantined.** Upstream can rotate only +the refresh grant and return the same access token. The store commits `G+1` regardless, +so quarantining against `G` was silently suppressed by the new fence — the account was +neither replayed nor retired, and the next request repeated the refresh. The refresh +result now reports the generation the credential actually sits at, and both endpoints +fence on that value. + +**An ordinary joiner could bump the generation twice.** With the refresh grant retained, +an ordinary same-account caller joins the forced caller's flight and CAS-writes the +identical credential, moving `G+1` to `G+2` and killing the handoff the owner just +performed. A joiner whose stored credential already equals the flight result now adopts +the stored state instead of rewriting it. + +**The fence covered only two synthetic call sites.** Mid-stream SSE terminals, a +replay's own second 401, and compact's ordinary recorder were all unfenced, so a stale +401 could still retire a replacement — which contradicted what the commit message +claimed. All three now pass `credentialGeneration` for a `pool` context. + +**Bare `invalid_grant` was classified transient.** The parser only looked for +`revoked`, `invalidated`, or `expired` in the description; upstream sends +`invalid_grant` with no description at all, so a genuinely dead grant read as +`unknown` and every request retried it forever. + +### One guard without an isolated regression + +The `resolvedGrantFingerprint` provenance check has no test that fails when only it is +removed: the adopt-stored branch intercepts the same scenario first, and both must be +removed together before the cross-account overwrite reappears. It is kept as +defence-in-depth rather than dropped, because the two guards answer different questions +— one asks whether the credential is the one already stored, the other whether it +belongs to this grant at all — and a future change to either branch would remove the +overlap. This is recorded rather than presented as proven. + + +## Second review round: both residuals closed + +**`invalid_grant` matched too loosely.** The first fix searched the combined +code-plus-description text, so a transient `server_error` whose description merely +mentioned the phrase was classified `revoked` and retired a healthy account — the +same failure mode this whole change exists to remove, reintroduced through the fix +for it. The match is now on the exact OAuth `error` code, with descriptive text +classified separately. + +**The cross-account test did not reach the branch it claimed to test.** It replaced +the owner's credential from inside `fetch`, which runs after the lock body has +already compared grants, so the alias-reuse and CAS paths handled the scenario and +the test passed with the provenance check removed. It now holds the shared grant's +file lock directly, replaces the owner's credential while its flight is parked in the +lock wait, then releases — so the lock body observes a different grant and returns +that replacement, which is the branch under test. The assertions are positive as well +as negative: the owner's replacement must survive intact. + +With that, the provenance guard has the isolated regression it previously lacked. +Removing `resolvedGrantFingerprint === refreshGrantFingerprint` on its own now fails +with the joiner holding `owner-secret` — a real cross-account credential leak, not an +inferred one. + diff --git a/src/codex/account-store.ts b/src/codex/account-store.ts index 32546f91a1..9f5853ddea 100644 --- a/src/codex/account-store.ts +++ b/src/codex/account-store.ts @@ -286,7 +286,25 @@ function withCredentialMutationLockSync(fn: () => T): T { } type CodexTokenResult = { accessToken: string; chatgptAccountId: string; generation: number }; -type CodexRefreshResult = CodexTokenResult & { credential?: CodexAccountCredentials }; +type CodexRefreshResult = CodexTokenResult & { + credential?: CodexAccountCredentials; + /** + * Grant the returned credential actually belongs to. + * + * Flights are keyed by refresh grant and shared across every account holding that + * grant, but a flight can resolve to a credential from a DIFFERENT grant: the + * owner's credential may be externally replaced while it waits for the file lock, + * and the grant-mismatch branch then hands back that replacement. A joiner that + * only checks its own current grant would CAS-write another account's access and + * refresh tokens onto itself. The result therefore carries its own provenance. + */ + resolvedGrantFingerprint?: string; + /** + * True when this call's own CAS write produced `generation` — the credential is a + * refresh of the one the caller was holding, not somebody else's replacement. + */ + selfRefreshed?: boolean; +}; const MAX_CODEX_REFRESH_FLIGHTS = 32; const CODEX_REFRESH_FLIGHT_STALE_MS = 120_000; interface RefreshFlight { @@ -375,12 +393,17 @@ export async function withCodexRefreshFileLock(lockKey: string, signal: Abort function findFreshCredentialForGrant( refreshGrantFingerprint: string, excludeId: string, + rejectedAccessToken?: string, ): CodexAccountCredentials | null { const now = Date.now(); const records = loadCodexAccountRecordStore(); for (const [candidateId, candidate] of Object.entries(records)) { if (candidateId === excludeId || candidate.deletedAt != null || !candidate.credential) continue; if (recordGrantFingerprint(candidate) !== refreshGrantFingerprint) continue; + // A sibling alias can hold a still-unexpired copy of the exact token upstream + // just rejected. Reusing it would bump the generation and replay the identical + // bearer — a second 401 dressed up as recovery. + if (rejectedAccessToken !== undefined && candidate.credential.accessToken === rejectedAccessToken) continue; if (candidate.credential.expiresAt > now + REFRESH_SKEW_MS) return candidate.credential; } return null; @@ -399,14 +422,78 @@ async function notePlanFromRefreshedAccessToken( } } +/** + * A forced refresh raised by a rejected bearer. Carries the generation the 401 was + * observed under so a credential someone else already replaced is never refreshed + * again, and the rejected token so a sibling alias holding that same token cannot + * satisfy the refresh. + */ +type ForcedRefreshFence = { rejectedGeneration: number; rejectedAccessToken: string }; + +/** True once the stored credential has moved off the generation the 401 belongs to. */ +function forcedFenceSuperseded(recordGeneration: number, forced: ForcedRefreshFence | undefined): boolean { + return forced !== undefined && recordGeneration !== forced.rejectedGeneration; +} + +/** + * Refresh a stored pool credential that upstream rejected with a 401, even though its + * `expiresAt` still looks valid. Ordinary callers must keep using + * {@link getValidCodexToken}: only a proven rejection justifies spending a refresh. + * + * `rotated` is false when the resolved token is byte-identical to the rejected one, + * which means replaying would earn the same 401 and the caller must not try. That can + * happen even on a SUCCESSFUL token response: upstream may rotate the refresh grant + * while returning the same access token. The generation has moved by then, so + * `generation` reports where the credential actually is — a caller that quarantines + * on `rotated === false` must fence on the returned value, not on the one it rejected. + */ +export async function forceRefreshCodexPoolToken( + id: string, + options: { rejectedGeneration: number; rejectedAccessToken: string; signal?: AbortSignal }, +): Promise { + const result = await resolveCodexToken( + id, + { rejectedGeneration: options.rejectedGeneration, rejectedAccessToken: options.rejectedAccessToken }, + options.signal, + ); + return { + accessToken: result.accessToken, + chatgptAccountId: result.chatgptAccountId, + generation: result.generation, + rotated: result.accessToken !== options.rejectedAccessToken, + // Only a CAS this call performed itself proves the new credential descends from the + // rejected one; anything else is somebody else's replacement and must not be treated + // as this request's own lineage. + selfRefreshed: result.selfRefreshed === true, + }; +} + export async function getValidCodexToken(id: string): Promise { + const result = await resolveCodexToken(id); + return { + accessToken: result.accessToken, + chatgptAccountId: result.chatgptAccountId, + generation: result.generation, + }; +} + +async function resolveCodexToken( + id: string, + forced?: ForcedRefreshFence, + callerSignal?: AbortSignal, +): Promise { const record = readCodexAccountRecord(id); const cred = record?.deletedAt == null ? record?.credential : undefined; if (!record || !cred) throw new Error("Codex account credential is unavailable; reauthenticate the account."); const refreshGrantFingerprint = recordGrantFingerprint(record); if (!refreshGrantFingerprint) throw new Error("Codex account credential is unavailable; reauthenticate the account."); - if (cred.expiresAt > Date.now() + REFRESH_SKEW_MS) { + // The freshness shortcut is exactly what makes a 401 on a time-valid token + // unrecoverable, so a forced caller skips it — but only while the stored credential + // is still the one that was rejected. Once it has been replaced, the shortcut is + // correct again and refreshing would burn a rotation for nothing. + const forcedTargetsStoredCredential = forced !== undefined && !forcedFenceSuperseded(record.generation, forced); + if (cred.expiresAt > Date.now() + REFRESH_SKEW_MS && !forcedTargetsStoredCredential) { return { accessToken: cred.accessToken, chatgptAccountId: cred.chatgptAccountId, generation: record.generation }; } @@ -419,10 +506,51 @@ export async function getValidCodexToken(id: string): Promise const refreshed = await existing.promise; const current = readCodexAccountRecord(id); const currentCred = current?.deletedAt == null ? current?.credential : undefined; + // The flight owner already committed this credential, and it is the one stored + // for this account: adopt the stored state instead of CAS-writing the identical + // bytes, which would bump the generation a second time and invalidate the + // affinity handoff the owner performed against generation+1. + if (current && currentCred && refreshed.credential + && currentCred.accessToken === refreshed.credential.accessToken + && currentCred.refreshToken === refreshed.credential.refreshToken) { + // A forced caller must still not accept the bearer upstream rejected. + if (!(forced !== undefined && currentCred.accessToken === forced.rejectedAccessToken)) { + return { + accessToken: currentCred.accessToken, + chatgptAccountId: currentCred.chatgptAccountId, + generation: current.generation, + }; + } + } + // Flights are keyed by refresh grant, not by account or generation, so this + // credential may belong to a flight started for a different generation of the + // same grant. Writing it onto a replacement would undo that replacement. + // + // The rejected-token test comes FIRST: a joined flight that resolved back to the + // bearer upstream rejected proves nothing, and reporting the replacement as + // "superseded" would hand the caller a token it must not replay. + if ( + current && currentCred + && forcedFenceSuperseded(current.generation, forced) + && !(forced !== undefined && currentCred.accessToken === forced.rejectedAccessToken) + ) { + return { + accessToken: currentCred.accessToken, + chatgptAccountId: currentCred.chatgptAccountId, + generation: current.generation, + }; + } if ( current && currentCred && refreshed.credential && + // Provenance: a flight can resolve to a credential from a DIFFERENT grant when + // the owner's own credential was replaced while it waited for the lock. Adopting + // that would copy another account's access and refresh tokens onto this one. + refreshed.resolvedGrantFingerprint === refreshGrantFingerprint && + // A joined flight that resolved to the rejected token proves nothing; fall + // through and open a real refresh instead of bumping the generation. + !(forced !== undefined && refreshed.credential.accessToken === forced.rejectedAccessToken) && recordGrantFingerprint(current) === refreshGrantFingerprint ) { if (!saveCodexAccountCredentialIfGeneration(id, current.generation, refreshed.credential)) { @@ -434,16 +562,22 @@ export async function getValidCodexToken(id: string): Promise accessToken: refreshed.credential.accessToken, chatgptAccountId: refreshed.credential.chatgptAccountId, generation, + // This joiner performed its own CAS onto its own record, so the resulting + // generation is its own lineage even though another caller drove the fetch. + selfRefreshed: true, + resolvedGrantFingerprint: refreshGrantFingerprint, }; } - return getValidCodexToken(id); + return resolveCodexToken(id, forced, callerSignal); } } if (refreshLocks.size >= MAX_CODEX_REFRESH_FLIGHTS) throw new CodexCredentialRefreshBusyError(); const abort = new AbortController(); - const signal = AbortSignal.any([abort.signal, AbortSignal.timeout(30_000)]); + const signal = AbortSignal.any( + callerSignal ? [abort.signal, AbortSignal.timeout(30_000), callerSignal] : [abort.signal, AbortSignal.timeout(30_000)], + ); let flight!: RefreshFlight; const refreshPromise = withCodexRefreshFileLock(refreshGrantFingerprint, signal, async (): Promise => { const current = readCodexAccountRecord(id); @@ -459,19 +593,34 @@ export async function getValidCodexToken(id: string): Promise chatgptAccountId: lockedCred.chatgptAccountId, generation: startGeneration, credential: lockedCred, + // This credential belongs to a DIFFERENT grant than the flight was opened + // for. Tagging it keeps a joiner from adopting it as its own. + ...(lockedRefreshGrantFingerprint !== undefined + ? { resolvedGrantFingerprint: lockedRefreshGrantFingerprint } + : {}), }; } throw new CodexCredentialGenerationConflictError(); } - if (lockedCred.expiresAt > Date.now() + REFRESH_SKEW_MS) { + // Third fence point: waiting for the lock can take long enough for another + // writer to replace the credential. Under the lock the stored generation is + // authoritative, so a superseded forced refresh stops here rather than + // spending a rotation on a credential nobody rejected. + const forcedStillTargetsStored = forced !== undefined && !forcedFenceSuperseded(startGeneration, forced); + if (lockedCred.expiresAt > Date.now() + REFRESH_SKEW_MS && !forcedStillTargetsStored) { return { accessToken: lockedCred.accessToken, chatgptAccountId: lockedCred.chatgptAccountId, generation: startGeneration, credential: lockedCred, + resolvedGrantFingerprint: refreshGrantFingerprint, }; } - const sameGrantFreshCredential = findFreshCredentialForGrant(refreshGrantFingerprint, id); + const sameGrantFreshCredential = findFreshCredentialForGrant( + refreshGrantFingerprint, + id, + forced?.rejectedAccessToken, + ); if (sameGrantFreshCredential) { if (!saveCodexAccountCredentialIfGeneration(id, startGeneration, sameGrantFreshCredential)) { throw new CodexCredentialGenerationConflictError(); @@ -481,6 +630,8 @@ export async function getValidCodexToken(id: string): Promise chatgptAccountId: sameGrantFreshCredential.chatgptAccountId, generation: startGeneration + 1, credential: sameGrantFreshCredential, + resolvedGrantFingerprint: refreshGrantFingerprint, + selfRefreshed: true, }; } const res = await fetch(CHATGPT_TOKEN_URL, { @@ -496,11 +647,22 @@ export async function getValidCodexToken(id: string): Promise if (!res.ok) { const errText = await res.text().catch(() => ""); let errDesc: string; + let errCodeExact: string | undefined; try { const parsed = JSON.parse(errText) as { error?: string; error_description?: string }; + errCodeExact = typeof parsed.error === "string" ? parsed.error.trim() : undefined; errDesc = [parsed.error, parsed.error_description].filter(Boolean).join(": ") || `HTTP ${res.status}`; } catch { errDesc = `HTTP ${res.status}`; } - const reason = errDesc.includes("invalidated") || errDesc.includes("revoked") ? "revoked" as const + // `invalid_grant` is the standard OAuth code for a refresh token that is no longer + // usable, and upstream sends it bare with no description. Without it here the dead + // grant is classified "unknown", which callers treat as transient — so the account + // is never retired and every request repeats the same doomed refresh (#2887). + // + // Matched on the exact `error` CODE, not anywhere in the combined text: a transient + // `server_error` whose description happens to mention invalid_grant would otherwise + // retire a healthy account, which is the failure this whole change exists to remove. + const reason = errCodeExact === "invalid_grant" + || errDesc.includes("invalidated") || errDesc.includes("revoked") ? "revoked" as const : errDesc.includes("expired") ? "expired" as const : "unknown" as const; throw new TokenRefreshError(reason, `Codex token refresh failed (${reason}); reauthenticate the account.`); @@ -527,7 +689,17 @@ export async function getValidCodexToken(id: string): Promise if (!saveCodexAccountCredentialIfGeneration(id, startGeneration, updated)) { throw new CodexCredentialGenerationConflictError(); } - return { accessToken: updated.accessToken, chatgptAccountId: updated.chatgptAccountId, generation: startGeneration + 1, credential: updated }; + return { + accessToken: updated.accessToken, + chatgptAccountId: updated.chatgptAccountId, + generation: startGeneration + 1, + credential: updated, + // The grant this flight was OPENED for, not the rotated one it produced. Joiners + // are waiting on that key, and a successful refresh normally rotates the refresh + // token — tagging the new grant would make every legitimate joiner look foreign. + resolvedGrantFingerprint: refreshGrantFingerprint, + selfRefreshed: true, + }; }).finally(() => { if (refreshLocks.get(refreshGrantFingerprint) === flight) refreshLocks.delete(refreshGrantFingerprint); }); @@ -540,5 +712,12 @@ export async function getValidCodexToken(id: string): Promise accessToken: result.accessToken, chatgptAccountId: result.chatgptAccountId, generation: result.generation, + // Carry the flight's provenance out to the caller: the owner is the one whose CAS + // produced this generation, and a forced caller needs that to know whether the new + // credential descends from the one it was holding. + ...(result.selfRefreshed !== undefined ? { selfRefreshed: result.selfRefreshed } : {}), + ...(result.resolvedGrantFingerprint !== undefined + ? { resolvedGrantFingerprint: result.resolvedGrantFingerprint } + : {}), }; } diff --git a/src/codex/routing.ts b/src/codex/routing.ts index d38a40645f..24fce08cfa 100644 --- a/src/codex/routing.ts +++ b/src/codex/routing.ts @@ -231,6 +231,15 @@ export type CodexUpstreamOutcomeMeta = { promoteAccountId?: string; /** Generation captured when this routed account was selected. */ writerGeneration?: number; + /** + * Credential generation this request's bearer was read at. Distinct from + * `writerGeneration`, which tracks the config store. + * + * A 401 that arrives after the credential was already replaced is evidence about a + * token nobody is using any more, so it must not quarantine the replacement. Absent + * means the caller cannot supply lineage and the historical unfenced handling stands. + */ + credentialGeneration?: number; }; function hasConfiguredPoolAccount( @@ -923,6 +932,55 @@ function isThreadAffinityGenerationLive(entry: ThreadAffinityEntry): boolean { return isCodexAccountGenerationLive(entry.accountId, entry.generation); } +/** Generations this account's affinity entries are bound at. Test observability only. */ +export function debugCodexAffinityGenerations(accountId: string): number[] { + const generations: number[] = []; + for (const affinities of threadAccountMap.values()) { + for (const entry of affinities.values()) { + if (entry.accountId === accountId) generations.push(entry.generation); + } + } + return generations; +} + +/** + * Advance this account's affinity entries from the generation a rejected credential + * was bound under to the generation its own refresh produced. + * + * A 401 refresh-and-replay keeps the request on the same account, but the CAS write + * moves the credential from G to G+1, and {@link isThreadAffinityGenerationLive} + * demands exact equality — so without this the entry the replay just preserved is + * dead on the next request. Not quarantining an account is not the same as keeping + * its affinity. + * + * Lineage is proven by the CALLER, which must pass only a generation its own refresh + * produced. Re-deriving it here from `replacedAt` cannot work: the caller reads that + * field after the refresh and this function would re-read the same record, so the + * comparison is tautological and an external replacement passes it. An external + * replacement must retire the affinity, because that credential may belong to a + * different upstream identity. + */ +export function handOffThreadAffinityGeneration( + accountId: string, + fromGeneration: number, + toGeneration: number, +): boolean { + if (accountId === MAIN_CODEX_ACCOUNT_ID) return false; + if (toGeneration !== fromGeneration + 1) return false; + const record = readCodexAccountRecord(accountId); + if (!record?.credential || record.deletedAt != null) return false; + if (record.generation !== toGeneration) return false; + let handedOff = false; + for (const affinities of threadAccountMap.values()) { + for (const entry of affinities.values()) { + if (entry.accountId !== accountId || entry.generation !== fromGeneration) continue; + entry.generation = toGeneration; + handedOff = true; + } + } + return handedOff; +} + function pruneExpiredThreadAffinities(now: number): void { for (const [threadId, affinities] of threadAccountMap) { for (const [scope, entry] of affinities) { @@ -2146,6 +2204,15 @@ export function recordCodexUpstreamOutcome( if (outcomeClass === "credential") { // 401/403 quarantines the account for reauth. That supersedes quota state // entirely: a cooldown (and any probe lease) on an unusable account is moot. + // Unless the rejected credential is already gone: a stale 401 racing a + // replacement would otherwise take the fresh credential out of rotation and + // sweep affinities that belong to it (#2887). + if ( + meta.credentialGeneration !== undefined + && !isCodexAccountGenerationLive(accountId, meta.credentialGeneration) + ) { + return; + } upstreamHealth.set(accountId, { consecutiveFailures: 1, lastFailureStatus, diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 641e6c6696..6eb6b7754e 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -59,9 +59,15 @@ import { } from "../../codex/main-account"; import { formatCodexProviderForLog, + handOffThreadAffinityGeneration, recordCodexUpstreamOutcome, type CodexUpstreamOutcome, } from "../../codex/routing"; +import { + TokenRefreshError, + forceRefreshCodexPoolToken, + readCodexAccountRecord, +} from "../../codex/account-store"; import { fetchWithResetRetry, fetchWithTransientRetry, @@ -261,6 +267,89 @@ async function refreshNativeMainCompactContext(args: { } } +/** See the core counterpart: only a dead grant is terminal (#2887). */ +function isTerminalCompactPoolRefreshFailure(error: unknown): boolean { + return error instanceof TokenRefreshError && (error.reason === "revoked" || error.reason === "expired"); +} + +/** + * Compact's forced refresh for an ordinary stored pool credential rejected with a + * pre-stream 401. Mirrors {@link refreshNativeMainCompactContext} so the two 401 + * contracts on this endpoint cannot drift. + */ +async function refreshPoolCompactContext(args: { + req: Request; + authCtx: CodexAuthContext & { kind: "pool" }; + provider: OcxProviderConfig; + codexAccountMode?: CodexAccountMode; + substituteMainCredential: boolean; + options: HandleResponsesCompactOptions; +}): Promise< + | { ok: true; authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } + | { ok: false; response: Response; quarantine: boolean; quarantineGeneration?: number } +> { + const { req, authCtx, provider, codexAccountMode, substituteMainCredential, options } = args; + const reauthResponse = () => formatErrorResponse( + 401, + "authentication_error", + "Selected Codex account needs reauthentication", + ); + try { + const refreshed = await forceRefreshCodexPoolToken(authCtx.accountId, { + rejectedGeneration: authCtx.generation, + rejectedAccessToken: authCtx.accessToken, + signal: req.signal, + }); + // See the core counterpart: a successful response can rotate only the refresh grant, + // so the credential may already sit at a later generation than the one we rejected. + if (!refreshed.rotated) { + return { ok: false, quarantine: true, quarantineGeneration: refreshed.generation, response: reauthResponse() }; + } + if (refreshed.selfRefreshed) { + handOffThreadAffinityGeneration(authCtx.accountId, authCtx.generation, refreshed.generation); + } + const refreshedAuthCtx: CodexAuthContext = { + ...authCtx, + accessToken: refreshed.accessToken, + chatgptAccountId: refreshed.chatgptAccountId, + generation: refreshed.generation, + }; + const refreshedProvider = applyCodexAuthContextToProvider( + stripCodexRuntimeProviderFields(provider), + refreshedAuthCtx, + codexAccountMode, + ); + const headers = new Headers({ "content-type": "application/json" }); + const selected = await materializeCodexUpstreamAuthAsync(req.headers, refreshedAuthCtx, { + substituteMainCredential, + signal: req.signal, + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, + }); + for (const name of FORWARD_HEADERS) { + const value = selected.get(name); + if (value) headers.set(name, value); + } + const override = (refreshedProvider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; + if (override) { + headers.set("authorization", `Bearer ${override.accessToken}`); + headers.set("chatgpt-account-id", override.chatgptAccountId); + } + return { ok: true, authCtx: refreshedAuthCtx, provider: refreshedProvider, headers }; + } catch (error) { + if (isTerminalCompactPoolRefreshFailure(error)) { + return { ok: false, quarantine: true, response: reauthResponse() }; + } + const response = formatErrorResponse( + 503, + "server_busy", + "Codex credential refresh did not complete; retry this request", + ); + const headers = new Headers(response.headers); + headers.set("Retry-After", "1"); + return { ok: false, quarantine: false, response: new Response(response.body, { status: response.status, headers }) }; + } +} + /** @@ -599,6 +688,10 @@ export async function handleResponsesCompact( fixedAccount: ctx.fixedAccount, modelId: selectedModelId, probeLeaseId: codexProbeLeaseId(ctx), + // Fence a stored pool account's outcome on the credential the request was + // holding, so a 401 that lost a race with re-authentication cannot retire the + // replacement (#2887). Also covers the replay's own second 401. + ...(ctx.kind === "pool" ? { credentialGeneration: ctx.generation } : {}), probeQuotaScope: codexProbeQuotaScope(ctx), writerGeneration: ctx.kind === "pool" || ctx.kind === "main-pool" ? ctx.writerGeneration @@ -678,20 +771,46 @@ export async function handleResponsesCompact( if ( upstream.status === 401 - && authCtx.kind === "main-pool" + && (authCtx.kind === "main-pool" || authCtx.kind === "pool") && usesCodexForwardPoolAuth(authCtx, compactProvider) && !req.signal.aborted ) { await upstream.body?.cancel().catch(() => undefined); - const replay = await refreshNativeMainCompactContext({ - req, - authCtx, - provider: compactProvider, - codexAccountMode: route.codexAccountMode, - substituteMainCredential, - options, - }); + const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined; + const poolReplay = poolAuthCtx + ? await refreshPoolCompactContext({ + req, + authCtx: poolAuthCtx, + provider: compactProvider, + codexAccountMode: route.codexAccountMode, + substituteMainCredential, + options, + }) + : undefined; + const replay = poolReplay + ?? await refreshNativeMainCompactContext({ + req, + authCtx, + provider: compactProvider, + codexAccountMode: route.codexAccountMode, + substituteMainCredential, + options, + }); if (!replay.ok) { + // A transient refresh failure must not retire the account; only a dead grant + // does, and only while the rejected credential is still the stored one (#2887). + if (poolAuthCtx) { + if (poolReplay && !poolReplay.ok && poolReplay.quarantine) { + recordCodexUpstreamOutcome(config, poolAuthCtx.accountId, 401, { + threadId: poolAuthCtx.affinityKey, + fixedAccount: poolAuthCtx.fixedAccount, + modelId: selectedModelId, + writerGeneration: poolAuthCtx.writerGeneration, + credentialGeneration: poolReplay.quarantineGeneration ?? poolAuthCtx.generation, + }); + } + return replay.response; + } recordCompactPoolOutcome(outcomeCtx, replay.response.status === 401 ? 401 : "connect_neutral"); return replay.response; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 3a9051b7e8..f0c1291c2a 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -168,10 +168,16 @@ import { computeQuotaCooldown, codexQuotaScopeForModel, formatCodexProviderForLog, + handOffThreadAffinityGeneration, previewCodexAccountForRequest, recordCodexUpstreamOutcome, type CodexUpstreamOutcome, } from "../../codex/routing"; +import { + TokenRefreshError, + forceRefreshCodexPoolToken, + readCodexAccountRecord, +} from "../../codex/account-store"; import { codexAuthContextLogLabel } from "../../codex/account-label"; import { applyUpstreamRecoveryInit, @@ -1311,6 +1317,10 @@ export function codexForwardTerminalOutcomeRecorder( probeLeaseId: codexProbeLeaseId(authCtx), probeQuotaScope: codexProbeQuotaScope(authCtx), writerGeneration: authCtx.writerGeneration, + // A mid-stream terminal can carry a semantic 401 long after the credential was + // replaced. It is never replayed — the client already saw output — but it must + // not retire the replacement either (#2887). + ...(authCtx.kind === "pool" ? { credentialGeneration: authCtx.generation } : {}), }); }; } @@ -1744,6 +1754,94 @@ async function resolveResponsesCodexAuth( } } +/** + * Terminal means the grant itself is dead and no retry can help. Everything else — + * an untyped network failure, a token-endpoint 5xx surfacing as `unknown`, an abort, + * refresh capacity, lock contention, a superseded flight — is transient, and treating + * it as terminal would quarantine a healthy account on an upstream blip, which is the + * defect this path exists to fix (#2887). + */ +function isTerminalPoolRefreshFailure(error: unknown): boolean { + return error instanceof TokenRefreshError && (error.reason === "revoked" || error.reason === "expired"); +} + +/** + * One forced refresh and one same-account rebuild for a stored pool credential that + * upstream rejected with a pre-stream 401. `quarantine` distinguishes a dead grant, + * which must retire the account, from a transient failure, which must not. + */ +async function refreshPoolForwardAuth(args: { + req: Request; + route: RouteResult; + authCtx: CodexAuthContext & { kind: "pool" }; + substituteMainCredential: boolean; + options: HandleResponsesOptions; +}): Promise< + | { ok: true; authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } + | { ok: false; response: Response; quarantine: boolean; quarantineGeneration?: number } +> { + const { req, route, authCtx, substituteMainCredential, options } = args; + try { + const refreshed = await forceRefreshCodexPoolToken(authCtx.accountId, { + rejectedGeneration: authCtx.generation, + rejectedAccessToken: authCtx.accessToken, + signal: options.abortSignal, + }); + if (!refreshed.rotated) { + // The store resolved to the same bearer upstream just rejected. Replaying it + // would spend another upstream call to earn the identical 401. Upstream can do + // this on a SUCCESSFUL response by rotating only the refresh grant, so the + // credential generation may already have moved — quarantine has to be fenced on + // where the credential actually is, not on the generation we started from. + return { + ok: false, + quarantine: true, + quarantineGeneration: refreshed.generation, + response: formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"), + }; + } + // Only a CAS this request performed itself proves the new credential descends from + // the rejected one. Somebody else's replacement may be a different identity, and + // its affinity must be retired rather than inherited. + if (refreshed.selfRefreshed) { + handOffThreadAffinityGeneration(authCtx.accountId, authCtx.generation, refreshed.generation); + } + const refreshedAuthCtx: CodexAuthContext = { + ...authCtx, + accessToken: refreshed.accessToken, + chatgptAccountId: refreshed.chatgptAccountId, + generation: refreshed.generation, + }; + const provider = applyCodexAuthContextToProvider( + stripCodexRuntimeProviderFields(route.provider), + refreshedAuthCtx, + route.codexAccountMode, + ); + const headers = await materializeCodexUpstreamAuthAsync(req.headers, refreshedAuthCtx, { + substituteMainCredential, + signal: options.abortSignal, + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, + }); + return { ok: true, authCtx: refreshedAuthCtx, provider, headers }; + } catch (error) { + if (isTerminalPoolRefreshFailure(error)) { + return { + ok: false, + quarantine: true, + response: formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"), + }; + } + const response = formatErrorResponse( + 503, + "server_busy", + "Codex credential refresh did not complete; retry this request", + ); + const headers = new Headers(response.headers); + headers.set("Retry-After", "1"); + return { ok: false, quarantine: false, response: new Response(response.body, { status: response.status, headers }) }; + } +} + async function refreshNativeMainForwardAuth(args: { req: Request; route: RouteResult; @@ -3814,20 +3912,31 @@ async function handleResponsesInner( if ( upstreamResponse.status === 401 - && authCtx.kind === "main-pool" + && (authCtx.kind === "main-pool" || authCtx.kind === "pool") && usesCodexForwardPoolAuth(authCtx, route.provider) && !codexMain401ReplayAttempted ) { codexMain401ReplayAttempted = true; try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already consumed */ } - const replay = await refreshNativeMainForwardAuth({ - req, - route, - authCtx, - substituteMainCredential, - options, - }); + const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined; + const poolReplay = poolAuthCtx + ? await refreshPoolForwardAuth({ req, route, authCtx: poolAuthCtx, substituteMainCredential, options }) + : undefined; + const replay = poolReplay + ?? await refreshNativeMainForwardAuth({ req, route, authCtx, substituteMainCredential, options }); if (!replay.ok) { + // Compact already records this; core historically returned without recording, + // so a dead grant stayed selectable and every request repeated the same doomed + // refresh. Fenced by the generation the 401 belongs to (#2887). + if (poolAuthCtx && poolReplay && !poolReplay.ok && poolReplay.quarantine) { + recordCodexUpstreamOutcome(config, poolAuthCtx.accountId, 401, { + threadId: poolAuthCtx.affinityKey, + fixedAccount: poolAuthCtx.fixedAccount, + modelId: route.modelId, + writerGeneration: poolAuthCtx.writerGeneration, + credentialGeneration: poolReplay.quarantineGeneration ?? poolAuthCtx.generation, + }); + } upstream.abort(); releaseCodexAuthContextProbeLease(authCtx); return replay.response; @@ -4212,6 +4321,9 @@ async function handleResponsesInner( probeLeaseId: codexProbeLeaseId(authCtx), probeQuotaScope: codexProbeQuotaScope(authCtx), writerGeneration: authCtx.writerGeneration, + // Includes a replay's second 401, which is the case that actually retires the + // account — fence it on the credential the request was holding. + ...(authCtx.kind === "pool" ? { credentialGeneration: authCtx.generation } : {}), }); } } diff --git a/tests/codex-account-store.test.ts b/tests/codex-account-store.test.ts index 63f19cf0b3..7dac3e4c7f 100644 --- a/tests/codex-account-store.test.ts +++ b/tests/codex-account-store.test.ts @@ -583,4 +583,351 @@ describe("codex-account-store CRUD", () => { globalThis.fetch = originalFetch; } }); + + test("a forced refresh rotates a time-valid credential that upstream rejected (#2887)", async () => { + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + // Far beyond the refresh skew: getValidCodexToken would return this untouched, which is + // exactly why a 401 on it was unrecoverable. + saveCodexAccountCredential("forced", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("forced")!.generation; + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = (async () => { + calls += 1; + return Response.json({ access_token: "rotated", refresh_token: "grant2", expires_in: 3600 }); + }) as typeof fetch; + + try { + const result = await forceRefreshCodexPoolToken("forced", { + rejectedGeneration: generation, + rejectedAccessToken: "rejected", + }); + expect(calls).toBe(1); + expect(result.accessToken).toBe("rotated"); + expect(result.rotated).toBe(true); + expect(result.generation).toBe(generation + 1); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("a forced refresh whose generation was already superseded spends no rotation (#2887)", async () => { + const { forceRefreshCodexPoolToken, saveCodexAccountCredential, readCodexAccountRecord } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("forced-stale", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const rejectedGeneration = readCodexAccountRecord("forced-stale")!.generation; + // An operator re-authenticated while the request was in flight. + saveCodexAccountCredential("forced-stale", { + accessToken: "replacement", + refreshToken: "grant-new", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = (async () => { + calls += 1; + return Response.json({ access_token: "should-not-happen", expires_in: 3600 }); + }) as typeof fetch; + + try { + const result = await forceRefreshCodexPoolToken("forced-stale", { + rejectedGeneration, + rejectedAccessToken: "rejected", + }); + // The replacement is handed back untouched: no token call, no generation bump. + expect(calls).toBe(0); + expect(result.accessToken).toBe("replacement"); + expect(result.generation).toBe(rejectedGeneration + 1); + expect(readCodexAccountRecord("forced-stale")!.credential!.accessToken).toBe("replacement"); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("concurrent forced refreshes of one rejected generation collapse to a single token call (#2887)", async () => { + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("forced-concurrent", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("forced-concurrent")!.generation; + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = (async () => { + calls += 1; + await new Promise(resolve => setTimeout(resolve, 10)); + return Response.json({ access_token: "rotated", refresh_token: "grant2", expires_in: 3600 }); + }) as typeof fetch; + + try { + const both = await Promise.allSettled([ + forceRefreshCodexPoolToken("forced-concurrent", { rejectedGeneration: generation, rejectedAccessToken: "rejected" }), + forceRefreshCodexPoolToken("forced-concurrent", { rejectedGeneration: generation, rejectedAccessToken: "rejected" }), + ]); + expect(calls).toBe(1); + // One generation increment, not two: a second bump would invalidate the affinity the + // first caller just handed forward. + expect(readCodexAccountRecord("forced-concurrent")!.generation).toBe(generation + 1); + expect(both.some(r => r.status === "fulfilled" && r.value.accessToken === "rotated")).toBe(true); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("a joined flight cannot copy a sibling account's replacement credential (#2887 review)", async () => { + // Flights are keyed by refresh GRANT and shared across every account holding it. If the + // owner's own credential is externally replaced BEFORE it takes the file lock, the + // grant-mismatch branch hands back that replacement. Without provenance on the result, a + // joiner CAS-writes another account's access AND refresh tokens onto itself. + // + // The replacement has to land before the lock body reads the record, which is why it is + // written from the lock-acquisition hook rather than from inside `fetch`: by fetch time + // the grant comparison has already happened and a different branch handles the case. + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + const shared = { refreshToken: "shared-grant", expiresAt: Date.now() + 3600_000, chatgptAccountId: "acc" }; + saveCodexAccountCredential("owner", { ...shared, accessToken: "owner-rejected" }); + saveCodexAccountCredential("joiner", { ...shared, accessToken: "joiner-rejected" }); + const ownerGeneration = readCodexAccountRecord("owner")!.generation; + const joinerGeneration = readCodexAccountRecord("joiner")!.generation; + + const originalFetch = globalThis.fetch; + // Hold the shared grant's file lock so the owner's flight is parked BEFORE its lock body + // reads the record. Replacing the owner's credential now means the lock body observes a + // different grant and returns that replacement, which is the branch under test. + const lockPath = refreshLockPathForToken("shared-grant"); + writeFileSync(lockPath, JSON.stringify({ acquiredAt: Date.now(), pid: process.pid }) + "\n"); + globalThis.fetch = (async () => Response.json({ access_token: "unused", expires_in: 3600 })) as typeof fetch; + + try { + const ownerFlight = forceRefreshCodexPoolToken("owner", { + rejectedGeneration: ownerGeneration, + rejectedAccessToken: "owner-rejected", + }).catch(() => undefined); + // Let the owner reach the lock wait, then re-authenticate it onto a DIFFERENT grant + // and release the lock so its body runs against the replacement. + await new Promise(resolve => setTimeout(resolve, 20)); + saveCodexAccountCredential("owner", { + accessToken: "owner-secret", + refreshToken: "owner-new-grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc-owner", + }); + unlinkSync(lockPath); + + const joiner = await forceRefreshCodexPoolToken("joiner", { + rejectedGeneration: joinerGeneration, + rejectedAccessToken: "joiner-rejected", + }).catch(() => undefined); + await ownerFlight; + + // The joiner must never end up holding the owner's credential, and the owner's own + // replacement must survive untouched. + const joinerRecord = readCodexAccountRecord("joiner"); + expect(joinerRecord?.credential?.accessToken).not.toBe("owner-secret"); + expect(joinerRecord?.credential?.refreshToken).not.toBe("owner-new-grant"); + expect(readCodexAccountRecord("owner")!.credential!.accessToken).toBe("owner-secret"); + expect(joiner?.accessToken).not.toBe("owner-secret"); + } finally { + globalThis.fetch = originalFetch; + if (existsSync(lockPath)) unlinkSync(lockPath); + } + }); + + test("a successful refresh that returns the SAME access token reports rotated=false at its real generation (#2887 review)", async () => { + // Upstream may rotate only the refresh grant. The store commits G+1 either way, so a + // caller that quarantines on rotated===false must fence on the RETURNED generation — + // fencing on the one it rejected silently suppresses its own quarantine. + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("same-bearer", { + accessToken: "still-rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("same-bearer")!.generation; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => Response.json({ + access_token: "still-rejected", + refresh_token: "grant-rotated", + expires_in: 3600, + })) as typeof fetch; + + try { + const result = await forceRefreshCodexPoolToken("same-bearer", { + rejectedGeneration: generation, + rejectedAccessToken: "still-rejected", + }); + expect(result.rotated).toBe(false); + // The generation reported must be where the credential actually is, not where it was. + expect(result.generation).toBe(readCodexAccountRecord("same-bearer")!.generation); + expect(result.generation).toBe(generation + 1); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("an ordinary joiner does not bump the generation a second time (#2887 review)", async () => { + // The forced owner commits G+1 and hands its affinity forward to G+1. An ordinary + // same-account joiner that re-writes the identical credential would move it to G+2 and + // invalidate that handoff. + const { forceRefreshCodexPoolToken, getValidCodexToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("double-bump", { + accessToken: "rejected", + refreshToken: "grant", + // Expired, so the ordinary caller actually joins the flight instead of taking the + // freshness shortcut — that shortcut is why an ordinary caller normally never sees + // a 401-driven refresh at all. + expiresAt: 0, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("double-bump")!.generation; + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = (async () => { + calls += 1; + await new Promise(resolve => setTimeout(resolve, 10)); + // Same refresh grant retained, so an ordinary caller joins this very flight. + return Response.json({ access_token: "rotated", refresh_token: "grant", expires_in: 3600 }); + }) as typeof fetch; + + try { + const forced = forceRefreshCodexPoolToken("double-bump", { + rejectedGeneration: generation, + rejectedAccessToken: "rejected", + }); + await new Promise(resolve => setTimeout(resolve, 2)); + const ordinary = getValidCodexToken("double-bump"); + const [forcedResult, ordinaryResult] = await Promise.all([forced, ordinary]); + + expect(calls).toBe(1); + expect(forcedResult.generation).toBe(generation + 1); + expect(ordinaryResult.generation).toBe(generation + 1); + expect(readCodexAccountRecord("double-bump")!.generation).toBe(generation + 1); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("a bare invalid_grant is terminal, not transient (#2887 review)", async () => { + // Upstream sends invalid_grant with no description. Classified "unknown" it reads as + // transient, so a dead grant is never retired and every request repeats the refresh. + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential, TokenRefreshError } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("dead-grant", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("dead-grant")!.generation; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => Response.json({ error: "invalid_grant" }, { status: 400 })) as typeof fetch; + + try { + await forceRefreshCodexPoolToken("dead-grant", { + rejectedGeneration: generation, + rejectedAccessToken: "rejected", + }); + throw new Error("expected a TokenRefreshError"); + } catch (error) { + expect(error).toBeInstanceOf(TokenRefreshError); + expect((error as InstanceType).reason).toBe("revoked"); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("a replacement landing mid-refresh is not reported as this call's own lineage (#2887 review)", async () => { + // `selfRefreshed` is what gates the affinity handoff. An external replacement must not + // set it: that credential may be a different upstream identity, so inheriting the + // rejected credential's thread bindings would silently move traffic onto it. Deriving + // lineage from the stored record instead is tautological — the caller reads the same + // record the check would re-read. + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("external", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: 0, + chatgptAccountId: "acc", + }); + const rejectedGeneration = readCodexAccountRecord("external")!.generation; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => { + // An operator re-authenticates while the token call is in flight. + saveCodexAccountCredential("external", { + accessToken: "external-access", + refreshToken: "external-grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + return Response.json({ access_token: "rotated", refresh_token: "grant2", expires_in: 3600 }); + }) as typeof fetch; + + try { + const result = await forceRefreshCodexPoolToken("external", { + rejectedGeneration, + rejectedAccessToken: "rejected", + }).catch(error => error as Error); + // Either the CAS is refused outright, or the replacement is returned without claiming + // this call produced it. What must never happen is selfRefreshed on someone else's write. + if (!(result instanceof Error)) { + expect(result.selfRefreshed).toBe(false); + } + // The replacement survives regardless. + expect(readCodexAccountRecord("external")!.credential!.accessToken).toBe("external-access"); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("a transient error merely mentioning invalid_grant stays transient (#2887 review 2)", async () => { + // Matching the phrase anywhere in the combined code+description text would retire a + // healthy account on an upstream blip — reintroducing the defect this path fixes. + const { forceRefreshCodexPoolToken, readCodexAccountRecord, saveCodexAccountCredential, TokenRefreshError } = + await import("../src/codex/account-store"); + saveCodexAccountCredential("blip", { + accessToken: "rejected", + refreshToken: "grant", + expiresAt: Date.now() + 3600_000, + chatgptAccountId: "acc", + }); + const generation = readCodexAccountRecord("blip")!.generation; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => Response.json({ + error: "server_error", + error_description: "upstream failed while validating invalid_grant handling", + }, { status: 503 })) as typeof fetch; + + try { + await forceRefreshCodexPoolToken("blip", { + rejectedGeneration: generation, + rejectedAccessToken: "rejected", + }); + throw new Error("expected a TokenRefreshError"); + } catch (error) { + expect(error).toBeInstanceOf(TokenRefreshError); + expect((error as InstanceType).reason).toBe("unknown"); + } finally { + globalThis.fetch = originalFetch; + } + }); }); diff --git a/tests/responses-pool-401-refresh.test.ts b/tests/responses-pool-401-refresh.test.ts new file mode 100644 index 0000000000..73c78a8d13 --- /dev/null +++ b/tests/responses-pool-401-refresh.test.ts @@ -0,0 +1,355 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createHash } from "node:crypto"; +import { clearAccountNeedsReauth, isAccountNeedsReauth } from "../src/codex/auth-api"; +import { + clearCodexUpstreamHealth, + clearThreadAccountMap, + resolveCodexAccountForThreadDetailed, +} from "../src/codex/routing"; +import { handleResponses, handleResponsesCompact } from "../src/server/responses"; +import type { RequestLogContext } from "../src/server/request-log"; +import type { OcxConfig } from "../src/types"; + +/** + * #2887: an ordinary stored pool account holding a TIME-VALID access token that upstream + * rejects with a pre-stream 401. Before the fix the refresh-and-replay branch admitted only + * `main-pool`, so this account was never refreshed: one send, zero token-endpoint calls, a + * 401 handed to the client, `needsReauth` set, and its affinity swept. + * + * These assertions are written against that failure signature, not against a value + * comparison, so restoring the `main-pool`-only predicate turns them red. + */ + +const ACCOUNT_ID = "work"; +const OTHER_ACCOUNT_ID = "other"; +const originalFetch = globalThis.fetch; +let home = ""; +let previousOcxHome: string | undefined; +let previousCodexHome: string | undefined; + +function config(options: { secondAccount?: boolean } = {}): OcxConfig { + return { + defaultProvider: "openai", + activeCodexAccountId: ACCOUNT_ID, + autoSwitchThreshold: 0, + // Round-robin over two accounts makes a lost binding observable. Under a single-account + // pool, selection re-picks and re-binds the same account, so a dropped affinity looks + // identical to a preserved one and the assertion would prove nothing. + ...(options.secondAccount ? { accountPoolStrategy: "round-robin" } : {}), + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "pool", + }, + }, + codexAccounts: options.secondAccount + ? [{ id: ACCOUNT_ID, label: "work" }, { id: OTHER_ACCOUNT_ID, label: "other" }] + : [{ id: ACCOUNT_ID, label: "work" }], + } as unknown as OcxConfig; +} + +const THREAD_ID = "thread-2887"; + +function request( + path: "/v1/responses" | "/v1/responses/compact", + options: { affined?: boolean } = {}, +): Request { + return new Request(`http://localhost${path}`, { + method: "POST", + headers: { + "content-type": "application/json", + // A bound thread is what makes affinity exist at all; without it there is no + // entry to carry across the refresh and the handoff cannot be observed. + ...(options.affined ? { "x-codex-parent-thread-id": THREAD_ID } : {}), + }, + body: JSON.stringify(path.endsWith("compact") + ? { model: "gpt-5.5", input: [] } + : { model: "gpt-5.5", input: "hello", stream: false }), + }); +} + +function storedRecord(options: { + accessToken: string; + refreshToken: string; + generation: number; + chatgptAccountId: string; +}) { + return { + credential: { + accessToken: options.accessToken, + refreshToken: options.refreshToken, + expiresAt: Date.now() + 3_600_000, + chatgptAccountId: options.chatgptAccountId, + }, + generation: options.generation, + refreshGrantFingerprint: createHash("sha256") + .update(`codex-refresh-grant:${options.refreshToken}`) + .digest("hex"), + }; +} + +/** A stored credential whose expiry is far beyond the refresh skew, as in the report. */ +function writeStoredAccount(extra: Record = {}): void { + writeFileSync(join(home, "codex-accounts.json"), JSON.stringify({ + [ACCOUNT_ID]: storedRecord({ + accessToken: "rejected-access", + refreshToken: "refresh-grant", + generation: 3, + chatgptAccountId: "acc-work", + }), + ...extra, + }, null, 2)); +} + +function readStoredGeneration(): number { + const raw = JSON.parse(readFileSync(join(home, "codex-accounts.json"), "utf8")) as + Record; + return raw[ACCOUNT_ID]!.generation; +} + +type Harness = { sends: string[]; refreshes: string[] }; + +/** + * Upstream rejects the old bearer once, the token endpoint rotates, and the replay with the + * new bearer succeeds — the reporter's deterministic harness. + */ +function installHarness(options: { refresh?: () => Response } = {}): Harness { + const sends: string[] = []; + const refreshes: string[] = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + if (url.hostname === "auth.openai.com") { + refreshes.push(new URLSearchParams(String(init?.body)).get("refresh_token") ?? ""); + if (options.refresh) return options.refresh(); + return Response.json({ + access_token: "refreshed-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + } + if (!url.pathname.endsWith("/responses") && !url.pathname.endsWith("/responses/compact")) { + return Response.json({ rate_limit: { primary_window: { used_percent: 10 } } }); + } + const authorization = new Headers(init?.headers).get("authorization") ?? ""; + sends.push(authorization); + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "expired bearer" } }, { status: 401 }); + } + return Response.json({ id: "resp_replayed", object: "response", status: "completed", output: [] }); + }) as typeof fetch; + return { sends, refreshes }; +} + +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "ocx-responses-pool-401-")); + previousOcxHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = home; + process.env.CODEX_HOME = home; + clearAccountNeedsReauth(ACCOUNT_ID); + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + writeStoredAccount(); +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + clearAccountNeedsReauth(ACCOUNT_ID); + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + if (previousOcxHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOcxHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + rmSync(home, { recursive: true, force: true }); +}); + +describe("ordinary pool 401 refresh and replay (#2887)", () => { + test("Responses refreshes a time-valid stored credential once and replays the same account", async () => { + const harness = installHarness(); + const response = await handleResponses( + request("/v1/responses"), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + // The defect surfaced as a 401 reaching the client with no refresh attempted. + expect(response.status).toBe(200); + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + // Quarantine is the other half of the report: the account must stay usable. + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expect(readStoredGeneration()).toBe(4); + }); + + test("compact refreshes a time-valid stored credential once and replays the same account", async () => { + const harness = installHarness(); + const response = await handleResponsesCompact( + request("/v1/responses/compact"), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(200); + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); + + test("the replayed account is still selectable on the NEXT request, not just this one", async () => { + // The affinity entry is bound under generation G; the forced refresh CAS-writes G+1 and + // isThreadAffinityGenerationLive demands exact equality. Without the same-lineage handoff + // the entry is dead the moment the replay succeeds, so the account this request just + // recovered is dropped on the following one. Asserting at replay time cannot see that. + installHarness(); + writeStoredAccount({ + [OTHER_ACCOUNT_ID]: storedRecord({ + accessToken: "other-access", + refreshToken: "other-grant", + generation: 1, + chatgptAccountId: "acc-other", + }), + }); + const cfg = config({ secondAccount: true }); + const response = await handleResponses( + request("/v1/responses", { affined: true }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + expect(response.status).toBe(200); + + // The thread must still resolve through its EXISTING binding. A dead entry is deleted and + // reported as expired, which is the behavior the missing handoff produces. + // The binding lives under the model's quota scope, so resolution must be asked in that + // same scope; a scopeless read looks in the legacy bucket and finds nothing. + expect(resolveCodexAccountForThreadDetailed(THREAD_ID, cfg, Date.now(), "shared")).toEqual({ + status: "selected", + accountId: ACCOUNT_ID, + }); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); + + test("a sibling alias holding the rejected token does not satisfy the forced refresh", async () => { + // findFreshCredentialForGrant scans by refresh grant, so a second account sharing the grant + // can hand back a still-unexpired copy of the very token upstream just rejected. Reusing it + // bumps the generation and replays the identical bearer: a second 401 dressed as recovery. + writeStoredAccount({ + alias: storedRecord({ + accessToken: "rejected-access", + refreshToken: "refresh-grant", + generation: 1, + chatgptAccountId: "acc-work", + }), + }); + const harness = installHarness(); + const response = await handleResponses( + request("/v1/responses"), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(200); + // The rejected bearer must never be sent twice. + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + + test("a transient refresh failure does not quarantine the account", async () => { + // A token-endpoint 5xx becomes TokenRefreshError("unknown"). Treating that as terminal + // would rebuild this very bug: an upstream blip would retire a healthy account. + const harness = installHarness({ + refresh: () => Response.json({ error: "server_error" }, { status: 503 }), + }); + const response = await handleResponses( + request("/v1/responses"), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + expect(harness.refreshes.length).toBe(1); + expect(response.status).toBe(503); + expect(response.headers.get("Retry-After")).toBe("1"); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); + + test("a revoked grant is terminal and retires the account", async () => { + // The mirror case: core historically returned the 401 without recording an outcome, so a + // genuinely dead grant stayed selectable and every request repeated the doomed refresh. + installHarness({ + refresh: () => Response.json( + { error: "invalid_grant", error_description: "refresh token revoked" }, + { status: 400 }, + ), + }); + const response = await handleResponses( + request("/v1/responses"), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(401); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(true); + }); + + test("a 401 carrying a superseded credential generation cannot quarantine the replacement", async () => { + // Two requests can be in flight while an operator re-authenticates. The slower one comes + // back 401 against a credential that no longer exists; without the generation fence it + // takes the fresh credential out of rotation and sweeps affinities that belong to it. + const { recordCodexUpstreamOutcome } = await import("../src/codex/routing"); + + // generation 3 is what the stored fixture was written at; 4 is the replacement. + writeStoredAccount({ + [ACCOUNT_ID]: storedRecord({ + accessToken: "replacement-access", + refreshToken: "replacement-grant", + generation: 4, + chatgptAccountId: "acc-work", + }), + }); + + recordCodexUpstreamOutcome(config(), ACCOUNT_ID, 401, { credentialGeneration: 3 }); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + + // The same evidence against the live generation still retires it, so the fence is a + // lineage check and not a blanket suppression of credential failures. + recordCodexUpstreamOutcome(config(), ACCOUNT_ID, 401, { credentialGeneration: 4 }); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(true); + }); + + test("a credential replaced before the request never reaches the 401 path at all", async () => { + // Establishes the boundary for the lineage rule: once the replacement is stored, it is + // picked up at selection time, so no rejected bearer is ever sent and no rotation is + // spent. The interesting case — a replacement landing WHILE the forced refresh runs — is + // covered at the store level, where the handoff's `selfRefreshed` gate is observable + // without racing the endpoint. + const { saveCodexAccountCredential } = await import("../src/codex/account-store"); + const harness = installHarness({ + refresh: () => { + throw new Error("the token endpoint must not be reached in this scenario"); + }, + }); + saveCodexAccountCredential(ACCOUNT_ID, { + accessToken: "externally-replaced", + refreshToken: "external-grant", + expiresAt: Date.now() + 3_600_000, + chatgptAccountId: "acc-work", + }); + + const response = await handleResponses( + request("/v1/responses", { affined: true }), + config(), + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(200); + expect(harness.refreshes).toEqual([]); + // One send, with the replacement bearer: the rejected token is never used. + expect(harness.sends).toEqual(["Bearer externally-replaced"]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); +});