diff --git a/src/codex/account-store.ts b/src/codex/account-store.ts index 7784cbe2b..a0d8422b4 100644 --- a/src/codex/account-store.ts +++ b/src/codex/account-store.ts @@ -1,7 +1,15 @@ import { createHash } from "node:crypto"; import { closeSync, existsSync, readFileSync, mkdirSync, openSync, unlinkSync, writeFileSync } from "node:fs"; import { join } from "node:path"; -import { getConfigDir, atomicWriteFile, backupInvalidConfig, hardenConfigDir, hardenExistingSecret } from "../config"; +import { + ConfigMutationLockError, + getConfigDir, + atomicWriteFile, + backupInvalidConfig, + hardenConfigDir, + hardenExistingSecret, + withConfigMutationLockSync, +} from "../config"; import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; import type { CodexAccountCredentialRecord, CodexAccountCredentials } from "../types"; @@ -121,44 +129,50 @@ export function getCodexAccountCredential(id: string): CodexAccountCredentials | } export function saveCodexAccountCredential(id: string, cred: CodexAccountCredentials): void { - const store = loadCodexAccountRecordStore(); - const current = store[id]; - const refreshGrantFingerprint = current?.credential?.refreshToken === cred.refreshToken - ? current.refreshGrantFingerprint ?? refreshGrantFingerprintForToken(cred.refreshToken) - : refreshGrantFingerprintForToken(cred.refreshToken); - store[id] = { - credential: cred, - generation: (current?.generation ?? 0) + 1, - refreshGrantFingerprint, - replacedAt: current ? Date.now() : undefined, - ...preservedValidationMetadata(current), - }; - persist(store); + withCredentialMutationLockSync(() => { + const store = loadCodexAccountRecordStore(); + const current = store[id]; + const refreshGrantFingerprint = current?.credential?.refreshToken === cred.refreshToken + ? current.refreshGrantFingerprint ?? refreshGrantFingerprintForToken(cred.refreshToken) + : refreshGrantFingerprintForToken(cred.refreshToken); + store[id] = { + credential: cred, + generation: (current?.generation ?? 0) + 1, + refreshGrantFingerprint, + replacedAt: current ? Date.now() : undefined, + ...preservedValidationMetadata(current), + }; + persist(store); + }); } export function markCodexAccountValidated(id: string, atMs: number = Date.now()): void { - const store = loadCodexAccountRecordStore(); - const current = store[id]; - if (!current || current.deletedAt != null || !current.credential) return; - store[id] = { - ...current, - lastCodexValidatedAt: atMs, - lastCodexValidationStatus: "ok", - lastCodexValidationError: undefined, - }; - persist(store); + withCredentialMutationLockSync(() => { + const store = loadCodexAccountRecordStore(); + const current = store[id]; + if (!current || current.deletedAt != null || !current.credential) return; + store[id] = { + ...current, + lastCodexValidatedAt: atMs, + lastCodexValidationStatus: "ok", + lastCodexValidationError: undefined, + }; + persist(store); + }); } export function markCodexAccountValidationFailed(id: string, reason: string): void { - const store = loadCodexAccountRecordStore(); - const current = store[id]; - if (!current || current.deletedAt != null || !current.credential) return; - store[id] = { - ...current, - lastCodexValidationStatus: "failed", - lastCodexValidationError: reason, - }; - persist(store); + withCredentialMutationLockSync(() => { + const store = loadCodexAccountRecordStore(); + const current = store[id]; + if (!current || current.deletedAt != null || !current.credential) return; + store[id] = { + ...current, + lastCodexValidationStatus: "failed", + lastCodexValidationError: reason, + }; + persist(store); + }); } export function removeCodexAccountCredential(id: string): void { @@ -183,32 +197,36 @@ export function saveCodexAccountCredentialIfGeneration( generation: number, cred: CodexAccountCredentials, ): boolean { - const store = loadCodexAccountRecordStore(); - const current = store[id]; - 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); - store[id] = { - credential: cred, - generation: generation + 1, - refreshGrantFingerprint, - replacedAt: current.replacedAt, - ...preservedValidationMetadata(current), - }; - persist(store); - return true; + return withCredentialMutationLockSync(() => { + const store = loadCodexAccountRecordStore(); + const current = store[id]; + 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); + store[id] = { + credential: cred, + generation: generation + 1, + refreshGrantFingerprint, + replacedAt: current.replacedAt, + ...preservedValidationMetadata(current), + }; + persist(store); + return true; + }); } export function tombstoneCodexAccount(id: string): number { - const store = loadCodexAccountRecordStore(); - const current = store[id]; - const generation = (current?.generation ?? 0) + 1; - store[id] = { generation, deletedAt: Date.now() }; - persist(store); - return generation; + return withCredentialMutationLockSync(() => { + const store = loadCodexAccountRecordStore(); + const current = store[id]; + const generation = (current?.generation ?? 0) + 1; + store[id] = { generation, deletedAt: Date.now() }; + persist(store); + return generation; + }); } const CHATGPT_TOKEN_URL = "https://auth.openai.com/oauth/token"; @@ -237,6 +255,16 @@ export class CodexCredentialRefreshLockTimeoutError extends Error { } } +/** Credential writers share the config mutation coordinator; contention is transient, not reauth. */ +function withCredentialMutationLockSync(fn: () => T): T { + try { + return withConfigMutationLockSync(fn); + } catch (error) { + if (error instanceof ConfigMutationLockError) throw new CodexCredentialRefreshLockTimeoutError(); + throw error; + } +} + type CodexTokenResult = { accessToken: string; chatgptAccountId: string; generation: number }; type CodexRefreshResult = CodexTokenResult & { credential?: CodexAccountCredentials }; const refreshLocks = new Map>(); diff --git a/src/codex/auth-api.ts b/src/codex/auth-api.ts index f2cea1bde..3a925cbe2 100644 --- a/src/codex/auth-api.ts +++ b/src/codex/auth-api.ts @@ -1,10 +1,11 @@ -import { loadConfig, saveConfigPreservingClaudeCode } from "../config"; +import { ConfigMutationLockError, loadConfig, mutatePersistedConfig, saveConfigPreservingClaudeCode } from "../config"; import { withCodexAccountLogLabel } from "./account-label"; import { getCodexAccountCredential, getValidCodexToken, isCodexAccountGenerationLive, markCodexAccountValidated, + readCodexAccountRecord, saveCodexAccountCredential, CodexCredentialGenerationConflictError, CodexCredentialRefreshLockTimeoutError, @@ -413,6 +414,8 @@ async function fetchMainAccountInfoAttempt(forceRefresh: boolean, retriesRemaini interface PoolQuotaResult { quota: StoredAccountQuota | null; needsReauth: boolean; + /** Credential generation whose cache or network result this DTO state belongs to. */ + credentialGeneration?: number; /** Present only when this call freshly parsed a WHAM usage response. */ freshQuota?: Omit; /** Present only when this call's WHAM response included a non-empty `plan_type`. */ @@ -423,6 +426,16 @@ interface PoolQuotaResult { freshResetCredits?: number; } +interface PoolQuotaRefreshFlight { + state: { + startCredentialGeneration?: number; + resolvedCredentialGeneration?: number; + }; + promise: Promise; +} + +const poolQuotaRefreshInFlight = new Map>(); + export interface CodexAuthAccountDto { id: string; alias?: string; @@ -440,39 +453,157 @@ export interface CodexAuthAccountDto { healthAction?: string; } -async function fetchPoolAccountQuota(accountId: string, forceRefresh = false, configuredPlan?: string): Promise { - const existing = getAccountQuota(accountId); - if (!forceRefresh && existing && Date.now() - existing.updatedAt < POOL_CACHE_TTL) { - return { quota: existing, needsReauth: false }; +interface FreshPoolPlanUpdate { + accountId: string; + plan: string; + credentialGeneration: number; +} + +/** + * Persist only validated plan leaves against the latest disk snapshot. A quota GET must not save + * the long-lived runtime object wholesale: unrelated manual/provider writes may have landed while + * WHAM requests were in flight. Missing or malformed files fail closed: a read path must not + * recreate a deleted config from the server's older in-memory snapshot. + */ +function reconcileFreshPoolAccountPlans(runtimeConfig: OcxConfig, updates: FreshPoolPlanUpdate[]): void { + if (updates.length === 0) return; + let outcome: ReturnType>; + try { + outcome = mutatePersistedConfig(persistedConfig => { + const accepted: FreshPoolPlanUpdate[] = []; + let changed = false; + for (const update of updates) { + if (!isCodexAccountGenerationLive(update.accountId, update.credentialGeneration)) continue; + const liveAccount = configuredPoolAccount(runtimeConfig, update.accountId); + const persistedAccount = configuredPoolAccount(persistedConfig, update.accountId); + if (!liveAccount || !persistedAccount) continue; + accepted.push(update); + if (persistedAccount.plan !== update.plan) { + persistedAccount.plan = update.plan; + changed = true; + } + } + return { changed, value: accepted }; + }); + } catch (error) { + // Plan persistence is derived metadata on a read route. Contention must fail closed without + // turning account listing into a 500; a later refresh can retry against the latest files. + if (error instanceof ConfigMutationLockError) return; + throw error; + } + if (outcome.status === "unavailable") return; + for (const update of outcome.value) { + // A replacement immediately after the durable commit is allowed to supersede the result, but + // the long-lived object must never be updated from that stale generation. + if (!isCodexAccountGenerationLive(update.accountId, update.credentialGeneration)) continue; + const liveAccount = configuredPoolAccount(runtimeConfig, update.accountId); + if (liveAccount) { + liveAccount.plan = update.plan; + } } +} + +async function fetchFreshPoolAccountQuota( + accountId: string, + existing: StoredAccountQuota | null, + configuredPlan?: string, + onCredentialGeneration?: (generation: number) => void, +): Promise { + let requestCredentialGeneration = readCodexAccountRecord(accountId)?.generation; try { const { accessToken, chatgptAccountId, generation } = await getValidCodexToken(accountId); + requestCredentialGeneration = generation; + onCredentialGeneration?.(generation); const resp = await fetch("https://chatgpt.com/backend-api/wham/usage", { headers: { Authorization: `Bearer ${accessToken}`, "ChatGPT-Account-Id": chatgptAccountId }, signal: AbortSignal.timeout(8000), }); - if (!resp.ok) return { quota: existing ?? null, needsReauth: resp.status === 401 }; + if (!resp.ok) { + return { + quota: existing ?? null, + needsReauth: resp.status === 401, + credentialGeneration: generation, + }; + } const data = (await resp.json()) as WhamUsageResponse; const freshPlan = nonEmptyPlan(data.plan_type) ?? undefined; const quota = parseUsageQuota({ ...data, plan_type: freshPlan ?? configuredPlan }); const freshResetCredits = quota?.resetCredits; - if (!quota) return { quota: existing ?? null, needsReauth: false }; + if (!quota) { + return { + quota: isCodexAccountGenerationLive(accountId, generation) ? existing ?? null : getAccountQuota(accountId), + needsReauth: false, + credentialGeneration: generation, + ...(freshPlan !== undefined + ? { freshPlan, freshCredentialGeneration: generation } + : {}), + }; + } if (!isCodexAccountGenerationLive(accountId, generation)) { - return { quota: getAccountQuota(accountId), needsReauth: false }; + return { quota: null, needsReauth: false, credentialGeneration: generation }; } setAccountQuotaFromParsed(accountId, quota); return { quota: getAccountQuota(accountId), needsReauth: false, + credentialGeneration: generation, freshQuota: quota, freshCredentialGeneration: generation, ...(freshPlan !== undefined ? { freshPlan } : {}), ...(freshResetCredits !== undefined ? { freshResetCredits } : {}), }; } catch (e) { - if (e instanceof CodexCredentialGenerationConflictError || e instanceof CodexCredentialRefreshLockTimeoutError) return { quota: existing ?? null, needsReauth: false }; - if (e instanceof TokenRefreshError) return { quota: existing ?? null, needsReauth: true }; - return { quota: existing ?? null, needsReauth: false }; + if (e instanceof CodexCredentialGenerationConflictError || e instanceof CodexCredentialRefreshLockTimeoutError) { + return { quota: null, needsReauth: false, credentialGeneration: requestCredentialGeneration }; + } + if (e instanceof TokenRefreshError) { + return { quota: existing ?? null, needsReauth: true, credentialGeneration: requestCredentialGeneration }; + } + return { quota: existing ?? null, needsReauth: false, credentialGeneration: requestCredentialGeneration }; + } +} + +async function fetchPoolAccountQuota(accountId: string, forceRefresh = false, configuredPlan?: string): Promise { + const existing = getAccountQuota(accountId); + if (!forceRefresh && existing && Date.now() - existing.updatedAt < POOL_CACHE_TTL) { + return { + quota: existing, + needsReauth: false, + credentialGeneration: readCodexAccountRecord(accountId)?.generation, + }; + } + // A token refresh may increment the generation (and rotate the refresh token) before WHAM + // completes. Join a flight whose starting or resolved generation is still current, but let a + // replacement credential with the same pool id start its own request. + const record = readCodexAccountRecord(accountId); + const flights = poolQuotaRefreshInFlight.get(accountId); + const current = flights && [...flights].find(flight => { + const generation = flight.state.resolvedCredentialGeneration + ?? flight.state.startCredentialGeneration; + return generation !== undefined && isCodexAccountGenerationLive(accountId, generation); + }); + if (current) return current.promise; + + const state: PoolQuotaRefreshFlight["state"] = { + startCredentialGeneration: record?.generation, + }; + const refresh = fetchFreshPoolAccountQuota( + accountId, + existing, + configuredPlan, + generation => { state.resolvedCredentialGeneration = generation; }, + ); + const flight: PoolQuotaRefreshFlight = { state, promise: refresh }; + const activeFlights = flights ?? new Set(); + activeFlights.add(flight); + if (!flights) poolQuotaRefreshInFlight.set(accountId, activeFlights); + try { + return await refresh; + } finally { + activeFlights.delete(flight); + if (activeFlights.size === 0 && poolQuotaRefreshInFlight.get(accountId) === activeFlights) { + poolQuotaRefreshInFlight.delete(accountId); + } } } @@ -542,12 +673,54 @@ export async function listCodexAuthAccounts(config: OcxConfig, forceRefresh = fa const runtimeConfig = getRuntimeConfig(config); const poolAccounts = (runtimeConfig.codexAccounts ?? []).filter(isSelectableCodexPoolAccount); const mainInfo = await fetchMainAccountInfo(forceRefresh); - const withQuota = await mapWithConcurrency(poolAccounts, POOL_QUOTA_REFRESH_CONCURRENCY, async a => { - const cred = getCodexAccountCredential(a.id); + const refreshedPool = await mapWithConcurrency(poolAccounts, POOL_QUOTA_REFRESH_CONCURRENCY, async account => { + const cred = getCodexAccountCredential(account.id); const quotaResult = cred - ? await fetchPoolAccountQuota(a.id, forceRefresh, a.plan) + ? await fetchPoolAccountQuota(account.id, forceRefresh, account.plan) : { quota: null, needsReauth: true }; - return poolAccountDto(a, quotaResult, !!cred, isCodexAccountPaused(runtimeConfig, a.id)); + return { accountId: account.id, quotaResult }; + }); + + // WHAM plan_type is authoritative only for the credential generation that fetched it. Collect + // changes after every parallel read settles, then apply one narrow disk patch for the batch. + const planUpdates = refreshedPool.flatMap(({ accountId, quotaResult }): FreshPoolPlanUpdate[] => { + const plan = quotaResult.freshPlan; + const credentialGeneration = quotaResult.freshCredentialGeneration; + return plan && credentialGeneration !== undefined + ? [{ accountId, plan, credentialGeneration }] + : []; + }); + reconcileFreshPoolAccountPlans(runtimeConfig, planUpdates); + + const withQuota = refreshedPool.flatMap(({ accountId, quotaResult }) => { + const currentAccount = configuredPoolAccount(runtimeConfig, accountId); + if (!currentAccount) return []; + const currentCredential = getCodexAccountCredential(accountId); + if (!currentCredential) { + return [poolAccountDto( + currentAccount, + { quota: null, needsReauth: true }, + false, + isCodexAccountPaused(runtimeConfig, accountId), + )]; + } + const resultGeneration = quotaResult.credentialGeneration ?? quotaResult.freshCredentialGeneration; + const generationLive = resultGeneration === undefined + || isCodexAccountGenerationLive(accountId, resultGeneration); + const effectiveQuotaResult = !generationLive + ? { quota: null, needsReauth: false } + : quotaResult; + // Response DTO can show the WHAM plan even when disk persistence fails closed (lock busy / + // missing config). Persistence still remains generation-gated via reconcileFreshPoolAccountPlans. + const dtoAccount = generationLive && quotaResult.freshPlan + ? { ...currentAccount, plan: quotaResult.freshPlan } + : currentAccount; + return [poolAccountDto( + dtoAccount, + effectiveQuotaResult, + true, + isCodexAccountPaused(runtimeConfig, accountId), + )]; }); const hasMainCredential = readCodexTokens() !== null; const mainNeedsReauth = !hasMainCredential || isAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); @@ -1006,169 +1179,185 @@ export async function handleCodexAuthAPI( } (async () => { - let completed = false; - for (let i = 0; i < 150; i++) { - await new Promise(r => setTimeout(r, 2000)); - const st = getLoginStatus("chatgpt"); - if (st.done && st.loggedIn) { - const { getCredential } = await import("../oauth/store"); - const cred = getCredential("chatgpt"); - if (cred) { - const oauthAccountId = cred.accountId; - if (!oauthAccountId) { - codexAuthLoginState.set(flowId, { - status: "error", - error: "Could not determine account identity from OAuth tokens. Please retry OAuth login.", - doneAt: Date.now(), - }); - completed = true; - break; - } - - let email = cred.email || accountId; - let plan: string | undefined; - let quota: Omit | null = null; - try { - const tokens = { access_token: cred.access, account_id: oauthAccountId }; - const resp = await fetch("https://chatgpt.com/backend-api/wham/usage", { - headers: { Authorization: `Bearer ${tokens.access_token}`, "ChatGPT-Account-Id": tokens.account_id }, - signal: AbortSignal.timeout(8000), - }); - if (resp.ok) { - const data = (await resp.json()) as WhamUsageResponse; - email = data.email ?? email; - plan = data.plan_type ?? undefined; - quota = parseUsageQuota(data); + try { + let completed = false; + for (let i = 0; i < 150; i++) { + await new Promise(r => setTimeout(r, 2000)); + const st = getLoginStatus("chatgpt"); + if (st.done && st.loggedIn) { + const { getCredential } = await import("../oauth/store"); + const cred = getCredential("chatgpt"); + if (cred) { + const oauthAccountId = cred.accountId; + if (!oauthAccountId) { + codexAuthLoginState.set(flowId, { + status: "error", + error: "Could not determine account identity from OAuth tokens. Please retry OAuth login.", + doneAt: Date.now(), + }); + completed = true; + break; } - } catch { /* wham fetch is non-blocking */ } - // Reauth must refresh the same ChatGPT identity already bound to this pool slot. - // Otherwise a different login would silently overwrite credentials under a trusted id. - if (reauth) { - const existingCred = getCodexAccountCredential(accountId); - const poolAccount = configuredPoolAccount(getRuntimeConfig(config), accountId); - const expectedChatgptId = existingCred?.chatgptAccountId?.trim(); - const expectedEmail = poolAccount?.email?.trim().toLowerCase(); - const gotEmail = email.trim().toLowerCase(); - if (expectedChatgptId) { - if (expectedChatgptId !== oauthAccountId) { - codexAuthLoginState.set(flowId, { - status: "error", - error: "Signed-in ChatGPT account does not match this pool account. Sign in with the same account, or remove it and add a new one.", - doneAt: Date.now(), - }); - completed = true; - break; + + let email = cred.email || accountId; + let plan: string | undefined; + let quota: Omit | null = null; + try { + const tokens = { access_token: cred.access, account_id: oauthAccountId }; + const resp = await fetch("https://chatgpt.com/backend-api/wham/usage", { + headers: { Authorization: `Bearer ${tokens.access_token}`, "ChatGPT-Account-Id": tokens.account_id }, + signal: AbortSignal.timeout(8000), + }); + if (resp.ok) { + const data = (await resp.json()) as WhamUsageResponse; + email = data.email ?? email; + plan = nonEmptyPlan(data.plan_type) ?? undefined; + quota = parseUsageQuota(data); } - } else if (expectedEmail) { - if (!gotEmail || gotEmail !== expectedEmail) { + } catch { /* wham fetch is non-blocking */ } + // Reauth must refresh the same ChatGPT identity already bound to this pool slot. + // Otherwise a different login would silently overwrite credentials under a trusted id. + if (reauth) { + const existingCred = getCodexAccountCredential(accountId); + const poolAccount = configuredPoolAccount(getRuntimeConfig(config), accountId); + const expectedChatgptId = existingCred?.chatgptAccountId?.trim(); + const expectedEmail = poolAccount?.email?.trim().toLowerCase(); + const gotEmail = email.trim().toLowerCase(); + if (expectedChatgptId) { + if (expectedChatgptId !== oauthAccountId) { + codexAuthLoginState.set(flowId, { + status: "error", + error: "Signed-in ChatGPT account does not match this pool account. Sign in with the same account, or remove it and add a new one.", + doneAt: Date.now(), + }); + completed = true; + break; + } + } else if (expectedEmail) { + if (!gotEmail || gotEmail !== expectedEmail) { + codexAuthLoginState.set(flowId, { + status: "error", + error: "Signed-in ChatGPT account does not match this pool account. Sign in with the same account, or remove it and add a new one.", + doneAt: Date.now(), + }); + completed = true; + break; + } + } else { + // No chatgptAccountId and no pool email — refuse silent identity replacement + // (including empty credential slots that still have a pool row). codexAuthLoginState.set(flowId, { status: "error", - error: "Signed-in ChatGPT account does not match this pool account. Sign in with the same account, or remove it and add a new one.", + error: "Cannot verify account identity for reauth. Remove this account and add it again.", doneAt: Date.now(), }); completed = true; break; } - } else { - // No chatgptAccountId and no pool email — refuse silent identity replacement - // (including empty credential slots that still have a pool row). + } + + // 1.2: Duplicate check is scoped by personal vs workspace plan bucket. + const collision = checkAccountIdCollision(oauthAccountId, email, plan, reauth ? accountId : undefined); + if (collision.collision) { + codexAuthLoginState.set(flowId, { + status: "error", error: collision.reason, doneAt: Date.now(), + }); + completed = true; + break; + } + + const warmup = await verifyCodexAccountWarmup(accountId, cred.access, oauthAccountId); + if (!warmup.ok) { + const body = await warmup.response.json().catch(() => ({})) as { error?: string; reason?: string }; codexAuthLoginState.set(flowId, { status: "error", - error: "Cannot verify account identity for reauth. Remove this account and add it again.", + error: body.reason ? `${body.error ?? "Codex account warmup failed"} (${body.reason})` : body.error ?? "Codex account warmup failed", doneAt: Date.now(), }); completed = true; break; } - } - // 1.2: Duplicate check is scoped by personal vs workspace plan bucket. - const collision = checkAccountIdCollision(oauthAccountId, email, plan, reauth ? accountId : undefined); - if (collision.collision) { - codexAuthLoginState.set(flowId, { - status: "error", error: collision.reason, doneAt: Date.now(), - }); - completed = true; - break; - } + const latestConfig = getRuntimeConfig(config); + const accounts = latestConfig.codexAccounts ?? []; + const existingIdx = accounts.findIndex(account => account.id === accountId); + const commitConflict = codexAccountPersistenceConflict( + latestConfig, + accountId, + reauth ? "reauth" : "create", + ); + if (commitConflict) { + codexAuthLoginState.set(flowId, { + status: "error", + error: commitConflict, + doneAt: Date.now(), + }); + completed = true; + break; + } - const warmup = await verifyCodexAccountWarmup(accountId, cred.access, oauthAccountId); - if (!warmup.ok) { - const body = await warmup.response.json().catch(() => ({})) as { error?: string; reason?: string }; - codexAuthLoginState.set(flowId, { - status: "error", - error: body.reason ? `${body.error ?? "Codex account warmup failed"} (${body.reason})` : body.error ?? "Codex account warmup failed", - doneAt: Date.now(), + saveCodexAccountCredential(accountId, { + accessToken: cred.access, + refreshToken: cred.refresh, + expiresAt: cred.expires, + chatgptAccountId: oauthAccountId, }); - completed = true; - break; - } + // A successful reauthentication replaces the credential generation. Do not let a + // failed optional WHAM probe make the replacement inherit quota from the old record. + if (reauth) clearAccountQuota(accountId); + markCodexAccountValidated(accountId, warmup.validatedAt); + clearAccountNeedsReauth(accountId); + if (quota) { + setAccountQuotaFromParsed(accountId, quota); + } - const latestConfig = getRuntimeConfig(config); - const accounts = latestConfig.codexAccounts ?? []; - const existingIdx = accounts.findIndex(account => account.id === accountId); - const commitConflict = codexAccountPersistenceConflict( - latestConfig, - accountId, - reauth ? "reauth" : "create", - ); - if (commitConflict) { - codexAuthLoginState.set(flowId, { - status: "error", - error: commitConflict, - doneAt: Date.now(), - }); + if (existingIdx >= 0) { + // Keep the pool id stable; refresh display metadata after a successful login/reauth. + accounts[existingIdx] = withCodexAccountLogLabel({ + ...accounts[existingIdx], + email, + plan: plan ?? accounts[existingIdx].plan, + isMain: false, + }, accounts); + latestConfig.codexAccounts = accounts; + saveRuntimeConfig(config, latestConfig); + } else { + accounts.push(withCodexAccountLogLabel({ id: accountId, email, plan, isMain: false }, accounts)); + latestConfig.codexAccounts = accounts; + saveRuntimeConfig(config, latestConfig); + } + codexAuthLoginState.set(flowId, { status: "done", accountId, email, doneAt: Date.now() }); completed = true; - break; } - - saveCodexAccountCredential(accountId, { - accessToken: cred.access, - refreshToken: cred.refresh, - expiresAt: cred.expires, - chatgptAccountId: oauthAccountId, - }); - markCodexAccountValidated(accountId, warmup.validatedAt); - clearAccountNeedsReauth(accountId); - if (quota) { - setAccountQuotaFromParsed(accountId, quota); - } - - if (existingIdx >= 0) { - // Keep the pool id stable; refresh display metadata after a successful login/reauth. - accounts[existingIdx] = withCodexAccountLogLabel({ - ...accounts[existingIdx], - email, - plan: plan ?? accounts[existingIdx].plan, - isMain: false, - }, accounts); - latestConfig.codexAccounts = accounts; - saveRuntimeConfig(config, latestConfig); - } else { - accounts.push(withCodexAccountLogLabel({ id: accountId, email, plan, isMain: false }, accounts)); - latestConfig.codexAccounts = accounts; - saveRuntimeConfig(config, latestConfig); - } - codexAuthLoginState.set(flowId, { status: "done", accountId, email, doneAt: Date.now() }); + break; + } + if (st.done && st.error) { + codexAuthLoginState.set(flowId, { status: "error", error: st.error, doneAt: Date.now() }); completed = true; + break; } - break; } - if (st.done && st.error) { - codexAuthLoginState.set(flowId, { status: "error", error: st.error, doneAt: Date.now() }); - completed = true; - break; + if (!completed) { + codexAuthLoginState.set(flowId, { + status: "error", + error: "Login timed out before OAuth completed.", + doneAt: Date.now(), + }); } - } - if (!completed) { + } catch (error) { + const message = error instanceof ConfigMutationLockError + || error instanceof CodexCredentialRefreshLockTimeoutError + ? "Configuration is busy; retry login shortly." + : error instanceof Error ? error.message : String(error); codexAuthLoginState.set(flowId, { status: "error", - error: "Login timed out before OAuth completed.", + error: message, doneAt: Date.now(), }); + } finally { + // TTL: keep completed flow state available for clients that miss a short polling window. + setTimeout(() => codexAuthLoginState.delete(flowId), 300_000); } - // TTL: keep completed flow state available for clients that miss a short polling window. - setTimeout(() => codexAuthLoginState.delete(flowId), 300_000); })(); codexAuthLoginState.set(flowId, { status: "pending" }); diff --git a/src/codex/auth-context.ts b/src/codex/auth-context.ts index aacea92ee..a643703af 100644 --- a/src/codex/auth-context.ts +++ b/src/codex/auth-context.ts @@ -4,6 +4,7 @@ import { getValidCodexToken, isCodexAccountGenerationLive, } from "./account-store"; +import { ConfigMutationLockError } from "../config"; import { markAccountNeedsReauth } from "./account-runtime-state"; import { isCodexAccountUsable } from "./account-usability"; import { reconcileMainCodexAccountRuntimeState } from "./account-lifecycle"; @@ -180,7 +181,9 @@ export class CodexThreadAffinityExpiredError extends Error { } export function shouldMarkAccountNeedsReauthForCodexAuthFailure(cause: unknown): boolean { - return !(cause instanceof CodexCredentialGenerationConflictError) && !(cause instanceof CodexCredentialRefreshLockTimeoutError); + return !(cause instanceof CodexCredentialGenerationConflictError) + && !(cause instanceof CodexCredentialRefreshLockTimeoutError) + && !(cause instanceof ConfigMutationLockError); } export interface ResolveCodexAuthContextOptions { diff --git a/src/config.ts b/src/config.ts index 8f4a523f2..2e9615cf5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,8 +1,9 @@ import { execFileSync } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { copyFileSync, existsSync, linkSync, mkdirSync, readFileSync, renameSync, truncateSync, unlinkSync, writeFileSync, chmodSync } from "node:fs"; +import { chmodSync, copyFileSync, existsSync, linkSync, mkdirSync, readFileSync, renameSync, truncateSync, unlinkSync, writeFileSync } from "node:fs"; import { homedir } from "node:os"; import { join, resolve } from "node:path"; +import { Database } from "bun:sqlite"; import * as z from "zod/v4"; import { CODEX_ACCOUNT_NAMESPACE_COMBO_ALIAS_COLLISION_ERROR, @@ -1309,6 +1310,12 @@ export type ConfigDiagnostics = { warnings?: string[]; }; +type ConfigFileSnapshot = { + diagnostics: ConfigDiagnostics; + /** Exact file contents, including a possible BOM, used as the optimistic revision. */ + raw?: string; +}; + function configPlaceholderWarnings(config: OcxConfig): string[] { const warnings: string[] = []; for (const [name, provider] of Object.entries(config.providers)) { @@ -1399,14 +1406,9 @@ export function validateConfigCandidate(value: unknown): { ok: true; config: Ocx return { ok: false, error: schemaDiagnosticsError(result.error) }; } -export function readConfigDiagnostics(): ConfigDiagnostics { - const configPath = getConfigPath(); - if (!existsSync(configPath)) { - return { config: getDefaultConfig(), source: "default", error: null }; - } +function configDiagnosticsFromRaw(raw: string): ConfigDiagnostics { try { - const raw = readFileSync(configPath, "utf-8").replace(/^\uFEFF/, ""); - const parsed = JSON.parse(raw); + const parsed = JSON.parse(raw.replace(/^\uFEFF/, "")); const result = configSchema.safeParse(parsed); if (result.success) { return validFileConfigDiagnostics(normalizeApiKeyIds(result.data as OcxConfig), parsed); @@ -1423,10 +1425,42 @@ export function readConfigDiagnostics(): ConfigDiagnostics { } } -export function saveConfig(config: OcxConfig): void { +function readConfigFileSnapshot(): ConfigFileSnapshot { + try { + const raw = readFileSync(getConfigPath(), "utf-8"); + return { diagnostics: configDiagnosticsFromRaw(raw), raw }; + } catch (error) { + if (isMissingPathError(error)) { + return { + diagnostics: { config: getDefaultConfig(), source: "default", error: null }, + }; + } + return { + diagnostics: { config: getDefaultConfig(), source: "fallback", error: "invalid_json" }, + }; + } +} + +export function readConfigDiagnostics(): ConfigDiagnostics { + return readConfigFileSnapshot().diagnostics; +} + +const CONFIG_MUTATION_DB_FILENAME = "config-mutation.sqlite"; +const CONFIG_MUTATION_DB_SIDECARS = ["-journal", "-wal", "-shm"] as const; + +export class ConfigMutationLockError extends Error { + readonly code = "CONFIG_MUTATION_LOCK_UNAVAILABLE"; + + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "ConfigMutationLockError"; + } +} + +function configMutationDatabasePath(): string { const dir = getConfigDir(); - // First statement on purpose: a rejected write must leave nothing behind, not a - // freshly created/chmod'd directory. See src/lib/test-home-guard.ts. + // First statement on purpose: a rejected mutation must leave nothing behind, not a + // freshly created/chmod'd directory or database. See src/lib/test-home-guard.ts. assertNotRealHomeUnderTest(dir); if (!existsSync(dir)) { mkdirSync(dir, { recursive: true, mode: 0o700 }); @@ -1436,10 +1470,165 @@ export function saveConfig(config: OcxConfig): void { if (process.platform === "win32") { hardenSecretDir(dir, { required: true }); } + const path = join(dir, CONFIG_MUTATION_DB_FILENAME); + recordOwnedConfigPath(dir, path); + for (const suffix of CONFIG_MUTATION_DB_SIDECARS) { + recordOwnedConfigPath(dir, `${path}${suffix}`); + } + return path; +} + +let configMutationLockDepth = 0; + +/** + * Serialize synchronous config and Codex credential-generation commits across processes with an + * OS-backed SQLite write transaction. `busy_timeout=0` is deliberate: runtime request paths must + * fail immediately under contention rather than freeze the Bun event loop. Process exit releases + * SQLite locks without stale-owner deletion or lease recovery races. + * + * Reentrancy is limited to the current synchronous call stack; never return a Promise from `fn`. + */ +export function withConfigMutationLockSync(fn: () => T): T { + if (configMutationLockDepth > 0) { + configMutationLockDepth += 1; + try { + return fn(); + } finally { + configMutationLockDepth -= 1; + } + } + const path = configMutationDatabasePath(); + let database: Database | undefined; + let transactionOpen = false; + try { + database = new Database(path, { create: true }); + try { chmodSync(path, 0o600); } catch { /* platform may ignore chmod */ } + database.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + transactionOpen = true; + } catch (cause) { + try { database?.close(); } catch { /* acquisition already failed */ } + const code = cause && typeof cause === "object" && "code" in cause + ? String((cause as { code?: unknown }).code) + : ""; + throw new ConfigMutationLockError( + code === "SQLITE_BUSY" ? "Config mutation already in progress" : "Could not acquire config mutation transaction", + { cause }, + ); + } + + configMutationLockDepth = 1; + try { + const value = fn(); + database.exec("COMMIT"); + transactionOpen = false; + return value; + } catch (error) { + if (transactionOpen) { + try { database.exec("ROLLBACK"); } catch { /* close below still releases the OS lock */ } + transactionOpen = false; + } + throw error; + } finally { + configMutationLockDepth = 0; + try { database.close(); } catch { /* the OS lock is released with the handle */ } + } +} + +function persistConfigUnlocked(config: OcxConfig): void { const configPath = getConfigPath(); atomicWriteFile(configPath, JSON.stringify(config, null, 2) + "\n"); } +export function saveConfig(config: OcxConfig): void { + // Keep the real-home assertion ahead of even lock-directory preparation. + assertNotRealHomeUnderTest(getConfigDir()); + withConfigMutationLockSync(() => persistConfigUnlocked(config)); +} + +export type PersistedConfigMutation = { + changed: boolean; + value: T; +}; + +export type PersistedConfigMutationOutcome = + | { status: "committed" | "unchanged"; value: T } + | { status: "unavailable"; reason: "missing" | "invalid" | "conflict" }; + +const CONFIG_MUTATION_MAX_REBASE_ATTEMPTS = 3; +let persistedConfigMutationBeforeCommitForTests: (() => void) | null = null; + +/** Test-only one-shot seam: inject a competing mutation after the first decision, before freshness revalidation. */ +export function setPersistedConfigMutationBeforeCommitForTests(hook: (() => void) | null): void { + persistedConfigMutationBeforeCommitForTests = hook; +} + +function unavailableConfigMutationReason(snapshot: ConfigFileSnapshot): "missing" | "invalid" { + return snapshot.diagnostics.source === "default" ? "missing" : "invalid"; +} + +/** + * Patch a schema-valid on-disk config under the shared mutation lock. Cooperating writers are + * serialized; the callback is rerun on the newest snapshot so observed direct byte changes rebase + * and credential predicates are re-evaluated immediately before the atomic commit. A writer that + * ignores the coordinator can still change bytes after the final check because the filesystem has + * no portable conditional rename. Missing or malformed config always fails closed and is never + * recreated from a prior snapshot. + */ +export function mutatePersistedConfig( + mutate: (config: OcxConfig) => PersistedConfigMutation, +): PersistedConfigMutationOutcome { + // Avoid creating/opening the coordinator database for a read-path update that already knows + // there is no valid config. The same check runs again under the transaction for authority. + const observed = readConfigFileSnapshot(); + if (observed.diagnostics.source !== "file" || observed.raw === undefined) { + return { status: "unavailable", reason: unavailableConfigMutationReason(observed) }; + } + return withConfigMutationLockSync(() => { + let base = readConfigFileSnapshot(); + for (let attempt = 0; attempt < CONFIG_MUTATION_MAX_REBASE_ATTEMPTS; attempt += 1) { + if (base.diagnostics.source !== "file" || base.raw === undefined) { + return { status: "unavailable", reason: unavailableConfigMutationReason(base) }; + } + + const tentativeConfig = structuredClone(base.diagnostics.config); + const tentative = mutate(tentativeConfig); + if (!tentative.changed) return { status: "unchanged", value: tentative.value }; + + const hook = persistedConfigMutationBeforeCommitForTests; + persistedConfigMutationBeforeCommitForTests = null; + hook?.(); + + const latest = readConfigFileSnapshot(); + if (latest.diagnostics.source !== "file" || latest.raw === undefined) { + return { status: "unavailable", reason: unavailableConfigMutationReason(latest) }; + } + if (latest.raw !== base.raw) { + base = latest; + continue; + } + + // Re-run against a fresh clone even when config bytes are unchanged: a Codex credential + // generation lives in a separate file and may have changed at the injected seam. + const confirmedConfig = structuredClone(latest.diagnostics.config); + const confirmed = mutate(confirmedConfig); + if (!confirmed.changed) return { status: "unchanged", value: confirmed.value }; + + const commitBase = readConfigFileSnapshot(); + if (commitBase.diagnostics.source !== "file" || commitBase.raw === undefined) { + return { status: "unavailable", reason: unavailableConfigMutationReason(commitBase) }; + } + if (commitBase.raw !== latest.raw) { + base = commitBase; + continue; + } + + persistConfigUnlocked(confirmedConfig); + return { status: "committed", value: confirmed.value }; + } + return { status: "unavailable", reason: "conflict" }; + }); +} + export function websocketsEnabled(config: Pick): boolean { return config.websockets === true; } @@ -1669,36 +1858,38 @@ function readPersistedServerBinding( * guarantee. */ export function saveConfigPreservingClaudeCode(config: OcxConfig): void { - const bindingBaseline = persistedLiveServerBinding.get(config); - const onDisk = claudeCodeBaseline.has(config) || bindingBaseline - ? readRawConfigJson() - : undefined; - if (claudeCodeBaseline.has(config)) { - if (onDisk !== undefined) { - const baseline = claudeCodeBaseline.get(config); - const persistedClaudeCode = normalizePersistedClaudeCode(onDisk.claudeCode); - const diskChanged = !deepEqual(persistedClaudeCode, baseline); - const weChanged = !deepEqual(config.claudeCode, baseline); - if (diskChanged && !weChanged) { - config.claudeCode = persistedClaudeCode; + withConfigMutationLockSync(() => { + const bindingBaseline = persistedLiveServerBinding.get(config); + const onDisk = claudeCodeBaseline.has(config) || bindingBaseline + ? readRawConfigJson() + : undefined; + if (claudeCodeBaseline.has(config)) { + if (onDisk !== undefined) { + const baseline = claudeCodeBaseline.get(config); + const persistedClaudeCode = normalizePersistedClaudeCode(onDisk.claudeCode); + const diskChanged = !deepEqual(persistedClaudeCode, baseline); + const weChanged = !deepEqual(config.claudeCode, baseline); + if (diskChanged && !weChanged) { + config.claudeCode = persistedClaudeCode; + } } } - } - const persistedBinding = bindingBaseline && onDisk - ? readPersistedServerBinding(onDisk, bindingBaseline) - : bindingBaseline; - if (persistedBinding) { - const persistedConfig: OcxConfig = { ...config, port: persistedBinding.port }; - if (persistedBinding.hostname === undefined) delete persistedConfig.hostname; - else persistedConfig.hostname = persistedBinding.hostname; - saveConfig(persistedConfig); - persistedLiveServerBinding.set(config, persistedBinding); - } else { - saveConfig(config); - } - if (claudeCodeBaseline.has(config)) { - claudeCodeBaseline.set(config, structuredClone(config.claudeCode)); - } + const persistedBinding = bindingBaseline && onDisk + ? readPersistedServerBinding(onDisk, bindingBaseline) + : bindingBaseline; + if (persistedBinding) { + const persistedConfig: OcxConfig = { ...config, port: persistedBinding.port }; + if (persistedBinding.hostname === undefined) delete persistedConfig.hostname; + else persistedConfig.hostname = persistedBinding.hostname; + persistConfigUnlocked(persistedConfig); + persistedLiveServerBinding.set(config, persistedBinding); + } else { + persistConfigUnlocked(config); + } + if (claudeCodeBaseline.has(config)) { + claudeCodeBaseline.set(config, structuredClone(config.claudeCode)); + } + }); } export function codexAutoStartEnabled(config: Pick): boolean { diff --git a/src/lib/config-ownership.ts b/src/lib/config-ownership.ts index 6c8b855f5..3dbfa2072 100644 --- a/src/lib/config-ownership.ts +++ b/src/lib/config-ownership.ts @@ -47,6 +47,10 @@ const INITIAL_OWNED_PATHS = [ "codex-shim.autorestore.lock", "codex-shim.json", "config.json", + "config-mutation.sqlite", + "config-mutation.sqlite-journal", + "config-mutation.sqlite-shm", + "config-mutation.sqlite-wal", "crash.log", "kimi-device-id", "mimo-client-id", diff --git a/src/server/management-api.ts b/src/server/management-api.ts index 07db2e453..28fd111f8 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -167,7 +167,23 @@ export async function handleManagementAPI(req: Request, url: URL, config: OcxCon if (url.pathname.startsWith("/api/codex-auth/")) { const { handleCodexAuthAPI } = await import("../codex/auth-api"); - return handleCodexAuthAPI(req, url, config); + const { ConfigMutationLockError } = await import("../config"); + const { CodexCredentialRefreshLockTimeoutError } = await import("../codex/account-store"); + try { + return await handleCodexAuthAPI(req, url, config); + } catch (error) { + // Credential writers remap ConfigMutationLockError to CodexCredentialRefreshLockTimeoutError; + // treat both as the same retryable busy response. + if (error instanceof ConfigMutationLockError || error instanceof CodexCredentialRefreshLockTimeoutError) { + return jsonResponse( + { error: "Configuration is busy; retry shortly", code: "CONFIG_MUTATION_LOCK_UNAVAILABLE" }, + 503, + req, + config, + ); + } + throw error; + } } return null; diff --git a/tests/codex-auth-api.test.ts b/tests/codex-auth-api.test.ts index 0b7fa9654..12f0b065c 100644 --- a/tests/codex-auth-api.test.ts +++ b/tests/codex-auth-api.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import type { ServerWebSocket } from "bun"; -import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { Database } from "bun:sqlite"; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { CODEX_ACCOUNT_LOG_LABEL_RE } from "../src/codex/account-label"; import { @@ -37,6 +38,12 @@ import { reconcileMainCodexAccountRuntimeState, resetMainCodexAccountIdentityTrackingForTests, } from "../src/codex/account-lifecycle"; +import { + getConfigPath, + loadConfig, + saveConfig, + setPersistedConfigMutationBeforeCommitForTests, +} from "../src/config"; const TEST_DIR = join(import.meta.dir, ".tmp-codex-auth-api-test"); const TEST_CODEX_HOME = join(TEST_DIR, "codex"); @@ -215,6 +222,7 @@ beforeEach(() => { }); afterEach(() => { + setPersistedConfigMutationBeforeCommitForTests(null); clearAccountNeedsReauth("__main__"); clearAccountQuota(); clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); @@ -684,6 +692,697 @@ describe("codex-auth API", () => { } }); + test("GET /api/codex-auth/accounts reconciles and persists a fresh pool plan on a cache miss", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-refresh", + email: "pool-plan-refresh@example.com", + plan: "plus", + }); + saveConfig(structuredClone(config)); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12, reset_at: 1782628379 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(data.accounts.find(account => account.id === "pool-plan-refresh")?.plan).toBe("prolite"); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-refresh")?.plan).toBe("prolite"); + expect(loadConfig().codexAccounts?.find(account => account.id === "pool-plan-refresh")?.plan).toBe("prolite"); + }); + + test("a fresh upgraded plan controls quota projection in the same accounts response", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-upgrade", + email: "pool-plan-upgrade@example.com", + plan: "free", + }); + saveConfig(structuredClone(config)); + globalThis.fetch = (async () => Response.json({ + plan_type: "pro", + rate_limit: { + primary_window: { used_percent: 17, reset_at: 1782628379 }, + tertiary_window: { used_percent: 23, reset_at: 1787401330 }, + }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { + accounts: Array<{ id: string; plan?: string; quota?: Record }>; + }; + const account = data.accounts.find(candidate => candidate.id === "pool-plan-upgrade"); + + expect(account?.plan).toBe("pro"); + expect(account?.quota).toMatchObject({ + weeklyPercent: 17, + weeklyResetAt: 1782628379, + monthlyPercent: 23, + monthlyResetAt: 1787401330, + }); + }); + + test("pool plan refresh ignores a missing upstream plan and performs no config write", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-missing", + email: "pool-plan-missing@example.com", + plan: "plus", + }); + saveConfig(structuredClone(config)); + let configCommits = 0; + setPersistedConfigMutationBeforeCommitForTests(() => { configCommits += 1; }); + globalThis.fetch = (async () => Response.json({ + rate_limit: { primary_window: { used_percent: 8, reset_at: 1782628379 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(data.accounts.find(account => account.id === "pool-plan-missing")?.plan).toBe("plus"); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-missing")?.plan).toBe("plus"); + expect(loadConfig().codexAccounts?.find(account => account.id === "pool-plan-missing")?.plan).toBe("plus"); + expect(configCommits).toBe(0); + }); + + test("pool plan refresh persists plan_type when WHAM omits rate_limit windows", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-only", + email: "pool-plan-only@example.com", + plan: "plus", + }); + saveConfig(structuredClone(config)); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(data.accounts.find(account => account.id === "pool-plan-only")?.plan).toBe("prolite"); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-only")?.plan).toBe("prolite"); + expect(loadConfig().codexAccounts?.find(account => account.id === "pool-plan-only")?.plan).toBe("prolite"); + }); + + test("pool plan refresh performs no config write when the authoritative plan is unchanged", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-unchanged", + email: "pool-plan-unchanged@example.com", + plan: "plus", + }); + saveConfig(structuredClone(config)); + let configCommits = 0; + setPersistedConfigMutationBeforeCommitForTests(() => { configCommits += 1; }); + globalThis.fetch = (async () => Response.json({ + plan_type: "plus", + rate_limit: { primary_window: { used_percent: 7, reset_at: 1782628379 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(data.accounts.find(account => account.id === "pool-plan-unchanged")?.plan).toBe("plus"); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-unchanged")?.plan).toBe("plus"); + expect(loadConfig().codexAccounts?.find(account => account.id === "pool-plan-unchanged")?.plan).toBe("plus"); + expect(configCommits).toBe(0); + }); + + test("pool plan refresh does not recreate a config file deleted while the server is running", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-config-deleted", + email: "pool-plan-config-deleted@example.com", + plan: "plus", + }); + globalThis.fetch = (async () => Response.json({ + plan_type: "pro", + rate_limit: { primary_window: { used_percent: 6, reset_at: 1782628379 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(resp!.status).toBe(200); + // Missing config fails closed on disk, but the response still surfaces the WHAM plan. + expect(data.accounts.find(account => account.id === "pool-plan-config-deleted")?.plan).toBe("pro"); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-config-deleted")?.plan).toBe("plus"); + expect(existsSync(join(TEST_DIR, "config.json"))).toBe(false); + }); + + test("pool plan refresh batches multiple authoritative changes into one config save", async () => { + const config = makeConfig(); + seedPoolAccount(config, { id: "pool-plan-a", email: "pool-plan-a@example.com", plan: "plus" }); + seedPoolAccount(config, { id: "pool-plan-b", email: "pool-plan-b@example.com", plan: "free" }); + saveConfig(structuredClone(config)); + let configCommits = 0; + setPersistedConfigMutationBeforeCommitForTests(() => { configCommits += 1; }); + globalThis.fetch = (async (_input, init) => { + const accountId = new Headers(init?.headers).get("ChatGPT-Account-Id"); + return Response.json({ + plan_type: accountId === "acct-pool-plan-a" ? "prolite" : "pro", + rate_limit: { primary_window: { used_percent: 9 } }, + }); + }) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + expect(resp!.status).toBe(200); + expect(configCommits).toBe(1); + expect(loadConfig().codexAccounts?.map(account => [account.id, account.plan])).toEqual([ + ["pool-plan-a", "prolite"], + ["pool-plan-b", "pro"], + ]); + }); + + test("concurrent pool plan refreshes share token rotation, one WHAM read, and one config save", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-concurrent", + email: "pool-plan-concurrent@example.com", + plan: "plus", + expiresAt: Date.now() - 1, + }); + saveConfig(structuredClone(config)); + let configCommits = 0; + setPersistedConfigMutationBeforeCommitForTests(() => { configCommits += 1; }); + let calls = 0; + let tokenRefreshCalls = 0; + let releaseFetch!: () => void; + let markFetchStarted!: () => void; + const fetchStarted = new Promise(resolve => { markFetchStarted = resolve; }); + const fetchGate = new Promise(resolve => { releaseFetch = resolve; }); + globalThis.fetch = (async input => { + if (String(input) === "https://auth.openai.com/oauth/token") { + tokenRefreshCalls += 1; + return Response.json({ + access_token: "rotated-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + } + calls += 1; + markFetchStarted(); + await fetchGate; + return Response.json({ + plan_type: "pro", + rate_limit: { primary_window: { used_percent: 16 } }, + }); + }) as typeof fetch; + + try { + const request = () => { + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + return handleCodexAuthAPI(req, new URL(req.url), config); + }; + const first = request(); + const second = request(); + await fetchStarted; + await new Promise(resolve => queueMicrotask(resolve)); + expect(calls).toBe(1); + releaseFetch(); + const responses = await Promise.all([first, second]); + const payloads = await Promise.all(responses.map(response => response!.json())) as Array<{ + accounts: Array<{ id: string; plan?: string }>; + }>; + + expect(payloads.map(payload => payload.accounts.find(account => account.id === "pool-plan-concurrent")?.plan)) + .toEqual(["pro", "pro"]); + expect(tokenRefreshCalls).toBe(1); + expect(calls).toBe(1); + expect(configCommits).toBe(1); + } finally { + setPersistedConfigMutationBeforeCommitForTests(null); + } + }); + + test("pool plan persistence preserves an unrelated edit from the latest disk config", async () => { + const config = makeConfig({ + providers: { + custom: { + adapter: "openai-chat", + baseUrl: "https://example.test/v1", + note: "runtime-note", + }, + }, + defaultProvider: "custom", + }); + seedPoolAccount(config, { id: "pool-plan-disk", email: "pool-plan-disk@example.com", plan: "plus" }); + saveConfig(structuredClone(config)); + let releaseFetch!: () => void; + let markFetchStarted!: () => void; + const fetchStarted = new Promise(resolve => { markFetchStarted = resolve; }); + const fetchGate = new Promise(resolve => { releaseFetch = resolve; }); + globalThis.fetch = (async () => { + markFetchStarted(); + await fetchGate; + return Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 14 } }, + }); + }) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const pending = handleCodexAuthAPI(req, new URL(req.url), config); + await fetchStarted; + const editedOnDisk = loadConfig(); + editedOnDisk.providers.custom!.note = "manual-disk-note"; + saveConfig(editedOnDisk); + releaseFetch(); + const resp = await pending; + expect(resp!.status).toBe(200); + + const persisted = loadConfig(); + expect(persisted.providers.custom?.note).toBe("manual-disk-note"); + expect(persisted.codexAccounts?.find(account => account.id === "pool-plan-disk")?.plan).toBe("prolite"); + expect(config.providers.custom?.note).toBe("runtime-note"); + }); + + test("pool plan commit rebases an unrelated config write injected after its first checks", async () => { + const config = makeConfig({ + providers: { + custom: { + adapter: "openai-chat", + baseUrl: "https://example.test/v1", + note: "runtime-note", + }, + }, + defaultProvider: "custom", + }); + seedPoolAccount(config, { id: "pool-plan-cas-edit", email: "pool-plan-cas-edit@example.com", plan: "plus" }); + saveConfig(structuredClone(config)); + let injected = false; + setPersistedConfigMutationBeforeCommitForTests(() => { + injected = true; + const concurrent = loadConfig(); + concurrent.providers.custom!.note = "concurrent-disk-note"; + saveConfig(concurrent); + }); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + const persisted = loadConfig(); + + expect(injected).toBe(true); + expect(persisted.providers.custom?.note).toBe("concurrent-disk-note"); + expect(persisted.codexAccounts?.find(account => account.id === "pool-plan-cas-edit")?.plan).toBe("prolite"); + expect(config.providers.custom?.note).toBe("runtime-note"); + expect(data.accounts.find(account => account.id === "pool-plan-cas-edit")?.plan).toBe("prolite"); + }); + + test("pool plan commit does not recreate a config deleted after its first checks", async () => { + const config = makeConfig(); + seedPoolAccount(config, { id: "pool-plan-cas-delete", email: "pool-plan-cas-delete@example.com", plan: "plus" }); + saveConfig(structuredClone(config)); + let absentAtRecheck = false; + setPersistedConfigMutationBeforeCommitForTests(() => { + rmSync(getConfigPath(), { force: true }); + absentAtRecheck = !existsSync(getConfigPath()); + }); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(absentAtRecheck).toBe(true); + expect(existsSync(getConfigPath())).toBe(false); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-cas-delete")?.plan).toBe("plus"); + expect(data.accounts.find(account => account.id === "pool-plan-cas-delete")?.plan).toBe("prolite"); + }); + + test("pool plan commit leaves config malformed when malformed bytes arrive after its first checks", async () => { + const config = makeConfig(); + seedPoolAccount(config, { id: "pool-plan-cas-malformed", email: "pool-plan-cas-malformed@example.com", plan: "plus" }); + saveConfig(structuredClone(config)); + const malformed = "{ malformed concurrent config"; + let malformedAtRecheck = false; + setPersistedConfigMutationBeforeCommitForTests(() => { + writeFileSync(getConfigPath(), malformed); + malformedAtRecheck = readFileSync(getConfigPath(), "utf8") === malformed; + }); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(malformedAtRecheck).toBe(true); + expect(readFileSync(getConfigPath(), "utf8")).toBe(malformed); + expect(config.codexAccounts?.find(account => account.id === "pool-plan-cas-malformed")?.plan).toBe("plus"); + expect(data.accounts.find(account => account.id === "pool-plan-cas-malformed")?.plan).toBe("prolite"); + }); + + test("pool plan commit rejects an old plan after same-id credential replacement at its commit seam", async () => { + const accountId = "pool-plan-cas-generation"; + const config = makeConfig(); + seedPoolAccount(config, { + id: accountId, + email: "old-generation@example.com", + plan: "plus", + accessToken: "old-generation-access", + refreshToken: "old-generation-refresh", + chatgptAccountId: "old-generation-account", + }); + saveConfig(structuredClone(config)); + const startingGeneration = readCodexAccountRecord(accountId)?.generation; + let replacementGeneration: number | undefined; + setPersistedConfigMutationBeforeCommitForTests(() => { + const replacementConfig = loadConfig(); + const replacementAccount = replacementConfig.codexAccounts?.find(account => account.id === accountId); + if (!replacementAccount) throw new Error("replacement account missing"); + replacementAccount.email = "replacement@example.test"; + replacementAccount.plan = "free"; + saveCodexAccountCredential(accountId, { + accessToken: "replacement-access", + refreshToken: "replacement-refresh", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "replacement-account", + }); + replacementGeneration = readCodexAccountRecord(accountId)?.generation; + saveConfig(replacementConfig); + }); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12 } }, + })) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + const persistedAccount = loadConfig().codexAccounts?.find(account => account.id === accountId); + const finalRecord = readCodexAccountRecord(accountId); + + expect(startingGeneration).toBe(1); + expect(replacementGeneration).toBe(2); + expect(finalRecord?.generation).toBe(2); + expect(finalRecord?.credential?.accessToken).toBe("replacement-access"); + expect(persistedAccount).toMatchObject({ email: "replacement@example.test", plan: "free" }); + expect(config.codexAccounts?.find(account => account.id === accountId)?.plan).toBe("plus"); + expect(data.accounts.find(account => account.id === accountId)?.plan).toBe("plus"); + }); + + test("pool plan persistence fails closed without waiting behind a busy config writer", async () => { + const accountId = "pool-plan-lock-busy"; + const config = makeConfig(); + seedPoolAccount(config, { id: accountId, email: "pool-plan-lock-busy@example.com", plan: "plus" }); + saveConfig(structuredClone(config)); + const lockDatabase = new Database(join(TEST_DIR, "config-mutation.sqlite"), { create: true }); + lockDatabase.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE"); + globalThis.fetch = (async () => Response.json({ + plan_type: "prolite", + rate_limit: { primary_window: { used_percent: 12 } }, + })) as typeof fetch; + + try { + const startedAt = performance.now(); + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const resp = await handleCodexAuthAPI(req, new URL(req.url), config); + const elapsedMs = performance.now() - startedAt; + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(elapsedMs).toBeLessThan(2_000); + expect(loadConfig().codexAccounts?.find(account => account.id === accountId)?.plan).toBe("plus"); + expect(config.codexAccounts?.find(account => account.id === accountId)?.plan).toBe("plus"); + // Disk persistence fails closed under lock contention, but the response still reflects WHAM. + expect(data.accounts.find(account => account.id === accountId)?.plan).toBe("prolite"); + } finally { + lockDatabase.exec("ROLLBACK"); + lockDatabase.close(); + } + }); + + test("pool plan refresh cannot update an account deleted and recreated during the WHAM request", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-reused", + email: "old-plan@example.com", + plan: "plus", + accessToken: "old-plan-access", + refreshToken: "old-plan-refresh", + chatgptAccountId: "old-plan-account", + }); + let releaseFetch!: () => void; + let markFetchStarted!: () => void; + const fetchStarted = new Promise(resolve => { markFetchStarted = resolve; }); + const fetchGate = new Promise(resolve => { releaseFetch = resolve; }); + globalThis.fetch = (async () => { + markFetchStarted(); + await fetchGate; + return Response.json({ + plan_type: "pro", + rate_limit: { primary_window: { used_percent: 11 } }, + }); + }) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const pending = handleCodexAuthAPI(req, new URL(req.url), config); + await fetchStarted; + deleteCodexAccount(config, "pool-plan-reused"); + config.codexAccounts = [{ + id: "pool-plan-reused", + email: "new-plan@example.com", + plan: "free", + isMain: false, + }]; + saveCodexAccountCredential("pool-plan-reused", { + accessToken: "new-plan-access", + refreshToken: "new-plan-refresh", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "new-plan-account", + }); + releaseFetch(); + const resp = await pending; + const data = await resp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(config.codexAccounts[0]?.plan).toBe("free"); + expect(data.accounts.find(account => account.id === "pool-plan-reused")?.plan).toBe("free"); + expect(existsSync(join(TEST_DIR, "config.json"))).toBe(false); + }); + + test("a replacement credential starts a separate refresh while the prior generation is still in flight", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-generation-flight", + email: "old-flight@example.com", + plan: "plus", + accessToken: "old-flight-access", + refreshToken: "old-flight-refresh", + chatgptAccountId: "old-flight-account", + }); + saveConfig(structuredClone(config)); + let releaseOldFetch!: () => void; + let markOldFetchStarted!: () => void; + const oldFetchStarted = new Promise(resolve => { markOldFetchStarted = resolve; }); + const oldFetchGate = new Promise(resolve => { releaseOldFetch = resolve; }); + const whamAccountIds: string[] = []; + globalThis.fetch = (async (_input, init) => { + const accountId = new Headers(init?.headers).get("ChatGPT-Account-Id") ?? ""; + whamAccountIds.push(accountId); + if (accountId === "old-flight-account") { + markOldFetchStarted(); + await oldFetchGate; + return Response.json({ + plan_type: "plus", + rate_limit: { primary_window: { used_percent: 41 } }, + }); + } + return Response.json({ + plan_type: "pro", + rate_limit: { primary_window: { used_percent: 19 } }, + }); + }) as typeof fetch; + + const request = () => { + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + return handleCodexAuthAPI(req, new URL(req.url), config); + }; + const oldPending = request(); + await oldFetchStarted; + deleteCodexAccount(config, "pool-plan-generation-flight"); + config.codexAccounts = [{ + id: "pool-plan-generation-flight", + email: "new-flight@example.com", + plan: "free", + isMain: false, + }]; + saveCodexAccountCredential("pool-plan-generation-flight", { + accessToken: "new-flight-access", + refreshToken: "new-flight-refresh", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "new-flight-account", + }); + saveConfig(structuredClone(config)); + + const replacementResp = await request(); + const replacementData = await replacementResp!.json() as { + accounts: Array<{ id: string; plan?: string; quota?: { weeklyPercent?: number } }>; + }; + const replacement = replacementData.accounts.find(account => account.id === "pool-plan-generation-flight"); + expect(replacement).toMatchObject({ plan: "pro", quota: { weeklyPercent: 19 } }); + + releaseOldFetch(); + const oldResp = await oldPending; + const oldData = await oldResp!.json() as { accounts: Array<{ id: string; plan?: string }> }; + + expect(whamAccountIds).toEqual(["old-flight-account", "new-flight-account"]); + expect(oldData.accounts.find(account => account.id === "pool-plan-generation-flight")?.plan).toBe("pro"); + expect(config.codexAccounts[0]?.plan).toBe("pro"); + expect(loadConfig().codexAccounts?.[0]?.plan).toBe("pro"); + }); + + test("final account assembly discards quota and credential state from a replaced account", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-finished", + email: "old-finished@example.com", + plan: "plus", + chatgptAccountId: "old-finished-account", + }); + seedPoolAccount(config, { + id: "pool-plan-blocker", + email: "blocker@example.com", + plan: "plus", + chatgptAccountId: "blocker-account", + }); + let releaseBlocker!: () => void; + let markBlockerStarted!: () => void; + const blockerStarted = new Promise(resolve => { markBlockerStarted = resolve; }); + const blockerGate = new Promise(resolve => { releaseBlocker = resolve; }); + globalThis.fetch = (async (_input, init) => { + const accountId = new Headers(init?.headers).get("ChatGPT-Account-Id"); + if (accountId === "blocker-account") { + markBlockerStarted(); + await blockerGate; + return Response.json({ + plan_type: "plus", + rate_limit: { primary_window: { used_percent: 5 } }, + }); + } + return Response.json({ + plan_type: "pro", + rate_limit: { primary_window: { used_percent: 31, reset_at: 1782628379 } }, + }); + }) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const pending = handleCodexAuthAPI(req, new URL(req.url), config); + await blockerStarted; + for (let attempt = 0; attempt < 100 && getAccountQuota("pool-plan-finished") === null; attempt += 1) { + await new Promise(resolve => queueMicrotask(resolve)); + } + expect(getAccountQuota("pool-plan-finished")?.weeklyPercent).toBe(31); + + deleteCodexAccount(config, "pool-plan-finished"); + config.codexAccounts = [ + { + id: "pool-plan-finished", + email: "new-finished@example.com", + plan: "free", + isMain: false, + }, + ...(config.codexAccounts ?? []), + ]; + releaseBlocker(); + const resp = await pending; + const data = await resp!.json() as { + accounts: Array<{ + id: string; + plan?: string; + quota: unknown; + hasCredential: boolean; + needsReauth?: boolean; + }>; + }; + const replacement = data.accounts.find(account => account.id === "pool-plan-finished"); + + expect(replacement).toMatchObject({ + plan: "free", + quota: null, + hasCredential: false, + needsReauth: true, + }); + expect(config.codexAccounts.find(account => account.id === "pool-plan-finished")?.plan).toBe("free"); + }); + + test("a stale 401 refresh cannot mark a replacement credential for reauthentication", async () => { + const config = makeConfig(); + seedPoolAccount(config, { + id: "pool-plan-stale-auth", + email: "old-auth@example.com", + plan: "plus", + accessToken: "old-auth-access", + refreshToken: "old-auth-refresh", + chatgptAccountId: "old-auth-account", + }); + let releaseFetch!: () => void; + let markFetchStarted!: () => void; + const fetchStarted = new Promise(resolve => { markFetchStarted = resolve; }); + const fetchGate = new Promise(resolve => { releaseFetch = resolve; }); + globalThis.fetch = (async () => { + markFetchStarted(); + await fetchGate; + return new Response(null, { status: 401 }); + }) as typeof fetch; + + const req = new Request("http://localhost/api/codex-auth/accounts?refresh=1", { method: "GET" }); + const pending = handleCodexAuthAPI(req, new URL(req.url), config); + await fetchStarted; + deleteCodexAccount(config, "pool-plan-stale-auth"); + config.codexAccounts = [{ + id: "pool-plan-stale-auth", + email: "new-auth@example.com", + plan: "free", + isMain: false, + }]; + saveCodexAccountCredential("pool-plan-stale-auth", { + accessToken: "new-auth-access", + refreshToken: "new-auth-refresh", + expiresAt: Date.now() + 5 * 60_000, + chatgptAccountId: "new-auth-account", + }); + releaseFetch(); + const resp = await pending; + const data = await resp!.json() as { + accounts: Array<{ + id: string; + plan?: string; + quota: unknown; + hasCredential: boolean; + needsReauth?: boolean; + }>; + }; + const replacement = data.accounts.find(account => account.id === "pool-plan-stale-auth"); + + expect(replacement).toMatchObject({ + plan: "free", + quota: null, + hasCredential: true, + needsReauth: false, + }); + expect(getCodexAccountCredential("pool-plan-stale-auth")?.accessToken).toBe("new-auth-access"); + }); + test("GET /api/codex-auth/accounts refresh=1 clears stale weekly for monthly primary-only WHAM", async () => { const config = makeConfig(); seedPoolAccount(config, { @@ -2286,6 +2985,7 @@ describe("codex-auth API", () => { expiresAt: Date.now() + 5 * 60_000, chatgptAccountId: "acct-reauth-plan", }); + updateAccountQuota("reauth-plan", 67, 1782628379); const result = await completeMockCodexOAuth({ config, @@ -2298,6 +2998,7 @@ describe("codex-auth API", () => { expect(result.state).toMatchObject({ status: "done" }); expect(config.codexAccounts?.[0]?.plan).toBe("business"); + expect(getAccountQuota("reauth-plan")).toBeNull(); }); test("OAuth pool login excludes self from collision check when reauth", async () => { @@ -2363,6 +3064,7 @@ describe("codex-auth API", () => { const pool = data.accounts.find(a => a.id === "cached-test"); expect(pool?.quota).toMatchObject({ weeklyPercent: 25 }); expect(called).toBe(false); + expect(existsSync(join(TEST_DIR, "config.json"))).toBe(false); } finally { globalThis.fetch = originalFetch; } diff --git a/tests/codex-auth-context.test.ts b/tests/codex-auth-context.test.ts index 16c9a921d..ff7efa21c 100644 --- a/tests/codex-auth-context.test.ts +++ b/tests/codex-auth-context.test.ts @@ -25,6 +25,7 @@ import { removeCodexAccountCredential, saveCodexAccountCredential, } from "../src/codex/account-store"; +import { ConfigMutationLockError } from "../src/config"; import { MAIN_CODEX_ACCOUNT_ID } from "../src/codex/main-account"; import { clearAccountNeedsReauth, isAccountNeedsReauth } from "../src/codex/auth-api"; import { @@ -519,6 +520,7 @@ describe("Codex auth context", () => { test("reauth marking is reserved for real token failures", () => { expect(shouldMarkAccountNeedsReauthForCodexAuthFailure(new CodexCredentialGenerationConflictError())).toBe(false); expect(shouldMarkAccountNeedsReauthForCodexAuthFailure(new CodexCredentialRefreshLockTimeoutError())).toBe(false); + expect(shouldMarkAccountNeedsReauthForCodexAuthFailure(new ConfigMutationLockError("busy"))).toBe(false); expect(shouldMarkAccountNeedsReauthForCodexAuthFailure(new Error("bad token"))).toBe(true); }); diff --git a/tests/config-mutation-lock.test.ts b/tests/config-mutation-lock.test.ts new file mode 100644 index 000000000..d984f2724 --- /dev/null +++ b/tests/config-mutation-lock.test.ts @@ -0,0 +1,183 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; +import { ConfigMutationLockError, loadConfig, saveConfig, withConfigMutationLockSync } from "../src/config"; +import { CodexCredentialRefreshLockTimeoutError, getCodexAccountCredential, saveCodexAccountCredential } from "../src/codex/account-store"; +import type { OcxConfig } from "../src/types"; +import { ManagementRequest, managementHeaders } from "./helpers/management-auth"; + +let testRoot = ""; +let previousOpencodexHome: string | undefined; + +function config(port = 10100): OcxConfig { + return { port, providers: {}, defaultProvider: "openai" }; +} + +async function waitForPath(path: string): Promise { + for (let attempt = 0; attempt < 500; attempt += 1) { + if (existsSync(path)) return; + await Bun.sleep(10); + } + throw new Error(`Timed out waiting for child marker ${path}`); +} + +async function waitForOwnedChild(child: ReturnType): Promise { + const result = await Promise.race([ + child.exited.then(exitCode => ({ exitCode })), + Bun.sleep(5_000).then(() => null), + ]); + if (result) return result.exitCode; + child.kill(); + await child.exited; + throw new Error("Timed out waiting for owned config-lock child"); +} + +beforeEach(() => { + previousOpencodexHome = process.env.OPENCODEX_HOME; + testRoot = mkdtempSync(join(import.meta.dir, ".tmp-config-mutation-lock-")); + process.env.OPENCODEX_HOME = testRoot; +}); + +afterEach(() => { + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + rmSync(testRoot, { recursive: true, force: true }); +}); + +test("a live cross-process holder is not stolen and runtime writers fail immediately", async () => { + saveConfig(config()); + const readyPath = join(testRoot, "holder-ready"); + const releasePath = join(testRoot, "holder-release"); + const configModuleUrl = pathToFileURL(join(import.meta.dir, "../src/config.ts")).href; + const childSource = ` + import { existsSync, writeFileSync } from "node:fs"; + import { withConfigMutationLockSync } from ${JSON.stringify(configModuleUrl)}; + withConfigMutationLockSync(() => { + writeFileSync(${JSON.stringify(readyPath)}, "ready"); + while (!existsSync(${JSON.stringify(releasePath)})) Bun.sleepSync(10); + }); + `; + const child = Bun.spawn([process.execPath, "-e", childSource], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env, OPENCODEX_HOME: testRoot }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + + try { + try { + await waitForPath(readyPath); + } catch (error) { + child.kill(); + await child.exited; + const stderr = await new Response(child.stderr).text().catch(() => ""); + throw new Error(`${(error as Error).message}\nchild stderr: ${stderr}`); + } + const startedAt = performance.now(); + expect(() => saveConfig(config(20200))).toThrow(ConfigMutationLockError); + expect(() => saveCodexAccountCredential("busy-account", { + accessToken: "busy-access", + refreshToken: "busy-refresh", + expiresAt: Date.now() + 60_000, + chatgptAccountId: "busy-chatgpt-account", + })).toThrow(CodexCredentialRefreshLockTimeoutError); + expect(performance.now() - startedAt).toBeLessThan(1_000); + expect(loadConfig().port).toBe(10100); + expect(getCodexAccountCredential("busy-account")).toBeNull(); + } finally { + writeFileSync(releasePath, "release"); + expect(await waitForOwnedChild(child)).toBe(0); + } + + saveConfig(config(20200)); + saveCodexAccountCredential("busy-account", { + accessToken: "fresh-access", + refreshToken: "fresh-refresh", + expiresAt: Date.now() + 60_000, + chatgptAccountId: "fresh-chatgpt-account", + }); + expect(loadConfig().port).toBe(20200); + expect(getCodexAccountCredential("busy-account")?.accessToken).toBe("fresh-access"); +}); + +test("an abruptly exited holder releases the OS-backed transaction without stale recovery", async () => { + saveConfig(config()); + const enteredPath = join(testRoot, "crashed-holder-entered"); + const configModuleUrl = pathToFileURL(join(import.meta.dir, "../src/config.ts")).href; + const childSource = ` + import { writeFileSync } from "node:fs"; + import { withConfigMutationLockSync } from ${JSON.stringify(configModuleUrl)}; + withConfigMutationLockSync(() => { + writeFileSync(${JSON.stringify(enteredPath)}, "entered"); + process.exit(0); + }); + `; + const child = Bun.spawn([process.execPath, "-e", childSource], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env, OPENCODEX_HOME: testRoot }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + + expect(await waitForOwnedChild(child)).toBe(0); + expect(existsSync(enteredPath)).toBe(true); + expect(() => saveConfig(config(30300))).not.toThrow(); + expect(loadConfig().port).toBe(30300); +}); + +test("a throwing mutation releases the lock and leaves writers available", () => { + saveConfig(config()); + expect(() => withConfigMutationLockSync(() => { + throw new Error("mutation failed"); + })).toThrow("mutation failed"); + expect(() => saveConfig(config(50500))).not.toThrow(); + expect(loadConfig().port).toBe(50500); +}); + +test("management API maps config mutation lock contention to retryable 503", async () => { + saveConfig(config()); + const readyPath = join(testRoot, "mgmt-holder-ready"); + const releasePath = join(testRoot, "mgmt-holder-release"); + const configModuleUrl = pathToFileURL(join(import.meta.dir, "../src/config.ts")).href; + const childSource = ` + import { existsSync, writeFileSync } from "node:fs"; + import { withConfigMutationLockSync } from ${JSON.stringify(configModuleUrl)}; + withConfigMutationLockSync(() => { + writeFileSync(${JSON.stringify(readyPath)}, "ready"); + while (!existsSync(${JSON.stringify(releasePath)})) Bun.sleepSync(10); + }); + `; + const child = Bun.spawn([process.execPath, "-e", childSource], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env, OPENCODEX_HOME: testRoot }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + + try { + await waitForPath(readyPath); + const { handleManagementAPI } = await import("../src/server/management-api"); + const url = new URL("http://localhost/api/codex-auth/auto-switch"); + const response = await handleManagementAPI( + new ManagementRequest(url, { + method: "PUT", + headers: managementHeaders({ "content-type": "application/json" }), + body: JSON.stringify({ threshold: 50 }), + }), + url, + config(), + ); + expect(response?.status).toBe(503); + expect(await response!.json()).toMatchObject({ + error: "Configuration is busy; retry shortly", + code: "CONFIG_MUTATION_LOCK_UNAVAILABLE", + }); + } finally { + writeFileSync(releasePath, "release"); + expect(await waitForOwnedChild(child)).toBe(0); + } +}); diff --git a/tests/server-auth.test.ts b/tests/server-auth.test.ts index 36816eb5e..bccf223e4 100644 --- a/tests/server-auth.test.ts +++ b/tests/server-auth.test.ts @@ -1923,6 +1923,8 @@ describe("server local API auth", () => { } }, { timeout: SERVER_BUDGET_MS }); + // Same windows-latest contention budget as the stalled-body case above: abort + // teardown and harness stop can exceed Bun's default 5s under runner load. test("aborted 400 inspection never authorizes a pool retry", async () => { let releaseDispatch!: () => void; const dispatched = new Promise(resolve => { releaseDispatch = resolve; }); @@ -1944,7 +1946,7 @@ describe("server local API auth", () => { } finally { await stopPoolRetryHarness(harness); } - }); + }, { timeout: SERVER_BUDGET_MS }); test("invalid JSON 400 never authorizes a pool retry", async () => { const body = '{"detail":'; @@ -1955,7 +1957,7 @@ describe("server local API auth", () => { } finally { await stopPoolRetryHarness(harness); } - }); + }, { timeout: SERVER_BUDGET_MS }); test("missing or non-string detail never authorizes a pool retry", async () => { const bodies = ["{}", '{"detail":null}', '{"detail":400}', '{"detail":{"message":"unsupported"}}']; diff --git a/tests/storage-mutation-race.test.ts b/tests/storage-mutation-race.test.ts index 820c0736c..5204e3745 100644 --- a/tests/storage-mutation-race.test.ts +++ b/tests/storage-mutation-race.test.ts @@ -137,6 +137,25 @@ async function waitForPolicyJob( throw new Error("policy job did not finish"); } +/** Windows can keep SQLite/job handles briefly after stop; retry only transient cleanup codes. */ +function removeTree(path: string): void { + let lastError: unknown; + for (let attempt = 0; attempt < 50; attempt += 1) { + try { + rmSync(path, { recursive: true, force: true }); + return; + } catch (error) { + const code = error && typeof error === "object" && "code" in error + ? String(error.code) + : ""; + if (!new Set(["EPERM", "EBUSY", "ENOTEMPTY"]).has(code)) throw error; + lastError = error; + Bun.sleepSync(50); + } + } + throw lastError; +} + beforeEach(() => { previousHome = process.env.OPENCODEX_HOME; isolatedCodexHome = installIsolatedCodexHome("ocx-storage-mutation-race-codex-"); @@ -161,7 +180,7 @@ afterEach(() => { else process.env.OPENCODEX_HOME = previousHome; isolatedCodexHome?.restore(); isolatedCodexHome = null; - if (testDir) rmSync(testDir, { recursive: true, force: true }); + if (testDir) removeTree(testDir); testDir = ""; }); @@ -231,7 +250,7 @@ describe("storage mutation coordinator", () => { const server = startServer(0); try { - await enablePolicyAndRun(server.url); + const { startedAt } = await enablePolicyAndRun(server.url); await Bun.sleep(80); const preview = await previewDigest(server.url, 50); @@ -250,6 +269,10 @@ describe("storage mutation coordinator", () => { }); expect(restoreRes.status).toBe(409); expect((await restoreRes.json()).error).toBe("storage_mutation_busy"); + + // Drain the blocked policy job before stop/teardown — leaving it mid-block leaves + // Windows holding OPENCODEX_HOME (SQLite/job handles) and afterEach rmSync fails EBUSY. + await waitForPolicyJob(server.url, startedAt); } finally { await server.stop(true); }