From d54acacd79368ea388a22e72cd425bdfd8f84d9b Mon Sep 17 00:00:00 2001 From: Watcher Bot Date: Thu, 20 Aug 2026 19:19:27 +0000 Subject: [PATCH] fix(codex): refresh native main account tokens --- src/codex/account-store.ts | 413 +++++++++++++----- src/codex/account-usability.ts | 6 +- src/codex/auth-collision.ts | 4 +- src/codex/auth-context.ts | 101 +++-- src/codex/main-account.ts | 203 ++++++++- src/oauth/chatgpt.ts | 61 ++- src/routing/analytics.ts | 1 + src/server/responses/compact.ts | 53 ++- src/server/responses/core.ts | 188 +++++++- src/usage/log.ts | 2 + tests/codex-account-store.test.ts | 6 +- tests/codex-main-account-refresh.test.ts | 236 ++++++++++ tests/codex-refresh-file-lock.test.ts | 158 +++++++ ...ponses-compact-native-main-refresh.test.ts | 61 +++ tests/responses-native-main-refresh.test.ts | 65 +++ 15 files changed, 1390 insertions(+), 168 deletions(-) create mode 100644 tests/codex-main-account-refresh.test.ts create mode 100644 tests/codex-refresh-file-lock.test.ts create mode 100644 tests/responses-compact-native-main-refresh.test.ts create mode 100644 tests/responses-native-main-refresh.test.ts diff --git a/src/codex/account-store.ts b/src/codex/account-store.ts index 85eaab3ef8..50d0dd55f6 100644 --- a/src/codex/account-store.ts +++ b/src/codex/account-store.ts @@ -1,5 +1,5 @@ -import { createHash } from "node:crypto"; -import { closeSync, existsSync, readFileSync, mkdirSync, openSync, unlinkSync, writeFileSync } from "node:fs"; +import { createHash, randomUUID } from "node:crypto"; +import { closeSync, existsSync, readFileSync, mkdirSync, openSync, renameSync, unlinkSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { ConfigMutationLockError, @@ -17,7 +17,7 @@ type LegacyCodexAccountStore = Record; type CodexAccountStore = Record; type RawCodexAccountStore = Record; -const REFRESH_SKEW_MS = 60_000; +export const CODEX_REFRESH_SKEW_MS = 60_000; const REFRESH_LOCK_STALE_MS = 60_000; const REFRESH_LOCK_WAIT_MS = REFRESH_LOCK_STALE_MS + 5_000; const REFRESH_LOCK_POLL_MS = 50; @@ -203,9 +203,8 @@ export function saveCodexAccountCredentialIfGeneration( if (!current || current.generation !== generation || current.deletedAt != null || !current.credential) { return false; } - const refreshGrantFingerprint = current.credential.refreshToken === cred.refreshToken - ? current.refreshGrantFingerprint ?? refreshGrantFingerprintForToken(cred.refreshToken) - : refreshGrantFingerprintForToken(cred.refreshToken); + const refreshGrantFingerprint = current.refreshGrantFingerprint + ?? refreshGrantFingerprintForToken(current.credential.refreshToken); store[id] = { credential: cred, generation: generation + 1, @@ -287,6 +286,9 @@ function withCredentialMutationLockSync(fn: () => T): T { type CodexTokenResult = { accessToken: string; chatgptAccountId: string; generation: number }; type CodexRefreshResult = CodexTokenResult & { credential?: CodexAccountCredentials }; +export interface CodexAccountStoreRefreshDependencies { + readonly fetch?: typeof fetch; +} const MAX_CODEX_REFRESH_FLIGHTS = 32; const CODEX_REFRESH_FLIGHT_STALE_MS = 120_000; interface RefreshFlight { @@ -295,10 +297,11 @@ interface RefreshFlight { abort: AbortController; } const refreshLocks = new Map(); +const abandonedRefreshLockOwners = new Set(); -function codexRefreshLockPath(lockKey: string): string { +function codexRefreshLockPath(lockKey: string, directory = getConfigDir()): string { const digest = createHash("sha256").update(lockKey).digest("hex").slice(0, 32); - return join(getConfigDir(), `codex-refresh-${digest}.lock`); + return join(directory, `codex-refresh-${digest}.lock`); } function sleep(ms: number, signal?: AbortSignal): Promise { @@ -321,58 +324,202 @@ function errCode(err: unknown): string | undefined { return err && typeof err === "object" && "code" in err ? String((err as { code?: unknown }).code) : undefined; } -function isRefreshLockStale(path: string): boolean { +interface RefreshLockOwner { + readonly owner: string; + readonly pid: number; + readonly acquiredAt: number; + readonly abandonedAt?: number; +} + +function refreshLockOwner(path: string): RefreshLockOwner | undefined { try { hardenExistingSecret(path); - const parsed = JSON.parse(readFileSync(path, "utf-8")) as { acquiredAt?: unknown }; - return typeof parsed.acquiredAt !== "number" || Date.now() - parsed.acquiredAt > REFRESH_LOCK_STALE_MS; + const parsed = JSON.parse(readFileSync(path, "utf-8")) as Partial; + if (typeof parsed.owner !== "string" || typeof parsed.pid !== "number" || typeof parsed.acquiredAt !== "number") { + return undefined; + } + return { + owner: parsed.owner, + pid: parsed.pid, + acquiredAt: parsed.acquiredAt, + ...(typeof parsed.abandonedAt === "number" ? { abandonedAt: parsed.abandonedAt } : {}), + }; } catch { + return undefined; + } +} + +function refreshLockOwnerIsLive(owner: RefreshLockOwner): boolean { + try { + process.kill(owner.pid, 0); return true; + } catch (error) { + return errCode(error) === "EPERM"; } } -async function withCodexRefreshFileLock(lockKey: string, signal: AbortSignal, fn: () => Promise): Promise { - hardenConfigDir(); - const dir = getConfigDir(); - if (!existsSync(dir)) mkdirSync(dir, { recursive: true, mode: 0o700 }); +function refreshLockIsStale(path: string): boolean { + const owner = refreshLockOwner(path); + if (owner?.abandonedAt !== undefined || (owner && abandonedRefreshLockOwners.has(owner.owner))) return true; + if (!owner) { + try { + const parsed = JSON.parse(readFileSync(path, "utf-8")) as { acquiredAt?: unknown }; + // Pre-owner lock records remain live for their bounded lease. This preserves + // rolling-upgrade compatibility; malformed records without a timestamp are reclaimable. + return typeof parsed.acquiredAt !== "number" || Date.now() - parsed.acquiredAt > REFRESH_LOCK_STALE_MS; + } catch { + return true; + } + } + return Date.now() - owner.acquiredAt > REFRESH_LOCK_STALE_MS && !refreshLockOwnerIsLive(owner); +} + +function abandonRefreshLock(path: string, owner: string): void { + abandonedRefreshLockOwners.add(owner); + const current = refreshLockOwner(path); + if (!current || current.owner !== owner) return; + writeFileSync(path, JSON.stringify({ ...current, abandonedAt: Date.now() }) + "\n"); +} + +function quarantineStaleRefreshLock(path: string): void { + const retiredPath = `${path}.stale-${randomUUID()}`; + const owner = refreshLockOwner(path); + try { + if (!refreshLockIsStale(path)) return; + // Acquirers hold the reclaim lock while creating the lock file, so this + // rename can only retire the stale entry that was just inspected. + renameSync(path, retiredPath); + } catch (error) { + if (errCode(error) !== "ENOENT") throw error; + } finally { + try { + unlinkSync(retiredPath); + } catch (error) { + if (errCode(error) !== "ENOENT") throw error; + } + if (owner) abandonedRefreshLockOwners.delete(owner.owner); + } +} - const path = codexRefreshLockPath(lockKey); +async function acquireRefreshReclaimLock(path: string, signal?: AbortSignal): Promise<{ fd: number; owner: RefreshLockOwner }> { + const reclaimPath = `${path}.reclaim`; const deadline = Date.now() + REFRESH_LOCK_WAIT_MS; - let fd: number | null = null; - while (fd == null) { - if (signal.aborted) throw signal.reason; + const reclaimOwner: RefreshLockOwner = { owner: randomUUID(), pid: process.pid, acquiredAt: Date.now() }; + while (true) { + if (signal?.aborted) throw signal.reason; + if (Date.now() >= deadline) throw new CodexCredentialRefreshLockTimeoutError(); try { - fd = openSync(path, "wx", 0o600); - writeFileSync(fd, JSON.stringify({ acquiredAt: Date.now(), pid: process.pid }) + "\n"); - break; - } catch (err) { - if (errCode(err) !== "EEXIST") throw err; - if (isRefreshLockStale(path)) { - try { - unlinkSync(path); - } catch (unlinkErr) { - if (errCode(unlinkErr) !== "ENOENT") throw unlinkErr; - } - continue; - } + const fd = openSync(reclaimPath, "wx", 0o600); + writeFileSync(fd, JSON.stringify(reclaimOwner) + "\n"); + return { fd, owner: reclaimOwner }; + } catch (error) { + if (errCode(error) !== "EEXIST") throw error; if (Date.now() >= deadline) throw new CodexCredentialRefreshLockTimeoutError(); await sleep(REFRESH_LOCK_POLL_MS, signal); } } +} +function tryAcquireRefreshReclaimLock(path: string): { fd: number; owner: RefreshLockOwner } | null { + const reclaimPath = `${path}.reclaim`; + const reclaimOwner: RefreshLockOwner = { owner: randomUUID(), pid: process.pid, acquiredAt: Date.now() }; try { - return await fn(); + const fd = openSync(reclaimPath, "wx", 0o600); + writeFileSync(fd, JSON.stringify(reclaimOwner) + "\n"); + return { fd, owner: reclaimOwner }; + } catch (error) { + if (errCode(error) === "EEXIST") return null; + throw error; + } +} + +function releaseRefreshReclaimLock(path: string, reclaim: { fd: number; owner: RefreshLockOwner }): void { + closeSync(reclaim.fd); + const reclaimPath = `${path}.reclaim`; + try { + if (refreshLockOwner(reclaimPath)?.owner !== reclaim.owner.owner) return; + unlinkSync(reclaimPath); + } catch (error) { + if (errCode(error) !== "ENOENT") throw error; + } +} + +async function releaseRefreshLockOnce(path: string, owner: string, signal: AbortSignal): Promise { + const reclaim = await acquireRefreshReclaimLock(path, signal); + try { + if (refreshLockOwner(path)?.owner !== owner) return true; + unlinkSync(path); + abandonedRefreshLockOwners.delete(owner); + return true; + } catch (error) { + if (errCode(error) !== "ENOENT") throw error; + return true; } finally { - if (fd != null) closeSync(fd); + releaseRefreshReclaimLock(path, reclaim); + } +} + +async function releaseRefreshLock(path: string, owner: string, signal: AbortSignal): Promise { + if (signal.aborted) { + const reclaim = tryAcquireRefreshReclaimLock(path); + if (!reclaim) { + abandonRefreshLock(path, owner); + throw signal.reason; + } try { + if (refreshLockOwner(path)?.owner !== owner) return; unlinkSync(path); - } catch (err) { - if (errCode(err) !== "ENOENT") throw err; + } catch (error) { + if (errCode(error) !== "ENOENT") throw error; + } finally { + releaseRefreshReclaimLock(path, reclaim); } + return; } + await releaseRefreshLockOnce(path, owner, AbortSignal.any([signal, AbortSignal.timeout(1_000)])); } -function findFreshCredentialForGrant( +export async function withCodexRefreshFileLock(args: { + lockKey: string; + signal: AbortSignal; + run: () => Promise; + directory?: string; +}): Promise { + const dir = args.directory ?? getConfigDir(); + if (!args.directory) hardenConfigDir(); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true, mode: 0o700 }); + + const path = codexRefreshLockPath(args.lockKey, dir); + const deadline = Date.now() + REFRESH_LOCK_WAIT_MS; + const owner = randomUUID(); + let fd: number | null = null; + while (fd == null) { + if (args.signal.aborted) throw args.signal.reason; + const reclaim = await acquireRefreshReclaimLock(path, args.signal); + try { + try { + fd = openSync(path, "wx", 0o600); + writeFileSync(fd, JSON.stringify({ acquiredAt: Date.now(), pid: process.pid, owner }) + "\n"); + } catch (err) { + if (errCode(err) !== "EEXIST") throw err; + quarantineStaleRefreshLock(path); + } + } finally { + releaseRefreshReclaimLock(path, reclaim); + } + if (fd == null && Date.now() >= deadline) throw new CodexCredentialRefreshLockTimeoutError(); + if (fd == null) await sleep(REFRESH_LOCK_POLL_MS, args.signal); + } + + try { + return await args.run(); + } finally { + if (fd != null) closeSync(fd); + await releaseRefreshLock(path, owner, args.signal); + } +} + +export function findFreshCredentialForGrant( refreshGrantFingerprint: string, excludeId: string, ): CodexAccountCredentials | null { @@ -381,11 +528,44 @@ function findFreshCredentialForGrant( for (const [candidateId, candidate] of Object.entries(records)) { if (candidateId === excludeId || candidate.deletedAt != null || !candidate.credential) continue; if (recordGrantFingerprint(candidate) !== refreshGrantFingerprint) continue; - if (candidate.credential.expiresAt > now + REFRESH_SKEW_MS) return candidate.credential; + if (candidate.credential.expiresAt > now + CODEX_REFRESH_SKEW_MS) return candidate.credential; } return null; } +export function publishFreshCredentialForGrant( + args: { + refreshGrantFingerprint: string; + credential: CodexAccountCredentials; + excludeId: string; + replaceAccessToken?: string; + }, +): void { + withCredentialMutationLockSync(() => { + const now = Date.now(); + const store = loadCodexAccountRecordStore(); + let changed = false; + for (const [candidateId, candidate] of Object.entries(store)) { + if (candidateId === args.excludeId || candidate.deletedAt != null || !candidate.credential) continue; + if (recordGrantFingerprint(candidate) !== args.refreshGrantFingerprint) continue; + if ( + candidate.credential.expiresAt > now + CODEX_REFRESH_SKEW_MS + && candidate.credential.accessToken !== args.replaceAccessToken + ) continue; + store[candidateId] = { + ...candidate, + credential: args.credential, + generation: candidate.generation + 1, + refreshGrantFingerprint: args.refreshGrantFingerprint, + replacedAt: Date.now(), + ...preservedValidationMetadata(candidate), + }; + changed = true; + } + if (changed) persist(store); + }); +} + async function notePlanFromRefreshedAccessToken( id: string, accessToken: string, @@ -399,14 +579,17 @@ async function notePlanFromRefreshedAccessToken( } } -export async function getValidCodexToken(id: string): Promise { +export async function getValidCodexToken( + id: string, + dependencies: CodexAccountStoreRefreshDependencies = {}, +): 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) { + if (cred.expiresAt > Date.now() + CODEX_REFRESH_SKEW_MS) { return { accessToken: cred.accessToken, chatgptAccountId: cred.chatgptAccountId, generation: record.generation }; } @@ -436,7 +619,7 @@ export async function getValidCodexToken(id: string): Promise generation, }; } - return getValidCodexToken(id); + return getValidCodexToken(id, dependencies); } } @@ -445,15 +628,28 @@ export async function getValidCodexToken(id: string): Promise const abort = new AbortController(); const signal = AbortSignal.any([abort.signal, AbortSignal.timeout(30_000)]); let flight!: RefreshFlight; - const refreshPromise = withCodexRefreshFileLock(refreshGrantFingerprint, signal, async (): Promise => { - const current = readCodexAccountRecord(id); - const lockedRecord = readCodexAccountRecord(id); - const lockedCred = lockedRecord?.deletedAt == null ? lockedRecord?.credential : undefined; - if (!lockedRecord || !lockedCred) throw new CodexCredentialGenerationConflictError(); - const startGeneration = lockedRecord.generation; - const lockedRefreshGrantFingerprint = recordGrantFingerprint(lockedRecord); - if (lockedRefreshGrantFingerprint !== refreshGrantFingerprint) { - if (lockedCred.expiresAt > Date.now() + REFRESH_SKEW_MS) { + const refreshPromise = withCodexRefreshFileLock({ + lockKey: refreshGrantFingerprint, + signal, + run: async (): Promise => { + const current = readCodexAccountRecord(id); + const lockedRecord = readCodexAccountRecord(id); + const lockedCred = lockedRecord?.deletedAt == null ? lockedRecord?.credential : undefined; + if (!lockedRecord || !lockedCred) throw new CodexCredentialGenerationConflictError(); + const startGeneration = lockedRecord.generation; + const lockedRefreshGrantFingerprint = recordGrantFingerprint(lockedRecord); + if (lockedRefreshGrantFingerprint !== refreshGrantFingerprint) { + if (lockedCred.expiresAt > Date.now() + CODEX_REFRESH_SKEW_MS) { + return { + accessToken: lockedCred.accessToken, + chatgptAccountId: lockedCred.chatgptAccountId, + generation: startGeneration, + credential: lockedCred, + }; + } + throw new CodexCredentialGenerationConflictError(); + } + if (lockedCred.expiresAt > Date.now() + CODEX_REFRESH_SKEW_MS) { return { accessToken: lockedCred.accessToken, chatgptAccountId: lockedCred.chatgptAccountId, @@ -461,73 +657,64 @@ export async function getValidCodexToken(id: string): Promise credential: lockedCred, }; } - throw new CodexCredentialGenerationConflictError(); - } - if (lockedCred.expiresAt > Date.now() + REFRESH_SKEW_MS) { - return { - accessToken: lockedCred.accessToken, + const sameGrantFreshCredential = findFreshCredentialForGrant(refreshGrantFingerprint, id); + if (sameGrantFreshCredential) { + if (!saveCodexAccountCredentialIfGeneration(id, startGeneration, sameGrantFreshCredential)) { + throw new CodexCredentialGenerationConflictError(); + } + return { + accessToken: sameGrantFreshCredential.accessToken, + chatgptAccountId: sameGrantFreshCredential.chatgptAccountId, + generation: startGeneration + 1, + credential: sameGrantFreshCredential, + }; + } + const res = await (dependencies.fetch ?? fetch)(CHATGPT_TOKEN_URL, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + grant_type: "refresh_token", + client_id: CHATGPT_CLIENT_ID, + refresh_token: lockedCred.refreshToken, + }).toString(), + signal, + }); + if (!res.ok) { + const errText = await res.text().catch(() => ""); + let errDesc: string; + try { + const parsed = JSON.parse(errText) as { error?: string; error_description?: string }; + 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 + : errDesc.includes("expired") ? "expired" as const + : "unknown" as const; + throw new TokenRefreshError(reason, `Codex token refresh failed (${reason}); reauthenticate the account.`); + } + const data = (await res.json()) as { access_token: string; refresh_token?: string; expires_in: number }; + // Guard against a missing/non-finite/negative expires_in (malformed upstream + // response): a NaN expiry would never compare as expired, and a negative + // duration would stamp an already-past expiry — both block refresh semantics. + const expiresIn = + typeof data.expires_in === "number" && Number.isFinite(data.expires_in) && data.expires_in >= 0 + ? data.expires_in + : 3600; + // The computed timestamp itself must stay finite: Number.MAX_VALUE passes + // Number.isFinite but overflows to Infinity once multiplied by 1000. + const expiresAt = Date.now() + expiresIn * 1000; + const safeExpiresAt = Number.isFinite(expiresAt) ? expiresAt : Date.now() + 3600 * 1000; + + const updated: CodexAccountCredentials = { + accessToken: data.access_token, + refreshToken: data.refresh_token ?? lockedCred.refreshToken, + expiresAt: safeExpiresAt, chatgptAccountId: lockedCred.chatgptAccountId, - generation: startGeneration, - credential: lockedCred, }; - } - const sameGrantFreshCredential = findFreshCredentialForGrant(refreshGrantFingerprint, id); - if (sameGrantFreshCredential) { - if (!saveCodexAccountCredentialIfGeneration(id, startGeneration, sameGrantFreshCredential)) { + if (!saveCodexAccountCredentialIfGeneration(id, startGeneration, updated)) { throw new CodexCredentialGenerationConflictError(); } - return { - accessToken: sameGrantFreshCredential.accessToken, - chatgptAccountId: sameGrantFreshCredential.chatgptAccountId, - generation: startGeneration + 1, - credential: sameGrantFreshCredential, - }; - } - const res = await fetch(CHATGPT_TOKEN_URL, { - method: "POST", - headers: { "Content-Type": "application/x-www-form-urlencoded" }, - body: new URLSearchParams({ - grant_type: "refresh_token", - client_id: CHATGPT_CLIENT_ID, - refresh_token: lockedCred.refreshToken, - }).toString(), - signal, - }); - if (!res.ok) { - const errText = await res.text().catch(() => ""); - let errDesc: string; - try { - const parsed = JSON.parse(errText) as { error?: string; error_description?: string }; - 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 - : errDesc.includes("expired") ? "expired" as const - : "unknown" as const; - throw new TokenRefreshError(reason, `Codex token refresh failed (${reason}); reauthenticate the account.`); - } - const data = (await res.json()) as { access_token: string; refresh_token?: string; expires_in: number }; - // Guard against a missing/non-finite/negative expires_in (malformed upstream - // response): a NaN expiry would never compare as expired, and a negative - // duration would stamp an already-past expiry — both block refresh semantics. - const expiresIn = - typeof data.expires_in === "number" && Number.isFinite(data.expires_in) && data.expires_in >= 0 - ? data.expires_in - : 3600; - // The computed timestamp itself must stay finite: Number.MAX_VALUE passes - // Number.isFinite but overflows to Infinity once multiplied by 1000. - const expiresAt = Date.now() + expiresIn * 1000; - const safeExpiresAt = Number.isFinite(expiresAt) ? expiresAt : Date.now() + 3600 * 1000; - - const updated: CodexAccountCredentials = { - accessToken: data.access_token, - refreshToken: data.refresh_token ?? lockedCred.refreshToken, - expiresAt: safeExpiresAt, - chatgptAccountId: lockedCred.chatgptAccountId, - }; - 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 }; + }, }).finally(() => { if (refreshLocks.get(refreshGrantFingerprint) === flight) refreshLocks.delete(refreshGrantFingerprint); }); diff --git a/src/codex/account-usability.ts b/src/codex/account-usability.ts index d508e19f4d..264bc5f29d 100644 --- a/src/codex/account-usability.ts +++ b/src/codex/account-usability.ts @@ -1,6 +1,6 @@ import { getCodexAccountCredential } from "./account-store"; import { isAccountNeedsReauth } from "./account-runtime-state"; -import { MAIN_CODEX_ACCOUNT_ID, isMainAccountTokenLive } from "./main-account"; +import { MAIN_CODEX_ACCOUNT_ID, isMainAccountCredentialUsable, isMainAccountTokenLive } from "./main-account"; import { hasLegacyMainCodexPoolAccount, isSelectableCodexPoolAccount } from "./account-id"; import type { OcxConfig } from "../types"; import { isNativeMainTrafficBlocked } from "./native-profile-startup"; @@ -32,8 +32,8 @@ export function isCodexAccountUsable( // before reservation or token materialization. Treat cached main as a routing // candidate without touching the credential file so affinity is not rebound. if (options.nativeMainSelectionOnly) return true; - // Main account: credential is the read-only ~/.codex/auth.json token (Option A). - return (options.isMainAccountTokenLive ?? isMainAccountTokenLive)(); + // Main account: credential is ~/.codex/auth.json and may be refreshed from its native refresh token. + return (options.isMainAccountTokenLive ?? isMainAccountCredentialUsable)(); } const exists = (config.codexAccounts ?? []) .some(account => isSelectableCodexPoolAccount(account) && account.id === accountId); diff --git a/src/codex/auth-collision.ts b/src/codex/auth-collision.ts index 52c9242e8c..8729c715aa 100644 --- a/src/codex/auth-collision.ts +++ b/src/codex/auth-collision.ts @@ -11,6 +11,7 @@ export interface CodexTokens { access_token: string; account_id: string; id_token?: string; + refresh_token?: string; } /** @@ -42,7 +43,7 @@ export function readCodexTokensResult(): CodexTokenReadResult { } try { const j = JSON.parse(raw) as { - tokens?: { access_token?: string; account_id?: string; id_token?: string }; + tokens?: { access_token?: string; account_id?: string; id_token?: string; refresh_token?: string }; }; if (!j?.tokens?.access_token) return { status: "invalid" }; return { @@ -51,6 +52,7 @@ export function readCodexTokensResult(): CodexTokenReadResult { access_token: j.tokens.access_token, account_id: j.tokens.account_id ?? "", id_token: j.tokens.id_token, + refresh_token: j.tokens.refresh_token, }, }; } catch { diff --git a/src/codex/auth-context.ts b/src/codex/auth-context.ts index 71a79b1b67..2e0be095d0 100644 --- a/src/codex/auth-context.ts +++ b/src/codex/auth-context.ts @@ -3,6 +3,7 @@ import { CodexCredentialRefreshLockTimeoutError, CodexCredentialRefreshBusyError, CodexCredentialRefreshStaleError, + TokenRefreshError, getValidCodexToken, isCodexAccountGenerationLive, } from "./account-store"; @@ -11,7 +12,8 @@ import { isCodexAccountPaused } from "./account-pause"; import { ConfigMutationLockError } from "../config"; import { isCodexAccountUsable } from "./account-usability"; import { reconcileMainCodexAccountRuntimeState } from "./account-lifecycle"; -import { MAIN_CODEX_ACCOUNT_ID, getMainAccountToken, isMainAccountTokenLive } from "./main-account"; +import { MAIN_CODEX_ACCOUNT_ID, getMainAccountToken, getValidMainAccountToken, isMainAccountTokenLive } from "./main-account"; +import type { NativeMainRefreshDependencies } from "./main-account"; import { isNativeMainTrafficBlocked, nativeMainStartupGateSnapshot } from "./native-profile-startup"; import type { NativeMainStartupBlockReason } from "./native-profile-startup"; import { @@ -39,6 +41,16 @@ import type { CodexAccountMode, OcxConfig, OcxProviderConfig } from "../types"; import { FORWARD_HEADERS } from "../adapters/openai-responses"; import { captureConfigGeneration } from "../lib/state-store-sweeper"; +function startLazyQuotaPrime(config: OcxConfig, prime?: (config: OcxConfig, reason: string) => Promise): void { + if (prime) { + void prime(config, "pre-route").catch(() => {}); + return; + } + void import("./auth-api") + .then(({ primeCodexPoolQuotas }) => primeCodexPoolQuotas(config, "pre-route")) + .catch(() => {}); +} + export type CodexAuthContext = | { kind: "main"; accountId: null } | { @@ -295,6 +307,8 @@ export interface ResolveCodexAuthContextOptions { /** Test-only native credential read seams. */ isMainAccountTokenLive?: () => boolean; getMainAccountToken?: typeof getMainAccountToken; + getValidMainAccountToken?: typeof getValidMainAccountToken; + nativeMainRefreshDependencies?: NativeMainRefreshDependencies; primeCodexPoolQuotas?: (config: OcxConfig, reason: string) => Promise; /** Test seam for account-gated native model discovery. */ resolveCodexModelEntitlements?: ( @@ -450,13 +464,7 @@ export async function resolveCodexAuthContext( // blocks the current request, and the helper's single-flight guard collapses // repeated triggers into one pass. if (fixedAccountId === undefined && !nativeMainReadsForbidden && !getAccountQuota(accountId)) { - if (options.primeCodexPoolQuotas) { - void options.primeCodexPoolQuotas(config, "pre-route").catch(() => {}); - } else { - import("./auth-api") - .then(({ primeCodexPoolQuotas }) => primeCodexPoolQuotas(config, "pre-route")) - .catch(() => {}); - } + startLazyQuotaPrime(config, options.primeCodexPoolQuotas); } // Snapshot (not just the deadline) so a refused request can report WHY it is cooled: // a literal Retry-After reads very differently to a user than a reset-derived guess. @@ -483,27 +491,48 @@ export async function resolveCodexAuthContext( } if (accountId === MAIN_CODEX_ACCOUNT_ID) { - // Main account in rotation: inject the read-only auth.json token and fail closed if it vanished. - const token = (options.getMainAccountToken ?? getMainAccountToken)(); - if (!token) { + try { + // Main account in rotation: inject a valid auth.json token, refreshing it when possible. + const token = options.getValidMainAccountToken + ? await options.getValidMainAccountToken() + : options.getMainAccountToken + ? options.getMainAccountToken() + : await getValidMainAccountToken({ dependencies: options.nativeMainRefreshDependencies }); + if (!token) { + // Nothing will reach upstream, so give the probe back instead of burning it. + if (probeLeaseId && probeQuotaScope) releaseCodexQuotaScopeProbeLease(accountId, probeQuotaScope, probeLeaseId); + else if (probeLeaseId) releaseCodexQuotaProbeLease(accountId, probeLeaseId); + throw new CodexPoolAuthenticationError( + fixedAccountId !== undefined ? "Selected Codex account is unavailable" : undefined, + ); + } + return { + kind: "main-pool", + accountId, + writerGeneration, + accessToken: token.accessToken, + chatgptAccountId: token.chatgptAccountId, + ...(fixedAccountId !== undefined ? { fixedAccount: true } : {}), + ...(quotaScope ? { quotaScope } : {}), + ...(probeLeaseId ? { probeLeaseId } : {}), + ...(probeQuotaScope ? { probeQuotaScope } : {}), + }; + } catch (cause) { // Nothing will reach upstream, so give the probe back instead of burning it. if (probeLeaseId && probeQuotaScope) releaseCodexQuotaScopeProbeLease(accountId, probeQuotaScope, probeLeaseId); else if (probeLeaseId) releaseCodexQuotaProbeLease(accountId, probeLeaseId); - throw new CodexPoolAuthenticationError( - fixedAccountId !== undefined ? "Selected Codex account is unavailable" : undefined, - ); + if ( + cause instanceof CodexPoolAuthenticationError + || cause instanceof TokenRefreshError + || cause instanceof CodexCredentialRefreshLockTimeoutError + || cause instanceof CodexCredentialRefreshBusyError + || cause instanceof CodexCredentialRefreshStaleError + ) throw cause; + if (shouldMarkAccountNeedsReauthForCodexAuthFailure(cause)) { + markAccountNeedsReauth(accountId, writerGeneration); + } + throw new CodexAuthContextError(accountId, cause); } - return { - kind: "main-pool", - accountId, - writerGeneration, - accessToken: token.accessToken, - chatgptAccountId: token.chatgptAccountId, - ...(fixedAccountId !== undefined ? { fixedAccount: true } : {}), - ...(quotaScope ? { quotaScope } : {}), - ...(probeLeaseId ? { probeLeaseId } : {}), - ...(probeQuotaScope ? { probeQuotaScope } : {}), - }; } try { @@ -605,6 +634,28 @@ export function materializeCodexUpstreamAuth( return selected; } +export async function materializeCodexUpstreamAuthAsync(args: { + headers: Headers; + ctx: CodexAuthContext; + options?: { substituteMainCredential?: boolean; nativeMainRefreshDependencies?: NativeMainRefreshDependencies }; +}): Promise { + const options = args.options ?? {}; + if (args.ctx.kind !== "main" || options.substituteMainCredential !== true) { + return materializeCodexUpstreamAuth(args.headers, args.ctx, options); + } + const selected = new Headers(); + for (const name of FORWARD_HEADERS) { + const value = args.headers.get(name); + if (value) selected.set(name, value); + } + const stored = await getValidMainAccountToken({ dependencies: options.nativeMainRefreshDependencies }); + // Fail BEFORE any upstream I/O. Falling through here would send the admission secret. + if (!stored?.accessToken) throw new CodexMainSubstitutionUnavailableError(); + selected.set("authorization", `Bearer ${stored.accessToken}`); + if (stored.chatgptAccountId) selected.set("chatgpt-account-id", stored.chatgptAccountId); + return selected; +} + /** @deprecated Prefer materializeCodexUpstreamAuth; kept for call sites without admission context. */ export function headersForCodexAuthContext(headers: Headers, ctx: CodexAuthContext): Headers { return materializeCodexUpstreamAuth(headers, ctx); diff --git a/src/codex/main-account.ts b/src/codex/main-account.ts index 30296b586a..5641cd5854 100644 --- a/src/codex/main-account.ts +++ b/src/codex/main-account.ts @@ -1,5 +1,21 @@ +import { existsSync, mkdirSync, readFileSync } from "node:fs"; +import { join } from "node:path"; +import { atomicWriteFile } from "../config"; import { readCodexTokens } from "./auth-collision"; -import { decodeJwtPayload } from "../oauth/chatgpt"; +import { decodeJwtPayload, extractAccountId, refreshChatGPTTokenRaw } from "../oauth/chatgpt"; +import { + CODEX_REFRESH_SKEW_MS, + CodexCredentialRefreshBusyError, + CodexCredentialRefreshLockTimeoutError, + CodexCredentialRefreshStaleError, + TokenRefreshError, + findFreshCredentialForGrant, + publishFreshCredentialForGrant, + refreshGrantFingerprintForToken, + withCodexRefreshFileLock, +} from "./account-store"; +import { clearAccountNeedsReauth, markAccountNeedsReauth } from "./account-runtime-state"; +import { resolveCodexHomeDir } from "./home"; import { extractChatgptPlanType } from "./plan"; import { MAIN_CODEX_ACCOUNT_ID } from "./account-id"; @@ -13,6 +29,15 @@ export { MAIN_CODEX_ACCOUNT_ID } from "./account-id"; let mainAccountPlan: string | null = null; let jwtPlanAttempted = false; +interface AuthJsonShape { + tokens?: Record; + [key: string]: unknown; +} + +export interface NativeMainRefreshDependencies { + readonly refreshToken?: typeof refreshChatGPTTokenRaw; +} + export function setMainAccountPlan(plan: string | null): void { mainAccountPlan = plan; if (plan === null) jwtPlanAttempted = false; @@ -37,6 +62,13 @@ export function getMainAccountToken(): { accessToken: string; chatgptAccountId: return { accessToken: tokens.access_token, chatgptAccountId: tokens.account_id }; } +function mainAccessTokenFresh(accessToken: string | undefined, now = Date.now()): boolean { + if (!accessToken) return false; + const payload = decodeJwtPayload(accessToken); + const exp = typeof payload?.exp === "number" ? payload.exp * 1000 : undefined; + return exp === undefined || exp > now + CODEX_REFRESH_SKEW_MS; +} + /** * The main token is usable when it exists and — if its JWT carries a decodable `exp` — is * not expired. When `exp` cannot be decoded we treat the token as live (best-effort); an @@ -44,10 +76,12 @@ export function getMainAccountToken(): { accessToken: string; chatgptAccountId: */ export function isMainAccountTokenLive(now = Date.now()): boolean { const tokens = readCodexTokens(); - if (!tokens?.access_token) return false; - const payload = decodeJwtPayload(tokens.access_token); - const exp = typeof payload?.exp === "number" ? payload.exp * 1000 : undefined; - return exp === undefined || exp > now; + return mainAccessTokenFresh(tokens?.access_token, now - CODEX_REFRESH_SKEW_MS); +} + +export function isMainAccountCredentialUsable(now = Date.now()): boolean { + const token = mainTokenFromAuth(readMainAuthJson()); + return mainAccessTokenFresh(token?.accessToken, now) || !!token?.refreshToken; } /** @@ -66,3 +100,162 @@ export function isMainAccountTokenVerifiablyLive(now = Date.now()): boolean { const exp = typeof payload?.exp === "number" ? payload.exp * 1000 : undefined; return exp !== undefined && exp > now; } + +function authJsonPath(): string { + return join(resolveCodexHomeDir(), "auth.json"); +} + +function readMainAuthJson(): AuthJsonShape | null { + try { + return JSON.parse(readFileSync(authJsonPath(), "utf-8")) as AuthJsonShape; + } catch { + return null; + } +} + +function persistMainAuthJson(auth: AuthJsonShape): void { + const dir = resolveCodexHomeDir(); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true, mode: 0o700 }); + atomicWriteFile(authJsonPath(), JSON.stringify(auth, null, 2) + "\n"); +} + +function mainTokenFromAuth(auth: AuthJsonShape | null): { + accessToken: string; + refreshToken?: string; + refreshGrantFingerprint?: string; + idToken?: string; + chatgptAccountId: string; +} | null { + const tokens = auth?.tokens; + const accessToken = typeof tokens?.access_token === "string" ? tokens.access_token : undefined; + if (!accessToken) return null; + const refreshToken = typeof tokens?.refresh_token === "string" && tokens.refresh_token.trim() + ? tokens.refresh_token.trim() + : undefined; + const refreshGrantFingerprint = typeof tokens?.refresh_grant_fingerprint === "string" + ? tokens.refresh_grant_fingerprint + : undefined; + const idToken = typeof tokens?.id_token === "string" ? tokens.id_token : undefined; + const storedAccountId = typeof tokens?.account_id === "string" ? tokens.account_id : ""; + return { + accessToken, + refreshToken, + refreshGrantFingerprint, + idToken, + chatgptAccountId: storedAccountId || extractAccountId(idToken, accessToken) || "", + }; +} + +function tokenRefreshReason(error: unknown): "expired" | "revoked" | "unknown" { + if (error instanceof TokenRefreshError) return error.reason; + const message = error instanceof Error ? error.message : String(error); + return message.includes("invalidated") || message.includes("invalid_grant") || message.includes("revoked") + ? "revoked" + : message.includes("expired") + ? "expired" + : "unknown"; +} + +function refreshedMainResult(auth: AuthJsonShape): { accessToken: string; chatgptAccountId: string } | null { + const token = mainTokenFromAuth(auth); + if (!token?.accessToken) return null; + setMainAccountPlan(extractChatgptPlanType(token.idToken, token.accessToken) ?? null); + return { accessToken: token.accessToken, chatgptAccountId: token.chatgptAccountId }; +} + +export async function forceRefreshMainAccountToken( + rejectedAccessToken?: string, + options: { signal?: AbortSignal; dependencies?: NativeMainRefreshDependencies } = {}, +): Promise<{ accessToken: string; chatgptAccountId: string } | null> { + const initial = mainTokenFromAuth(readMainAuthJson()); + if (!initial?.refreshToken) return null; + const refreshGrantFingerprint = initial.refreshGrantFingerprint ?? refreshGrantFingerprintForToken(initial.refreshToken); + const signal = options.signal + ? AbortSignal.any([options.signal, AbortSignal.timeout(30_000)]) + : AbortSignal.timeout(30_000); + try { + const refreshed = await withCodexRefreshFileLock({ + lockKey: refreshGrantFingerprint, + signal, + run: async () => { + const lockedAuth = readMainAuthJson(); + const locked = mainTokenFromAuth(lockedAuth); + if (!lockedAuth || !locked?.refreshToken) return null; + if ( + rejectedAccessToken + && locked.accessToken !== rejectedAccessToken + && mainAccessTokenFresh(locked.accessToken) + ) { + return refreshedMainResult(lockedAuth); + } + const lockedRefreshGrantFingerprint = locked.refreshGrantFingerprint ?? refreshGrantFingerprintForToken(locked.refreshToken); + if (lockedRefreshGrantFingerprint !== refreshGrantFingerprint) { + if (mainAccessTokenFresh(locked.accessToken)) return refreshedMainResult(lockedAuth); + return null; + } + const sameGrantFreshCredential = findFreshCredentialForGrant(refreshGrantFingerprint, MAIN_CODEX_ACCOUNT_ID); + if (sameGrantFreshCredential) { + if (!rejectedAccessToken || sameGrantFreshCredential.accessToken !== rejectedAccessToken) { + lockedAuth.tokens = { + ...(lockedAuth.tokens ?? {}), + access_token: sameGrantFreshCredential.accessToken, + refresh_token: sameGrantFreshCredential.refreshToken, + refresh_grant_fingerprint: refreshGrantFingerprint, + account_id: sameGrantFreshCredential.chatgptAccountId, + }; + persistMainAuthJson(lockedAuth); + return refreshedMainResult(lockedAuth); + } + } + const token = await (options.dependencies?.refreshToken ?? refreshChatGPTTokenRaw)(locked.refreshToken, { signal }); + const updatedAccessToken = token.access; + const updatedRefreshToken = token.refresh || locked.refreshToken; + publishFreshCredentialForGrant({ + refreshGrantFingerprint, + credential: { + accessToken: updatedAccessToken, + refreshToken: updatedRefreshToken, + expiresAt: token.expires, + chatgptAccountId: token.accountId ?? extractAccountId(token.idToken, updatedAccessToken) ?? locked.chatgptAccountId, + }, + excludeId: MAIN_CODEX_ACCOUNT_ID, + replaceAccessToken: rejectedAccessToken, + }); + lockedAuth.tokens = { + ...(lockedAuth.tokens ?? {}), + access_token: updatedAccessToken, + refresh_token: updatedRefreshToken, + refresh_grant_fingerprint: refreshGrantFingerprint, + ...(token.idToken ? { id_token: token.idToken } : {}), + account_id: token.accountId ?? extractAccountId(token.idToken, updatedAccessToken) ?? locked.chatgptAccountId, + }; + persistMainAuthJson(lockedAuth); + return refreshedMainResult(lockedAuth); + }, + }); + if (refreshed) clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); + return refreshed; + } catch (error) { + if ( + error instanceof CodexCredentialRefreshLockTimeoutError + || error instanceof CodexCredentialRefreshBusyError + || error instanceof CodexCredentialRefreshStaleError + ) { + throw error; + } + if (error instanceof Error && (error.name === "AbortError" || error.name === "TimeoutError")) throw error; + const reason = tokenRefreshReason(error); + if (reason === "expired" || reason === "revoked") markAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); + throw new TokenRefreshError(reason, "Codex main token refresh failed; reauthenticate the main account."); + } +} + +export async function getValidMainAccountToken( + options: { dependencies?: NativeMainRefreshDependencies } = {}, +): Promise<{ accessToken: string; chatgptAccountId: string } | null> { + const auth = readMainAuthJson(); + const token = mainTokenFromAuth(auth); + if (!token) return null; + if (mainAccessTokenFresh(token.accessToken)) return { accessToken: token.accessToken, chatgptAccountId: token.chatgptAccountId }; + return forceRefreshMainAccountToken(token.accessToken, options); +} diff --git a/src/oauth/chatgpt.ts b/src/oauth/chatgpt.ts index f4ecc7f8a9..330604841c 100644 --- a/src/oauth/chatgpt.ts +++ b/src/oauth/chatgpt.ts @@ -2,9 +2,9 @@ import { OAuthCallbackFlow } from "./callback-server"; import type { OAuthController, OAuthCredentials } from "./types"; import { generatePKCE } from "./pkce"; -const CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"; +export const CHATGPT_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"; const AUTH_URL = "https://auth.openai.com/oauth/authorize"; -const TOKEN_URL = "https://auth.openai.com/oauth/token"; +export const CHATGPT_TOKEN_URL = "https://auth.openai.com/oauth/token"; const SCOPE = "openid profile email offline_access api.connectors.read api.connectors.invoke"; const CALLBACK_PORT = 1455; const CALLBACK_PATH = "/auth/callback"; @@ -46,9 +46,19 @@ export function extractEmail(idToken?: string, accessToken?: string): string | u return undefined; } -function credsFromToken(data: Record): OAuthCredentials { +export interface ChatGPTTokenResponse { + access: string; + refresh: string; + expires: number; + accountId?: string; + email?: string; + idToken?: string; +} + +function tokenResponseFromData(data: Record): ChatGPTTokenResponse { const idToken = typeof data.id_token === "string" ? data.id_token : undefined; - const accessToken = data.access_token as string; + const accessToken = typeof data.access_token === "string" ? data.access_token : undefined; + if (!accessToken) throw new Error("ChatGPT token response is missing a string access_token"); // ?? only guards null/undefined; NaN or a string expires_in would otherwise // produce a NaN expiry that never compares as expired, and a negative duration // would stamp an already-past expiry — both block refresh semantics. @@ -62,10 +72,22 @@ function credsFromToken(data: Record): OAuthCredentials { const expires = Number.isFinite(computedExpires) ? computedExpires : Date.now() + 3600 * 1000; return { access: accessToken, - refresh: (data.refresh_token as string) ?? "", + refresh: typeof data.refresh_token === "string" ? data.refresh_token : "", expires, accountId: extractAccountId(idToken, accessToken), email: extractEmail(idToken, accessToken), + idToken, + }; +} + +function credsFromToken(data: Record): OAuthCredentials { + const token = tokenResponseFromData(data); + return { + access: token.access, + refresh: token.refresh, + expires: token.expires, + accountId: token.accountId, + email: token.email, }; } @@ -88,7 +110,7 @@ export class ChatGPTOAuthFlow extends OAuthCallbackFlow { this.#verifier = pkce.verifier; const params = new URLSearchParams({ response_type: "code", - client_id: CLIENT_ID, + client_id: CHATGPT_CLIENT_ID, redirect_uri: redirectUri, scope: SCOPE, code_challenge: pkce.challenge, @@ -107,12 +129,12 @@ export class ChatGPTOAuthFlow extends OAuthCallbackFlow { async exchangeToken(code: string, _state: string, redirectUri: string): Promise { if (!this.#verifier) throw new Error("ChatGPT PKCE verifier not initialized"); - const resp = await fetch(TOKEN_URL, { + const resp = await fetch(CHATGPT_TOKEN_URL, { method: "POST", headers: { "Content-Type": "application/x-www-form-urlencoded" }, body: new URLSearchParams({ grant_type: "authorization_code", - client_id: CLIENT_ID, + client_id: CHATGPT_CLIENT_ID, code, redirect_uri: redirectUri, code_verifier: this.#verifier, @@ -143,19 +165,34 @@ export async function loginChatGPT(ctrl: OAuthController, opts?: { forceLogin?: // Note: uses form-urlencoded per OAuth 2.0 spec (RFC 6749 §6). // Codex-rs uses JSON for refresh — intentional divergence; both accepted by auth.openai.com. -export async function refreshChatGPTToken(refreshToken: string): Promise { - const resp = await fetch(TOKEN_URL, { +export async function refreshChatGPTTokenRaw( + refreshToken: string, + options: { signal?: AbortSignal } = {}, +): Promise { + const resp = await fetch(CHATGPT_TOKEN_URL, { method: "POST", headers: { "Content-Type": "application/x-www-form-urlencoded" }, body: new URLSearchParams({ grant_type: "refresh_token", - client_id: CLIENT_ID, + client_id: CHATGPT_CLIENT_ID, refresh_token: refreshToken, }).toString(), + signal: options.signal, }); if (!resp.ok) { const errDesc = await safeErrorDescription(resp); throw new Error(`ChatGPT refresh failed: ${resp.status} ${errDesc}`); } - return credsFromToken((await resp.json()) as Record); + return tokenResponseFromData((await resp.json()) as Record); +} + +export async function refreshChatGPTToken(refreshToken: string): Promise { + const token = await refreshChatGPTTokenRaw(refreshToken); + return { + access: token.access, + refresh: token.refresh, + expires: token.expires, + accountId: token.accountId, + email: token.email, + }; } diff --git a/src/routing/analytics.ts b/src/routing/analytics.ts index 9e7b5a0696..f0ee281f9b 100644 --- a/src/routing/analytics.ts +++ b/src/routing/analytics.ts @@ -117,6 +117,7 @@ interface Bucket extends AnalyticsBreakdownRow { const COOLDOWN_RECOVERY_KINDS = new Set([ "rate-limit-429", "key-429", + "codex-main-401", "oauth-401", "anthropic-oauth-429", ]); diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index adc9415ec6..3092cfaeb3 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -51,7 +51,7 @@ import { CodexPoolAuthenticationError, CodexThreadAffinityExpiredError, headersForCodexAuthContext, - materializeCodexUpstreamAuth, + materializeCodexUpstreamAuthAsync, CodexMainSubstitutionUnavailableError, isCodexAuthContextUsable, resolveCodexAuthContext, @@ -60,6 +60,7 @@ import { releaseCodexAuthContextProbeLease, type CodexAuthContext, } from "../../codex/auth-context"; +import type { NativeMainRefreshDependencies } from "../../codex/main-account"; import { formatCodexProviderForLog, recordCodexUpstreamOutcome, @@ -129,14 +130,32 @@ import { codexAccountGatedCanonicalWireModel, decodeRequestErrorResponse, handleResponses, + nativeMainRefreshFailureResponse, preAuthUpstreamHostCircuitKey, upstreamHostCircuitOpenResponse, usesCodexForwardPoolAuth, } from "./core"; +import { + CodexCredentialRefreshBusyError, + CodexCredentialRefreshLockTimeoutError, + CodexCredentialRefreshStaleError, + TokenRefreshError, +} from "../../codex/account-store"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers"; export const COMPACT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; +export type HandleResponsesCompactOptions = { + admission?: DataPlaneAdmission; + nativeMainRefreshDependencies?: NativeMainRefreshDependencies; +}; + +function compactOptions(admissionOrOptions?: DataPlaneAdmission | HandleResponsesCompactOptions): HandleResponsesCompactOptions { + if (!admissionOrOptions) return {}; + if ("kind" in admissionOrOptions && "source" in admissionOrOptions) return { admission: admissionOrOptions }; + return admissionOrOptions; +} + export function compactResponseTooLargeError(): Response { return new Response(JSON.stringify({ error: { @@ -165,14 +184,16 @@ async function resolveAlternateCompactContext(args: { selectedModelId: string | undefined; excludeAccountId: string | null; turnAdmissionLease?: AdmissionLease; + nativeMainRefreshDependencies?: NativeMainRefreshDependencies; }): Promise<{ authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } | null> { - const { req, config, route, selectedModelId, excludeAccountId, turnAdmissionLease } = args; + const { req, config, route, selectedModelId, excludeAccountId, turnAdmissionLease, nativeMainRefreshDependencies } = args; if (!route.codexAccountMode || !excludeAccountId) return null; try { const authCtx = await resolveCodexAuthContext(req.headers, config, route.codexAccountMode, { ...(selectedModelId ? { modelId: selectedModelId } : {}), excludeAccountId, beginCodexAccountSelection: codexAccountSelectionForTurn(turnAdmissionLease), + nativeMainRefreshDependencies, }); if (!authCtx.accountId || authCtx.accountId === excludeAccountId) return null; const provider = applyCodexAuthContextToProvider(route.provider, authCtx, route.codexAccountMode); @@ -273,8 +294,10 @@ export async function handleResponsesCompact( config: OcxConfig, logCtx: RequestLogContext, turnAdmissionLease?: AdmissionLease, - admission?: DataPlaneAdmission, + admissionOrOptions?: DataPlaneAdmission | HandleResponsesCompactOptions, ): Promise { + const options = compactOptions(admissionOrOptions); + const admission = options.admission; let body: unknown; try { body = await readJsonRequestBody(req); @@ -380,9 +403,17 @@ export async function handleResponsesCompact( modelId: selectedModelId, substituteMainCredentialForDirect: substituteMainCredential, beginCodexAccountSelection: codexAccountSelectionForTurn(turnAdmissionLease), + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); logCtx.accountLogLabel = codexAuthContextLogLabel(authCtx, config); - const selected = materializeCodexUpstreamAuth(req.headers, authCtx, { substituteMainCredential }); + const selected = await materializeCodexUpstreamAuthAsync({ + headers: req.headers, + ctx: authCtx, + options: { + substituteMainCredential, + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, + }, + }); compactProvider = applyCodexAuthContextToProvider(route.provider, authCtx, route.codexAccountMode); for (const name of FORWARD_HEADERS) { const value = selected.get(name); @@ -395,6 +426,12 @@ export async function handleResponsesCompact( } } } catch (err) { + if ( + err instanceof TokenRefreshError + || err instanceof CodexCredentialRefreshLockTimeoutError + || err instanceof CodexCredentialRefreshBusyError + || err instanceof CodexCredentialRefreshStaleError + ) return nativeMainRefreshFailureResponse(err, req.signal); if (err instanceof CodexAccountCooldownError) { return cooldownErrorResponse(err, Date.now(), route.codexAccountNamespace); } @@ -585,6 +622,7 @@ export async function handleResponsesCompact( selectedModelId, excludeAccountId: authCtx.accountId, turnAdmissionLease, + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); // Resolution can await a credential refresh, so the client may have gone away // while we were choosing B. Re-check before spending anything: recording A, @@ -692,7 +730,12 @@ export async function handleResponsesCompact( headers: internalHeaders, body: JSON.stringify(internalBody), }); - const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, ...(admission ? { admission } : {}) }); + const response = await handleResponses(internalReq, config, logCtx, { + abortSignal: req.signal, + turnAdmissionLease, + ...(admission ? { admission } : {}), + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, + }); if (!response.ok) return response; let json: { output?: unknown[]; status?: unknown; error?: unknown }; if (response.headers.get("content-type")?.includes("text/event-stream")) { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 22bf3c18c3..f0aa04771c 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -121,6 +121,7 @@ import { CodexThreadAffinityExpiredError, headersForCodexAuthContext, materializeCodexUpstreamAuth, + materializeCodexUpstreamAuthAsync, CodexMainSubstitutionUnavailableError, isCodexAuthContextUsable, resolveCodexAuthContext, @@ -130,6 +131,14 @@ import { stripCodexRuntimeProviderFields, type CodexAuthContext, } from "../../codex/auth-context"; +import { forceRefreshMainAccountToken } from "../../codex/main-account"; +import type { NativeMainRefreshDependencies } from "../../codex/main-account"; +import { + CodexCredentialRefreshBusyError, + CodexCredentialRefreshLockTimeoutError, + CodexCredentialRefreshStaleError, + TokenRefreshError, +} from "../../codex/account-store"; import { entitledCodexAccountIdsForModel, invalidateCodexModelEntitlementsForAccount, @@ -1227,6 +1236,8 @@ export interface HandleResponsesOptions { onConsumedComboFailure?: (failure: ConsumedComboFailure) => void; /** Caller-owned for Chat/Claude replay; omitted only at genuine Responses ingress. */ translatorBudget?: TranslatorBudget; + /** Internal immutable refresh transport seam for native-main credential tests. */ + nativeMainRefreshDependencies?: NativeMainRefreshDependencies; } @@ -1239,6 +1250,23 @@ export function clientCancelledResponse(): Response { return formatErrorResponse(499, "client_cancelled", "Client cancelled request"); } +export function nativeMainRefreshFailureResponse(error: unknown, signal?: AbortSignal): Response { + if (signal?.aborted) { + return clientCancelledResponse(); + } + if (error instanceof TokenRefreshError && (error.reason === "revoked" || error.reason === "expired")) { + return formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"); + } + if ( + error instanceof CodexCredentialRefreshLockTimeoutError + || error instanceof CodexCredentialRefreshBusyError + || error instanceof CodexCredentialRefreshStaleError + ) { + return formatErrorResponse(503, "upstream_error", "Native Codex credential refresh is temporarily unavailable"); + } + return formatErrorResponse(503, "upstream_error", "Native Codex credential refresh failed temporarily"); +} + export function sanitizedRetryAfter(value: string | null, now: number): string | undefined { @@ -1466,6 +1494,7 @@ async function resolveResponsesCodexAuth( modelId: route.modelId, substituteMainCredentialForDirect: substituteMainCredential, beginCodexAccountSelection: codexAccountSelectionForTurn(options.turnAdmissionLease), + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); options.onCodexAuthContextResolved?.(authCtx); } else { @@ -1482,10 +1511,25 @@ async function resolveResponsesCodexAuth( return { ok: true, authCtx, - headers: materializeCodexUpstreamAuth(req.headers, authCtx, { substituteMainCredential }), + headers: await materializeCodexUpstreamAuthAsync({ + headers: req.headers, + ctx: authCtx, + options: { + substituteMainCredential, + nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, + }, + }), substituteMainCredential, }; } catch (err) { + if ( + err instanceof TokenRefreshError + || err instanceof CodexCredentialRefreshLockTimeoutError + || err instanceof CodexCredentialRefreshBusyError + || err instanceof CodexCredentialRefreshStaleError + ) { + return { ok: false, response: nativeMainRefreshFailureResponse(err, options.abortSignal) }; + } if (err instanceof CodexAccountCooldownError) { return { ok: false, response: cooldownErrorResponse(err, Date.now(), route.codexAccountNamespace) }; } @@ -3051,6 +3095,7 @@ async function handleResponsesInner( } const opaqueBlobRecoveryGuard: OpaqueBlobRecoveryGuard = { attempted: false }; + let codexMain401ReplayAttempted = false; let oauth401ReplayAttempted = false; const rateLimitPolicy = rateLimitRetryPolicyFor(route.provider); let rateLimitRetries = 0; @@ -3351,6 +3396,90 @@ async function handleResponsesInner( } } } + // Intentional duplication with the recovery-loop branch below: consolidating these paths + // would introduce connascence to two distant request builders and weaken each branch's + // contract stability; their distance keeps pre-stream and continuation cleanup explicit. + if ( + upstreamResponse.status === 401 + && authCtx.kind === "main-pool" + && usesCodexForwardPoolAuth(authCtx, route.provider) + && !codexMain401ReplayAttempted + ) { + codexMain401ReplayAttempted = true; + try { await upstreamResponse.body?.cancel(); } catch { /* already consumed/closed */ } + let refreshed: { accessToken: string; chatgptAccountId: string } | null; + try { + refreshed = await forceRefreshMainAccountToken(authCtx.accessToken, { + signal: options.abortSignal, + dependencies: options.nativeMainRefreshDependencies, + }); + } catch (error) { + upstream.abort(); + return nativeMainRefreshFailureResponse(error, options.abortSignal); + } + if (!refreshed) { + upstream.abort(); + return formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"); + } + authCtx = { + ...authCtx, + accessToken: refreshed.accessToken, + chatgptAccountId: refreshed.chatgptAccountId, + }; + selectedForwardHeaders = headersForCodexAuthContext(req.headers, authCtx); + const retryProvider = applyCodexAuthContextToProvider( + stripCodexRuntimeProviderFields(route.provider), + authCtx, + route.codexAccountMode, + ); + route.provider = retryProvider; + const retryAdapter = resolveAdapter( + resolveWireProtocolOverride(route.providerName, route.modelId, retryProvider, inboundWire), + config.cacheRetention, + ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: retryProvider, + adapterName: retryAdapter.name, + codexAuthContext: authCtx, + forwardHeaders: selectedForwardHeaders, + }); + request = await retryAdapter.buildRequest(parsed, { + headers: selectedForwardHeaders, + translatorBudget, + }); + recordAdapterReasoning(logCtx, request); + recordAdapterTier(logCtx, request); + const retryEstimate = typeof request.usageLog?.inputTokens === "number" + ? request.usageLog.inputTokens + : passthroughEstimate; + if (retryEstimate !== undefined) logCtx.usageLogInputTokens = retryEstimate; + logCtx.accountLogLabel = codexAuthContextLogLabel(authCtx, config); + sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, retryAdapter.name, logCtx.accountLogLabel); + noteAttemptSend(logCtx.activeAttempt, retryEstimate, "codex-main-401"); + try { + upstreamResponse = await fetchWithHeaderTimeout(request.url, { + method: request.method, + headers: request.headers, + body: request.body, + }, upstream.signal, connectMs, parsed.stream, + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + providerName: route.providerName, + modelId: route.modelId, + }), + route.provider.authMode === "forward") + .then(res => { + settleObservedHostResponse(); + return res; + }); + } catch (err) { + return transportFailureResponse(err); + } finally { + request.releaseBodyObservation?.(); + } + continue passthroughRecovery; + } // The deterministic route record cannot classify history it never observed (restart, expiry, // eviction, or an older transcript). Inspect only a bounded clone of a 4xx whose exact outbound // Responses body still carries opaque state, then rebuild once through the ordinary adapter @@ -4457,6 +4586,7 @@ async function handleResponsesInner( // 413→429 rotation cannot silently undo the tightening. let imageRetryAttempted = false; const opaqueBlobRecoveryGuard: OpaqueBlobRecoveryGuard = { attempted: false }; + let codexMain401ReplayAttempted = false; let oauth401ReplayAttempted = false; /** * Rebuild the request from the current parsed input (and any image-tier bias) and refetch @@ -4537,6 +4667,62 @@ async function handleResponsesInner( }; // Keep recovery kinds in sync with the native Responses `passthroughRecovery:` loop above. recovery: for (;;) { + // See the pre-stream branch above: this continuation owns different cleanup and replay + // state, so preserving distance avoids connascence and protects contract stability. + if ( + upstreamResponse.status === 401 + && authCtx.kind === "main-pool" + && usesCodexForwardPoolAuth(authCtx, route.provider) + && !codexMain401ReplayAttempted + ) { + codexMain401ReplayAttempted = true; + try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already consumed/closed */ } + let refreshed: { accessToken: string; chatgptAccountId: string } | null; + try { + refreshed = await forceRefreshMainAccountToken(authCtx.accessToken, { + signal: options.abortSignal, + dependencies: options.nativeMainRefreshDependencies, + }); + } catch (error) { + cleanupUpstreamAbort(); + return nativeMainRefreshFailureResponse(error, options.abortSignal); + } + if (!refreshed) { + cleanupUpstreamAbort(); + return formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"); + } + authCtx = { + ...authCtx, + accessToken: refreshed.accessToken, + chatgptAccountId: refreshed.chatgptAccountId, + }; + selectedForwardHeaders = headersForCodexAuthContext(req.headers, authCtx); + const refreshedProvider = applyCodexAuthContextToProvider( + stripCodexRuntimeProviderFields(route.provider), + authCtx, + route.codexAccountMode, + ); + route.provider = refreshedProvider; + logCtx.accountLogLabel = codexAuthContextLogLabel(authCtx, config); + invalidateSameTargetRequest(); + activeAdapter = resolveAdapter( + resolveWireProtocolOverride(route.providerName, route.modelId, refreshedProvider, inboundWire), + config.cacheRetention, + ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: refreshedProvider, + adapterName: activeAdapter.name, + codexAuthContext: authCtx, + forwardHeaders: selectedForwardHeaders, + }); + const result = await rebuildAndRefetch("codex-main-401"); + if ("failed" in result) return result.failed; + upstreamResponse = result; + continue recovery; + } + if ( upstreamResponse.status === 401 && isOAuth401ReplayProvider diff --git a/src/usage/log.ts b/src/usage/log.ts index 66654b8b74..1fed09c28b 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -24,6 +24,7 @@ export function isCodexUsageAccountLogLabel(value: unknown): value is CodexUsage export type AttemptRecoveryKind = | "transient-5xx" | "connection-reset" + | "codex-main-401" | "oauth-401" | "key-429" | "rate-limit-429" @@ -214,6 +215,7 @@ function normalizeUsageValue(usage: OcxUsage | undefined): OcxUsage | undefined const ATTEMPT_RECOVERY_KINDS = new Set([ "transient-5xx", "connection-reset", + "codex-main-401", "oauth-401", "key-429", "rate-limit-429", diff --git a/tests/codex-account-store.test.ts b/tests/codex-account-store.test.ts index 63f19cf0b3..d6f8867778 100644 --- a/tests/codex-account-store.test.ts +++ b/tests/codex-account-store.test.ts @@ -232,7 +232,7 @@ describe("codex-account-store CRUD", () => { expect(JSON.stringify(record)).not.toContain("sensitive-access revoked"); }); - test("successful refresh returns bumped generation and persists rotated refresh token", async () => { + test("successful refresh returns bumped generation and preserves the logical refresh grant", async () => { const { getCodexAccountCredential, getValidCodexToken, @@ -254,8 +254,8 @@ describe("codex-account-store CRUD", () => { const result = await getValidCodexToken("refresh-success"); expect(result).toEqual({ accessToken: "new", chatgptAccountId: "acc", generation: startGeneration + 1 }); expect(getCodexAccountCredential("refresh-success")).toMatchObject({ accessToken: "new", refreshToken: "new-r" }); - expect(readCodexAccountRecord("refresh-success")!.refreshGrantFingerprint).not.toBe(startFingerprint); - expect(readCodexAccountRecord("refresh-success")!.refreshGrantFingerprint).toBe(refreshGrantFingerprintForToken("new-r")); + expect(readCodexAccountRecord("refresh-success")!.refreshGrantFingerprint).toBe(startFingerprint); + expect(readCodexAccountRecord("refresh-success")!.refreshGrantFingerprint).toBe(refreshGrantFingerprintForToken("old-r")); } finally { globalThis.fetch = originalFetch; } diff --git a/tests/codex-main-account-refresh.test.ts b/tests/codex-main-account-refresh.test.ts new file mode 100644 index 0000000000..b7c15893f9 --- /dev/null +++ b/tests/codex-main-account-refresh.test.ts @@ -0,0 +1,236 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { clearAccountNeedsReauth, isAccountNeedsReauth } from "../src/codex/auth-api"; +import { MAIN_CODEX_ACCOUNT_ID } from "../src/codex/account-id"; +import { + TokenRefreshError, + getValidCodexToken, + readCodexAccountRecord, + refreshGrantFingerprintForToken, + saveCodexAccountCredential, + withCodexRefreshFileLock, +} from "../src/codex/account-store"; +import { + forceRefreshMainAccountToken, + getValidMainAccountToken, + isMainAccountCredentialUsable, + type NativeMainRefreshDependencies, +} from "../src/codex/main-account"; + +let testDir: string; +let codexHome: string; +let previousOpencodexHome: string | undefined; +let previousCodexHome: string | undefined; + +function jwt(expSecondsFromNow: number): string { + const payload = Buffer.from(JSON.stringify({ exp: Math.floor(Date.now() / 1000) + expSecondsFromNow })).toString("base64url"); + return `header.${payload}.signature`; +} + +function writeAuth(payload: Record): void { + writeFileSync(join(codexHome, "auth.json"), JSON.stringify(payload, null, 2) + "\n"); +} + +function refreshDependencies(accessToken: string): NativeMainRefreshDependencies { + return Object.freeze({ + refreshToken: async () => ({ access: accessToken, refresh: "rotated-refresh", expires: Date.now() + 3_600_000, accountId: "main-account" }), + }); +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} + +beforeEach(() => { + testDir = mkdtempSync(join(tmpdir(), "ocx-main-refresh-")); + codexHome = join(testDir, "codex"); + previousOpencodexHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = testDir; + process.env.CODEX_HOME = codexHome; + mkdirSync(codexHome, { recursive: true }); + clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); +}); + +afterEach(() => { + clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + rmSync(testDir, { recursive: true, force: true }); +}); + +describe("native main credential refresh", () => { + test("keeps an expired credential selectable when its refresh grant is valid", () => { + writeAuth({ tokens: { access_token: jwt(-60), refresh_token: "native-refresh", account_id: "main-account" } }); + expect(isMainAccountCredentialUsable()).toBe(true); + }); + + test("atomically persists rotated fields while preserving unrelated auth fields", async () => { + const freshAccess = jwt(3_600); + const grantFingerprint = refreshGrantFingerprintForToken("native-refresh"); + writeAuth({ + retained: { profile: "keep" }, + tokens: { access_token: jwt(-60), refresh_token: "native-refresh", account_id: "main-account", retained: "keep" }, + }); + const credential = await getValidMainAccountToken({ dependencies: refreshDependencies(freshAccess) }); + const persisted = JSON.parse(readFileSync(join(codexHome, "auth.json"), "utf8")) as Record; + expect(credential).toEqual({ accessToken: freshAccess, chatgptAccountId: "main-account" }); + expect(persisted.retained).toEqual({ profile: "keep" }); + expect(persisted.tokens).toMatchObject({ + access_token: freshAccess, + refresh_token: "rotated-refresh", + refresh_grant_fingerprint: grantFingerprint, + retained: "keep", + }); + expect(isAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID)).toBe(false); + }); + + test("does not partially overwrite malformed refresh results", async () => { + const before = { tokens: { access_token: jwt(-60), refresh_token: "native-refresh", account_id: "main-account", retained: "keep" } }; + writeAuth(before); + const dependencies: NativeMainRefreshDependencies = Object.freeze({ + refreshToken: async () => { throw new TokenRefreshError("unknown", "malformed refresh result"); }, + }); + await expect(getValidMainAccountToken({ dependencies })).rejects.toBeInstanceOf(TokenRefreshError); + expect(JSON.parse(readFileSync(join(codexHome, "auth.json"), "utf8"))).toEqual(before); + }); + + test("marks terminal revoked grants for reauthentication", async () => { + writeAuth({ tokens: { access_token: jwt(-60), refresh_token: "native-refresh", account_id: "main-account" } }); + const dependencies: NativeMainRefreshDependencies = Object.freeze({ + refreshToken: async () => { throw new TokenRefreshError("revoked", "revoked"); }, + }); + await expect(getValidMainAccountToken({ dependencies })).rejects.toBeInstanceOf(TokenRefreshError); + expect(isAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID)).toBe(true); + }); + + test("serializes native refresh behind the shared grant lock", async () => { + writeAuth({ tokens: { access_token: jwt(-60), refresh_token: "native-refresh", account_id: "main-account" } }); + const entered = deferred(); + const release = deferred(); + const lock = withCodexRefreshFileLock({ + lockKey: refreshGrantFingerprintForToken("native-refresh"), + signal: AbortSignal.timeout(5_000), + run: async () => { + entered.resolve(); + await release.promise; + }, + }); + await entered.promise; + let refreshCalls = 0; + try { + await expect(forceRefreshMainAccountToken(undefined, { + signal: AbortSignal.timeout(20), + dependencies: Object.freeze({ + refreshToken: async () => { + refreshCalls += 1; + return { access: jwt(3_600), refresh: "rotated-refresh", expires: Date.now() + 3_600_000, accountId: "main-account" }; + }, + }), + })).rejects.toBeInstanceOf(DOMException); + expect(refreshCalls).toBe(0); + } finally { + release.resolve(); + await lock; + } + }); + + test("publishes native-first refreshes to stored accounts sharing the grant", async () => { + const freshAccess = jwt(3_600); + const secondFreshAccess = jwt(3_700); + writeAuth({ tokens: { access_token: jwt(-60), refresh_token: "shared-refresh", account_id: "main-account" } }); + saveCodexAccountCredential("pool-shared", { + accessToken: "expired-pool", + refreshToken: "shared-refresh", + expiresAt: Date.now() - 60_000, + chatgptAccountId: "pool-account", + }); + + await forceRefreshMainAccountToken(undefined, { dependencies: refreshDependencies(freshAccess) }); + const poolToken = await getValidCodexToken("pool-shared"); + const persistedAfterFirst = JSON.parse(readFileSync(join(codexHome, "auth.json"), "utf8")) as Record; + + expect(poolToken.accessToken).toBe(freshAccess); + expect(poolToken.chatgptAccountId).toBe("main-account"); + expect(persistedAfterFirst.tokens.refresh_grant_fingerprint).toBe(refreshGrantFingerprintForToken("shared-refresh")); + + writeAuth({ + tokens: { + ...persistedAfterFirst.tokens, + access_token: jwt(-60), + }, + }); + const second = await forceRefreshMainAccountToken(undefined, { + dependencies: Object.freeze({ + refreshToken: async () => ({ access: secondFreshAccess, refresh: "second-rotated-refresh", expires: Date.now() + 3_600_000, accountId: "main-account" }), + }), + }); + const poolAfterSecond = await getValidCodexToken("pool-shared"); + + expect(second?.accessToken).toBe(freshAccess); + expect(poolAfterSecond.accessToken).toBe(freshAccess); + }); + + test("adopts stored-first refreshes into native auth without another refresh", async () => { + const freshAccess = jwt(3_600); + writeAuth({ tokens: { access_token: jwt(-60), refresh_token: "shared-refresh", account_id: "main-account" } }); + saveCodexAccountCredential("pool-shared", { + accessToken: "expired-pool", + refreshToken: "shared-refresh", + expiresAt: Date.now() - 60_000, + chatgptAccountId: "pool-account", + }); + const stored = await getValidCodexToken("pool-shared", { + fetch: async () => Response.json({ access_token: freshAccess, refresh_token: "rotated-refresh", expires_in: 3600 }), + }); + const record = readCodexAccountRecord("pool-shared"); + + const refreshed = await forceRefreshMainAccountToken(undefined, { + dependencies: Object.freeze({ + refreshToken: async () => { + throw new Error("native refresh must not run when a stored same-grant credential is fresh"); + }, + }), + }); + const persisted = JSON.parse(readFileSync(join(codexHome, "auth.json"), "utf8")) as Record; + + expect(stored.accessToken).toBe(freshAccess); + expect(record?.refreshGrantFingerprint).toBe(refreshGrantFingerprintForToken("shared-refresh")); + expect(refreshed).toEqual({ accessToken: freshAccess, chatgptAccountId: "pool-account" }); + expect(persisted.tokens).toMatchObject({ + access_token: freshAccess, + refresh_token: "rotated-refresh", + refresh_grant_fingerprint: refreshGrantFingerprintForToken("shared-refresh"), + account_id: "pool-account", + }); + }); + + test("401 refresh ignores a same-grant stored credential carrying the rejected bearer", async () => { + const freshAccess = jwt(3_600); + const rejectedAccess = jwt(3_500); + writeAuth({ tokens: { access_token: rejectedAccess, refresh_token: "shared-refresh", account_id: "main-account" } }); + saveCodexAccountCredential("pool-shared", { + accessToken: rejectedAccess, + refreshToken: "shared-refresh", + expiresAt: Date.now() + 3_600_000, + chatgptAccountId: "pool-account", + }); + const refreshed = await forceRefreshMainAccountToken(rejectedAccess, { + dependencies: refreshDependencies(freshAccess), + }); + const persisted = JSON.parse(readFileSync(join(codexHome, "auth.json"), "utf8")) as Record; + const record = readCodexAccountRecord("pool-shared"); + + expect(refreshed?.accessToken).toBe(freshAccess); + expect(persisted.tokens.access_token).toBe(freshAccess); + expect(record?.credential?.accessToken).toBe(freshAccess); + expect(record?.credential?.refreshToken).toBe("rotated-refresh"); + expect(record?.refreshGrantFingerprint).toBe(refreshGrantFingerprintForToken("shared-refresh")); + }); +}); diff --git a/tests/codex-refresh-file-lock.test.ts b/tests/codex-refresh-file-lock.test.ts new file mode 100644 index 0000000000..f1431bc739 --- /dev/null +++ b/tests/codex-refresh-file-lock.test.ts @@ -0,0 +1,158 @@ +import { describe, expect, test } from "bun:test"; +import { createHash } from "node:crypto"; +import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { withCodexRefreshFileLock } from "../src/codex/account-store"; + +function lockPath(directory: string, key: string): string { + const digest = createHash("sha256").update(key).digest("hex").slice(0, 32); + return join(directory, `codex-refresh-${digest}.lock`); +} + +describe("CODEX_HOME refresh file lock", () => { + test("does not steal a fresh reclaim owner", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "fresh-reclaim"; + const path = lockPath(directory, key); + writeFileSync(`${path}.reclaim`, JSON.stringify({ owner: "fresh", pid: process.pid, acquiredAt: Date.now() }) + "\n"); + const abort = AbortSignal.abort(new DOMException("cancelled", "AbortError")); + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: abort, + run: async () => undefined, + directory, + })).rejects.toBeInstanceOf(DOMException); + expect(readdirSync(directory)).toContain(`${path.split("/").at(-1)}.reclaim`); + rmSync(directory, { recursive: true, force: true }); + }); + + test("does not replace a reclaim owner while waiting for a primary lock", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "fresh-reclaim-wait"; + const path = lockPath(directory, key); + const reclaimOwner = { owner: "fresh-reclaim-owner", pid: process.pid, acquiredAt: Date.now() }; + writeFileSync(path, JSON.stringify({ owner: "primary-owner", pid: 0, acquiredAt: 0 }) + "\n"); + writeFileSync(`${path}.reclaim`, JSON.stringify(reclaimOwner) + "\n"); + + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: AbortSignal.timeout(20), + run: async () => undefined, + directory, + })).rejects.toBeInstanceOf(DOMException); + expect(JSON.parse(readFileSync(`${path}.reclaim`, "utf8"))).toEqual(reclaimOwner); + rmSync(directory, { recursive: true, force: true }); + }); + + test("release contention follows the caller signal bound", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "release-contention"; + const path = lockPath(directory, key); + let runEntered = false; + + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: AbortSignal.timeout(20), + run: async () => { + runEntered = true; + writeFileSync(`${path}.reclaim`, JSON.stringify({ owner: "fresh", pid: process.pid, acquiredAt: Date.now() }) + "\n"); + }, + directory, + })).rejects.toBeInstanceOf(DOMException); + expect(runEntered).toBe(true); + expect(readdirSync(directory).sort()).toEqual([`${path.split("/").at(-1)}`, `${path.split("/").at(-1)}.reclaim`].sort()); + rmSync(directory, { recursive: true, force: true }); + }); + + test("release contention honors an already expired caller signal", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "release-expired-signal"; + const path = lockPath(directory, key); + const controller = new AbortController(); + let runEntered = false; + + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: controller.signal, + run: async () => { + runEntered = true; + writeFileSync(`${path}.reclaim`, JSON.stringify({ owner: "fresh", pid: process.pid, acquiredAt: Date.now() }) + "\n"); + controller.abort(new DOMException("deadline", "TimeoutError")); + }, + directory, + })).rejects.toBeInstanceOf(DOMException); + expect(runEntered).toBe(true); + expect(readdirSync(directory).sort()).toEqual([`${path.split("/").at(-1)}`, `${path.split("/").at(-1)}.reclaim`].sort()); + rmSync(directory, { recursive: true, force: true }); + }); + + test("release contention cleanup lets a later caller acquire after reclaim owner exits", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "release-eventual-cleanup"; + const path = lockPath(directory, key); + const controller = new AbortController(); + + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: controller.signal, + run: async () => { + writeFileSync(`${path}.reclaim`, JSON.stringify({ owner: "fresh", pid: process.pid, acquiredAt: Date.now() }) + "\n"); + controller.abort(new DOMException("deadline", "TimeoutError")); + }, + directory, + })).rejects.toBeInstanceOf(DOMException); + rmSync(`${path}.reclaim`, { force: true }); + + await withCodexRefreshFileLock({ + lockKey: key, + signal: AbortSignal.timeout(2_000), + run: async () => undefined, + directory, + }); + expect(readdirSync(directory)).toEqual([]); + rmSync(directory, { recursive: true, force: true }); + }); + + test("release contention cleanup survives reclaim ownership beyond one retry", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "release-long-contention"; + const path = lockPath(directory, key); + const controller = new AbortController(); + + await expect(withCodexRefreshFileLock({ + lockKey: key, + signal: controller.signal, + run: async () => { + writeFileSync(`${path}.reclaim`, JSON.stringify({ owner: "fresh", pid: process.pid, acquiredAt: Date.now() }) + "\n"); + controller.abort(new DOMException("deadline", "TimeoutError")); + setTimeout(() => rmSync(`${path}.reclaim`, { force: true }), 1_200); + }, + directory, + })).rejects.toBeInstanceOf(DOMException); + + await withCodexRefreshFileLock({ + lockKey: key, + signal: AbortSignal.timeout(4_000), + run: async () => undefined, + directory, + }); + expect(readdirSync(directory)).toEqual([]); + rmSync(directory, { recursive: true, force: true }); + }); + + test("quarantines malformed debris and releases owner-safe locks", async () => { + const directory = mkdtempSync(join(tmpdir(), "ocx-refresh-lock-")); + const key = "malformed-orphan"; + const path = lockPath(directory, key); + writeFileSync(path, "not-json\n"); + await withCodexRefreshFileLock({ + lockKey: key, + signal: AbortSignal.timeout(5_000), + run: async () => undefined, + directory, + }); + expect(readdirSync(directory)).toEqual([]); + rmSync(directory, { recursive: true, force: true }); + }); +}); diff --git a/tests/responses-compact-native-main-refresh.test.ts b/tests/responses-compact-native-main-refresh.test.ts new file mode 100644 index 0000000000..3a6d7a5fb7 --- /dev/null +++ b/tests/responses-compact-native-main-refresh.test.ts @@ -0,0 +1,61 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { NativeMainRefreshDependencies } from "../src/codex/main-account"; +import { handleResponsesCompact } from "../src/server/responses"; +import type { OcxConfig } from "../src/types"; + +let directory: string; +let previousHome: string | undefined; + +function jwt(offset: number): string { + return `header.${Buffer.from(JSON.stringify({ exp: Math.floor(Date.now() / 1000) + offset })).toString("base64url")}.signature`; +} + +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), "ocx-compact-refresh-")); + previousHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = directory; + mkdirSync(directory, { recursive: true }); +}); + +afterEach(() => { + if (previousHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousHome; + rmSync(directory, { recursive: true, force: true }); +}); + +test("substitutes a refreshed native credential before compact upstream I/O", async () => { + const fresh = jwt(3_600); + const observedBearers: string[] = []; + const upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: req => { + observedBearers.push(req.headers.get("authorization") ?? ""); + return Response.json({ id: "compact_1", output: [] }); + } }); + try { + writeFileSync(join(directory, "auth.json"), JSON.stringify({ tokens: { access_token: jwt(-60), refresh_token: "refresh", account_id: "main" } })); + const config = { + defaultProvider: "openai", + providers: { openai: { + adapter: "openai-responses", baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward", codexAccountMode: "direct", + fetch: (_input: RequestInfo | URL, init?: RequestInit) => fetch(`http://127.0.0.1:${upstream.port}`, init), + } }, + codexAccounts: [], + } as unknown as OcxConfig; + const dependencies: NativeMainRefreshDependencies = Object.freeze({ + refreshToken: async () => ({ access: fresh, refresh: "refresh-2", expires: Date.now() + 3_600_000, accountId: "main" }), + }); + const response = await handleResponsesCompact(new Request("http://localhost/v1/responses/compact", { + method: "POST", headers: { "content-type": "application/json", authorization: "Bearer admission" }, + body: JSON.stringify({ model: "openai/gpt-5", input: [] }), + }), config, { model: "", provider: "" }, undefined, { + admission: { source: "bearer" } as never, + nativeMainRefreshDependencies: dependencies, + }); + expect(response.status).toBe(200); + expect(observedBearers).toEqual([`Bearer ${fresh}`]); + } finally { + upstream.stop(true); + } +}); diff --git a/tests/responses-native-main-refresh.test.ts b/tests/responses-native-main-refresh.test.ts new file mode 100644 index 0000000000..0805244c90 --- /dev/null +++ b/tests/responses-native-main-refresh.test.ts @@ -0,0 +1,65 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { MAIN_CODEX_ACCOUNT_ID } from "../src/codex/account-id"; +import type { NativeMainRefreshDependencies } from "../src/codex/main-account"; +import { handleResponses } from "../src/server/responses"; +import type { OcxConfig } from "../src/types"; + +let directory: string; +let previousHome: string | undefined; + +function jwt(offset: number): string { + return `header.${Buffer.from(JSON.stringify({ exp: Math.floor(Date.now() / 1000) + offset })).toString("base64url")}.signature`; +} + +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), "ocx-responses-refresh-")); + previousHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = directory; + mkdirSync(directory, { recursive: true }); +}); + +afterEach(() => { + if (previousHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousHome; + rmSync(directory, { recursive: true, force: true }); +}); + +test("replays one native-main 401 with the refreshed bearer", async () => { + const stale = jwt(3_600); + const fresh = jwt(7_200); + const observedBearers: string[] = []; + const upstream = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(req) { + observedBearers.push(req.headers.get("authorization") ?? ""); + if (observedBearers.length === 1) return Response.json({ error: { message: "expired" } }, { status: 401 }); + return Response.json({ id: "resp_1", object: "response", status: "completed", output: [] }); + }, + }); + try { + writeFileSync(join(directory, "auth.json"), JSON.stringify({ tokens: { access_token: stale, refresh_token: "refresh", account_id: "main" } })); + const config = { + defaultProvider: "openai", + activeCodexAccountId: MAIN_CODEX_ACCOUNT_ID, + providers: { openai: { + adapter: "openai-responses", baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward", codexAccountMode: "pool", + fetch: (_input: RequestInfo | URL, init?: RequestInit) => fetch(`http://127.0.0.1:${upstream.port}`, init), + } }, + codexAccounts: [], + } as unknown as OcxConfig; + const dependencies: NativeMainRefreshDependencies = Object.freeze({ + refreshToken: async () => ({ access: fresh, refresh: "refresh-2", expires: Date.now() + 3_600_000, accountId: "main" }), + }); + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "openai/gpt-5", input: "hello", stream: false }), + }), config, { model: "", provider: "" }, { nativeMainRefreshDependencies: dependencies }); + expect(response.status).toBe(200); + expect(observedBearers).toEqual([`Bearer ${stale}`, `Bearer ${fresh}`]); + } finally { + upstream.stop(true); + } +});