From 6dd3a34e85c0fb8f8fb3a11b0ad5146e9bd0bccf Mon Sep 17 00:00:00 2001 From: Fred Amartey <43480311+FredAmartey@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:36:45 -0400 Subject: [PATCH] feat(retry): opt-in replay of a pre-response reset for self-contained Responses sends A native Responses send whose upstream connection closes before any response byte is answered with the non-replayable 429 refusal since #4798. Codex does not retry a 429, so on a long thread that close ends the turn, while the direct path would retry it as a transport error. Add `providers..retryOnReset`: off by default, one replay by default, up to three total sends. Only a request the proxy can judge self-contained is replayed (`store: false`, complete input, client-executed tools only, no server-side continuation state), decided once per request on the parsed inbound body and carried by every native passthrough leg. The replay is a ceiling inside the leg's existing send budget, never an addition to it. When it is spent, or a later attempt fails any other way, the same refusal is returned, so no exit can invite the client to resend. Replay-safe sidecar callers are unchanged. Validated at the management write boundary like `retryOn429`; a malformed block degrades to absent at load like `webSearchBridge`. Documented in the provider reference (all locales), the server notes and the owning structure sections. --- .../fr/reference/configuration/providers.md | 1 + .../ja/reference/configuration/providers.md | 1 + .../ko/reference/configuration/providers.md | 1 + .../docs/reference/configuration/providers.md | 1 + .../docs/reference/configuration/server.md | 12 + .../ru/reference/configuration/providers.md | 1 + .../tr/reference/configuration/providers.md | 1 + .../reference/configuration/providers.md | 1 + .../reference/configuration/providers.md | 1 + scripts/test-layout/layout.json | 4 +- src/config.ts | 2 +- src/config/load-degrade.ts | 32 +- src/config/schema/leaf-validators.ts | 15 + src/lib/upstream-retry.ts | 97 +++++- src/providers/key-failover.ts | 28 +- src/server/auth-cors.ts | 9 + src/server/responses/core-codex-account.ts | 84 +++-- src/server/responses/passthrough-dispatch.ts | 89 ++++-- src/server/responses/reset-replay.ts | 91 ++++++ src/types.ts | 2 + src/types/provider.ts | 26 ++ structure/transports/responses.md | 39 ++- tests/fixtures/test-layout-expected.json | 4 +- tests/lib/upstream-retry.test.ts | 80 +++++ .../upstream-transient-retry.test.ts | 21 +- .../responses/responses-core-modules.test.ts | 3 +- .../responses-pool-401-refresh.test.ts | 158 +++++++++- .../responses/responses-reset-replay.test.ts | 289 ++++++++++++++++++ .../management-provider-reset-replay.test.ts | 39 +++ 29 files changed, 1049 insertions(+), 83 deletions(-) create mode 100644 src/server/responses/reset-replay.ts create mode 100644 tests/responses/responses-reset-replay.test.ts create mode 100644 tests/server/management-provider-reset-replay.test.ts diff --git a/docs-site/src/content/docs/fr/reference/configuration/providers.md b/docs-site/src/content/docs/fr/reference/configuration/providers.md index e96e79f5f2..ac4c08fee9 100644 --- a/docs-site/src/content/docs/fr/reference/configuration/providers.md +++ b/docs-site/src/content/docs/fr/reference/configuration/providers.md @@ -139,6 +139,7 @@ sauvegarde dont le contenu diffère, puis réécrit en identifiants sans préfix | `responsesSnapshotRepair?` | `boolean` | Réparation côté client désactivée par défaut pour les instantanés du cycle de vie des réponses clairsemés dans SSE et JSON. Remplit les métadonnées d'état canonique, de sortie et d'outil manquantes tandis que l'inspection brute et la persistance restent inchangées. | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | Fournisseurs à clé API uniquement (`authMode: "key"`). Nouvelle tentative facultative sur la même cible après un 429 : lorsque `retryOn429` est absent, la fonctionnalité est désactivée ; la présence d'un objet l'active, sauf avec `enabled: false`. Après un 429, le proxy attend selon `Retry-After` reçu en amont ou selon l'intervalle fixe, puis relit la requête à l'identique avec la même clé avant tout basculement de clé. Ce comportement couvre la boucle principale de récupération d'un tour textuel, le protocole de transfert Responses, le pont d'images et de vidéos, le service auxiliaire de recherche Web et les continuations du terminal. Seules les réponses HTTP 429 reçues avant le début de la diffusion peuvent être relues ; les transports `runTurn` personnalisés ne font pas partie de la boucle de nouvelle tentative HTTP. `attempts` compte les relectures avec la même clé après le premier 429, soit `attempts` + 1 envois au total, et constitue un budget commun à toute la requête, partagé entre la boucle principale de récupération, la continuation de la garde du terminal et les nouvelles tentatives du pont. L'épuisement de `attempts` arrête uniquement les relectures supplémentaires avec la même clé : le basculement normal de clé ou la gestion de l'erreur finale s'applique ensuite selon les cibles disponibles. Sur le protocole de transfert authentifié par clé, aucun basculement n'est possible ; le 429 final est donc renvoyé sans modification. Codex ne retente jamais lui-même une requête après un 429 : cette option constitue ainsi la seule protection pour les fournisseurs à clé unique. Valeurs par défaut : `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000` (chaque attente est plafonnée à `maxIntervalMs`, lui-même plafonné à 600000), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | Fournisseurs `openai-chat` et `openai-responses` authentifiés par clé uniquement. Les fournisseurs `authMode: "forward"` (le pool de comptes ChatGPT) ne lisent jamais cette option et conservent l'échelle par défaut. Nouvelle tentative facultative pour les états transitoires reçus en amont avant le début de la diffusion (500, 502, 503, 504, 520, 521, 522) : l'absence de l'option la désactive ; la présence d'un objet l'active, sauf avec `enabled: false`. Ce comportement couvre la requête Responses initiale, la continuation de la garde du terminal, le point de terminaison natif `/v1/chat/completions` et les réémissions liées à la récupération après un 429 ou à la récupération de compte. `attempts` représente le nombre TOTAL d'envois en amont autorisés pour une requête, premier envoi compris (de 1 à 10, valeur par défaut : 3). Il constitue un budget commun à la requête, partagé avec la récupération après une réinitialisation de connexion ; ainsi, `3` signifie qu'au plus trois requêtes réelles atteignent le fournisseur. Les attentes utilisent une temporisation exponentielle à base fixe de 400 ms, plafonnée à 5 s, et respectent `Retry-After`. Cette option est distincte de `retryOn429`, qui traite la limitation de débit ; les échecs en cours de diffusion ne sont jamais relus. | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | Envois Responses natifs uniquement (`adapter: "openai-responses"`, backend ChatGPT canonique compris). Renvoi optionnel lorsque la connexion amont se ferme avant le moindre octet de réponse : absent signifie désactivé, la présence de l'objet l'active sauf `enabled: false`. Désactivé, le proxy répond à cette coupure par le HTTP 429 non rejouable `upstream_reset_replay_refused` décrit dans [server](server.md). Activé, une requête que le proxy peut juger autonome — `store: false`, `input` complet, uniquement des outils exécutés côté client (`function`, `custom`, `tool_search` client, groupes `namespace` de ceux-ci) et aucun `previous_response_id`, `conversation`, `background` ni `stream_id` — est renvoyée sur une nouvelle connexion. `attempts` est le nombre TOTAL d'envois que le renvoi peut atteindre, premier compris (1..3, défaut 2), et ne dépasse jamais le budget d'envois dont dispose déjà l'étape. Une fois les renvois épuisés, ou si un renvoi échoue pour une raison autre qu'une annulation de l'appelant, le même refus est retourné, jamais un statut qui inviterait le client à réémettre. Une annulation pendant un renvoi remonte comme l'annulation elle-même. Toute autre forme de requête conserve le refus. L'inférence rejouée peut être facturée si l'origine avait déjà commencé la première. | | `autoToolChoiceOnlyModels?` | `string[]` | Modèles dont `tool_choice` accepte uniquement `auto` ou `none` ; les choix forcés sont dévalorisés. | | `preserveReasoningContentModels?` | `string[]` | Modèles nécessitant un assistant préalable `reasoning_content` dans l'historique des discussions. | | `reasoningDetailsModels?` | `string[]` | Modèles dont le point de terminaison renvoie la réflexion sous forme de tableau structuré `reasoning_details` (MiniMax série M avec `reasoning_split`) ; les deltas de flux sont des instantanés cumulatifs comparés par préfixe, et la réflexion conservée est rejouée sous forme de tableau `reasoning_details` plutôt que de chaîne `reasoning_content`. | 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 6a2364e23c..a641ea2d2f 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ja/reference/configuration/providers.md @@ -131,6 +131,7 @@ account を削除しても mapping は保持され、同じ id を再追加す | `responsesSnapshotRepair?` | `boolean` | デフォルトで無効のクライアント向け修復です。SSE と JSON の Responses ライフサイクルで欠落した status、output、ツールメタデータを補完し、raw 検査と永続化は変更しません。 | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | API-key プロバイダーのみ(`authMode: "key"`)。オプトインの同一ターゲット 429 リトライ: `retryOn429` が無ければ無効で、オブジェクトがあれば `enabled: false` でない限り有効になります。429 時に待機(上流の `Retry-After` または固定間隔)してから、キー フェイルオーバーの前に同一キーで同一リクエストを再送します — メインのテキストターン回復ループ、Responses passthrough、画像/動画ブリッジ、web-search サイドカー、ターミナル継続要求をすべてカバーします。再送の対象はプリストリームの HTTP 429 応答のみで、カスタム `runTurn` トランスポートは HTTP リトライループの対象外です。`attempts` は最初の 429 以降の同一キー再送回数(合計送信数 = `attempts` + 1)で、メインの回復ループ・ターミナルガード継続・ブリッジ再試行で共有されるリクエスト単位の予算です。`attempts` を使い切っても同一キーでの再送が止まるだけで、通常のキー フェイルオーバーまたは最終エラー処理が利用可能なターゲットに応じて続きます — キー認証の passthrough ワイヤにはフェイルオーバーがないため、使い切った 429 はそのまま返ります。Codex 自体は 429 をリトライしないため、単一キーのプロバイダーでは唯一の防御です。デフォルト: `enabled: true`、`attempts: 3`、`intervalMs: 5000`、`maxIntervalMs: 60000`(1回の待機は `maxIntervalMs` で上限、その上限は 600000)、`respectRetryAfter: true`。 | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | キー認証の `openai-chat` および `openai-responses` プロバイダーのみ。`authMode: "forward"` のプロバイダー（ChatGPT アカウントプール）はこのオプションを読まず、既定の再試行段数を維持します。ストリーム開始前に上流から返される一時的なステータス（500、502、503、504、520、521、522）に対するオプトインの再試行です。設定がなければ無効で、オブジェクトを指定すると `enabled: false` でない限り有効になります。最初の Responses リクエスト、ターミナルガード継続、ネイティブの `/v1/chat/completions`、および 429／アカウント回復時の再取得が対象です。`attempts` は最初の送信を含め、1 回のリクエストで許可される上流への送信総数です（1～10、デフォルトは 3）。接続リセット回復と共有するリクエスト単位の単一予算であるため、`3` を指定した場合、プロバイダーに到達する実リクエストは最大 3 回です。待機には 400 ms を基準とする固定式の指数バックオフを使用し、上限は 5 秒で、`Retry-After` に従います。レート制限を扱う `retryOn429` とは別の機能であり、ストリーム開始後の失敗は再送されません。 | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | ネイティブ Responses 送信のみ（`adapter: "openai-responses"`、正規の ChatGPT バックエンドを含む）。応答バイトが 1 つも届く前に上流接続が閉じた場合のオプトイン再送です。未指定なら無効、オブジェクトがあれば `enabled: false` でない限り有効。無効時、プロキシはこの切断に対して [server](server.md) に記載の再送不可な HTTP 429 `upstream_reset_replay_refused` を返します。有効時、プロキシが自己完結と判断できるリクエスト（`store: false`、完全な `input`、クライアント実行のツールのみ（`function`、`custom`、client `tool_search`、それらの `namespace` グループ）、かつ `previous_response_id`、`conversation`、`background`、`stream_id` なし）を新しい接続で再送します。`attempts` は最初の送信を含む再送到達可能な合計送信回数（1..3、既定 2）で、そのレグが既に持つ送信予算を超えることはありません。再送を使い切った場合や、呼び出し側のキャンセル以外の理由で再送が失敗した場合は同じ拒否応答を返し、クライアントに再送を促すステータスは返しません。再送中のキャンセルはキャンセルそのものとして返ります。それ以外の形のリクエストは拒否のままです。オリジンが最初の推論を開始済みなら、再送分も課金される可能性があります。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` が `auto` または `none` のみを受け入れるモデル。強制的な選択は格下げされます。 | | `preserveReasoningContentModels?` | `string[]` |チャット履歴に以前のアシスタント `reasoning_content` が必要なモデル。 | | `reasoningDetailsModels?` | `string[]` | thinking を構造化された `reasoning_details` 配列で返すモデル（`reasoning_split` 使用の MiniMax M シリーズ）。ストリーム差分は累積スナップショットとして prefix-diff され、保持された reasoning は `reasoning_content` 文字列ではなく `reasoning_details` 配列としてリプレイされます。 | 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 824a982694..a4b3ea8c5a 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ko/reference/configuration/providers.md @@ -131,6 +131,7 @@ managed map을 활성화하면 privacy-safe selector를 만들고, 이후 계정 | `responsesSnapshotRepair?` | `boolean` | 기본값이 꺼진 클라이언트용 복구입니다. SSE와 JSON의 Responses 수명 주기에서 누락된 status, output, 도구 메타데이터를 채우며 raw 검사와 영속화는 변경하지 않습니다. | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | API-key 프로바이더 전용(`authMode: "key"`). 동일 대상 429 재시도: `retryOn429`가 없으면 기능이 꺼져 있고, 객체가 있으면 `enabled: false`가 아닌 한 활성화됩니다. 429 시 대기(업스트림 `Retry-After` 또는 고정 간격) 후 키 장애 조치 전에 동일 키로 동일 요청을 재전송합니다 — 일반 텍스트 턴 복구 루프, Responses passthrough, 이미지/비디오 브리지, web-search 사이드카, 터미널 연속 요청을 모두 포함합니다. 재전송 대상은 프리스트림 HTTP 429 응답뿐이며, 커스텀 `runTurn` 전송은 HTTP 재시도 루프에서 제외됩니다. `attempts`는 첫 429 이후의 동일 키 재전송 횟수(총 전송 = `attempts` + 1)이며, 메인 복구 루프·터미널 가드 연속 요청·브리지 재시도가 공유하는 요청 단위 예산입니다. `attempts`를 모두 소진해도 동일 키 재전송만 중단되며, 이후에는 일반 키 장애 조치 또는 최종 오류 처리가 사용 가능한 대상에 따라 진행됩니다 — 키 인증 passthrough 와이어에는 장애 조치가 없으므로 소진된 429가 그대로 반환됩니다. Codex 자체는 429를 재시도하지 않으므로 단일 키 프로바이더의 유일한 방어선입니다. 기본값: `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000`(단일 대기는 `maxIntervalMs`로 상한, 그 자체는 600000으로 상한), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | 키 인증 `openai-chat` 및 `openai-responses` 프로바이더 전용입니다. `authMode: "forward"` 프로바이더(ChatGPT 계정 풀)는 이 옵션을 읽지 않고 기본 재시도 단계를 유지합니다. 스트림 시작 전의 일시적인 업스트림 상태(500, 502, 503, 504, 520, 521, 522)를 선택적으로 재시도합니다. 이 옵션이 없으면 꺼져 있고, 객체가 있으면 `enabled: false`가 아닌 한 활성화됩니다. 최초 Responses 요청, 터미널 가드 연속 요청, 네이티브 `/v1/chat/completions`, 429/계정 복구 재조회를 포함합니다. `attempts`는 최초 전송을 포함하여 요청 하나에 허용되는 업스트림 전송의 총횟수(1..10, 기본값 3)입니다. 연결 재설정 복구와 요청 단위 예산 하나를 공유하므로 `3`이면 실제로 프로바이더에 도달하는 요청은 최대 세 번입니다. 대기에는 400ms로 고정된 지수 백오프를 사용하고 상한은 5초이며 `Retry-After`를 따릅니다. 속도 제한을 처리하는 `retryOn429`와는 별개이며, 스트림 도중의 실패는 절대 재전송하지 않습니다. | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | 네이티브 Responses 전송 전용(`adapter: "openai-responses"`, 정식 ChatGPT 백엔드 포함). 응답 바이트가 하나도 도착하기 전에 업스트림 연결이 닫혔을 때의 옵트인 재전송입니다. 없으면 꺼짐, 객체가 있으면 `enabled: false`가 아닌 한 켜짐. 꺼져 있으면 프록시는 이런 끊김에 [server](server.md)에 설명된 재전송 불가 HTTP 429 `upstream_reset_replay_refused`로 응답합니다. 켜져 있으면 프록시가 자기완결적이라고 판단할 수 있는 요청(`store: false`, 완전한 `input`, 클라이언트 실행 도구만(`function`, `custom`, client `tool_search`, 이들의 `namespace` 그룹), 그리고 `previous_response_id`, `conversation`, `background`, `stream_id` 없음)을 새 연결로 다시 보냅니다. `attempts`는 첫 전송을 포함해 재전송이 도달할 수 있는 총 전송 횟수(1..3, 기본 2)이며 해당 레그가 이미 가진 전송 예산을 넘지 않습니다. 재전송을 다 쓰거나 호출자 취소가 아닌 이유로 재전송이 실패하면 같은 거부 응답을 반환하며, 클라이언트에게 다시 보내라고 유도하는 상태 코드는 절대 반환하지 않습니다. 재전송 중의 취소는 취소 그대로 반환됩니다. 그 외 형태의 요청은 거부가 유지됩니다. 오리진이 첫 추론을 이미 시작했다면 재전송분도 과금될 수 있습니다. | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice`가 `auto` 또는 `none`만 받는 모델입니다. 강제 선택은 낮은 수준으로 바뀝니다. | | `preserveReasoningContentModels?` | `string[]` | chat 기록에서 이전 assistant `reasoning_content`가 필요한 모델입니다. | | `reasoningDetailsModels?` | `string[]` | thinking을 구조화된 `reasoning_details` 배열로 반환하는 모델(`reasoning_split` 사용 MiniMax M 시리즈). 스트림 델타는 누적 스냅샷이라 prefix-diff로 처리하고, 보존된 reasoning은 `reasoning_content` 문자열 대신 `reasoning_details` 배열로 리플레이합니다. | diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 6f780ed4ed..8536bdab61 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -212,6 +212,7 @@ Providers can expose a built-in shorthand, such as `agy` for `google-antigravity | `webSearchBridge?` | `{ enabled?: boolean; backend?: "ollama" \| "openai" \| "anthropic" \| "xai" \| "gemini" \| "exa"; maxSearches?: number; timeoutMs?: number; endpoint?: string }` | Key-auth `openai-responses` passthrough providers only. Off by default. Codex always declares the hosted `web_search` tool, and the passthrough relays it on the assumption the destination executes it. A gateway that does not run hosted search answers with a `function_call` named `web_search` that nothing runs, and the undeclared-tool guard ends the turn. With `enabled: true` and an explicit `backend` OpenCodex intercepts that call, runs the search itself, feeds the result back to the same upstream, and shows Codex a hosted `web_search_call` cell. Never armed for `authMode: "forward"` (ChatGPT already searches) or for a provider that executes hosted search upstream. `backend` is required; there is no implicit default and a missing credential for the named backend leaves the bridge disarmed rather than falling through to another paid search. `ollama` reuses this provider's own API key on `POST /api/web_search`, so the origin must be `https://ollama.com` unless the operator names `endpoint` explicitly. `openai` / `anthropic` / `xai` / `gemini` / `exa` reuse the matching sidecar executor and that executor's own credential (`webSearchSidecar.exaApiKey` for Exa). The search model comes from `webSearchSidecar.model` only when `webSearchSidecar.backend` resolves to the same backend this bridge names; otherwise the bridge runs that backend's own default, because a model chosen for one vendor is rejected by another. An unset `webSearchSidecar.backend` resolves to `openai`, so an unset-backend model reaches an `openai` bridge and no other. There is no per-provider bridge model override. Streaming turns only. A turn that mixes `web_search` with another client tool call still fails closed rather than dropping the client's call. Assistant text such as XML-like `` prose is not executed. Defaults: `maxSearches: 3` (1..10), `timeoutMs: 60000` (1000..600000). | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | API-key providers only (`authMode: "key"`). Opt-in same-target 429 retry: when `retryOn429` is absent the feature is off; object presence enables it unless `enabled: false`. On 429 the proxy waits (upstream `Retry-After` or the fixed interval) and replays the identical request on the same key before any key failover — across the main text-turn recovery loop, the Responses passthrough wire, the image/video bridge, the web-search sidecar, and terminal continuations. Only pre-stream HTTP 429 responses are eligible for replay; custom `runTurn` transports are outside the HTTP retry loop. `attempts` counts same-key replays after the first 429 (total sends = `attempts` + 1) and is one request-wide budget shared by the main recovery loop, the terminal-guard continuation, and bridge retries. Exhausting `attempts` only stops further same-key replays: normal key failover or final-error handling then applies per the available targets — on the key-auth passthrough wire there is no failover, so the exhausted 429 surfaces as-is. Codex itself never retries 429, so this is the only defense for single-key providers. Defaults: `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000` (any single wait is capped at `maxIntervalMs`, itself capped at 600000), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | Key-auth `openai-chat` and `openai-responses` providers only. `authMode: "forward"` providers (the ChatGPT account pool) never read this option and keep the default ladder. Opt-in retry for pre-stream transient upstream statuses (500, 502, 503, 504, 520, 521, 522): absent means off, object presence enables it unless `enabled: false`. Covers the initial Responses request, the Responses passthrough lane and each of its recovery legs (OAuth-401 replay, same-target 429 replay, validated rebuild), the terminal-guard continuation, and native `/v1/chat/completions`. `attempts` is the TOTAL number of upstream sends allowed for one request including the first (1..10, default 3) — it is one budget shared with connection-reset recovery, so `3` means at most three real requests reach the provider. On the Responses passthrough lane the configured value is additionally intersected with the request-wide send allowance, so a value below that allowance narrows the ladder exactly while a value above it does not raise the bound. Waits use a fixed 400 ms exponential backoff capped at 5 s and honor `Retry-After`. Separate from `retryOn429`, which handles rate limiting; mid-stream failures are never replayed. | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | Native Responses sends only (`adapter: "openai-responses"`, including the canonical ChatGPT backend). The decision is made once per request from the inbound body and threaded to every send of that request, the account move included, so the answer never depends on which leg reset. Opt-in replay when the upstream connection closes before any response byte: absent means off, object presence enables it unless `enabled: false`. Off, the proxy answers such a close with the non-replayable HTTP 429 `upstream_reset_replay_refused` described in [server](server.md). On, a request the proxy can judge self-contained — `store: false`, complete `input`, only client-executed tools (`function`, `custom`, client `tool_search`, `namespace` groups of those) and no `previous_response_id`, `conversation`, `background` or `stream_id` — is sent again on a fresh connection. `attempts` is the TOTAL number of sends the replay may reach including the first (1..3, default 2) and never exceeds the send budget the leg already has. When the replays are spent, or a replay fails for any reason other than a caller cancellation, the same refusal is returned, never a status that invites the client to resend. A cancellation during a replay surfaces as the cancellation itself. Any other request shape keeps the refusal. The replayed inference may be billed if the origin had already started the first one. | | `autoToolChoiceOnlyModels?` | `string[]` | Models whose `tool_choice` accepts only `auto` or `none`; forced choices are downgraded. | | `preserveReasoningContentModels?` | `string[]` | Models requiring prior assistant `reasoning_content` in chat history. | | `reasoningDetailsModels?` | `string[]` | Models whose endpoint returns thinking as a structured `reasoning_details` array (MiniMax M-series with `reasoning_split`); stream deltas are cumulative snapshots that are prefix-diffed, and preserved reasoning replays as a `reasoning_details` array instead of a `reasoning_content` string. | diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 6deae1998c..20c0292010 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -68,6 +68,18 @@ it, nor does it record the refusal as rate-limit or quota evidence against the c was holding. Tool-call side requests such as vision and web search are replayed normally, because repeating them cannot duplicate a turn. +This default is the shared failure model's verdict for a reset at the pre-header stage, not a +rule this path holds on its own. + +A native Responses provider can opt into replaying this case with +[`retryOnReset`](providers.md#provider-fields). The proxy then sends the request again on a +fresh connection when, and only when, the request is self-contained (`store: false`, complete +input, client-executed tools only, no server-side continuation state). Replays continue until +the configured total send count is reached, bounded by the send budget the leg already has, so +the default of two allows one replay. The refusal returns when those sends are spent or a +replay fails for another reason. A caller that cancels mid-replay gets the cancellation, not +the refusal. Any other request shape keeps the refusal. + `noProxy` accepts either a comma-separated string or an array. Both forms add entries without replacing an inherited `NO_PROXY`: 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 7107a9cb12..c3ea9ceaef 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -144,6 +144,7 @@ cross-route credential fallback не существует. Строки API GPT- | `responsesSnapshotRepair?` | `boolean` | По умолчанию выключенная клиентская repair для неполных lifecycle snapshot'ов Responses в SSE и JSON. Добавляет отсутствующие status, output и tool metadata, не меняя raw inspection и persistence. | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | Только для провайдеров с API-ключом (`authMode: "key"`). Опциональный повтор при 429 на том же таргете: если `retryOn429` отсутствует, функция выключена; наличие объекта включает её, если только `enabled: false`. При 429: ожидание (`Retry-After` апстрима или фиксированный интервал) и повтор идентичного запроса на том же ключе до любого фейловера ключей — покрывает основной цикл восстановления текстовых ходов, passthrough-канал Responses, мост изображений/видео, sidecar web-search и терминальные продолжения. Повтор допустим только для HTTP 429, полученных до начала потока; пользовательские транспорты `runTurn` не входят в цикл HTTP-повторов. `attempts` — это число повторов на том же ключе после первого 429 (всего отправок = `attempts` + 1) и единый бюджет на запрос, общий для основного цикла восстановления, терминального продолжения и повторов моста. Исчерпание `attempts` лишь останавливает дальнейшие повторы на том же ключе; далее применяется обычный фейловер ключей или финальная обработка ошибки в зависимости от доступных таргетов — на passthrough-канале с ключевой аутентификацией фейловера нет, поэтому исчерпанный 429 возвращается как есть. Codex сам никогда не повторяет 429, поэтому это единственная защита для провайдеров с одним ключом. По умолчанию: `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000` (любое ожидание ограничено `maxIntervalMs`, который сам ограничен 600000), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | Только для провайдеров `openai-chat` и `openai-responses` с аутентификацией по ключу. Провайдеры с `authMode: "forward"` (пул аккаунтов ChatGPT) никогда не читают эту настройку и сохраняют число повторов по умолчанию. Опциональный повтор при временных статусах апстрима до начала потока (500, 502, 503, 504, 520, 521, 522): если параметр отсутствует, функция выключена; наличие объекта включает её, если только `enabled: false`. Покрывает исходный запрос `Responses`, продолжение терминального предохранителя, нативный `/v1/chat/completions`, а также повторные запросы при восстановлении после 429 или ошибки учётной записи. `attempts` — ОБЩЕЕ число разрешённых отправок в апстрим для одного запроса, включая первую (1..10, по умолчанию 3). Это единый бюджет на запрос, общий с восстановлением после сброса соединения, поэтому `3` означает, что до провайдера дойдут не более трёх реальных запросов. Ожидание использует экспоненциальную задержку с фиксированной начальной величиной 400 мс, ограниченную 5 с, и учитывает `Retry-After`. Параметр не связан с `retryOn429`, который обрабатывает ограничение частоты запросов; сбои после начала потока никогда не воспроизводятся. | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | Только нативные отправки Responses (`adapter: "openai-responses"`, включая канонический бэкенд ChatGPT). Опциональная повторная отправка, когда апстрим закрывает соединение до первого байта ответа: отсутствие означает «выключено», наличие объекта включает, если не `enabled: false`. В выключенном состоянии прокси отвечает на такой обрыв неповторяемым HTTP 429 `upstream_reset_replay_refused`, описанным в [server](server.md). Во включённом запрос, который прокси может считать самодостаточным — `store: false`, полный `input`, только клиентские инструменты (`function`, `custom`, клиентский `tool_search`, их группы `namespace`) и без `previous_response_id`, `conversation`, `background` и `stream_id` — отправляется снова по новому соединению. `attempts` — ОБЩЕЕ число отправок, которого может достичь повтор, включая первую (1..3, по умолчанию 2); оно никогда не превышает бюджет отправок, уже имеющийся у этого этапа. Когда повторы исчерпаны или повтор не удался по любой причине, кроме отмены вызывающей стороной, возвращается тот же отказ и никогда — статус, приглашающий клиент отправить снова. Отмена во время повтора возвращается как сама отмена. Любая другая форма запроса сохраняет отказ. Повторный инференс может быть оплачен, если источник уже начал первый. | | `autoToolChoiceOnlyModels?` | `string[]` | Модели, у которых `tool_choice` принимает только `auto` или `none`; forced choice понижается. | | `preserveReasoningContentModels?` | `string[]` | Модели, которым нужен предыдущий assistant `reasoning_content` в chat history. | | `reasoningDetailsModels?` | `string[]` | Модели, чей endpoint возвращает thinking как структурированный массив `reasoning_details` (MiniMax M-series с `reasoning_split`); потоковые дельты — кумулятивные снимки, сравниваемые по префиксу, а сохранённый reasoning воспроизводится массивом `reasoning_details` вместо строки `reasoning_content`. | diff --git a/docs-site/src/content/docs/tr/reference/configuration/providers.md b/docs-site/src/content/docs/tr/reference/configuration/providers.md index 5a0c198d75..b8cfc76c4e 100644 --- a/docs-site/src/content/docs/tr/reference/configuration/providers.md +++ b/docs-site/src/content/docs/tr/reference/configuration/providers.md @@ -145,6 +145,7 @@ alanlı seçilmiş kimlikleri yalın kimliklere yeniden yazar. | `responsesSnapshotRepair?` | `boolean` | SSE ve JSON'daki seyrek Responses yaşam döngüsü anlık görüntüleri için varsayılan olarak devre dışı bırakılmış istemciye yönelik onarım. Ham inceleme ve kalıcılık değişmeden kalırken eksik kurallı durumu, çıktıyı ve araç meta verilerini doldurur. | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | Yalnızca API anahtarı sağlayıcıları (`authMode: "key"`). İsteğe bağlı aynı hedef 429 yeniden denemesi: `retryOn429` olmadığında özellik kapalıdır; nesnenin varlığı `enabled: false` olmadığı sürece özelliği etkinleştirir. 429'da proxy bekler (yukarı akış `Retry-After` veya sabit aralık) ve herhangi bir anahtar yük devretmesinden önce aynı istek üzerinde aynı anahtarla aynı isteği yeniden oynatır — ana metin turu kurtarma döngüsü, Responses doğrudan geçiş hattı, görsel/video köprüsü, web araması sidecar'ı ve terminal devamları genelinde. Yalnızca akış öncesi HTTP 429 yanıtları yeniden oynatma için uygundur; özel `runTurn` aktarımları HTTP yeniden deneme döngüsünün dışındadır. `attempts`, ilk 429'dan sonraki aynı anahtar yeniden oynatmalarını sayar (toplam gönderim = `attempts` + 1) ve ana kurtarma döngüsü, terminal koruma devamı ve köprü yeniden denemeleri tarafından paylaşılan tek bir istek genelinde bütçedir. `attempts`'ı tüketmek yalnızca daha fazla aynı anahtar yeniden oynatmasını durdurur: normal anahtar yük devretmesi veya nihai hata işleme daha sonra kullanılabilir hedeflere göre geçerli olur — anahtar kimlik doğrulamalı doğrudan geçiş hattında yük devretme yoktur, bu nedenle tükenen 429 olduğu gibi görünür. Codex'in kendisi 429'u asla yeniden denemez, bu nedenle tek anahtarlı sağlayıcılar için tek savunma budur. Varsayılanlar: `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000` (tek bir bekleme `maxIntervalMs` ile sınırlandırılır, kendisi de 600000 ile sınırlandırılır), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | Yalnızca anahtarla kimlik doğrulanan `openai-chat` ve `openai-responses` sağlayıcıları. `authMode: "forward"` sağlayıcıları (ChatGPT hesap havuzu) bu seçeneği hiç okumaz ve varsayılan merdiveni korur. Akış öncesi geçici yukarı akış durumları (500, 502, 503, 504, 520, 521, 522) için isteğe bağlı yeniden deneme: seçenek belirtilmezse kapalıdır; nesnenin varlığı, `enabled: false` olmadığı sürece özelliği etkinleştirir. İlk Responses isteğini, terminal koruma devamını, yerel `/v1/chat/completions` isteklerini ve 429/hesap kurtarma yeniden getirmelerini kapsar. `attempts`, bir istek için ilk gönderim dahil izin verilen yukarı akış gönderimlerinin TOPLAM sayısıdır (1..10, varsayılan 3) — bağlantı sıfırlama kurtarmasıyla paylaşılan, istek kapsamlı tek bütçedir; dolayısıyla `3`, sağlayıcıya en fazla üç gerçek isteğin ulaşması anlamına gelir. Beklemelerde 400 ms'lik sabit üstel geri çekilme uygulanır, süre 5 sn ile sınırlandırılır ve `Retry-After` dikkate alınır. Hız sınırlamasını işleyen `retryOn429` seçeneğinden ayrıdır; akış ortası hataları hiçbir zaman yeniden oynatılmaz. | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | Yalnızca native Responses gönderimleri (`adapter: "openai-responses"`, kanonik ChatGPT arka ucu dahil). Üst akış bağlantısı herhangi bir yanıt baytından önce kapandığında isteğe bağlı yeniden gönderim: yoksa kapalı, nesne varsa `enabled: false` olmadıkça açık. Kapalıyken proxy bu kopmaya [server](server.md) içinde anlatılan, yeniden gönderilemez HTTP 429 `upstream_reset_replay_refused` ile yanıt verir. Açıkken proxy'nin kendi başına yeterli sayabildiği bir istek — `store: false`, eksiksiz `input`, yalnızca istemci tarafında çalışan araçlar (`function`, `custom`, istemci `tool_search`, bunların `namespace` grupları) ve `previous_response_id`, `conversation`, `background` ya da `stream_id` olmadan — yapılandırılan toplam gönderim sayısına ulaşılana kadar yeni bir bağlantı üzerinden yeniden gönderilir. `attempts`, ilk gönderim dahil yeniden gönderimin ulaşabileceği TOPLAM gönderim sayısıdır (1..3, varsayılan 2) ve adımın zaten sahip olduğu gönderim bütçesini asla aşmaz. Yeniden gönderimler tükendiğinde ya da bir yeniden gönderim çağıran tarafın iptali dışında bir nedenle başarısız olduğunda aynı ret döndürülür; istemciyi yeniden göndermeye davet eden bir durum kodu asla döndürülmez. Yeniden gönderim sırasındaki bir iptal, iptalin kendisi olarak döner. Başka biçimdeki istekler reddedilmeye devam eder. Kaynak ilk çıkarımı zaten başlatmışsa, yeniden gönderilen çıkarım faturalandırılabilir. | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice`'u yalnızca `auto` veya `none` kabul eden modeller; zorunlu seçimlerin derecesi düşürülür. | | `preserveReasoningContentModels?` | `string[]` | Sohbet geçmişinde önceki asistan `reasoning_content`'ini gerektiren modeller. | | `reasoningDetailsModels?` | `string[]` | Thinking'i yapılandırılmış bir `reasoning_details` dizisi olarak döndüren modeller (`reasoning_split` ile MiniMax M-serisi); akış deltaları önek farkıyla işlenen kümülatif anlık görüntülerdir ve korunan reasoning, `reasoning_content` dizesi yerine `reasoning_details` dizisi olarak yeniden oynatılır. | 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 90ab7966d1..594826f353 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 @@ -131,6 +131,7 @@ selector，而不是分配一个新名称。 | `responsesSnapshotRepair?` | `boolean` | 默认关闭的客户端修复，用于补全 SSE 与 JSON 中稀疏 Responses 生命周期快照缺失的 status、output 和工具元数据；原始检查与持久化保持不变。 | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | 仅限 API-key 提供商（`authMode: "key"`）。可选的同目标 429 重试：未配置 `retryOn429` 时功能关闭；对象存在即启用，除非 `enabled: false`。收到 429 时等待（上游 `Retry-After` 或固定间隔）后在相同 key 上重放完全相同请求，再进入任何 key 故障转移——覆盖主文本恢复循环、Responses passthrough、图像/视频桥、web-search 侧车与终结续接。重放仅适用于流开始前的 HTTP 429 响应；自定义 `runTurn` 传输不在 HTTP 重试循环范围内。`attempts` 是首个 429 之后的同 key 重放次数（总发送次数 = `attempts` + 1），是主恢复循环、终结守卫续接与桥接重试共享的按请求统一预算；`attempts` 耗尽只会停止进一步的同 key 重放：随后按可用目标进行正常的 key 故障转移或最终错误处理——key 认证的 passthrough 线路上没有故障转移，因此耗尽的 429 会原样透出。Codex 自身从不重试 429，因此这是单 key 提供商唯一的防线。默认值：`enabled: true`、`attempts: 3`、`intervalMs: 5000`、`maxIntervalMs: 60000`（单次等待以 `maxIntervalMs` 为上限，其本身上限 600000）、`respectRetryAfter: true`。 | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | 仅限使用 key 认证的 `openai-chat` 与 `openai-responses` 提供商。`authMode: "forward"` 的提供商（ChatGPT 账号池）从不读取此选项，保持默认重试次数。可选的流开始前上游瞬态状态码（500、502、503、504、520、521、522）重试：未配置时关闭；对象存在即启用，除非 `enabled: false`。覆盖初始 Responses 请求、终结守卫续接、原生 `/v1/chat/completions`，以及 429/账户恢复重新获取。`attempts` 是单个请求允许向上游发送的总次数，包含首次发送（1..10，默认 3）；它是与连接重置恢复共享的按请求预算，因此 `3` 表示最多只有三个实际请求到达提供商。等待采用固定 400 毫秒的指数退避，上限为 5 秒，并遵循 `Retry-After`。此选项独立于处理速率限制的 `retryOn429`；流开始后的故障绝不会重放。 | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | 仅原生 Responses 发送（`adapter: "openai-responses"`，含规范的 ChatGPT 后端）。当上游连接在收到任何响应字节之前关闭时的可选重放：缺省为关闭，存在对象即启用，除非 `enabled: false`。关闭时，代理对这种断开返回 [server](server.md) 中描述的不可重放 HTTP 429 `upstream_reset_replay_refused`。开启时，代理能判定为自包含的请求（`store: false`、完整的 `input`、仅客户端执行的工具（`function`、`custom`、client `tool_search` 及其 `namespace` 分组），且没有 `previous_response_id`、`conversation`、`background` 或 `stream_id`）会通过新连接重新发送，直到达到配置的总发送次数。`attempts` 是重放可达到的总发送次数（含首次，1..3，默认 2），且永不超过该段已有的发送预算。重放用尽，或重放因调用方取消以外的原因失败时，返回同样的拒绝，绝不返回诱使客户端重发的状态码。重放期间的取消按取消本身返回。其他形状的请求保持拒绝。若源站已开始首次推理，重放的推理可能会被计费。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` 只接受 `auto` 或 `none` 的模型；强制选择会被降级。 | | `preserveReasoningContentModels?` | `string[]` | 需要在聊天历史中保留先前 assistant `reasoning_content` 的模型。 | | `reasoningDetailsModels?` | `string[]` | 以结构化 `reasoning_details` 数组返回思考内容的模型（启用 `reasoning_split` 的 MiniMax M 系列）；流式增量为累积快照，按前缀差分处理，保留的推理以 `reasoning_details` 数组而非 `reasoning_content` 字符串回放。 | diff --git a/docs-site/src/content/docs/zh-tw/reference/configuration/providers.md b/docs-site/src/content/docs/zh-tw/reference/configuration/providers.md index d90eea6e7c..b9bfbe4196 100644 --- a/docs-site/src/content/docs/zh-tw/reference/configuration/providers.md +++ b/docs-site/src/content/docs/zh-tw/reference/configuration/providers.md @@ -103,6 +103,7 @@ ocx models provider openrouter on | `parallelToolCalls?` | `boolean` | 切換平行工具呼叫。OpenAI Chat 預設開啟；非 chat adapter 僅在明確 `true` 時廣告。 | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | 預設停用的下游 SSE 修復，用於精確佔位 id 與缺失的終端 id。Function-call id 永不被重寫。 | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | 僅限使用金鑰認證的 `openai-chat` 與 `openai-responses` 供應商。`authMode: "forward"` 的供應商（ChatGPT 帳號池）從不讀取此選項，維持預設重試次數。選擇性重試串流開始前的暫時性上游狀態（500、502、503、504、520、521、522）：未設定時停用；只要有此物件即啟用，除非 `enabled: false`。涵蓋初始 `Responses` 請求、終止防護續接、原生 `/v1/chat/completions`，以及 429／帳號復原的重新擷取。`attempts` 是單一請求允許傳送至上游的總次數，包含第一次（1..10，預設 3）；這是與連線重設復原共用的單一請求範圍預算，因此 `3` 表示最多只有三個實際請求會送達供應商。等待採固定 400 毫秒、上限 5 秒的指數退避，並遵循 `Retry-After`。此機制獨立於處理速率限制的 `retryOn429`；串流中的失敗絕不重播。 | +| `retryOnReset?` | `{ enabled?: boolean; attempts?: number }` | 僅原生 Responses 傳送（`adapter: "openai-responses"`，含正規的 ChatGPT 後端）。當上游連線在收到任何回應位元組之前關閉時的選擇性重送：未設定為關閉，存在物件即啟用，除非 `enabled: false`。關閉時，代理對這種斷線回傳 [server](server.md) 所述、不可重送的 HTTP 429 `upstream_reset_replay_refused`。啟用時，代理能判定為自包含的請求（`store: false`、完整的 `input`、僅客戶端執行的工具（`function`、`custom`、client `tool_search` 及其 `namespace` 群組），且沒有 `previous_response_id`、`conversation`、`background` 或 `stream_id`）會以新連線重新傳送，直到達到設定的總傳送次數。`attempts` 是重送可達到的總傳送次數（含首次，1..3，預設 2），且永不超過該段既有的傳送預算。重送用盡，或重送因呼叫方取消以外的原因失敗時，回傳同樣的拒絕，絕不回傳誘使客戶端重送的狀態碼。重送期間的取消會以取消本身回傳。其他形狀的請求維持拒絕。若來源已開始首次推論，重送的推論可能會被計費。 | | `autoToolChoiceOnlyModels?` | `string[]` | 其 `tool_choice` 僅接受 `auto` 或 `none` 的模型；強制選擇被降級。 | | `preserveReasoningContentModels?` | `string[]` | 需要在 chat 歷史中保留先前 assistant `reasoning_content` 的模型。 | | `reasoningDetailsModels?` | `string[]` | 以結構化 `reasoning_details` 陣列回傳思考內容的模型（啟用 `reasoning_split` 的 MiniMax M 系列）；串流增量為累積快照，以前綴差分處理，保留的推理以 `reasoning_details` 陣列而非 `reasoning_content` 字串重播。 | diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index a8d43bc015..6098ab9e99 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1610,7 +1610,9 @@ "claude-intercept-settings.test.ts": "claude-integration", "claude-desktop-first-party.test.ts": "claude-integration", "claude-desktop-mode-explanation.test.ts": "claude-integration", - "claude-intercept-integration.test.ts": "server" + "claude-intercept-integration.test.ts": "server", + "responses-reset-replay.test.ts": "responses", + "management-provider-reset-replay.test.ts": "server" }, "migrated": [ "adapters", diff --git a/src/config.ts b/src/config.ts index 9788f25214..0c63a168ba 100644 --- a/src/config.ts +++ b/src/config.ts @@ -110,7 +110,7 @@ export { sanitizeModelCostsForDisplay, modelPreferHostedToolsConfigError, } from "./config/schema/leaf-validators"; -export { hardenExistingSecret, retryOn429PolicyConfigError } from "./config/load-degrade"; +export { hardenExistingSecret, retryOn429PolicyConfigError, retryOnResetPolicyConfigError } from "./config/load-degrade"; export { backupInvalidConfig } from "./config/salvage"; export type { ConfigDiagnostics, ConfigAdmissionSnapshot } from "./config/diagnostics"; export { diff --git a/src/config/load-degrade.ts b/src/config/load-degrade.ts index 1787ad109a..91ad938689 100644 --- a/src/config/load-degrade.ts +++ b/src/config/load-degrade.ts @@ -32,6 +32,7 @@ import { quotaResetNotifySchema, remoteGuiConfigSchema, retryOn429PolicySchema, + retryOnResetPolicySchema, runtimeRoleSchema, spendSchema, } from "./schema/leaf-validators"; @@ -195,18 +196,35 @@ export function sanitizeRetryOn429ForLoad(parsed: unknown): void { * redacted (a malformed write can place a secret in a property name). */ export function retryOn429PolicyConfigError(policy: unknown): string | null { + return strictPolicyConfigError("retryOn429", retryOn429PolicySchema, policy); +} + +/** + * Management write-boundary validation for `retryOnReset`, with the same fail-closed + * contract as `retryOn429PolicyConfigError`: the load-time schema degrades a malformed block + * to "absent", so this is the one place a bad value is refused instead of silently dropped. + */ +export function retryOnResetPolicyConfigError(policy: unknown): string | null { + return strictPolicyConfigError("retryOnReset", retryOnResetPolicySchema, policy); +} + +function strictPolicyConfigError( + field: string, + schema: { safeParse: (value: unknown) => { success: true } | { success: false; error: { issues: Array<{ code: string; message: string; path: PropertyKey[]; keys?: string[] }> } } }, + policy: unknown, +): string | null { if (policy === undefined) return null; - const result = retryOn429PolicySchema.safeParse(policy); + const result = schema.safeParse(policy); if (result.success) return null; const first = result.error.issues[0]; - if (!first) return "retryOn429 is invalid"; - if (first.code === "unrecognized_keys") { + if (!first) return `${field} is invalid`; + if (first.code === "unrecognized_keys" && first.keys) { const names = first.keys.map(key => JSON.stringify(redactSecretString(key))).join(", "); - return `retryOn429 has unrecognized field${first.keys.length > 1 ? "s" : ""}: ${names}`; + return `${field} has unrecognized field${first.keys.length > 1 ? "s" : ""}: ${names}`; } - if (first.path.length === 0) return `retryOn429 is invalid (${first.message})`; - const field = String(first.path[first.path.length - 1]); - return `retryOn429.${field} is invalid (${first.message})`; + if (first.path.length === 0) return `${field} is invalid (${first.message})`; + const last = String(first.path[first.path.length - 1]); + return `${field}.${last} is invalid (${first.message})`; } export function sanitizeCapabilityDeclarationsForLoad(parsed: unknown): void { diff --git a/src/config/schema/leaf-validators.ts b/src/config/schema/leaf-validators.ts index 0906b75671..474f3790d0 100644 --- a/src/config/schema/leaf-validators.ts +++ b/src/config/schema/leaf-validators.ts @@ -81,6 +81,17 @@ const transientRetryOn5xxPolicySchema = z.object({ attempts: z.number().int().min(1).max(10).optional(), }).strict(); +/** + * `retryOnReset` accepts only these keys. `attempts` is the TOTAL number of sends the replay + * may reach including the first, so the ceiling is the reset helper's own maximum + * (`RESET_RETRY_MAX_ATTEMPTS`, 3): a replay of a possibly-executed model POST is the one send + * this proxy otherwise refuses, and the operator gets at most two of them per leg. + */ +export const retryOnResetPolicySchema = z.object({ + enabled: z.boolean().optional(), + attempts: z.number().int().min(1).max(3).optional(), +}).strict(); + const requestPacingRuleSchema = z.object({ // Keep the RPM-derived timer within the same one-hour bound as minIntervalMs. requestsPerMinute: z.number().min(1 / 60).max(60_000).optional(), @@ -331,6 +342,10 @@ export const providerConfigSchema = z.object({ .optional(), retryOn429: retryOn429PolicySchema.optional(), transientRetryOn5xx: transientRetryOn5xxPolicySchema.optional(), + // Degrades to "absent" like `webSearchBridge`: a malformed hand edit of an opt-in feature + // that is off by default must not send the operator through invalid-config recovery. The + // management write boundary still rejects it loudly (`retryOnResetPolicyConfigError`). + retryOnReset: retryOnResetPolicySchema.optional().catch(undefined), codexAccountMode: z.enum(["pool", "direct"]).optional(), // Validated rather than passed through: this schema ends in `.passthrough()`, so an // undeclared key survives verbatim. A misspelled `codexToolMode` therefore used to be diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 8dd6f1146f..00f7df903d 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -59,6 +59,34 @@ export function isReplayRefusalResponse(response: Response): boolean { } /** Origin never produced a response event; the turn may still be executing. */ +import { + causeForRecoveryKind, + permitsResend, + resendPermission, + resendSendClass, +} from "./request-failure-model"; + +/** + * This failure, in the shared vocabulary: a `connection-reset` is `transport-ambiguous`, and a + * send that died before a response head is at the `pre-header` stage. + */ +const RESET_CAUSE = causeForRecoveryKind("connection-reset"); +/** + * Whether the table permits resending it on its own. It does not: the pair is + * `refused-ambiguous`, which is what makes the refusal below the default rather than a rule + * this module holds privately. That verdict forbids an AUTOMATIC resend and, by its own + * contract, leaves room for a bounded recovery an operator opted into. Reading it here means a + * change to the table reaches this path instead of drifting from it. + */ +const AUTOMATIC_RESET_REPLAY = permitsResend(resendPermission("pre-header", RESET_CAUSE)); +/** + * The budget class that cause draws on: `transient`, the same allowance every other send comes + * from. This is why `replayResets` is capped by `attempts` rather than added to it. One logical + * request funds one pool of sends, so a second bounded recovery cannot buy an independent + * replacement for the same request. + */ +const RESET_REPLAY_SEND_CLASS: ReturnType = resendSendClass(RESET_CAUSE); + export const UPSTREAM_NO_RESPONSE_CODE = "upstream_no_response"; /** Transport closed after the send, before any response event. */ export const UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE = "upstream_closed_before_response"; @@ -421,6 +449,19 @@ export interface ResetRetryOptions { * uncounted but UNCOUNTABLE: the callback existed on a type those call sites never reach. */ onSendsConsumed?: (sends: number) => void; + /** + * Operator-policy replay of a pre-header connection reset on a model POST: the TOTAL + * number of sends the replay may reach, including the first. Never above `attempts`; it + * says how much of the budget the leg already has a reset may spend, and widens nothing. + * + * This is not `replaySafe`. A replay-safe operation cannot duplicate anything, so it may + * rethrow when its retries run out and let the caller's own error path take over. A policy + * replay is a possibly duplicated inference the operator chose to risk, so once the replays + * are spent, or a later attempt fails any other way, the leg settles as the same + * non-replayable refusal a reset gets without the policy. Nothing on this path may hand a + * client a status that invites the whole turn to be sent again. + */ + replayResets?: number; } export interface TransientRetryOptions extends ResetRetryOptions { @@ -494,6 +535,22 @@ export function applyUpstreamRecoveryInit( return { ...init, headers, keepalive: false }; } +/** + * The refusal this proxy returns for a pre-header reset it will not replay. The WeakSet + * markers protect in-process recovery; the code survives JSON re-wrapping. The raw exception + * is never exposed, because it can carry credentials or request data. + */ +function replayRefusalResponse(): Response { + const response = new Response(JSON.stringify({ error: { + type: "upstream_error", + code: UPSTREAM_RESET_REPLAY_REFUSED_CODE, + message: "The upstream connection closed before a response was received. The request may already have been processed; automatic replay was stopped.", + } }), { status: REPLAY_REFUSED_STATUS, headers: { "content-type": "application/json" } }); + markResponseNonReplayable(response); + markReplayRefusalResponse(response); + return response; +} + /** * Run `doFetch` within one send budget. Connection-reset-shaped rejections are * terminal by default; only an explicitly replay-safe operation receives reset retries @@ -509,6 +566,14 @@ export async function fetchWithResetRetry( // more send on every recovery leg, which is most of what made a bounded per-layer retry // compose into an unbounded per-request count. if (attempts === 0) throw new SendBudgetExhaustedError(opts.label); + // An operator replay spends the budget this leg already has; it never adds to it. `attempts` + // measures the transient allowance, so the opt-in is funded only while the table keys this + // cause to that same class. If it ever moved, this leg would be spending the wrong budget, + // and not replaying is the safe reading of that. + const replayCeiling = opts.replayResets === undefined || RESET_REPLAY_SEND_CLASS !== "transient" + ? 0 + : Math.min(attempts, normalizeSendAttempts(opts.replayResets, 0)); + const policyReplay = opts.replaySafe !== true && replayCeiling > 0; let lastError: unknown; let sawReset = false; for (let attempt = 0; attempt < attempts; attempt++) { @@ -522,31 +587,35 @@ export async function fetchWithResetRetry( } catch (err) { if (opts.abortSignal?.aborted) throw err; if (!isConnectionResetError(err)) { + // A policy replay that already saw a reset is an ambiguous request whatever ended + // it: the first send may have run. Settle it as the refusal rather than throwing + // into a caller whose transport-failure path answers with a client-retryable 502. + if (sawReset && policyReplay) return replayRefusalResponse(); // A reset that already reached the origin is credential-visible // evidence: keep it attached so the terminal rejection cannot be // downgraded to the pre-connection neutral class (#914 review). if (sawReset) throw new UpstreamRetryEvidenceError([], err, true); throw err; } - if (opts.replaySafe !== true) { - // Return evidence instead of throwing a generic transport error: outer catches + // A replay-safe caller and the operator opt-in are the two bounded recoveries the + // `refused-ambiguous` verdict allows. With neither, the table's answer stands. + if (opts.replaySafe !== true && !policyReplay && !AUTOMATIC_RESET_REPLAY) { + // Return evidence instead of throwing a generic transport error. Outer catches // otherwise turn it into a replayable 502 and a combo/account recovery resends it. - // The WeakSet protects in-process recovery; the code survives JSON re-wrapping. - // Never expose the raw exception, which can contain credentials or request data. - const response = new Response(JSON.stringify({ error: { - type: "upstream_error", - code: UPSTREAM_RESET_REPLAY_REFUSED_CODE, - message: "The upstream connection closed before a response was received. The request may already have been processed; automatic replay was stopped.", - } }), { status: REPLAY_REFUSED_STATUS, headers: { "content-type": "application/json" } }); - markResponseNonReplayable(response); - markReplayRefusalResponse(response); - return response; + return replayRefusalResponse(); + } + if (opts.replaySafe !== true) { + // The opt-in spends its own ceiling, then answers exactly as the default would. + if (!policyReplay || attempt + 1 >= replayCeiling) return replayRefusalResponse(); + } else if (attempt === attempts - 1) { + throw err; } - if (attempt === attempts - 1) throw err; sawReset = true; lastError = err; console.warn( - `[upstream-retry] connection reset${opts.label ? ` (${opts.label})` : ""} — retrying (${attempt + 2}/${attempts})`, + `[upstream-retry] connection reset${opts.label ? ` (${opts.label})` : ""} — ${ + policyReplay ? `replaying (${attempt + 2}/${replayCeiling}, retryOnReset)` : `retrying (${attempt + 2}/${attempts})` + }`, ); await sleepWithAbort(retryBackoffDelayMs(attempt, { baseDelayMs: RESET_RETRY_BASE_DELAY_MS, diff --git a/src/providers/key-failover.ts b/src/providers/key-failover.ts index e4552df141..c8958297d7 100644 --- a/src/providers/key-failover.ts +++ b/src/providers/key-failover.ts @@ -13,7 +13,7 @@ import type { ProviderApiKeySelection } from "../types/provider"; import { routedProviderConfig } from "../router"; import { getProviderRegistryEntry } from "./registry"; import { normalizedBaseUrl } from "./quota/vendor-probes-key"; -import type { OcxConfig, OcxProviderConfig, RateLimitRetryPolicy, TransientRetryPolicy } from "../types"; +import type { OcxConfig, OcxProviderConfig, RateLimitRetryPolicy, ResetReplayPolicy, TransientRetryPolicy } from "../types"; import { OPENCODE_GO_SESSION_HEADER } from "./opencode-go-transport"; import { resolveProviderTransport, type OcxProviderTransport } from "./xai-transport"; import { sweepExpiredOnWrite } from "../lib/state-store-sweeper"; @@ -326,6 +326,15 @@ const DEFAULT_TRANSIENT_RETRY = { attempts: 3, } as const satisfies Required; +/** + * Default reset replay used when a provider opts in with a bare `retryOnReset: {}`: one + * replay. `attempts` is the TOTAL sends the replay may reach, not extra retries. + */ +const DEFAULT_RESET_REPLAY = { + enabled: true, + attempts: 2, +} as const satisfies Required; + /** Map<`${providerName}\0${keyId}`, KeyCooldown> */ const keyCooldowns = new Map(); @@ -616,6 +625,23 @@ export function transientRetryPolicyFor( }; } +/** + * Normalize a provider's `retryOnReset` policy, or return null when it is absent or + * explicitly disabled. No auth-mode gate: the canonical ChatGPT backend is `forward` auth and + * is the send this policy exists for. Whether a given request may actually be replayed is a + * per-body decision made by `selfContainedResponsesBody`, not by the provider. + */ +export function resetReplayPolicyFor( + provider: Pick, +): Required | null { + const policy = provider.retryOnReset; + if (!policy || policy.enabled === false) return null; + return { + enabled: policy.enabled ?? DEFAULT_RESET_REPLAY.enabled, + attempts: policy.attempts ?? DEFAULT_RESET_REPLAY.attempts, + }; +} + /** * Wait before the next same-target replay: upstream Retry-After (seconds or HTTP-date) when * `respectRetryAfter` is on and the header parses, capped at `maxIntervalMs`; otherwise the diff --git a/src/server/auth-cors.ts b/src/server/auth-cors.ts index ad987bf4e5..bc121ae23b 100644 --- a/src/server/auth-cors.ts +++ b/src/server/auth-cors.ts @@ -11,6 +11,7 @@ import { providerWebSearchBridgeConfigError, requestPacingConfigError, retryOn429PolicyConfigError, + retryOnResetPolicyConfigError, sanitizeModelCostsForDisplay, } from "../config"; import { @@ -723,6 +724,9 @@ export function providerManagementConfigError( delete canonicalCandidate.modelCosts; // requestPacing is a user-owned transport overlay, not part of the canonical seed. delete canonicalCandidate.requestPacing; + // retryOnReset is the same kind of overlay and is meant for this provider's native + // Responses sends; it is validated below (retryOnResetPolicyConfigError). + delete canonicalCandidate.retryOnReset; // Context windows are the same kind of user-owned overlay as requestPacing: the operator // narrowing what their own native rows advertise. They can only ever LOWER the measured // window (see nativeOpenAiContextWindow), so admitting them cannot widen what the proxy @@ -769,6 +773,10 @@ export function providerManagementConfigError( // it before it reaches the management API response. return `provider ${JSON.stringify(redactSecretString(name))} ${retryOn429Error}`; } + const retryOnResetError = retryOnResetPolicyConfigError(raw.retryOnReset); + if (retryOnResetError) { + return `provider ${JSON.stringify(redactSecretString(name))} ${retryOnResetError}`; + } const requestPacingError = requestPacingConfigError(raw.requestPacing); if (requestPacingError) { return `provider ${JSON.stringify(redactSecretString(name))} ${requestPacingError}`; @@ -1049,6 +1057,7 @@ const PROVIDER_CONFIG_FIELD_POLICY = { showThinkingSummary: "editor", retryOn429: "editor", transientRetryOn5xx: "editor", + retryOnReset: "editor", reasoningSplitModels: "editor", reasoningDetailsModels: "editor", thinkingToggleModels: "editor", diff --git a/src/server/responses/core-codex-account.ts b/src/server/responses/core-codex-account.ts index 99796846d8..3a05d8ed46 100644 --- a/src/server/responses/core-codex-account.ts +++ b/src/server/responses/core-codex-account.ts @@ -17,6 +17,7 @@ import { resetUpstreamHostHealth, } from "../../codex/upstream-host-health"; import { safeOriginLabel, fetchWithHeaderTimeout, providerFetch } from "./fetch-helpers"; +import { fetchWithResetRetry } from "../../lib/upstream-retry"; import { classifyPoolRecoveryDispatch } from "../../routing/probe-lease"; import { formatErrorResponse } from "../../bridge"; import { readBoundedResponseBody } from "../../lib/bounded-body"; @@ -352,6 +353,20 @@ export interface CodexPoolAccountRetryArgs { /** Root workflow this turn belongs to, so the move is charged there as well. */ workflowRootId?: string; }; + /** + * The reset-replay decision this request already made, plus the counters the dispatcher owns. + * + * Passed rather than recomputed. This leg rebuilds the request for another account, and + * judging a rebuilt body could reach a different answer than the one the request started + * with, which would make the behaviour depend on which account leg reset. `attempts` and + * `noteSendsConsumed` are the same request-wide budget every other send draws on, so a replay + * here cannot buy a send the logical request has not got. + */ + resetReplay?: { + options: { replayResets?: number }; + attempts: () => number; + noteSendsConsumed: (sends: number) => void; + }; firstAuthCtx: Extract; firstResponse: Response; outcomeStatus: number; @@ -800,35 +815,60 @@ export async function retryCodexPoolOnAlternateAccount( // The move is a physical send like any other, so the root workflow is charged too. chargeWorkflowSends(args.options.workflowRootId, 1); } - noteProviderAttemptSend(logCtx, route.providerName, route.provider, passthroughEstimate); + const movedAuthCtx = retryAuthCtx; + const sendOnce = (): Promise => fetchWithHeaderTimeout( + request.url, + { + method: request.method, + headers: request.headers, + body: request.body, + }, + upstream.signal, + connectMs, + stream, + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + providerName: route.providerName, + modelId: route.modelId, + onCodexWsQuota: codexWsQuotaObserver(movedAuthCtx, route.provider, route.modelId), + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexReserveDispatchGuard(movedAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + }), + // Credential-bearing forward send: never follow a redirect into a + // dead-host rejection after the credential was seen (#914). + route.provider.authMode === "forward", + ); + // The move goes through the shared reset layer like every other send. Without a policy + // that layer changes nothing but the answer to a pre-header reset: the refusal this + // request would get on any other leg, instead of a transport throw the caller turns into + // a client-retryable 502 for an ambiguous send. + let movePhysicalSends = 0; try { - upstreamResponse = await fetchWithHeaderTimeout( - request.url, - { - method: request.method, - headers: request.headers, - body: request.body, + upstreamResponse = await fetchWithResetRetry(sendOnce, { + abortSignal: options.abortSignal, + label: safeOriginLabel(request.url), + // The account move already reserved one send. Anything past it is an extra physical + // send, so it is measured against what the logical request has left. + attempts: Math.max(1, args.resetReplay?.attempts() ?? 1), + onSendsConsumed: sends => { + for (let i = 0; i < sends; i += 1) { + noteProviderAttemptSend(logCtx, route.providerName, route.provider, passthroughEstimate); + } + // The first send is the one the move permit and the workflow charge already bought. + const extra = Math.max(0, movePhysicalSends + sends - 1); + movePhysicalSends += sends; + if (extra > 0) { + args.resetReplay?.noteSendsConsumed(extra); + chargeWorkflowSends(args.options.workflowRootId, extra); + } }, - upstream.signal, - connectMs, - stream, - providerFetch(route.provider, options.codexWsRuntimeIdentity, { - providerName: route.providerName, - modelId: route.modelId, - onCodexWsQuota: codexWsQuotaObserver(retryAuthCtx, route.provider, route.modelId), - beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(retryAuthCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, - }), - // Credential-bearing forward send: never follow a redirect into a - // dead-host rejection after the credential was seen (#914). - route.provider.authMode === "forward", - ); + ...(args.resetReplay?.options ?? {}), + }); } catch (error) { // Only the forward send is a transport boundary. Entitlement resolver throws below are // deliberately outside this catch so programming errors retain their original path. return { kind: "transport", error, authCtx: retryAuthCtx }; } - retrySendCount += 1; + retrySendCount += Math.max(1, movePhysicalSends); args.onResponse?.(upstreamResponse, retryAuthCtx, request); // The alternate account can refuse the same model, and that refusal is evidence about the // account that produced it. Read BEFORE the ladder's own break, so the ordinary diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 6676bbd1d4..c8cff18774 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -126,6 +126,7 @@ import { rateLimitRetryDelayMs, transientRetryPolicyFor, } from "../../providers/key-failover"; +import { resetReplayOptions } from "./reset-replay"; import type { AttemptRecoveryKind } from "../../usage/log"; import { resolveWireProtocolOverride } from "../adapter-resolve"; import { refreshPoolForwardAuth, refreshNativeMainForwardAuth, withClaudeNativeSession } from "./core-auth"; @@ -809,6 +810,11 @@ export async function preparePassthroughExchange( }; const initialBodyRefusal = refuseOversizedOutboundBody(request); if (initialBodyRefusal) return initialBodyRefusal; + // Decided once for the request and carried by every leg below: the rotation, refresh and + // same-target 429 legs rebuild the request but send the same turn, so a reset on any of + // them is the same question. Empty unless the provider opted in AND the body is one the + // proxy can judge self-contained (see reset-replay.ts). + const resetReplay = resetReplayOptions(route.provider, parsed._rawBody); try { // Transient-5xx pre-stream retry (devlog/_plan/260716_claudecode_hardening/010): // the ChatGPT backend emits transient 502/520s that an immediate retry absorbs. @@ -851,6 +857,7 @@ export async function preparePassthroughExchange( // failing the turn. Recovery legs keep the fail-closed refusal; only this // initial send is replay-eligible. Attempts stay budget-bounded via attempts. replaySafe: isOpenCodeGoDestination(route.provider), + ...resetReplay, }, ); } catch (err) { @@ -949,7 +956,7 @@ export async function preparePassthroughExchange( route.provider.authMode === "forward") .then(adoptObservedResponse); }, - { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: allowance.attempts, onSendsConsumed: noteTransientSends }, + { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: allowance.attempts, onSendsConsumed: noteTransientSends, ...resetReplay }, ); } catch (err) { return { failed: transportFailureResponse(err) }; @@ -1030,35 +1037,47 @@ export async function preparePassthroughExchange( // every other build site; a replay is exactly when a grown payload reappears. const replayBodyRefusal = refuseOversizedOutboundBody(request); if (replayBodyRefusal) return replayBodyRefusal; - transportState.noteRoutedAttemptSend(passthroughEstimate, "oauth-401"); - upstreamResponse = await fetchWithHeaderTimeout( - request.url, - { method: request.method, headers: request.headers, body: request.body }, - upstream.signal, - connectMs, - parsed.stream, - // The replay-dispatched signal is what bounds the rest of this logical request, so it - // has to describe a send that actually happened. fetchWithHeaderTimeout awaits pacing - // admission BEFORE calling the executor, so signalling at the call site would spend the - // budget even when a rejected pacing wait means nothing reaches the network. Wrapping - // the executor moves the signal to the last moment before the send, where a throw from - // here on is a genuine transport attempt. - storedPoolReplayDispatchNotifier( - providerFetch(route.provider, options.codexWsRuntimeIdentity, { - nativeControl: nativeResponseControlEligible(route.provider, options.nativeControl) && options.inboundTransport === "websocket" && !options.comboAttempt - && responseEffects.plaintextV2AgentMessageToolNames.size === 0 - ? options.nativeControl : undefined, - dispatchOverride: oauthDispatch(request), - providerName: route.providerName, - modelId: route.modelId, - onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), - beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) - ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, - }), - codex401ReplayKind === "stored" ? options.onStoredPool401ReplayDispatched : undefined, - ), - route.provider.authMode === "forward", - ).then(adoptObservedResponse); + // The replay-dispatched signal is what bounds the rest of this logical request, so it + // has to describe a send that actually happened. fetchWithHeaderTimeout awaits pacing + // admission BEFORE calling the executor, so signalling at the call site would spend the + // budget even when a rejected pacing wait means nothing reaches the network. Wrapping + // the executor moves the signal to the last moment before the send, where a throw from + // here on is a genuine transport attempt. The notifier is built once for the whole leg: + // it fires on the first dispatch, and a reset replay is another send of the same replay, + // not a second one to announce. + const oauthReplayExecutor = storedPoolReplayDispatchNotifier( + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeControl: nativeResponseControlEligible(route.provider, options.nativeControl) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeControl : undefined, + dispatchOverride: oauthDispatch(request), + providerName: route.providerName, + modelId: route.modelId, + onCodexWsQuota: codexWsQuotaObserver(admissionState.authCtx, route.provider, route.modelId), + beforeDispatch: isCanonicalOpenAiForwardProvider(route.provider) + ? createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config, route.modelId, options.admission, options.visionDescribeTerminal === true) : undefined, + }), + codex401ReplayKind === "stored" ? options.onStoredPool401ReplayDispatched : undefined, + ); + upstreamResponse = await fetchWithTransientRetry( + recovery => { + transportState.noteRoutedAttemptSend(passthroughEstimate, recovery ?? "oauth-401"); + return fetchWithHeaderTimeout( + request.url, + applyUpstreamRecoveryInit({ + method: request.method, + headers: request.headers, + body: request.body, + }, recovery), + upstream.signal, + connectMs, + parsed.stream, + oauthReplayExecutor, + route.provider.authMode === "forward", + ).then(adoptObservedResponse); + }, + { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends, ...resetReplay }, + ); } catch (err) { return transportFailureResponse(err); } finally { @@ -1182,7 +1201,7 @@ export async function preparePassthroughExchange( route.provider.authMode === "forward") .then(adoptObservedResponse); }, - { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends }, + { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends, ...resetReplay }, ); } catch (err) { return transportFailureResponse(err); @@ -1314,7 +1333,7 @@ export async function preparePassthroughExchange( route.provider.authMode === "forward") .then(adoptObservedResponse); }, - { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends }, + { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends, ...resetReplay }, ); } catch (err) { return transportFailureResponse(err); @@ -1412,6 +1431,12 @@ export async function preparePassthroughExchange( firstResponse: upstreamResponse, outcomeStatus: poolRetryOutcome, sameAccountOnly: storedReplaySpent, + // The decision this request already made, with the counters it already spends from. + resetReplay: { + options: resetReplay, + attempts: () => remainingTransientSendBudget(transientSendAttempts()), + noteSendsConsumed: noteTransientSends, + }, upstream, connectMs, passthroughEstimate, diff --git a/src/server/responses/reset-replay.ts b/src/server/responses/reset-replay.ts new file mode 100644 index 0000000000..597f077bae --- /dev/null +++ b/src/server/responses/reset-replay.ts @@ -0,0 +1,91 @@ +/** + * Operator-opted replay of a native Responses send that died before any response byte. + * + * `fetchWithResetRetry` refuses to send a model POST again after a pre-header connection + * reset, because no response is not proof the origin never processed the request. This + * module is the one place that decision can be overridden, and it is deliberately narrow on + * both axes: the provider has to opt in (`retryOnReset`), and the request has to be one whose + * second send cannot do more than run the same inference again. Anything the proxy cannot + * judge keeps the refusal. + * + * The judgment is made on the inbound body the client sent, which is already parsed. It is + * conservative for the outbound request: the proxy expands `previous_response_id` and lowers + * hosted tools into client execution, so every hazard that reaches the wire was visible here, + * and a hazard visible here may already have been removed. A cheap fail-closed answer beats + * re-parsing a multi-megabyte outbound body on every send. + */ +import type { OcxProviderConfig } from "../../types"; +import { resetReplayPolicyFor } from "../../providers/key-failover"; + +/** Input items a client owns end to end: replaying them re-runs nothing but the model. */ +const CLIENT_INPUT_ITEM_TYPES: ReadonlySet = new Set([ + "message", "reasoning", "compaction", + "function_call", "function_call_output", + "custom_tool_call", "custom_tool_call_output", + "tool_search_call", +]); +const MESSAGE_ROLES: ReadonlySet = new Set(["user", "assistant", "system", "developer"]); +/** Bounded traversal: a catalog is operator data, not a reason to walk forever. */ +const MAX_TOOL_ENTRIES = 4096; +const MAX_TOOL_DEPTH = 4; + +function record(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +/** + * True when every tool in the catalog is executed by the client. Hosted tools (`web_search`, + * `mcp`, `code_interpreter`, ...) run on the origin during the turn, so an unknown or hosted + * type fails the whole catalog rather than being skipped: a tool this proxy does not + * recognise is a tool it cannot vouch for. + */ +function clientExecutedTools(tools: unknown, budget: { remaining: number }, depth = 0): boolean { + if (!Array.isArray(tools) || depth > MAX_TOOL_DEPTH) return false; + return tools.every(tool => { + budget.remaining -= 1; + if (budget.remaining < 0 || !record(tool)) return false; + if (tool.type === "function" || tool.type === "custom") return true; + if (tool.type === "tool_search") return tool.execution === "client"; + return tool.type === "namespace" && typeof tool.name === "string" + && clientExecutedTools(tool.tools, budget, depth + 1); + }); +} + +/** + * A Responses body whose second send can only repeat the inference: nothing stored, no + * server-side continuation state, complete input, and only client-executed tools. Deferred + * tool declarations inside `input` are checked by the same rule as the root catalog, so a + * hosted tool cannot ride in through `additional_tools` or a `tool_search_output`. + */ +export function selfContainedResponsesBody(body: unknown): boolean { + if (!record(body)) return false; + if (body.store !== false || body.background === true) return false; + if (body.previous_response_id != null || body.conversation != null || Object.hasOwn(body, "stream_id")) return false; + const input = body.input; + if (typeof input !== "string" && !Array.isArray(input)) return false; + const budget = { remaining: MAX_TOOL_ENTRIES }; + if (body.tools !== undefined && !clientExecutedTools(body.tools, budget)) return false; + if (typeof input === "string") return true; + return input.every(item => { + if (!record(item)) return false; + if (item.type === "additional_tools" || item.type === "tool_search_output") { + return clientExecutedTools(item.tools, budget); + } + if (item.type === undefined) return typeof item.role === "string" && MESSAGE_ROLES.has(item.role); + return typeof item.type === "string" && CLIENT_INPUT_ITEM_TYPES.has(item.type); + }); +} + +/** + * The `replayResets` option for one native Responses leg, or nothing. Computed once per + * request from the provider policy and the inbound body, then spread into every send of that + * request: the rebuilt legs (rotation, refresh, same-target 429 wait) carry the same turn. + */ +export function resetReplayOptions( + provider: Pick, + inboundBody: unknown, +): { replayResets: number } | Record { + const policy = resetReplayPolicyFor(provider); + if (policy === null || !selfContainedResponsesBody(inboundBody)) return {}; + return { replayResets: policy.attempts }; +} diff --git a/src/types.ts b/src/types.ts index 747fc17c75..76dd25f3ca 100644 --- a/src/types.ts +++ b/src/types.ts @@ -106,6 +106,8 @@ export type { ResponsesItemIdRepairConfig, RateLimitRetryPolicy, TransientRetryPolicy, + + ResetReplayPolicy, ProviderWebSearchBridgeBackend, ProviderWebSearchBridgeConfig, ProviderCostOverlay, diff --git a/src/types/provider.ts b/src/types/provider.ts index 5e0bd79b58..262b1e38b3 100644 --- a/src/types/provider.ts +++ b/src/types/provider.ts @@ -67,6 +67,26 @@ export interface TransientRetryPolicy { attempts?: number; } +/** + * Opt-in replay of a native Responses send whose upstream connection closed before any + * response byte (`providers..retryOnReset`). + * + * Disabled unless the object is present; a bare `{}` opts in with defaults. Only a request + * the proxy can judge self-contained is ever replayed; see + * `src/server/responses/reset-replay.ts`. The replayed inference may still be billed if the + * origin had already started the first one. + */ +export interface ResetReplayPolicy { + /** Master switch. Presence of the object also enables the policy (default true). */ + enabled?: boolean; + /** + * TOTAL upstream sends the replay may reach for one leg, including the first (1..3, + * default 2). It never widens the send budget the leg already has; it only says how much + * of that budget a connection reset may spend. + */ + attempts?: number; +} + /** * Same-target 429 wait-and-retry policy (`providers..retryOn429`). When present and not * explicitly disabled, the proxy waits and replays the identical request on the same key before @@ -914,6 +934,12 @@ export interface OcxProviderConfig { * with defaults. Key-auth `openai-chat` only. */ transientRetryOn5xx?: TransientRetryPolicy; + /** + * Opt-in replay of a native Responses send that died before any response byte + * (`providers..retryOnReset`). Disabled unless present; a bare `{}` opts in with + * defaults. Native `openai-responses` sends only, and only for self-contained requests. + */ + retryOnReset?: ResetReplayPolicy; /** * Model ids whose OpenAI-compatible chat endpoint accepts `reasoning_split: true` and returns * thinking separately in `reasoning_content` / `reasoning_details` instead of visible content. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 09bf0d1ccb..299e4b4068 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -849,6 +849,40 @@ compact, and native Chat — are deliberately not opted in. Adapters with their `fetchResponse` (kiro, cursor, google) keep their own retry policies; kiro imports the shared abort/sleep helpers from this module. +One operator opt-in reaches a model-POST path, and it is not `replaySafe`. + +The default is not a rule this path holds privately. In the shared vocabulary +(`src/lib/request-failure-model.ts`) a connection reset is `transport-ambiguous` and a send +that died before a response head is at the `pre-header` stage; that pair is +`refused-ambiguous`, which is what the refusal restates. `fetchWithResetRetry` reads the table +rather than repeating it, so a change there reaches this path. `refused-ambiguous` forbids an +AUTOMATIC resend and, by its own contract, leaves room for a bounded recovery an operator +opted into. This is that recovery. + +`providers..retryOnReset` (`src/providers/key-failover.ts::resetReplayPolicyFor`) lets +the native Responses passthrough pass `replayResets` on every send of a request — initial, +rotation, forward-auth 401 refresh, OAuth rotation, same-target 429 and the alternate-account +move alike — when the inbound body is one the proxy can judge self-contained: `store: false`, a +complete `input`, only client-executed tools and no `previous_response_id`, `conversation`, +`background` or `stream_id` (`src/server/responses/reset-replay.ts::selfContainedResponsesBody`, +fail-closed on any unknown tool or item type). The judgment is made once per request and threaded +to each leg, including the account move in +`src/server/responses/core-codex-account.ts::retryCodexPoolOnAlternateAccount`, so the answer +cannot depend on which leg reset. + +`replayResets` is a ceiling inside the leg's existing `attempts`, never an addition to it. The +table keys `transport-ambiguous` to the `transient` send class, which is the allowance +`attempts` measures, and the helper funds the opt-in only while that stays true. One logical +request therefore funds one pool of sends: under the default three-send allowance a request that +already moved accounts has nothing left for a replay, and the refusal stands. Two bounded +recoveries cannot each buy an independent replacement send for one request. + +It settles differently from a replay-safe retry: once the ceiling is reached, or a later attempt +of that leg fails any other way, the helper returns the same refusal a reset gets without the +policy, so no exit of this path can hand the client a status that invites the turn to be sent +again. A caller cancellation during a replay surfaces as the cancellation. The generic adapter +dispatch, its continuation, compact and native Chat keep the refusal unconditionally. + ## Console upload rejection recovery `src/providers/opencode-zen-rate-limit.ts` recognizes the complete Console upload-rejection envelope only at the effective HTTPS opencode.ai Zen/Go generation endpoint. A provider row name cannot authorize another destination. The two recovery loops in `src/server/responses/core.ts` wait 800 ms and replay the captured serialized request once; cancellation, nonreplayable responses, other errors and a second upload rejection keep their failure semantics. The recovery kind is persisted as `console-go-upload-retry` and has a localized Logs label. @@ -1250,7 +1284,10 @@ turn up to four more times, and a 429 is where the client stops. `upstream_reset_replay_refused`. No response headers is not evidence that the model POST was never processed, so the decision not to replay is ours, made before any response existed — the same shape as `request_send_budget_exhausted`, and it takes the same status -for the same reason. Only an explicitly replay-safe operation opts into reset retries. +for the same reason. Only an explicitly replay-safe operation opts into reset retries, or a +provider the operator opted in through `retryOnReset` for a request the proxy judged +self-contained; that replay spends the leg's own send budget and ends in this refusal when +it is spent (see [upstream reset retry](#upstream-reset-retry)). **An upstream reset observed mid-stream or after a terminal keeps its existing behaviour.** The passthrough read path still settles a genuine upstream reset as a synthetic 502, and the diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 53fc4cbab3..dd627bb06f 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1442,5 +1442,7 @@ "claude-intercept-settings.test.ts": "claude-integration", "claude-desktop-first-party.test.ts": "claude-integration", "claude-desktop-mode-explanation.test.ts": "claude-integration", - "claude-intercept-integration.test.ts": "server" + "claude-intercept-integration.test.ts": "server", + "responses-reset-replay.test.ts": "responses", + "management-provider-reset-replay.test.ts": "server" } diff --git a/tests/lib/upstream-retry.test.ts b/tests/lib/upstream-retry.test.ts index e8d09b309e..d0e2c40166 100644 --- a/tests/lib/upstream-retry.test.ts +++ b/tests/lib/upstream-retry.test.ts @@ -509,6 +509,86 @@ describe("ambiguous reset safety", () => { }); }); +describe("operator reset replay (replayResets)", () => { + test("replays a reset once and returns the second attempt's response", async () => { + silenceWarn(); + const reports: number[] = []; + const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]); + const res = await fetchWithResetRetry(mock.doFetch, { + attempts: 3, replayResets: 2, label: "test", onSendsConsumed: count => reports.push(count), + }); + expect(res.status).toBe(200); + expect(mock.calls).toHaveLength(2); + expect(reports).toEqual([1, 1]); + expect(warnSpies[0]).toHaveBeenCalledTimes(1); + expect(String(warnSpies[0]!.mock.calls[0]?.[0])).toContain("retryOnReset"); + }); + + test("a spent ceiling settles as the refusal, never a throw", async () => { + silenceWarn(); + const mock = mockDoFetch([bunResetError(), bunResetError(), new Response("duplicate")]); + const response = await fetchWithResetRetry(mock.doFetch, { attempts: 3, replayResets: 2 }); + expect(response.status).toBe(429); + expect(isNonReplayableResponse(response)).toBe(true); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); + // Two sends: the original and the one replay. The third result was never requested. + expect(mock.calls).toHaveLength(2); + }); + + test("the ceiling never widens the leg's own budget", async () => { + const mock = mockDoFetch([bunResetError(), new Response("duplicate")]); + const response = await fetchWithResetRetry(mock.doFetch, { attempts: 1, replayResets: 3 }); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); + expect(mock.calls).toHaveLength(1); + }); + + test("a replay that fails any other way is still the refusal, not a transport rejection", async () => { + silenceWarn(); + const refused = Object.assign(new Error("Unable to connect"), { code: "ECONNREFUSED" }); + const mock = mockDoFetch([bunResetError(), refused]); + // Without the policy the same sequence rethrows as credential-visible evidence, and the + // caller's transport path would answer with a 502 the client is invited to resend. + const response = await fetchWithResetRetry(mock.doFetch, { attempts: 3, replayResets: 3 }); + expect(response.status).toBe(429); + expect(isNonReplayableResponse(response)).toBe(true); + expect(mock.calls).toHaveLength(2); + }); + + test("a zero ceiling is the plain refusal and a replay-safe caller is unaffected", async () => { + silenceWarn(); + const off = mockDoFetch([bunResetError(), new Response("duplicate")]); + const refusal = await fetchWithResetRetry(off.doFetch, { attempts: 3, replayResets: 0 }); + expect(refusal.status).toBe(429); + expect(off.calls).toHaveLength(1); + const safe = mockDoFetch([bunResetError(), bunResetError(), bunResetError()]); + await expect(fetchWithResetRetry(safe.doFetch, { replaySafe: true, replayResets: 1 })) + .rejects.toThrow("socket connection was closed unexpectedly"); + expect(safe.calls).toHaveLength(3); + }); + + test("shares one budget with the transient layer", async () => { + silenceWarn(); + const reports: number[] = []; + const mock = mockDoFetch([ + new Response("busy", { status: 503 }), bunResetError(), new Response("ok", { status: 200 }), + ]); + const res = await fetchWithTransientRetry(mock.doFetch, { + attempts: 3, replayResets: 3, onSendsConsumed: count => reports.push(count), + }); + expect(res.status).toBe(200); + expect(mock.calls).toHaveLength(3); + expect(reports).toEqual([3]); + const spent = mockDoFetch([ + new Response("busy", { status: 503 }), new Response("busy", { status: 503 }), bunResetError(), new Response("ok"), + ]); + const response = await fetchWithTransientRetry(spent.doFetch, { attempts: 3, replayResets: 3 }); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); + expect(spent.calls).toHaveLength(3); + }); +}); + describe("ambiguous reset safety through error formatting", () => { test("every terminal code survives formatting without advertising Retry-After", async () => { for (const code of ["upstream_no_response", "upstream_closed_before_response", "upstream_reset_replay_refused"]) { diff --git a/tests/providers/upstream-transient-retry.test.ts b/tests/providers/upstream-transient-retry.test.ts index 516f31850a..80951187d5 100644 --- a/tests/providers/upstream-transient-retry.test.ts +++ b/tests/providers/upstream-transient-retry.test.ts @@ -6,7 +6,7 @@ import { isTransientUpstreamStatus, markResponseNonReplayable, } from "../../src/lib/upstream-retry"; -import { transientRetryPolicyFor } from "../../src/providers/key-failover"; +import { resetReplayPolicyFor, transientRetryPolicyFor } from "../../src/providers/key-failover"; import { handleChatCompletions } from "../../src/server/chat-completions"; import type { OcxConfig, OcxProviderConfig } from "../../src/types"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; @@ -72,6 +72,25 @@ describe("transientRetryPolicyFor", () => { }); }); +describe("resetReplayPolicyFor", () => { + test("is off unless the provider opts in", () => { + expect(resetReplayPolicyFor({} as OcxProviderConfig)).toBeNull(); + expect(resetReplayPolicyFor({ retryOnReset: { enabled: false } } as OcxProviderConfig)).toBeNull(); + }); + + test("a bare object opts in with one replay; attempts is a total-send ceiling", () => { + expect(resetReplayPolicyFor({ retryOnReset: {} } as OcxProviderConfig)).toEqual({ enabled: true, attempts: 2 }); + expect(resetReplayPolicyFor({ retryOnReset: { attempts: 3 } } as OcxProviderConfig)).toEqual({ enabled: true, attempts: 3 }); + }); + + test("no auth-mode or adapter gate: forward auth is the send it exists for", () => { + for (const authMode of ["forward", "oauth", "key", undefined]) { + expect(resetReplayPolicyFor({ authMode, retryOnReset: {} } as unknown as OcxProviderConfig)) + .toEqual({ enabled: true, attempts: 2 }); + } + }); +}); + describe("fetchWithTransientRetry", () => { test("a non-replayable gateway status is returned after one send, body intact", async () => { // The Codex WebSocket relay settles a 504 when the origin never acknowledged a frame it diff --git a/tests/responses/responses-core-modules.test.ts b/tests/responses/responses-core-modules.test.ts index 6d92a3ab55..e836f3c40b 100644 --- a/tests/responses/responses-core-modules.test.ts +++ b/tests/responses/responses-core-modules.test.ts @@ -16,7 +16,8 @@ const EXISTING_BOUNDARIES = new Set([ "combo-session-recall.ts", "combo-stream-preflight.ts", "context-overflow.ts", "empty-completion-guard.ts", "encrypted-payload.ts", "fetch-helpers.ts", "input-admission.ts", "outbound-body-guard.ts", "passthrough-error.ts", - "responses-field-backfill.ts", "terminal-guard.ts", "upstream-error.ts", "ws-upstream.ts", + "reset-replay.ts", "responses-field-backfill.ts", "terminal-guard.ts", "upstream-error.ts", + "ws-upstream.ts", ]); function siblingImports(source: string): string[] { diff --git a/tests/responses/responses-pool-401-refresh.test.ts b/tests/responses/responses-pool-401-refresh.test.ts index 2fc4409c13..4bdc63e3e0 100644 --- a/tests/responses/responses-pool-401-refresh.test.ts +++ b/tests/responses/responses-pool-401-refresh.test.ts @@ -87,6 +87,8 @@ function request( headers?: HeadersInit; stream?: boolean; input?: unknown; + /** Emitted only when asked: `selfContainedResponsesBody` requires it for a reset replay. */ + store?: false; } = {}, ): Request { const headers = new Headers(options.headers); @@ -99,7 +101,12 @@ function request( headers, body: JSON.stringify(compact ? { model: options.model ?? "gpt-5.5", input } - : { model: options.model ?? "gpt-5.5", input, stream: options.stream ?? false }), + : { + model: options.model ?? "gpt-5.5", + input, + stream: options.stream ?? false, + ...(options.store === false ? { store: false } : {}), + }), }); } @@ -356,6 +363,55 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { expect(readStoredGeneration()).toBe(4); }); + /** + * The refreshed send is a send like any other: a connection that dies before response + * headers is ambiguous there too. It reached upstream unwrapped, so it observed neither + * the operator replay nor the refusal that stands in for it, and its physical sends were + * not charged to the logical request's budget. CodeRabbit found this on #4942. + */ + test("a pre-header reset on the refreshed send replays under retryOnReset", async () => { + const cfg = config(); + (cfg.providers as Record>).openai!.retryOnReset = {}; + let resetOnce = false; + const harness = installHarness({ + responseForSend: (authorization) => { + if (authorization !== "Bearer refreshed-access" || resetOnce) return undefined; + resetOnce = true; + throw Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }); + }, + }); + + const logCtx = { model: "", provider: "" } as RequestLogContext; + const response = await handleResponses(request("/v1/responses"), cfg, logCtx); + + expect(response.status).toBe(200); + // One refresh, then the refreshed bearer sends twice: the reset and its replay. + expect(harness.refreshes).toEqual(["refresh-grant"]); + expect(harness.sends).toEqual([ + "Bearer rejected-access", "Bearer refreshed-access", "Bearer refreshed-access", + ]); + expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); + }); + + test("a pre-header reset on the refreshed send without the policy is the refusal, not a transport error", async () => { + let resetOnce = false; + const harness = installHarness({ + responseForSend: (authorization) => { + if (authorization !== "Bearer refreshed-access" || resetOnce) return undefined; + resetOnce = true; + throw Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }); + }, + }); + + const response = await handleResponses( + request("/v1/responses"), config(), { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + }); + test("compact refreshes a time-valid stored credential once and replays the same account", async () => { const harness = installHarness(); const response = await handleResponsesCompact( @@ -1035,6 +1091,106 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { }); }); +/** + * The account move is a send like every other leg, so a connection that dies before a response + * head there is the same ambiguous failure. It reached upstream unwrapped, so an eligible + * request that moved accounts and then reset observed neither the operator replay nor the + * refusal, and the answer depended on which leg happened to reset. Found in review on #4942. + */ +describe("pre-header reset on the alternate-account send", () => { + /** Both accounts hold live bearers, so the move is reached without a 401 replay first. */ + function writeTwoLiveAccounts(): void { + writeFileSync(join(home, "codex-accounts.json"), JSON.stringify({ + [ACCOUNT_ID]: storedRecord({ + accessToken: "work-access", + refreshToken: "work-grant", + generation: 1, + chatgptAccountId: "acc-work", + }), + [OTHER_ACCOUNT_ID]: storedRecord({ + accessToken: "other-access", + refreshToken: "other-grant", + generation: 1, + chatgptAccountId: "acc-other", + }), + }, null, 2)); + } + + function movingConfig(extra: Record = {}): OcxConfig { + const cfg = config({ secondAccount: true }); + // fill-first keeps the process-wide round-robin cursor where the other regressions expect it. + cfg.accountPoolStrategy = "fill-first"; + Object.assign(cfg.providers.openai as Record, extra); + return cfg; + } + + function resetOnce(): { harness: Harness; movedSends: () => number } { + let reset = false; + let moved = 0; + const harness = installHarness({ + responseForSend: authorization => { + if (authorization === "Bearer work-access") { + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + } + if (authorization === "Bearer other-access") { + moved += 1; + if (!reset) { + reset = true; + throw Object.assign( + new Error("The socket connection was closed unexpectedly."), + { code: "ECONNRESET" }, + ); + } + return Response.json({ + id: "resp_alternate", object: "response", status: "completed", + model: "gpt-5.5", output: [], usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + } + return undefined; + }, + }); + return { harness, movedSends: () => moved }; + } + + test("with retryOnReset the leg still refuses, because the request has no send left", async () => { + writeTwoLiveAccounts(); + const { harness, movedSends } = resetOnce(); + + const response = await handleResponses( + // `store: false` is what makes the body self-contained; without it the policy never + // reaches the wire and this case would pass for the wrong reason. + request("/v1/responses", { store: false }), + movingConfig({ retryOnReset: {} }), + { model: "", provider: "" } as RequestLogContext, + ); + + // The policy reaches this leg, and the request-wide allowance is what stops it: the first + // send and the account move already spent two of three, so the replay would be a fourth + // physical send. A per-leg counter would have bought it; one shared budget does not. + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(movedSends()).toBe(1); + expect(harness.sends).toEqual(["Bearer work-access", "Bearer other-access"]); + }); + + test("without the policy the alternate leg answers the refusal, not a transport error", async () => { + writeTwoLiveAccounts(); + const { harness, movedSends } = resetOnce(); + + const response = await handleResponses( + request("/v1/responses"), + movingConfig(), + { model: "", provider: "" } as RequestLogContext, + ); + + // 502 here would invite the client to resend a request that may already be running. + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(movedSends()).toBe(1); + expect(harness.sends).toEqual(["Bearer work-access", "Bearer other-access"]); + }); +}); + describe("stored pool 401 replay then encrypted combo recovery", () => { const assignment = "RECOVERED-POOL-PLAINTEXT-SENTINEL"; diff --git a/tests/responses/responses-reset-replay.test.ts b/tests/responses/responses-reset-replay.test.ts new file mode 100644 index 0000000000..713d1ad99e --- /dev/null +++ b/tests/responses/responses-reset-replay.test.ts @@ -0,0 +1,289 @@ +/** + * `retryOnReset` through the public Responses dispatch: a native `openai-responses` provider + * whose upstream closes the connection before any response byte. The counted quantity is the + * number of physical sends, read back from the request log the same way + * `responses-send-budget-counts.test.ts` does, because the whole contract is "one more send, + * and only for a request the proxy can judge self-contained". + */ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; +import { clearKeyCooldowns } from "../../src/providers/key-failover"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { handleResponses } from "../../src/server/responses/core"; +import { selfContainedResponsesBody } from "../../src/server/responses/reset-replay"; +import type { RequestLogContext } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; + +const originalFetch = globalThis.fetch; +const warnSpies: Array> = []; + +let releaseSpendHome: (() => void) | undefined; + +beforeEach(() => { + // Direct handler dispatches need the writer lease startServer normally holds (#5157). + releaseSpendHome = acquireOwnedSpendHome(); + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); + warnSpies.push(spyOn(console, "warn").mockImplementation(() => {})); +}); + +afterEach(() => { + releaseSpendHome?.(); + releaseSpendHome = undefined; + globalThis.fetch = originalFetch; + for (const spy of warnSpies.splice(0)) spy.mockRestore(); + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); +}); + +function responsesProvider(name: string, extra: Record = {}): Record { + return { + adapter: "openai-responses", + baseUrl: `https://${name}.example/v1`, + authMode: "key", + apiKey: `sk-${name}`, + models: [`model-${name}`], + ...extra, + }; +} + +function singleProvider(extra: Record = {}): OcxConfig { + return { + defaultProvider: "t0", + providers: { t0: responsesProvider("t0", extra) }, + } as unknown as OcxConfig; +} + +function comboOverTwo(extra: Record = {}): OcxConfig { + return { + defaultProvider: "t0", + providers: { t0: responsesProvider("t0", extra), t1: responsesProvider("t1", extra) }, + combos: { fan: { strategy: "failover", targets: [ + { provider: "t0", model: "model-t0" }, { provider: "t1", model: "model-t1" }, + ] } }, + } as unknown as OcxConfig; +} + +/** A self-contained turn: nothing stored, complete input, one client tool. */ +function selfContained(model: string, fields: Record = {}): Record { + return { + model, stream: false, store: false, input: "hello", + tools: [{ type: "function", name: "read_fixture", parameters: { type: "object", properties: {} } }], + ...fields, + }; +} + +function request(body: Record): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + +function reset(): Error { + return Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }); +} + +function completed(id: string): Response { + return Response.json({ + id, object: "response", status: "completed", model: "model-t0", output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); +} + +interface Wire { sends: Array<{ authorization: string; connection: string | null; keepalive: unknown; body: string }> } + +/** Fake upstream that answers each send in order; the last entry repeats. */ +function upstream(answers: Array): Wire { + const wire: Wire = { sends: [] }; + let index = 0; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + const headers = new Headers(init?.headers); + wire.sends.push({ + authorization: headers.get("authorization") ?? "", + connection: headers.get("connection"), + keepalive: (init as { keepalive?: unknown } | undefined)?.keepalive, + body: typeof init?.body === "string" ? init.body : "", + }); + const answer = answers[index] ?? answers[answers.length - 1]!; + index += 1; + if (answer instanceof Error) throw answer; + return answer.clone(); + }) as typeof fetch; + return wire; +} + +const totalSends = (logCtx: RequestLogContext): number => + (logCtx.attempts ?? []).reduce((sum, attempt) => sum + attempt.sendCount, 0); + +describe("selfContainedResponsesBody", () => { + const base = selfContained("m"); + + test("accepts a stored-nothing turn with complete input and client tools", () => { + expect(selfContainedResponsesBody(base)).toBe(true); + expect(selfContainedResponsesBody({ ...base, tools: undefined })).toBe(true); + expect(selfContainedResponsesBody({ ...base, input: [ + { role: "user", content: "hi" }, + { type: "message", role: "assistant", content: [] }, + { type: "reasoning", summary: [] }, + { type: "function_call", call_id: "c", name: "read_fixture", arguments: "{}" }, + { type: "function_call_output", call_id: "c", output: "ok" }, + { type: "custom_tool_call", call_id: "d", name: "x", input: "" }, + { type: "custom_tool_call_output", call_id: "d", output: "" }, + { type: "compaction", encrypted_content: "..." }, + { type: "tool_search_call", id: "s" }, + ] })).toBe(true); + expect(selfContainedResponsesBody({ ...base, tools: [ + { type: "custom", name: "fixture" }, + { type: "tool_search", execution: "client" }, + { type: "namespace", name: "group", tools: [{ type: "function", name: "inner" }] }, + ] })).toBe(true); + expect(selfContainedResponsesBody({ ...base, input: [ + { role: "user", content: "hi" }, + { type: "additional_tools", tools: [{ type: "function", name: "late" }] }, + { type: "tool_search_output", tools: [{ type: "namespace", name: "n", tools: [{ type: "custom", name: "c" }] }] }, + ] })).toBe(true); + }); + + test("refuses anything stored, continued, backgrounded or server-owned", () => { + for (const fields of [ + { store: true }, { store: undefined }, { background: true }, + { previous_response_id: "resp_prior" }, { conversation: "conv_1" }, { stream_id: "lane" }, + { input: undefined }, { input: null }, { input: 5 }, + ]) { + expect(selfContainedResponsesBody({ ...base, ...fields })).toBe(false); + } + expect(selfContainedResponsesBody("not an object")).toBe(false); + expect(selfContainedResponsesBody(null)).toBe(false); + }); + + test("refuses hosted, server-executed and unknown tools wherever they are declared", () => { + for (const tool of [ + { type: "web_search" }, { type: "mcp", server_url: "https://example.test" }, + { type: "code_interpreter" }, { type: "file_search" }, { type: "image_generation" }, + { type: "tool_search", execution: "server" }, { type: "future_unknown" }, + { type: "namespace", name: "mixed", tools: [{ type: "function", name: "ok" }, { type: "mcp" }] }, + { type: "namespace", tools: [{ type: "function", name: "unnamed-group" }] }, + "not-an-object", + ]) { + expect(selfContainedResponsesBody({ ...base, tools: [tool] })).toBe(false); + expect(selfContainedResponsesBody({ ...base, input: [{ type: "additional_tools", tools: [tool] }] })).toBe(false); + expect(selfContainedResponsesBody({ ...base, input: [{ type: "tool_search_output", tools: [tool] }] })).toBe(false); + } + expect(selfContainedResponsesBody({ ...base, tools: {} })).toBe(false); + }); + + test("refuses input items the client does not own or the proxy does not know", () => { + for (const item of [ + { type: "item_reference", id: "stored" }, + { type: "mcp_approval_response", approval_request_id: "a", approve: true }, + { type: "computer_call_output", call_id: "c", output: {} }, + { type: "future_unknown_state" }, + { role: "tool", content: "x" }, + { content: "no role, no type" }, + "not-an-object", + ]) { + expect(selfContainedResponsesBody({ ...base, input: [{ role: "user", content: "hi" }, item] })).toBe(false); + } + }); + + test("catalog traversal is bounded", () => { + let tool: unknown = { type: "function", name: "leaf" }; + for (let i = 0; i < 6; i++) tool = { type: "namespace", name: "deep", tools: [tool] }; + expect(selfContainedResponsesBody({ ...base, tools: [tool] })).toBe(false); + expect(selfContainedResponsesBody({ ...base, tools: Array(4097).fill({ type: "function", name: "many" }) })).toBe(false); + }); +}); + +describe("retryOnReset through native Responses dispatch", () => { + test("a self-contained turn is sent again on a fresh connection and completes", async () => { + const wire = upstream([reset(), completed("resp_replayed")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(request(selfContained("model-t0")), singleProvider({ retryOnReset: {} }), logCtx); + expect(response.status).toBe(200); + expect((await response.json()).id).toBe("resp_replayed"); + expect(wire.sends).toHaveLength(2); + expect(wire.sends.map(send => send.authorization)).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); + // The replay is byte-identical and leaves Bun's keep-alive pool behind. + expect(wire.sends[1]!.body).toBe(wire.sends[0]!.body); + expect(wire.sends[1]!.connection).toBe("close"); + expect(wire.sends[1]!.keepalive).toBe(false); + expect(totalSends(logCtx)).toBe(2); + }); + + test("a spent ceiling is the same refusal the request would get without the policy", async () => { + const wire = upstream([reset(), reset(), completed("resp_never")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(request(selfContained("model-t0")), singleProvider({ retryOnReset: {} }), logCtx); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(wire.sends).toHaveLength(2); + expect(totalSends(logCtx)).toBe(2); + }); + + test("attempts raises the ceiling to the leg's own budget and no further", async () => { + const wire = upstream([reset(), reset(), reset(), completed("resp_never")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses( + request(selfContained("model-t0")), singleProvider({ retryOnReset: { attempts: 3 } }), logCtx, + ); + expect(response.status).toBe(429); + expect(wire.sends).toHaveLength(3); + expect(totalSends(logCtx)).toBe(3); + }); + + test("a request the proxy cannot judge self-contained keeps the refusal after one send", async () => { + for (const fields of [ + { store: undefined }, { previous_response_id: "resp_prior" }, + { tools: [{ type: "web_search" }] }, { input: [{ type: "item_reference", id: "stored" }] }, + ]) { + const wire = upstream([reset(), completed("resp_never")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses( + request(selfContained("model-t0", fields)), singleProvider({ retryOnReset: {} }), logCtx, + ); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(wire.sends).toHaveLength(1); + expect(totalSends(logCtx)).toBe(1); + } + }); + + test("without the policy nothing changes: one send, then the refusal", async () => { + for (const extra of [{}, { retryOnReset: { enabled: false } }]) { + const wire = upstream([reset(), completed("resp_never")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(request(selfContained("model-t0")), singleProvider(extra), logCtx); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(wire.sends).toHaveLength(1); + } + }); + + test("a spent replay cannot reach a combo sibling", async () => { + const wire = upstream([reset(), reset(), completed("resp_never")]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(request(selfContained("combo/fan")), comboOverTwo({ retryOnReset: {} }), logCtx); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); + expect(wire.sends.map(send => send.authorization)).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); + expect(totalSends(logCtx)).toBe(2); + }); + + test("a streaming turn replays the same way", async () => { + const sse = 'event: response.created\ndata: {"type":"response.created","response":{"id":"resp_s"}}\n\n' + + 'event: response.completed\ndata: {"type":"response.completed","response":{"id":"resp_s","status":"completed","output":[]}}\n\n'; + const wire = upstream([reset(), new Response(sse, { status: 200, headers: { "content-type": "text/event-stream" } })]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses( + request(selfContained("model-t0", { stream: true })), singleProvider({ retryOnReset: {} }), logCtx, + ); + expect(response.status).toBe(200); + expect(await response.text()).toContain("response.completed"); + expect(wire.sends).toHaveLength(2); + expect(wire.sends[1]!.connection).toBe("close"); + }); +}); diff --git a/tests/server/management-provider-reset-replay.test.ts b/tests/server/management-provider-reset-replay.test.ts new file mode 100644 index 0000000000..4b3c34b062 --- /dev/null +++ b/tests/server/management-provider-reset-replay.test.ts @@ -0,0 +1,39 @@ +import { expect, test } from "bun:test"; +import { providerManagementConfigError } from "../../src/server/auth-cors"; + +// Lives apart from management-provider-validation.test.ts because that file sits at its +// file-size ratchet cap; the helpers it needs are small enough to repeat here. +const canonicalDirect = { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", +} as const; + +test("provider management validates retryOnReset bounds and unknown keys", () => { + const base = { adapter: "openai-responses", baseUrl: "https://api.openai.com/v1" }; + expect(providerManagementConfigError("custom", { ...base, retryOnReset: {} })).toBeNull(); + expect(providerManagementConfigError("custom", { ...base, retryOnReset: { enabled: true, attempts: 3 } })).toBeNull(); + expect(providerManagementConfigError("custom", { ...base, retryOnReset: { attempts: 0 } })) + .toContain("retryOnReset.attempts is invalid"); + expect(providerManagementConfigError("custom", { ...base, retryOnReset: { attempts: 4 } })) + .toContain("retryOnReset.attempts is invalid"); + expect(providerManagementConfigError("custom", { ...base, retryOnReset: { attempt: 2 } })) + .toContain("retryOnReset has unrecognized field"); + expect(providerManagementConfigError("custom", { ...base, retryOnReset: true })) + .toContain("retryOnReset is invalid"); + // The canonical openai row is the main target of this policy, and a full-object write + // compares it against the seed with an exact key match: the field must be admitted + // there like requestPacing is, while its value is still validated. + expect(providerManagementConfigError("openai", { ...canonicalDirect, retryOnReset: { attempts: 3 } })).toBeNull(); + expect(providerManagementConfigError("openai", { ...canonicalDirect, retryOnReset: { attempts: 4 } })) + .toContain("retryOnReset.attempts is invalid"); + // A secret-shaped unknown field name and a secret-shaped provider name are both redacted. + const secretError = providerManagementConfigError("custom", { ...base, retryOnReset: { "sk-super-secret-9876": true } })!; + expect(secretError).toContain("retryOnReset has unrecognized field"); + expect(secretError).not.toContain("sk-super-secret-9876"); + const secretNameError = providerManagementConfigError("sk-super-secret-9876", { ...base, retryOnReset: { attempts: 0 } })!; + expect(secretNameError).toContain("retryOnReset.attempts is invalid"); + expect(secretNameError).not.toContain("sk-super-secret-9876"); + expect(secretNameError).toContain("[REDACTED]"); +});