diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 2e31c8b2431..a52e63d77ea 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -892,6 +892,7 @@ "kiro-retry.test.ts": "providers/kiro", "kiro-review-regressions.test.ts": "providers/kiro", "kiro-stream.test.ts": "providers/kiro", + "kiro-fallback-error-body.test.ts": "providers/kiro", "kiro-usage-quota.test.ts": "providers/kiro", "kiro-windows-cli-db-path.test.ts": "providers/kiro", "kiro-windows-cli-executable-path.test.ts": "providers/kiro", @@ -1280,6 +1281,7 @@ "responses-account-label.test.ts": "responses", "responses-canonical-only-top-level-fields.test.ts": "responses", "responses-compact-handoff-admission.test.ts": "responses", + "responses-compaction-policy-identity.test.ts": "responses", "responses-compaction-override.test.ts": "responses", "responses-compaction-routing.test.ts": "responses", "responses-compaction.test.ts": "responses", @@ -1363,6 +1365,7 @@ "self-launch-argv.test.ts": "lib", "server-403-permission-e2e.test.ts": "server", "server-agent-task-recovery-replay.test.ts": "server", + "server-auth-scoped-quota.test.ts": "server", "server-auth.test.ts": "server", "server-background-lifecycle.test.ts": "server", "server-clickjacking-headers.test.ts": "server", diff --git a/src/adapters/kiro/adapter.ts b/src/adapters/kiro/adapter.ts index b6a374cf4cc..0300c52907f 100644 --- a/src/adapters/kiro/adapter.ts +++ b/src/adapters/kiro/adapter.ts @@ -250,6 +250,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter }); return { response, + abortSignal: requestAbortSignal, inputTokens: retry.inputTokens, contextInputEstimate: retry.contextInputEstimate, nameMap: retry.nameMap, diff --git a/src/adapters/kiro/stream.ts b/src/adapters/kiro/stream.ts index d10ab1105c0..1e89d420f7c 100644 --- a/src/adapters/kiro/stream.ts +++ b/src/adapters/kiro/stream.ts @@ -19,6 +19,7 @@ import { noteKiroTransientThrottle } from "../kiro-retry"; import { KiroThinkingParser } from "../kiro-thinking"; import { isCompleteKiroToolInput, kiroTruncationErrorMessage } from "../kiro-truncation"; import { isValidKiroConversationId } from "../kiro-wire"; +import { readDisplaySafeErrorPayloadText } from "../upstream-http-error"; import { tagKiroReasoningBlob } from "./reasoning"; import { estimateKiroTokens, kiroUpstreamContextWindow } from "./usage"; @@ -74,6 +75,7 @@ function createKiroAttemptRetention(budget: TranslatorBudget): KiroAttemptRetent interface KiroFallbackAttempt { response: Response; + abortSignal?: AbortSignal; inputTokens: number; contextInputEstimate: number; nameMap: Map; @@ -1092,7 +1094,7 @@ export async function* parseKiroStream( firstResult.releaseRetained(); fallback.releaseRequestBody?.(); if (!fallback.response.ok) { - const payload = await fallback.response.text().catch(() => ""); + const payload = await readDisplaySafeErrorPayloadText(fallback.response, fallback.abortSignal); const failure = classifyKiroHttpError(fallback.response.status, fallback.response.headers, payload); yield { type: "error", diff --git a/src/codex/auth-context.ts b/src/codex/auth-context.ts index db100fd6c26..206036cbbe1 100644 --- a/src/codex/auth-context.ts +++ b/src/codex/auth-context.ts @@ -598,6 +598,20 @@ function selectedCodexToken(headers: Headers): { accessToken: string; chatgptAcc }; } +/** + * The workspace account id a request-owned `main` credential materializes under, or + * `undefined` when the caller's headers carry none. This is the `chatgpt-account-id` + * `materializeCodexUpstreamAuth` would set for a caller-owned `{ kind: "main" }` context, + * read here without touching a credential store so a rotation gate can compare workspace + * scope before a send is ever built. + */ +export function callerCodexWorkspaceAccountId(headers: Headers): string | undefined { + const explicit = headers.get("chatgpt-account-id"); + if (explicit) return explicit; + const bearer = headers.get("authorization")?.replace(/^Bearer\s+/i, "").trim(); + return bearer ? extractAccountId(undefined, bearer) : undefined; +} + function assertMaterializedReserve(headers: Headers, ctx: CodexAuthContext, options: CodexAuthMaterializationOptions): void { if (!requiresReserveAuthorization(options.config, options.modelId, options.admission)) return; assertReserveAdmission(options.config!); diff --git a/src/codex/quota-rejection.ts b/src/codex/quota-rejection.ts index cde0a8272d1..d4559c1187c 100644 --- a/src/codex/quota-rejection.ts +++ b/src/codex/quota-rejection.ts @@ -349,9 +349,10 @@ export async function codexScopedExhaustionCode( * Status alone and message text are intentionally insufficient. The broad * alternate-account retry remains eligible for 429/402 to preserve #584. * - * The one carve-out from that breadth is an organization- or project-scoped exhaustion - * ({@link SCOPED_EXHAUSTION_CODE_VALUES}), which reports `alternateRetryEligible: false` - * because every credential inside the refusing limit would be refused by the same counter. + * Organization- or project-scoped exhaustion ({@link SCOPED_EXHAUSTION_CODE_VALUES}) remains + * alternate-retry eligible here because the response does not identify the refusing scope. The + * account-rotation path may suppress the send later when the resolved alternate carries binding + * evidence that it shares an organization-level counter. */ export async function classifyCodexPreStreamRejection( response: Response, @@ -377,7 +378,10 @@ export async function classifyCodexPreStreamRejection( }); } if (scoped) { - return rejection(status, "scoped-quota-exhaustion", { scopedExhaustionCode: scoped }); + return rejection(status, "scoped-quota-exhaustion", { + alternateRetryEligible: true, + scopedExhaustionCode: scoped, + }); } return rejection( status, diff --git a/src/lib/bounded-body.ts b/src/lib/bounded-body.ts index 0b3769eaacf..effb247fb71 100644 --- a/src/lib/bounded-body.ts +++ b/src/lib/bounded-body.ts @@ -15,6 +15,8 @@ export interface BoundedBodyOptions { * Reader cancellation and lock release still run. Defaults to false. */ fatalUtf8?: boolean; + /** Report UTF-8 validity without rejecting malformed bodies. */ + reportUtf8Validity?: boolean; /** * Byte ceiling for retained body data. Defaults to BOUNDED_BODY_MAX_BYTES (64 KiB), * which suits error bodies; callers materializing whole success payloads (e.g. a @@ -44,6 +46,8 @@ export interface BoundedBodyResult { oversized: boolean; /** False means callers should use a status-only fallback, not `text`. */ displaySafe: boolean; + /** Present when reportUtf8Validity was requested and the retained body reached EOF. */ + utf8Valid?: boolean; } export interface BoundedBytesOptions { @@ -238,6 +242,14 @@ function decodeUtf8(chunks: readonly Uint8Array[], fatal: boolean, timedOut = fa } } +function decodeUtf8WithValidity(bytes: Uint8Array): { text: string; utf8Valid: boolean } { + try { + return { text: decodeUtf8([bytes], true), utf8Valid: true }; + } catch { + return { text: decodeUtf8([bytes], false), utf8Valid: false }; + } +} + /** * Consume the original response body under strict memory and time bounds. * @@ -326,6 +338,24 @@ export async function readBoundedResponseBody( const { value, done } = outcome as ReadableStreamReadResult; if (done) { + if (options.reportUtf8Validity) { + const bytes = retained.subarray(0, retainedBytes); + // A fatal decode that returned already proved the bytes valid; still + // honour the reporting contract instead of dropping utf8Valid. + const decoded = options.fatalUtf8 === true + ? { text: decodeUtf8([bytes], true), utf8Valid: true } + : decodeUtf8WithValidity(bytes); + return { + text: decoded.text, + truncated: false, + timedOut: false, + totalTimedOut: false, + inactivityTimedOut: false, + oversized: false, + displaySafe: true, + utf8Valid: decoded.utf8Valid, + }; + } return { text: decodeUtf8([retained.subarray(0, retainedBytes)], options.fatalUtf8 === true), truncated: false, diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 89e40123789..195c26230e5 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -51,6 +51,7 @@ import { describeImagesInPlace, planVisionSidecar, shouldResolveOpenAiVisionSide import { createAdapterEventQueue, preflightAdapterEvents } from "../../adapters/run-turn-queue"; import { applyCodexAuthContextToProvider, + callerCodexWorkspaceAccountId, createCodexReserveDispatchGuard, unwrapUpstreamRetryEvidenceError, CodexMainProfileDrainingError, @@ -186,6 +187,7 @@ import { handleResponses, preAuthUpstreamHostCircuitKey, poolCredentialRefreshIncompleteResponse, + shouldRetryCodexScopedQuotaOnAlternate, upstreamHostCircuitOpenResponse, usesCodexForwardPoolAuth, } from "./core"; @@ -1211,9 +1213,36 @@ export async function handleResponsesCompact( if (alternate && req.signal.aborted) { releaseCodexAuthContextProbeLease(alternate.authCtx); recordCompactPoolOutcome(outcomeCtx, 499); + void upstream.body?.cancel(req.signal.reason).catch(() => undefined); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } - if (alternate) { + // The same scope binding the regular path applies: an organization-scoped + // exhaustion refuses every credential in that workspace, so a proven + // same-workspace alternate pays a cold prompt prefix for no new capacity. + // Suppression is not silence — the buffered recorder below still attributes + // the 429/402 to the account that produced it. + const sharedWorkspaceScope = alternate != null + && !await shouldRetryCodexScopedQuotaOnAlternate( + upstream, + authCtx.chatgptAccountId, + alternate.authCtx.kind === "pool" || alternate.authCtx.kind === "main-pool" + ? alternate.authCtx.chatgptAccountId + : callerCodexWorkspaceAccountId(req.headers), + req.signal, + ); + // The scope check reads the rejection body asynchronously — the same window the + // comment above covers. Re-check before the branch below records A, cancels its + // body, and sends B for a caller that is gone. + if (alternate && req.signal.aborted) { + releaseCodexAuthContextProbeLease(alternate.authCtx); + recordCompactPoolOutcome(outcomeCtx, 499); + void upstream.body?.cancel(req.signal.reason).catch(() => undefined); + return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + } + if (alternate && sharedWorkspaceScope) { + releaseCodexAuthContextProbeLease(alternate.authCtx); + } + if (alternate && !sharedWorkspaceScope) { // Same order the regular path uses (core.ts:349-357): a 429/402 carries the // quota snapshot that produced it, so refresh A's cache before recording its // rejection. Skipping this leaves quota-strategy routing and the dashboard diff --git a/src/server/responses/compaction-routing.ts b/src/server/responses/compaction-routing.ts index a7d50962db5..4660d636a51 100644 --- a/src/server/responses/compaction-routing.ts +++ b/src/server/responses/compaction-routing.ts @@ -3,6 +3,8 @@ import { isDeclaredReasoningEffort } from "../../reasoning-effort"; import { COMPACTION_TRIGGERS } from "../../config/schema/compaction-triggers"; import { routeConcreteModel, type RouteResult } from "../../router"; import { resolveComboId } from "../../combos/identifiers"; +import { resolvePolicyProfileId } from "../../routing/profile"; +import { parseSyntheticRowId } from "../fast-row"; import { recallComboForLane } from "./combo-session-recall"; import { sessionLaneIdFromRequest } from "../request-log-conversation"; @@ -83,7 +85,7 @@ export function applyCompactionRoutingOverride( if (trigger === undefined) return null; const sourceModel = raw.model; - const sourceCombo = recallComboForLane(config, sessionLaneIdFromRequest(headers), sourceModel); + const sourceCombo = recallComboForLane(config, sessionLaneIdFromRequest(headers), sourceSelectorOf(config, sourceModel)); const targetCombo = resolveComboId(config, override.model.trim()) ?? undefined; raw.model = override.model.trim(); if (override.reasoningEffort !== undefined) { @@ -92,19 +94,42 @@ export function applyCompactionRoutingOverride( return { sourceModel, ...(sourceCombo ? { sourceCombo } : {}), ...(targetCombo ? { targetCombo } : {}) }; } +/** + * The selector a synthetic-row grammar actually routed on. `--fast`/`--effort` suffixes are + * decoration applied at ingress; identity checks must see the base id or a decorated + * virtual selector (`alias--fast`) slips past them. + */ +function sourceSelectorOf(config: OcxConfig, sourceModel: string): string { + const { fastRow, effortRow } = parseSyntheticRowId(sourceModel, config); + return fastRow?.baseId ?? effortRow?.baseId ?? sourceModel; +} + /** Same provider identity keeps caller auth and may use native compact; its ciphertext replays only there. */ export function compactionRoutingKeepsProviderIdentity( config: OcxConfig, override: CompactionRoutingOverride, route: RouteResult, ): boolean { - if (route.combo || override.sourceCombo || override.targetCombo || resolveComboId(config, override.sourceModel)) return false; + // `sourceModel` is the selector as the client sent it, so a synthetic `--fast` or + // effort suffix can still be attached. The base id is what the conversation routed on, + // and only the base can match the combo/policy guards below. + const sourceSelector = sourceSelectorOf(config, override.sourceModel); + if (route.combo || override.sourceCombo || override.targetCombo || resolveComboId(config, sourceSelector)) return false; + // A policy selector does not identify one stable serving backend: its route depends on + // request evidence and live candidate state that this post-rewrite check no longer has. + // Treat it as crossing identity rather than reconstructing it through concrete routing, + // which deliberately bypasses policy evaluation and may fall through to defaultProvider. + if (resolvePolicyProfileId(config, sourceSelector) !== null) return false; let source: RouteResult; try { - source = routeConcreteModel(config, override.sourceModel); + source = routeConcreteModel(config, sourceSelector); } catch { return false; } + // The default-provider branch is where every unrecognized selector lands — including a + // policy/combo alias that was renamed or deleted since the conversation began. Such a + // selector cannot prove which backend served it, so it can never match an identity. + if (source.routeReason === "default-provider") return false; return source.providerName === route.providerName && source.codexAccountMode === route.codexAccountMode && source.codexAccountNamespace === route.codexAccountNamespace; diff --git a/src/server/responses/core-codex-account.ts b/src/server/responses/core-codex-account.ts index 0645dda4cd3..7ce9ec43889 100644 --- a/src/server/responses/core-codex-account.ts +++ b/src/server/responses/core-codex-account.ts @@ -41,6 +41,7 @@ import { isNativeMainTrafficBlocked } from "../../codex/native-profile-startup"; import { MAIN_CODEX_ACCOUNT_ID } from "../../codex/main-account"; import { slugsEquivalent } from "../../providers/slug-codec"; import { + callerCodexWorkspaceAccountId, codexProbeLeaseId, codexTransientProbeGrant, codexProbeQuotaScope, @@ -280,15 +281,11 @@ export async function shouldRetryCodexPoolAccountQuota( // body carries no quota evidence either, but the marker is the contract, not the prose. if (isNonReplayableResponse(response)) return false; if (response.status === 402 || response.status === 429) { - // Status alone used to authorize the move, which is right for a limit the ACCOUNT owns and - // wrong for one it merely belongs to. An organization- or project-scoped exhaustion refuses - // every credential inside that organization, so the second account meets the same counter - // and the only thing the rotation buys is a second cold prompt prefix (#4546). Positive - // evidence is required to withhold it: the helper fails closed, so an unreadable or - // ambiguous body keeps the broad #584 behaviour unchanged, and `rate_limit_exceeded`, - // `slow_down` and plan-level exhaustion still rotate exactly as before. - const { codexScopedExhaustionCode } = await import("../../codex/quota-rejection"); - return await codexScopedExhaustionCode(response, { signal }) === undefined; + // The response does not identify the organization or project whose quota was exhausted. + // Resolve the alternate before deciding whether its known workspace identity proves that an + // organization-scoped retry would be futile. Until then, preserve the broad #584 behaviour. + void signal; + return true; } if (response.status < 500 || response.status >= 600) return false; try { @@ -305,6 +302,20 @@ export async function shouldRetryCodexPoolAccountQuota( } +export async function shouldRetryCodexScopedQuotaOnAlternate( + response: Response, + firstWorkspaceAccountId: string, + alternateWorkspaceAccountId: string | undefined, + signal?: AbortSignal, +): Promise { + if (!firstWorkspaceAccountId || firstWorkspaceAccountId !== alternateWorkspaceAccountId) return true; + const { codexScopedExhaustionCode } = await import("../../codex/quota-rejection"); + const code = await codexScopedExhaustionCode(response, { signal }); + // Workspace identity binds organization-level limits, but the response supplies no project id. + return code === undefined || code === "project_spend_limit_exceeded"; +} + + /** * A pre-stream upstream 5xx another Codex account may still be able to serve. * @@ -532,6 +543,22 @@ export async function retryCodexPoolOnAlternateAccount( writerGeneration: firstAuthCtx.writerGeneration, }); }; + // A body-confirmed quota response may arrive under HTTP 5xx. A path that returns the + // first response without a move must still record the NORMALIZED outcome: the ordinary + // terminal recorder sees only that wire status and would misclassify it as transient, + // leaving the exhausted account immediately selectable next turn. + const recordWrappedQuotaOutcome = (): void => { + if (outcomeStatus === firstResponse.status || (outcomeStatus !== 429 && outcomeStatus !== 402)) return; + recordCodexUpstreamOutcome(config, firstAuthCtx.accountId, outcomeStatus, { + ...codexQuotaOutcomeMeta(firstResponse), + threadId: firstAuthCtx.affinityKey, + modelId: route.modelId, + probeLeaseId: codexProbeLeaseId(firstAuthCtx), + probeQuotaScope: codexProbeQuotaScope(firstAuthCtx), + transientProbe: codexTransientProbeGrant(firstAuthCtx), + writerGeneration: firstAuthCtx.writerGeneration, + }); + }; if (outcomeStatus === 400 && ACCOUNT_GATED_NATIVE_OPENAI_MODELS.has(route.modelId)) { invalidateCodexModelEntitlementsForAccount(firstAuthCtx.accountId); let refreshed; @@ -628,26 +655,45 @@ export async function retryCodexPoolOnAlternateAccount( && retryAuthCtx?.kind !== "main-pool" && retryAuthCtx?.kind !== "main" ) { - // A body-confirmed quota response may arrive under HTTP 5xx. Without an alternate, - // the ordinary terminal recorder sees only that wire status and would misclassify it - // as transient, leaving the exhausted account immediately selectable next turn. - if (outcomeStatus !== firstResponse.status && (outcomeStatus === 429 || outcomeStatus === 402)) { - recordCodexUpstreamOutcome(config, firstAuthCtx.accountId, outcomeStatus, { - ...codexQuotaOutcomeMeta(firstResponse), - threadId: firstAuthCtx.affinityKey, - modelId: route.modelId, - probeLeaseId: codexProbeLeaseId(firstAuthCtx), - probeQuotaScope: codexProbeQuotaScope(firstAuthCtx), - transientProbe: codexTransientProbeGrant(firstAuthCtx), - writerGeneration: firstAuthCtx.writerGeneration, - }); - } + recordWrappedQuotaOutcome(); // No usable alternate was resolved, so the reserved move never becomes a send. accountMovePermit?.release(); recordUnmovedTransientOutcome(); return { kind: "no-alternate" }; } + if ( + (outcomeStatus === 429 || outcomeStatus === 402) + && !await shouldRetryCodexScopedQuotaOnAlternate( + firstResponse, + firstAuthCtx.chatgptAccountId, + retryAuthCtx.kind === "pool" || retryAuthCtx.kind === "main-pool" + ? retryAuthCtx.chatgptAccountId + // A request-owned `main` alternate has no stored account id; its workspace + // identity is what the caller's own credential materializes upstream. + : callerCodexWorkspaceAccountId(callerAuthHeaders), + options.abortSignal, + ) + ) { + // Suppressing the move is not suppressing the evidence: a same-workspace refusal + // still records its normalized quota outcome on the account that produced it. + recordWrappedQuotaOutcome(); + accountMovePermit?.release(); + releaseCodexAuthContextProbeLease(retryAuthCtx); + return { kind: "no-alternate" }; + } + + // The scope classification above reads the rejection body asynchronously, so the + // request may have been cancelled while it ran. Re-check before the send below + // mutates routing state or spends the alternate on a caller that is gone. + if (options.abortSignal?.aborted) { + recordWrappedQuotaOutcome(); + recordUnmovedTransientOutcome(); + accountMovePermit?.release(); + releaseCodexAuthContextProbeLease(retryAuthCtx); + return { kind: "no-alternate" }; + } + const quotaMeta = { ...codexQuotaOutcomeMeta(firstResponse), ...(await codexDenialOutcomeMeta(firstResponse)) }; if (outcomeStatus === 429 || outcomeStatus === 402) { const { applyAccountQuotaFromUpstreamHeaders } = await import("../../codex/auth-api"); diff --git a/src/server/responses/core-combo-failure.ts b/src/server/responses/core-combo-failure.ts index c14c3bb680b..67bf1eb9021 100644 --- a/src/server/responses/core-combo-failure.ts +++ b/src/server/responses/core-combo-failure.ts @@ -48,13 +48,14 @@ export async function consumeComboFailure( try { const body = await readBoundedResponseBody(response, { signal, - // Match shouldRetryCodexPoolAccountQuota before treating a 5xx body as quota evidence. - fatalUtf8: response.status >= 500 && response.status < 600, + // Quota evidence requires valid UTF-8, while other classifications still use the + // bounded replacement-decoded text (notably cyber-policy failures, which must stop). + reportUtf8Validity: response.status >= 500 && response.status < 600, }); usage = usageFromComboFailureText(body.text); if ( response.status >= 500 && response.status < 600 - && body.displaySafe && !body.truncated + && body.displaySafe && !body.truncated && body.utf8Valid === true ) { const quotaMessage = codexQuotaFailureMessage(body.text); quotaConfirmedByBody = quotaMessage !== undefined diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 89bf34dad3a..9905ed7dc5f 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -189,7 +189,7 @@ export { readDisplaySafeErrorText } from "./core-errors"; export { usesCodexForwardPoolAuth } from "./core-codex-account"; export { preAuthUpstreamHostCircuitKey } from "./core-codex-account"; export { upstreamHostCircuitOpenResponse } from "./core-codex-account"; -export { shouldRetryCodexPoolAccountQuota } from "./core-codex-account"; +export { shouldRetryCodexPoolAccountQuota, shouldRetryCodexScopedQuotaOnAlternate } from "./core-codex-account"; export { shouldRetryCodexPoolAccountTransient } from "./core-codex-account"; export { codexAccountGatedCanonicalWireModel } from "./core-codex-account"; export { codexForwardTerminalOutcomeRecorder } from "./core-codex-account"; diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index cebc5e9fb9c..29111c5efb6 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -122,7 +122,6 @@ import { recordCodexUpstreamOutcome } from "../../codex/routing"; import { describeUpstreamConnectFailure } from "./upstream-error"; import type { OpaqueBlobRecoveryGuard } from "./core-opaque-recovery"; import { - isOpenCodeGoDestination, rateLimitRetryPolicyFor, rateLimitRetryDelayMs, transientRetryPolicyFor, @@ -889,12 +888,6 @@ export async function preparePassthroughExchange( { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends, claimAmbiguousResend: claimPreHeaderResend, - // The OpenCode Go destination stalls-then-drops inference sends (ambiguous - // pre-header resets surfacing as refused 429s); its subscription traffic is - // inference-only, so a bounded reset replay here absorbs the blip instead of - // failing the turn. Recovery legs keep the fail-closed refusal; only this - // initial send is replay-eligible. Attempts stay budget-bounded via attempts. - replaySafe: isOpenCodeGoDestination(route.provider), }, ); } catch (err) { diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 0f910254d1a..c69ffd00cd4 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -146,7 +146,12 @@ export async function handleResponsesWithPolicyFallback( } : {}), onRequestBodyParsed: body => { options.onRequestBodyParsed?.(body); - if (body && typeof body === "object" && !Array.isArray(body)) rawBody = body as Record; + if (rawBody === null && body && typeof body === "object" && !Array.isArray(body)) { + // Recovery and other core preparation may mutate the parsed body in place. Keep an + // immutable snapshot of the original wire body so a retry cannot serialize those + // mutations. Object-identity metadata is re-established by each attempt, not serialized. + rawBody = structuredClone(body as Record); + } }, onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; diff --git a/structure/providers/kiro.md b/structure/providers/kiro.md index b1d0b503d55..4fcc55b5caf 100644 --- a/structure/providers/kiro.md +++ b/structure/providers/kiro.md @@ -24,6 +24,10 @@ reserves the private completion tool. Meta Muse 64-character MCP aliases live in ## Kiro Responses text controls +Fallback HTTP errors in `src/adapters/kiro/stream.ts` use the shared bounded display-safe body +reader with the originating request's abort signal. Oversized or incomplete bodies contribute no +classification text; cancellation releases the reader and turn retention without another fallback. + Kiro shares the Responses freeform restoration boundary in `src/responses/apply-patch-envelope.ts`: contractual `input` wrappers are unwrapped, while alternate field and outer-fence recovery is limited to unambiguous bare `exec` and `apply_patch` bodies. diff --git a/structure/providers/openai-tiers.md b/structure/providers/openai-tiers.md index e6f650c78d3..009fbe71cc0 100644 --- a/structure/providers/openai-tiers.md +++ b/structure/providers/openai-tiers.md @@ -149,11 +149,12 @@ and credential/transport failures retain their ordinary handling. `credit_balance_exhausted`, `organization_spend_limit_exceeded`, `project_spend_limit_exceeded` and `organization_usage_limit_exceeded` name a balance or cap held by the organization or project, so `classifyCodexPreStreamRejection` reports `scoped-quota-exhaustion` with `alternateRetryEligible` -false and `scopedExhaustionCode` set, and never `resetCreditEligible` — a reset credit reconciles a +true and `scopedExhaustionCode` set, and never `resetCreditEligible` — a reset credit reconciles a ChatGPT plan window and cannot pay an organization's bill. The two sets are disjoint and share one parser, so a `code`/`type` pair that disagrees, a duplicate key at any depth, or a case or whitespace near-miss yields no code at all. `codexScopedExhaustionCode` exposes the scoped answer -alone for the rotation gate and fails closed, so only positive evidence changes a routing decision. +alone for the post-resolution rotation gate and fails closed. A code by itself cannot bind the +refusal to every credential in a heterogeneous pool. `pausedCodexAccountIds` is a persisted Pool eligibility boundary. A paused added account or the stable `__main__` alias remains visible for maintenance and quota reads, but is excluded from new diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index bfd34825f7d..090904ee892 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -83,6 +83,11 @@ both execution paths. ## Bounded response ingestion and OrcaRouter login +`readBoundedResponseBody` can report UTF-8 validity independently of replacement decoding. +At complete EOF, `reportUtf8Validity` returns `utf8Valid`; combined with `fatalUtf8`, valid input +reports true and malformed input still rejects. Incomplete, oversized or timed-out bodies do not +provide positive UTF-8 evidence. Existing byte, deadline, cancellation and reader-release limits apply. + `src/lib/bounded-body.ts` owns `readBoundedResponseBytes`: it consumes the original response body without cloning or teeing and retains at most the caller's `maxBytes`. An exact-cap body requires EOF to succeed; observing an additional byte discards the retained prefix, returns an diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 023bddbe114..67ca2889f3f 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -11,6 +11,11 @@ Plaintext collaboration restoration treats a null namespace as absent, rejects n ## Responses HTTP/SSE +`src/server/responses/core-combo-failure.ts` preserves cyber-policy stops from bounded +replacement-decoded error text even when a 5xx body has malformed UTF-8. Quota/reset evidence +requires complete, display-safe, valid UTF-8. Rebuilt failures retain the non-replayable marker; +cyber-policy failures carry neither Retry-After nor quota-reset metadata. + `/v1/responses` is the main Codex-facing endpoint. The server parses Responses input, routes to a provider, lets the selected adapter speak the upstream protocol, then bridges adapter events back to Responses-compatible streaming output. For an opted-in key-auth provider, a hosted-search continuation stays bound to the API-key selection that served the first leg; the contract is the [hosted-search continuation binding](../providers-and-adapters.md#hosted-search-continuation-binding). @@ -428,11 +433,19 @@ bridge. It uses the existing account quorum, cooldown and three-rotation request the complete credential/transport/replay identity, and attributes usage to the serving account. Single-account installs do not retry; a missing alternate credential preserves the original error. -`shouldRetryCodexPoolAccountQuota` withholds that rotation when the 429 or 402 body names an -organization- or project-scoped exhaustion (`codexScopedExhaustionCode` in -`src/codex/quota-rejection.ts`). Every credential inside the refusing organization meets the same -counter, so the move would pay a second cold prompt prefix for no new capacity. Withholding the -move does not withhold the accounting: `src/server/responses/passthrough-delivery.ts` applies the +`shouldRetryCodexPoolAccountQuota` admits that rotation when the 429 or 402 body names an +organization- or project-scoped exhaustion because the response does not identify the refusing +scope. After resolving an alternate — on `/v1/responses` and on the single bounded send the +native `/responses/compact` path resolves — the rotation path uses `codexScopedExhaustionCode` +from `src/codex/quota-rejection.ts` to withhold organization-level retries only when both +credentials have the same known workspace account id. A stored Pool or main-pool alternate +supplies that id directly; a request-owned `main` alternate is bound by the caller credential's +own `chatgpt-account-id` via `callerCodexWorkspaceAccountId`. Project exhaustion remains +retryable because no project identity is available. Credentials in distinct or unknown +workspaces therefore retain failover, while a proven same-workspace move cannot pay a second +cold prompt prefix for no new capacity. A suppressed move still records the normalized 429/402 +on the refused account, so a 5xx-wrapped quota body cools it rather than letting its wire +status record as transient. `src/server/responses/passthrough-delivery.ts` applies the response's quota headers to the serving account and records the 429 outcome on the ordinary delivery path, so the account still earns its cooldown and leaves the selection pool. The gate fails closed — an empty, truncated, unparseable, duplicate-keyed or aborted body keeps the broad @@ -928,6 +941,16 @@ counter rather than holding a second. A replacement never widens a send budget: fit inside the allowance the leg already had, and it is charged to the same counter every other send goes through. +OpenCode Go inference POSTs obey this same operator gate; the destination itself does not +authorize a replay. If the granted pre-header replacement returns a transient 5xx, the reset +layer cancels that body and returns the non-replayable refusal. Policy fallback and account +rotation preserve that marker instead of interpreting its 429 as fresh quota evidence. + +`src/server/responses/policy-fallback.ts` retains one deep snapshot of the first parsed wire +body. Candidate retries serialize that snapshot, so in-place recovery or sanitizer mutations +from a previous attempt cannot become another provider's input. Object-identity metadata is +not serialized and must be established independently by each attempt. + The number of replacements is the request's as well. A leg reads it from `route.provider`, which credential rotation, OAuth refresh, transport resolution and each combo target reassign inside one request, so the grant is held to the smallest ceiling any leg has presented rather than to @@ -1085,11 +1108,12 @@ combo recall, and do not publish replacement combo/handoff recall. They never ch conversation's configured model or any compaction request outside the configured triggers. `compactionRoutingKeepsProviderIdentity` compares the source model's concrete route with the -selected route (provider name, Codex account mode and namespace; combos on either side never -match, and a bare source model the lane remembers as a combo target counts as a combo source, -recorded as `sourceCombo` when the override is applied, and a configured combo target is recorded as -`targetCombo` so its concretely routed children stay portable too). A matching identity keeps the caller's credential and may use the native compact -endpoint. A mismatch marks the credential domain as rewritten, exactly like a shadow +selected route (provider name, Codex account mode and namespace; policy selectors and combos on +either side never match, and a bare source model the lane remembers as a combo target counts as a +combo source, recorded as `sourceCombo` when the override is applied, and a configured combo target +is recorded as `targetCombo` so its concretely routed children stay portable too). A matching +identity keeps the caller's credential and may use the native compact endpoint. A mismatch marks +the credential domain as rewritten, exactly like a shadow intercept, and forces the portable summarizer even for a native-capable target: `compact.ts` skips `/responses/compact`, and `request-prepare.ts` sets `parsed._portableCompaction`, which `request-sidecar-auth.ts` (`routedCompaction`) and the passthrough adapter's compaction body @@ -1097,6 +1121,9 @@ build both honor for canonical ChatGPT destinations. Native ciphertext is replay backend that minted it; the conversation model would otherwise resume with an omission marker in place of its history. +Identity checks remove synthetic fast/effort suffixes first. A stale selector that only resolves +through the default provider cannot establish the original serving identity and stays portable. + `tests/responses/responses-compaction-override.test.ts` covers trigger selection, config validation, native and routed handlers, same-provider credential retention, cross-provider portable summaries and their replay, combo failover, and subsequent conversation settings. diff --git a/tests/ci-workflows/cold-spawn-warmup.test.ts b/tests/ci-workflows/cold-spawn-warmup.test.ts index 29e86fdc944..7df6eb92181 100644 --- a/tests/ci-workflows/cold-spawn-warmup.test.ts +++ b/tests/ci-workflows/cold-spawn-warmup.test.ts @@ -6,11 +6,12 @@ import { COLD_SPAWN_WARMUP_HOOK_BUDGET_MS, moduleGraphSpecifiers, resetColdSpawnWarmupForTests, + spawnModuleGraphWarmupChild, warmColdSpawn, warmModuleGraph, } from "../helpers/cold-spawn-warmup"; import { repoPath, repoRoot } from "../helpers/repo-root"; -import { SPAWN_BUDGET_MS } from "../helpers/test-budget"; +import { INTERNAL_DEADLINE_MS, SPAWN_BUDGET_MS } from "../helpers/test-budget"; import { analyzeWarmupRegistration, dispositionComplaints, @@ -264,6 +265,42 @@ describe("warm-up failure policy", () => { .rejects.toThrow("needs either an entry or a source"); }); + test("a warm-up child that never exits is killed at the deadline, not awaited forever", async () => { + resetColdSpawnWarmupForTests(); + // Run 35511743422's macos 2/2 leg held this shape for eighteen silent minutes: a child + // that could not be observed to exit, waited on through a synchronous spawn whose own + // timeout rode the dead event loop. The bound has to live on the parent's live loop — + // SIGKILL at the deadline, then settle. + const startedAt = performance.now(); + const result = await spawnModuleGraphWarmupChild( + "setInterval(() => undefined, 60_000)", + repoRoot(), + undefined, + 1_000, + ); + expect(performance.now() - startedAt).toBeLessThan(INTERNAL_DEADLINE_MS); + expect(result.timedOut).toBe(true); + expect(result.exitCode).not.toBe(0); + }); + + test("a descendant holding the child's pipes does not turn exit into a wait for EOF", async () => { + resetColdSpawnWarmupForTests(); + // `close` is what a clean exit earns. A grandchild that keeps the write end open must not + // convert it into an unbounded wait, so exit starts a reap grace instead. + const script = [ + 'const { spawn } = require("node:child_process");', + 'spawn(process.execPath, ["--eval", "setTimeout(() => process.exit(0), 8_000)"], { detached: true, stdio: "inherit" }).unref();', + 'process.stdout.write("ok\\n");', + "process.exit(0);", + ].join("\n"); + const startedAt = performance.now(); + const result = await spawnModuleGraphWarmupChild(script, repoRoot(), undefined, INTERNAL_DEADLINE_MS); + expect(performance.now() - startedAt).toBeLessThan(INTERNAL_DEADLINE_MS); + expect(result.exitCode).toBe(0); + expect(result.timedOut).toBe(false); + expect(result.stdout).toContain("ok"); + }); + test("a real module graph loads, and reports what it loaded", async () => { resetColdSpawnWarmupForTests(); // The end-to-end path: scan a child source, spawn one Bun child, import what it named, exit. diff --git a/tests/codex-integration/codex-quota-rejection.test.ts b/tests/codex-integration/codex-quota-rejection.test.ts index 9327b33db23..6aa05d3c4b4 100644 --- a/tests/codex-integration/codex-quota-rejection.test.ts +++ b/tests/codex-integration/codex-quota-rejection.test.ts @@ -4,6 +4,7 @@ import { BOUNDED_BODY_MAX_BYTES } from "../../src/lib/bounded-body"; import { consumeComboFailure, shouldRetryCodexPoolAccountQuota, + shouldRetryCodexScopedQuotaOnAlternate, shouldRetryCodexPoolAccountTransient, } from "../../src/server/responses/core"; import { markResponseNonReplayable } from "../../src/lib/upstream-retry"; @@ -187,7 +188,7 @@ describe("Codex pre-stream quota rejection classification", () => { expect(failure.classificationText).toContain("The usage limit has been reached"); } else { expect(failure.resetAt).toBeUndefined(); - expect(failure.classificationText).toBe("Provider error 503"); + expect(failure.classificationText).toContain("The usage limit has been reached"); } expect(response.bodyUsed).toBe(true); }); @@ -488,7 +489,8 @@ describe("Codex pre-stream quota rejection classification", () => { }); /** - * Rotating inside the limit that refused is the send amplification #4546 exists to stop. + * Rotating inside a proven-shared limit is the send amplification #4546 exists to stop. A code + * alone cannot prove that a prospective alternate belongs to the same organization or project. * * openai/codex #44492 and #45602 reclassified exactly these HTTP 429 codes as terminal quota * exhaustion while deliberately keeping `rate_limit_exceeded` and `slow_down` retryable, and @@ -499,7 +501,7 @@ describe("Codex pre-stream quota rejection classification", () => { * user-level rate limit stops failing over, and that regression would be invisible until a pool * stopped rotating in production. */ -describe("organization-scoped quota exhaustion withholds the account rotation (#4546)", () => { +describe("scoped quota exhaustion preserves unbound account rotation (#4546)", () => { const SCOPED_CODES = [ "credit_balance_exhausted", "organization_spend_limit_exceeded", @@ -512,7 +514,7 @@ describe("organization-scoped quota exhaustion withholds the account rotation (# expect(result).toEqual({ kind: "scoped-quota-exhaustion", status: 429, - alternateRetryEligible: false, + alternateRetryEligible: true, resetCreditEligible: false, scopedExhaustionCode: code, }); @@ -520,18 +522,33 @@ describe("organization-scoped quota exhaustion withholds the account rotation (# expect(result).not.toHaveProperty("semanticCode"); }); - test.each(SCOPED_CODES)("%s withholds the alternate-account send", async code => { + test.each(SCOPED_CODES)("%s keeps an unresolved alternate-account send eligible", async code => { await expect(shouldRetryCodexPoolAccountQuota(jsonRejection(429, { code }))) - .resolves.toBe(false); + .resolves.toBe(true); }); test("a root-level code and a 402 are read the same way", async () => { await expect(shouldRetryCodexPoolAccountQuota( jsonPayload(429, { code: "organization_spend_limit_exceeded" }), - )).resolves.toBe(false); + )).resolves.toBe(true); await expect(shouldRetryCodexPoolAccountQuota( jsonRejection(402, { code: "credit_balance_exhausted" }), - )).resolves.toBe(false); + )).resolves.toBe(true); + }); + + test("only proven shared organization scope withholds the resolved alternate", async () => { + const rejection = () => jsonRejection(429, { code: "organization_spend_limit_exceeded" }); + await expect(shouldRetryCodexScopedQuotaOnAlternate(rejection(), "workspace-a", "workspace-b")) + .resolves.toBe(true); + await expect(shouldRetryCodexScopedQuotaOnAlternate(rejection(), "workspace-a", undefined)) + .resolves.toBe(true); + await expect(shouldRetryCodexScopedQuotaOnAlternate(rejection(), "workspace-a", "workspace-a")) + .resolves.toBe(false); + await expect(shouldRetryCodexScopedQuotaOnAlternate( + jsonRejection(429, { code: "project_spend_limit_exceeded" }), + "workspace-a", + "workspace-a", + )).resolves.toBe(true); }); test.each([ diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 25bf5372234..bce48e0348b 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -718,6 +718,7 @@ "kiro-retry.test.ts": "providers/kiro", "kiro-review-regressions.test.ts": "providers/kiro", "kiro-stream.test.ts": "providers/kiro", + "kiro-fallback-error-body.test.ts": "providers/kiro", "kiro-usage-quota.test.ts": "providers/kiro", "kiro-windows-cli-db-path.test.ts": "providers/kiro", "kiro-windows-cli-executable-path.test.ts": "providers/kiro", @@ -1107,6 +1108,7 @@ "responses-bare-echo-helper-fence.test.ts": "responses", "responses-canonical-only-top-level-fields.test.ts": "responses", "responses-compact-handoff-admission.test.ts": "responses", + "responses-compaction-policy-identity.test.ts": "responses", "responses-compaction-override.test.ts": "responses", "responses-compaction-routing.test.ts": "responses", "responses-compaction.test.ts": "responses", @@ -1192,6 +1194,7 @@ "self-launch-argv.test.ts": "lib", "server-403-permission-e2e.test.ts": "server", "server-agent-task-recovery-replay.test.ts": "server", + "server-auth-scoped-quota.test.ts": "server", "server-auth.test.ts": "server", "server-background-lifecycle.test.ts": "server", "server-clickjacking-headers.test.ts": "server", diff --git a/tests/helpers/cold-spawn-warmup.ts b/tests/helpers/cold-spawn-warmup.ts index cc8b5289c78..aada38df630 100644 --- a/tests/helpers/cold-spawn-warmup.ts +++ b/tests/helpers/cold-spawn-warmup.ts @@ -1,3 +1,4 @@ +import { spawn } from "node:child_process"; import { readFileSync } from "node:fs"; import { dirname, isAbsolute, resolve } from "node:path"; import { repoRoot } from "./repo-root"; @@ -201,7 +202,111 @@ export async function warmModuleGraph(options: ColdSpawnWarmup): Promise { return warmColdSpawn(options.graph, deadlineMs => runModuleGraphWarmup(options, deadlineMs)); } -function runModuleGraphWarmup(options: ColdSpawnWarmup, deadlineMs: number): void { +export interface ModuleGraphWarmupResult { + stdout: string; + stderr: string; + exitCode: number | null; + signal: NodeJS.Signals | null; + timedOut: boolean; +} + +/** + * Spawn the warm-up child asynchronously and bound it on a live event loop. + * + * A blocking `Bun.spawnSync` made its own `timeout` the only bound it could honour, and that + * turned out to be no bound at all: while the synchronous wait runs, the event loop is dead, so + * the calling hook's budget and the suite's per-test timeout freeze inside the same wait and + * nothing can report anything. Run 35511743422's macos 2/2 leg held that shape for eighteen + * silent minutes inside tests/clients/client-connect.test.ts before the job ceiling cut it and + * reported `cancelled` — a result the `ci` gate reads as failure rather than evidence. Whether + * the child or the spawn primitive wedged is not observable from the outside, so the bound here + * does not depend on either: SIGKILL at the deadline, a short reap grace, and the call settles + * with or without the child's exit or EOF. A child that outlives its kill — or a descendant + * holding its pipes — cannot turn a warm-up into an unbounded wait. + */ +export function spawnModuleGraphWarmupChild( + script: string, + cwd: string, + env: Record | undefined, + deadlineMs: number, +): Promise { + const maxCaptureBytes = 1024 * 1024; + return new Promise((resolve, reject) => { + let child: ReturnType; + try { + child = spawn(process.execPath, ["--eval", script], { + cwd, + env: { ...process.env, ...env }, + stdio: ["ignore", "pipe", "pipe"], + }); + } catch { + reject(new Error("[cold-spawn-warmup] the warm-up child could not be spawned")); + return; + } + const stdoutChunks: Buffer[] = []; + const stderrChunks: Buffer[] = []; + let bytes = 0; + let settled = false; + let timedOut = false; + let exitCode: number | null = null; + let signal: NodeJS.Signals | null = null; + let deadline: ReturnType | undefined; + let reap: ReturnType | undefined; + const finish = () => { + if (settled) return; + settled = true; + clearTimeout(deadline); + clearTimeout(reap); + child.stdout?.destroy(); + child.stderr?.destroy(); + child.unref(); + resolve({ + stdout: Buffer.concat(stdoutChunks).toString("utf8"), + stderr: Buffer.concat(stderrChunks).toString("utf8"), + exitCode, + signal, + timedOut, + }); + }; + const beginReapGrace = () => { + if (settled) return; + reap ??= setTimeout(finish, WARMUP_REAP_RESERVE_MS); + }; + const stop = () => { + if (settled || timedOut) return; + timedOut = true; + clearTimeout(deadline); + beginReapGrace(); + try { child.kill("SIGKILL"); } catch { /* The kill's own failure must not extend the wait. */ } + }; + const capture = (chunk: Buffer, into: Buffer[]) => { + if (settled || timedOut) return; + bytes += chunk.length; + if (bytes > maxCaptureBytes) { stop(); return; } + into.push(chunk); + }; + child.stdout?.on("data", (chunk: Buffer) => capture(chunk, stdoutChunks)); + child.stderr?.on("data", (chunk: Buffer) => capture(chunk, stderrChunks)); + child.stdout?.on("error", stop); + child.stderr?.on("error", stop); + // The child was never started or died at launch; there is nothing to reap. + child.on("error", finish); + child.once("exit", (code, exitSignal) => { + exitCode = code; + signal = exitSignal; + // A descendant retaining a pipe must not turn a clean exit into a wait for EOF. + beginReapGrace(); + }); + child.once("close", (code, exitSignal) => { + exitCode = code; + signal = exitSignal; + finish(); + }); + deadline = setTimeout(stop, deadlineMs); + }); +} + +async function runModuleGraphWarmup(options: ColdSpawnWarmup, deadlineMs: number): Promise { const cwd = options.cwd ?? repoRoot(); const source = options.source ?? readFileSync(requireEntry(options), "utf8"); const resolveDir = options.entry === undefined ? cwd : dirname(options.entry); @@ -214,21 +319,26 @@ function runModuleGraphWarmup(options: ColdSpawnWarmup, deadlineMs: number): voi } const startedAt = performance.now(); - const result = Bun.spawnSync([process.execPath, "--eval", warmupScript(specifiers, deadlineMs)], { + const result = await spawnModuleGraphWarmupChild( + warmupScript(specifiers, deadlineMs), cwd, - env: { ...process.env, ...options.env }, - stdout: "pipe", - stderr: "pipe", - timeout: deadlineMs, - }); + options.env, + deadlineMs, + ); const elapsedMs = (performance.now() - startedAt).toFixed(0); - const stdout = result.stdout.toString(); - const report = parseWarmupReport(stdout); + const report = parseWarmupReport(result.stdout); + if (result.timedOut) { + throw new Error( + `[cold-spawn-warmup] graph=${options.graph} warm-up child did not exit within ${deadlineMs}ms ` + + `and was killed (specifiers=${specifiers.length}). ` + + `stderr: ${result.stderr.trim().slice(0, 600)}`, + ); + } if (result.exitCode !== 0 || report === undefined || report.loaded === 0) { throw new Error( `[cold-spawn-warmup] graph=${options.graph} loaded nothing in ${elapsedMs}ms ` + `(exitCode=${String(result.exitCode)}, specifiers=${specifiers.length}). ` - + `stderr: ${result.stderr.toString().trim().slice(0, 600)}`, + + `stderr: ${result.stderr.trim().slice(0, 600)}`, ); } console.log( diff --git a/tests/helpers/pool-retry-harness.ts b/tests/helpers/pool-retry-harness.ts new file mode 100644 index 00000000000..28f44589712 --- /dev/null +++ b/tests/helpers/pool-retry-harness.ts @@ -0,0 +1,245 @@ +import { existsSync, mkdirSync, mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { saveCodexAccountCredential } from "../../src/codex/account-store"; +import { + clearAccountNeedsReauth, + clearAccountQuota, + markAccountNeedsReauth, + updateAccountQuota, +} from "../../src/codex/auth-api"; +import { resetCodexModelEntitlementCacheForTests } from "../../src/codex/model-entitlements"; +import { clearCodexUpstreamHealth, clearThreadAccountMap } from "../../src/codex/routing"; +import { clearCodexWebSocketRegistry } from "../../src/codex/websocket-registry"; +import { saveConfig } from "../../src/config"; +import { startServer } from "../../src/server"; +import { clearRequestLogsForTests } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; +import { removeTreeWithRetry } from "./remove-tree"; + +const originalGlobalFetch = globalThis.fetch; + +// A per-run directory, not a fixed path, for the same reason server-auth.test.ts gives: +// `bun test --isolate` gives each file its own module registry but all files share one +// filesystem, so a literal here would collide with whichever file imported this harness. +export const POOL_RETRY_TEST_DIR = mkdtempSync(join(tmpdir(), "ocx-pool-retry-")); + +export const canonicalDirect = { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", +} as const; + +export function redirectCanonicalCodexTo(baseUrl: string): void { + const prefix = "/backend-api/codex"; + const currentWebSocket = globalThis.WebSocket; + // These fixtures serve HTTP/SSE only. Refuse the native upstream upgrade + // deterministically so its existing SSE fallback stays on the mocked fetch; + // downstream loopback WebSockets and other destinations remain real. + globalThis.WebSocket = new Proxy(currentWebSocket, { + construct(target, args, newTarget) { + const url = new URL(String(args[0])); + if (url.protocol === "wss:" && url.hostname === "chatgpt.com" + && (url.pathname === prefix || url.pathname.startsWith(`${prefix}/`))) { + throw new Error("HTTP-only Codex fixture rejects native upstream WebSocket"); + } + return Reflect.construct(target, args, newTarget); + }, + }); + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const requestUrl = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + const url = new URL(requestUrl); + if (url.hostname === "chatgpt.com" && url.pathname.startsWith(prefix)) { + const target = new URL(`${url.pathname.slice(prefix.length)}${url.search}`, baseUrl); + return originalGlobalFetch(target, init); + } + return originalGlobalFetch(input, init); + }) as typeof fetch; +} + +export const POOL_RETRY_MODEL = "gpt-5.5"; + +export function unsupportedModelBody(model = POOL_RETRY_MODEL): string { + return JSON.stringify({ + detail: `The '${model}' model is not supported when using Codex with a ChatGPT account.`, + }); +} + +export type PoolRetryHarness = { + config: OcxConfig; + dispatches: string[]; + request: (init?: { + stream?: boolean; + signal?: AbortSignal; + model?: string; + path?: "/v1/responses" | "/v1/responses/compact"; + callerBearer?: boolean; + headers?: Record; + extraBody?: Record; + }) => Promise; + restoreFetch: () => void; + server: ReturnType; + upstream: ReturnType; +}; + +async function removeTestDirBestEffort(dir: string): Promise { + if (!existsSync(dir)) return; + // Windows can keep the prior harness's ACL/icacls handles for a beat after + // stop; a single EBUSY must not take down the rest of the file. + for (let attempt = 0; attempt < 8; attempt++) { + try { + removeTreeWithRetry(dir); + return; + } catch (err) { + const code = err && typeof err === "object" && "code" in err ? String((err as { code: unknown }).code) : ""; + if (code !== "EBUSY" && code !== "EPERM" && code !== "ENOTEMPTY") throw err; + await Bun.sleep(25 * (attempt + 1)); + } + } + removeTreeWithRetry(dir); +} + +export async function startPoolRetryHarness( + reply: (accountId: string, request: Request) => Response | Promise, + options: { + secondAccount?: boolean; + streamMode?: "legacy-tee" | "eager-relay"; + accountMode?: "direct" | "pool"; + activeAccountId?: string; + accountNamespaces?: Record; + noVisionModels?: string[]; + visionSidecarModel?: string; + websockets?: boolean; + forwardApiKey?: string; + pausedAccountIds?: string[]; + reauthAccountIds?: string[]; + omitCredentialAccountIds?: string[]; + combos?: OcxConfig["combos"]; + modelRosterByAccount?: Record; + } = {}, +): Promise { + await removeTestDirBestEffort(POOL_RETRY_TEST_DIR); + mkdirSync(POOL_RETRY_TEST_DIR, { recursive: true }); + process.env.OPENCODEX_HOME = POOL_RETRY_TEST_DIR; + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + clearAccountQuota(); + resetCodexModelEntitlementCacheForTests(); + clearRequestLogsForTests(); + clearAccountNeedsReauth("pool-a"); + clearAccountNeedsReauth("pool-b"); + // The registry is process-global and survives a harness teardown. WS-REBIND-01 + // asserts exact per-account socket counts, so a socket leaked by any earlier test + // in this file shifts its snapshots and fails it in milliseconds — which reads as + // a flake next to the timeouts, but is ordinary shared state. Reset it with the + // rest rather than leaving one of six kinds of state uncleaned. + clearCodexWebSocketRegistry(); + + const dispatches: string[] = []; + const upstream = Bun.serve({ + port: 0, + async fetch(request) { + const accountId = request.headers.get("chatgpt-account-id") ?? "missing"; + if (new URL(request.url).pathname === "/models") { + return Response.json({ + models: (options.modelRosterByAccount?.[accountId] ?? []).map(slug => ({ + slug, + supported_in_api: true, + visibility: "list", + })), + }); + } + dispatches.push(accountId); + return reply(accountId, request); + }, + }); + redirectCanonicalCodexTo(upstream.url.toString()); + const redirectedFetch = globalThis.fetch; + + const secondAccount = options.secondAccount ?? true; + const config = { + port: 0, + defaultProvider: "openai", + openaiProviderTierVersion: 2, + providers: { + openai: { + ...canonicalDirect, + codexAccountMode: options.accountMode ?? "pool", + ...(options.noVisionModels ? { noVisionModels: options.noVisionModels } : {}), + ...(options.forwardApiKey ? { apiKey: options.forwardApiKey } : {}), + }, + }, + codexAccounts: [ + { id: "main", email: "main@example.test", isMain: true }, + { id: "pool-a", email: "pool-a@example.test", isMain: false, chatgptAccountId: "acct-pool-a" }, + ...(secondAccount + ? [{ id: "pool-b", email: "pool-b@example.test", isMain: false, chatgptAccountId: "acct-pool-b" }] + : []), + ], + activeCodexAccountId: options.activeAccountId ?? "pool-a", + ...(options.accountNamespaces ? { codexAccountNamespaces: options.accountNamespaces } : {}), + ...(options.pausedAccountIds ? { pausedCodexAccountIds: options.pausedAccountIds } : {}), + ...(options.visionSidecarModel ? { visionSidecar: { model: options.visionSidecarModel } } : {}), + ...(options.websockets ? { websockets: true } : {}), + ...(options.streamMode ? { streamMode: options.streamMode } : {}), + ...(options.combos ? { combos: options.combos } : {}), + } as OcxConfig; + saveConfig(config); + if (!options.omitCredentialAccountIds?.includes("pool-a")) { + saveCodexAccountCredential("pool-a", { + accessToken: "pool-a-token", + refreshToken: "pool-a-refresh", + expiresAt: Date.now() + 10 * 60_000, + chatgptAccountId: "acct-pool-a", + }); + } + updateAccountQuota("pool-a", 10); + if (secondAccount) { + if (!options.omitCredentialAccountIds?.includes("pool-b")) { + saveCodexAccountCredential("pool-b", { + accessToken: "pool-b-token", + refreshToken: "pool-b-refresh", + expiresAt: Date.now() + 10 * 60_000, + chatgptAccountId: "acct-pool-b", + }); + } + updateAccountQuota("pool-b", 20); + } + for (const accountId of options.reauthAccountIds ?? []) markAccountNeedsReauth(accountId); + + const server = startServer(0); + return { + config, + dispatches, + restoreFetch: () => { + if (globalThis.fetch === redirectedFetch) globalThis.fetch = originalGlobalFetch; + }, + server, + upstream, + request: ({ + stream = false, + signal, + model = POOL_RETRY_MODEL, + path = "/v1/responses", + callerBearer = true, + headers = {}, + extraBody = {}, + } = {}) => originalGlobalFetch(new URL(path, server.url), { + method: "POST", + headers: { + "content-type": "application/json", + ...(callerBearer ? { authorization: "Bearer inbound-token" } : {}), + ...headers, + }, + body: JSON.stringify({ model, input: path.endsWith("/compact") ? [] : "hello", stream, ...extraBody }), + signal, + }), + }; +} + +export async function stopPoolRetryHarness(harness: PoolRetryHarness): Promise { + harness.restoreFetch(); + await harness.server.stop(true); + await harness.upstream.stop(true); +} diff --git a/tests/providers/cyber-policy-error-fidelity.test.ts b/tests/providers/cyber-policy-error-fidelity.test.ts index 7adc39f3f22..85508880ee5 100644 --- a/tests/providers/cyber-policy-error-fidelity.test.ts +++ b/tests/providers/cyber-policy-error-fidelity.test.ts @@ -18,6 +18,7 @@ import { consumeComboFailure } from "../../src/server/responses/core"; import { handleResponses } from "../../src/server/responses"; import type { AdapterEvent, OcxConfig } from "../../src/types"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { isNonReplayableResponse, markResponseNonReplayable } from "../../src/lib/upstream-retry"; import { createTestTranslatorBudget, withTestTranslatorBudget } from "../helpers/translator-budget"; const createOpenAIChatAdapter = (...args: Parameters) => @@ -180,6 +181,37 @@ describe("cyber_policy error fidelity", () => { }); }); + test("malformed UTF-8 does not erase a cyber-policy stop", async () => { + const bytes = new Uint8Array([ + ...new TextEncoder().encode(JSON.stringify(CYBER_ERROR_BODY)), + 0xff, + ]); + const failure = await consumeComboFailure(new Response(bytes, { status: 502 })); + expect(failure.response.status).toBe(400); + expect(failure.upstreamCode).toBe(CYBER_POLICY_ERROR_CODE); + expect(comboFailureDecision(502, failure.classificationText, { + code: failure.upstreamCode, + })).toBe("stop"); + }); + + test("bounded recovery preserves a non-replayable malformed cyber stop", async () => { + const bytes = new Uint8Array([...new TextEncoder().encode(JSON.stringify(CYBER_ERROR_BODY)), 0xff]); + const upstream = new Response(bytes, { + status: 502, + headers: { "retry-after": "120", "x-codex-primary-reset-at": "2000000000" }, + }); + markResponseNonReplayable(upstream); + const failure = await consumeComboFailure(upstream); + expect(failure.response.status).toBe(400); + expect(failure.upstreamCode).toBe(CYBER_POLICY_ERROR_CODE); + expect(failure.nonReplayable).toBe(true); + expect(isNonReplayableResponse(failure.response)).toBe(true); + expect(failure.retryAfter).toBeUndefined(); + expect(failure.resetAt).toBeUndefined(); + expect(failure.response.headers.get("retry-after")).toBeNull(); + expect(comboFailureDecision(502, failure.classificationText, { code: failure.upstreamCode })).toBe("stop"); + }); + test("drops Codex reset headers as well as Retry-After for a cyber-policy failure", async () => { const upstream = new Response(JSON.stringify(CYBER_ERROR_BODY), { status: 429, diff --git a/tests/providers/kiro/kiro-fallback-error-body.test.ts b/tests/providers/kiro/kiro-fallback-error-body.test.ts new file mode 100644 index 00000000000..614521a28be --- /dev/null +++ b/tests/providers/kiro/kiro-fallback-error-body.test.ts @@ -0,0 +1,94 @@ +import { afterEach, test, expect } from "bun:test"; +import { createKiroAdapter as createKiroAdapterProduction, parseKiroStream } from "../../../src/adapters/kiro"; +import { encodeMessage } from "../../../src/lib/eventstream-decoder"; +import { createTranslatorBudget } from "../../../src/lib/translator-budget"; +import type { AdapterEvent, OcxParsedRequest, OcxProviderConfig } from "../../../src/types"; +import { withTestTranslatorBudget } from "../../helpers/translator-budget"; + +const realFetch = globalThis.fetch; +afterEach(() => { globalThis.fetch = realFetch; }); +const createKiroAdapter = (...args: Parameters) => + withTestTranslatorBudget(createKiroAdapterProduction(...args)); +const provider = { adapter: "kiro", baseUrl: "https://runtime.us-east-1.kiro.dev", authMode: "oauth", apiKey: "tok-123" } as OcxProviderConfig; +const bashTool = { name: "bash", description: "Run a shell command", parameters: { type: "object" } }; +function parsedWith(messages: unknown[], tools?: unknown[]): OcxParsedRequest { + return { modelId: "claude-sonnet-4.5", stream: true, options: {}, context: { messages, tools } } as OcxParsedRequest; +} +const eventFrame = (obj: unknown) => encodeMessage( + { ":message-type": "event", ":event-type": "assistantResponseEvent" }, + new TextEncoder().encode(JSON.stringify(obj)), +); +function streamOf(...frames: Uint8Array[]): ReadableStream { + let index = 0; + return new ReadableStream({ pull(controller) { + if (index < frames.length) controller.enqueue(frames[index++]); + else controller.close(); + } }); +} +async function collectAdapterEvents(events: AsyncGenerator): Promise { + const result: AdapterEvent[] = []; + for await (const event of events) result.push(event); + return result; +} + test("fallback HTTP errors stop reading oversized upstream bodies", async () => { + const chunk = new TextEncoder().encode("A".repeat(32 * 1024)); + let pulls = 0; + let cancelled = false; + globalThis.fetch = (async () => new Response(new ReadableStream({ + pull(controller) { + pulls += 1; + controller.enqueue(chunk); + }, + cancel() { + cancelled = true; + }, + }), { + status: 400, + headers: { "content-type": "text/plain" }, + })) as typeof fetch; + const adapter = createKiroAdapter(provider); + await adapter.buildRequest(parsedWith([{ role: "user", content: "do it" }], [bashTool])); + + const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf( + eventFrame({ content: "I am checking." }), + )))); + + expect(cancelled).toBe(true); + expect(pulls).toBeLessThan(10); + expect(events.at(-1)).toMatchObject({ + type: "error", + status: 400, + retryable: false, + }); + }); + + test("fallback error-body cancellation releases the reader and turn budget without another send", async () => { + const controller = new AbortController(); + const reason = new Error("fixture request cancelled"); + const budget = createTranslatorBudget(); + let cancelled = false; + let sends = 0; + const body = new ReadableStream({ + pull() { queueMicrotask(() => controller.abort(reason)); }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 }); + let caught: unknown; + try { + await collectAdapterEvents(parseKiroStream( + new Response(streamOf(eventFrame({ content: "I am checking." }))), + budget, "claude-sonnet-4.5", 0, undefined, undefined, "cancelled-turn", "required", + async () => { + sends++; + return { + response: new Response(body, { status: 400 }), abortSignal: controller.signal, + inputTokens: 0, contextInputEstimate: 0, nameMap: new Map(), conversationId: "cancelled-turn", + }; + }, + )); + } catch (error) { caught = error; } + expect(caught).toBe(reason); + expect(sends).toBe(1); + expect(cancelled).toBe(true); + expect(body.locked).toBe(false); + expect(budget.snapshot().currentBytes).toBe(0); + }); diff --git a/tests/responses/responses-compaction-policy-identity.test.ts b/tests/responses/responses-compaction-policy-identity.test.ts new file mode 100644 index 00000000000..36f1a029e2d --- /dev/null +++ b/tests/responses/responses-compaction-policy-identity.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, test } from "bun:test"; +import { getDefaultConfig } from "../../src/config"; +import { routeConcreteModel } from "../../src/router"; +import { compactionRoutingKeepsProviderIdentity } from "../../src/server/responses/compaction-routing"; +import type { OcxConfig } from "../../src/types"; + +function policyConfig(): OcxConfig { + return { + ...getDefaultConfig(), + defaultProvider: "openai-apikey", + providers: { + openai: { + adapter: "openai-responses", authMode: "forward", + baseUrl: "https://chatgpt.com/backend-api/codex", + }, + "openai-apikey": { + adapter: "openai-responses", authMode: "key", apiKey: "fixture-key", + baseUrl: "https://api.openai.com/v1", + }, + }, + routingProfiles: { + primary: { + alias: "ocx/primary", + candidates: [{ provider: "openai", model: "gpt-5.6-luna" }], + }, + }, + }; +} + +describe("compaction routing policy identity", () => { + test.each(["policy/primary", "ocx/primary"])("treats policy source %s as cross-identity", sourceModel => { + const config = policyConfig(); + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity(config, { sourceModel }, target)).toBe(false); + }); + + test.each(["policy/primary--fast", "ocx/primary--fast"])( + "treats synthetic policy selector %s as cross-identity", + sourceModel => { + const config = policyConfig(); + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity(config, { sourceModel }, target)).toBe(false); + }, + ); + + test("treats a stale policy alias as cross-identity after the profile is deleted", () => { + const config = policyConfig(); + delete config.routingProfiles; + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity(config, { sourceModel: "ocx/primary" }, target)).toBe(false); + }); + + test("fails closed for a selector that only resolves through the default provider", () => { + const config = policyConfig(); + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity( + config, + { sourceModel: "unconfigured-model" }, + target, + )).toBe(false); + }); + + test("retains identity for a concrete source on the target provider", () => { + const config = policyConfig(); + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity( + config, + { sourceModel: "openai-apikey/gpt-6-astra" }, + target, + )).toBe(true); + }); + + test("retains identity for a concrete fast selector on the target provider", () => { + const config = policyConfig(); + const target = routeConcreteModel(config, "openai-apikey/gpt-5.6-luna"); + + expect(compactionRoutingKeepsProviderIdentity( + config, + { sourceModel: "openai-apikey/gpt-6-astra--fast" }, + target, + )).toBe(true); + }); +}); diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index f111aafe71b..bf2a4555991 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -1330,6 +1330,46 @@ describe("compact alternate-account attempt (#913)", () => { expect(getCodexUpstreamHealth("pool-b")).toBeNull(); }); }); + + test(`a same-workspace alternate is withheld for a scoped ${rejection} refusal`, async () => { + await withPoolEnv(`ocx-compact-same-scope-${rejection}-`, async config => { + // pool-b shares pool-a's workspace: an organization-scoped exhaustion binds + // every credential in that workspace, so the alternate send cannot pay. + saveCodexAccountCredential("pool-b", { + accessToken: "pool-b-access-token", + refreshToken: "pool-b-refresh-token", + expiresAt: Date.now() + 300_000, + chatgptAccountId: "pool_acc_a", + }); + const bearers: string[] = []; + const accountIds: string[] = []; + const body = JSON.stringify({ + error: { + code: "organization_spend_limit_exceeded", + message: "The usage limit has been reached", + }, + }); + globalThis.fetch = (async (_url: string, init?: RequestInit) => { + const headers = new Headers(init?.headers); + bearers.push(headers.get("authorization") ?? ""); + accountIds.push(headers.get("chatgpt-account-id") ?? ""); + return new Response(body, { + status: rejection, + headers: { "content-type": "application/json", "retry-after": "42" }, + }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(bearers).toEqual(["Bearer pool-a-access-token"]); + expect(accountIds).toEqual(["pool_acc_a"]); + expect(res.status).toBe(rejection); + }); + }); } test("a native-main drain starting between attempts preserves the first rejection", async () => { diff --git a/tests/responses/responses-passthrough-transient-policy.test.ts b/tests/responses/responses-passthrough-transient-policy.test.ts index 211d30d1ed7..c2ccf954785 100644 --- a/tests/responses/responses-passthrough-transient-policy.test.ts +++ b/tests/responses/responses-passthrough-transient-policy.test.ts @@ -181,9 +181,7 @@ describe("a configured ladder is bounded by the request budget", () => { }); }); - const goPacked = dense(readResponsesCoreModule("passthrough-dispatch.ts")); -describe("the Go destination replays ambiguous resets on the initial send", () => { - test("replaySafe is destination-scoped to exactly one leg", () => { - expect(occurrences(goPacked, "replaySafe:isOpenCodeGoDestination(route.provider)")).toBe(1); - }); -}); +// The OpenCode Go replaySafe exception is gone for good: the behavioral contract is pinned +// by an execution test in responses-send-budget-counts.test.ts ("an OpenCode Go destination +// refuses an ambiguous pre-answer reset instead of replaying"), which fails if any name for +// the option ever returns. diff --git a/tests/responses/responses-send-budget-counts.test.ts b/tests/responses/responses-send-budget-counts.test.ts index b97e6f57832..b741792ed3c 100644 --- a/tests/responses/responses-send-budget-counts.test.ts +++ b/tests/responses/responses-send-budget-counts.test.ts @@ -417,6 +417,27 @@ describe("ambiguous reset safety across Responses recovery", () => { expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); expect(sends).toBe(1); }); + + test("an OpenCode Go destination refuses an ambiguous pre-answer reset instead of replaying", async () => { + // The removed replaySafe exception let the first send to this destination retry a + // dropped inference once. With it gone the destination behaves like every other: + // reset before the answer -> refusal 429, exactly one send on the wire. + const config = { + defaultProvider: "go", + providers: { go: transientChatProvider("go", { baseUrl: "https://opencode.ai/zen/go/v1" }) }, + } as unknown as OcxConfig; + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + throw Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + takeSpendHome(); + const response = await handleResponses(responsesRequest("go/model-go"), config, logCtx); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(sends).toBe(1); + }); }); describe("ambiguous reset safety after outer recovery", () => { diff --git a/tests/routing/routing-policy-fallback.test.ts b/tests/routing/routing-policy-fallback.test.ts index 347b9bdaaac..90f42435dd8 100644 --- a/tests/routing/routing-policy-fallback.test.ts +++ b/tests/routing/routing-policy-fallback.test.ts @@ -2,6 +2,8 @@ import { describe, expect, test } from "bun:test"; import { formatErrorResponse } from "../../src/bridge"; import { RequestPacingQueueOverloadError } from "../../src/providers/request-pacing"; +import { fetchWithTransientRetry, isNonReplayableResponse } from "../../src/lib/upstream-retry"; +import { shouldRetryCodexPoolAccountQuota } from "../../src/server/responses/core-codex-account"; import type { OcxConfig } from "../../src/types"; import { beginRequestAttempt, type RequestLogContext } from "../../src/server/request-log"; import type { RouteDecisionTraceV1 } from "../../src/routing/trace"; @@ -49,6 +51,32 @@ function seedAttempt(logCtx: RequestLogContext, provider: string, model: string) } describe("policy candidate fallback", () => { + test.each([false, true])("reset refusal stays terminal across policy and account recovery (replacement=%s)", async replacement => { + let sends = 0; + let coreCalls = 0; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, {} as RequestLogContext, {}, { + runCore: async (req, _config, context, options) => { + coreCalls += 1; + const body = await req.json(); + options.onRequestBodyParsed?.(body); + body.input = "attempt-local recovered text"; + context.routeDecision = policyTrace(); + return fetchWithTransientRetry(async () => { + sends += 1; + if (sends === 1) throw Object.assign(new Error("connection reset"), { code: "ECONNRESET" }); + return new Response("busy", { status: 502 }); + }, { attempts: 3, claimAmbiguousResend: () => replacement }); + }, + }); + + expect(response.status).toBe(429); + expect(isNonReplayableResponse(response)).toBe(true); + await expect(shouldRetryCodexPoolAccountQuota(response)).resolves.toBe(false); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(coreCalls).toBe(1); + expect(sends).toBe(replacement ? 2 : 1); + }); + test("policy hops retain only the original sidecar snapshot outside primary headers", async () => { const authorization = `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "sidecar-account" })}`; const initial = request(); @@ -110,6 +138,66 @@ describe("policy candidate fallback", () => { expect(cloneCalls).toBe(0); }); + test("retries from an immutable snapshot of the initially parsed body", async () => { + const trace = policyTrace(); + const logCtx = { routeDecision: trace } as RequestLogContext; + const seenInputs: unknown[] = []; + let calls = 0; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { + runCore: async (req, _config, context, options) => { + calls += 1; + const body = await req.json() as { input: unknown; model: string }; + options.onRequestBodyParsed?.(body); + seenInputs.push(body.input); + context.routeDecision = trace; + if (calls === 1) { + body.input = "recovered plaintext"; + return Response.json({ error: { type: "rate_limit_error" } }, { status: 429 }); + } + return Response.json({ status: "completed" }); + }, + }); + + expect(response.status).toBe(200); + expect(seenInputs).toEqual(["hello", "hello"]); + }); + + test("the retry snapshot survives mutation inside the input array", async () => { + // The top-level field swap above also passes under a shallow `{...body}` copy. The + // real leaks mutate deeper: the sanitizer splices input entries in place and the + // assignment injector rewrites inside the same array. Pin a nested mutation so a + // shallow-copy regression cannot stay green. + const trace = policyTrace(); + const logCtx = { routeDecision: trace } as RequestLogContext; + const seenInputs: unknown[] = []; + let calls = 0; + const req = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "policy/daily", input: [{ role: "user", content: "hello" }], stream: false }), + }); + const response = await handleResponsesWithPolicyFallback(req, {} as OcxConfig, logCtx, {}, { + runCore: async (req, _config, context, options) => { + calls += 1; + const body = await req.json() as { input: { role: string; content: string }[]; model: string }; + options.onRequestBodyParsed?.(body); + seenInputs.push(JSON.parse(JSON.stringify(body.input))); + context.routeDecision = trace; + if (calls === 1) { + body.input.splice(0, 1, { role: "assistant", content: "recovered plaintext" }); + return Response.json({ error: { type: "rate_limit_error" } }, { status: 429 }); + } + return Response.json({ status: "completed" }); + }, + }); + + expect(response.status).toBe(200); + expect(seenInputs).toEqual([ + [{ role: "user", content: "hello" }], + [{ role: "user", content: "hello" }], + ]); + }); + test("a local input-admission refusal hops instead of ending the chain (#1524)", async () => { // #1524: a candidate whose context window cannot fit the request used to TERMINATE the // fallback chain. It is a local preflight verdict about ONE candidate, not about the diff --git a/tests/server/bounded-body.test.ts b/tests/server/bounded-body.test.ts index 2a142719ae6..54bbe6ddcd8 100644 --- a/tests/server/bounded-body.test.ts +++ b/tests/server/bounded-body.test.ts @@ -22,6 +22,22 @@ function responseFromChunks(...chunks: Uint8Array[]): Response { } describe("readBoundedResponseBody", () => { + test("reportUtf8Validity is honoured on the fatal decode path at EOF", async () => { + const valid = await readBoundedResponseBody(responseFromChunks(encoder.encode('{"ok":true}')), { + fatalUtf8: true, + reportUtf8Validity: true, + }); + expect(valid.utf8Valid).toBe(true); + let caught: unknown; + try { + await readBoundedResponseBody(responseFromChunks(new Uint8Array([0xff])), { + fatalUtf8: true, + reportUtf8Validity: true, + }); + } catch (error) { caught = error; } + expect(boundedBodyDecodeFailure(caught)).toBe("invalid_utf8"); + }); + test("only actual decoder exceptions carry the decode discriminator", async () => { for (const bytes of [new Uint8Array([0xff]), new Uint8Array([0xe2, 0x82])]) { let caught: unknown; diff --git a/tests/server/server-auth-scoped-quota.test.ts b/tests/server/server-auth-scoped-quota.test.ts new file mode 100644 index 00000000000..297fdc0b9fd --- /dev/null +++ b/tests/server/server-auth-scoped-quota.test.ts @@ -0,0 +1,144 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { existsSync } from "node:fs"; +import { saveCodexAccountCredential } from "../../src/codex/account-store"; +import { clearAccountNeedsReauth, clearAccountQuota } from "../../src/codex/auth-api"; +import { resetCodexModelEntitlementCacheForTests } from "../../src/codex/model-entitlements"; +import { clearCodexUpstreamHealth, clearThreadAccountMap, getCodexUpstreamHealth } from "../../src/codex/routing"; +import { resetDebugLogBufferForTests } from "../../src/lib/debug-log-buffer"; +import { resetDebugSettingsForTests } from "../../src/lib/debug-settings"; +import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; +import { + POOL_RETRY_TEST_DIR, + startPoolRetryHarness, + stopPoolRetryHarness, +} from "../helpers/pool-retry-harness"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { SERVER_BUDGET_MS } from "../helpers/test-budget"; + +const previousApiToken = process.env.OPENCODEX_API_AUTH_TOKEN; +const previousOpencodexHome = process.env.OPENCODEX_HOME; +const originalGlobalFetch = globalThis.fetch; +const originalGlobalWebSocket = globalThis.WebSocket; +let isolatedCodexHome: IsolatedCodexHome | null = null; + +beforeEach(() => { + isolatedCodexHome = installIsolatedCodexHome("ocx-server-auth-codex-"); +}); + +afterEach(() => { + globalThis.fetch = originalGlobalFetch; + globalThis.WebSocket = originalGlobalWebSocket; + if (previousApiToken === undefined) delete process.env.OPENCODEX_API_AUTH_TOKEN; + else process.env.OPENCODEX_API_AUTH_TOKEN = previousApiToken; + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + isolatedCodexHome?.restore(); + isolatedCodexHome = null; + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + clearAccountNeedsReauth("pool-a"); + clearAccountNeedsReauth("pool-b"); + clearAccountQuota(); + resetCodexModelEntitlementCacheForTests(); + resetDebugSettingsForTests(); + resetDebugLogBufferForTests(); + if (existsSync(POOL_RETRY_TEST_DIR)) removeTreeWithRetry(POOL_RETRY_TEST_DIR); +}); + +describe("server local API auth", () => { + test.each([429, 402] as const)( + "a same-workspace caller main is bound by its workspace id and never sees a %i scoped refusal", + async rejection => { + // The alternate resolved here is the request's own main credential: it has no + // stored account id, so the scope gate can only bind it by the workspace id the + // caller credential would materialize upstream. + const model = "gpt-daybreak-blue-latest"; + const harness = await startPoolRetryHarness(() => new Response( + JSON.stringify({ + error: { + code: "organization_spend_limit_exceeded", + message: "The usage limit has been reached", + }, + }), + { status: rejection, headers: { "content-type": "application/json", "retry-after": "60" } }, + ), { + secondAccount: false, + modelRosterByAccount: { "acct-pool-a": [model] }, + }); + try { + const response = await harness.request({ + model, + headers: { "chatgpt-account-id": "acct-pool-a" }, + }); + expect(response.status).toBe(rejection); + expect(harness.dispatches).toEqual(["acct-pool-a"]); + } finally { + await stopPoolRetryHarness(harness); + } + }, + { timeout: SERVER_BUDGET_MS }, + ); + + test("a same-workspace caller main is also bound by the bearer token's account claim", async () => { + const model = "gpt-daybreak-blue-latest"; + const harness = await startPoolRetryHarness(() => new Response( + JSON.stringify({ + error: { + code: "organization_spend_limit_exceeded", + message: "The usage limit has been reached", + }, + }), + { status: 429, headers: { "content-type": "application/json", "retry-after": "60" } }, + ), { + secondAccount: false, + modelRosterByAccount: { "acct-pool-a": [model] }, + }); + try { + const response = await harness.request({ + model, + headers: { + authorization: `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "acct-pool-a" })}`, + }, + }); + expect(response.status).toBe(429); + expect(harness.dispatches).toEqual(["acct-pool-a"]); + } finally { + await stopPoolRetryHarness(harness); + } + }, { timeout: SERVER_BUDGET_MS }); + + test("a suppressed 5xx-wrapped scoped refusal still records its normalized quota outcome", async () => { + // ChatGPT sometimes wraps quota exhaustion in a generic 5xx. Suppressing the + // same-workspace alternate must still record the normalized 429 on the refused + // account — otherwise it earns only a transient failure and stays selectable. + const harness = await startPoolRetryHarness(() => new Response( + JSON.stringify({ + error: { + code: "organization_spend_limit_exceeded", + message: "The usage limit has been reached", + }, + }), + // No Retry-After: the send layer honours it as a real wait, so the cooldown must + // come from the normalized quota record's default, not the wire header. + { status: 502, headers: { "content-type": "application/json" } }, + )); + try { + // pool-b shares pool-a's workspace, so the resolved alternate is suppressed. + saveCodexAccountCredential("pool-b", { + accessToken: "pool-b-token", + refreshToken: "pool-b-refresh", + expiresAt: Date.now() + 10 * 60_000, + chatgptAccountId: "acct-pool-a", + }); + const response = await harness.request(); + expect(response.status).toBe(502); + expect(harness.dispatches).not.toContain("acct-pool-b"); + const health = getCodexUpstreamHealth("pool-a"); + expect(health).toMatchObject({ cooldownSource: "default" }); + expect(health?.cooldownUntil).toBeGreaterThan(Date.now()); + } finally { + await stopPoolRetryHarness(harness); + } + }, { timeout: SERVER_BUDGET_MS }); +}); diff --git a/tests/server/server-auth.test.ts b/tests/server/server-auth.test.ts index e7bf1bd9d6b..bca54523705 100644 --- a/tests/server/server-auth.test.ts +++ b/tests/server/server-auth.test.ts @@ -9,7 +9,7 @@ import { request as httpRequest } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { saveCodexAccountCredential } from "../../src/codex/account-store"; -import { clearCodexWebSocketRegistry, getTrackedCodexWebSocketCountForAccount } from "../../src/codex/websocket-registry"; +import { getTrackedCodexWebSocketCountForAccount } from "../../src/codex/websocket-registry"; import { INTERNAL_DEADLINE_MS, SERVER_BUDGET_MS } from "../helpers/test-budget"; import { clearAccountNeedsReauth, clearAccountQuota, getAccountQuota, isAccountNeedsReauth, markAccountNeedsReauth, updateAccountQuota } from "../../src/codex/auth-api"; import { @@ -57,6 +57,15 @@ import { resetDebugSettingsForTests, setDebugSettings } from "../../src/lib/debu import { watchdogMs } from "../helpers/ci-watchdog"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { deferredResetSseUpstream } from "../helpers/deferred-reset-sse-upstream"; +import { + POOL_RETRY_MODEL, + POOL_RETRY_TEST_DIR, + canonicalDirect, + redirectCanonicalCodexTo, + startPoolRetryHarness, + stopPoolRetryHarness, + unsupportedModelBody, +} from "../helpers/pool-retry-harness"; const previousApiToken = process.env.OPENCODEX_API_AUTH_TOKEN; const previousOpencodexHome = process.env.OPENCODEX_HOME; const originalGlobalFetch = globalThis.fetch; @@ -114,46 +123,12 @@ function managementHeaders(initial?: HeadersInit): Headers { return headers; } -const canonicalDirect = { - adapter: "openai-responses", - baseUrl: "https://chatgpt.com/backend-api/codex", - authMode: "forward", - codexAccountMode: "direct", -} as const; - function poolProviders(): OcxConfig["providers"] { return { openai: { ...canonicalDirect, codexAccountMode: "pool" }, }; } -function redirectCanonicalCodexTo(baseUrl: string): void { - const prefix = "/backend-api/codex"; - const currentWebSocket = globalThis.WebSocket; - // These fixtures serve HTTP/SSE only. Refuse the native upstream upgrade - // deterministically so its existing SSE fallback stays on the mocked fetch; - // downstream loopback WebSockets and other destinations remain real. - globalThis.WebSocket = new Proxy(currentWebSocket, { - construct(target, args, newTarget) { - const url = new URL(String(args[0])); - if (url.protocol === "wss:" && url.hostname === "chatgpt.com" - && (url.pathname === prefix || url.pathname.startsWith(`${prefix}/`))) { - throw new Error("HTTP-only Codex fixture rejects native upstream WebSocket"); - } - return Reflect.construct(target, args, newTarget); - }, - }); - globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { - const requestUrl = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; - const url = new URL(requestUrl); - if (url.hostname === "chatgpt.com" && url.pathname.startsWith(prefix)) { - const target = new URL(`${url.pathname.slice(prefix.length)}${url.search}`, baseUrl); - return originalGlobalFetch(target, init); - } - return originalGlobalFetch(input, init); - }) as typeof fetch; -} - function stubModelDiscoveryFor(...origins: string[]): void { const allowed = new Set(origins); globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { @@ -188,194 +163,9 @@ afterEach(() => { resetDebugSettingsForTests(); resetDebugLogBufferForTests(); if (existsSync(TEST_DIR)) removeTreeWithRetry(TEST_DIR); + if (existsSync(POOL_RETRY_TEST_DIR)) removeTreeWithRetry(POOL_RETRY_TEST_DIR); }); -const POOL_RETRY_MODEL = "gpt-5.5"; - -function unsupportedModelBody(model = POOL_RETRY_MODEL): string { - return JSON.stringify({ - detail: `The '${model}' model is not supported when using Codex with a ChatGPT account.`, - }); -} - -type PoolRetryHarness = { - config: OcxConfig; - dispatches: string[]; - request: (init?: { - stream?: boolean; - signal?: AbortSignal; - model?: string; - path?: "/v1/responses" | "/v1/responses/compact"; - callerBearer?: boolean; - headers?: Record; - extraBody?: Record; - }) => Promise; - restoreFetch: () => void; - server: ReturnType; - upstream: ReturnType; -}; - -async function removeTestDirBestEffort(dir: string): Promise { - if (!existsSync(dir)) return; - // Windows can keep the prior harness's ACL/icacls handles for a beat after - // stop; a single EBUSY must not take down the rest of the file. - for (let attempt = 0; attempt < 8; attempt++) { - try { - removeTreeWithRetry(dir); - return; - } catch (err) { - const code = err && typeof err === "object" && "code" in err ? String((err as { code: unknown }).code) : ""; - if (code !== "EBUSY" && code !== "EPERM" && code !== "ENOTEMPTY") throw err; - await Bun.sleep(25 * (attempt + 1)); - } - } - removeTreeWithRetry(dir); -} - -async function startPoolRetryHarness( - reply: (accountId: string, request: Request) => Response | Promise, - options: { - secondAccount?: boolean; - streamMode?: "legacy-tee" | "eager-relay"; - accountMode?: "direct" | "pool"; - activeAccountId?: string; - accountNamespaces?: Record; - noVisionModels?: string[]; - visionSidecarModel?: string; - websockets?: boolean; - forwardApiKey?: string; - pausedAccountIds?: string[]; - reauthAccountIds?: string[]; - omitCredentialAccountIds?: string[]; - combos?: OcxConfig["combos"]; - modelRosterByAccount?: Record; - } = {}, -): Promise { - await removeTestDirBestEffort(TEST_DIR); - mkdirSync(TEST_DIR, { recursive: true }); - process.env.OPENCODEX_HOME = TEST_DIR; - clearCodexUpstreamHealth(); - clearThreadAccountMap(); - clearAccountQuota(); - resetCodexModelEntitlementCacheForTests(); - clearRequestLogsForTests(); - clearAccountNeedsReauth("pool-a"); - clearAccountNeedsReauth("pool-b"); - // The registry is process-global and survives a harness teardown. WS-REBIND-01 - // asserts exact per-account socket counts, so a socket leaked by any earlier test - // in this file shifts its snapshots and fails it in milliseconds — which reads as - // a flake next to the timeouts, but is ordinary shared state. Reset it with the - // rest rather than leaving one of six kinds of state uncleaned. - clearCodexWebSocketRegistry(); - - const dispatches: string[] = []; - const upstream = Bun.serve({ - port: 0, - async fetch(request) { - const accountId = request.headers.get("chatgpt-account-id") ?? "missing"; - if (new URL(request.url).pathname === "/models") { - return Response.json({ - models: (options.modelRosterByAccount?.[accountId] ?? []).map(slug => ({ - slug, - supported_in_api: true, - visibility: "list", - })), - }); - } - dispatches.push(accountId); - return reply(accountId, request); - }, - }); - redirectCanonicalCodexTo(upstream.url.toString()); - const redirectedFetch = globalThis.fetch; - - const secondAccount = options.secondAccount ?? true; - const config = { - port: 0, - defaultProvider: "openai", - openaiProviderTierVersion: 2, - providers: { - openai: { - ...canonicalDirect, - codexAccountMode: options.accountMode ?? "pool", - ...(options.noVisionModels ? { noVisionModels: options.noVisionModels } : {}), - ...(options.forwardApiKey ? { apiKey: options.forwardApiKey } : {}), - }, - }, - codexAccounts: [ - { id: "main", email: "main@example.test", isMain: true }, - { id: "pool-a", email: "pool-a@example.test", isMain: false, chatgptAccountId: "acct-pool-a" }, - ...(secondAccount - ? [{ id: "pool-b", email: "pool-b@example.test", isMain: false, chatgptAccountId: "acct-pool-b" }] - : []), - ], - activeCodexAccountId: options.activeAccountId ?? "pool-a", - ...(options.accountNamespaces ? { codexAccountNamespaces: options.accountNamespaces } : {}), - ...(options.pausedAccountIds ? { pausedCodexAccountIds: options.pausedAccountIds } : {}), - ...(options.visionSidecarModel ? { visionSidecar: { model: options.visionSidecarModel } } : {}), - ...(options.websockets ? { websockets: true } : {}), - ...(options.streamMode ? { streamMode: options.streamMode } : {}), - ...(options.combos ? { combos: options.combos } : {}), - } as OcxConfig; - saveConfig(config); - if (!options.omitCredentialAccountIds?.includes("pool-a")) { - saveCodexAccountCredential("pool-a", { - accessToken: "pool-a-token", - refreshToken: "pool-a-refresh", - expiresAt: Date.now() + 10 * 60_000, - chatgptAccountId: "acct-pool-a", - }); - } - updateAccountQuota("pool-a", 10); - if (secondAccount) { - if (!options.omitCredentialAccountIds?.includes("pool-b")) { - saveCodexAccountCredential("pool-b", { - accessToken: "pool-b-token", - refreshToken: "pool-b-refresh", - expiresAt: Date.now() + 10 * 60_000, - chatgptAccountId: "acct-pool-b", - }); - } - updateAccountQuota("pool-b", 20); - } - for (const accountId of options.reauthAccountIds ?? []) markAccountNeedsReauth(accountId); - - const server = startServer(0); - return { - config, - dispatches, - restoreFetch: () => { - if (globalThis.fetch === redirectedFetch) globalThis.fetch = originalGlobalFetch; - }, - server, - upstream, - request: ({ - stream = false, - signal, - model = POOL_RETRY_MODEL, - path = "/v1/responses", - callerBearer = true, - headers = {}, - extraBody = {}, - } = {}) => originalGlobalFetch(new URL(path, server.url), { - method: "POST", - headers: { - "content-type": "application/json", - ...(callerBearer ? { authorization: "Bearer inbound-token" } : {}), - ...headers, - }, - body: JSON.stringify({ model, input: path.endsWith("/compact") ? [] : "hello", stream, ...extraBody }), - signal, - }), - }; -} - -async function stopPoolRetryHarness(harness: PoolRetryHarness): Promise { - harness.restoreFetch(); - await harness.server.stop(true); - await harness.upstream.stop(true); -} - function rejectionResponse(body: BodyInit, headers: Record = {}): Response { return new Response(body, { status: 400, @@ -3675,7 +3465,7 @@ describe("server local API auth", () => { test("valid JSON wrong top-level shape never authorizes a pool retry", async () => { // One harness, five bodies — same reason as the sibling above. Each - // startPoolRetryHarness() wipes and recreates TEST_DIR, binds a server, and + // startPoolRetryHarness() wipes and recreates its OPENCODEX_HOME directory, binds a server, and // redirects global fetch; five of those did not fit Bun's 5s default on a // Windows runner, and the request still in flight when the budget expired // raced the next test through that same global fetch.