diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 27881bc9c5..3ca45b3256 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -940,6 +940,8 @@ threshold. The Desktop/main account keeps its separate 98% hard lock. Set `codexPool.lowQuotaProtection` in configuration to pause accounts, record a log-and-API alert, or both; see [routing configuration](/reference/configuration/routing/#codex-pool-low-quota-protection). A pause takes effect for the next selection immediately, while saving it to disk is deferred. +Pool quota responses can authorize a pause only with valid usage and a captured credential +that is still current. A response without that proof may refresh displayed usage without pausing an account. Check this server’s authenticated `GET /api/codex-auth/low-quota-events` history for `logged` alerts or save failures. Manual resume remains in force for the current quota episode. The default alert reaches only the log and API; it does not produce a desktop or OS notification. diff --git a/src/codex/auth-api/pool-quota-probe.ts b/src/codex/auth-api/pool-quota-probe.ts index b95c4afb8e..d267f00540 100644 --- a/src/codex/auth-api/pool-quota-probe.ts +++ b/src/codex/auth-api/pool-quota-probe.ts @@ -373,7 +373,8 @@ export async function commitPoolQuotaResponse( } const validPolicyObservation = isValidWhamHistoryObservation(data); setAccountQuotaFromParsed(accountId, quota, writerGeneration, undefined, validPolicyObservation ? quota : null, - ctx.poolWriter && validPolicyObservation ? { writer: ctx.poolWriter, observedAt, source: "wham", raw: quota } : undefined); + ctx.poolWriter && validPolicyObservation ? { writer: ctx.poolWriter, observedAt, source: "wham", raw: quota } : undefined, + true); return { quota: getAccountQuota(accountId), needsReauth: false, diff --git a/src/codex/quota-auto-refresh.ts b/src/codex/quota-auto-refresh.ts index 54cd196aac..42fa040ee5 100644 --- a/src/codex/quota-auto-refresh.ts +++ b/src/codex/quota-auto-refresh.ts @@ -226,7 +226,7 @@ async function warmAccount(config: OcxConfig, accountId: string): Promise { if (isCodexAccountGenerationLive(accountId, token.generation)) { - applyAccountQuotaFromUpstreamHeaders(accountId, headers, writerGeneration, undefined, { poolWriter }); + applyAccountQuotaFromUpstreamHeaders(accountId, headers, writerGeneration, undefined, { poolWriter, poolResponse: true }); } } }); } catch (error) { diff --git a/src/codex/quota.ts b/src/codex/quota.ts index 7720c33c67..49c761110d 100644 --- a/src/codex/quota.ts +++ b/src/codex/quota.ts @@ -288,6 +288,10 @@ export function setAccountQuotaFromParsed( mainWriter?: MainQuotaWriter, policyQuota: MainPolicyQuotaObservation | null = accountId === MAIN_CODEX_ACCOUNT_ID ? quota : null, historyEvidence?: QuotaObservationEvidence, + // Pool responses carry their credential provenance separately from the optional history + // writer: when capture failed before dispatch, an absent writer must fail closed instead + // of reading as a writer-free legacy/login observation. + poolRequest = false, ): void { quota = withoutRetiredCodexQuota(quota); policyQuota = withoutRetiredCodexQuota(policyQuota); @@ -298,7 +302,10 @@ export function setAccountQuotaFromParsed( hydrateAccountQuotasFromDisk(); const legacyExisting = accountQuota.get(accountId); const updatedAt = Date.now(); - if (historyEvidence && historyEvidence.writer.accountId === accountId && isPoolQuotaWriterLive(historyEvidence.writer)) { + const livePoolEvidence = historyEvidence + ? historyEvidence.writer.accountId === accountId && isPoolQuotaWriterLive(historyEvidence.writer) + : !poolRequest; + if (historyEvidence && livePoolEvidence) { quotaHistory.append(historyEvidence.writer, { observedAt: historyEvidence.observedAt, source: historyEvidence.source, credentialGeneration: historyEvidence.writer.credentialGeneration, windows: historyWindows(historyEvidence.raw), }, updatedAt); @@ -323,7 +330,7 @@ export function setAccountQuotaFromParsed( schedulePersistAccountQuotas(); // Credits carry the previous usage tuple; they must not refresh its observation clock. if (!(quota.resetCredits !== undefined && !snapshotHasUsage(quota))) { - if (!isMain && policyQuota) observeCodexLowQuota(accountId, policyQuota); + if (!isMain && policyQuota && livePoolEvidence) observeCodexLowQuota(accountId, policyQuota); notifyCodexQuotaSnapshot(accountId, next); } } @@ -581,7 +588,7 @@ export function applyAccountQuotaFromUpstreamHeaders( headers: Headers, writerGeneration = captureConfigGeneration(), mainWriter?: MainQuotaWriter, - options?: { modelId?: string; poolWriter?: PoolQuotaWriter }, + options?: { modelId?: string; poolWriter?: PoolQuotaWriter; poolResponse?: boolean }, ): void { const quota = parseUpstreamQuotaHeaders(headers, options); if (!quota) return; @@ -591,7 +598,8 @@ export function applyAccountQuotaFromUpstreamHeaders( const validHistory = !["x-codex-primary-used-percent", "x-codex-secondary-used-percent", "x-codex-tertiary-used-percent"] .some(name => isInvalidPolicyUsagePercent(headers.get(name))); setAccountQuotaFromParsed(accountId, quota, writerGeneration, mainWriter, policyQuota, - options?.poolWriter && validHistory ? { writer: options.poolWriter, observedAt: Date.now(), source: "response-header", raw: quota } : undefined); + options?.poolWriter && validHistory ? { writer: options.poolWriter, observedAt: Date.now(), source: "response-header", raw: quota } : undefined, + options?.poolResponse === true); } export function updateAccountQuota( diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 4371aa8d73..526a8caeaa 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -1259,7 +1259,8 @@ export async function handleResponsesCompact( upstream.headers, authCtx.writerGeneration, authCtx.kind === "main-pool" ? authCtx.mainQuotaWriter : undefined, - { modelId: route.modelId, poolWriter: authCtx.kind === "pool" ? authCtx.poolQuotaWriter : undefined }, + { modelId: route.modelId, poolWriter: authCtx.kind === "pool" ? authCtx.poolQuotaWriter : undefined, + poolResponse: authCtx.kind === "pool" }, ); } recordCompactPoolOutcome(authCtx, upstream.status, { @@ -1328,7 +1329,7 @@ export async function handleResponsesCompact( if (outcomeCtx.kind === "pool") { const { applyAccountQuotaFromUpstreamHeaders } = await import("../../codex/quota"); applyAccountQuotaFromUpstreamHeaders(outcomeCtx.accountId, upstream.headers, outcomeCtx.writerGeneration, - undefined, { modelId: route.modelId, poolWriter: outcomeCtx.poolQuotaWriter }); + undefined, { modelId: route.modelId, poolWriter: outcomeCtx.poolQuotaWriter, poolResponse: true }); } const retryAfter = upstream.headers.get("retry-after"); const resetAt = [ diff --git a/src/server/responses/core-codex-account.ts b/src/server/responses/core-codex-account.ts index 1f6f299258..2f2c741336 100644 --- a/src/server/responses/core-codex-account.ts +++ b/src/server/responses/core-codex-account.ts @@ -127,7 +127,8 @@ export function codexWsQuotaObserver(authCtx: CodexAuthContext, provider: OcxPro const mainWriter = authCtx.kind === "main-pool" ? authCtx.mainQuotaWriter : undefined; return headers => { if (credentialGeneration !== undefined && !isCodexAccountGenerationLive(accountId, credentialGeneration)) return; - applyCapturedCodexQuota(accountId, headers, writerGeneration, mainWriter, { modelId, poolWriter: authCtx.kind === "pool" ? authCtx.poolQuotaWriter : undefined }); + applyCapturedCodexQuota(accountId, headers, writerGeneration, mainWriter, { modelId, poolWriter: authCtx.kind === "pool" ? authCtx.poolQuotaWriter : undefined, + poolResponse: authCtx.kind === "pool" }); }; } @@ -717,7 +718,8 @@ export async function retryCodexPoolOnAlternateAccount( firstResponse.headers, firstAuthCtx.writerGeneration, firstAuthCtx.kind === "main-pool" ? firstAuthCtx.mainQuotaWriter : undefined, - { modelId: route.modelId, poolWriter: firstAuthCtx.kind === "pool" ? firstAuthCtx.poolQuotaWriter : undefined }, + { modelId: route.modelId, poolWriter: firstAuthCtx.kind === "pool" ? firstAuthCtx.poolQuotaWriter : undefined, + poolResponse: firstAuthCtx.kind === "pool" }, ); } const deferFirstOutcome = shouldDeferCodexResetDerivedCooldown( diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index c3d55b1f44..eae2a72617 100644 --- a/src/server/responses/passthrough-delivery.ts +++ b/src/server/responses/passthrough-delivery.ts @@ -487,7 +487,8 @@ export async function deliverPassthroughResponse( if (!isCodexWsQuotaObservedResponse(upstreamResponse)) { applyAccountQuotaFromUpstreamHeaders(admissionState.authCtx.accountId, upstreamResponse.headers, admissionState.authCtx.writerGeneration, admissionState.authCtx.kind === "main-pool" ? admissionState.authCtx.mainQuotaWriter : undefined, - { modelId: route.modelId, poolWriter: admissionState.authCtx.kind === "pool" ? admissionState.authCtx.poolQuotaWriter : undefined }); + { modelId: route.modelId, poolWriter: admissionState.authCtx.kind === "pool" ? admissionState.authCtx.poolQuotaWriter : undefined, + poolResponse: admissionState.authCtx.kind === "pool" }); } if (terminalBodyWillRecord) { options.setTerminalOutcomeRecorder?.((status, httpStatusOverride) => { diff --git a/structure/providers/openai-accounts.md b/structure/providers/openai-accounts.md index 7c719ce7b9..ab490244b8 100644 --- a/structure/providers/openai-accounts.md +++ b/structure/providers/openai-accounts.md @@ -291,7 +291,7 @@ The credits module beside `src/codex/quota-types.ts` retains validated WHAM cred ## Low-quota protection -`src/codex/quota.ts` sends accepted usage observations through `src/codex/low-quota-observer.ts`; credits-only updates never replay carried usage into protection, and pool WHAM/header observations reach the policy only after raw percentages pass validation; clamped display bars cannot authorize a pause. The pool-account-only [configuration policy](../config.md#codex-pool-low-quota-protection) pauses live selection immediately, defers a bounded config save, and deduplicates account/window notices. Manual resume is respected across every qualifying window already active for that account until recovery or a new reset episode. Native main keeps its separate 98% hard lock. Each server registration owns a bounded status ledger; its authenticated management route exposes only its own account ids. The default alert is a log line plus a `logged` event, with no OS notification. +`src/codex/quota.ts` sends accepted usage observations through `src/codex/low-quota-observer.ts`; credits-only updates never replay carried usage into protection, and pool WHAM/header observations reach the policy only after raw percentages pass validation and the captured credential writer is still live; clamped display bars and responses from retired credentials cannot authorize a pause. The pool-account-only [configuration policy](../config.md#codex-pool-low-quota-protection) pauses live selection immediately, defers a bounded config save, and deduplicates account/window notices. Manual resume is respected across every qualifying window already active for that account until recovery or a new reset episode. Native main keeps its separate 98% hard lock. Each server registration owns a bounded status ledger; its authenticated management route exposes only its own account ids. The default alert is a log line plus a `logged` event, with no OS notification. A pool response that could not capture a writer at dispatch (credential already replaced, or the store unreadable) fails closed the same way: its snapshot still refreshes the display cache, but it never authorizes a policy pause. Writer-free legacy and login paths are explicitly distinct from that pool contract and keep their existing observation behavior. ## Bounded pool quota observations diff --git a/tests/codex-integration/low-quota-protection.test.ts b/tests/codex-integration/low-quota-protection.test.ts index ff4261c251..c09c98bfd4 100644 --- a/tests/codex-integration/low-quota-protection.test.ts +++ b/tests/codex-integration/low-quota-protection.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -9,8 +9,13 @@ import { createLowQuotaEventLedger } from "../../src/codex/low-quota-events"; import { registerCodexLowQuotaProtection, type LowQuotaRegistration } from "../../src/codex/low-quota-protection"; import { setCodexAccountPaused } from "../../src/codex/account-pause"; import { MAIN_CODEX_ACCOUNT_ID } from "../../src/codex/account-id"; -import { saveCodexAccountCredential } from "../../src/codex/account-store"; +import { capturePoolQuotaWriter, removeCodexAccountCredential, saveCodexAccountCredential } from "../../src/codex/account-store"; import { commitPoolQuotaResponse } from "../../src/codex/auth-api/pool-quota-probe"; +import * as authContextModule from "../../src/codex/auth-context"; +import { handleResponsesCompact } from "../../src/server/responses"; +import { baseCompactionBody, compactionRequest, completedPayload, drainCompactionResponseState } from "../helpers/compaction-routing-fixtures"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { codexWsQuotaObserver } from "../../src/server/responses/core-codex-account"; import { captureConfigGeneration } from "../../src/lib/state-store-sweeper"; import { applyAccountQuotaFromUpstreamHeaders, clearAccountQuota, getAccountQuota, setAccountQuotaFromParsed, updateAccountQuota } from "../../src/codex/quota"; import type { CodexAccount } from "../../src/types/accounts"; @@ -275,16 +280,19 @@ describe("low quota protection", () => { test("invalid raw WHAM usage stays display-only while valid usage pauses", async () => { const config = configWith(protection({ actions: { pause: true, notify: false } })); const registration = register(config, { persist: () => {} }); - const generation = saveCodexAccountCredential(ACCOUNT_A, { + const credential = { accessToken: "fixture-access", refreshToken: "fixture-refresh", expiresAt: Date.now() + 60_000, chatgptAccountId: "fixture-chatgpt-account", - }); + }; + const generation = saveCodexAccountCredential(ACCOUNT_A, credential); + const poolWriter = capturePoolQuotaWriter(ACCOUNT_A, { ...credential, generation }); + expect(poolWriter).toBeDefined(); const publish = (usedPercent: number) => commitPoolQuotaResponse( new Response(JSON.stringify({ rate_limit: { primary_window: { used_percent: usedPercent, limit_window_seconds: 604_800, } } }), { status: 200 }), { accountId: ACCOUNT_A, existing: null, configuredPlan: "plus", generation, - writerGeneration: captureConfigGeneration() }, + writerGeneration: captureConfigGeneration(), poolWriter }, ); await publish(150); expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(100); @@ -309,6 +317,163 @@ describe("low quota protection", () => { await registration.flush(); }); + test("a stale pool credential response cannot pause its replacement", () => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + const registration = register(config, { persist: () => {} }); + const first = { accessToken: "first-access", refreshToken: "first-refresh", + expiresAt: Date.now() + 60_000, chatgptAccountId: "first-chatgpt" }; + const generation = saveCodexAccountCredential(ACCOUNT_A, first); + const writer = capturePoolQuotaWriter(ACCOUNT_A, { ...first, generation }); + expect(writer).toBeDefined(); + removeCodexAccountCredential(ACCOUNT_A); + saveCodexAccountCredential(ACCOUNT_A, { ...first, accessToken: "replacement-access" }); + + applyAccountQuotaFromUpstreamHeaders(ACCOUNT_A, new Headers({ + "x-codex-primary-used-percent": "90", "x-codex-primary-window-minutes": "10080", + }), captureConfigGeneration(), undefined, { poolWriter: writer }); + + expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(90); + expect(config.pausedCodexAccountIds).toBeUndefined(); + expect(registration.hasPendingSave()).toBe(false); + }); + + test("a pool response whose writer capture failed cannot pause its replacement", () => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + const registration = register(config, { persist: () => {} }); + const first = { accessToken: "first-access", refreshToken: "first-refresh", + expiresAt: Date.now() + 60_000, chatgptAccountId: "first-chatgpt" }; + saveCodexAccountCredential(ACCOUNT_A, first); + // The credential is replaced before the delayed response lands, exactly when a + // dispatch-time capture could no longer produce a writer for the retired credential. + removeCodexAccountCredential(ACCOUNT_A); + saveCodexAccountCredential(ACCOUNT_A, { ...first, accessToken: "replacement-access" }); + + applyAccountQuotaFromUpstreamHeaders(ACCOUNT_A, new Headers({ + "x-codex-primary-used-percent": "90", "x-codex-primary-window-minutes": "10080", + }), captureConfigGeneration(), undefined, { poolResponse: true }); + + // The display snapshot still commits; only the policy observation fails closed. + expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(90); + expect(config.pausedCodexAccountIds).toBeUndefined(); + expect(registration.hasPendingSave()).toBe(false); + }); + + test("valid WHAM usage without a captured pool writer remains display-only", async () => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + const registration = register(config, { persist: () => {} }); + const generation = saveCodexAccountCredential(ACCOUNT_A, { + accessToken: "fixture-access", refreshToken: "fixture-refresh", + expiresAt: Date.now() + 60_000, chatgptAccountId: "fixture-chatgpt-account", + }); + await commitPoolQuotaResponse(new Response(JSON.stringify({ rate_limit: { primary_window: { + used_percent: 90, limit_window_seconds: 604_800, + } } }), { status: 200 }), { + accountId: ACCOUNT_A, existing: null, configuredPlan: "plus", generation, + writerGeneration: captureConfigGeneration(), + }); + expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(90); + expect(config.pausedCodexAccountIds).toBeUndefined(); + expect(registration.hasPendingSave()).toBe(false); + }); + + test.each([false, true])("WS pool metadata requires a captured live writer (captured=%s)", captured => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + const registration = register(config, { persist: () => {} }); + const credential = { accessToken: "fixture-access", refreshToken: "fixture-refresh", + expiresAt: Date.now() + 60_000, chatgptAccountId: "fixture-chatgpt-account" }; + const generation = saveCodexAccountCredential(ACCOUNT_A, credential); + const writer = capturePoolQuotaWriter(ACCOUNT_A, { ...credential, generation }); + expect(writer).toBeDefined(); + const observer = codexWsQuotaObserver({ + kind: "pool", accountId: ACCOUNT_A, generation, ...credential, + writerGeneration: captureConfigGeneration(), poolQuotaWriter: captured ? writer : undefined, + }, { adapter: "openai-responses", authMode: "forward", baseUrl: "https://chatgpt.com/backend-api/codex" }); + expect(observer).toBeDefined(); + observer!(new Headers({ "x-codex-primary-used-percent": "90", "x-codex-primary-window-minutes": "10080" })); + expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(90); + expect(config.pausedCodexAccountIds).toEqual(captured ? [ACCOUNT_A] : undefined); + expect(registration.hasPendingSave()).toBe(captured); + }); + + test("WS metadata from a retired generation cannot refresh or pause its replacement", () => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + const registration = register(config, { persist: () => {} }); + const credential = { accessToken: "fixture-access", refreshToken: "fixture-refresh", + expiresAt: Date.now() + 60_000, chatgptAccountId: "fixture-chatgpt-account" }; + const generation = saveCodexAccountCredential(ACCOUNT_A, credential); + const writer = capturePoolQuotaWriter(ACCOUNT_A, { ...credential, generation }); + expect(writer).toBeDefined(); + const observer = codexWsQuotaObserver({ + kind: "pool", accountId: ACCOUNT_A, generation, ...credential, + writerGeneration: captureConfigGeneration(), poolQuotaWriter: writer, + }, { adapter: "openai-responses", authMode: "forward", baseUrl: "https://chatgpt.com/backend-api/codex" }); + expect(observer).toBeDefined(); + removeCodexAccountCredential(ACCOUNT_A); + saveCodexAccountCredential(ACCOUNT_A, { ...credential, accessToken: "replacement-access" }); + observer!(new Headers({ "x-codex-primary-used-percent": "90", "x-codex-primary-window-minutes": "10080" })); + expect(getAccountQuota(ACCOUNT_A)).toBeNull(); + expect(config.pausedCodexAccountIds).toBeUndefined(); + expect(registration.hasPendingSave()).toBe(false); + }); + + test.each([false, true])("compact pool headers require a captured live writer (captured=%s)", async captured => { + const config = configWith(protection({ actions: { pause: true, notify: false } })); + config.defaultProvider = "openai"; + config.activeCodexAccountId = ACCOUNT_A; + config.providers = { openai: { adapter: "openai-responses", authMode: "forward", + baseUrl: "https://chatgpt.com/backend-api/codex", codexAccountMode: "pool" } }; + saveCodexAccountCredential(ACCOUNT_A, { accessToken: "fixture-access", refreshToken: "fixture-refresh", + expiresAt: Date.now() + 600_000, chatgptAccountId: "fixture-chatgpt-account" }); + updateAccountQuota(ACCOUNT_A, 10); + const registration = register(config, { persist: () => {} }); + const originalResolve = authContextModule.resolveCodexAuthContext; + let poolContexts = 0; + // The resolver still chooses a real fixture credential. Omit only its optional writer + // to exercise dispatch-time capture failure through the actual compact handler. + const resolve = spyOn(authContextModule, "resolveCodexAuthContext").mockImplementation(async (...args) => { + const ctx = await originalResolve(...args); + if (ctx.kind !== "pool") return ctx; + expect(ctx.poolQuotaWriter).toBeDefined(); + poolContexts++; + return captured ? ctx : { ...ctx, poolQuotaWriter: undefined }; + }); + const previousCodexHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = home; + const releaseSpendHome = acquireOwnedSpendHome(); + const originalFetch = globalThis.fetch; + let compactSends = 0; + globalThis.fetch = (async (input, init) => { + const request = new Request(input, init); + const url = new URL(request.url); + if (request.method === "GET" && url.pathname.endsWith("/models")) { + return Response.json({ models: [{ slug: "gpt-5.6-sol", supported_in_api: true, visibility: "list" }] }); + } + expect(url.pathname.endsWith("/responses/compact")).toBe(true); + compactSends++; + return Response.json(completedPayload("fixture compact result"), { headers: { + "x-codex-primary-used-percent": "90", "x-codex-primary-window-minutes": "10080", + } }); + }) as typeof fetch; + try { + const response = await handleResponsesCompact(compactionRequest(baseCompactionBody({ model: "gpt-5.6-sol" })), + config, { model: "", provider: "" }); + expect(response.status).toBe(200); + expect(compactSends).toBe(1); + expect(poolContexts).toBeGreaterThan(0); + expect(getAccountQuota(ACCOUNT_A)?.weeklyPercent).toBe(90); + expect(config.pausedCodexAccountIds).toEqual(captured ? [ACCOUNT_A] : undefined); + expect(registration.hasPendingSave()).toBe(captured); + } finally { + resolve.mockRestore(); + globalThis.fetch = originalFetch; + try { await drainCompactionResponseState(); } finally { + releaseSpendHome(); + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + } + } + }); + test("invalid raw legacy weekly usage stays display-only while valid usage pauses", async () => { const config = configWith(protection({ actions: { pause: true, notify: false } })); const registration = register(config, { persist: () => {} });