import { CodexInstallationIdUnavailableError } from "../../codex/installation-id"; import { isolateCodexPoolClientMetadata } from "../../codex/forward-transport-headers"; import { captureAccountAutoPauseAttempt, wasAccountAutoPaused, wasPermanentAccountRefusal } from "../../oauth/account-auto-pause"; import { spendLedgerStorageErrorResponse } from "./spend-storage-error"; import { captureAdmissionPolicy } from "../capture-admission-policy"; import { createPhysicalSendReporter, rebindPhysicalSend } from "../../lib/request-execution-budget"; import { hasResponseMediaType } from "../../lib/http-response-semantics"; import { createInferenceSendBudget } from "../inference/context"; import { unboundPoolSpendRefusalResponse } from "../workflow-refusal"; import { capturePoolQuotaWriter } from "../../codex/account-store"; import { join } from "node:path"; import { rmSync } from "node:fs"; import { previewXaiOauthWireModel } from "./core-normalize"; import { admissionModelDeniedResponse, AdmissionModelDeniedError, assertRouteAllowedByScope, resolveAdmissionModelScope, } from "../admission-model-scope"; import type { Server } from "bun"; import { bridgeToResponsesSSE, buildResponseJSON, formatErrorResponse, type ResponsesTerminalStatus } from "../../bridge"; import { getConfigPath, getConfigDir, multiAgentGuidanceEnabled, } from "../../config"; import { resolveProviderApiKey } from "../../providers/key-store"; import { parseRequest } from "../../responses/parser"; import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractCompactUserMessages } from "../../responses/compaction"; import { applyReasoningRetention, appendRetentionNotice, type ReasoningRetentionResult } from "../../responses/reasoning-retention"; import { FORWARD_HEADERS, sanitizeReasoningInputContent } from "../../adapters/openai-responses"; import { repairUrlCitationAnnotations, stripProviderInputItemMetadata } from "../../adapters/openai-responses/request-strips"; import { expandPreviousResponseInput, previousResponseProviderState, rememberResponseState } from "../../responses/state"; import { repairLegacyDottedToolCallNames } from "../../responses/legacy-dotted-tool-name-repair"; import { NoEligiblePolicyCandidateError, routeCompactionModel } from "../../router"; import { evidenceFromBody } from "../../routing/request-evidence"; import { advanceComboAfterFailure, comboDefaultEffort, comboFailureDecision, comboIdFromRawBody, concreteComboRequestBody, getCombo, resolveComboId, isComboTargetInCooldown, NoAvailableComboTargetsError, noteComboSuccess, parseRetryAfterMs, pickComboTarget, targetKey, } from "../../combos"; import { isInjectionDebugEnabled } from "../../lib/debug-settings"; import { injectionDebugLog } from "../../lib/injection-debug-log"; import { modelInList, namespacedToolName } from "../../types"; import type { AdapterEvent, CodexAccountMode, OcxConfig, OcxParsedRequest, OcxProviderConfig, OcxProviderContinuationState, OcxUsage } from "../../types"; import { forceRefreshOAuthAccessSnapshot, getOAuthCredentialApiBaseUrl, getOAuthCredentialProjectId, getValidAccessTokenSnapshot, type OAuthAccessSnapshot, UnsupportedOAuthProviderError, } from "../../oauth"; import { buildWebSearchTool, planWebSearch, runWithWebSearch, shouldResolveOpenAiWebSearchSidecar } from "../../web-search"; import { describeImagesInPlace, planVisionSidecar, shouldResolveOpenAiVisionSidecar, stripImagesInPlace } from "../../vision"; import { createAdapterEventQueue, preflightAdapterEvents } from "../../adapters/run-turn-queue"; import { applyCodexAuthContextToProvider, callerCodexWorkspaceAccountId, createCodexAuthDispatchGuard, unwrapUpstreamRetryEvidenceError, CodexMainProfileDrainingError, headersForCodexAuthContext, materializeCodexUpstreamAuthAsync, isCodexAuthContextUsable, resolveCodexAuthContext, codexPoolAffinityKey, codexProbeLeaseId, codexTransientProbeGrant, codexProbeQuotaScope, releaseCodexAuthContextProbeLease, stripCodexRuntimeProviderFields, type CodexAuthContext, } from "../../codex/auth-context"; import { forceRefreshMainAccountToken, mainAccountTokenQuotaWriter, type NativeMainRefreshDependencies, } from "../../codex/main-account"; import { formatCodexProviderForLog, handOffThreadAffinityGeneration, recordCodexUpstreamOutcome, type CodexUpstreamOutcome, } from "../../codex/routing"; import { codexFailureWindowOutcomeMeta } from "../../codex/routing/failure-window"; import { applyAccountChangeConversationStateScrub, conversationStateBindingFromAuth, accountChangeFileReferenceRefusal, rememberServingConversationStateIssuer, conversationCarriesUploadedFiles, } from "./account-change-state"; import { TokenRefreshError, forceRefreshCodexPoolToken, readCodexAccountRecord, } from "../../codex/account-store"; import { fetchWithResetRetry, fetchWithTransientRetry, applyUpstreamRecoveryInit, isNonReplayableResponse, SendBudgetExhaustedError, TRANSIENT_RETRY_MAX_ATTEMPTS, type UpstreamSendRecovery, } from "../../lib/upstream-retry"; import { createRequestExecutionBudget, type RequestExecutionBudget, } from "../../lib/request-execution-budget"; import { classifyTransportFailureKind, transportErrorCode } from "../../lib/upstream-reachability"; import { acquireUpstreamHostAdmission, disableUpstreamHostCircuitForKey, normalizeUpstreamHostCircuitThreshold, recordUpstreamHostFailure, releaseUpstreamHostAdmission, resetUpstreamHostHealth, upstreamHostHealthKey, type UpstreamHostAdmissionLease, } from "../../codex/upstream-host-health"; import { ForwardAdmissionCredentialError, contextPrincipalIdOf, hasForwardableCodexBearer, resolveContextPrincipal, validateForwardAdmissionCredential, } from "../auth-cors"; import type { DataPlaneAdmission } from "../auth-cors"; import { listOpenAiForwardSidecarCandidates, resolveFirstUsableOpenAiSidecar, type ResolvedOpenAiForwardSidecar } from "../../providers/openai-sidecar"; import { CODEX_FORWARD_BASE_URL, isCanonicalOpenAiForwardProvider, supportsNativeResponsesCompactEndpoint } from "../../providers/openai-tiers"; import { buildResponsesTransport, copyResponsesClientMetadata } from "../../adapters/openai-responses/transport"; import { scrubOcxCompactionItems, stripCanonicalOnlyInputItems, stripCanonicalOnlyTopLevelFields, stripInternalChatMessageMetadataPassthrough } from "../../adapters/openai-responses/request-strips"; import { dropResponsesReasoningInputItems } from "../../adapters/openai-responses/reasoning"; import { requiresPlaintextReasoningReplay } from "../../adapters/openai-responses/passthrough"; import { destinationDecodesNativeCompactionBlob, isOpenAiOperatedResponsesDestination } from "../../providers/openai-tiers-destination"; import { bindRouteReasoningReplayScope } from "./core-replay"; import { commitReasoningReplayServingIdentity } from "../../responses/reasoning-replay-cache"; import { nativeCompactResponseSchema } from "../../responses/schema"; import { NATIVE_RESERVE_MODEL } from "../../codex/catalog/native-models"; import { isCodexReserveOptInMissing, isCodexReserveRequestEligible, CODEX_RESERVE_OPT_IN_REQUIRED_MESSAGE, } from "../../codex/loopback-target"; import { slugsEquivalent } from "../../providers/slug-codec"; import { decideTier, tierValueAfterDecision } from "../../providers/fastwire"; import { fastPolicyForModel } from "../../providers/service-tier"; import { parseFastOnlyRowId } from "../fast-row"; import { captureOpenAiVirtualWirePolicy, resolveOpenAiCompactModel } from "../../providers/openai-virtual-models"; import { applyCompactionRoutingOverride, compactionRoutingKeepsProviderIdentity, type CompactionRoutingOverride, } from "./compaction-routing"; import { isUsageDebugEnabled } from "../../usage/debug"; import { readJsonRequestBody, resolveInboundBodyLimitBytes, DecompressedBodyTooLargeError, UnsupportedContentEncodingError, } from "../request-decompress"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; import { hasKeyPoolFailover, rotateProviderTransportOn429, selectProactiveApiKeyTransport, transientRetryPolicyFor } from "../../providers/key-failover"; import { transientSendCapFor } from "./request-send-budget"; import { distributedKeyPhysicalFetch, recordDistributedResponseFailure } from "../../providers/api-key-response-ownership"; import { providerApiKeySelectionIsCurrent, resolveCurrentProviderApiKeyTransport } from "../../providers/api-key-selection"; import { stampApiKeyAccountLabel } from "../../providers/label"; import { shouldAttemptImageTierRetry } from "../image-retry"; import { resolveProviderTransport } from "../../providers/xai-transport"; import type { WsData } from "../ws-bridge"; import { codexAccountSelectionForTurn, registerTurn, trackStreamLifetime, unregisterTurn } from "../lifecycle"; import type { AdmissionLease } from "../../lib/admission"; import { redactSecretString } from "../../lib/redact"; import { readBoundedResponseBytes } from "../../lib/bounded-body"; import { resolveConnectTimeoutMs, resolveStallTimeoutMs } from "../../stall-timeout"; import { isLocalUpstream } from "../../lib/local-upstream"; import { isRateLimitOrQuotaFailureMessage } from "../../lib/errors"; import { supportedLadderFor } from "../effort-policy"; import { beginRequestAttempt, catalogModelSupportsServiceTier, finishRequestAttempt, inspectResponseLogJson, noteProviderAttemptSend, readConfiguredCodexServiceTier, requestLogSpeedLabel, sealRequestAttemptIdentity, usageFromResponsesPayload, type RequestLogContext, } from "../request-log"; import type { AttemptRecoveryKind } from "../../usage/log"; import { consumeForInspection, consumeForResponseLogMetadata, markNativePassthroughSseResponse, relaySseWithFailedTail, relayWithAbort, sanitizePassthroughHeaders, } from "../relay"; import { hasResponsesItemIdRepair, relaySseWithResponsesItemIdRepair } from "../responses-item-id-repair"; import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; import { codexAuthContextLogLabel } from "../../codex/account-label"; import { rebindPoolCreditPolicy } from "../../codex/pool-credit-policy"; import { clientCancelledResponse } from "./core-errors"; import { codexAccountGatedCanonicalWireModel, decodeRequestErrorResponse, handleResponses, preAuthUpstreamHostCircuitKey, poolCredentialRefreshIncompleteResponse, shouldRetryCodexScopedQuotaOnAlternate, upstreamHostCircuitOpenResponse, usesCodexForwardPoolAuth, } from "./core"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel, sendWithConnectionPolicy } from "./fetch-helpers"; import { requestPacingOverloadResponse } from "./pacing-overload"; import { shouldRetryCodexPoolAccountQuota, shouldRetryCodexPoolAccountTransient } from "./core-codex-account"; import { mapCodexAuthContextErrorToResponse, nativeMainRefreshFailureResponse } from "./codex-auth-error"; import { linkRequestSessionLane, sessionLaneIdFromRequest, reasoningReplayConversationIdFromResponsesRequest, sessionIdHeaderFromRequest } from "../request-log-conversation"; import { recallComboForLane } from "./combo-session-recall"; import { redactHostedImageDisplayPaths } from "../responses-hosted-image-display"; export const COMPACT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; const COMPACT_HANDOFF_ROUTE_TTL_MS = 24 * 60 * 60_000; const COMPACT_HANDOFF_ROUTE_MAX_ENTRIES = 2_048; const COMPACT_HANDOFF_MODEL_MAX_LENGTH = 512; interface CompactHandoffRoute { model: string; lastUsedAt: number; } /** * The Codex client does not send its newly selected model on an automatic * previous-model compact request. Keep the last route that demonstrably compacted * this same thread so a quota-blocked previous model has one safe fallback target. */ const compactHandoffRoutes = new Map(); export function clearCompactHandoffRoutesForTests(): void { compactHandoffRoutes.clear(); } function pruneCompactHandoffRoutes(now: number): void { for (const [key, entry] of compactHandoffRoutes) { if (now - entry.lastUsedAt > COMPACT_HANDOFF_ROUTE_TTL_MS) compactHandoffRoutes.delete(key); } while (compactHandoffRoutes.size > COMPACT_HANDOFF_ROUTE_MAX_ENTRIES) { const oldest = compactHandoffRoutes.keys().next().value; if (typeof oldest !== "string") return; compactHandoffRoutes.delete(oldest); } } /** * The handoff map is process-global, so a caller-controlled lane header alone * cannot be the key: two authenticated clients sending the same lane header * would share one fallback route. Namespace the lane by the admitted principal. * A loopback or missing admission has no authenticated identity to bind this * cross-request state to, so it is ineligible rather than trusted. */ function compactHandoffRouteKey(req: Request, admission: DataPlaneAdmission | undefined): string | null { const lane = sessionLaneIdFromRequest(req.headers); if (!lane || !admission || admission.kind === "loopback") return null; // Fail closed rather than substituting a weaker identity. `keyId` survives a // rotation and every identity-less environment admission would collapse into // one bucket, which is the collision this key exists to prevent. Production // admission always mints `contextPrincipalId` for configured and environment // holders, so no real authenticated caller loses the route. const principal = contextPrincipalIdOf(admission); if (!principal) return null; return `${principal}\u0000${lane}`; } function rememberCompactHandoffRoute( req: Request, admission: DataPlaneAdmission | undefined, model: string, now = Date.now(), ): void { const key = compactHandoffRouteKey(req, admission); if (!key || model.length > COMPACT_HANDOFF_MODEL_MAX_LENGTH) return; pruneCompactHandoffRoutes(now); compactHandoffRoutes.delete(key); compactHandoffRoutes.set(key, { model, lastUsedAt: now }); pruneCompactHandoffRoutes(now); } function forgetCompactHandoffRoute(req: Request, admission?: DataPlaneAdmission): void { const key = compactHandoffRouteKey(req, admission); if (key) compactHandoffRoutes.delete(key); } function compactHandoffRoute( req: Request, admission: DataPlaneAdmission | undefined, previousModel: string, now = Date.now(), ): string | null { const key = compactHandoffRouteKey(req, admission); if (!key) return null; pruneCompactHandoffRoutes(now); const entry = compactHandoffRoutes.get(key); if (!entry || entry.model === previousModel) return null; compactHandoffRoutes.delete(key); compactHandoffRoutes.set(key, { ...entry, lastUsedAt: now }); return entry.model; } export interface HandleResponsesCompactOptions { advisorAdmissionPolicy?: import("../../types/admission-policy").CapturedAdmissionPolicy; compactionRoutingOverride?: CompactionRoutingOverride | null; nativeMainRefreshDependencies?: NativeMainRefreshDependencies; /** Release the listener's idle guard only after the complete request body is accepted. */ onRequestBodyRead?: () => void; /** * The logical request's send budget (#4546). Compact used to hold its own: the normal send * took a fresh transient allowance of three, the 401 replay and the 429 alternate each added * one -- and the guard that was supposed to make those two mutually exclusive keys on * `kind === "pool"`, so a main-pool credential could spend all five. The recursive handoff * child then started over, so one compact could reach ten. */ sendBudget?: RequestExecutionBudget; } export function compactResponseTooLargeError(): Response { return new Response(JSON.stringify({ error: { message: "Compact response exceeded 32 MiB", type: "compact_response_too_large", code: "compact_response_too_large", }, }), { status: 502, headers: { "Content-Type": "application/json" } }); } async function refreshNativeMainCompactContext(args: { req: Request; config: OcxConfig; modelId?: string; admission?: DataPlaneAdmission; authCtx: CodexAuthContext; provider: OcxProviderConfig; codexAccountMode?: CodexAccountMode; substituteMainCredential: boolean; options: HandleResponsesCompactOptions; }): Promise< | { ok: true; authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } | { ok: false; response: Response } > { const { req, config, authCtx, provider, codexAccountMode, substituteMainCredential, options } = args; if (authCtx.kind !== "main-pool") { return { ok: false, response: formatErrorResponse(401, "authentication_error", "No native main credential to refresh") }; } try { const refreshed = await forceRefreshMainAccountToken(authCtx.accessToken, { signal: req.signal, ...(options.nativeMainRefreshDependencies ?? {}), }); if (!refreshed) { return { ok: false, response: formatErrorResponse(401, "authentication_error", "Codex main account needs reauthentication") }; } const refreshedAuthCtx: CodexAuthContext = { ...authCtx, accessToken: refreshed.accessToken, chatgptAccountId: refreshed.chatgptAccountId, mainQuotaWriter: mainAccountTokenQuotaWriter(refreshed) ?? authCtx.mainQuotaWriter, }; const refreshedProvider = applyCodexAuthContextToProvider( stripCodexRuntimeProviderFields(provider), refreshedAuthCtx, codexAccountMode, ); const headers = new Headers({ "content-type": "application/json" }); const selected = await materializeCodexUpstreamAuthAsync(req.headers, refreshedAuthCtx, { admission: args.admission, config, modelId: args.modelId, substituteMainCredential, signal: req.signal, nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); for (const name of FORWARD_HEADERS) { const value = selected.get(name); if (value !== null) headers.set(name, value); } const override = (refreshedProvider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; if (override) { headers.set("authorization", `Bearer ${override.accessToken}`); headers.set("chatgpt-account-id", override.chatgptAccountId); } return { ok: true, authCtx: refreshedAuthCtx, provider: refreshedProvider, headers }; } catch (error) { if (req.signal.aborted) { return { ok: false, response: formatErrorResponse(499, "client_cancelled", "Client cancelled compact request") }; } return { ok: false, response: mapCodexAuthContextErrorToResponse(error, { now: Date.now(), }) ?? nativeMainRefreshFailureResponse(error) }; } } /** See the core counterpart: only a dead grant is terminal (#2887). */ function isTerminalCompactPoolRefreshFailure(error: unknown): boolean { return error instanceof TokenRefreshError && (error.reason === "revoked" || error.reason === "expired"); } /** * Compact's forced refresh for an ordinary stored pool credential rejected with a * pre-stream 401. Mirrors {@link refreshNativeMainCompactContext} so the two 401 * contracts on this endpoint cannot drift. */ async function refreshPoolCompactContext(args: { req: Request; config: OcxConfig; modelId?: string; admission?: DataPlaneAdmission; authCtx: CodexAuthContext & { kind: "pool" }; provider: OcxProviderConfig; codexAccountMode?: CodexAccountMode; /** * Public selector for the account this refresh is for, when the request carried one. The * caller has it and this function does not, because compact takes no `RouteResult` — which * is the whole reason the refusal here used to be less specific than the one core returns * for the identical failure. */ codexAccountNamespace?: string; substituteMainCredential: boolean; options: HandleResponsesCompactOptions; }): Promise< | { ok: true; authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } | { ok: false; response: Response; quarantine: boolean; quarantineGeneration?: number } > { const { req, config, authCtx, provider, codexAccountMode, substituteMainCredential, options } = args; const reauthResponse = () => formatErrorResponse( 401, "authentication_error", "Selected Codex account needs reauthentication", ); try { const refreshed = await forceRefreshCodexPoolToken(authCtx.accountId, { rejectedGeneration: authCtx.generation, rejectedAccessToken: authCtx.accessToken, signal: req.signal, }); // See the core counterpart: a successful response can rotate only the refresh grant, // so the credential may already sit at a later generation than the one we rejected. if (!refreshed.rotated) { return { ok: false, quarantine: true, quarantineGeneration: refreshed.generation, response: reauthResponse() }; } if (refreshed.selfRefreshed) { handOffThreadAffinityGeneration(authCtx.accountId, authCtx.generation, refreshed.generation); } const refreshedAuthCtx: CodexAuthContext = { ...authCtx, accessToken: refreshed.accessToken, chatgptAccountId: refreshed.chatgptAccountId, generation: refreshed.generation, poolQuotaWriter: capturePoolQuotaWriter(authCtx.accountId, refreshed), }; // See the core counterpart: the spread dropped the WeakMap-bound credit policy, so a hold // or opt-out landing during the refresh await would go unchecked on the copy. rebindPoolCreditPolicy(authCtx, refreshedAuthCtx); const refreshedProvider = applyCodexAuthContextToProvider( stripCodexRuntimeProviderFields(provider), refreshedAuthCtx, codexAccountMode, ); const headers = new Headers({ "content-type": "application/json" }); const selected = await materializeCodexUpstreamAuthAsync(req.headers, refreshedAuthCtx, { admission: args.admission, config, modelId: args.modelId, substituteMainCredential, signal: req.signal, nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); for (const name of FORWARD_HEADERS) { const value = selected.get(name); if (value !== null) headers.set(name, value); } const override = (refreshedProvider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; if (override) { headers.set("authorization", `Bearer ${override.accessToken}`); headers.set("chatgpt-account-id", override.chatgptAccountId); } return { ok: true, authCtx: refreshedAuthCtx, provider: refreshedProvider, headers }; } catch (error) { if (isTerminalCompactPoolRefreshFailure(error)) { return { ok: false, quarantine: true, response: reauthResponse() }; } if (req.signal.aborted) { return { ok: false, quarantine: false, response: clientCancelledResponse() }; } // See the core counterpart: a policy refusal raised after a successful refresh maps like // the admission path — a reset-bound 429, never a 503 mislabeled as a refresh failure. const policyResponse = mapCodexAuthContextErrorToResponse(error, { now: Date.now(), accountSelector: args.codexAccountNamespace, }); if (policyResponse) return { ok: false, quarantine: false, response: policyResponse }; return { ok: false, quarantine: false, response: poolCredentialRefreshIncompleteResponse({ authCtx, config, accountSelector: args.codexAccountNamespace, }), }; } } /** * Resolve one eligible pool account other than `excludeAccountId`, and build everything * the alternate send needs. Returns null when no alternate exists or construction fails, * in which case the caller keeps the first account's rejection intact. * * Mirrors the auth resolution the native compact branch already does for the first * account, so the alternate is built the same way rather than through a second, * divergent path. */ async function resolveAlternateCompactContext(args: { req: Request; config: OcxConfig; route: { provider: OcxProviderConfig; codexAccountMode?: CodexAccountMode }; selectedModelId: string | undefined; excludeAccountId: string | null; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission; }): Promise<{ authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } | { failure: Response } | null> { const { req, config, route, selectedModelId, excludeAccountId, turnAdmissionLease } = args; if (!route.codexAccountMode || !excludeAccountId) return null; let authCtx: CodexAuthContext | undefined; try { authCtx = await resolveCodexAuthContext(req.headers, config, route.codexAccountMode, { admission: args.admission, ...(selectedModelId ? { modelId: selectedModelId } : {}), excludeAccountId, requestScopedMainCredential: hasForwardableCodexBearer(req.headers, config), beginCodexAccountSelection: codexAccountSelectionForTurn(turnAdmissionLease), }); // Caller-owned main has no Pool account id. It is still a valid one-shot alternate after a // stored account fails; resolveCodexAuthContext already prevents returning it when main is the // excluded credential. if (authCtx.accountId === excludeAccountId) return null; const provider = applyCodexAuthContextToProvider(route.provider, authCtx, route.codexAccountMode); const headers = new Headers({ "content-type": "application/json" }); const selected = headersForCodexAuthContext(req.headers, authCtx, config, selectedModelId, args.admission); for (const name of FORWARD_HEADERS) { const value = selected.get(name); if (value !== null) headers.set(name, value); } const override = (provider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; if (override) { headers.set("authorization", `Bearer ${override.accessToken}`); headers.set("chatgpt-account-id", override.chatgptAccountId); } if (provider.apiKey) headers.set("authorization", `Bearer ${resolveProviderApiKey(provider.apiKey)}`); return { authCtx, provider, headers }; } catch (err) { releaseCodexAuthContextProbeLease(authCtx); if (err instanceof CodexInstallationIdUnavailableError) return { failure: formatErrorResponse(503, "server_busy", err.message) }; if (err instanceof CodexMainProfileDrainingError) { // The native-main fence can start after account A has already rejected the // request. Treat the now-fenced main profile as no alternate and preserve A's // real rejection instead of replacing it with a synthetic drain response. return null; } // No eligible alternate (all cooled, affinity expired, reauth needed) — the caller // returns the first account's rejection unchanged, which is today's behavior. return null; } } /** * Headers a client needs to back off correctly after a pool rejection. The buffered * response is rebuilt from scratch, so anything not listed here is dropped — which is * what silently discarded `Retry-After` and the reset hints from a 429 before. * Deliberately narrow: no hop-by-hop headers, no content-length (the buffered body sets * its own), no cookies. */ const COMPACT_PASSTHROUGH_HEADERS = [ "retry-after", "x-codex-primary-reset-at", "x-codex-secondary-reset-at", "x-codex-tertiary-reset-at", // A relayed 3xx keeps its Location so the client can follow it (#914). "location", ]; function compactResponseHeaders(upstream: Response): Headers { const headers = new Headers({ "Content-Type": upstream.headers.get("content-type") ?? "application/json" }); for (const name of COMPACT_PASSTHROUGH_HEADERS) { const value = upstream.headers.get(name); if (value !== null) headers.set(name, value); } return headers; } export async function bufferCompactResponse( upstream: Response, signal: AbortSignal, stallTimeoutSec?: number, localUpstream?: boolean, ): Promise { const headers = compactResponseHeaders(upstream); try { if (signal.aborted) { // No reader is attached yet. Cancellation must not wait for a broken source's cleanup. void upstream.body?.cancel(signal.reason).catch(() => undefined); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } if (!upstream.body) return new Response(null, { status: upstream.status, statusText: upstream.statusText, headers }); const declaredLength = Number(upstream.headers.get("content-length")); if (Number.isFinite(declaredLength) && declaredLength > COMPACT_RESPONSE_MAX_BYTES) { void upstream.body.cancel("compact_response_too_large").catch(() => undefined); return compactResponseTooLargeError(); } // Header admission has finished; only non-empty body chunks re-arm this deadline. // The raw reader preserves bytes and cancels/releases without awaiting source cleanup. const result = await readBoundedResponseBytes(upstream, { signal, maxBytes: COMPACT_RESPONSE_MAX_BYTES, // Compaction buffers the complete body while holding an active-turn lease. Keep its // default bounded even for local destinations so silent bodies cannot exhaust that gate. inactivityTimeoutMs: resolveStallTimeoutMs(stallTimeoutSec), }); if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); if (result.oversized) return compactResponseTooLargeError(); return new Response(result.bytes, { status: upstream.status, statusText: upstream.statusText, headers }); } catch (error) { if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); if (error instanceof DOMException && error.name === "TimeoutError") { return Response.json({ error: { message: "Compact response body stalled", type: "upstream_stall_timeout", code: "upstream_stall_timeout", } }, { status: 504 }); } return formatErrorResponse(502, "upstream_error", "Failed to read compact response"); } } export async function handleResponsesCompact( req: Request, config: OcxConfig, logCtx: RequestLogContext, turnAdmissionLease?: AdmissionLease, admission?: DataPlaneAdmission, options: HandleResponsesCompactOptions = {}, ): Promise { options = { ...options, advisorAdmissionPolicy: options.advisorAdmissionPolicy ?? captureAdmissionPolicy(config, admission) }; let body: unknown; try { body = await readJsonRequestBody(req, undefined, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); } catch (err) { return decodeRequestErrorResponse(err, "responses-compact"); } if (!body || typeof body !== "object" || Array.isArray(body)) { return formatErrorResponse(400, "invalid_request_error", "Invalid compaction request body"); } // Compaction forwards client history upstream too; keep local artifact paths out of it. redactHostedImageDisplayPaths(body); if (!options.compactionRoutingOverride) { options = { ...options, compactionRoutingOverride: applyCompactionRoutingOverride(body, req.headers, config, { endpoint: "compact" }) }; } const raw = body as { model?: unknown; input?: unknown }; if (typeof raw.model !== "string" || raw.model.length === 0) { return formatErrorResponse(400, "invalid_request_error", "compaction request requires a model"); } options.onRequestBodyRead?.(); // Correct the IDENTITY before routing, or the synthetic id does not route at all. Held in // a local rather than written back to `raw.model`: assigning to the property widens it out // of the `string` narrowing the guard above just established. const compactFastRow = parseFastOnlyRowId(config, () => raw.model as string); let compactModel = compactFastRow ? compactFastRow.baseId : raw.model; if (compactFastRow) (raw as Record).model = compactModel; // The client's own selector, kept for the request log: `raw.model` is rewritten to the // base id above, and logCtx.requestedModel is assigned from it further down, so without // this the log would lose which id the client actually asked for. const compactRequestedModel = options.compactionRoutingOverride?.sourceModel ?? (compactFastRow ? compactFastRow.baseId + "--fast" : raw.model); // Recall the last completed client-visible bare model after a combo switch (#3891). // Configured selectors take precedence over this implicit session hint. if (!options.compactionRoutingOverride && typeof compactModel === "string" && !compactModel.includes("/") && !compactFastRow && !resolveComboId(config, compactModel)) { const recalledComboId = recallComboForLane(config, sessionLaneIdFromRequest(req.headers), compactModel); if (recalledComboId) { (raw as Record).model = `combo/${recalledComboId}`; // Keep the routed identity in sync: the bare model can 404 outright (no // canonical openai provider) or resolve straight onto a native-compact // provider, both bypassing combo failover. The combo selector resolves // through tryPickComboModel, whose route.combo skips the native compact // endpoint. compactModel = `combo/${recalledComboId}`; } } let route; try { // Compact requests route through the same policy evaluation as normal // turns, so body-derived evidence (tools/image) must reach the first // evaluation too - not only the later handleResponses dispatch. // Codex selects a bare native model for compaction even when the operator // routes ordinary turns elsewhere (#2901); the compaction-scoped router // may land that on the configured default provider instead of 404. route = routeCompactionModel(config, compactModel, evidenceFromBody(raw)); // A compaction override picks the model, not the caller, so the key's scope // is applied to what the override resolved to rather than to the selector // the client sent. // Routed xAI OAuth compaction must preview its Fast wire model here; // native compact keeps checking the destination it dispatches directly. const callerTier = (raw as Record).service_tier; assertRouteAllowedByScope(resolveAdmissionModelScope(config, admission), compactRequestedModel, { providerName: route.providerName, modelId: previewXaiOauthWireModel({ options: { serviceTier: compactFastRow ? "priority" : typeof callerTier === "string" ? callerTier : undefined, } }, route, config, "responses"), }); } catch (err) { if (err instanceof AdmissionModelDeniedError) return admissionModelDeniedResponse(err); if (err instanceof NoEligiblePolicyCandidateError) { // Persist the evaluation trace (per-candidate exclusions + the // no-eligible reason) so a failed compact policy request stays // auditable, matching the other request handlers. logCtx.routeDecision = err.trace; } return formatErrorResponse(404, "invalid_request_error", err instanceof Error ? err.message : String(err)); } const selectedModelId = route.modelId; // Derive from the RESOLVED route model, not the caller's raw string. An account-qualified // selector like `side/gpt-daybreak-blue-latest` does not match the gated map — `slugsEquivalent` // reads the account namespace as a routed provider prefix — so keying on `raw.model` sent // exactly the selector form back down the native compact endpoint this guard exists to avoid. // `route.modelId` is the same value `applyCodexAccountGatedWireNormalization` uses in core.ts. const accountGatedCompactWireModel = isCanonicalOpenAiForwardProvider(route.provider) ? codexAccountGatedCanonicalWireModel(selectedModelId) : undefined; logCtx.requestedModel = compactRequestedModel; logCtx.model = selectedModelId; logCtx.routeDecision = route.routeDecision; logCtx.policyEligibility = route.policyEligibility; logCtx.provider = route.codexAccountNamespace ? `${route.providerName}-${route.codexAccountNamespace}` : route.providerName; logCtx.spendPoolId = route.providerName; logCtx.providerAdapter = route.provider.adapter; // #4940, and the same refusal the ordinary Responses path makes in request-prepare.ts. Compact has // to repeat it rather than inherit it: the native branch below dispatches straight to // `/responses/compact` and only the routed fallback replays through handleResponses, so relying on // that one seam would leave the native compact turn forwarding a Reserve model the opt-in has made // unservable. Composed from the same three facts, in the same order, as `customReserveForward` // further down, and placed before the virtual-model rewrite so it reads the model the caller // actually selected -- and before auth, host-circuit admission, or any upstream byte. // // No terminal-helper or inbound-wire qualifier is needed here the way it is on the ordinary path: // this endpoint is the native Codex compaction wire, and a vision/search helper never reaches it. if (isCodexReserveOptInMissing(config, selectedModelId, admission) && isCanonicalOpenAiForwardProvider(route.provider)) { return formatErrorResponse(400, "invalid_request_error", CODEX_RESERVE_OPT_IN_REQUIRED_MESSAGE); } const virtual = resolveOpenAiCompactModel(route.providerName, selectedModelId); if (virtual) { route.modelId = virtual.wireModelId; captureOpenAiVirtualWirePolicy(route, virtual); logCtx.model = virtual.selectedModelId; logCtx.resolvedModel = virtual.wireModelId; logCtx.wireModel = virtual.wireModelId; } else { logCtx.resolvedModel = route.modelId; } if (compactFastRow) { // Resolved AFTER the virtual-model rewrite above, and against `route.modelId`, which is // now the WIRE model: `resolveOpenAiCompactModel` maps an alias like `gpt-5.6-sol-pro` // onto a different wire id, and capability overrides are keyed by exact model id, so // deciding before the rewrite could set `priority` on a wire model that does not support // it. `capabilityProvider` is passed for the same reason core.ts:2103 passes it. const decision = decideTier( fastPolicyForModel( route.provider, route.modelId, route.providerName, "responses", config.providers[route.providerName], ), config.fastMode, "priority", ); // The WHOLE decision: native compact spreads `raw` into the forwarded body, so on a drop // a caller's pre-existing service_tier must be REMOVED rather than left to ride along // past the suppression. const serviceTier = tierValueAfterDecision(decision, "priority"); if (serviceTier === undefined) delete (raw as Record).service_tier; else (raw as Record).service_tier = serviceTier; } // #1686: a bearer-presented admission secret is one of ours, so the stored main credential // is substituted below instead of the caller bearer being forwarded. // #2132: and only when the route is a native Codex one, which is the only route that can // consume that credential. See the longer note in core.ts resolveResponsesCodexAuth. const customReserveForward = selectedModelId === NATIVE_RESERVE_MODEL && isCodexReserveRequestEligible(config, admission) && isCanonicalOpenAiForwardProvider(route.provider); const substituteMainCredential = admission?.source === "bearer" && (route.codexAccountMode !== undefined || customReserveForward); const requestScopedMainCredential = route.codexAccountMode !== undefined && !substituteMainCredential && hasForwardableCodexBearer(req.headers, config); if (route.codexAccountMode === "direct" && !substituteMainCredential) { try { validateForwardAdmissionCredential(req.headers, config); } catch (err) { if (err instanceof ForwardAdmissionCredentialError) return formatErrorResponse(401, "authentication_error", err.message); throw err; } } // Native /responses/compact exists on the canonical ChatGPT backend and on the // official OpenAI API, or a gateway explicitly declaring the native endpoint. Other // Responses-shaped gateways still take the routed summarizer path (#422). // Combo-resolved targets skip native compact so failover can advance through the // combo target list when the picked model returns 429/5xx — the routed path below // dispatches through handleResponses → handleComboResponses with full failover. // // One holder for the WHOLE logical compact, declared above the native branch because the // routed fallback below is not a different request: a native attempt that 404s, or a quota // failure that hands off, continues here. The routed turn used to call handleResponses with // no budget at all, so `handleResponsesInner` minted a fresh four after the native attempt // had already spent some of the first one. const sendBudget: RequestExecutionBudget = options.sendBudget ?? createInferenceSendBudget(req, logCtx); // A manual override onto another backend must not mint ciphertext the conversation model cannot replay. const manualOverrideCrossesProvider = options.compactionRoutingOverride ? !compactionRoutingKeepsProviderIdentity(config, options.compactionRoutingOverride, route) : false; if (supportsNativeResponsesCompactEndpoint(route.providerName, route.provider) && !accountGatedCompactWireModel && !route.combo && !manualOverrideCrossesProvider) { if (req.signal.aborted) { return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } // The enclosing native-compact guard already restricts this path to // supported backends, so compact intentionally does not require the // regular Responses adapter check here. const preAuthCompactHostKey = preAuthUpstreamHostCircuitKey(route, config, { requireResponsesAdapter: false, }); let compactHostAdmissionLease: UpstreamHostAdmissionLease | null = null; let authCtx: CodexAuthContext = { kind: "main", accountId: null }; if (preAuthCompactHostKey) { const admission = acquireUpstreamHostAdmission( preAuthCompactHostKey, config.upstreamHostCircuitThreshold, ); if (admission.kind === "blocked") { return upstreamHostCircuitOpenResponse(admission.retryAfterSeconds); } compactHostAdmissionLease = admission.lease; } try { // Native-capable destination: forward the compact request to its configured backend. // Resolve the SAME pool/thread auth context as /v1/responses — forwarding the caller's raw // headers would run compaction on the wrong account (or 401) whenever a pool account is // active for this thread while normal turns succeed. let compactProvider = route.provider; let headers = new Headers({ "content-type": "application/json" }); try { if (route.codexAccountMode || customReserveForward) { if (route.codexAccountMode) authCtx = await resolveCodexAuthContext(req.headers, config, route.codexAccountMode, { admission, accountId: route.codexAccountId, modelId: selectedModelId, substituteMainCredentialForDirect: substituteMainCredential, requestScopedMainCredential, beginCodexAccountSelection: codexAccountSelectionForTurn(turnAdmissionLease), signal: req.signal, nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, // #4778: the same retention the regular Responses path passes at its final auth. The // post-429 guard below only declines to move AFTER this resolution has already bound // an account, so without the bit here a quota-driven rebind could have carried the // conversation off its issuing account before that guard is ever consulted -- and an // uploaded file is readable only by the account that received it. Answered from `raw`, // the same object and the same predicate the guard below uses, so the two can never // disagree about which conversations are in scope. retainAccountForUploadedFiles: conversationCarriesUploadedFiles(raw), }); logCtx.accountLogLabel = codexAuthContextLogLabel(authCtx, config); const selected = await materializeCodexUpstreamAuthAsync(req.headers, authCtx, { admission, config: isCanonicalOpenAiForwardProvider(route.provider) ? config : undefined, modelId: selectedModelId, beginCodexAccountSelection: codexAccountSelectionForTurn(turnAdmissionLease), substituteMainCredential, signal: req.signal, nativeMainRefreshDependencies: options.nativeMainRefreshDependencies, }); compactProvider = applyCodexAuthContextToProvider(route.provider, authCtx, route.codexAccountMode); for (const name of FORWARD_HEADERS) { const value = selected.get(name); if (value !== null) headers.set(name, value); } const override = (compactProvider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; if (override) { headers.set("authorization", `Bearer ${override.accessToken}`); headers.set("chatgpt-account-id", override.chatgptAccountId); } } } catch (err) { if (req.signal.aborted) { return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } const response = mapCodexAuthContextErrorToResponse(err, { accountSelector: route.codexAccountNamespace, now: Date.now(), }); if (response) return response; throw err; } const base = isCanonicalOpenAiForwardProvider(compactProvider) ? CODEX_FORWARD_BASE_URL : (compactProvider.baseUrl ?? "").replace(/\/+$/, ""); if (compactProvider.authMode !== "forward" && compactProvider.apiKey) { // Native compact never enters handleResponses, so it needs its own pre-dispatch key // pick. Kept inside this branch on purpose: the overlay above owns the forward and // codexAccountMode cases, and the picker returns null for them anyway. // Transport variant, for the same reason as the Responses core: the persisted row has no // registry backfills, and compactProvider is read for its transport fields below. let warmKeyProvider; try { warmKeyProvider = selectProactiveApiKeyTransport(config, route.providerName, compactProvider); } catch (error) { const refusal = requestPacingOverloadResponse(error); if (refusal) return refusal; throw error; } if (warmKeyProvider?.apiKey) compactProvider = warmKeyProvider; headers.set("authorization", `Bearer ${resolveProviderApiKey(compactProvider.apiKey)}`); stampApiKeyAccountLabel(logCtx, route.providerName, compactProvider); } const { reasoning: _reasoning, ...compactBodyRaw } = raw as typeof raw & { reasoning?: unknown }; const requireCompactPlaintextReasoning = !isOpenAiOperatedResponsesDestination(compactProvider) && requiresPlaintextReasoningReplay(compactProvider); // The regular /v1/responses path applies sanitizeReasoningInputContent via the adapter's // buildRequest, but the compact endpoint forwards directly. Apply the same sanitizer here // so canonical reasoning stays blank and custom gateways honor their explicit plaintext policy. // #5095: the compact endpoint forwards `raw` directly, so it needs the same legacy dotted // call-name repair the adapter applies. A damaged item here refuses the compaction itself, // which is the request a long task depends on to keep going. // Direct compact bypasses the adapter's Google replay projection (#7111). const compactBody = repairLegacyDottedToolCallNames( stripProviderInputItemMetadata(sanitizeReasoningInputContent(compactBodyRaw, { preserveRawReasoningContent: !isOpenAiOperatedResponsesDestination(compactProvider) && compactProvider.preserveResponsesReasoningContent === true, dropNullContentChannel: !isOpenAiOperatedResponsesDestination(compactProvider), stripEncryptedContent: requireCompactPlaintextReasoning, requirePlaintextReasoning: requireCompactPlaintextReasoning, })), ) as typeof compactBodyRaw; { const binding = conversationStateBindingFromAuth(authCtx, codexPoolAffinityKey(req.headers)); if (binding) { // Refused rather than scrubbed: an uploaded file is content the caller attached, not // continuation state the turn can do without. const refusal = accountChangeFileReferenceRefusal({ body: raw, bindingKey: binding.bindingKey, servingAccountId: binding.accountId, }); if (refusal) return refusal; applyAccountChangeConversationStateScrub({ body: raw, headers, bindingKey: binding.bindingKey, servingAccountId: binding.accountId, logCtx, }); applyAccountChangeConversationStateScrub({ body: compactBody, bindingKey: binding.bindingKey, servingAccountId: binding.accountId, logCtx, }); } } // Gateway protocol capability never grants official credential authority. Reuse the normal // Responses transport for custom paths, static auth and explicitly forwarded client metadata. const gatewayTransport = !isCanonicalOpenAiForwardProvider(compactProvider) && compactProvider.supportsNativeCompactEndpoint === true ? buildResponsesTransport({ ...compactProvider, apiKey: compactProvider.apiKey ? resolveProviderApiKey(compactProvider.apiKey) : undefined }, { headers: req.headers }) : undefined; if (gatewayTransport) { headers = new Headers(gatewayTransport.headers); const binding = conversationStateBindingFromAuth(authCtx, codexPoolAffinityKey(req.headers)); if (binding) applyAccountChangeConversationStateScrub({ body: compactBody, headers, bindingKey: binding.bindingKey, servingAccountId: binding.accountId, logCtx }); } // Native v1 bypasses the adapter serializer. Bind the same replay provenance as ordinary // turns, but leave canonical compact's established account-state handling unchanged. const gatewayReplayParsed = gatewayTransport ? parseRequest(raw) : undefined; if (gatewayReplayParsed) { gatewayReplayParsed.modelId = route.modelId; gatewayReplayParsed._clientThreadId = req.headers.get("x-codex-parent-thread-id")?.trim() || undefined; const clientThreadId = reasoningReplayConversationIdFromResponsesRequest({ clientThreadId: gatewayReplayParsed._clientThreadId, threadIdHeader: req.headers.get("thread-id"), cursorConversationId: gatewayReplayParsed._cursorConversationId, sessionIdHeader: sessionIdHeaderFromRequest(req.headers), }); if (clientThreadId) { gatewayReplayParsed._reasoningReplayScope = { clientThreadId, clientPrincipalId: resolveContextPrincipal(req, config, admission), }; } } let compactUrl = gatewayTransport ? `${gatewayTransport.url.replace(/\/+$/, "")}/compact` : `${base}/responses/compact`; const compactTargetKey = `${route.providerName}|${route.modelId}|compact`; let actualCompactHostKey = upstreamHostHealthKey( route.providerName, safeOriginLabel(compactUrl), ); const compactHostKey = compactProvider.authMode === "forward" ? actualCompactHostKey : null; const compactHostCircuitEnabled = compactHostKey !== null && normalizeUpstreamHostCircuitThreshold(config.upstreamHostCircuitThreshold) > 0; if (compactHostKey !== null && !compactHostCircuitEnabled) { disableUpstreamHostCircuitForKey(actualCompactHostKey); } if (compactHostAdmissionLease && compactHostAdmissionLease.key !== compactHostKey) { releaseCodexAuthContextProbeLease(authCtx); return formatErrorResponse(502, "upstream_error", "Provider host changed after circuit admission"); } if (req.signal.aborted) { releaseCodexAuthContextProbeLease(authCtx); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } if (!compactHostAdmissionLease && compactHostCircuitEnabled) { const admission = acquireUpstreamHostAdmission( compactHostKey!, config.upstreamHostCircuitThreshold, ); if (admission.kind === "blocked") { releaseCodexAuthContextProbeLease(authCtx); return upstreamHostCircuitOpenResponse(admission.retryAfterSeconds); } compactHostAdmissionLease = admission.lease; } const settleObservedCompactHostResponse = (): void => { if (compactHostCircuitEnabled) { resetUpstreamHostHealth(actualCompactHostKey, compactHostAdmissionLease); } else { resetUpstreamHostHealth(actualCompactHostKey); } compactHostAdmissionLease = null; }; const connectMs = () => resolveConnectTimeoutMs(config.connectTimeoutMs, { localUpstream: isLocalUpstream(compactUrl) }); const applyCompactPoolQuotaHeaders = async (ctx: CodexAuthContext, response: Response) => { if (ctx.kind !== "pool" && ctx.kind !== "main-pool") return; const { applyAccountQuotaFromUpstreamHeaders } = await import("../../codex/quota"); applyAccountQuotaFromUpstreamHeaders(ctx.accountId, response.headers, ctx.writerGeneration, ctx.kind === "main-pool" ? ctx.mainQuotaWriter : undefined, { modelId: route.modelId, poolWriter: ctx.kind === "pool" ? ctx.poolQuotaWriter : undefined, poolResponse: ctx.kind === "pool" }); }; // Takes its context explicitly: the alternate-account flow below records a rejection // against A while promoting B, then records B's own outcome. A closure over a single // `authCtx` cannot express either. const recordCompactPoolOutcome = ( ctx: CodexAuthContext, outcome: CodexUpstreamOutcome, meta: { retryAfter?: string | null; resetAt?: unknown | unknown[]; promoteAccountId?: string; } = {}, ) => { // A gateway response is not evidence about any Codex account. if (gatewayTransport || !usesCodexForwardPoolAuth(ctx, route.provider)) return; if ((outcome === 401 || outcome === 403) && upstream && wasPermanentAccountRefusal(upstream)) return; recordCodexUpstreamOutcome(config, ctx.accountId, outcome, { ...meta, ...codexFailureWindowOutcomeMeta(ctx), threadId: ctx.kind === "pool" || ctx.kind === "main-pool" ? ctx.affinityKey : undefined, fixedAccount: ctx.fixedAccount, modelId: selectedModelId, probeLeaseId: codexProbeLeaseId(ctx), // Fence a stored pool account's outcome on the credential the request was // holding, so a 401 that lost a race with re-authentication cannot retire the // replacement (#2887). Also covers the replay's own second 401. ...(ctx.kind === "pool" ? { credentialGeneration: ctx.generation } : {}), probeQuotaScope: codexProbeQuotaScope(ctx), transientProbe: codexTransientProbeGrant(ctx), writerGeneration: ctx.kind === "pool" || ctx.kind === "main-pool" ? ctx.writerGeneration : undefined, }); }; // Two recovery modes, mirroring retryCodexPoolOnAlternateAccount() on the regular // path (core.ts:396). The first account keeps the full ladder — transient-5xx retry // wrapping reset retry — because those retries happen before any alternate is even // considered. The alternate is one bounded send: a second ladder would multiply the // work an already-rejecting pool is doing. // // Both modes now draw one shared budget. The comment below used to say the 401 replay // spends the account budget so the 429 alternate is skipped, but that guard is keyed on // `kind === "pool"` and a main-pool credential left it false -- so 401 then 429 really did // reach five. The single sends spend the base allowance first and then the one shared // final-recovery reserve, which is the same rule the Responses path follows. const sendSingleCompactAttempt = async ( doFetch: () => Promise, ): Promise => { if (sendBudget.remainingBaseSends(TRANSIENT_RETRY_MAX_ATTEMPTS) > 0) { const report = createPhysicalSendReporter(sendBudget, () => ({ poolId: logCtx.spendPoolId ?? route.providerName, identityId: logCtx.accountLogLabel })); let started = false; try { if (report.beforeSend?.() === false) throw new SendBudgetExhaustedError(safeHostLabel(compactUrl)); started = true; return await (report.execute ? report.execute(doFetch) : doFetch()); } finally { try { if (started) report(1); } finally { report.close?.(); } } } const decision = sendBudget.reserveDispatch({ sendClass: "auth-recovery", targetKey: compactTargetKey, }); if (!decision.allowed) return Promise.reject(new SendBudgetExhaustedError(safeHostLabel(compactUrl))); const dispatch = () => { if (!decision.permit.use()) { decision.permit.release(); throw new SendBudgetExhaustedError(safeHostLabel(compactUrl)); } return doFetch(); }; return decision.permit.execute ? decision.permit.execute(dispatch) : dispatch(); }; const sendCompactAttempt = ( sendProvider: OcxProviderConfig, sendHeaders: Headers, recovery: "normal" | "single", sendAuthCtx: CodexAuthContext, onPhysicalDispatch?: () => void, ): Promise => { // A gateway send carries only its own transport headers, whatever recovery handed in. // An account-change scrub that removed the turn state from the send headers still applies. let wireHeaders = gatewayTransport ? new Headers(gatewayTransport.headers) : sendHeaders; if (gatewayTransport && !sendHeaders.has("x-codex-turn-state")) wireHeaders.delete("x-codex-turn-state"); sendHeaders = wireHeaders; const buildCompactSendBody = () => { if (gatewayReplayParsed) bindRouteReasoningReplayScope({ parsed: gatewayReplayParsed, providerName: route.providerName, provider: sendProvider, adapterName: sendProvider.adapter, codexAuthContext: sendAuthCtx, forwardHeaders: wireHeaders, }); const requireSendPlaintextReasoning = !isOpenAiOperatedResponsesDestination(sendProvider) && requiresPlaintextReasoningReplay(sendProvider); let replayBody = gatewayReplayParsed ? sanitizeReasoningInputContent(scrubOcxCompactionItems( compactBody, destinationDecodesNativeCompactionBlob(sendProvider), gatewayReplayParsed._stripReasoningEncryptedContent === true, ), { preserveRawReasoningContent: !isOpenAiOperatedResponsesDestination(sendProvider) && sendProvider.preserveResponsesReasoningContent === true, dropNullContentChannel: !isOpenAiOperatedResponsesDestination(sendProvider), stripEncryptedContent: gatewayReplayParsed._stripReasoningEncryptedContent === true || requireSendPlaintextReasoning, dropForeignItemId: gatewayReplayParsed._dropForeignReasoningItemIds === true, requirePlaintextReasoning: requireSendPlaintextReasoning, }) : compactBody; if (!isOpenAiOperatedResponsesDestination(sendProvider) && sendProvider.dropResponsesReasoningItems === true) { replayBody = dropResponsesReasoningInputItems(replayBody) as typeof replayBody; } if (!isOpenAiOperatedResponsesDestination(sendProvider) && sendProvider.preserveResponsesMessageMetadata !== true) { replayBody = stripInternalChatMessageMetadataPassthrough(replayBody) as typeof replayBody; } if (sendAuthCtx.kind === "pool" && isCanonicalOpenAiForwardProvider(sendProvider)) { replayBody = isolateCodexPoolClientMetadata(replayBody, sendAuthCtx.chatgptAccountId) as typeof replayBody; } // A queued key selection can refresh the destination; project only for this send. return isOpenAiOperatedResponsesDestination(sendProvider) ? repairUrlCitationAnnotations(replayBody) : stripCanonicalOnlyInputItems(stripCanonicalOnlyTopLevelFields(replayBody)); }; const doFetch = (upstreamRecovery?: UpstreamSendRecovery) => fetchWithHeaderTimeout( compactUrl, applyUpstreamRecoveryInit({ method: "POST", headers: wireHeaders, body: JSON.stringify({ ...buildCompactSendBody() as typeof compactBody, model: route.modelId }), }, upstreamRecovery), req.signal, // Keyed sends can refresh their destination after pacing. Their physical send below // owns the header clock; the outer wrapper owns pacing and client cancellation only. sendProvider.authMode !== "forward" && sendProvider.apiKey ? 0 : connectMs(), false, providerFetch(sendProvider, undefined, { providerName: route.providerName, modelId: route.modelId, accountAutoPause: (name, provider, headers) => captureAccountAutoPauseAttempt(name, provider, { config, codex: sendAuthCtx }, headers), beforeDispatch: isCanonicalOpenAiForwardProvider(sendProvider) ? createCodexAuthDispatchGuard(sendAuthCtx, config, selectedModelId, admission) : undefined, dispatchOverride: sendProvider.authMode !== "forward" && sendProvider.apiKey ? async (_input, init, execute, reacquirePacing) => { for (let attempt = 0; !providerApiKeySelectionIsCurrent(config, route.providerName, sendProvider); attempt++) { if (attempt >= 3) throw new Error("Provider key selection changed repeatedly before compact dispatch"); let current = resolveCurrentProviderApiKeyTransport(config, route.providerName, sendProvider); if (!current && sendProvider.requestPacing?.enabled && (sendProvider.requestPacing.keys || sendProvider.requestPacing.keySelection === "round-robin")) { const fresh = resolveCurrentProviderApiKeyTransport(config, route.providerName, { ...sendProvider, _apiKeyAttempt: undefined }); if (fresh) current = selectProactiveApiKeyTransport(config, route.providerName, fresh) ?? fresh; } if (!current?.apiKey || !supportsNativeResponsesCompactEndpoint(route.providerName, current)) { throw new Error("Provider key selection is no longer available for compact"); } const rebuilt = new Headers(sendHeaders); for (const name of Object.keys(sendProvider.headers ?? {})) rebuilt.delete(name); for (const [name, value] of Object.entries(current.headers ?? {})) rebuilt.set(name, value); rebuilt.set("authorization", `Bearer ${resolveProviderApiKey(current.apiKey)}`); const currentGateway = gatewayTransport ? buildResponsesTransport({ ...current, apiKey: resolveProviderApiKey(current.apiKey) }, { headers: req.headers }) : undefined; sendHeaders = wireHeaders = headers = currentGateway ? new Headers(currentGateway.headers) : rebuilt; sendProvider = compactProvider = current; compactUrl = currentGateway ? `${currentGateway.url.replace(/\/+$/, "")}/compact` : `${(current.baseUrl ?? "").replace(/\/+$/, "")}/responses/compact`; actualCompactHostKey = upstreamHostHealthKey(route.providerName, safeOriginLabel(compactUrl)); if (current.requestPacing?.enabled && (current.requestPacing.keys || current.requestPacing.keySelection === "round-robin")) { if (!reacquirePacing) throw new Error("Refreshed compact dispatch requires pacing readmission"); await reacquirePacing(current, route.modelId); } } if (req.signal.aborted) throw req.signal.reason; stampApiKeyAccountLabel(logCtx, route.providerName, sendProvider); if (!rebindPhysicalSend(sendBudget, { poolId: logCtx.spendPoolId ?? route.providerName, identityId: logCtx.accountLogLabel })) throw new SendBudgetExhaustedError(safeHostLabel(compactUrl)); noteProviderAttemptSend(logCtx, route.providerName, sendProvider, logCtx.usageLogInputTokens, upstreamRecovery); const physicalFetch = ((input, physicalInit) => sendWithConnectionPolicy( distributedKeyPhysicalFetch((sendProvider as { fetch?: typeof fetch }).fetch ?? execute, route.providerName, sendProvider, compactUrl), String(input), physicalInit, { providerName: route.providerName, provider: sendProvider })) as typeof fetch; return fetchWithHeaderTimeout(compactUrl, { ...init, headers: sendHeaders, body: JSON.stringify({ ...buildCompactSendBody() as typeof compactBody, model: route.modelId }) }, init.signal ?? req.signal, connectMs(), false, physicalFetch); } : undefined, ...(onPhysicalDispatch ? { onPhysicalDispatch } : {}), }), // Every credential-bearing forward send gets manual redirects, not only // pool sends: direct mode carries the caller's credential too (#914). sendProvider.authMode === "forward", ).then(res => { // Every real attempt response — including an intermediate 5xx the retry // wrapper replaces — proves the host was reached (#914 review). settleObservedCompactHostResponse(); return res; }); return recovery === "single" ? sendSingleCompactAttempt(doFetch) : fetchWithTransientRetry(doFetch, { abortSignal: req.signal, label: safeHostLabel(compactUrl), // Key-auth policy is a logical-request total; canonical forward keeps its default // ladder. Both intersect the same shared base remainder as ordinary Responses. attempts: sendBudget.remainingBaseSends(transientSendCapFor( transientRetryPolicyFor(sendProvider)?.attempts, sendBudget.used, )), onSendsConsumed: createPhysicalSendReporter(sendBudget, () => ({ poolId: logCtx.spendPoolId ?? route.providerName, identityId: logCtx.accountLogLabel })), }); }; // The account each outcome belongs to. Reassigned only when the alternate send below // actually happens, so every recorder call names the context that produced it. let outcomeCtx = authCtx; const localDispatchRefusal = (error: unknown): Response | undefined => { const cause = unwrapUpstreamRetryEvidenceError(error); const response = requestPacingOverloadResponse(cause) ?? spendLedgerStorageErrorResponse(error, logCtx) ?? (cause instanceof SendBudgetExhaustedError ? unboundPoolSpendRefusalResponse(logCtx) ?? formatErrorResponse(429, "request_send_budget_exhausted", cause.message) : mapCodexAuthContextErrorToResponse(cause, { now: Date.now(), accountSelector: route.codexAccountNamespace, })); if (response) { releaseUpstreamHostAdmission(compactHostAdmissionLease); compactHostAdmissionLease = null; releaseCodexAuthContextProbeLease(outcomeCtx); } return response; }; let upstream: Response; let storedPool401ReplayAttempted = false; try { // Same connect timeout + keep-alive reset + transient-5xx recovery as /v1/responses — // compact hits the same ChatGPT host and must soft-avoid / clear affinity (#186). upstream = await sendCompactAttempt(compactProvider, headers, "normal", authCtx); } catch (err) { if (req.signal.aborted) { recordCompactPoolOutcome(outcomeCtx, 499); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } const localRefusal = localDispatchRefusal(err); if (localRefusal) return localRefusal; const outcome = classifyTransportFailureKind(err); // Host-level evidence stands regardless of pool membership (#914 review). if (outcome === "connect_neutral") { if (compactHostCircuitEnabled) { recordUpstreamHostFailure(actualCompactHostKey, { code: transportErrorCode(err), threshold: config.upstreamHostCircuitThreshold, lease: compactHostAdmissionLease, }); } else { recordUpstreamHostFailure(actualCompactHostKey, { code: transportErrorCode(err) }); } } else { releaseUpstreamHostAdmission(compactHostAdmissionLease); } compactHostAdmissionLease = null; recordCompactPoolOutcome(outcomeCtx, outcome); return formatErrorResponse(502, "upstream_error", "Failed to connect to compact upstream"); } recordDistributedResponseFailure(upstream, config, route.providerName, req.signal); if ( upstream.status === 401 && !wasPermanentAccountRefusal(upstream) && (authCtx.kind === "main-pool" || authCtx.kind === "pool") && !gatewayTransport && usesCodexForwardPoolAuth(authCtx, compactProvider) && !req.signal.aborted ) { await upstream.body?.cancel().catch(() => undefined); const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined; storedPool401ReplayAttempted = poolAuthCtx !== undefined; const poolReplay = poolAuthCtx ? await refreshPoolCompactContext({ req, admission, config, modelId: selectedModelId, authCtx: poolAuthCtx, provider: compactProvider, codexAccountMode: route.codexAccountMode, codexAccountNamespace: route.codexAccountNamespace, substituteMainCredential, options, }) : undefined; const replay = poolReplay ?? await refreshNativeMainCompactContext({ req, admission, config, modelId: selectedModelId, authCtx, provider: compactProvider, codexAccountMode: route.codexAccountMode, substituteMainCredential, options, }); if (!replay.ok) { // A transient refresh failure must not retire the account; only a dead grant // does, and only while the rejected credential is still the stored one (#2887). if (poolAuthCtx) { if (poolReplay && !poolReplay.ok && poolReplay.quarantine) { recordCodexUpstreamOutcome(config, poolAuthCtx.accountId, 401, { threadId: poolAuthCtx.affinityKey, fixedAccount: poolAuthCtx.fixedAccount, modelId: selectedModelId, writerGeneration: poolAuthCtx.writerGeneration, credentialGeneration: poolReplay.quarantineGeneration ?? poolAuthCtx.generation, }); } return replay.response; } recordCompactPoolOutcome(outcomeCtx, replay.response.status === 401 ? 401 : "connect_neutral"); return replay.response; } authCtx = replay.authCtx; outcomeCtx = replay.authCtx; compactProvider = replay.provider; headers = replay.headers; logCtx.accountLogLabel = codexAuthContextLogLabel(replay.authCtx, config); try { upstream = await sendCompactAttempt(compactProvider, headers, "single", authCtx); } catch (err) { if (req.signal.aborted) { recordCompactPoolOutcome(outcomeCtx, 499); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } const localRefusal = localDispatchRefusal(err); if (localRefusal) return localRefusal; recordCompactPoolOutcome(outcomeCtx, classifyTransportFailureKind(err)); return formatErrorResponse(502, "upstream_error", "Failed to connect to compact upstream"); } } // The settle for an alternate send that threw, shared by the quota and transient moves. By the // time it runs the request already names the alternate (outcomeCtx), so health goes to it. const failedAlternateCompactSend = (err: unknown): Response => { if (req.signal.aborted) { recordCompactPoolOutcome(outcomeCtx, 499); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } const localRefusal = localDispatchRefusal(err); if (localRefusal) return localRefusal; const outcome = classifyTransportFailureKind(err); // Host-level evidence stands regardless of pool membership (#914 review). if (outcome === "connect_neutral") { if (compactHostCircuitEnabled) { recordUpstreamHostFailure(actualCompactHostKey, { code: transportErrorCode(err), threshold: config.upstreamHostCircuitThreshold, lease: compactHostAdmissionLease, }); } else { recordUpstreamHostFailure(actualCompactHostKey, { code: transportErrorCode(err) }); } } else { releaseUpstreamHostAdmission(compactHostAdmissionLease); } compactHostAdmissionLease = null; recordCompactPoolOutcome(outcomeCtx, outcome); return formatErrorResponse(502, "upstream_error", "Failed to connect to compact upstream"); }; const scrubCompactStateForAlternate = (alternateAuthCtx: CodexAuthContext, alternateHeaders: Headers): void => { const binding = conversationStateBindingFromAuth( alternateAuthCtx, (authCtx.kind === "pool" || authCtx.kind === "main-pool") ? authCtx.affinityKey : codexPoolAffinityKey(req.headers), ); if (!binding) return; for (const body of [raw, compactBody]) { applyAccountChangeConversationStateScrub({ body, headers: alternateHeaders, bindingKey: binding.bindingKey, servingAccountId: binding.accountId, priorAccountId: authCtx.accountId, logCtx, }); } }; // Bounded same-request alternate: the regular /v1/responses path already does this // (core.ts:319-423) and recognizes exactly 429/402. Without it a pool rejection // surfaces to the client, which retries the compact task OUTSIDE the logical request // — reporting exhausted retries while another pool account sat idle (#913). if ( (upstream.status === 429 || upstream.status === 402 || wasAccountAutoPaused(upstream)) // A replay refusal this proxy synthesized carries 429 for the client's benefit only. // It is not pool quota evidence, and the alternate account below is another send of a // compact turn that may already have been processed. && !isNonReplayableResponse(upstream) && (!storedPool401ReplayAttempted || wasAccountAutoPaused(upstream)) && !gatewayTransport && usesCodexForwardPoolAuth(authCtx, route.provider) && !authCtx.fixedAccount && route.codexAccountMode && !req.signal.aborted ) { const firstRetryAfter = upstream.headers.get("retry-after"); const firstResetAt = [ upstream.headers.get("x-codex-primary-reset-at"), upstream.headers.get("x-codex-secondary-reset-at"), upstream.headers.get("x-codex-tertiary-reset-at"), ].filter(Boolean); // Build the alternate COMPLETELY before cancelling the first body: if construction // throws, the first rejection is still intact and can be returned to the client. // The same reasoning refuses an uploaded-file move here: no alternate can read a file the // issuing account received, so which one is chosen is irrelevant and asking before the // resolution costs nothing. It also reuses the guarantee the comment above depends on -- // the first body is still uncancelled -- so the client gets the original rejection rather // than an inaccessible-file error from account B (#4710). const alternate = conversationCarriesUploadedFiles(raw) ? undefined : await resolveAlternateCompactContext({ req, admission, config, route, selectedModelId, excludeAccountId: authCtx.accountId, turnAdmissionLease, }); if (alternate && "failure" in alternate) { await applyCompactPoolQuotaHeaders(authCtx, upstream); recordCompactPoolOutcome(authCtx, upstream.status, { retryAfter: firstRetryAfter, resetAt: firstResetAt }); await upstream.body?.cancel().catch(() => undefined); return alternate.failure; } // Resolution can await a credential refresh, so the client may have gone away // while we were choosing B. Re-check before spending anything: recording A, // cancelling its body, and sending B are all observable side effects, and B's // quota is not ours to spend on a request nobody is waiting for. 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"); } // 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 // reading numbers from before the account ran out. await applyCompactPoolQuotaHeaders(authCtx, upstream); recordCompactPoolOutcome(authCtx, upstream.status, { retryAfter: firstRetryAfter, resetAt: firstResetAt, ...(alternate.authCtx.accountId ? { promoteAccountId: alternate.authCtx.accountId } : {}), }); await upstream.body?.cancel().catch(() => undefined); outcomeCtx = alternate.authCtx; logCtx.accountLogLabel = codexAuthContextLogLabel(alternate.authCtx, config); scrubCompactStateForAlternate(alternate.authCtx, alternate.headers); try { upstream = await sendCompactAttempt(alternate.provider, alternate.headers, "single", alternate.authCtx); } catch (err) { return failedAlternateCompactSend(err); } } } // A transient 5xx that survived the same-account ladder above (#6969). The regular path moves // it to another Pool account; compact used to move only on 429/402, so the client retried the // compact task outside this request while another account sat idle. One bounded alternate send, // the same guards as the quota move, and a 5xx whose body confirms quota keeps today's path. if ( shouldRetryCodexPoolAccountTransient(upstream) && (!storedPool401ReplayAttempted || wasAccountAutoPaused(upstream)) && !gatewayTransport && usesCodexForwardPoolAuth(authCtx, route.provider) && !authCtx.fixedAccount && route.codexAccountMode && outcomeCtx === authCtx && !req.signal.aborted && !conversationCarriesUploadedFiles(raw) && !await shouldRetryCodexPoolAccountQuota(upstream, req.signal) ) { const firstStatus = upstream.status; const firstRetryAfter = upstream.headers.get("retry-after"); const firstResetAt = [ upstream.headers.get("x-codex-primary-reset-at"), upstream.headers.get("x-codex-secondary-reset-at"), upstream.headers.get("x-codex-tertiary-reset-at"), ].filter(Boolean); const alternate = await resolveAlternateCompactContext({ req, admission, config, route, selectedModelId, excludeAccountId: authCtx.accountId, turnAdmissionLease, }); if (alternate && "failure" in alternate) { await applyCompactPoolQuotaHeaders(authCtx, upstream); recordCompactPoolOutcome(authCtx, firstStatus, { retryAfter: firstRetryAfter, resetAt: firstResetAt }); await upstream.body?.cancel().catch(() => undefined); return alternate.failure; } 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) { const firstLabel = logCtx.accountLogLabel; // B's identity before B's admission, so every budget and spend check runs against B. A's // response stays intact until B is actually sent. logCtx.accountLogLabel = codexAuthContextLogLabel(alternate.authCtx, config); scrubCompactStateForAlternate(alternate.authCtx, alternate.headers); const recordFirstMove = () => { recordCompactPoolOutcome(authCtx, firstStatus, { retryAfter: firstRetryAfter, resetAt: firstResetAt }); void upstream.body?.cancel().catch(() => undefined); outcomeCtx = alternate.authCtx; }; let alternateResponse: Response | undefined; // Whether B's send actually left this process. Every local refusal before that point -- a // spent budget, a dispatch guard such as disabled credits, a pacing refusal -- keeps A. let alternateDispatched = false; try { alternateResponse = await sendCompactAttempt(alternate.provider, alternate.headers, "single", alternate.authCtx, () => { alternateDispatched = true; }); } catch (err) { if (alternateDispatched) { recordFirstMove(); return failedAlternateCompactSend(err); } // Refused before anything left: A's answer and identity stand, recorded once below. logCtx.accountLogLabel = firstLabel; releaseCodexAuthContextProbeLease(alternate.authCtx); } if (alternateResponse) { recordFirstMove(); upstream = alternateResponse; } } } // Capture the final serving account as well as an earlier rejected account, once per response. if (outcomeCtx.kind === "pool") { const { applyAccountQuotaFromUpstreamHeaders } = await import("../../codex/quota"); applyAccountQuotaFromUpstreamHeaders(outcomeCtx.accountId, upstream.headers, outcomeCtx.writerGeneration, undefined, { modelId: route.modelId, poolWriter: outcomeCtx.poolQuotaWriter, poolResponse: true }); } const retryAfter = upstream.headers.get("retry-after"); const resetAt = [ upstream.headers.get("x-codex-primary-reset-at"), upstream.headers.get("x-codex-secondary-reset-at"), upstream.headers.get("x-codex-tertiary-reset-at"), ].filter(Boolean); const buffered = await bufferCompactResponse(upstream, req.signal, config.stallTimeoutSec, isLocalUpstream(compactUrl)); const bufferedErrorText = buffered.ok ? "" : await buffered.clone().text().catch(() => ""); // The client-facing 429 of a synthesized replay refusal says nothing about this // account's quota. Pool accounting keeps reading it as the transport failure it is, // which is also what it recorded before the status was corrected for the client. const replayRefused = isNonReplayableResponse(upstream); const explicitQuotaStatus = !replayRefused && (buffered.status === 429 || buffered.status === 402); const bodyInferredQuota = !replayRefused && !buffered.ok && !explicitQuotaStatus && isRateLimitOrQuotaFailureMessage(bufferedErrorText); const quotaFailure = explicitQuotaStatus || bodyInferredQuota; // Record pool health only after the body is fully delivered (or definitively failed). // A premature 200 would clear soft-avoid while the client still sees a buffer 502. if (buffered.status === 499) { recordCompactPoolOutcome(outcomeCtx, 499); return buffered; } // A body-confirmed quota failure can arrive behind a generic 5xx. Record it as // quota evidence; otherwise preserve the real upstream status so a local buffering // failure after a 200 cannot soft-avoid a healthy account or rotate a thread. recordCompactPoolOutcome( outcomeCtx, bodyInferredQuota ? 429 : replayRefused ? 502 : upstream.status, { retryAfter, resetAt }, ); // Lift usage and response metadata from the buffered upstream JSON into the // request log; the routed branch gets the same through handleResponses. The // synthetic buffer errors are not upstream bodies and stay uninspected. if (buffered.ok) { const compactText = await buffered.clone().text(); if (gatewayReplayParsed) { try { if (nativeCompactResponseSchema.safeParse(JSON.parse(compactText)).success) { commitReasoningReplayServingIdentity(gatewayReplayParsed._reasoningReplayScope); } } catch { /* A malformed body cannot establish serving provenance. */ } } inspectResponseLogJson(logCtx, compactText); if (!options.compactionRoutingOverride) forgetCompactHandoffRoute(req, admission); rememberServingConversationStateIssuer(outcomeCtx, codexPoolAffinityKey(req.headers)); } else if (!options.compactionRoutingOverride && quotaFailure && (!storedPool401ReplayAttempted || wasAccountAutoPaused(upstream))) { const fallbackModel = compactHandoffRoute(req, admission, raw.model); if (fallbackModel && !req.signal.aborted) { const fallbackReq = new Request(req.url, { method: "POST", headers: req.headers, body: JSON.stringify({ ...raw, model: fallbackModel }), signal: req.signal, }); try { const fallback = await handleResponsesCompact( fallbackReq, config, logCtx, turnAdmissionLease, admission, // The handoff child is the same logical compact on a second model, so it inherits // the holder. Forwarding `options` alone was not enough: the child minted its own. { ...options, sendBudget }, ); if (fallback.ok || fallback.status === 499) return fallback; await fallback.body?.cancel().catch(() => undefined); } catch { // The previous-model rejection is the authoritative failure when the // remembered handoff route can no longer compact this thread. } } } // A native compact 404 falls back to a regular Responses compaction turn. if (buffered.status !== 404) return buffered; } finally { releaseUpstreamHostAdmission(compactHostAdmissionLease); releaseCodexAuthContextProbeLease(authCtx); } } // ROUTED model: run the v2 synthetic-compaction turn internally (appends COMPACT_PROMPT, no // tools) and decode the resulting ocx1 envelope into plain v1 replacement-history items. const rawInputItems = Array.isArray(raw.input) ? (raw.input as unknown[]) : []; // Keep readable reasoning local until success, outside the summarizer request. // Opaque compaction responses below keep their original protocol and ciphertext. const retention: ReasoningRetentionResult = config.reasoningRetention?.enabled === true && !isCanonicalOpenAiForwardProvider(route.provider) ? applyReasoningRetention(rawInputItems, { archiveDir: join(getConfigDir(), "reasoning-archive"), contextWindow: route.staticPolicy.model.contextWindow ?? route.staticPolicy.model.maxInputTokens, maxContextPercent: config.reasoningRetention?.maxContextPercent, maxTokens: config.reasoningRetention?.maxTokens, }) : { input: rawInputItems }; let keepArchive = false; try { const inputItems = retention.input; const internalBody = { ...raw, // Canonical ChatGPT Responses rejects non-streaming turns. Daybreak cannot use the // native compact endpoint either, so run its synthetic compaction as SSE and collapse // the completed event back into the v1 compact JSON contract below. Combo-dispatched // turns also go out as SSE: failover can land on a canonical child that rejects a // non-streaming turn, and every combo-capable provider already serves streaming traffic. stream: isCanonicalOpenAiForwardProvider(route.provider) || accountGatedCompactWireModel || route.combo ? true : false, input: [...inputItems, { type: "compaction_trigger" }], }; const internalHeaders = new Headers({ "content-type": "application/json" }); for (const name of FORWARD_HEADERS) { const value = req.headers.get(name); if (value || (value !== null && (name === "x-codex-installation-id" || name === "x-codex-turn-metadata"))) internalHeaders.set(name, value); } copyResponsesClientMetadata(internalHeaders, req.headers, route.provider); const internalReq = new Request("http://localhost/v1/responses", { method: "POST", headers: internalHeaders, body: JSON.stringify(internalBody), }); linkRequestSessionLane(req, internalReq); // The routed compaction turn is a handoff inside the same logical request, so it draws the // REMAINDER. Minting here is what let a native attempt spend three sends and the routed // fallback spend four more. // internalReq keeps only FORWARD_HEADERS, so the caller principal is resolved from req. const replayPrincipal = { clientPrincipalId: resolveContextPrincipal(req, config, admission) }; const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, sendBudget, compactionRecoveryKind: "compaction-v1", compactionRoutingOverride: options.compactionRoutingOverride, advisorAdmissionPolicy: options.advisorAdmissionPolicy, replayPrincipal, ...(admission ? { admission } : {}) }); if (req.signal.aborted) { void response.body?.cancel(req.signal.reason).catch(() => undefined); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } if (!response.ok) return response; let json: { output?: unknown[]; status?: unknown; error?: unknown }; if (hasResponseMediaType(response.headers, "text/event-stream")) { if (!response.body) { return formatErrorResponse(502, "server_error", "compaction turn returned an empty event stream"); } const terminal = { status: "incomplete" as "completed" | "failed" | "incomplete" }; let completed: { id?: unknown; output?: unknown; status?: unknown } | undefined; await new Promise(resolve => { consumeForInspection( response.body!, status => { terminal.status = status; }, req.signal, resolve, undefined, undefined, value => { completed = value; }, ); }); if (req.signal.aborted) { return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } if (terminal.status !== "completed" || !completed) { return formatErrorResponse(502, "upstream_error", `compaction turn did not complete (status: ${terminal.status})`); } json = completed as { output?: unknown[]; status?: unknown; error?: unknown }; } else { try { json = await response.json() as { output?: unknown[]; status?: unknown; error?: unknown }; } catch { return formatErrorResponse(502, "server_error", "compaction turn returned a non-JSON response"); } } if (req.signal.aborted) { return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } // The internal turn answers 200 even when it failed or was truncated, so the body // has to be inspected. Reporting a failure beats installing "(no summary // available)" as replacement history and silently losing the conversation (#422). if (json.error) { const message = typeof json.error === "string" ? json.error : (json.error as { message?: unknown })?.message; return formatErrorResponse(502, "upstream_error", typeof message === "string" ? message : "compaction turn failed"); } if (json.status !== "completed") { return formatErrorResponse( 502, "upstream_error", `compaction turn did not complete (status: ${String(json.status ?? "unknown")})`, ); } const compactionItems = (json.output ?? []).filter( (item): item is { type: string; encrypted_content?: string } => !!item && typeof item === "object" && (item as { type?: string }).type === "compaction", ); if (compactionItems.length !== 1) { return formatErrorResponse( 502, "invalid_response_error", `compaction turn produced ${compactionItems.length} compaction items, expected exactly 1`, ); } // Native Responses backends return a real opaque OpenAI-encrypted compaction item. OCX cannot // and should not decrypt it; preserve that item for /responses/compact callers. Synthetic // routed summaries are our `ocx1:` envelope and must be decoded into v1 history items. if (typeof compactionItems[0]!.encrypted_content === "string" && compactionItems[0]!.encrypted_content.trim().length > 0 && !compactionItems[0]!.encrypted_content.startsWith("ocx1:")) { // Failover may reach native compaction after readable reasoning was withheld locally. // The opaque item cannot contain that text, so publish it (or its archive notice) alongside. const output = [...compactionItems, ...(retention.retainedReasoning ?? [])]; if (retention.archived) output.push({ type: "message", role: "user", content: [{ type: "input_text", text: appendRetentionNotice("", retention.archived).trimStart() }] }); const result = new Response(JSON.stringify({ output }), { headers: { "Content-Type": "application/json" }, }); if (!options.compactionRoutingOverride) rememberCompactHandoffRoute(req, admission, raw.model); keepArchive = true; return result; } const encrypted = compactionItems[0]!.encrypted_content; const decoded = typeof encrypted === "string" ? decodeCompactionSummary(encrypted) : null; // An empty `ocx1:` envelope decodes to "" rather than null, so length is what matters. if (decoded === null || decoded.trim().length === 0) { return formatErrorResponse(502, "invalid_response_error", "compaction turn produced an empty summary"); } const summary = appendRetentionNotice(decoded, retention.archived); const output = [...buildCompactV1Output(extractCompactUserMessages(inputItems), summary), ...(retention.retainedReasoning ?? [])]; if (!options.compactionRoutingOverride) rememberCompactHandoffRoute(req, admission, raw.model); const result = new Response(JSON.stringify({ output }), { headers: { "Content-Type": "application/json" } }); keepArchive = true; return result; } finally { // Keep an archive only after successful replacement history publishes its path. if (!keepArchive && retention.archived) rmSync(retention.archived.path, { force: true }); } }