From 199382912964810cb16dc3995f3ad76b831faaf0 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 5 Aug 2026 10:04:29 +0900 Subject: [PATCH] feat(codex): preserve opt-in upstream host circuit --- .../ja/reference/configuration/providers.md | 3 +- .../ko/reference/configuration/providers.md | 3 +- .../docs/reference/configuration/providers.md | 3 +- .../ru/reference/configuration/providers.md | 3 +- .../reference/configuration/providers.md | 3 +- src/codex/upstream-host-health.ts | 320 ++++++++++++++++-- src/server/responses/compact.ts | 122 ++++++- src/server/responses/core.ts | 132 +++++++- src/types.ts | 5 + tests/responses-compaction-routing.test.ts | 58 ++++ tests/upstream-host-circuit.test.ts | 174 ++++++++++ 11 files changed, 769 insertions(+), 57 deletions(-) create mode 100644 tests/upstream-host-circuit.test.ts diff --git a/docs-site/src/content/docs/ja/reference/configuration/providers.md b/docs-site/src/content/docs/ja/reference/configuration/providers.md index 19e07d53d..d585aa496 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ja/reference/configuration/providers.md @@ -21,7 +21,8 @@ description: プロバイダー エントリ、認証、エンドポイント、 | `autoSwitchThreshold?` | `number` | `80` | 使用量ベースのプロアクティブ切り替えしきい値。`quota` は紐付け済み/未紐付けタスクの次のリクエストを再評価でき、`fill-first` は未紐付け割り当ての使い切り基準としてのみ使用し、通常の `round-robin` 選択は使用しません。既知の 5 時間、週次、30 日 quota window の最大スコアを使います。`0` は使用量ベースの切り替えだけを無効にし、未紐付け割り当てや障害回復は無効にしません。 | | `accountPoolStrategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | 新規/未紐付け Codex リクエストの割り当て戦略。live な `(parent thread id, quota scope)` affinity がなければ未紐付けで、プロキシ再起動や affinity リセット後は既存の表示タスクも未紐付けになり得ます。`quota` はアクティブアカウントがなければ既知 usage 最小の適格アカウントを選び、適格なアクティブアカウントが `autoSwitchThreshold` 未満なら維持します。しきい値到達後は、未紐付けリクエストまたは紐付け済みタスクの次のリクエストを usage の低い適格アカウントへ移せます。`round-robin` は未紐付けリクエストを均等分散し、`fill-first` は cooldown、使用不可、または drain threshold までアクティブアカウントへ割り当てます。 | | `accountPoolStickyLimit?` | `number` | `1` | 1 回の round-robin 選択で次へ進む前に保持する新規/未紐付けタスク割り当て数。カウンターは上流の成功後ではなくタスクの紐付け時に増えます。範囲 1–100。`accountPoolStrategy` が `round-robin` のときのみ。 | -| `upstreamFailoverThreshold?` | `number` | `3` |今後の新しいセッションがフェイルオーバーする前に一時的なエラーが連続して発生する。 `0` を無効に設定します。実証済みの接続前DNS/TCP到達不能障害はprovider-host単位で記録され、アカウントの健全性、クールダウン、スレッド/セッションの親和性、アクティブアカウントの選択、Poolルーティングには影響せず、この閾値にもカウントされません。 | +| `upstreamFailoverThreshold?` | `number` | `3` |今後の新しいセッションがフェイルオーバーする前に一時的なエラーが連続して発生する。 `0` を無効に設定します。通常のResponses送信とネイティブcompact送信では、実証済みの接続前DNS/TCP到達不能障害はprovider-host単位で記録され、アカウントの健全性、クールダウン、スレッド/セッションの親和性、アクティブアカウントの選択、Poolルーティングには影響せず、この閾値にもカウントされません。 | +| `upstreamHostCircuitThreshold?` | `number` | `0` | ネイティブOpenAI forwardのResponses送信とcompact送信で、実証済みの接続前DNS/TCP障害に適用するオプトインのサーキットしきい値です。`0`で無効、`1`〜`20`ではその回数の終端論理リクエストが失敗するとprovider-originを30秒間遮断します。遮断中はアカウント選択やupstream送信の前に`Retry-After`付き`503`を返し、時間経過後はhalf-openリクエストを1件だけ許可します。タイムアウトとHTTP応答は数えず、HTTP応答が1件でもあれば回路を閉じます。 | | `modelCacheTtlMs?` | `number` | `300000` |プロバイダーごとの `/models` キャッシュの鮮度ウィンドウ。 | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Anthropic プロンプト キャッシュ ポリシー: 無効、5 分間の一時的、または 1 時間の延長。 | | `tokenGuardian?` | `OcxTokenGuardianConfig` |オフ |オプションのプロアクティブな OAuth 更新および Codex アカウントのウォームアップ ポリシー。 | diff --git a/docs-site/src/content/docs/ko/reference/configuration/providers.md b/docs-site/src/content/docs/ko/reference/configuration/providers.md index 11ed1e2e7..a17a5ffd2 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ko/reference/configuration/providers.md @@ -21,7 +21,8 @@ description: 공급자 항목, 인증, 엔드포인트, 모델 카탈로그, 할 | `autoSwitchThreshold?` | `number` | `80` | 사용량 기반 선제 전환 임계값입니다. `quota`는 바인딩된 작업과 바인딩 없는 작업의 다음 요청을 모두 재평가할 수 있고, `fill-first`는 바인딩 없는 작업 배정의 소진 기준으로만 사용하며, 기본 `round-robin` 선택은 이 값을 사용하지 않습니다. 알려진 5시간, 주간, 30일 quota window 중 가장 높은 점수를 씁니다. `0`은 사용량 기반 전환만 끄며 바인딩 없는 작업 배정이나 실패 복구는 끄지 않습니다. | | `accountPoolStrategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | 새 작업/바인딩 없는 Codex 요청의 계정 배정 전략입니다. `(parent thread id, quota scope)`의 live affinity가 없으면 바인딩 없는 요청이며, 프록시 재시작이나 affinity 초기화 뒤에는 기존에 보이던 작업도 바인딩이 없어질 수 있습니다. `quota`는 활성 계정이 없을 때 알려진 usage가 가장 낮은 적격 계정을 선택하고, 적격 활성 계정이 `autoSwitchThreshold` 미만이면 유지합니다. 임계값 도달 뒤에는 바인딩 없는 요청이나 바인딩된 작업의 다음 요청을 usage가 더 낮은 적격 계정으로 옮길 수 있습니다. `round-robin`은 바인딩 없는 요청을 균등 분배하고, `fill-first`는 cooldown, 사용 불가 또는 drain threshold까지 활성 계정에 배정합니다. | | `accountPoolStickyLimit?` | `number` | `1` | 한 round-robin 선택이 다음으로 넘어가기 전에 유지하는 새 작업/바인딩 없는 작업 배정 수입니다. 카운터는 업스트림 성공 뒤가 아니라 작업을 바인딩할 때 증가합니다. 범위 1–100이며 `accountPoolStrategy`가 `round-robin`일 때만 적용됩니다. | -| `upstreamFailoverThreshold?` | `number` | `3` | 연속된 일시적 실패가 이 횟수에 도달하면 이후 새 세션은 failover됩니다. `0`으로 두면 비활성화됩니다. 입증된 연결 전 DNS/TCP 도달 불가 실패는 provider-host 범위로 기록되며 계정 상태, 쿨다운, 스레드/세션 선호도, 활성 계정 선택 또는 Pool 라우팅에 영향을 주지 않고 이 임계값에도 집계되지 않습니다. | +| `upstreamFailoverThreshold?` | `number` | `3` | 연속된 일시적 실패가 이 횟수에 도달하면 이후 새 세션은 failover됩니다. `0`으로 두면 비활성화됩니다. 일반 Responses와 네이티브 compact 전송에서 입증된 연결 전 DNS/TCP 도달 불가 실패는 provider-host 범위로 기록되며 계정 상태, 쿨다운, 스레드/세션 선호도, 활성 계정 선택 또는 Pool 라우팅에 영향을 주지 않고 이 임계값에도 집계되지 않습니다. | +| `upstreamHostCircuitThreshold?` | `number` | `0` | 네이티브 OpenAI forward Responses와 compact 전송에서 입증된 연결 전 DNS/TCP 실패에 적용하는 선택적 회로 차단 임계값입니다. `0`은 비활성화하며, `1`~`20`은 이 횟수만큼 최종 논리 요청이 실패하면 provider-origin을 30초 동안 차단합니다. 차단 중에는 계정 선택이나 업스트림 전송 전에 `Retry-After`가 포함된 `503`을 반환하고, 시간이 지나면 반개방 요청 하나만 허용합니다. 타임아웃과 HTTP 응답은 집계하지 않으며, HTTP 응답이 하나라도 오면 회로를 닫습니다. | | `modelCacheTtlMs?` | `number` | `300000` | 공급자별 `/models` 캐시의 최신성 창입니다. | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Anthropic 프롬프트 캐시 정책입니다. 비활성, 5분짜리 임시, 1시간짜리 확장 중 하나입니다. | | `tokenGuardian?` | `OcxTokenGuardianConfig` | 꺼짐 | 선택적 선제 OAuth 갱신과 Codex 계정 워밍업 정책입니다. | diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 70962e4c3..a866e6c60 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -22,7 +22,8 @@ authenticated. | `autoSwitchThreshold?` | `number` | `80` | Usage threshold for proactive switching. `quota` can re-evaluate both bound and unbound tasks on their next request; `fill-first` uses it only as the drain point for unbound assignment; normal `round-robin` selection does not use it. The score uses the hottest known 5h, weekly, or 30d quota window. `0` disables usage-based proactive switching only, not unbound assignment or failure recovery. | | `accountPoolStrategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | Assignment strategy for new/unbound Codex requests. A request is unbound when it has no live (parent thread id, quota scope) affinity; a visible existing task can become unbound after proxy restart or affinity reset. `quota` picks the lowest-usage eligible account when no active account exists, keeps an eligible active account below `autoSwitchThreshold`, and after the threshold may move an unbound request or proactively rebind a bound task to a lower-usage eligible account. `round-robin` distributes unbound requests evenly; `fill-first` keeps assigning unbound requests to the active account until cooldown, unavailability, or the configured drain threshold. | | `accountPoolStickyLimit?` | `number` | `1` | New/unbound task assignments retained on one round-robin selection before advancing; the counter advances when a task is bound, not after an upstream success. Range 1–100. | -| `upstreamFailoverThreshold?` | `number` | `3` | Consecutive transient failures before future new sessions fail over. Set `0` to disable. Proven pre-connection DNS/TCP reachability failures are tracked at the provider-host level: they never affect account health, cooldowns, thread/session affinity, active-account selection, or Pool routing, and never count toward this threshold. | +| `upstreamFailoverThreshold?` | `number` | `3` | Consecutive transient failures before future new sessions fail over. Set `0` to disable. For regular Responses and native compact sends, proven pre-connection DNS/TCP reachability failures are tracked at the provider-host level: they never affect account health, cooldowns, thread/session affinity, active-account selection, or Pool routing, and never count toward this threshold. | +| `upstreamHostCircuitThreshold?` | `number` | `0` | Opt-in circuit threshold for proven pre-connection DNS/TCP failures on native OpenAI forward Responses and compact sends. `0` disables it; `1`–`20` opens a 30-second provider-origin cooldown after that many terminal logical requests. While open, requests receive `503` with `Retry-After` before account selection or upstream send; after cooldown, one half-open request is admitted. Timeouts and HTTP responses never count, and any HTTP response closes the circuit. | | `modelCacheTtlMs?` | `number` | `300000` | Freshness window for the per-provider `/models` cache. | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Anthropic prompt-cache policy: disabled, 5-minute ephemeral, or 1-hour extended. | | `tokenGuardian?` | `OcxTokenGuardianConfig` | off | Optional proactive OAuth refresh and Codex-account warmup policy. | diff --git a/docs-site/src/content/docs/ru/reference/configuration/providers.md b/docs-site/src/content/docs/ru/reference/configuration/providers.md index 7dc87358a..db743e3d7 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -22,7 +22,8 @@ description: Записи провайдеров, аутентификация, | `autoSwitchThreshold?` | `number` | `80` | Порог проактивного переключения по использованию. `quota` может повторно оценить следующий запрос как привязанной, так и непривязанной задачи; `fill-first` использует его только как точку исчерпания для непривязанных назначений; обычный `round-robin` его не использует. Оценка берёт самое горячее из окон 5 часов, недели и 30 дней. `0` отключает только переключение по использованию, но не назначение непривязанных задач и не восстановление после сбоев. | | `accountPoolStrategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | Стратегия назначения для новых/непривязанных запросов Codex. Запрос непривязан, если у него нет live affinity `(parent thread id, quota scope)`; видимая существующая задача может стать непривязанной после перезапуска прокси или сброса affinity. `quota` выбирает подходящий аккаунт с наименьшим известным usage, когда активного аккаунта нет, сохраняет подходящий активный аккаунт ниже `autoSwitchThreshold`, а после порога может перевести непривязанный запрос или следующий запрос привязанной задачи на подходящий аккаунт с меньшим usage. `round-robin` равномерно распределяет непривязанные запросы; `fill-first` назначает их активному аккаунту до cooldown, недоступности или порога исчерпания. | | `accountPoolStickyLimit?` | `number` | `1` | Число назначений новых/непривязанных задач на одном выборе round-robin перед переходом дальше. Счётчик растёт при привязке задачи, а не после успеха upstream. Диапазон 1–100; только при `accountPoolStrategy` = `round-robin`. | -| `upstreamFailoverThreshold?` | `number` | `3` | Сколько подряд transient failure допустить, прежде чем новые сессии начнут делать failover. `0` отключает эту логику. Доказанные ошибки доступности DNS/TCP до соединения учитываются на уровне пары «провайдер, хост» и не влияют на здоровье аккаунта, кулдауны, привязку потока/сессии, выбор активного аккаунта или маршрутизацию пула, а также не учитываются в этом пороге. | +| `upstreamFailoverThreshold?` | `number` | `3` | Сколько подряд transient failure допустить, прежде чем новые сессии начнут делать failover. `0` отключает эту логику. Для обычных Responses-запросов и нативных compact-отправок доказанные ошибки доступности DNS/TCP до соединения учитываются на уровне пары «провайдер, хост» и не влияют на здоровье аккаунта, кулдауны, привязку потока/сессии, выбор активного аккаунта или маршрутизацию пула, а также не учитываются в этом пороге. | +| `upstreamHostCircuitThreshold?` | `number` | `0` | Опциональный порог circuit breaker для доказанных DNS/TCP-сбоев до соединения в нативных OpenAI forward Responses- и compact-отправках. `0` отключает его; `1`–`20` открывает 30-секундный cooldown для provider-origin после такого числа завершившихся логических запросов. Пока circuit открыт, до выбора аккаунта и upstream-отправки возвращается `503` с `Retry-After`; после cooldown допускается один half-open запрос. Таймауты и HTTP-ответы не учитываются, а любой HTTP-ответ закрывает circuit. | | `modelCacheTtlMs?` | `number` | `300000` | Окно свежести для кэша `/models` на уровне провайдера. | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Политика prompt-cache Anthropic: отключено, 5-минутный ephemeral или 1-часовой extended. | | `tokenGuardian?` | `OcxTokenGuardianConfig` | off | Необязательная политика proactive OAuth refresh и warmup'а аккаунтов Codex. | diff --git a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md index 450db5437..7fc83478e 100644 --- a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md +++ b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md @@ -21,7 +21,8 @@ description: 提供者条目、身份验证、端点、模型目录、配额、 | `autoSwitchThreshold?` | `number` | `80` | 基于用量的主动切换阈值。`quota` 可在下一次请求中重新评估已绑定和未绑定任务；`fill-first` 仅把它用作未绑定分配的耗尽点；正常 `round-robin` 不使用它。分数取已知 5 小时、周或 30 天 quota window 的最高值。`0` 只关闭基于用量的主动切换，不关闭未绑定任务分配或故障恢复。 | | `accountPoolStrategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | 新建/未绑定 Codex 请求的分配策略。没有 live `(parent thread id, quota scope)` affinity 的请求属于未绑定；代理重启或 affinity 重置后，已有可见任务也可能未绑定。`quota` 在没有活跃账号时选择已知 usage 最低的合格账号；活跃账号合格且低于 `autoSwitchThreshold` 时继续使用；达到阈值后，可把未绑定请求或已绑定任务的下一次请求切换到 usage 更低的合格账号。`round-robin` 均匀分配未绑定请求；`fill-first` 在 cooldown、不可用或耗尽阈值前持续分配给活跃账号。 | | `accountPoolStickyLimit?` | `number` | `1` | 一次 round-robin 选择在推进前保留的新建/未绑定任务分配数。计数在任务绑定时增加，而不是在上游成功后增加。范围 1–100；仅当 `accountPoolStrategy` 为 `round-robin` 时生效。 | -| `upstreamFailoverThreshold?` | `number` | `3` | 连续发生多少次瞬态故障后，后续新会话会切换到备用上游。设为 `0` 可禁用。已证明的连接前 DNS/TCP 不可达故障按 provider-host 粒度记录，不影响账户健康、冷却、线程/会话亲和性、活动账户选择或 Pool 路由，也不会计入此阈值；未确认的失败仍归属账户。 | +| `upstreamFailoverThreshold?` | `number` | `3` | 连续发生多少次瞬态故障后，后续新会话会切换到备用上游。设为 `0` 可禁用。对于常规 Responses 和原生 compact 发送，已证明的连接前 DNS/TCP 不可达故障按 provider-host 粒度记录，不影响账户健康、冷却、线程/会话亲和性、活动账户选择或 Pool 路由，也不会计入此阈值；未确认的失败仍归属账户。 | +| `upstreamHostCircuitThreshold?` | `number` | `0` | 原生 OpenAI forward Responses 与 compact 发送的可选断路器阈值，仅统计已证明的连接前 DNS/TCP 故障。`0` 表示禁用；`1`–`20` 表示在这么多个终止逻辑请求失败后，对 provider-origin 冷却 30 秒。断路期间会在账户选择和上游发送之前返回带 `Retry-After` 的 `503`；冷却结束后只允许一个半开请求。超时和 HTTP 响应不计数，任意 HTTP 响应都会关闭断路器。 | | `modelCacheTtlMs?` | `number` | `300000` | 每个提供者 `/models` 缓存的新鲜度窗口。 | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Anthropic 提示缓存策略：禁用、5 分钟临时缓存，或 1 小时扩展缓存。 | | `tokenGuardian?` | `OcxTokenGuardianConfig` | 关闭 | 可选的主动 OAuth 刷新与 Codex 账户预热策略。 | diff --git a/src/codex/upstream-host-health.ts b/src/codex/upstream-host-health.ts index 0ca863b0b..a8ac8f267 100644 --- a/src/codex/upstream-host-health.ts +++ b/src/codex/upstream-host-health.ts @@ -1,67 +1,325 @@ /** - * Observational ledger for pre-connection upstream reachability failures, - * keyed by (provider, host). Records ONLY — no circuit breaker, no admission - * change (issue #914 scope). Rotation decisions stay with account health; - * this ledger exists so a host-wide outage is visible as host-wide. + * Provider-origin reachability health and the optional logical-request circuit. * - * Retention: bounded at 128 entries; on overflow the stalest entries by - * last-touch are pruned before insert, and failure timestamps older than the - * window are reconciled away on the next record — repeated provider/base-URL - * churn cannot grow the map for the process lifetime. + * Issue #914 owns the observational ledger. The circuit is deliberately opt-in: + * a threshold of 0 preserves the ledger-only behavior. Only proven + * `connect_neutral` failures may settle a lease as host failure; timeout, reset, + * HTTP, redirect, authentication, and local failures remain outside this module. */ export const UPSTREAM_HOST_HEALTH_MAX_ENTRIES = 128; export const UPSTREAM_HOST_FAILURE_WINDOW_MS = 10 * 60_000; +export const UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS = 30_000; +export const UPSTREAM_HOST_CIRCUIT_MAX_THRESHOLD = 20; export type UpstreamHostHealthEntry = { consecutiveFailures: number; lastFailureAt: number; lastFailureCode?: string; + cooldownUntil?: number; /** Recency marker for stalest-first pruning (not health semantics). */ lastTouch: number; }; -const hostHealth = new Map(); +/** Opaque ownership token for one admitted logical request. */ +export type UpstreamHostAdmissionLease = Readonly<{ + key: string; + leaseId: symbol; + generation: number; + halfOpen: boolean; +}>; + +export type UpstreamHostAdmission = + | { kind: "admitted"; lease: UpstreamHostAdmissionLease | null } + | { kind: "blocked"; retryAfterSeconds: number }; + +type InternalUpstreamHostHealth = UpstreamHostHealthEntry & { + generation: number; + activeLeaseIds: Set; + halfOpenLeaseId?: symbol; + /** True only when the entry is owned by opt-in circuit admissions. */ + circuitManaged: boolean; +}; + +const hostHealth = new Map(); +let nextGenerationValue = 0; export function upstreamHostHealthKey(provider: string, host: string): string { return `${provider}|${host.toLowerCase()}`; } -function pruneForInsert(): void { - if (hostHealth.size < UPSTREAM_HOST_HEALTH_MAX_ENTRIES) return; - const entries = [...hostHealth.entries()].sort((a, b) => a[1].lastTouch - b[1].lastTouch); - for (const [key] of entries) { - if (hostHealth.size < UPSTREAM_HOST_HEALTH_MAX_ENTRIES) return; +export function normalizeUpstreamHostCircuitThreshold(value: unknown): number { + if (typeof value !== "number" || !Number.isInteger(value) || value <= 0) return 0; + return Math.min(value, UPSTREAM_HOST_CIRCUIT_MAX_THRESHOLD); +} + +function nextGeneration(): number { + nextGenerationValue = nextGenerationValue >= Number.MAX_SAFE_INTEGER ? 1 : nextGenerationValue + 1; + return nextGenerationValue; +} + +function snapshot(entry: InternalUpstreamHostHealth): UpstreamHostHealthEntry { + return { + consecutiveFailures: entry.consecutiveFailures, + lastFailureAt: entry.lastFailureAt, + lastTouch: entry.lastTouch, + ...(entry.lastFailureCode !== undefined ? { lastFailureCode: entry.lastFailureCode } : {}), + ...(entry.cooldownUntil !== undefined ? { cooldownUntil: entry.cooldownUntil } : {}), + }; +} + +function newEntry(now: number, circuitManaged: boolean): InternalUpstreamHostHealth { + return { + consecutiveFailures: 0, + lastFailureAt: 0, + lastTouch: now, + generation: nextGeneration(), + activeLeaseIds: new Set(), + circuitManaged, + }; +} + +function advanceGeneration(entry: InternalUpstreamHostHealth): void { + entry.generation = nextGeneration(); + entry.activeLeaseIds.clear(); + delete entry.halfOpenLeaseId; +} + +function removeExpiredUnleased(now: number): void { + for (const [key, entry] of hostHealth) { + if (entry.activeLeaseIds.size > 0) continue; + if ( + entry.consecutiveFailures === 0 + || (entry.cooldownUntil === undefined + && now - entry.lastFailureAt > UPSTREAM_HOST_FAILURE_WINDOW_MS) + ) { + hostHealth.delete(key); + } + } +} + +function oldestUnleasedKey(now: number): string | undefined { + let preferred: [string, number] | undefined; + let cooling: [string, number] | undefined; + for (const [key, entry] of hostHealth) { + if (entry.activeLeaseIds.size > 0) continue; + const candidate: [string, number] = [key, entry.lastTouch]; + if (entry.cooldownUntil !== undefined && entry.cooldownUntil > now) { + if (!cooling || candidate[1] < cooling[1]) cooling = candidate; + } else if (!preferred || candidate[1] < preferred[1]) { + preferred = candidate; + } + } + return preferred?.[0] ?? cooling?.[0]; +} + +function pruneTo(limit: number, now: number): void { + removeExpiredUnleased(now); + while (hostHealth.size > limit) { + const key = oldestUnleasedKey(now); + if (!key) return; // Active leases may temporarily exceed the retention cap. + hostHealth.delete(key); + } +} + +function makeRoom(now: number): void { + pruneTo(UPSTREAM_HOST_HEALTH_MAX_ENTRIES - 1, now); +} + +function issueLease( + key: string, + entry: InternalUpstreamHostHealth, + halfOpen: boolean, + now: number, +): UpstreamHostAdmissionLease { + const leaseId = Symbol(halfOpen ? "upstream-host-half-open" : "upstream-host-admission"); + entry.activeLeaseIds.add(leaseId); + entry.lastTouch = now; + if (halfOpen) entry.halfOpenLeaseId = leaseId; + return { key, leaseId, generation: entry.generation, halfOpen }; +} + +function matchingEntry(lease: UpstreamHostAdmissionLease): InternalUpstreamHostHealth | null { + const entry = hostHealth.get(lease.key); + if (!entry || entry.generation !== lease.generation || !entry.activeLeaseIds.has(lease.leaseId)) { + return null; + } + if (lease.halfOpen && entry.halfOpenLeaseId !== lease.leaseId) return null; + return entry; +} + +function settleLease(entry: InternalUpstreamHostHealth, lease: UpstreamHostAdmissionLease): void { + entry.activeLeaseIds.delete(lease.leaseId); + if (entry.halfOpenLeaseId === lease.leaseId) delete entry.halfOpenLeaseId; +} + +/** + * Admit one logical request. A disabled threshold returns a null lease and has + * byte-for-byte compatible call-site behavior with the observational ledger. + */ +export function acquireUpstreamHostAdmission( + key: string, + thresholdValue: unknown, + now = Date.now(), +): UpstreamHostAdmission { + const threshold = normalizeUpstreamHostCircuitThreshold(thresholdValue); + if (threshold === 0) return { kind: "admitted", lease: null }; + + pruneTo(UPSTREAM_HOST_HEALTH_MAX_ENTRIES, now); + let entry = hostHealth.get(key); + if ( + entry?.activeLeaseIds.size === 0 + && entry.cooldownUntil === undefined + && entry.consecutiveFailures > 0 + && now - entry.lastFailureAt > UPSTREAM_HOST_FAILURE_WINDOW_MS + ) { hostHealth.delete(key); + entry = undefined; + } + if (!entry) { + makeRoom(now); + entry = newEntry(now, true); + hostHealth.set(key, entry); + } else if (!entry.circuitManaged) { + // Observational history predating opt-in admission must not count toward + // opening the circuit. Start a fresh fenced generation. + entry.consecutiveFailures = 0; + entry.lastFailureAt = 0; + entry.circuitManaged = true; + delete entry.lastFailureCode; + delete entry.cooldownUntil; + advanceGeneration(entry); } + + if (entry.cooldownUntil !== undefined) { + if (entry.cooldownUntil > now) { + return { + kind: "blocked", + retryAfterSeconds: Math.max(1, Math.ceil((entry.cooldownUntil - now) / 1_000)), + }; + } + if (entry.halfOpenLeaseId !== undefined) { + return { kind: "blocked", retryAfterSeconds: 1 }; + } + advanceGeneration(entry); + return { kind: "admitted", lease: issueLease(key, entry, true, now) }; + } + + return { kind: "admitted", lease: issueLease(key, entry, false, now) }; } +/** Release an admitted request without recording transport evidence. */ +export function releaseUpstreamHostAdmission( + lease: UpstreamHostAdmissionLease | null | undefined, + now = Date.now(), +): boolean { + if (!lease) return false; + const entry = matchingEntry(lease); + if (!entry) return false; + settleLease(entry, lease); + entry.lastTouch = now; + if (entry.activeLeaseIds.size === 0 && entry.consecutiveFailures === 0) { + hostHealth.delete(lease.key); + } + pruneTo(UPSTREAM_HOST_HEALTH_MAX_ENTRIES, now); + return true; +} + +/** Record one terminal logical `connect_neutral` failure. */ export function recordUpstreamHostFailure( key: string, - opts: { code?: string; now?: number } = {}, + opts: { + code?: string; + now?: number; + threshold?: unknown; + lease?: UpstreamHostAdmissionLease | null; + } = {}, ): void { const now = opts.now ?? Date.now(); - const prior = hostHealth.get(key); - // Prune only for a genuinely new key: updating an existing entry must never - // evict an unrelated one. - if (prior === undefined) pruneForInsert(); - const stale = prior !== undefined && now - prior.lastFailureAt > UPSTREAM_HOST_FAILURE_WINDOW_MS; - const code = typeof opts.code === "string" && opts.code !== "" ? opts.code : prior?.lastFailureCode; - hostHealth.set(key, { - consecutiveFailures: stale || prior === undefined ? 1 : prior.consecutiveFailures + 1, - lastFailureAt: now, - lastTouch: now, - ...(code !== undefined ? { lastFailureCode: code } : {}), - }); + const hasCircuitSettlement = Object.hasOwn(opts, "lease"); + // An integrated logical request may already have settled its lease on an + // earlier physical response. Later retry completions carry explicit null and + // must not mutate a newer generation opened by another request. + if (hasCircuitSettlement && !opts.lease) return; + // Legacy callers without an admission lease continue to populate only the + // observational ledger. Circuit mutation is fenced to explicitly admitted + // logical requests so an unwired side path cannot open it accidentally. + const threshold = opts.lease + ? normalizeUpstreamHostCircuitThreshold(opts.threshold) + : 0; + let entry: InternalUpstreamHostHealth | undefined; + if (opts.lease) { + entry = matchingEntry(opts.lease) ?? undefined; + if (!entry || opts.lease.key !== key) return; // stale completion cannot mutate a newer generation + settleLease(entry, opts.lease); + } else { + entry = hostHealth.get(key); + if (entry?.circuitManaged) { + const stale = entry.activeLeaseIds.size === 0 + && entry.cooldownUntil === undefined + && now - entry.lastFailureAt > UPSTREAM_HOST_FAILURE_WINDOW_MS; + if (!stale) return; + hostHealth.delete(key); + entry = undefined; + } + } + if (!entry) { + makeRoom(now); + entry = newEntry(now, opts.lease !== undefined); + hostHealth.set(key, entry); + } + + const reopens = opts.lease?.halfOpen === true || entry.cooldownUntil !== undefined; + const stale = entry.consecutiveFailures === 0 + || (!reopens && now - entry.lastFailureAt > UPSTREAM_HOST_FAILURE_WINDOW_MS); + entry.consecutiveFailures = reopens && threshold > 0 + ? Math.max(threshold, entry.consecutiveFailures + 1) + : stale ? 1 : entry.consecutiveFailures + 1; + entry.lastFailureAt = now; + entry.lastTouch = now; + const code = typeof opts.code === "string" && opts.code !== "" + ? opts.code + : entry.lastFailureCode; + if (code !== undefined) entry.lastFailureCode = code; + + if (threshold > 0 && (reopens || entry.consecutiveFailures >= threshold)) { + entry.cooldownUntil = now + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS; + advanceGeneration(entry); + } else { + delete entry.cooldownUntil; + } + pruneTo(UPSTREAM_HOST_HEALTH_MAX_ENTRIES, now); } -/** Any real HTTP response from the host clears its reachability streak. */ -export function resetUpstreamHostHealth(key: string): void { - hostHealth.delete(key); +/** Any real HTTP response from the admitted logical request proves reachability. */ +export function resetUpstreamHostHealth( + key: string, + lease?: UpstreamHostAdmissionLease | null, + now = Date.now(), +): boolean { + if (lease === null) return false; + if (lease === undefined) { + const entry = hostHealth.get(key); + if (entry?.circuitManaged) return false; + return hostHealth.delete(key); + } + const entry = matchingEntry(lease); + if (!entry || lease.key !== key) return false; + settleLease(entry, lease); + entry.consecutiveFailures = 0; + entry.lastFailureAt = 0; + entry.lastTouch = now; + delete entry.lastFailureCode; + delete entry.cooldownUntil; + if (entry.activeLeaseIds.size === 0) hostHealth.delete(key); + pruneTo(UPSTREAM_HOST_HEALTH_MAX_ENTRIES, now); + return true; } export function getUpstreamHostHealth(key: string): UpstreamHostHealthEntry | null { - return hostHealth.get(key) ?? null; + const entry = hostHealth.get(key); + if (!entry || entry.consecutiveFailures === 0) return null; + return snapshot(entry); } /** Test hook: clear the whole ledger. */ diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 883f076fb..e87b9ae62 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -69,7 +69,15 @@ import { type UpstreamSendRecovery, } from "../../lib/upstream-retry"; import { classifyTransportFailureKind, transportErrorCode } from "../../lib/upstream-reachability"; -import { recordUpstreamHostFailure, resetUpstreamHostHealth, upstreamHostHealthKey } from "../../codex/upstream-host-health"; +import { + acquireUpstreamHostAdmission, + normalizeUpstreamHostCircuitThreshold, + recordUpstreamHostFailure, + releaseUpstreamHostAdmission, + resetUpstreamHostHealth, + upstreamHostHealthKey, + type UpstreamHostAdmissionLease, +} from "../../codex/upstream-host-health"; import { ForwardAdmissionCredentialError, validateForwardAdmissionCredential } from "../auth-cors"; import { listOpenAiForwardSidecarCandidates, resolveFirstUsableOpenAiSidecar, type ResolvedOpenAiForwardSidecar } from "../../providers/openai-sidecar"; import { isCanonicalOpenAiForwardProvider, supportsNativeResponsesCompactEndpoint } from "../../providers/openai-tiers"; @@ -302,12 +310,43 @@ export async function handleResponsesCompact( // official OpenAI API. Any other Responses-shaped gateway must take the routed // summarizer path below, or compaction fails against an endpoint it never had (#422). if (supportsNativeResponsesCompactEndpoint(route.providerName, route.provider)) { + if (req.signal.aborted) { + return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + } + const preAuthCompactHostKey = normalizeUpstreamHostCircuitThreshold( + config.upstreamHostCircuitThreshold, + ) > 0 + && route.codexAccountMode === "pool" + && route.codexAccountId === undefined + && route.provider.authMode === "forward" + ? upstreamHostHealthKey( + route.providerName, + safeOriginLabel(route.provider.baseUrl ?? ""), + ) + : null; + 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 formatErrorResponse( + 503, + "upstream_host_circuit_open", + "Provider host is temporarily unavailable", + { retryAfter: String(admission.retryAfterSeconds) }, + ); + } + compactHostAdmissionLease = admission.lease; + } + try { // Native ChatGPT/OpenAI model: forward the compact request verbatim to the real 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 authCtx: CodexAuthContext = { kind: "main", accountId: null }; const headers = new Headers({ "content-type": "application/json" }); try { if (route.codexAccountMode) { @@ -354,6 +393,47 @@ export async function handleResponsesCompact( // so routed-model reasoning items (reasoning_text content) don't 400 the ChatGPT backend. const compactBody = sanitizeReasoningInputContent(compactBodyRaw) as typeof compactBodyRaw; const compactUrl = `${base}/responses/compact`; + const actualCompactHostKey = upstreamHostHealthKey( + route.providerName, + safeOriginLabel(compactUrl), + ); + const compactHostKey = compactProvider.authMode === "forward" + ? actualCompactHostKey + : null; + const compactHostCircuitEnabled = compactHostKey !== null + && normalizeUpstreamHostCircuitThreshold(config.upstreamHostCircuitThreshold) > 0; + 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 formatErrorResponse( + 503, + "upstream_host_circuit_open", + "Provider host is temporarily unavailable", + { retryAfter: String(admission.retryAfterSeconds) }, + ); + } + compactHostAdmissionLease = admission.lease; + } + const settleObservedCompactHostResponse = (): void => { + if (compactHostCircuitEnabled) { + resetUpstreamHostHealth(actualCompactHostKey, compactHostAdmissionLease); + } else { + resetUpstreamHostHealth(actualCompactHostKey); + } + compactHostAdmissionLease = null; + }; const compactThreadId = req.headers.get("x-codex-parent-thread-id"); const connectMs = config.connectTimeoutMs ?? 200_000; // Takes its context explicitly: the alternate-account flow below records a rejection @@ -408,7 +488,7 @@ export async function handleResponsesCompact( ).then(res => { // Every real attempt response — including an intermediate 5xx the retry // wrapper replaces — proves the host was reached (#914 review). - resetUpstreamHostHealth(upstreamHostHealthKey(route.providerName, safeOriginLabel(compactUrl))); + settleObservedCompactHostResponse(); return res; }); return recovery === "single" @@ -432,11 +512,19 @@ export async function handleResponsesCompact( const outcome = classifyTransportFailureKind(err); // Host-level evidence stands regardless of pool membership (#914 review). if (outcome === "connect_neutral") { - recordUpstreamHostFailure( - upstreamHostHealthKey(route.providerName, safeOriginLabel(compactUrl)), - { code: transportErrorCode(err) }, - ); + 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"); } @@ -507,11 +595,19 @@ export async function handleResponsesCompact( const outcome = classifyTransportFailureKind(err); // Host-level evidence stands regardless of pool membership (#914 review). if (outcome === "connect_neutral") { - recordUpstreamHostFailure( - upstreamHostHealthKey(route.providerName, safeOriginLabel(compactUrl)), - { code: transportErrorCode(err) }, - ); + 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"); } @@ -538,6 +634,10 @@ export async function handleResponsesCompact( // synthetic buffer errors are not upstream bodies and stay uninspected. if (buffered.ok) inspectResponseLogJson(logCtx, await buffered.clone().text()); return buffered; + } finally { + releaseUpstreamHostAdmission(compactHostAdmissionLease); + releaseCodexAuthContextProbeLease(authCtx); + } } // ROUTED model: run the v2 synthetic-compaction turn internally (appends COMPACT_PROMPT, no diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index f1e20a6ba..87b26279d 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -177,7 +177,15 @@ import { buildToolBridgeMaps, collabSurface, injectDeveloperMessage, multiAgentG import { hasUnreadableEncryptedAgentTask, looksLikeBackendCiphertext, sanitizeEncryptedContentInPlace } from "./encrypted-payload"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers"; import { classifyTransportFailureKind, transportErrorCode } from "../../lib/upstream-reachability"; -import { recordUpstreamHostFailure, resetUpstreamHostHealth, upstreamHostHealthKey } from "../../codex/upstream-host-health"; +import { + acquireUpstreamHostAdmission, + normalizeUpstreamHostCircuitThreshold, + recordUpstreamHostFailure, + releaseUpstreamHostAdmission, + resetUpstreamHostHealth, + upstreamHostHealthKey, + type UpstreamHostAdmissionLease, +} from "../../codex/upstream-host-health"; import { createResponsesSnapshotBlockRewrite, hasResponsesSnapshotRepair, @@ -241,6 +249,26 @@ export function usesCodexForwardPoolAuth( && provider.authMode === "forward" && provider.adapter === "openai-responses"; } +function preAuthUpstreamHostCircuitKey(route: RouteResult, config: OcxConfig): string | null { + if ( + normalizeUpstreamHostCircuitThreshold(config.upstreamHostCircuitThreshold) === 0 + || route.codexAccountMode !== "pool" + || route.codexAccountId !== undefined + || route.provider.authMode !== "forward" + || route.provider.adapter !== "openai-responses" + ) return null; + return upstreamHostHealthKey(route.providerName, safeOriginLabel(route.provider.baseUrl ?? "")); +} + +function upstreamHostCircuitOpenResponse(retryAfterSeconds: number): Response { + return formatErrorResponse( + 503, + "upstream_host_circuit_open", + "Provider host is temporarily unavailable", + { retryAfter: String(retryAfterSeconds) }, + ); +} + function normalizeCodexUnsupportedModelDetail(value: string): string { return value.trim().replace(/\s+/gu, " ").toLocaleLowerCase("en-US"); } @@ -455,7 +483,12 @@ async function retryCodexPoolOnAlternateAccount( route.provider.authMode === "forward", ); // A real HTTP response proves the host was reached (#914). - resetUpstreamHostHealth(upstreamHostHealthKey(route.providerName, safeOriginLabel(request.url))); + const retryHostKey = upstreamHostHealthKey(route.providerName, safeOriginLabel(request.url)); + if (normalizeUpstreamHostCircuitThreshold(config.upstreamHostCircuitThreshold) > 0) { + resetUpstreamHostHealth(retryHostKey, null); + } else { + resetUpstreamHostHealth(retryHostKey); + } return { kind: "retried", authCtx: retryAuthCtx, @@ -1296,6 +1329,9 @@ async function handleResponsesInner( logCtx: RequestLogContext, options: HandleResponsesOptions & { translatorBudget: TranslatorBudget }, ): Promise { + let pendingHostAdmissionLease: UpstreamHostAdmissionLease | null = null; + let authCtx: CodexAuthContext = { kind: "main", accountId: null }; + try { // The Chat and Anthropic surfaces replay through here with a Responses-shaped body, // so an omitted value means a genuine Responses inbound. const inboundWire = options.inboundWire ?? "responses"; @@ -1425,7 +1461,6 @@ async function handleResponsesInner( nativeMainSelectionOnly: !nativeMainRecoveryBlocked && previewSelectionAdmission?.mainProfileDraining === true, }; - let authCtx: CodexAuthContext = { kind: "main", accountId: null }; let selectedForwardHeaders = req.headers; let subagentFallbackAccountId = config.activeCodexAccountId ?? null; let subagentQuotaFailureModel = parsed.modelId; @@ -1527,6 +1562,19 @@ async function handleResponsesInner( logCtx.provider = `${route.providerName}-${route.codexAccountNamespace}`; } + if (options.abortSignal?.aborted) return clientCancelledResponse(); + const preAuthHostKey = preAuthUpstreamHostCircuitKey(route, config); + if (preAuthHostKey) { + const admission = acquireUpstreamHostAdmission( + preAuthHostKey, + config.upstreamHostCircuitThreshold, + ); + if (admission.kind === "blocked") { + return upstreamHostCircuitOpenResponse(admission.retryAfterSeconds); + } + pendingHostAdmissionLease = admission.lease; + } + { const finalAuth = await resolveResponsesCodexAuth(req, config, route, options); if (!finalAuth.ok) return finalAuth.response; @@ -1728,6 +1776,9 @@ async function handleResponsesInner( } if ("passthrough" in adapter && adapter.passthrough && !routedCompaction) { + let hostAdmissionLease = pendingHostAdmissionLease; + pendingHostAdmissionLease = null; + try { const imageGenCallAliases = route.provider.authMode === "forward" ? new Map() : imageGenToolCallAliases(toolBridgeMaps.toolNsMap, parsed._rawBody, translatorBudget); @@ -1754,6 +1805,41 @@ async function handleResponsesInner( } let request = await adapter.buildRequest(parsed, { headers: selectedForwardHeaders, translatorBudget }); recordAdapterReasoning(logCtx, request); + const actualHostKey = upstreamHostHealthKey( + route.providerName, + safeOriginLabel(request.url), + ); + const hostKey = route.provider.authMode === "forward" + ? actualHostKey + : null; + const hostCircuitEnabled = hostKey !== null + && normalizeUpstreamHostCircuitThreshold(config.upstreamHostCircuitThreshold) > 0; + if (hostAdmissionLease && hostAdmissionLease.key !== hostKey) { + return formatErrorResponse(502, "upstream_error", "Provider host changed after circuit admission"); + } + if (options.abortSignal?.aborted) { + releaseCodexAuthContextProbeLease(authCtx); + return clientCancelledResponse(); + } + if (!hostAdmissionLease && hostCircuitEnabled) { + const admission = acquireUpstreamHostAdmission( + hostKey!, + config.upstreamHostCircuitThreshold, + ); + if (admission.kind === "blocked") { + releaseCodexAuthContextProbeLease(authCtx); + return upstreamHostCircuitOpenResponse(admission.retryAfterSeconds); + } + hostAdmissionLease = admission.lease; + } + const settleObservedHostResponse = (): void => { + if (hostCircuitEnabled) { + resetUpstreamHostHealth(actualHostKey, hostAdmissionLease); + } else { + resetUpstreamHostHealth(actualHostKey); + } + hostAdmissionLease = null; + }; const passthroughEstimate = typeof request.usageLog?.inputTokens === "number" ? request.usageLog.inputTokens : undefined; @@ -1769,16 +1855,30 @@ async function handleResponsesInner( let upstreamResponse: Response; const transportFailureResponse = (err: unknown): Response => { upstream.abort(); - if (options.abortSignal?.aborted) return clientCancelledResponse(); + if (options.abortSignal?.aborted) { + releaseUpstreamHostAdmission(hostAdmissionLease); + hostAdmissionLease = null; + releaseCodexAuthContextProbeLease(authCtx); + return clientCancelledResponse(); + } const outcome = classifyTransportFailureKind(err); // Host-level evidence stands regardless of pool membership: a direct // forward send has no pool accounting, but the reachability failure is // still host-wide, not account evidence (#914 review). if (outcome === "connect_neutral") { - recordUpstreamHostFailure( - upstreamHostHealthKey(route.providerName, safeOriginLabel(request.url)), - { code: transportErrorCode(err) }, - ); + if (hostCircuitEnabled) { + recordUpstreamHostFailure(actualHostKey, { + code: transportErrorCode(err), + threshold: config.upstreamHostCircuitThreshold, + lease: hostAdmissionLease, + }); + } else { + recordUpstreamHostFailure(actualHostKey, { code: transportErrorCode(err) }); + } + hostAdmissionLease = null; + } else { + releaseUpstreamHostAdmission(hostAdmissionLease); + hostAdmissionLease = null; } if (usesCodexForwardPoolAuth(authCtx, route.provider)) { recordCodexUpstreamOutcome(config, authCtx.accountId, outcome, { @@ -1811,7 +1911,7 @@ async function handleResponsesInner( // Every real attempt response — including an intermediate 5xx the // retry wrapper replaces — proves the host was reached (#914 review). .then(res => { - resetUpstreamHostHealth(upstreamHostHealthKey(route.providerName, safeOriginLabel(request.url))); + settleObservedHostResponse(); return res; }); }, @@ -1870,7 +1970,7 @@ async function handleResponsesInner( }, recovery), upstream.signal, connectMs, parsed.stream, providerFetch(route.provider), route.provider.authMode === "forward") .then(res => { - resetUpstreamHostHealth(upstreamHostHealthKey(route.providerName, safeOriginLabel(request.url))); + settleObservedHostResponse(); return res; }); }, @@ -2306,6 +2406,12 @@ async function handleResponsesInner( status: upstreamResponse.status, headers, }); + } finally { + if (hostAdmissionLease) { + releaseUpstreamHostAdmission(hostAdmissionLease); + releaseCodexAuthContextProbeLease(authCtx); + } + } } // Image / web-search sidecars: plan once, then dispatch with runTurn-aware priority. @@ -3331,6 +3437,12 @@ async function handleResponsesInner( } return formatErrorResponse(400, "invalid_request_error", "Non-streaming not supported by this adapter"); + } finally { + if (pendingHostAdmissionLease) { + releaseUpstreamHostAdmission(pendingHostAdmissionLease); + releaseCodexAuthContextProbeLease(authCtx); + } + } } diff --git a/src/types.ts b/src/types.ts index 67b3f8bec..4c8381858 100644 --- a/src/types.ts +++ b/src/types.ts @@ -757,6 +757,11 @@ export interface OcxConfig { accountPoolStickyLimit?: number; /** Consecutive non-2xx upstream responses before switching future new threads. Default 3. 0 = disabled. */ upstreamFailoverThreshold?: number; + /** + * Opt-in provider-origin circuit threshold for proven pre-connection reachability failures. + * Default 0 (disabled); range 0..20. The circuit never counts timeouts or HTTP responses. + */ + upstreamHostCircuitThreshold?: number; /** * Opt-in Anthropic OAuth account pool (#294). Default OFF. * Failover on 429 + sticky affinity; new sessions may pick lowest known 5h usage. diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index d6cd644b4..c431633e8 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -23,6 +23,7 @@ import { releaseCodexAuthContextProbeLease, resolveCodexAuthContext, } from "../src/codex/auth-context"; +import { clearUpstreamHostHealth } from "../src/codex/upstream-host-health"; import { supportsNativeResponsesCompactEndpoint } from "../src/providers/openai-tiers"; import type { RequestLogContext } from "../src/server/request-log"; import { acquireNativeMainProfileDrain, tryAdmitTurn } from "../src/server/lifecycle"; @@ -571,6 +572,7 @@ describe("compact alternate-account attempt (#913)", () => { process.env.OPENCODEX_HOME = testDir; process.env.CODEX_HOME = testDir; clearCodexUpstreamHealth(); + clearUpstreamHostHealth(); clearAccountQuota(); for (const id of ["pool-a", "pool-b"]) { saveCodexAccountCredential(id, { @@ -584,6 +586,7 @@ describe("compact alternate-account attempt (#913)", () => { return run(twoAccountPoolConfig()).finally(() => { globalThis.fetch = originalFetch; clearCodexUpstreamHealth(); + clearUpstreamHostHealth(); clearAccountQuota(); rmSync(testDir, { recursive: true, force: true }); if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; @@ -929,4 +932,59 @@ describe("compact alternate-account attempt (#913)", () => { expect(statuses).toEqual([402, 429]); }); }); + + test("an opt-in regular circuit blocks before selecting another pool account", async () => { + await withPoolEnv("ocx-regular-host-circuit-", async config => { + config.upstreamHostCircuitThreshold = 1; + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + throw Object.assign(new Error("connection refused"), { code: "ECONNREFUSED" }); + }) as typeof fetch; + + const request = () => new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "gpt-5.6-sol", input: "hello", stream: false }), + }); + const first = await handleResponses(request(), config, { model: "", provider: "" }); + const second = await handleResponses(request(), config, { model: "", provider: "" }); + + expect(first.status).toBe(502); + expect(second.status).toBe(503); + expect(second.headers.get("retry-after")).toBe("30"); + expect(sends).toBe(1); + expect(getCodexUpstreamHealth("pool-a")).toBeNull(); + expect(getCodexUpstreamHealth("pool-b")).toBeNull(); + }); + }); + + test("an opt-in compact circuit blocks before selecting another pool account", async () => { + await withPoolEnv("ocx-compact-host-circuit-", async config => { + config.upstreamHostCircuitThreshold = 1; + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + throw Object.assign(new Error("connection refused"), { code: "ECONNREFUSED" }); + }) as typeof fetch; + + const first = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + const second = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(first.status).toBe(502); + expect(second.status).toBe(503); + expect(second.headers.get("retry-after")).toBe("30"); + expect(sends).toBe(1); + expect(getCodexUpstreamHealth("pool-a")).toBeNull(); + expect(getCodexUpstreamHealth("pool-b")).toBeNull(); + }); + }); }); diff --git a/tests/upstream-host-circuit.test.ts b/tests/upstream-host-circuit.test.ts new file mode 100644 index 000000000..38c90a12d --- /dev/null +++ b/tests/upstream-host-circuit.test.ts @@ -0,0 +1,174 @@ +import { beforeEach, describe, expect, test } from "bun:test"; +import { + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS, + UPSTREAM_HOST_CIRCUIT_MAX_THRESHOLD, + acquireUpstreamHostAdmission, + clearUpstreamHostHealth, + getUpstreamHostHealth, + normalizeUpstreamHostCircuitThreshold, + recordUpstreamHostFailure, + releaseUpstreamHostAdmission, + resetUpstreamHostHealth, + upstreamHostHealthKey, + type UpstreamHostAdmissionLease, +} from "../src/codex/upstream-host-health"; + +beforeEach(() => clearUpstreamHostHealth()); + +function admit(key: string, threshold: number, now: number): UpstreamHostAdmissionLease { + const admission = acquireUpstreamHostAdmission(key, threshold, now); + expect(admission.kind).toBe("admitted"); + if (admission.kind !== "admitted" || !admission.lease) { + throw new Error("expected a circuit admission lease"); + } + return admission.lease; +} + +function fail(key: string, threshold: number, now: number): void { + recordUpstreamHostFailure(key, { + code: "ECONNREFUSED", + now, + threshold, + lease: admit(key, threshold, now), + }); +} + +describe("opt-in upstream host circuit", () => { + test("normalizes the opt-in threshold and leaves zero disabled", () => { + expect(normalizeUpstreamHostCircuitThreshold(undefined)).toBe(0); + expect(normalizeUpstreamHostCircuitThreshold(-1)).toBe(0); + expect(normalizeUpstreamHostCircuitThreshold(1.5)).toBe(0); + expect(normalizeUpstreamHostCircuitThreshold(3)).toBe(3); + expect(normalizeUpstreamHostCircuitThreshold(999)).toBe(UPSTREAM_HOST_CIRCUIT_MAX_THRESHOLD); + + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + expect(acquireUpstreamHostAdmission(key, 0, 1_000)).toEqual({ + kind: "admitted", + lease: null, + }); + }); + + test("legacy observations cannot open the opt-in circuit without a lease", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + for (let attempt = 0; attempt < 3; attempt++) { + recordUpstreamHostFailure(key, { + code: "ECONNREFUSED", + now: 2_000 + attempt, + threshold: 1, + }); + } + expect(getUpstreamHostHealth(key)).toMatchObject({ + consecutiveFailures: 3, + lastFailureCode: "ECONNREFUSED", + }); + expect(getUpstreamHostHealth(key)?.cooldownUntil).toBeUndefined(); + }); + + test("opens exactly at the configured threshold", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + const threshold = 3; + fail(key, threshold, 3_001); + fail(key, threshold, 3_002); + expect(getUpstreamHostHealth(key)).toMatchObject({ consecutiveFailures: 2 }); + expect(getUpstreamHostHealth(key)?.cooldownUntil).toBeUndefined(); + + fail(key, threshold, 3_003); + expect(getUpstreamHostHealth(key)).toMatchObject({ + consecutiveFailures: 3, + cooldownUntil: 3_003 + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS, + }); + expect(acquireUpstreamHostAdmission(key, threshold, 3_004)).toEqual({ + kind: "blocked", + retryAfterSeconds: 30, + }); + }); + + test("admits one half-open request and an HTTP response closes the circuit", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + fail(key, 1, 4_000); + const probeAt = 4_000 + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS; + const probe = admit(key, 1, probeAt); + expect(probe.halfOpen).toBe(true); + expect(acquireUpstreamHostAdmission(key, 1, probeAt)).toEqual({ + kind: "blocked", + retryAfterSeconds: 1, + }); + expect(resetUpstreamHostHealth(key, probe, probeAt + 1)).toBe(true); + expect(getUpstreamHostHealth(key)).toBeNull(); + }); + + test("a half-open reachability failure immediately reopens the cooldown", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + fail(key, 1, 5_000); + const probeAt = 5_000 + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS; + fail(key, 1, probeAt); + expect(getUpstreamHostHealth(key)).toMatchObject({ + cooldownUntil: probeAt + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS, + }); + }); + + test("releasing a half-open request adds no evidence and permits another probe", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + fail(key, 1, 6_000); + const probeAt = 6_000 + UPSTREAM_HOST_CIRCUIT_COOLDOWN_MS; + const before = getUpstreamHostHealth(key); + const first = admit(key, 1, probeAt); + expect(releaseUpstreamHostAdmission(first, probeAt)).toBe(true); + expect(getUpstreamHostHealth(key)).toMatchObject({ + consecutiveFailures: before!.consecutiveFailures, + lastFailureAt: before!.lastFailureAt, + lastFailureCode: before!.lastFailureCode, + cooldownUntil: before!.cooldownUntil, + }); + expect(admit(key, 1, probeAt).halfOpen).toBe(true); + }); + + test("an HTTP response preserves a concurrent lease and its later failure authority", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + const first = admit(key, 3, 7_000); + const concurrent = admit(key, 3, 7_000); + expect(resetUpstreamHostHealth(key, first, 7_001)).toBe(true); + expect(getUpstreamHostHealth(key)).toBeNull(); + + recordUpstreamHostFailure(key, { + code: "ECONNREFUSED", + now: 7_002, + threshold: 3, + lease: concurrent, + }); + expect(getUpstreamHostHealth(key)).toMatchObject({ consecutiveFailures: 1 }); + }); + + test("a stale completion cannot mutate the generation that opened the circuit", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + const stale = admit(key, 1, 8_000); + fail(key, 1, 8_001); + const before = getUpstreamHostHealth(key); + + recordUpstreamHostFailure(key, { + code: "ECONNREFUSED", + now: 8_002, + threshold: 1, + lease: stale, + }); + expect(getUpstreamHostHealth(key)).toEqual(before); + }); + + test("a later physical retry without its lease cannot close a newer circuit", () => { + const key = upstreamHostHealthKey("openai", "https://chatgpt.com"); + fail(key, 1, 9_000); + const before = getUpstreamHostHealth(key); + + expect(resetUpstreamHostHealth(key, null, 9_001)).toBe(false); + recordUpstreamHostFailure(key, { + code: "ECONNREFUSED", + now: 9_002, + threshold: 1, + lease: null, + }); + // Unwired observational callers are also unable to mutate circuit-owned state. + recordUpstreamHostFailure(key, { code: "ECONNREFUSED", now: 9_003 }); + + expect(getUpstreamHostHealth(key)).toEqual(before); + }); +});