From 1ee8b0f9e243b07a8e64b8e9cd38ea65ce69cc8f Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 5 Oct 2026 00:47:48 +0000 Subject: [PATCH 1/4] fix(codex): enforce stored-account credit policy through dispatch Enforce credits-after-limit for exact stored selectors, credential acquisition, materialization, forced refresh copies and physical Responses sends. Map late policy refusals without credential quarantine and release unused recovery resources. Preserve spend-control parsing and independent main/Reserve policy. Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../content/docs/guides/codex-integration.md | 5 + .../docs/ko/guides/codex-integration.md | 5 + scripts/test-layout/layout.json | 1 + src/codex/account-credit-use.ts | 10 +- src/codex/auth-context.ts | 48 +++- src/codex/pool-credit-policy.ts | 67 ++++++ src/server/responses/adapter-dispatch.ts | 27 +++ src/server/responses/compact.ts | 15 +- src/server/responses/core-auth.ts | 13 ++ src/server/responses/core-codex-account.ts | 26 ++- src/server/responses/passthrough-dispatch.ts | 40 ++-- structure/codex-account-controls.md | 5 + structure/transports/responses.md | 2 +- .../codex-pool-credit-policy.test.ts | 209 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 1 + tests/responses/reserve-dispatch-ws.test.ts | 29 ++- .../responses-pool-401-refresh.test.ts | 133 ++++++++++- 17 files changed, 599 insertions(+), 37 deletions(-) create mode 100644 src/codex/pool-credit-policy.ts create mode 100644 tests/codex-integration/codex-pool-credit-policy.test.ts diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 114376d265b..bb838aca696 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -1424,3 +1424,8 @@ the experimental feature for production work. ## Streaming line endings The shared SSE decoder accepts LF, CRLF and standalone CR line endings, even when a delimiter spans network chunks. This allows compatible providers to stream events without requiring LF-only framing. + + +### Account-qualified requests and credits + +Choosing an account-qualified model does not enable credit spending. Stored accounts obey **Use credits after limit** during authentication, when credentials are prepared, and before Responses HTTP or WebSocket dispatch after pacing or retry waits. A held request reports the credit policy, not an authentication failure; wait for the reset, choose another account, or explicitly enable that account’s credit spending. diff --git a/docs-site/src/content/docs/ko/guides/codex-integration.md b/docs-site/src/content/docs/ko/guides/codex-integration.md index 1c069f19f19..96ccb3f82a9 100644 --- a/docs-site/src/content/docs/ko/guides/codex-integration.md +++ b/docs-site/src/content/docs/ko/guides/codex-integration.md @@ -408,3 +408,8 @@ opencodex가 managed [background service](/ko/reference/cli/lifecycle/#백그라 ## 스트리밍 줄바꿈 공유 SSE 디코더는 LF, CRLF 및 단독 CR 줄바꿈을 처리하며 네트워크 청크 사이에서 구분자가 나뉘어도 동작합니다. 호환 제공자는 LF 형식으로만 스트림을 만들 필요가 없습니다. + + +### 계정 직접 지정과 크레딧 사용 + +계정이 지정된 모델을 선택해도 크레딧 사용이 자동으로 허용되지는 않습니다. 저장된 계정은 인증 및 전송용 인증정보 준비 시에도 **한도 이후 크레딧 사용** 설정을 따릅니다. 페이싱이나 재시도 대기 후 Responses HTTP 또는 WebSocket 요청을 보내기 직전에도 현재 정책을 확인합니다. 제한된 요청은 인증 오류가 아니라 크레딧 정책을 안내합니다. 한도 초기화를 기다리거나 다른 계정을 선택하고, 크레딧 사용이 필요한 경우 해당 계정에서 명시적으로 허용하세요. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 61c524498e8..6cbbff3b4a7 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -68,6 +68,7 @@ "codex-credits-settings.test.ts": "codex-integration", "codex-credits-probes.test.ts": "codex-integration", "codex-credits-after-limit.test.ts": "codex-integration", + "codex-pool-credit-policy.test.ts": "codex-integration", "codex-credits-after-limit-main.test.ts": "codex-integration", "service-desktop-startup-health.test.ts": "service", "service-desktop-startup.test.ts": "service", diff --git a/src/codex/account-credit-use.ts b/src/codex/account-credit-use.ts index 4a8cc1fa916..1c7e256694f 100644 --- a/src/codex/account-credit-use.ts +++ b/src/codex/account-credit-use.ts @@ -3,6 +3,7 @@ import { deleteConfigTopLevelKey } from "../config/rebase-provenance"; import { isThirtyDayOnlyCodexPlan } from "./plan"; import { CODEX_EXHAUSTED_USAGE_PERCENT, + TERMINAL_SHORT_WINDOW_FRESHNESS_MS, isTerminalShortWindow, hasSpendableCodexCredits, resetAtToMs, @@ -73,8 +74,13 @@ export function codexUsageLimitResetAt(quota: StoredAccountQuota | null, plan: u if (!quota) return undefined; const full: number[] = []; if (isTerminalShortWindow(quota, now)) { - // A reset-less burst reading counts while it is fresh, so there is no instant to report. - full.push(typeof quota.shortResetAt === "number" && quota.shortResetAt > 0 ? resetAtToMs(quota.shortResetAt) : now); + // A reset-less burst reading holds until its observation goes stale: the freshness horizon + // is the deadline a retried request still meets, so report it rather than `now` (which + // collapsed the refusal's Retry-After to 1s against an account blocked for minutes). + // `isTerminalShortWindow` guarantees `shortObservedAt` is finite on this branch. + full.push(typeof quota.shortResetAt === "number" && quota.shortResetAt > 0 + ? resetAtToMs(quota.shortResetAt) + : (quota.shortObservedAt ?? now) + TERMINAL_SHORT_WINDOW_FRESHNESS_MS); } const longWindows: Array<[number | undefined, number | undefined]> = isThirtyDayOnlyCodexPlan(plan) ? [[quota.monthlyPercent, quota.monthlyResetAt]] diff --git a/src/codex/auth-context.ts b/src/codex/auth-context.ts index 13ef41a0adb..84929653161 100644 --- a/src/codex/auth-context.ts +++ b/src/codex/auth-context.ts @@ -1,3 +1,4 @@ +import { bindPoolCreditPolicy, poolCreditHoldResetAt, poolContextCreditHoldResetAt } from "./pool-credit-policy"; import { hasSpendableCodexCredits } from "./quota-types"; import { noteMainAccountActivity } from "./main-account-external-usage"; import { codexAccountPriorityFailbackEnabled } from "./account-priority"; @@ -525,6 +526,23 @@ export class CodexMainAccountHardLockError extends CodexAccountCooldownError { } } +/** A stored pool account's spending policy is independent of account selection. */ +export class CodexPoolAccountCreditsOffError extends CodexAccountCooldownError { + readonly resetAt: number; + + constructor(accountId: string, resetAt: number) { + super(accountId, resetAt); + this.name = "CodexPoolAccountCreditsOffError"; + this.resetAt = resetAt; + this.message = "The selected Codex account reached its usage limit, and spending credits is off or no fresh spendable balance is available." + + " Wait for the limit to reset, choose another account, or explicitly allow credits for this account in Codex Auth."; + } +} + +function assertPoolAccountCredits(accountId: string, resetAt: number | undefined): void { + if (resetAt !== undefined) throw new CodexPoolAccountCreditsOffError(accountId, resetAt); +} + /** The main login may not spend credits and one of its usage windows is full (#6334). */ export class CodexMainAccountCreditsOffError extends CodexAccountCooldownError { readonly resetAt?: number; @@ -597,7 +615,7 @@ export class CodexRecoveryWithheldError extends CodexAccountCooldownError { export type CodexAuthPolicyConfig = Readonly>; interface CodexAuthMaterializationOptions { @@ -725,6 +743,22 @@ export function unwrapUpstreamRetryEvidenceError(error: unknown): unknown { * Same evidence the hard lock reads, and no plan lookup: the plan can live in the physical auth * file, which several callers are forbidden to open, so every long window counts instead. */ +/** Recheck live account policy at the physical send, after pacing or retry backoff. */ +export function createCodexAuthDispatchGuard( + ctx: CodexAuthContext, + config: CodexAuthPolicyConfig, + modelId: string, + admission?: Pick, + terminalHelper = false, +): ((headers: Headers) => void) | undefined { + const reserveGuard = createCodexReserveDispatchGuard(ctx, config, modelId, admission, terminalHelper); + if (ctx.kind !== "pool") return reserveGuard; + return headers => { + assertPoolAccountCredits(ctx.accountId, poolContextCreditHoldResetAt(ctx, config)); + reserveGuard?.(headers); + }; +} + function mainCreditsHoldResetAt(config: Pick): number | undefined { if (codexAccountUsesCreditsAfterLimit(config, MAIN_CODEX_ACCOUNT_ID) && hasSpendableCodexCredits(getMainPolicyQuota())) return undefined; @@ -868,6 +902,7 @@ export function cooldownErrorMessage(err: CodexAccountCooldownError, accountSele // A credits-off refusal is a configuration policy, not a cooldown: clearing a cooldown // cannot lift it, so its own wording (wait for the reset or allow credits) is the remedy. || err instanceof CodexMainAccountCreditsOffError + || err instanceof CodexPoolAccountCreditsOffError || err instanceof CodexReserveUnavailableError // A transient-hold refusal is not a quota cooldown. Its own wording is the only accurate // one, and the quota recovery advice below would send the operator after a cooldown that @@ -917,6 +952,7 @@ export class CodexThreadAffinityExpiredError extends Error { export function shouldMarkAccountNeedsReauthForCodexAuthFailure(cause: unknown): boolean { return !(cause instanceof CodexMainAccountHardLockError) && !(cause instanceof CodexMainAccountCreditsOffError) + && !(cause instanceof CodexPoolAccountCreditsOffError) && !(cause instanceof CodexAccountValidationPendingError) && !(cause instanceof CodexReserveUnavailableError) && !(cause instanceof CodexCredentialGenerationConflictError) @@ -1408,6 +1444,7 @@ export async function resolveCodexAuthContext( // deferred credential must never become request auth through that fallback. try { assertCodexAccountValidationReady(accountId); + assertPoolAccountCredits(accountId, poolCreditHoldResetAt(policy, accountId, config)); } catch (cause) { // Nothing will reach upstream, so give the trial back instead of leaving the held account // unprobeable until the lease deadline lapses (#4701). @@ -1533,7 +1570,8 @@ export async function resolveCodexAuthContext( try { const token = await getValidCodexToken(accountId, { signal: options.signal }); assertCodexAccountValidationReady(accountId); - return { + assertPoolAccountCredits(accountId, poolCreditHoldResetAt(policy, accountId, config)); + return bindPoolCreditPolicy({ kind: "pool", accountId, writerGeneration, @@ -1548,11 +1586,12 @@ export async function resolveCodexAuthContext( ...(probeQuotaScope ? { probeQuotaScope } : {}), ...(affinityDecision ? { affinityDecision } : {}), ...(transientProbe ? { transientProbe } : {}), - }; + }, config, policy); } catch (cause) { releaseTransientProbeGrant(); if (probeLeaseId && probeQuotaScope) releaseCodexQuotaScopeProbeLease(accountId, probeQuotaScope, probeLeaseId); else if (probeLeaseId) releaseCodexQuotaProbeLease(accountId, probeLeaseId); + if (cause instanceof CodexPoolAccountCreditsOffError) throw cause; if (!options.signal?.aborted && shouldMarkAccountNeedsReauthForCodexAuthFailure(cause)) { markAccountNeedsReauth(accountId, writerGeneration); } @@ -1562,6 +1601,7 @@ export async function resolveCodexAuthContext( export function assertCodexAuthContextNotCooled(ctx: CodexAuthContext | undefined): void { if (ctx?.kind !== "pool" && ctx?.kind !== "main-pool") return; + if (ctx.kind === "pool") assertPoolAccountCredits(ctx.accountId, poolContextCreditHoldResetAt(ctx)); // A context holding the probe lease was deliberately admitted through the cooldown. if (ctx.probeLeaseId) return; const cooldown = getCodexQuotaHealthSnapshot(ctx.accountId, ctx.quotaScope); @@ -1577,6 +1617,7 @@ export function applyCodexAuthContextToProvider( ): OcxRuntimeProviderConfig { if (mode !== "pool" || (ctx.kind !== "pool" && ctx.kind !== "main-pool") || provider.authMode !== "forward") return provider; assertCodexAccountValidationReady(ctx.accountId); + if (ctx.kind === "pool") assertPoolAccountCredits(ctx.accountId, poolContextCreditHoldResetAt(ctx)); return { ...provider, _codexAccountOverride: { @@ -1620,6 +1661,7 @@ export function materializeCodexUpstreamAuth( } if (ctx.kind === "pool" || ctx.kind === "main-pool") { assertCodexAccountValidationReady(ctx.accountId); + if (ctx.kind === "pool") assertPoolAccountCredits(ctx.accountId, poolContextCreditHoldResetAt(ctx, options.config)); selected.set("authorization", `Bearer ${ctx.accessToken}`); selected.set("chatgpt-account-id", ctx.chatgptAccountId); if (ctx.kind === "main-pool") { diff --git a/src/codex/pool-credit-policy.ts b/src/codex/pool-credit-policy.ts new file mode 100644 index 00000000000..d1a69e1eadd --- /dev/null +++ b/src/codex/pool-credit-policy.ts @@ -0,0 +1,67 @@ +import type { OcxConfig } from "../types"; +import { codexUsageLimitResetAt, isCodexAccountHeldForCredits } from "./account-credit-use"; +import { MAIN_CODEX_ACCOUNT_ID } from "./main-account"; +import { isSelectableCodexPoolAccount } from "./account-id"; +import { getAccountQuota } from "./quota"; + +export type PoolCreditPolicyConfig = Readonly>; +type PoolContext = { kind: "pool"; accountId: string }; + +// Keep live policy references private: they must not enter DTOs or serialized auth context. +const policies = new WeakMap(); + +/** Read cached usage only. This helper never opens or refreshes a credential. */ +export function poolCreditHoldResetAt( + policy: PoolCreditPolicyConfig, + accountId: string, + catalog: PoolCreditPolicyConfig = policy, + now = Date.now(), +): number | undefined { + // Native main has an independent identity-bound quota policy. Never read its physical plan. + if (accountId === MAIN_CODEX_ACCOUNT_ID) return undefined; + const quota = getAccountQuota(accountId); + const plan = catalog.codexAccounts + ?.find(account => isSelectableCodexPoolAccount(account) && account.id === accountId)?.plan; + if (!isCodexAccountHeldForCredits(policy, accountId, quota, plan, now)) return undefined; + return codexUsageLimitResetAt(quota, plan, now); +} + +/** Remember policy by reference so later credential materialization rechecks current values. */ +export function bindPoolCreditPolicy( + context: T, + config: PoolCreditPolicyConfig, + policy: PoolCreditPolicyConfig = config, +): T { + policies.set(context, { config, policy }); + return context; +} + +/** + * Carry the resolver's live policy onto a context rebuilt by spread copy. The WeakMap keys on + * object identity, so `{ ...ctx }` alone loses the binding and every caller that omits an + * explicit override would skip credit-hold enforcement on the copy. A source with no binding + * leaves the target unbound rather than inventing a policy. + */ +export function rebindPoolCreditPolicy(source: S, target: T): T { + const bound = policies.get(source); + if (bound) policies.set(target, bound); + return target; +} + +/** Explicit policy wins; omitted materialization options retain the resolver's policy. */ +export function poolContextCreditHoldResetAt( + context: PoolContext, + override?: PoolCreditPolicyConfig, + now = Date.now(), +): number | undefined { + const bound = policies.get(context); + const policy = override ?? bound?.policy; + // Unbound contexts skip credit-hold enforcement entirely: callers that require + // enforcement must supply `override` (or resolve through bindPoolCreditPolicy). + if (!policy) return undefined; + const catalog = override?.codexAccounts ? override : bound?.config ?? policy; + return poolCreditHoldResetAt(policy, context.accountId, catalog, now); +} diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index e41c219f3f2..b109e2012d5 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/src/server/responses/adapter-dispatch.ts @@ -100,6 +100,9 @@ import { } from "../../lib/errors"; import { resolveClientRetryAfter } from "../../lib/retry-after"; import { cancelBodyOnAbort } from "../../lib/abort"; +import { createCodexAuthDispatchGuard, releaseCodexAuthContextProbeLease, unwrapUpstreamRetryEvidenceError } from "../../codex/auth-context"; +import { isCanonicalOpenAiForwardProvider } from "../../providers/openai-tiers"; +import { mapCodexAuthContextErrorToResponse } from "./codex-auth-error"; import { chargeWorkflowSends } from "../../lib/workflow-budget"; import { isAntigravityValidationRefusal } from "./antigravity-validation-refusal"; @@ -344,6 +347,8 @@ export async function prepareAdapterExchange( dispatchOverride: oauthDispatch(builtInitialRequest), providerName: route.providerName, modelId: route.modelId, + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), })); } else { @@ -384,6 +389,8 @@ export async function prepareAdapterExchange( dispatchOverride: oauthDispatch(builtInitialRequest), providerName: route.providerName, modelId: route.modelId, + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, })); }, { @@ -406,6 +413,13 @@ export async function prepareAdapterExchange( upstream.abort(); if (options.abortSignal?.aborted) return clientCancelledResponse(); const refusal = err instanceof UpstreamRetryEvidenceError ? err.cause : err; + const codexRefusal = mapCodexAuthContextErrorToResponse(unwrapUpstreamRetryEvidenceError(err), { + now: Date.now(), accountSelector: route.codexAccountNamespace, + }); + if (codexRefusal) { + releaseCodexAuthContextProbeLease(admissionState.authCtx); + return codexRefusal; + } // A pause committed during pacing is local admission policy, not a failed upstream. if (refusal instanceof OAuthAccountPausedError) { return formatErrorResponse(403, "permission_error", publicOAuthAuthenticationErrorMessage(refusal)); @@ -550,6 +564,8 @@ export async function prepareAdapterExchange( dispatchOverride: oauthDispatch(retryRequest), providerName: route.providerName, modelId: route.modelId, + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), }); }); @@ -601,6 +617,8 @@ export async function prepareAdapterExchange( dispatchOverride: oauthDispatch(retryRequest), providerName: route.providerName, modelId: route.modelId, + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, })); }, { @@ -624,6 +642,15 @@ export async function prepareAdapterExchange( retryRequest.releaseBodyObservation?.(); } } catch (err) { + const codexRefusal = !options.abortSignal?.aborted && mapCodexAuthContextErrorToResponse(unwrapUpstreamRetryEvidenceError(err), { + now: Date.now(), accountSelector: route.codexAccountNamespace, + }); + if (codexRefusal) { + cleanupUpstreamAbort(); + upstream.abort(); + releaseCodexAuthContextProbeLease(admissionState.authCtx); + return { failed: codexRefusal }; + } if (preserveFailureResponse && !replacementAdmitted && !options.abortSignal?.aborted) return { failed: preserveFailureResponse }; cleanupUpstreamAbort(); diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index b9def1d63df..4d2577399f4 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -53,7 +53,7 @@ import { createAdapterEventQueue, preflightAdapterEvents } from "../../adapters/ import { applyCodexAuthContextToProvider, callerCodexWorkspaceAccountId, - createCodexReserveDispatchGuard, + createCodexAuthDispatchGuard, unwrapUpstreamRetryEvidenceError, CodexMainProfileDrainingError, headersForCodexAuthContext, @@ -182,6 +182,7 @@ import { import { hasResponsesItemIdRepair, relaySseWithResponsesItemIdRepair } from "../responses-item-id-repair"; import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; import { codexAuthContextLogLabel } from "../../codex/account-label"; +import { rebindPoolCreditPolicy } from "../../codex/pool-credit-policy"; import { codexAccountGatedCanonicalWireModel, @@ -435,6 +436,9 @@ async function refreshPoolCompactContext(args: { generation: refreshed.generation, poolQuotaWriter: capturePoolQuotaWriter(authCtx.accountId, refreshed), }; + // See the core counterpart: the spread dropped the WeakMap-bound credit policy, so a hold + // or opt-out landing during the refresh await would go unchecked on the copy. + rebindPoolCreditPolicy(authCtx, refreshedAuthCtx); const refreshedProvider = applyCodexAuthContextToProvider( stripCodexRuntimeProviderFields(provider), refreshedAuthCtx, @@ -463,6 +467,13 @@ async function refreshPoolCompactContext(args: { if (isTerminalCompactPoolRefreshFailure(error)) { return { ok: false, quarantine: true, response: reauthResponse() }; } + // See the core counterpart: a policy refusal raised after a successful refresh maps like + // the admission path — a reset-bound 429, never a 503 mislabeled as a refresh failure. + const policyResponse = mapCodexAuthContextErrorToResponse(error, { + now: Date.now(), + accountSelector: args.codexAccountNamespace, + }); + if (policyResponse) return { ok: false, quarantine: false, response: policyResponse }; return { ok: false, quarantine: false, @@ -1042,7 +1053,7 @@ export async function handleResponsesCompact( providerName: route.providerName, modelId: route.modelId, beforeDispatch: isCanonicalOpenAiForwardProvider(sendProvider) - ? createCodexReserveDispatchGuard(sendAuthCtx, config, selectedModelId, admission) : undefined, + ? createCodexAuthDispatchGuard(sendAuthCtx, config, selectedModelId, admission) : undefined, }), // Every credential-bearing forward send gets manual redirects, not only // pool sends: direct mode carries the caller's credential too (#914). diff --git a/src/server/responses/core-auth.ts b/src/server/responses/core-auth.ts index 32d6f56d115..456bc2a42ee 100644 --- a/src/server/responses/core-auth.ts +++ b/src/server/responses/core-auth.ts @@ -41,6 +41,7 @@ import { markLocalRequestLogRefusal } from "../request-log"; import { CODEX_POOL_REFRESH_INCOMPLETE_LOG_REASON } from "../../codex/pool-refresh-backoff"; import { codexAuthContextLogLabel } from "../../codex/account-label"; import { forceRefreshMainAccountToken } from "../../codex/main-account"; +import { rebindPoolCreditPolicy } from "../../codex/pool-credit-policy"; /** Keep synthesized Claude identity out of request headers reused by policy/combo fallback. */ export function withClaudeNativeSession(headers: Headers, provider: OcxProviderConfig, sessionId?: string): Headers { @@ -474,6 +475,9 @@ export async function refreshPoolForwardAuth(args: { generation: refreshed.generation, poolQuotaWriter: capturePoolQuotaWriter(authCtx.accountId, refreshed), }; + // The spread dropped the WeakMap-bound credit policy: without the live binding a hold or + // opt-out landing during the refresh await would go unchecked on every no-override path. + rebindPoolCreditPolicy(authCtx, refreshedAuthCtx); const provider = applyCodexAuthContextToProvider( stripCodexRuntimeProviderFields(route.provider), refreshedAuthCtx, @@ -496,6 +500,15 @@ export async function refreshPoolForwardAuth(args: { response: formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"), }; } + // The credential itself refreshed; a policy refusal raised afterwards (a credit hold or an + // opt-out that landed during the await) is not a refresh failure. Map it the same way the + // admission path does — a reset-bound 429, never a 503 telling the caller to sign in — and + // never quarantine a valid account for a policy refusal. + const policyResponse = mapCodexAuthContextErrorToResponse(error, { + accountSelector: route.codexAccountNamespace, + now: Date.now(), + }); + if (policyResponse) return { ok: false, quarantine: false, response: policyResponse }; return { ok: false, quarantine: false, diff --git a/src/server/responses/core-codex-account.ts b/src/server/responses/core-codex-account.ts index 2c57bb798da..1864a7f6535 100644 --- a/src/server/responses/core-codex-account.ts +++ b/src/server/responses/core-codex-account.ts @@ -54,7 +54,8 @@ import { headersForCodexAuthContext, applyCodexAuthContextToProvider, stripCodexRuntimeProviderFields, - createCodexReserveDispatchGuard, + createCodexAuthDispatchGuard, + CodexPoolAccountCreditsOffError, } from "../../codex/auth-context"; import { ACCOUNT_GATED_NATIVE_OPENAI_MODELS } from "../../codex/catalog/native-models"; import { isRequestExecutionBudget } from "../../lib/request-execution-budget"; @@ -745,12 +746,21 @@ export async function retryCodexPoolOnAlternateAccount( // Only a combo reset-derived outcome is deferred. Retry-After, defaults, and // ordinary requests must block the first account before the alternate send. if (!deferFirstOutcome) recordFirstOutcome(); - const retryHeaders = headersForCodexAuthContext(callerAuthHeaders, retryAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission); - const retryProvider = applyCodexAuthContextToProvider( - stripCodexRuntimeProviderFields(route.provider), - retryAuthCtx, - "pool", - ); + let retryHeaders: Headers; + let retryProvider: ReturnType; + try { + retryHeaders = headersForCodexAuthContext(callerAuthHeaders, retryAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission); + retryProvider = applyCodexAuthContextToProvider(stripCodexRuntimeProviderFields(route.provider), retryAuthCtx, "pool"); + } catch (error) { + if (!(error instanceof CodexPoolAccountCreditsOffError)) throw error; + // Body/entitlement reads above can outlive credit consent. No alternate will send: + // release its reservation and probes, then let the owner map the policy refusal. + accountMovePermit?.release(); + releaseCodexAuthContextProbeLease(firstAuthCtx); + releaseCodexAuthContextProbeLease(retryAuthCtx); + await firstResponse.body?.cancel().catch(() => undefined); + return { kind: "transport", error, authCtx: retryAuthCtx }; + } const retryAdapter = resolveAdapter( resolveWireProtocolOverride(route.providerName, route.modelId, retryProvider, inboundWire, route.staticPolicy), config.cacheRetention, @@ -889,7 +899,7 @@ export async function retryCodexPoolOnAlternateAccount( modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(retryAuthCtx, route.provider, route.modelId), beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(retryAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + ? createCodexAuthDispatchGuard(retryAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), // Credential-bearing forward send: never follow a redirect into a // dead-host rejection after the credential was seen (#914). diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index ec47d4e6e77..21c10964526 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -37,7 +37,7 @@ import { codexProbeLeaseId, codexProbeQuotaScope, codexTransientProbeGrant, - createCodexReserveDispatchGuard, + createCodexAuthDispatchGuard, } from "../../codex/auth-context"; import { NamespaceToolCollisionError, @@ -87,7 +87,6 @@ import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, - storedPoolReplayDispatchNotifier, } from "./fetch-helpers"; import { classifyPoolRecoveryDispatch } from "../../routing/probe-lease"; import { clientCancelledResponse } from "./core-errors"; @@ -816,7 +815,7 @@ export async function preparePassthroughExchange( throw new Error("Credential selection changed before pre-output stream recovery"); } if (isCanonicalOpenAiForwardProvider(route.provider)) { - createCodexReserveDispatchGuard( + createCodexAuthDispatchGuard( admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, @@ -984,7 +983,7 @@ export async function preparePassthroughExchange( modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), route.provider.authMode === "forward") // Every real attempt response — including an intermediate 5xx the @@ -1089,7 +1088,7 @@ export async function preparePassthroughExchange( modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), route.provider.authMode === "forward") .then(adoptObservedResponse); @@ -1180,13 +1179,13 @@ export async function preparePassthroughExchange( // The replay-dispatched signal is what bounds the rest of this logical request, so it // has to describe a send that actually happened. fetchWithHeaderTimeout awaits pacing // admission BEFORE calling the executor, so signalling at the call site would spend the - // budget even when a rejected pacing wait means nothing reaches the network. Wrapping - // the executor moves the signal to the last moment before the send, where a throw from - // here on is a genuine transport attempt. The notifier is built once for the whole leg: - // it fires on the first dispatch, and a replacement is another send of the same replay - // rather than a second one to announce. - const oauthReplayExecutor = storedPoolReplayDispatchNotifier( - providerFetch(route.provider, options.codexWsRuntimeIdentity, { + // budget even when a rejected pacing wait means nothing reaches the network. Signal + // only after the physical-dispatch policy check succeeds; a local refusal is no send. + // A replacement is another send of the same replay, rather than a second announcement. + const replayGuard = isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined; + let replayNotified = false; + const oauthReplayExecutor = providerFetch(route.provider, options.codexWsRuntimeIdentity, { httpOnly: canonicalBufferedJson, nativeControl: nativeResponseControlEligible(route.provider, options.nativeControl) && options.inboundTransport === "websocket" && !options.comboAttempt && responseEffects.plaintextV2AgentMessageToolNames.size === 0 @@ -1195,11 +1194,14 @@ export async function preparePassthroughExchange( providerName: route.providerName, modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), - beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, - }), - codex401ReplayKind === "stored" ? options.onStoredPool401ReplayDispatched : undefined, - ); + beforeDispatch: headers => { + replayGuard?.(headers); + if (codex401ReplayKind === "stored" && !replayNotified) { + replayNotified = true; + options.onStoredPool401ReplayDispatched?.(); + } + }, + }); // Routed through the shared helper so an ambiguous reset on THIS leg answers with the // same refusal every other leg gives. A bare fetch here rejected instead, and the // caller's transport-failure path turns a rejection into a client-retryable 502 -- @@ -1346,7 +1348,7 @@ export async function preparePassthroughExchange( modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), route.provider.authMode === "forward") .then(adoptObservedResponse); @@ -1481,7 +1483,7 @@ export async function preparePassthroughExchange( modelId: route.modelId, onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + ? createCodexAuthDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, }), route.provider.authMode === "forward") .then(adoptObservedResponse); diff --git a/structure/codex-account-controls.md b/structure/codex-account-controls.md index b1e1fe4254b..b16526d0e3e 100644 --- a/structure/codex-account-controls.md +++ b/structure/codex-account-controls.md @@ -72,3 +72,8 @@ marks an account allowed to spend, independent of the display switch, which stil routing. Coverage: `tests/codex-integration/codex-credits-after-limit.test.ts`, `tests/codex-integration/codex-credits-after-limit-main.test.ts` and `gui/tests/codex-credit-spend.test.tsx`. + + +## Stored-account authentication policy + +Exact pool selectors also honor the credits-after-limit policy. `pool-credit-policy.ts` reads cached usage and configured plan metadata without credential I/O. Authentication checks before and after credential acquisition; provider overrides, materialization and recovery rechecks retain the live policy privately by context identity. `createCodexAuthDispatchGuard` rechecks live credit policy at HTTP and Responses WebSocket dispatch after pacing or retry backoff, alongside the independent Reserve guard. A policy refusal releases unused probes and never quarantines a valid credential. Coverage: `tests/codex-integration/codex-pool-credit-policy.test.ts`. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index ce5ac093c47..77ccc3bd134 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -231,7 +231,7 @@ post-namespace request fail with HTTP 400. The option-aware `openai` provider uses `openai-responses` with `authMode: "forward"`. Pool mode resolves main plus added accounts through affinity/quota/cooldown ownership; Direct forwards only the allowed Codex/OpenAI auth/session headers from the current request and short-circuits pool -state. `openai-apikey` uses its configured key and canonical API base URL. Missing credentials fail +state. Stored Pool credentials follow the [credit policy](../codex-account-controls.md#stored-account-authentication-policy) through refresh copies and physical HTTP/WebSocket dispatch; a policy refusal before transport contact announces no replay send. `openai-apikey` uses its configured key and canonical API base URL. Missing credentials fail within their route; neither route falls through to the other. See [`openai-tiers.md`](../providers/openai-tiers.md). diff --git a/tests/codex-integration/codex-pool-credit-policy.test.ts b/tests/codex-integration/codex-pool-credit-policy.test.ts new file mode 100644 index 00000000000..4592194f961 --- /dev/null +++ b/tests/codex-integration/codex-pool-credit-policy.test.ts @@ -0,0 +1,209 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { OcxConfig } from "../../src/types"; +import { + resolveCodexAuthContext, materializeCodexUpstreamAuth, applyCodexAuthContextToProvider, + assertCodexAuthContextNotCooled, CodexPoolAccountCreditsOffError, createCodexAuthDispatchGuard, + cooldownErrorMessage, cooldownErrorResponse, shouldMarkAccountNeedsReauthForCodexAuthFailure, +} from "../../src/codex/auth-context"; +import { rebindPoolCreditPolicy } from "../../src/codex/pool-credit-policy"; +import { TERMINAL_SHORT_WINDOW_FRESHNESS_MS } from "../../src/codex/quota-types"; +import { saveCodexAccountCredential } from "../../src/codex/account-store"; +import { isAccountNeedsReauth } from "../../src/codex/account-runtime-state"; +import { clearAccountNeedsReauth, clearAccountQuota, updateAccountQuota } from "../../src/codex/auth-api"; +import { getAccountQuota, parseUsageQuota, setAccountQuotaFromParsed } from "../../src/codex/quota"; +import { providerFetch, fetchWithHeaderTimeout } from "../../src/server/responses/fetch-helpers"; +import { + CODEX_QUOTA_PROBE_INTERVAL_MS, clearCodexUpstreamHealth, clearThreadAccountMap, + recordCodexUpstreamOutcome, tryAcquireCodexQuotaProbeLease, +} from "../../src/codex/routing"; +import { clearPoolRotationState } from "../../src/codex/pool-rotation"; +import { setAsyncIcaclsRunnerForTests, setIcaclsRunnerForTests } from "../../src/lib/windows-secret-acl"; +import { flushConfigDirHardeningForTests } from "../../src/config/paths"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +const ID = "credit-policy-fixture"; +const ICACLS_OK = { success: true, exitCode: 0, timedOut: false, stdout: "" }; +let dir = ""; +let previousHome: string | undefined; +let previousCodexHome: string | undefined; + +function config(): OcxConfig { + return { + providers: {}, port: 10100, hostname: "127.0.0.1", codexMainAccountHardLock: false, + codexAccounts: [{ id: ID, email: "fixture@example.test", isMain: false, plan: "pro" }], + creditCodexAccountIds: [], activeCodexAccountId: ID, autoSwitchThreshold: 0, + upstreamFailoverThreshold: 3, + } as OcxConfig; +} +function quota(percent = 100, resetAt = Date.now() + 3_600_000): void { + updateAccountQuota(ID, percent, resetAt); + setAccountQuotaFromParsed(ID, { credits: { hasCredits: true, balance: 42.5, observedAt: Date.now() } }); +} +const resolve = (cfg: OcxConfig) => resolveCodexAuthContext(new Headers(), cfg, "pool", { accountId: ID, modelId: "gpt-5.5" }); + +describe("stored-account credit policy at authentication", () => { + beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + dir = mkdtempSync(join(tmpdir(), "ocx-pool-credit-policy-")); + process.env.OPENCODEX_HOME = dir; + process.env.CODEX_HOME = dir; + setIcaclsRunnerForTests(() => ICACLS_OK); + setAsyncIcaclsRunnerForTests(async () => ICACLS_OK); + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); clearPoolRotationState(); + clearAccountNeedsReauth(ID); + saveCodexAccountCredential(ID, { + accessToken: "access-token-value-credit-policy", refreshToken: "fixture-refresh-credit-policy", + expiresAt: Date.now() + 3_600_000, chatgptAccountId: "fixture-workspace-credit-policy", + }); + }); + afterEach(async () => { + try { + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); clearPoolRotationState(); + clearAccountNeedsReauth(ID); + await flushConfigDirHardeningForTests(); + } finally { + setIcaclsRunnerForTests(null); setAsyncIcaclsRunnerForTests(null); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; else process.env.CODEX_HOME = previousCodexHome; + if (dir) removeTreeWithRetry(dir); + dir = ""; + } + }); + test("an exact selector respects the credits-off hold without quarantining the account", async () => { + quota(); + await expect(resolve(config())).rejects.toBeInstanceOf(CodexPoolAccountCreditsOffError); + expect(isAccountNeedsReauth(ID)).toBe(false); + }); + test("an exact selector still respects credits-off when the default mode is Direct", async () => { + quota(); + await expect(resolveCodexAuthContext(new Headers(), config(), "direct", { accountId: ID, modelId: "gpt-5.5" })) + .rejects.toBeInstanceOf(CodexPoolAccountCreditsOffError); + expect(isAccountNeedsReauth(ID)).toBe(false); + }); + test("a spending-control-only refusal revokes an opted-in exact account's cached permission", async () => { + const cfg = config(); cfg.creditCodexAccountIds = [ID]; quota(); + const revoked = parseUsageQuota({ spend_control: { reached: true } }); + expect(revoked?.credits).toBeNull(); + setAccountQuotaFromParsed(ID, revoked!); + await expect(resolve(cfg)).rejects.toBeInstanceOf(CodexPoolAccountCreditsOffError); + }); + test("explicit opt-in with current spendable evidence still materializes the selected credential", async () => { + const cfg = config(); cfg.creditCodexAccountIds = [ID]; quota(); + const ctx = await resolve(cfg); + expect(ctx.kind).toBe("pool"); + expect(materializeCodexUpstreamAuth(new Headers(), ctx).get("authorization")).toBe("Bearer access-token-value-credit-policy"); + }); + test("a below-limit exact account remains usable", async () => { + quota(99); + expect((await resolve(config())).kind).toBe("pool"); + }); + test("a later full window is checked even when materialization options omit config", async () => { + quota(99); const ctx = await resolve(config()); quota(); + expect(() => materializeCodexUpstreamAuth(new Headers(), ctx)).toThrow(CodexPoolAccountCreditsOffError); + }); + test("provider overrides recheck the same live policy", async () => { + const cfg = config(); cfg.creditCodexAccountIds = [ID]; quota(); + const ctx = await resolve(cfg); cfg.creditCodexAccountIds = []; + expect(() => applyCodexAuthContextToProvider({ adapter: "openai-responses", baseUrl: "https://example.test", authMode: "forward" }, ctx, "pool")) + .toThrow(CodexPoolAccountCreditsOffError); + }); + test("a recovery probe does not grant credit-spending permission", async () => { + quota(99); const ctx = await resolve(config()); + if (ctx.kind !== "pool") throw new Error("expected pool fixture"); + ctx.probeLeaseId = "fixture-probe"; quota(); + expect(() => assertCodexAuthContextNotCooled(ctx)).toThrow(CodexPoolAccountCreditsOffError); + }); + test("an explicit current materialization policy takes precedence", async () => { + const cfg = config(); cfg.creditCodexAccountIds = [ID]; quota(); + const ctx = await resolve(cfg); + expect(() => materializeCodexUpstreamAuth(new Headers(), ctx, { config: { creditCodexAccountIds: [] } })) + .toThrow(CodexPoolAccountCreditsOffError); + }); + test("an elapsed full window does not hold the account indefinitely", async () => { + quota(100, Date.now() - 1_000); + expect((await resolve(config())).kind).toBe("pool"); + }); + test("a reset-less full short window holds until the observation goes stale", async () => { + setAccountQuotaFromParsed(ID, { shortPercent: 100 }); + const observedAt = getAccountQuota(ID)?.shortObservedAt; + expect(observedAt).toBeDefined(); + const error = await resolve(config()).catch(cause => cause); + expect(error).toBeInstanceOf(CodexPoolAccountCreditsOffError); + // The refusal must tell the client when the hold ends — the freshness horizon — instead + // of reporting `now` and inviting an immediate retry against a still-blocked account. + expect(error.resetAt).toBe(observedAt! + TERMINAL_SHORT_WINDOW_FRESHNESS_MS); + const retryAfter = Number(cooldownErrorResponse(error).headers.get("Retry-After")); + expect(retryAfter).toBeGreaterThan(1); + expect(retryAfter).toBeLessThanOrEqual(Math.ceil(TERMINAL_SHORT_WINDOW_FRESHNESS_MS / 1_000)); + }); + test("a supplied short reset is the hold deadline", async () => { + const resetAt = Date.now() + 120_000; + setAccountQuotaFromParsed(ID, { shortPercent: 100, shortObservedAt: Date.now(), shortResetAt: resetAt }); + const error = await resolve(config()).catch(cause => cause); + expect(error).toBeInstanceOf(CodexPoolAccountCreditsOffError); + expect(error.resetAt).toBe(resetAt); + }); + test("a pool context rebuilt by copy keeps the resolver's live credit policy", async () => { + quota(99); + const ctx = await resolve(config()); + if (ctx.kind !== "pool") throw new Error("expected pool fixture"); + // The 401-refresh replay rebuilds the context by spread; the rebind carries the WeakMap + // entry so a hold landing during the refresh await is enforced on the copy too. + const copy = rebindPoolCreditPolicy(ctx, { ...ctx, accessToken: "rotated-access" }); + quota(); + expect(() => applyCodexAuthContextToProvider({ adapter: "openai-responses", baseUrl: "https://example.test", authMode: "forward" }, copy, "pool")) + .toThrow(CodexPoolAccountCreditsOffError); + expect(() => materializeCodexUpstreamAuth(new Headers(), copy)).toThrow(CodexPoolAccountCreditsOffError); + }); + test.each(["quota", "consent", "allowed"])("ordinary physical dispatch rechecks live policy after pacing: %s", async change => { + const cfg = config(); cfg.creditCodexAccountIds = [ID]; quota(change === "quota" ? 99 : 100); + const ctx = await resolve(cfg); + const headers = materializeCodexUpstreamAuth(new Headers(), ctx); + let sends = 0; + const provider = { adapter: "openai-responses" as const, authMode: "forward" as const, + baseUrl: "https://chatgpt.com/backend-api/codex", fetch: Object.assign(async () => { + sends++; return Response.json({}); + }, { preconnect() {} }) as typeof fetch }; + const executor = providerFetch(provider, undefined, { httpOnly: true, + beforeDispatch: createCodexAuthDispatchGuard(ctx, cfg, "gpt-5.5") }); + let release!: () => void; + executor.waitForPacing = () => new Promise(resolve => { release = resolve; }); + const pending = fetchWithHeaderTimeout(`${provider.baseUrl}/responses`, { method: "POST", headers, body: "{}" }, + new AbortController().signal, 1000, false, executor); + const observed = pending.catch(error => error); + if (change === "quota") { cfg.creditCodexAccountIds = []; quota(); } + else if (change === "consent") cfg.creditCodexAccountIds = []; + release(); + if (change === "allowed") { expect((await observed).status).toBe(200); expect(sends).toBe(1); } + else { expect(await observed).toBeInstanceOf(CodexPoolAccountCreditsOffError); expect(sends).toBe(0); } + }); + test("a mid-acquisition credit hold releases the due cooldown probe lease", async () => { + const cfg = config(); + quota(99); + const recordedAt = Date.now() - CODEX_QUOTA_PROBE_INTERVAL_MS - 1_000; + recordCodexUpstreamOutcome(cfg, ID, 429, { + resetAt: Math.floor((recordedAt + 4 * 24 * 60 * 60_000) / 1_000), + now: recordedAt, + fixedAccount: true, + }); + // The resolver stays synchronous until the credential await, so this lands between the + // two stored-account credit checks exactly as a limit observed mid-flight would. + const pending = resolveCodexAuthContext(new Headers(), cfg, "pool", { modelId: "gpt-5.5" }); + quota(); + await expect(pending).rejects.toBeInstanceOf(CodexPoolAccountCreditsOffError); + // The catch must hand the probe lease back: an unreleased lease would keep + // `probeLeaseId` set and block every subsequent probe regardless of the interval. + expect(tryAcquireCodexQuotaProbeLease(ID, Date.now() + CODEX_QUOTA_PROBE_INTERVAL_MS)).toBeTruthy(); + }); + test("policy refusal retains actionable wording and is not a reauthentication failure", () => { + const error = new CodexPoolAccountCreditsOffError(ID, Date.now() + 3_600_000); + expect(shouldMarkAccountNeedsReauthForCodexAuthFailure(error)).toBe(false); + expect(cooldownErrorMessage(error)).toBe(error.message); + expect(error.message).not.toContain(ID); + expect(error.message).not.toContain("clear-cooldown"); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e50b3203b50..b555d52c979 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -74,6 +74,7 @@ "codex-config-preservation.test.ts": "codex-integration", "codex-credits.test.ts": "codex-integration", "codex-credits-settings.test.ts": "codex-integration", "codex-credits-probes.test.ts": "codex-integration", "codex-credits-after-limit.test.ts": "codex-integration", + "codex-pool-credit-policy.test.ts": "codex-integration", "codex-credits-after-limit-main.test.ts": "codex-integration", "service-desktop-startup-health.test.ts": "service", "service-desktop-startup.test.ts": "service", "startup-health-packaged-probe.test.ts": "server", "discovered-native-models.test.ts": "codex-integration", "responses-hosted-image-delivery.test.ts": "responses", diff --git a/tests/responses/reserve-dispatch-ws.test.ts b/tests/responses/reserve-dispatch-ws.test.ts index 6bbf6cbb82e..bd9dce9c208 100644 --- a/tests/responses/reserve-dispatch-ws.test.ts +++ b/tests/responses/reserve-dispatch-ws.test.ts @@ -1,7 +1,8 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { codexWsUpstreamFetch } from "../../src/server/responses/ws-upstream"; import { providerFetch } from "../../src/server/responses/fetch-helpers"; -import { CodexReserveHelperUnsupportedError, CodexReserveUnavailableError, createCodexReserveDispatchGuard } from "../../src/codex/auth-context"; +import { CodexPoolAccountCreditsOffError, CodexReserveHelperUnsupportedError, CodexReserveUnavailableError, createCodexAuthDispatchGuard, createCodexReserveDispatchGuard } from "../../src/codex/auth-context"; +import { clearAccountQuota, setAccountQuotaFromParsed } from "../../src/codex/quota"; import { clearAccountNeedsReauth } from "../../src/codex/account-runtime-state"; import { clearCodexUpstreamHealthForAccount } from "../../src/codex/routing"; import { clearMainAccountInfoCache, observeMainQuotaCredential, observeMainQuotaIdentity } from "../../src/codex/main-account-cache"; @@ -58,6 +59,32 @@ afterEach(() => { }); describe("synchronous Reserve dispatch callbacks on WebSocket", () => { + test("a pool credit hold during delayed WS open refuses create and releases the socket without fallback", async () => { + install(); + const accountId = "credit-policy-ws-fixture"; + const config = { creditCodexAccountIds: [] }; + const ctx = { kind: "pool" as const, accountId, writerGeneration: 0, generation: 0, + accessToken: "fixture-reserve", chatgptAccountId: "fixture-workspace" }; + setAccountQuotaFromParsed(accountId, { weeklyPercent: 99, weeklyResetAt: Date.now() + 3_600_000 }); + let fallbacks = 0; + const fallback = Object.assign(async () => { fallbacks++; return new Response("unexpected"); }, { preconnect() {} }); + const requestInit = init(); + requestInit.body = JSON.stringify({ model: "gpt-5.5", input: "ping", stream: true }); + try { + const pending = codexWsUpstreamFetch(URL, requestInit, fallback, "1.4.0", undefined, + createCodexAuthDispatchGuard(ctx, config, "gpt-5.5")); + const observed = pending.catch(error => error); + const socket = DelayedWebSocket.instances[0]!; + setAccountQuotaFromParsed(accountId, { weeklyPercent: 100, weeklyResetAt: Date.now() + 3_600_000 }); + socket.dispatchEvent(new Event("open")); + expect(await observed).toBeInstanceOf(CodexPoolAccountCreditsOffError); + expect(socket.sent).toEqual([]); + expect(socket.closed).toBe(true); + expect(socket.listeners.size).toBe(0); + expect(fallbacks).toBe(0); + } finally { clearAccountQuota(accountId); } + }); + test.each([true, false])("valid-proof terminal helper with enabled-at-open=%s cannot confuse helper permission with conversation permission", async enabledAtOpen => { install(); clearAccountNeedsReauth("__main__"); diff --git a/tests/responses/responses-pool-401-refresh.test.ts b/tests/responses/responses-pool-401-refresh.test.ts index c6a6f0450bc..a538b5e1168 100644 --- a/tests/responses/responses-pool-401-refresh.test.ts +++ b/tests/responses/responses-pool-401-refresh.test.ts @@ -3,8 +3,9 @@ import { existsSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { createHash } from "node:crypto"; -import { clearAccountNeedsReauth, isAccountNeedsReauth } from "../../src/codex/auth-api"; +import { clearAccountNeedsReauth, clearAccountQuota, isAccountNeedsReauth, updateAccountQuota } from "../../src/codex/auth-api"; import { codexPoolAffinityKey } from "../../src/codex/auth-context"; +import { setAccountQuotaFromParsed } from "../../src/codex/quota"; import { clearCodexUpstreamHealth, clearThreadAccountMap, @@ -17,6 +18,8 @@ import { REQUEST_PACING_MAX_QUEUE_DEPTH, resetProviderRequestPacingForTest, setProviderRequestPacingLimitsForTest, + providerRequestPacingStatus, + waitForProviderRequestSlot, } from "../../src/providers/request-pacing"; import { clearResponseStateForTests, @@ -314,6 +317,7 @@ beforeEach(() => { releaseSpendHome = acquireOwnedSpendHome(); clearAccountNeedsReauth(ACCOUNT_ID); clearAccountNeedsReauth(OTHER_ACCOUNT_ID); + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); clearResponseStateMemoryForTests(); @@ -329,6 +333,7 @@ afterEach(() => { clearCompactHandoffRoutesForTests(); clearAccountNeedsReauth(ACCOUNT_ID); clearAccountNeedsReauth(OTHER_ACCOUNT_ID); + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); resetAgentTaskRecoveryState(); @@ -458,6 +463,132 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); }); + test.each(["/v1/responses", "/v1/responses/compact"] as const)( + "a credit hold landing during the forced refresh answers the credit policy, not a refresh failure (%s)", + async path => { + const harness = installHarness({ + refresh() { + // The account crosses the limit while the token endpoint is in flight: the + // replay's materialization must read the live hold, not the admission-time one. + updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + return Response.json({ + access_token: "refreshed-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + }, + }); + const response = path === "/v1/responses/compact" + ? await handleResponsesCompact(request(path), config(), { model: "", provider: "" } as RequestLogContext) + : await handleResponses(request(path), config(), { model: "", provider: "" } as RequestLogContext); + const body = await response.text(); + + // A policy refusal is not an incomplete refresh: the client gets the actionable 429, + // the rotated grant stays stored, and nothing replays on the refreshed bearer. + expect(response.status).toBe(429); + expect(body).toContain("spending credits"); + expect(body).not.toContain("credential refresh did not complete"); + expect(harness.sends).toEqual(["Bearer rejected-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expect(readStoredGeneration()).toBe(4); + }, + ); + + test.each(["/v1/responses", "/v1/responses/compact"] as const)( + "a credit hold while the initial request waits for pacing refuses without sending (%s)", async path => { + const cfg = config(); + cfg.providers.openai!.requestPacing = { enabled: true, maxConcurrentRequests: 1 }; + const blocker = await waitForProviderRequestSlot("openai", cfg.providers.openai!, "gpt-5.5"); + const harness = installHarness(); + const pending = path === "/v1/responses/compact" + ? handleResponsesCompact(request(path, { model: "work/gpt-5.5" }), cfg, { model: "", provider: "" } as RequestLogContext) + : handleResponses(request(path, { model: "work/gpt-5.5" }), cfg, { model: "", provider: "" } as RequestLogContext); + try { + for (let attempt = 0; attempt < 500 && providerRequestPacingStatus("openai", cfg.providers.openai!).queued === 0; attempt++) await Bun.sleep(1); + expect(providerRequestPacingStatus("openai", cfg.providers.openai!).queued).toBe(1); + updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + blocker?.release(); + const response = await pending; + expect(response.status).toBe(429); + expect(Number(response.headers.get("Retry-After"))).toBeGreaterThan(1); + expect(await response.text()).toContain("spending credits"); + expect(harness.sends).toEqual([]); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + } finally { blocker?.release(); resetProviderRequestPacingForTest(); } + }, + ); + + test.each([false, true])("portable compaction rechecks queued stored-account credits (opt-in=%s)", async optedIn => { + const cfg = config(); cfg.codexMainAccountHardLock = false; + cfg.providers.openai!.requestPacing = { enabled: true, maxConcurrentRequests: 1 }; + cfg.providers.backup = { adapter: "openai-chat", authMode: "key", apiKey: "backup-test-key", baseUrl: "https://backup.example.test" }; + cfg.creditCodexAccountIds = optedIn ? [ACCOUNT_ID] : []; + updateAccountQuota(ACCOUNT_ID, 99, Date.now() + 3_600_000); + const blocker = await waitForProviderRequestSlot("openai", cfg.providers.openai!, "gpt-5.5"); + const harness = installHarness({ responseForSend: () => Response.json({ id: "resp_summary", object: "response", status: "completed", + output: [{ type: "message", role: "assistant", content: [{ type: "output_text", text: "summary" }] }] }) }); + const pending = handleResponses(request("/v1/responses", { model: "work/gpt-5.5", input: [ + { type: "message", role: "user", content: "history" }, { type: "compaction_trigger" }, + ] }), cfg, { model: "", provider: "" } as RequestLogContext, + { compactionRoutingOverride: { sourceModel: "backup/model" } }); + try { + for (let attempt = 0; attempt < 500 && providerRequestPacingStatus("openai", cfg.providers.openai!).queued === 0; attempt++) await Bun.sleep(1); + expect(providerRequestPacingStatus("openai", cfg.providers.openai!).queued).toBe(1); + updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + setAccountQuotaFromParsed(ACCOUNT_ID, { credits: { hasCredits: true, balance: 42.5, observedAt: Date.now() } }); + blocker?.release(); + const response = await pending; + expect(response.status).toBe(optedIn ? 200 : 429); + const body = await response.text(); + expect(body).toContain(optedIn ? "compaction" : "spending credits"); + expect(harness.sends).toEqual(optedIn ? ["Bearer rejected-access"] : []); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + } finally { blocker?.release(); resetProviderRequestPacingForTest(); } + }); + + test("a post-refresh credit hold in the pacing queue does not announce a replay send", async () => { + const cfg = config(); + cfg.providers.openai!.requestPacing = { enabled: true, minIntervalMs: 200 }; + const harness = installHarness(); + let dispatchSignals = 0; + const pending = handleResponses(request("/v1/responses"), cfg, { model: "", provider: "" } as RequestLogContext, + { onStoredPool401ReplayDispatched: () => { dispatchSignals++; } }); + try { + for (let attempt = 0; attempt < 500 && !(harness.refreshes.length && providerRequestPacingStatus("openai", cfg.providers.openai!).queued); attempt++) await Bun.sleep(1); + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(providerRequestPacingStatus("openai", cfg.providers.openai!).queued).toBe(1); + updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + const response = await pending; + expect(response.status).toBe(429); + expect(dispatchSignals).toBe(0); + expect(harness.sends).toEqual(["Bearer rejected-access"]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + } finally { resetProviderRequestPacingForTest(); } + }); + + test("a credit hold during same-account model recovery returns a policy response without replay", async () => { + const gatedModel = "gpt-daybreak-blue-latest"; + const harness = installHarness({ responseForSend: () => Response.json({ + detail: `The '${gatedModel}' model is not supported when using Codex with a ChatGPT account.`, + }, { status: 400 }) }); + const response = await handleResponses(request("/v1/responses", { model: gatedModel }), config(), + { model: "", provider: "" } as RequestLogContext, { + resolveCodexModelEntitlements: async () => { + if (harness.sends.length) updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + return { modelsByAccount: new Map([[ACCOUNT_ID, new Set([gatedModel])]]), + confirmedAccountIds: new Set([ACCOUNT_ID]), credentialIdentities: new Map() }; + }, + }); + expect(response.status).toBe(429); + expect(await response.text()).toContain("spending credits"); + expect(harness.sends).toEqual(["Bearer rejected-access"]); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); + test("Responses does not compose a stored-account replay 429 with another Pool account", async () => { writeStoredAccount({ [OTHER_ACCOUNT_ID]: storedRecord({ From d9c8e561dbe6c0797bcddac2d6aeffc5f18d7c3b Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:30:14 +0900 Subject: [PATCH 2/4] fix(codex): recheck stored credit policy before sidecar sends --- .../content/docs/guides/codex-integration.md | 2 + .../docs/ko/guides/codex-integration.md | 2 + scripts/test-layout/layout.json | 1 + src/providers/openai-sidecar-credit.ts | 26 ++ src/providers/openai-sidecar.ts | 4 + src/server/responses/sidecar-execution.ts | 1 + src/server/search.ts | 4 + src/vision/describe.ts | 27 +- src/vision/index.ts | 1 + src/web-search/executor.ts | 27 +- src/web-search/loop.ts | 3 +- src/web-search/passthrough-bridge.ts | 1 + src/web-search/run-turn-loop.ts | 2 +- structure/codex-account-controls.md | 2 + structure/providers-and-adapters.md | 1 + structure/runtime.md | 2 +- .../codex-sidecar-credit-policy.test.ts | 311 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 1 + 18 files changed, 395 insertions(+), 23 deletions(-) create mode 100644 src/providers/openai-sidecar-credit.ts create mode 100644 tests/codex-integration/codex-sidecar-credit-policy.test.ts diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index bb838aca696..d6674c4a270 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -1429,3 +1429,5 @@ The shared SSE decoder accepts LF, CRLF and standalone CR line endings, even whe ### Account-qualified requests and credits Choosing an account-qualified model does not enable credit spending. Stored accounts obey **Use credits after limit** during authentication, when credentials are prepared, and before Responses HTTP or WebSocket dispatch after pacing or retry waits. A held request reports the credit policy, not an authentication failure; wait for the reset, choose another account, or explicitly enable that account’s credit spending. + +Stored-account vision and web-search helpers also recheck this policy before sending, including retries. If consent changes or a limit is reached after helper selection, the helper reports the policy refusal without sending that attempt. The standalone search relay returns a reset-bound 429. Caller-owned Direct credentials retain their existing behavior. diff --git a/docs-site/src/content/docs/ko/guides/codex-integration.md b/docs-site/src/content/docs/ko/guides/codex-integration.md index 96ccb3f82a9..7b5bf5ecd5f 100644 --- a/docs-site/src/content/docs/ko/guides/codex-integration.md +++ b/docs-site/src/content/docs/ko/guides/codex-integration.md @@ -413,3 +413,5 @@ opencodex가 managed [background service](/ko/reference/cli/lifecycle/#백그라 ### 계정 직접 지정과 크레딧 사용 계정이 지정된 모델을 선택해도 크레딧 사용이 자동으로 허용되지는 않습니다. 저장된 계정은 인증 및 전송용 인증정보 준비 시에도 **한도 이후 크레딧 사용** 설정을 따릅니다. 페이싱이나 재시도 대기 후 Responses HTTP 또는 WebSocket 요청을 보내기 직전에도 현재 정책을 확인합니다. 제한된 요청은 인증 오류가 아니라 크레딧 정책을 안내합니다. 한도 초기화를 기다리거나 다른 계정을 선택하고, 크레딧 사용이 필요한 경우 해당 계정에서 명시적으로 허용하세요. + +저장된 계정을 사용하는 이미지 설명과 웹 검색 도우미도 재시도를 포함한 매 전송 직전에 이 정책을 확인합니다. 도우미 선택 이후 허용 설정이 바뀌거나 한도에 도달하면 해당 요청을 보내지 않고 정책 제한을 안내합니다. 독립 검색 릴레이는 초기화 시점에 맞춘 429를 반환하며, 호출자가 직접 제공한 Direct 인증정보의 기존 동작은 유지합니다. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 6cbbff3b4a7..fbce126f29c 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -69,6 +69,7 @@ "codex-credits-probes.test.ts": "codex-integration", "codex-credits-after-limit.test.ts": "codex-integration", "codex-pool-credit-policy.test.ts": "codex-integration", + "codex-sidecar-credit-policy.test.ts": "codex-integration", "codex-credits-after-limit-main.test.ts": "codex-integration", "service-desktop-startup-health.test.ts": "service", "service-desktop-startup.test.ts": "service", diff --git a/src/providers/openai-sidecar-credit.ts b/src/providers/openai-sidecar-credit.ts new file mode 100644 index 00000000000..8d70fb38cf2 --- /dev/null +++ b/src/providers/openai-sidecar-credit.ts @@ -0,0 +1,26 @@ +import { + CodexPoolAccountCreditsOffError, + releaseCodexAuthContextProbeLease, + unwrapUpstreamRetryEvidenceError, + type CodexAuthContext, + type CodexAuthPolicyConfig, +} from "../codex/auth-context"; +import { poolContextCreditHoldResetAt } from "../codex/pool-credit-policy"; + +export function createOpenAiSidecarCreditGuard( + context: CodexAuthContext, + policy: CodexAuthPolicyConfig, +): (() => void) | undefined { + if (context.kind !== "pool") return undefined; + return () => { + const resetAt = poolContextCreditHoldResetAt(context, policy); + if (resetAt === undefined) return; + releaseCodexAuthContextProbeLease(context); + throw new CodexPoolAccountCreditsOffError(context.accountId, resetAt); + }; +} + +export function openAiSidecarCreditRefusal(error: unknown): CodexPoolAccountCreditsOffError | undefined { + const cause = unwrapUpstreamRetryEvidenceError(error); + return cause instanceof CodexPoolAccountCreditsOffError ? cause : undefined; +} diff --git a/src/providers/openai-sidecar.ts b/src/providers/openai-sidecar.ts index 8fa158667b4..b0ce2103aab 100644 --- a/src/providers/openai-sidecar.ts +++ b/src/providers/openai-sidecar.ts @@ -1,4 +1,5 @@ import { resolveProviderApiKey } from "./key-store"; +import { createOpenAiSidecarCreditGuard } from "./openai-sidecar-credit"; import { CodexPoolAuthenticationError, headersForCodexAuthContext, @@ -31,6 +32,7 @@ export interface OpenAiForwardSidecarCandidate { export interface ResolvedOpenAiForwardSidecar extends OpenAiForwardSidecarCandidate { authContext: CodexAuthContext; headers: Headers; + beforeDispatch?: () => void; recordOutcome?: (outcome: CodexUpstreamOutcome) => void; /** Hand back an acquired recovery probe when no sidecar request reached upstream. */ releaseProbeLease?: () => void; @@ -176,6 +178,7 @@ export async function resolveFirstUsableOpenAiSidecar( ...candidate, authContext, headers: selectedHeaders, + beforeDispatch: createOpenAiSidecarCreditGuard(authContext, policy), recordOutcome: (outcome: CodexUpstreamOutcome) => recordCodexUpstreamOutcome( config, authContext.accountId, @@ -226,6 +229,7 @@ export async function resolveFirstUsableOpenAiSidecar( ...candidate, authContext, headers: selectedHeaders, + beforeDispatch: createOpenAiSidecarCreditGuard(authContext, policy), ...(authContext.kind === "pool" || authContext.kind === "main-pool" ? { recordOutcome: (outcome: CodexUpstreamOutcome) => recordCodexUpstreamOutcome( diff --git a/src/server/responses/sidecar-execution.ts b/src/server/responses/sidecar-execution.ts index e34b4e80ef0..d3e71f07a23 100644 --- a/src/server/responses/sidecar-execution.ts +++ b/src/server/responses/sidecar-execution.ts @@ -510,6 +510,7 @@ export async function executeResponsesSidecars( transportState.bindKeyUsageFromBridge(usage); }, recordSidecarOutcome: wsPlan.forwardSidecar?.recordOutcome, + beforeSidecarDispatch: wsPlan.forwardSidecar?.beforeDispatch, connectTimeoutMs: config.connectTimeoutMs ?? 200_000, routedModelStallTimeoutMs: wsPlan.routedModelStallTimeoutMs, stallTimeoutSec: wsPlan.stallTimeoutSec, diff --git a/src/server/search.ts b/src/server/search.ts index 33d52c71528..7dd14464839 100644 --- a/src/server/search.ts +++ b/src/server/search.ts @@ -11,6 +11,7 @@ * paid backend than the one the operator named. */ import { formatErrorResponse } from "../bridge"; +import { openAiSidecarCreditRefusal } from "../providers/openai-sidecar-credit"; import { CodexAccountCooldownError, codexMainProfileDrainingResponse, @@ -202,6 +203,7 @@ export async function handleSearch( const sidecarExit = sidecarEnter("search"); let upstreamResponse: Response | undefined; try { + upstream.beforeDispatch?.(); upstreamResponse = await fetch(url, { method: "POST", headers, @@ -233,6 +235,8 @@ export async function handleSearch( if (req.signal.aborted) { return formatErrorResponse(499, "client_closed_request", "search request canceled by client"); } + const policyRefusal = openAiSidecarCreditRefusal(err); + if (policyRefusal) return cooldownErrorResponse(policyRefusal, Date.now(), accountNamespace); if (linkedSignal.signal.aborted || (err instanceof Error && err.name === "TimeoutError")) { upstream.recordOutcome?.("timeout"); return formatErrorResponse(504, "upstream_error", "search upstream timed out"); diff --git a/src/vision/describe.ts b/src/vision/describe.ts index 14629f01664..5b8324a4c60 100644 --- a/src/vision/describe.ts +++ b/src/vision/describe.ts @@ -8,6 +8,7 @@ import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream- import { parseSidecarSSE } from "../web-search/parse"; import type { SidecarOutcomeRecorder } from "../web-search/executor"; import { NATIVE_RESERVE_MODEL } from "../codex/catalog/native-models"; +import { openAiSidecarCreditRefusal } from "../providers/openai-sidecar-credit"; export interface VisionSettings { model: string; @@ -60,6 +61,7 @@ export async function describeImage( settings: VisionSettings, abortSignal?: AbortSignal, recordOutcome?: SidecarOutcomeRecorder, + beforeDispatch?: () => void, ): Promise { if (settings.reserveCompatibility && settings.model === NATIVE_RESERVE_MODEL) { return { text: "", error: "Luna Reserve compatibility is only available as a conversation model, not a vision helper. Choose another vision helper model." }; @@ -98,16 +100,19 @@ export async function describeImage( const res = await fetchWithResetRetry( // The replay needs `keepalive: false` to abandon the half-closed pooled socket; Bun has // ignored a bare `Connection: close` (oven-sh/bun#20492). - recovery => fetch(`${forwardProvider.baseUrl}/responses`, applyUpstreamRecoveryInit({ - method: "POST", - headers, - body: JSON.stringify(body), - signal: linkedSignal.signal, - // Credential-bearing: do not follow a cross-origin 3xx. Bun strips `Authorization` - // across origins but forwards nonstandard headers such as `chatgpt-account-id`, - // `session_id`, and `x-codex-turn-metadata` to the redirect target. - redirect: "manual", - }, recovery)), + recovery => { + beforeDispatch?.(); + return fetch(`${forwardProvider.baseUrl}/responses`, applyUpstreamRecoveryInit({ + method: "POST", + headers, + body: JSON.stringify(body), + signal: linkedSignal.signal, + // Credential-bearing: do not follow a cross-origin 3xx. Bun strips `Authorization` + // across origins but forwards nonstandard headers such as `chatgpt-account-id`, + // `session_id`, and `x-codex-turn-metadata` to the redirect target. + redirect: "manual", + }, recovery)); + }, { replaySafe: true, abortSignal: linkedSignal.signal, label: "vision-sidecar" }, ); const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); @@ -129,6 +134,8 @@ export async function describeImage( detachBodyGuard(); } } catch (e) { + const policyRefusal = openAiSidecarCreditRefusal(e); + if (policyRefusal) return { text: "", error: policyRefusal.message }; const kind = e instanceof Error && e.name === "TimeoutError" ? "timeout" : "connect_error"; const callerAborted = abortSignal?.aborted === true && linkedSignal.signal.aborted diff --git a/src/vision/index.ts b/src/vision/index.ts index d22d5502f96..a38fc148069 100644 --- a/src/vision/index.ts +++ b/src/vision/index.ts @@ -269,6 +269,7 @@ async function executeDescription( plan.settings, abortSignal, recordSidecarOutcome, + plan.forwardSidecar.beforeDispatch, ); } diff --git a/src/web-search/executor.ts b/src/web-search/executor.ts index e6abc5004ae..c2b6503d78f 100644 --- a/src/web-search/executor.ts +++ b/src/web-search/executor.ts @@ -15,6 +15,7 @@ import { withUpstreamHttpVersion } from "../lib/upstream-http-version"; import { parseSidecarSSE, type WebSearchResult } from "./parse"; import type { CodexUpstreamOutcome } from "../codex/routing"; import { NATIVE_RESERVE_MODEL } from "../codex/catalog/native-models"; +import { openAiSidecarCreditRefusal } from "../providers/openai-sidecar-credit"; export interface SidecarSettings { model: string; @@ -78,6 +79,7 @@ export async function runWebSearch( settings: SidecarSettings, abortSignal?: AbortSignal, recordOutcome?: SidecarOutcomeRecorder, + beforeDispatch?: () => void, ): Promise { if (settings.reserveCompatibility && settings.model === NATIVE_RESERVE_MODEL) { return { text: "", sources: [], error: "Luna Reserve compatibility is only available as a conversation model, not a search helper. Choose another search helper model." }; @@ -115,16 +117,19 @@ export async function runWebSearch( // defined init, and withUpstreamHttpVersion spreads the result, so `protocol` and the // recovery fields (`connection: close` + Bun's transport-level `keepalive: false`) survive // together. The reverse order needs a `?? init` fallback to type-check at all. - recovery => fetch(url, withUpstreamHttpVersion(url, applyUpstreamRecoveryInit({ - method: "POST", - headers, - body: JSON.stringify(body), - signal: linkedSignal.signal, - // Credential-bearing: do not follow a cross-origin 3xx. Bun strips `Authorization` - // across origins but forwards nonstandard headers such as `chatgpt-account-id`, - // `session_id`, and `x-codex-turn-metadata` to the redirect target. - redirect: "manual", - }, recovery), forwardProvider)), + recovery => { + beforeDispatch?.(); + return fetch(url, withUpstreamHttpVersion(url, applyUpstreamRecoveryInit({ + method: "POST", + headers, + body: JSON.stringify(body), + signal: linkedSignal.signal, + // Credential-bearing: do not follow a cross-origin 3xx. Bun strips `Authorization` + // across origins but forwards nonstandard headers such as `chatgpt-account-id`, + // `session_id`, and `x-codex-turn-metadata` to the redirect target. + redirect: "manual", + }, recovery), forwardProvider)); + }, { replaySafe: true, abortSignal: linkedSignal.signal, @@ -180,6 +185,8 @@ export async function runWebSearch( detachBodyGuard(); } } catch (e) { + const policyRefusal = openAiSidecarCreditRefusal(e); + if (policyRefusal) return { text: "", sources: [], error: policyRefusal.message }; const kind = e instanceof Error && e.name === "TimeoutError" ? "timeout" : "connect_error"; const callerAborted = abortSignal?.aborted === true && linkedSignal.signal.aborted diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index dee7c0fa4df..2af4c4bb301 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -298,6 +298,7 @@ export interface WebSearchLoopDeps { forceEmptyResponseId?: boolean; abortSignal?: AbortSignal; recordSidecarOutcome?: SidecarOutcomeRecorder; + beforeSidecarDispatch?: () => void; /** Cumulative per-iteration deadline for DNS/TCP/TLS and final response headers only. */ connectTimeoutMs?: number; /** Continuous routed-model response-body raw-byte inactivity deadline. Default 200000ms. */ @@ -761,7 +762,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { + previousHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + home = mkdtempSync(join(tmpdir(), "ocx-sidecar-credit-")); + process.env.OPENCODEX_HOME = home; + process.env.CODEX_HOME = home; + setIcaclsRunnerForTests(() => aclOk); + setAsyncIcaclsRunnerForTests(async () => aclOk); + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); clearPoolRotationState(); + clearAccountNeedsReauth(accountId); + resetVisionDescriptionCache(); + saveCodexAccountCredential(accountId, { + accessToken: "access-token-value-sidecar-credit", refreshToken: "fixture-sidecar-refresh", + expiresAt: Date.now() + 3_600_000, chatgptAccountId: "fixture-sidecar-workspace", + }); + globalThis.fetch = Object.assign(async () => { throw new Error("unexpected fixture fetch"); }, { preconnect() {} }) as typeof fetch; +}); + +afterEach(async () => { + globalThis.fetch = originalFetch; + clearAccountQuota(); clearCodexUpstreamHealth(); clearThreadAccountMap(); clearPoolRotationState(); + clearAccountNeedsReauth(accountId); + await flushConfigDirHardeningForTests(); + setIcaclsRunnerForTests(null); setAsyncIcaclsRunnerForTests(null); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; else process.env.CODEX_HOME = previousCodexHome; + removeTreeWithRetry(home); +}); + +describe("resolved sidecar credit policy at physical dispatch", () => { + for (const helper of ["vision", "search"] as const) { + test(`${helper} policy refusal stays local during caller abort`, async () => { + const cfg = config(); quota(100); + const sidecar = await resolve(cfg); + cfg.creditCodexAccountIds = []; + const controller = new AbortController(); + const recorded: unknown[] = []; + let sends = 0; + globalThis.fetch = Object.assign(async () => { sends++; return sse(); }, { preconnect() {} }) as typeof fetch; + const settings = { model: "gpt-5.5", reasoning: "low" as const, timeoutMs: 5_000 }; + const pending = helper === "vision" + ? describeImage("data:image/png;base64,iVBORw0KGgo=", "high", "inspect", sidecar.provider, + sidecar.headers, settings, controller.signal, value => recorded.push(value), sidecar.beforeDispatch) + : runWebSearch("fixture query", { type: "web_search" }, sidecar.provider, + sidecar.headers, settings, controller.signal, value => recorded.push(value), sidecar.beforeDispatch); + controller.abort(new Error("fixture caller cancellation")); + await pending; + expect(sends).toBe(0); + expect(recorded).toEqual([]); + expect(isAccountNeedsReauth(accountId)).toBe(false); + }); + + test.each(["quota", "consent", "allowed"])(`${helper} rechecks after resolution: %s`, async change => { + const cfg = config(); + quota(change === "quota" ? 99 : 100); + const sidecar = await resolve(cfg); + if (change === "quota") { cfg.creditCodexAccountIds = []; quota(100); } + if (change === "consent") cfg.creditCodexAccountIds = []; + let sends = 0; + const recorded: unknown[] = []; + globalThis.fetch = Object.assign(async () => { sends++; return sse(); }, { preconnect() {} }) as typeof fetch; + const settings = { model: "gpt-5.5", reasoning: "low" as const, timeoutMs: 5_000 }; + const beforeDispatch = sidecar.beforeDispatch; + const outcome = helper === "vision" + ? await describeImage("data:image/png;base64,iVBORw0KGgo=", "high", "inspect", sidecar.provider, + sidecar.headers, settings, undefined, value => recorded.push(value), beforeDispatch) + : await runWebSearch("fixture query", { type: "web_search" }, sidecar.provider, + sidecar.headers, settings, undefined, value => recorded.push(value), beforeDispatch); + expect(sends).toBe(change === "allowed" ? 1 : 0); + if (change === "allowed") { expect(outcome.error).toBeUndefined(); expect(recorded).toEqual([200]); } + else { expect(outcome.error).toContain("spending credits is off"); expect(recorded).toEqual([]); } + expect(isAccountNeedsReauth(accountId)).toBe(false); + expect(getCodexUpstreamHealth(accountId)).toBeNull(); + }); + } + + for (const helper of ["vision", "search"] as const) { + test(`${helper} reset retry rechecks consent without reporting a connection failure`, async () => { + const cfg = config(); quota(100); + const sidecar = await resolve(cfg); + let sends = 0; + const recorded: unknown[] = []; + globalThis.fetch = Object.assign(async () => { + sends++; + if (sends === 1) { + cfg.creditCodexAccountIds = []; + throw Object.assign(new Error("The socket connection was closed unexpectedly"), { code: "ECONNRESET" }); + } + return sse(); + }, { preconnect() {} }) as typeof fetch; + const settings = { model: "gpt-5.5", reasoning: "low" as const, timeoutMs: 5_000 }; + const outcome = helper === "vision" + ? await describeImage("data:image/png;base64,iVBORw0KGgo=", "high", "inspect", sidecar.provider, + sidecar.headers, settings, undefined, value => recorded.push(value), sidecar.beforeDispatch) + : await runWebSearch("fixture query", { type: "web_search" }, sidecar.provider, + sidecar.headers, settings, undefined, value => recorded.push(value), sidecar.beforeDispatch); + expect(sends).toBe(1); + expect(outcome.error).toContain("spending credits is off"); + expect(recorded).toEqual([]); + expect(isAccountNeedsReauth(accountId)).toBe(false); + }); + } + + test("search refuses the 429 replay after consent changes", async () => { + const cfg = config(); quota(100); + const sidecar = await resolve(cfg); + let sends = 0; + const recorded: unknown[] = []; + globalThis.fetch = Object.assign(async () => { + sends++; + cfg.creditCodexAccountIds = []; + return new Response("rate limited", { status: 429, headers: { "retry-after": "0.01" } }); + }, { preconnect() {} }) as typeof fetch; + const outcome = await runWebSearch("fixture query", { type: "web_search" }, sidecar.provider, sidecar.headers, + { model: "gpt-5.5", reasoning: "low", timeoutMs: 5_000 }, undefined, + value => recorded.push(value), sidecar.beforeDispatch); + expect(sends).toBe(1); + expect(outcome.error).toContain("spending credits is off"); + expect(recorded).toEqual([]); + }); + + test("production vision and passthrough search wiring retain the resolver guard", async () => { + const cfg = config(); quota(100); + const sidecar = await resolve(cfg); + cfg.creditCodexAccountIds = []; + let sends = 0; + globalThis.fetch = Object.assign(async () => { sends++; return sse(); }, { preconnect() {} }) as typeof fetch; + const parsed = { modelId: "fixture", stream: false, options: {}, context: { messages: [ + { role: "user" as const, content: [{ type: "image" as const, imageUrl: "data:image/png;base64,iVBORw0KGgo=" }] }, + ] } }; + await describeImagesInPlace(parsed, { backend: "openai", forwardSidecar: sidecar, + settings: { model: "gpt-5.5", reasoning: "low", timeoutMs: 5_000 }, maxDescriptionsPerTurn: 1 }, sidecar.headers); + expect(JSON.stringify(parsed.context.messages)).toContain("spending credits is off"); + const execute = createPassthroughWebSearchBridgeExecutor({ backend: "openai", maxSearches: 1, timeoutMs: 5_000 }, + { auth: { openAiSidecar: sidecar } }); + expect((await execute(["fixture query"])).error).toContain("spending credits is off"); + expect(sends).toBe(0); + expect(getCodexUpstreamHealth(accountId)).toBeNull(); + }); + + test("a late search relay hold returns reset-bound 429 without sending or quarantining", async () => { + const cfg = config(); quota(100); + const resolveOriginal = sidecarModule.resolveFirstUsableOpenAiSidecar; + const resolver = spyOn(sidecarModule, "resolveFirstUsableOpenAiSidecar").mockImplementation(async (...args) => { + const result = await resolveOriginal(...args); + cfg.creditCodexAccountIds = []; + return result; + }); + let sends = 0; + globalThis.fetch = Object.assign(async () => { sends++; return Response.json({ output: "fixture" }); }, { preconnect() {} }) as typeof fetch; + try { + const response = await handleSearch(new Request("http://localhost/v1/alpha/search", { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ query: "fixture" }), + }), cfg, { model: "", provider: "" }); + expect(response.status).toBe(429); + expect(Number(response.headers.get("retry-after"))).toBeGreaterThan(0); + expect(await response.text()).toContain("spending credits is off"); + expect(sends).toBe(0); + expect(isAccountNeedsReauth(accountId)).toBe(false); + expect(getCodexUpstreamHealth(accountId)).toBeNull(); + } finally { resolver.mockRestore(); } + }); + + test("caller-owned Direct auth is not governed by a stored account credit hold", async () => { + const cfg = config(); cfg.providers.openai!.codexAccountMode = "direct"; + cfg.creditCodexAccountIds = []; quota(100); + const token = fakeChatGptJwt({ chatgpt_account_id: "caller-workspace" }); + const sidecar = await resolveFirstUsableOpenAiSidecar(listOpenAiForwardSidecarCandidates(cfg), + new Headers({ authorization: `Bearer ${token}`, "chatgpt-account-id": "caller-workspace" }), cfg); + expect(sidecar?.authContext.kind).toBe("main"); + expect(sidecar?.beforeDispatch).toBeUndefined(); + let sends = 0; + globalThis.fetch = Object.assign(async () => { sends++; return sse(); }, { preconnect() {} }) as typeof fetch; + const outcome = await runWebSearch("fixture query", { type: "web_search" }, sidecar!.provider, sidecar!.headers, + { model: "gpt-5.5", reasoning: "low", timeoutMs: 5_000 }, undefined, undefined, sidecar!.beforeDispatch); + expect(outcome.error).toBeUndefined(); + expect(sends).toBe(1); + }); + + test("a dispatch refusal returns an unused quota probe lease", async () => { + const cfg = config(); quota(99); + const sidecar = await resolve(cfg); + const recordedAt = Date.now() - CODEX_QUOTA_PROBE_INTERVAL_MS - 1_000; + recordCodexUpstreamOutcome(cfg, accountId, 429, { + resetAt: Math.floor((recordedAt + 86_400_000) / 1_000), now: recordedAt, fixedAccount: true, + }); + const lease = tryAcquireCodexQuotaProbeLease(accountId); + expect(lease).toBeTruthy(); + sidecar.authContext.probeLeaseId = lease!; + quota(100); cfg.creditCodexAccountIds = []; + const outcome = await runWebSearch("fixture query", { type: "web_search" }, sidecar.provider, sidecar.headers, + { model: "gpt-5.5", reasoning: "low", timeoutMs: 5_000 }, undefined, undefined, sidecar.beforeDispatch); + expect(outcome.error).toContain("spending credits is off"); + expect(tryAcquireCodexQuotaProbeLease(accountId, Date.now() + CODEX_QUOTA_PROBE_INTERVAL_MS)).toBeTruthy(); + }); + + for (const transport of ["fetch", "runTurn"] as const) { + test.each(["blocked", "allowed"])(`${transport} search loop carries live policy: %s`, async policyState => { + const allowed = policyState === "allowed"; + const cfg = config(); quota(100); + const sidecar = await resolve(cfg); + if (!allowed) cfg.creditCodexAccountIds = []; + let searches = 0; + globalThis.fetch = Object.assign(async input => { + if (String(input).startsWith(sidecar.provider.baseUrl)) { searches++; return sse(); } + if (String(input) === "https://routed.example.test/completions") return Response.json({}); + throw new Error("unexpected fixture destination"); + }, { preconnect() {} }) as typeof fetch; + const parsed: OcxParsedRequest = { modelId: "fixture", stream: true, options: {}, + context: { messages: [{ role: "user", content: "search fixture" }], tools: [] } }; + const settings = { model: "gpt-5.5", reasoning: "low", timeoutMs: 5_000 }; + const searchEvents: AdapterEvent[] = [ + { type: "tool_call_start", id: "fixture-search", name: "web_search" }, + { type: "tool_call_delta", arguments: '{"query":"fixture query"}' }, + { type: "tool_call_end" }, { type: "done" }, + ]; + const answerEvents: AdapterEvent[] = [{ type: "text_delta", text: "fixture answer" }, { type: "done" }]; + async function* stream(events: AdapterEvent[]) { yield* events; } + const releaseSpendHome = acquireOwnedSpendHome(); + try { + if (transport === "runTurn") { + const output: AdapterEvent[] = []; + for await (const event of runTurnWebSearchLoop(stream(searchEvents), { + parsed, forwardProvider: sidecar.provider, forwardHeaders: sidecar.headers, + plan: { backend: "openai", forwardSidecar: sidecar, hostedTool: { type: "web_search" }, + settings, maxSearches: 1, routedModelStallTimeoutMs: 5_000, stallTimeoutSec: 5, streamRoutedModelOutput: false }, + dispatch: () => stream(answerEvents), + })) output.push(event); + expect(output.some(event => event.type === "web_search_call_end")).toBe(true); + } else { + let iteration = 0; + const adapter: ProviderAdapter = { name: "sidecar-credit-fixture", + buildRequest: () => ({ url: "https://routed.example.test/completions", method: "POST", headers: {}, body: "{}" }), + parseStream: () => stream(iteration++ === 0 ? searchEvents : answerEvents), + }; + const response = await runWithWebSearch({ parsed, adapter, + incomingMeta: { headers: new Headers(), translatorBudget: createTestTranslatorBudget() }, + forwardProvider: sidecar.provider, selectedForwardHeaders: sidecar.headers, + hostedTool: { type: "web_search" }, settings, maxSearches: 1, + beforeSidecarDispatch: sidecar.beforeDispatch, + }); + expect(await response.text()).toContain("response.completed"); + } + expect(searches).toBe(allowed ? 1 : 0); + expect(isAccountNeedsReauth(accountId)).toBe(false); + } finally { releaseSpendHome(); } + }); + } +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b555d52c979..ea8a382611c 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -75,6 +75,7 @@ "codex-credits-settings.test.ts": "codex-integration", "codex-credits-probes.test.ts": "codex-integration", "codex-credits-after-limit.test.ts": "codex-integration", "codex-pool-credit-policy.test.ts": "codex-integration", + "codex-sidecar-credit-policy.test.ts": "codex-integration", "codex-credits-after-limit-main.test.ts": "codex-integration", "service-desktop-startup-health.test.ts": "service", "service-desktop-startup.test.ts": "service", "startup-health-packaged-probe.test.ts": "server", "discovered-native-models.test.ts": "codex-integration", "responses-hosted-image-delivery.test.ts": "responses", From 9ea18a2f3ffe66660ea55e2db0f2b4d3174bd9df Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 6 Oct 2026 01:15:36 +0900 Subject: [PATCH 3/4] fix(codex): release probe leases on late credit-policy refusals Two policy refusals could leave a held pool probe behind: a credit hold that lands after header materialization in resolveResponsesCodexAuth (authCtx was scoped inside the try, so the catch could not release it), and a hold thrown by applyCodexAuthContextToProvider in prepareResponsesRequest, which had no mapper and escaped as a rejection instead of the actionable 429. Hoist the context and release it in the catch; wrap provider materialization, release, map to 429 (499 when cancelled), and rethrow unknown errors. Deterministic regressions in responses-pool-401-refresh cover both boundaries for quota and transient probes, cancellation, unknown errors, and the success path that keeps the probe for dispatch. --- src/server/responses/core-auth.ts | 3 +- src/server/responses/request-prepare.ts | 12 +- .../responses-pool-401-refresh.test.ts | 216 +++++++++++++++++- 3 files changed, 228 insertions(+), 3 deletions(-) diff --git a/src/server/responses/core-auth.ts b/src/server/responses/core-auth.ts index f7b5d26d03a..ffc951280d9 100644 --- a/src/server/responses/core-auth.ts +++ b/src/server/responses/core-auth.ts @@ -201,6 +201,7 @@ export async function resolveResponsesCodexAuth( credentialDomainWasRewritten = false, retainAccountForUploadedFiles = false, ): Promise { + let authCtx: CodexAuthContext | undefined; try { let authInputHeaders = codexRouteCredentialDomainHeaders( req, @@ -261,7 +262,6 @@ export async function resolveResponsesCodexAuth( if (route.codexAccountMode === "direct" && !substituteMainCredential) { validateForwardAdmissionCredential(authInputHeaders, config); } - let authCtx: CodexAuthContext; if (route.codexAccountMode) { authCtx = await resolveCodexAuthContext(authInputHeaders, config, route.codexAccountMode, { admission: options.admission, @@ -343,6 +343,7 @@ export async function resolveResponsesCodexAuth( substituteMainCredential, }; } catch (err) { + releaseCodexAuthContextProbeLease(authCtx); if (options.abortSignal?.aborted || req.signal.aborted) { return { ok: false, response: clientCancelledResponse() }; } diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 2f251c26b74..df7e5e39c6e 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -46,6 +46,7 @@ import { codexPoolAffinityKey, previewCodexPoolLineage, applyCodexAuthContextToProvider, + releaseCodexAuthContextProbeLease, hasCallerCodexBearer, requestOwnedMainPinState, requestOwnedMainCredentialIsLive, @@ -58,6 +59,7 @@ import { previousResponseProviderState, } from "../../responses/state"; import { formatErrorResponse } from "../../bridge"; +import { mapCodexAuthContextErrorToResponse } from "./codex-auth-error"; import type { OcxParsedRequest } from "../../types"; import { buildToolBridgeMaps } from "./collaboration"; import { parseRequest } from "../../responses/parser"; @@ -1352,7 +1354,15 @@ export async function prepareResponsesRequest( substituteMainCredential = finalAuth.substituteMainCredential; } - route.provider = applyCodexAuthContextToProvider(route.provider, admissionState.authCtx, route.codexAccountMode); + try { + route.provider = applyCodexAuthContextToProvider(route.provider, admissionState.authCtx, route.codexAccountMode); + } catch (error) { + releaseCodexAuthContextProbeLease(admissionState.authCtx); + if (options.abortSignal?.aborted || req.signal.aborted) return clientCancelledResponse(); + const mapped = mapCodexAuthContextErrorToResponse(error, { now: Date.now(), accountSelector: route.codexAccountNamespace }); + if (mapped) return mapped; + throw error; + } applyCodexAccountGatedWireNormalization(parsed, route, logCtx); try { recordPolicyPreparedDestination(policyScope, route.providerName, parsed._wireModelOverride ?? parsed.modelId); diff --git a/tests/responses/responses-pool-401-refresh.test.ts b/tests/responses/responses-pool-401-refresh.test.ts index a538b5e1168..2fe6e561b02 100644 --- a/tests/responses/responses-pool-401-refresh.test.ts +++ b/tests/responses/responses-pool-401-refresh.test.ts @@ -1,17 +1,34 @@ -import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { existsSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { createHash } from "node:crypto"; import { clearAccountNeedsReauth, clearAccountQuota, isAccountNeedsReauth, updateAccountQuota } from "../../src/codex/auth-api"; import { codexPoolAffinityKey } from "../../src/codex/auth-context"; +import * as authContextModule from "../../src/codex/auth-context"; +import * as coreAuthModule from "../../src/server/responses/core-auth"; +import { prepareResponsesRequest } from "../../src/server/responses/request-prepare"; +import { handleComboResponses } from "../../src/server/responses/core"; +import type { ResponsesAdmissionState } from "../../src/server/responses/core-options"; +import { createTranslatorBudget } from "../../src/lib/translator-budget"; +import { + canAcquireTransientProbe, + clearPoolRecoveryState, + transientProbeDiagnostics, + TRANSIENT_PROBE_INTERVAL_MS, +} from "../../src/routing/probe-lease"; import { setAccountQuotaFromParsed } from "../../src/codex/quota"; import { clearCodexUpstreamHealth, clearThreadAccountMap, + CODEX_QUOTA_PROBE_INTERVAL_MS, peekConversationStateIssuer, + recordCodexUpstreamOutcome, + releaseCodexQuotaScopeProbeLease, resolveCodexAccountForThreadDetailed, + tryAcquireCodexQuotaScopeProbeLease, } from "../../src/codex/routing"; +import { scopedHealthFor } from "../../src/codex/routing/health-store"; import { handleResponses, handleResponsesCompact } from "../../src/server/responses"; import { clearCompactHandoffRoutesForTests } from "../../src/server/responses/compact"; import { @@ -345,6 +362,203 @@ afterEach(() => { removeTreeWithRetry(home); }); +describe("stored-account credit refusal before Responses dispatch (#6606)", () => { + let now: number; + let clock: ReturnType; + + beforeEach(() => { + now = Date.now(); + clock = spyOn(Date, "now").mockReturnValue(now); + clearPoolRecoveryState(); + }); + afterEach(() => { + clock.mockRestore(); + clearPoolRecoveryState(); + }); + + function probeFixture(kind: "quota" | "transient") { + const cfg = config(); + cfg.creditCodexAccountIds = []; + cfg.upstreamHostCircuitThreshold = 0; + cfg.upstreamFailoverThreshold = 3; + updateAccountQuota(ACCOUNT_ID, 99, now + 3_600_000); + const req = request("/v1/responses", { affined: true }); + if (kind === "quota") { + recordCodexUpstreamOutcome(cfg, ACCOUNT_ID, 429, { + now: now - CODEX_QUOTA_PROBE_INTERVAL_MS - 1_000, + resetAt: Math.floor((now + 4 * 24 * 60 * 60_000) / 1_000), + modelId: "gpt-5.5", + fixedAccount: true, + }); + } else { + const key = codexPoolAffinityKey(req.headers)!; + expect(resolveCodexAccountForThreadDetailed(key, cfg, now, "shared")) + .toMatchObject({ status: "selected", accountId: ACCOUNT_ID }); + for (let attempt = 0; attempt < 3; attempt++) { + recordCodexUpstreamOutcome(cfg, ACCOUNT_ID, 503, { now, modelId: "gpt-5.5" }); + } + } + return { cfg, req }; + } + + function expectProbeHeld(ctx: authContextModule.CodexAuthContext, kind: "quota" | "transient") { + expect(ctx.kind).toBe("pool"); + if (ctx.kind !== "pool") throw new Error("expected a stored pool probe"); + if (kind === "quota") { + expect(ctx.probeLeaseId).toBeDefined(); + expect(ctx.probeQuotaScope).toBe("shared"); + expect(scopedHealthFor(ACCOUNT_ID, "shared")?.probeLeaseId).toBe(ctx.probeLeaseId); + } else { + expect(ctx.transientProbe).toBeDefined(); + expect(transientProbeDiagnostics(ACCOUNT_ID, now)).toMatchObject({ + held: true, leaseId: ctx.transientProbe!.lease.leaseId, + }); + expect(canAcquireTransientProbe(ACCOUNT_ID, now + TRANSIENT_PROBE_INTERVAL_MS)).toBe(false); + } + } + + function expectProbeReleased(kind: "quota" | "transient") { + if (kind === "quota") { + expect(scopedHealthFor(ACCOUNT_ID, "shared")?.probeLeaseId).toBeUndefined(); + const lease = tryAcquireCodexQuotaScopeProbeLease(ACCOUNT_ID, "shared", now + CODEX_QUOTA_PROBE_INTERVAL_MS); + expect(lease).toBeTruthy(); + if (lease) releaseCodexQuotaScopeProbeLease(ACCOUNT_ID, "shared", lease, now); + } else { + expect(transientProbeDiagnostics(ACCOUNT_ID, now).held).toBe(false); + expect(canAcquireTransientProbe(ACCOUNT_ID, now + TRANSIENT_PROBE_INTERVAL_MS)).toBe(true); + } + } + + async function prepareProbe(req: Request, cfg: OcxConfig, state: ResponsesAdmissionState, abortSignal?: AbortSignal) { + const translatorBudget = createTranslatorBudget(); + try { + return await prepareResponsesRequest({ + req, config: cfg, logCtx: { model: "", provider: "" } as RequestLogContext, + options: { translatorBudget, abortSignal }, + }, state, { handleResponses, handleComboResponses }); + } finally { + translatorBudget.dispose(); + } + } + + // D1: change policy only after real materialization has acquired and used the quota probe. + for (const cancelled of [false, true]) { + test(`late materialization credit hold releases its quota probe (${cancelled ? "499" : "429"})`, async () => { + const { cfg, req } = probeFixture("quota"); + const harness = installHarness(); + const abort = new AbortController(); + const materialize = authContextModule.materializeCodexUpstreamAuthAsync; + const materializeSpy = spyOn(authContextModule, "materializeCodexUpstreamAuthAsync") + .mockImplementation(async (...args) => { + const headers = await materialize(...args); + expectProbeHeld(args[1], "quota"); + updateAccountQuota(ACCOUNT_ID, 100, now + 3_600_000); + if (cancelled) abort.abort(); + return headers; + }); + try { + const response = await handleResponses(req, cfg, { model: "", provider: "" } as RequestLogContext, + { abortSignal: abort.signal }); + expect(materializeSpy).toHaveBeenCalledTimes(1); + expect(response.status).toBe(cancelled ? 499 : 429); + if (cancelled) expect(await response.json()).toMatchObject({ error: { code: "client_cancelled" } }); + else { + expect(Number(response.headers.get("Retry-After"))).toBeGreaterThan(0); + expect(await response.text()).toContain("spending credits"); + } + expect(harness.sends).toEqual([]); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expect(readStoredGeneration()).toBe(3); + expectProbeReleased("quota"); + } finally { materializeSpy.mockRestore(); } + }); + } + + // D2: auth resolution admits either kind of probe; a quota probe and transient trial never coexist. + for (const cancelled of [false, true]) { + test.each(["quota", "transient"] as const)( + `provider credit hold returns ${cancelled ? "499" : "429"} and releases the %s probe with host circuit disabled`, + async kind => { + const { cfg, req } = probeFixture(kind); + const harness = installHarness(); + const abort = new AbortController(); + const state: ResponsesAdmissionState = { pendingHostAdmissionLease: null, authCtx: { kind: "main", accountId: null } }; + const resolveAuth = coreAuthModule.resolveResponsesCodexAuth; + const authSpy = spyOn(coreAuthModule, "resolveResponsesCodexAuth").mockImplementation(async (...args) => { + const result = await resolveAuth(...args); + expect(result.ok).toBe(true); + if (result.ok) { + expectProbeHeld(result.authCtx, kind); + updateAccountQuota(ACCOUNT_ID, 100, now + 3_600_000); + if (cancelled) abort.abort(); + } + return result; + }); + try { + const result = await prepareProbe(req, cfg, state, abort.signal); + expect(authSpy).toHaveBeenCalledTimes(1); + expect(result).toBeInstanceOf(Response); + if (!(result instanceof Response)) throw new Error("expected a pre-dispatch refusal"); + expect(result.status).toBe(cancelled ? 499 : 429); + if (cancelled) expect(await result.json()).toMatchObject({ error: { code: "client_cancelled" } }); + else { + expect(Number(result.headers.get("Retry-After"))).toBeGreaterThan(0); + expect(await result.text()).toContain("spending credits"); + } + expect(state.pendingHostAdmissionLease).toBeNull(); + expect(harness.sends).toEqual([]); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expectProbeReleased(kind); + } finally { + authSpy.mockRestore(); + authContextModule.releaseCodexAuthContextProbeLease(state.authCtx); + } + }, + ); + } + + test.each(["quota", "transient"] as const)("unknown provider materialization error propagates after releasing the %s probe", async kind => { + const { cfg, req } = probeFixture(kind); + const harness = installHarness(); + const state: ResponsesAdmissionState = { pendingHostAdmissionLease: null, authCtx: { kind: "main", accountId: null } }; + const error = new Error("fixture provider materialization failure"); + const applySpy = spyOn(authContextModule, "applyCodexAuthContextToProvider").mockImplementation((_provider, ctx) => { + expectProbeHeld(ctx, kind); + throw error; + }); + try { + await expect(prepareProbe(req, cfg, state)).rejects.toBe(error); + expect(applySpy).toHaveBeenCalledTimes(1); + expect(state.pendingHostAdmissionLease).toBeNull(); + expect(harness.sends).toEqual([]); + expect(harness.refreshes).toEqual([]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expectProbeReleased(kind); + } finally { + applySpy.mockRestore(); + authContextModule.releaseCodexAuthContextProbeLease(state.authCtx); + } + }); + + test.each(["quota", "transient"] as const)("successful preparation retains the %s probe for downstream dispatch", async kind => { + const { cfg, req } = probeFixture(kind); + const harness = installHarness(); + const state: ResponsesAdmissionState = { pendingHostAdmissionLease: null, authCtx: { kind: "main", accountId: null } }; + try { + const result = await prepareProbe(req, cfg, state); + expect(result).not.toBeInstanceOf(Response); + if (result instanceof Response) throw new Error("expected successful preparation"); + expect(result.selectedForwardHeaders.get("authorization")).toBe("Bearer rejected-access"); + expectProbeHeld(state.authCtx, kind); + expect(state.pendingHostAdmissionLease).toBeNull(); + expect(harness.sends).toEqual([]); + expect(harness.refreshes).toEqual([]); + } finally { authContextModule.releaseCodexAuthContextProbeLease(state.authCtx); } + }); +}); + describe("ordinary pool 401 refresh and replay (#2887)", () => { test.each([false, true])("failed terminal cannot echo the selected pool credential or retain it for replay (stream:%s)", async stream => { const harness = installHarness({ From 4fa86eca9ee8ccbcfb8e5cee16177daecd32e61f Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 6 Oct 2026 01:30:18 +0900 Subject: [PATCH 4/4] fix(codex): let cancellation win over post-refresh credit refusals After a successful pool 401 refresh, a credit hold raised by the refreshed materialization was mapped to a 429 even when the client had already aborted. Check the request and effective signals first in both the Responses and compact refresh catches and return 499 without quarantine. Adds deterministic cancellation cases for request and effective signals on /v1/responses and the compact path; the non-aborted case now also asserts Retry-After. --- src/server/responses/compact.ts | 4 ++ src/server/responses/core-auth.ts | 3 ++ .../responses-pool-401-refresh.test.ts | 47 +++++++++++++++++++ 3 files changed, 54 insertions(+) diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 4d2577399f4..592c7583093 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -183,6 +183,7 @@ import { hasResponsesItemIdRepair, relaySseWithResponsesItemIdRepair } from "../ import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; import { codexAuthContextLogLabel } from "../../codex/account-label"; import { rebindPoolCreditPolicy } from "../../codex/pool-credit-policy"; +import { clientCancelledResponse } from "./core-errors"; import { codexAccountGatedCanonicalWireModel, @@ -467,6 +468,9 @@ async function refreshPoolCompactContext(args: { if (isTerminalCompactPoolRefreshFailure(error)) { return { ok: false, quarantine: true, response: reauthResponse() }; } + if (req.signal.aborted) { + return { ok: false, quarantine: false, response: clientCancelledResponse() }; + } // See the core counterpart: a policy refusal raised after a successful refresh maps like // the admission path — a reset-bound 429, never a 503 mislabeled as a refresh failure. const policyResponse = mapCodexAuthContextErrorToResponse(error, { diff --git a/src/server/responses/core-auth.ts b/src/server/responses/core-auth.ts index ffc951280d9..2f0833a9fcd 100644 --- a/src/server/responses/core-auth.ts +++ b/src/server/responses/core-auth.ts @@ -507,6 +507,9 @@ export async function refreshPoolForwardAuth(args: { response: formatErrorResponse(401, "authentication_error", "Selected Codex account needs reauthentication"), }; } + if (options.abortSignal?.aborted || req.signal.aborted) { + return { ok: false, quarantine: false, response: clientCancelledResponse() }; + } // The credential itself refreshed; a policy refusal raised afterwards (a credit hold or an // opt-out that landed during the await) is not a refresh failure. Map it the same way the // admission path does — a reset-bound 429, never a 503 telling the caller to sign in — and diff --git a/tests/responses/responses-pool-401-refresh.test.ts b/tests/responses/responses-pool-401-refresh.test.ts index 2fe6e561b02..17da8c768be 100644 --- a/tests/responses/responses-pool-401-refresh.test.ts +++ b/tests/responses/responses-pool-401-refresh.test.ts @@ -700,6 +700,7 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { // A policy refusal is not an incomplete refresh: the client gets the actionable 429, // the rotated grant stays stored, and nothing replays on the refreshed bearer. expect(response.status).toBe(429); + expect(Number(response.headers.get("Retry-After"))).toBeGreaterThan(0); expect(body).toContain("spending credits"); expect(body).not.toContain("credential refresh did not complete"); expect(harness.sends).toEqual(["Bearer rejected-access"]); @@ -709,6 +710,52 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { }, ); + for (const [path, signalSource] of [ + ["/v1/responses", "request"], + ["/v1/responses", "effective"], + ["/v1/responses/compact", "request"], + ] as const) { + test(`post-refresh credit refusal yields to cancellation (${path}, ${signalSource})`, async () => { + const client = new AbortController(); + const cfg = config(); + cfg.creditCodexAccountIds = []; + const harness = installHarness(); + const materialize = authContextModule.materializeCodexUpstreamAuthAsync; + const materializeSpy = spyOn(authContextModule, "materializeCodexUpstreamAuthAsync") + .mockImplementation(async (...args) => { + if (args[1].kind !== "pool" || args[1].accessToken !== "refreshed-access") { + return materialize(...args); + } + // Refresh has settled; abort when the real materializer rejects the new hold. + expect(readStoredGeneration()).toBe(4); + updateAccountQuota(ACCOUNT_ID, 100, Date.now() + 3_600_000); + try { + return await materialize(...args); + } catch (error) { + client.abort(); + throw error; + } + }); + try { + const req = signalSource === "request" + ? new Request(request(path), { signal: client.signal }) + : request(path); + const response = path === "/v1/responses/compact" + ? await handleResponsesCompact(req, cfg, { model: "", provider: "" } as RequestLogContext) + : await handleResponses(req, cfg, { model: "", provider: "" } as RequestLogContext, + signalSource === "effective" ? { abortSignal: client.signal } : {}); + expect(response.status).toBe(499); + expect(await response.json()).toMatchObject({ error: { code: "client_cancelled" } }); + expect(harness.sends).toEqual(["Bearer rejected-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + expect(readStoredGeneration()).toBe(4); + } finally { + materializeSpy.mockRestore(); + } + }); + } + test.each(["/v1/responses", "/v1/responses/compact"] as const)( "a credit hold while the initial request waits for pacing refuses without sending (%s)", async path => { const cfg = config();