diff --git a/docs-site/src/content/docs/ja/reference/architecture.md b/docs-site/src/content/docs/ja/reference/architecture.md index d1f436e3c6..d787b62ed7 100644 --- a/docs-site/src/content/docs/ja/reference/architecture.md +++ b/docs-site/src/content/docs/ja/reference/architecture.md @@ -48,7 +48,7 @@ HTTP の境界は `server/index.ts` が担い、Responses データプレーン `POST /v1/live` / `POST /v1/realtime/calls`（ChatGPT / Codex App 音声と OpenAI Realtime の call-create、`server/live.ts` が中継）と `/v1/live/{callId}` サイドバンド WebSocket、 `/v1/responses` のオプション WebSocket アップグレードを提供します。 -2. `server/responses/core.ts` が展開し JSON を読みます。覚えておいた `previous_response_id` 入力があれば展開したのち `responses/parser.ts` に渡します。 +2. `server/responses/core.ts` が圧縮を解除して JSON を解析し、記憶済みの `previous_response_id` 入力を展開します。完全な履歴再送をそのまま保持するのは、保持された provider item id または tool call id が完全な接頭辞の再送を証明する場合だけです。内容が同じだけなら各出現を保守的に保持します。その後 `responses/parser.ts` に渡し、実効入力上限と推定誤差帯を超える入力を上流 I/O 前に `413 request_too_large` で拒否します。 3. `router.ts` が通常のモデル id または `provider/model` id を解決します。続いて Codex アカウント affinity を決定し、必要ならプロバイダー OAuth を更新して選択された認証情報を route に適用します。 4. 本リクエストの前に `vision/` が `noVisionModels` モデル用の画像説明を作ります。安全なサイドカー経路がないときはテキスト専用の上流に画像を送らず取り除きます。 5. `server/adapter-resolve.ts` がモデル別の wire override を適用し、7つのアダプターのいずれかを作ります。 diff --git a/docs-site/src/content/docs/ja/reference/proxy-formats.md b/docs-site/src/content/docs/ja/reference/proxy-formats.md index 2d6f4fdf6a..ff0255997b 100644 --- a/docs-site/src/content/docs/ja/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ja/reference/proxy-formats.md @@ -47,6 +47,14 @@ provider events → internal adapter events → client dialect 未知の項目タイプは、前方互換性のためにルーズタイプの項目として受け入れられます。変換されたアダプターは、認識する項目タイプのみを処理し、プロバイダーが表現できない機能を拒否する場合があります。 +解析済みの `input` の近似見積もりが、ルーティング先モデルの実効入力上限に 10% の推定誤差帯を +加えた値を超える場合、`413 request_too_large`（code `input_context_window_exceeded`）で +拒否されます。見積もりはメッセージ、`instructions`、ツール、構造化出力スキーマを集計し、画像と +後続の guidance、圧縮プロンプト、bridge ツール注入には別の上限予約を使います。base64 を通常テキストとして +数えません。ガードはクォータ、sidecar、adapter、モデル上流 I/O より前に動作し、terminal-guard continuation +も自身の送信前に再検査されます。誤差帯内は provider の tokenizer に委ねます。完全履歴は provider item id +または tool call id が完全な保存済み接頭辞の再送を証明するときだけ重複除去されます。 + ### JSON および SSE 出力 `stream: true` の場合、応答は `text/event-stream` となります。ブリッジは、`response.created`、出力項目およびテキスト/ツール デルタ、および 1 つの端末 `response.completed`、`response.failed`、または `response.incomplete` イベントなどの応答イベントを発行します。通常のストリームは `data: [DONE]` で終了します。 @@ -211,6 +219,7 @@ Responses-family および Chat リクエストは、プロバイダーまたは | 503 | `combo_unavailable` |選択したコンボ内のすべてのターゲットは使用不可、クールダウン中、無効、またはその他の理由で不適格です。 | 400 | `unreadable_encrypted_agent_task` |暗号化された v2 ワーカー タスクには、それを使用できる適格なネイティブ ChatGPT ターゲットがありません。 | 426 | `upgrade_required` |応答 WebSocket トランスポートが無効になっているか、アップグレードが失敗しました。 HTTP を使用する | +| 413 | `request_too_large` | 推定 `input` が実効入力上限と 10% の推定誤差帯を超える（code `input_context_window_exceeded`）。クォータ、sidecar、adapter、モデル上流 I/O の前に拒否 | Anthropic オリジンの失敗は Anthropic のエラー エンベロープでレンダリングされるため、オリジンの拒否は OpenAI スタイルの `origin_rejected` 本体ではなく、その方言上の 403 `permission_error` になります。 diff --git a/docs-site/src/content/docs/ko/reference/architecture.md b/docs-site/src/content/docs/ko/reference/architecture.md index 0166f48292..981e98cb7f 100644 --- a/docs-site/src/content/docs/ko/reference/architecture.md +++ b/docs-site/src/content/docs/ko/reference/architecture.md @@ -50,8 +50,11 @@ HTTP 경계는 `server/index.ts`가 맡고, Responses 데이터 플레인은 `se `POST /v1/live` / `POST /v1/realtime/calls`(ChatGPT / Codex App 음성 및 OpenAI Realtime 호출 생성, `server/live.ts`가 중계)와 `/v1/live/{callId}` 사이드밴드 WebSocket, 그리고 `/v1/responses`의 선택적 WebSocket 업그레이드를 제공합니다. -2. `server/responses/core.ts`가 압축을 풀고 JSON을 읽습니다. 기억해 둔 `previous_response_id` 입력이 있으면 - 펼친 다음 `responses/parser.ts`로 넘깁니다. +2. `server/responses/core.ts`가 압축을 풀고 JSON을 읽은 뒤 기억해 둔 `previous_response_id` 입력을 + 펼칩니다. 전체 기록 재전송은 보존된 provider item id 또는 tool call id가 전체 접두사의 재전송임을 + 증명할 때만 그대로 유지됩니다. 내용만 같은 경우에는 각 출현을 보수적으로 보존합니다. 이후 + `responses/parser.ts`로 넘기며, 유효 입력 한도와 추정 오차 대역을 넘는 입력은 업스트림 I/O 전에 + `413 request_too_large`로 거부됩니다. 3. `router.ts`가 일반 모델 id 또는 `provider/model` id를 해석합니다. 이어서 Codex 계정 affinity를 결정하고, 필요하면 프로바이더 OAuth를 갱신해 선택된 자격 증명을 route에 적용합니다. 4. 본 요청 전에 `vision/`이 `noVisionModels` 모델용 이미지 설명을 만듭니다. 안전한 사이드카 경로가 diff --git a/docs-site/src/content/docs/ko/reference/proxy-formats.md b/docs-site/src/content/docs/ko/reference/proxy-formats.md index 39497eef4b..dc3db86377 100644 --- a/docs-site/src/content/docs/ko/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ko/reference/proxy-formats.md @@ -54,6 +54,14 @@ Responses 표현이 이 연결의 중심입니다. 네이티브 호환 경로는 알 수 없는 항목 유형은 앞으로의 호환성을 위해 느슨한 형식의 typed item으로 허용됩니다. 변환된 어댑터는 자신이 인식하는 항목 유형만 처리하며, 제공자가 표현할 수 없는 기능은 거부할 수 있습니다. +파싱된 `input`의 근사 추정치가 라우팅된 모델의 유효 입력 한도에 10% 추정 오차 대역을 더한 +값을 넘으면 `413 request_too_large`(code `input_context_window_exceeded`)로 거부됩니다. +추정치는 메시지, `instructions`, 도구, 구조화 출력 스키마를 합산하고 이미지 및 이후의 guidance, +압축 프롬프트, bridge 도구 주입에는 별도의 제한된 여유를 둡니다. base64를 일반 텍스트로 계산하지 +않습니다. 가드는 할당량, sidecar, adapter, 모델 업스트림 I/O 전에 실행되며 terminal-guard continuation도 +자체 전송 전에 다시 검사됩니다. 오차 대역 안의 요청은 provider tokenizer가 결정합니다. 전체 기록은 +provider item id 또는 tool call id가 저장된 전체 접두사의 재전송임을 증명할 때만 중복 제거됩니다. + ### JSON과 SSE 출력 `stream: true`이면 응답은 `text/event-stream`입니다. 브리지는 `response.created`, output-item과 text/tool @@ -254,6 +262,7 @@ data-plane key는 management credential이 아닙니다. management API는 별 | 503 | `combo_unavailable` | 선택한 combo의 모든 대상이 사용할 수 없거나, cooldown 중이거나, 비활성화되어 있거나, 다른 이유로 부적합합니다 | | 400 | `unreadable_encrypted_agent_task` | 암호화된 v2 worker task를 소비할 수 있는 적격 네이티브 ChatGPT 대상이 없습니다 | | 426 | `upgrade_required` | Responses WebSocket transport가 비활성화되어 있거나 업그레이드에 실패했습니다. HTTP를 사용하십시오 | +| 413 | `request_too_large` | 추정 `input`이 유효 입력 한도와 10% 추정 오차 대역을 초과합니다(code `input_context_window_exceeded`). 할당량, sidecar, adapter, 모델 업스트림 I/O 전에 거부됩니다 | Anthropic-origin 실패는 Anthropic의 error envelope로 렌더링됩니다. 따라서 해당 방언에서 origin 거부는 OpenAI 스타일 `origin_rejected` body가 아니라 403 `permission_error`입니다. diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index d47e140dca..c77124c189 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -53,7 +53,11 @@ the `server/responses.ts` facade and its `server/responses/*.ts` modules: `/v1/live/{callId}` (and `/v1/realtime?call_id=`), and the optional WebSocket upgrade on `/v1/responses`. 2. `server/responses/core.ts` decompresses and parses JSON, expands locally remembered - `previous_response_id` input when available, then calls `responses/parser.ts`. + `previous_response_id` input when available. A full-history resend is preserved only when + a retained provider item id or tool call id proves the complete prefix is replayed; ambiguous + content equality keeps every occurrence. It then calls `responses/parser.ts`; input estimated + above the routed model's effective limit plus the admission uncertainty band is rejected with + `413 request_too_large` before any upstream I/O. 3. `router.ts` resolves a bare or `provider/model` id. The server then resolves Codex account affinity, refreshes provider OAuth when needed, and applies the selected credential to the route. 4. Before the main call, `vision/` describes images for models in `noVisionModels`; if no safe diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index bd02453019..291e5b5021 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -55,6 +55,25 @@ non-empty `model`. `input` may be a string or an array of Responses items. Unknown item types are accepted as loose typed items for forward compatibility. Translated adapters handle only the item types they recognize, and may reject a feature their provider cannot represent. +Requests whose parsed `input` is estimated — using an approximate, model-aware token estimate over +message text, `instructions`, tool definitions, and structured-output schemas, plus separate +bounded reserves for images and deterministic later injections — to exceed the routed model's +effective input limit (per-model maximum input, falling back to the context window resolved from +provider, registry, or catalog metadata) by more than the 10% estimator uncertainty band are rejected with +`413 request_too_large` (code `input_context_window_exceeded`) before adapter construction and +model-serving upstream I/O. Body admission (decompression, size caps, parsing) precedes the guard. +HTTP requests are also rejected before authentication; WebSocket frames have already passed +handshake authentication and origin admission, so for them the guard runs before per-turn adapter +construction and upstream I/O. The guard reserves pending guidance, compaction, bridge-tool, and +vision-description mutations before quota or sidecar work, and rechecks terminal-guard continuations +before their upstream send. Estimates inside the uncertainty band are forwarded for the provider's +tokenizer to decide. Codex compacts well before this limit, so an oversized +body indicates abnormal duplication — for example a chained continuation that resends the full +conversation to a stateless provider. Such a resend is deduplicated only when a retained provider +item id or tool call id proves the complete stored prefix; content equality alone never drops an +occurrence. Compact the conversation or start a new thread and retry; the rejected request is never +forwarded upstream. + ### JSON and SSE output With `stream: true`, the response is `text/event-stream`. The bridge emits Responses events such as @@ -278,6 +297,7 @@ Errors use the client dialect's envelope where needed, but these status/code mea | 503 | `combo_unavailable` | Every target in the selected combo is unavailable, in cooldown, disabled, or otherwise ineligible | | 400 | `unreadable_encrypted_agent_task` | An encrypted v2 worker task has no eligible native ChatGPT target that can consume it | | 426 | `upgrade_required` | The Responses WebSocket transport is disabled or the upgrade failed; use HTTP | +| 413 | `request_too_large` | Estimated parsed `input` exceeds the routed model's effective input limit plus the 10% estimator uncertainty band (code `input_context_window_exceeded`); rejected before quota, sidecar, adapter, or model-serving upstream I/O | Anthropic-origin failures are rendered in Anthropic's error envelope, so the origin rejection is a 403 `permission_error` on that dialect rather than the OpenAI-style `origin_rejected` body. diff --git a/docs-site/src/content/docs/ru/reference/architecture.md b/docs-site/src/content/docs/ru/reference/architecture.md index 569652c6a4..74cab95ef4 100644 --- a/docs-site/src/content/docs/ru/reference/architecture.md +++ b/docs-site/src/content/docs/ru/reference/architecture.md @@ -54,8 +54,12 @@ src/ (создание голосового/Realtime-вызова ChatGPT / Codex App, ретранслируется `server/live.ts`), sideband WebSocket на `/v1/live/{callId}`, а также необязательный WebSocket-апгрейд на `/v1/responses`. -2. `server/responses/core.ts` распаковывает и парсит JSON, разворачивает локально запомненный вход - `previous_response_id`, когда он доступен, затем вызывает `responses/parser.ts`. +2. `server/responses/core.ts` распаковывает и парсит JSON и разворачивает локально запомненный + `previous_response_id`. Полная история сохраняется без повторного добавления только когда + сохранённый provider item id или tool call id доказывает повтор полной префиксной истории; одно + совпадение содержимого сохраняет оба вхождения. Затем вызывается `responses/parser.ts`, а вход, + превышающий действующий лимит вместе с полосой погрешности оценки, отклоняется с + `413 request_too_large` до любого upstream-I/O. 3. `router.ts` разрешает «голый» id или id вида `provider/model`. Затем сервер определяет привязку (affinity) аккаунта Codex, при необходимости обновляет OAuth провайдера и применяет выбранные учётные данные к маршруту. diff --git a/docs-site/src/content/docs/ru/reference/proxy-formats.md b/docs-site/src/content/docs/ru/reference/proxy-formats.md index 6a461e8b9e..6fc1b48196 100644 --- a/docs-site/src/content/docs/ru/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ru/reference/proxy-formats.md @@ -57,6 +57,15 @@ control и safety ответа всё равно происходят на гр Translated-adapter'ы обрабатывают только известные им типы и могут отвергнуть функцию, которую их провайдер не умеет выразить. +Запрос отклоняется с `413 request_too_large` (code `input_context_window_exceeded`), когда +приблизительная оценка разобранного `input` превышает действующий лимит маршрутизируемой модели +вместе с 10% полосой погрешности. Оценка суммирует сообщения, `instructions`, инструменты и схемы +структурированного вывода, а для изображений и последующих вставок guidance, compaction prompt и bridge +tools использует отдельные ограниченные резервы; base64 не считается обычным текстом. Проверка проходит +до quota, sidecar, adapter и модельного upstream-I/O, а terminal-guard continuation повторно проверяется +перед своей отправкой. Значения внутри полосы оставляются tokenizer'у провайдера. Полная история +дедуплицируется только когда provider item id или tool call id доказывает повтор всего сохранённого префикса. + ### JSON и SSE-вывод При `stream: true` ответ идёт как `text/event-stream`. Мост испускает события Responses вроде @@ -272,6 +281,7 @@ Direct, поэтому remote proxy key здесь обязан идти чер | 503 | `combo_unavailable` | Все цели выбранной combo недоступны, в cooldown, отключены или иным образом не подходят | | 400 | `unreadable_encrypted_agent_task` | У шифрованной задачи воркера v2 нет подходящей нативной цели ChatGPT, способной её прочитать | | 426 | `upgrade_required` | Транспорт Responses WebSocket выключен или upgrade не удался; используйте HTTP | +| 413 | `request_too_large` | Оценённый `input` превышает действующий лимит и 10% полосу погрешности (code `input_context_window_exceeded`); отклоняется до quota, sidecar, adapter и модельного upstream-I/O | Сбои, пришедшие с Anthropic-side, отрисовываются в error envelope Anthropic, поэтому отклонение origin превращается в 403 `permission_error`, а не в OpenAI-style body `origin_rejected`. diff --git a/docs-site/src/content/docs/zh-cn/reference/architecture.md b/docs-site/src/content/docs/zh-cn/reference/architecture.md index 926a6d096f..83fa66c59a 100644 --- a/docs-site/src/content/docs/zh-cn/reference/architecture.md +++ b/docs-site/src/content/docs/zh-cn/reference/architecture.md @@ -52,7 +52,10 @@ src/ 建连，由 `server/live.ts` 中继）、`/v1/live/{callId}` 旁路 WebSocket， 以及 `/v1/responses` 上可选的 WebSocket upgrade。 2. `server/responses/core.ts` 解压并解析 JSON；如果本地记住了对应输入，则展开 - `previous_response_id`，随后调用 `responses/parser.ts`。 + `previous_response_id`。只有保留的 provider item id 或 tool call id 能证明完整前缀确为 + 重放时，完整历史才会原样保留；仅内容相同仍会保守地保留每次出现。随后调用 + `responses/parser.ts`；估算超过目标模型有效输入上限及准入误差带的输入会在任何上游 I/O + 之前以 `413 request_too_large` 被拒绝。 3. `router.ts` 解析 bare id 或 `provider/model` id。server 随后确定 Codex account affinity， 必要时刷新 provider OAuth，并把选中的 credential 应用到 route。 4. 主请求发出前，`vision/` 会为 `noVisionModels` 中的模型描述图像。如果没有安全的 sidecar diff --git a/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md b/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md index a469e2f628..6b84cb31f6 100644 --- a/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md +++ b/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md @@ -52,6 +52,16 @@ Responses 表示是这座桥的中心。原生兼容的路由可以跳过部分 未知项目类型会作为宽松的类型化项被接受，以保证前向兼容。已翻译的适配器只处理它们能识别的项目类型，并且可能会拒绝其提供方无法表示的特性。 +如果解析后的 `input` 估算超过目标模型的有效输入上限（按模型配置的最大输入，缺失时回退到 +由提供方、注册表或目录元数据解析出的上下文窗口）再加 10% 的估算误差带，代理会以 +`413 request_too_large`（code `input_context_window_exceeded`）拒绝。估算会累计消息文本、 +`instructions`、工具定义和结构化输出 schema，并为图片及后续确定性的 guidance、压缩提示和 +bridge 工具注入单独预留空间；不会把 base64 图片字节当作普通文本 token。请求体准入先于该守卫。 +HTTP 请求还会在认证之前被拒绝；WebSocket 帧已通过握手认证和来源准入。准入检查发生在配额探测、 +sidecar、adapter 构建和模型上游 I/O 之前，terminal-guard continuation 也会在自己的上游发送前再次检查。 +误差带内的请求会交给 provider 的 tokenizer 决定。完整历史重发只有在保留的 provider item id 或 tool +call id 证明整个已存储前缀确为重放时才会去重；仅内容相同绝不会丢弃一次出现。 + ### JSON 和 SSE 输出 当 `stream: true` 时，响应为 `text/event-stream`。桥会发出 Responses 事件，例如 @@ -232,6 +242,7 @@ Responses 家族和 Chat 请求会把 `Authorization` 留给提供方或 Codex D | 503 | `combo_unavailable` | 所选 combo 中的所有目标都不可用、处于冷却、已禁用或以其他方式不具备资格 | | 400 | `unreadable_encrypted_agent_task` | 一个加密的 v2 worker task 没有任何可消费它的合格原生 ChatGPT 目标 | | 426 | `upgrade_required` | Responses WebSocket 传输被禁用，或升级失败；请改用 HTTP | +| 413 | `request_too_large` | 估算的 `input` 超过目标模型的有效输入上限及 10% 估算误差带（code `input_context_window_exceeded`）；在配额、sidecar、adapter 或模型上游 I/O 之前被拒绝 | Anthropic 来源的失败会以 Anthropic 的错误封装呈现，因此该方言中的 origin 拒绝会是 403 `permission_error`，而不是 OpenAI 风格的 `origin_rejected` body。 diff --git a/src/images/index.ts b/src/images/index.ts index f9f0fd5b99..59bc9c043f 100644 --- a/src/images/index.ts +++ b/src/images/index.ts @@ -1,4 +1,4 @@ -export { planImageBridge, planVideoBridge, findXaiProvider, resolveXaiImageApiKey } from "./plan"; +export { planImageBridge, planImageBridgeSync, planVideoBridge, planVideoBridgeSync, findXaiProvider, resolveXaiImageApiKey } from "./plan"; export { runWithImageBridge, clampImageMaxRounds, DEFAULT_MAX_ROUNDS, MAX_ROUNDS_HARD_LIMIT } from "./loop"; export type { ImageBridgePlan, ImageCallResult, VideoBridgePlan, VideoCallResult } from "./types"; export { buildImageTool, buildVideoTool, extractHostedImageGeneration, IMAGE_GEN_TOOL_NAME, VIDEO_GEN_TOOL_NAME, isImageGenName, isVideoGenName } from "./synthetic-tool"; diff --git a/src/images/plan.ts b/src/images/plan.ts index b8780d0fb4..d2f87db2cc 100644 --- a/src/images/plan.ts +++ b/src/images/plan.ts @@ -40,11 +40,15 @@ export function resolveXaiImageApiKey(provider: OcxProviderConfig): string | und return apiKey || undefined; } -export async function planImageBridge( +/** + * Build the image bridge plan without performing provider or filesystem I/O. + * Returns undefined when the request or routed provider should not use the bridge. + */ +export function planImageBridgeSync( config: OcxConfig, parsed: OcxParsedRequest, routedProvider: OcxProviderConfig, -): Promise { +): ImageBridgePlan | undefined { if (config.images?.bridgeEnabled !== true) return undefined; if (!parsed._imageGeneration) return undefined; const toolAllowed = toolChoiceToolPredicate(parsed.options.toolChoice); @@ -84,6 +88,15 @@ export async function planImageBridge( }; } +/** Async wrapper for {@link planImageBridgeSync}; the plan body is fully synchronous. */ +export async function planImageBridge( + config: OcxConfig, + parsed: OcxParsedRequest, + routedProvider: OcxProviderConfig, +): Promise { + return planImageBridgeSync(config, parsed, routedProvider); +} + const DEFAULT_VIDEO_MODEL = "grok-imagine-video"; /** @@ -94,11 +107,11 @@ const DEFAULT_VIDEO_MODEL = "grok-imagine-video"; * 2. the routed provider is NOT api.openai.com (native passthrough) * 3. an xAI provider with a valid API key is available */ -export async function planVideoBridge( +export function planVideoBridgeSync( config: OcxConfig, parsed: OcxParsedRequest, routedProvider: OcxProviderConfig, -): Promise { +): VideoBridgePlan | undefined { if (config.images?.videoBridgeEnabled !== true) return undefined; const toolNames = new Set(); toolNames.add(VIDEO_GEN_TOOL_NAME); @@ -141,3 +154,12 @@ export async function planVideoBridge( ...(artifactsKeepCount !== undefined ? { artifactsKeepCount } : {}), }; } + +/** Async wrapper for {@link planVideoBridgeSync}; the plan body is fully synchronous. */ +export async function planVideoBridge( + config: OcxConfig, + parsed: OcxParsedRequest, + routedProvider: OcxProviderConfig, +): Promise { + return planVideoBridgeSync(config, parsed, routedProvider); +} diff --git a/src/lib/json-walk.ts b/src/lib/json-walk.ts new file mode 100644 index 0000000000..23ddfa8032 --- /dev/null +++ b/src/lib/json-walk.ts @@ -0,0 +1,79 @@ +/** + * Shared read-only JSON tree traversal for admission/accounting walks. + * + * Both request-size accounting (`boundedJsonSerializedByteLength`) and token estimation + * (`countJsonTokens`) walk parsed client payloads with the same value/array/object frame + * mechanics: iterative over container/index frames (deep nesting cannot overflow the + * call stack), lazy own-key enumeration (no key arrays proportional to the payload), + * and early termination once the caller's budget is exceeded. This primitive owns that + * machinery; the two walkers only differ in what they count per node. + */ + +export interface JsonWalkHooks { + /** Called for every leaf value: string, number, boolean, null, and undefined. */ + onValue?(value: unknown): void; + /** Called for every own enumerable object key (before its value). */ + onObjectKey?(key: string): void; + onArrayStart?(): void; + onArrayEnd?(): void; + onArraySeparator?(): void; + onObjectStart?(): void; + onObjectEnd?(): void; + onObjectSeparator?(): void; + /** Stop the walk as soon as this returns true (checked before each frame). */ + isDone?(): boolean; +} + +type Frame = + | { kind: "value"; value: unknown } + | { kind: "array"; array: unknown[]; index: number } + | { kind: "object"; keys: Generator; record: Record; count: number }; + +function* ownEnumerableKeys(record: Record): Generator { + for (const key in record) { + if (Object.prototype.hasOwnProperty.call(record, key)) yield key; + } +} + +/** + * Iteratively walk a parsed JSON value, invoking the hooks in structural order. The + * walk stops when `hooks.isDone()` returns true; the caller decides how much it counted + * and whether that crossed its budget. + */ +export function walkJsonTree(value: unknown, hooks: JsonWalkHooks): void { + const stack: Frame[] = [{ kind: "value", value }]; + while (stack.length > 0 && !(hooks.isDone?.() ?? false)) { + const frame = stack.pop()!; + if (frame.kind === "value") { + const current = frame.value; + if (Array.isArray(current)) { + hooks.onArrayStart?.(); + stack.push({ kind: "array", array: current, index: 0 }); + } else if (current && typeof current === "object") { + const record = current as Record; + hooks.onObjectStart?.(); + stack.push({ kind: "object", keys: ownEnumerableKeys(record), record, count: 0 }); + } else { + hooks.onValue?.(current); + } + } else if (frame.kind === "array") { + if (frame.index < frame.array.length) { + if (frame.index > 0) hooks.onArraySeparator?.(); + stack.push({ kind: "array", array: frame.array, index: frame.index + 1 }); + stack.push({ kind: "value", value: frame.array[frame.index] }); + } else { + hooks.onArrayEnd?.(); + } + } else { + const next = frame.keys.next(); + if (next.done) { + hooks.onObjectEnd?.(); + } else { + if (frame.count > 0) hooks.onObjectSeparator?.(); + hooks.onObjectKey?.(next.value); + stack.push({ kind: "object", keys: frame.keys, record: frame.record, count: frame.count + 1 }); + stack.push({ kind: "value", value: frame.record[next.value] }); + } + } + } +} diff --git a/src/lib/token-estimate.ts b/src/lib/token-estimate.ts index 10b535548f..11aedf05aa 100644 --- a/src/lib/token-estimate.ts +++ b/src/lib/token-estimate.ts @@ -55,15 +55,21 @@ function cjkRatio(text: string): number { return sampled === 0 ? 0 : cjk / sampled; } -/** - * Estimate the token count of a text blob. Pure and deterministic. - * Returns 0 for empty/whitespace-free-empty input; otherwise ceil(length / ratio), min 1. - */ -export function estimateTokens(text: string, modelId?: string): number { +/** Unrounded estimate used when many fields share one admission budget. */ +export function estimateTokenFraction(text: string, modelId?: string): number { if (!text) return 0; const len = text.length; if (len === 0) return 0; let ratio = charsPerToken(modelId); if (cjkRatio(text) > CJK_RATIO_THRESHOLD) ratio = Math.min(ratio, CJK_CHARS_PER_TOKEN); - return Math.max(1, Math.ceil(len / ratio)); + return len / ratio; +} + +/** + * Estimate the token count of a text blob. Pure and deterministic. + * Returns 0 for empty input; otherwise ceil(length / ratio), min 1. + */ +export function estimateTokens(text: string, modelId?: string): number { + const estimate = estimateTokenFraction(text, modelId); + return estimate === 0 ? 0 : Math.max(1, Math.ceil(estimate)); } diff --git a/src/responses/custom-tool-compat.ts b/src/responses/custom-tool-compat.ts index c5feace809..fc5fe08ec2 100644 --- a/src/responses/custom-tool-compat.ts +++ b/src/responses/custom-tool-compat.ts @@ -18,14 +18,27 @@ export function customToolItemId(id: unknown): unknown { return id.startsWith("fc_") ? `ctc_${id.slice(3)}` : id; } +/** + * Lazily enumerate a parsed object's own enumerable string keys. One generator stays + * alive per open object level, so the iterative walks below never materialize key + * arrays proportional to the payload. + */ +function* ownEnumerableKeys(record: Record): Generator { + for (const key in record) { + if (Object.prototype.hasOwnProperty.call(record, key)) yield key; + } +} + export function collectRoutedCustomToolNames(body: unknown): Set { const names = new Set(); - const visit = (value: unknown): void => { + const stack: unknown[] = [body]; + while (stack.length > 0) { + const value = stack.pop()!; if (Array.isArray(value)) { - for (const entry of value) visit(entry); - return; + for (let i = value.length - 1; i >= 0; i--) stack.push(value[i]); + continue; } - if (!isPlainObject(value)) return; + if (!isPlainObject(value)) continue; if ( value.type === "custom" && typeof value.name === "string" @@ -33,42 +46,52 @@ export function collectRoutedCustomToolNames(body: unknown): Set { ) { names.add(value.name); } - for (const entry of Object.values(value)) visit(entry); - }; - visit(body); + for (const key in value) { + if (Object.prototype.hasOwnProperty.call(value, key)) stack.push(value[key]); + } + } return names; } function collectConvertedCallIds(value: unknown, names: ReadonlySet, out: Set): void { - if (Array.isArray(value)) { - for (const entry of value) collectConvertedCallIds(entry, names, out); - return; - } - if (!isPlainObject(value)) return; - if ( - (value.type === "custom_tool_call" || value.type === "function_call") - && typeof value.name === "string" - && names.has(value.name) - && typeof value.call_id === "string" - ) { - out.add(value.call_id); + const stack: unknown[] = [value]; + while (stack.length > 0) { + const current = stack.pop()!; + if (Array.isArray(current)) { + for (let i = current.length - 1; i >= 0; i--) stack.push(current[i]); + continue; + } + if (!isPlainObject(current)) continue; + if ( + (current.type === "custom_tool_call" || current.type === "function_call") + && typeof current.name === "string" + && names.has(current.name) + && typeof current.call_id === "string" + ) { + out.add(current.call_id); + } + for (const key in current) { + if (Object.prototype.hasOwnProperty.call(current, key)) stack.push(current[key]); + } } - for (const entry of Object.values(value)) collectConvertedCallIds(entry, names, out); } -function rewriteForUpstream( - value: unknown, +/** + * Replace a node that the routed-custom-tool rewrite consumes wholesale, or return + * undefined when the node must be traversed normally. These nodes are leaves for the + * rewrite: their inner fields are replaced or dropped rather than rewritten. + */ +function rewriteCustomToolNode( + node: unknown, names: ReadonlySet, callIds: ReadonlySet, ): unknown { - if (Array.isArray(value)) return value.map(entry => rewriteForUpstream(entry, names, callIds)); - if (!isPlainObject(value)) return value; - - if (value.type === "custom" && typeof value.name === "string" && names.has(value.name)) { - const { format: _format, ...rest } = value; - const isDefinition = typeof value.description === "string" - || isPlainObject(value.format) - || isPlainObject(value.parameters); + if (!isPlainObject(node)) return undefined; + if (node.type === "custom" && typeof node.name === "string" && names.has(node.name)) { + const { format: _format, ...rest } = node; + const isDefinition = typeof node.description === "string" + || isPlainObject(node.format) + || isPlainObject(node.parameters); if (!isDefinition) return { ...rest, type: "function" }; return { ...rest, @@ -86,36 +109,146 @@ function rewriteForUpstream( }, }; } - if ( - value.type === "custom_tool_call" - && typeof value.name === "string" - && names.has(value.name) + node.type === "custom_tool_call" + && typeof node.name === "string" + && names.has(node.name) ) { - const { input, id: _id, ...rest } = value; + const { input, id: _id, ...rest } = node; return { ...rest, type: "function_call", arguments: JSON.stringify({ input: typeof input === "string" ? input : "" }), }; } - if ( - value.type === "custom_tool_call_output" - && typeof value.call_id === "string" - && callIds.has(value.call_id) + node.type === "custom_tool_call_output" + && typeof node.call_id === "string" + && callIds.has(node.call_id) ) { - return { ...value, type: "function_call_output" }; + return { ...node, type: "function_call_output" }; } + return undefined; +} + +/** + * Node transform for {@link mapJsonTree}. Return the replacement node, or undefined to + * keep the node — pre-order mode then descends into it (a replacement is a leaf), while + * post-order mode keeps its rebuilt form. + */ +type JsonTreeTransform = (node: Record) => unknown | undefined; - let changed = false; - const next: Record = {}; - for (const [key, entry] of Object.entries(value)) { - const rewritten = rewriteForUpstream(entry, names, callIds); - next[key] = rewritten; - changed ||= rewritten !== entry; +/** + * Copy-on-write JSON tree walk shared by the routed custom-tool upstream rewrite + * (pre-order: a replaced node is a leaf) and the client-facing restore (post-order: + * replacement happens after children are rebuilt). Iterative over container/index + * frames with lazy key enumeration, so deeply nested client payloads cannot overflow + * the call stack; unchanged subtrees keep their original references (change detection + * is reference-based per child). Returns the transformed value and whether any node + * (including the root) was replaced. + */ +export function mapJsonTree( + value: unknown, + transform: JsonTreeTransform, + mode: "pre" | "post", +): { value: unknown; changed: boolean } { + type Slot = { value: unknown }; + type Changed = { flag: boolean }; + type Frame = + | { kind: "node"; node: unknown; slot: Slot } + | { kind: "array"; array: unknown[]; index: number; next: unknown[]; changed: Changed; slot: Slot } + | { kind: "object"; record: Record; keys: Generator; next: Record; changed: Changed; slot: Slot } + | { kind: "array-assign"; next: unknown[]; changed: Changed; index: number; child: unknown; slot: Slot } + | { kind: "object-assign"; next: Record; changed: Changed; key: string; child: unknown; slot: Slot }; + + const rootChanged: Changed = { flag: false }; + const rootSlot: Slot = { value }; + const stack: Frame[] = [{ kind: "node", node: value, slot: rootSlot }]; + while (stack.length > 0) { + const frame = stack.pop()!; + if (frame.kind === "node") { + const node = frame.node; + if (mode === "pre" && isPlainObject(node)) { + const leaf = transform(node); + if (leaf !== undefined) { + frame.slot.value = leaf; + if (frame.slot === rootSlot) rootChanged.flag = true; + continue; + } + } + if (Array.isArray(node)) { + const changed = frame.slot === rootSlot ? rootChanged : { flag: false }; + stack.push({ + kind: "array", + array: node, + index: 0, + next: new Array(node.length), + changed, + slot: frame.slot, + }); + } else if (isPlainObject(node)) { + const changed = frame.slot === rootSlot ? rootChanged : { flag: false }; + stack.push({ + kind: "object", + record: node, + keys: ownEnumerableKeys(node), + next: {}, + changed, + slot: frame.slot, + }); + } else { + frame.slot.value = node; + } + } else if (frame.kind === "array") { + if (frame.index < frame.array.length) { + const child = frame.array[frame.index]; + const childSlot: Slot = { value: child }; + stack.push({ kind: "array", array: frame.array, index: frame.index + 1, next: frame.next, changed: frame.changed, slot: frame.slot }); + stack.push({ kind: "array-assign", next: frame.next, changed: frame.changed, index: frame.index, child, slot: childSlot }); + stack.push({ kind: "node", node: child, slot: childSlot }); + } else { + frame.slot.value = frame.changed.flag ? frame.next : frame.array; + } + } else if (frame.kind === "object") { + const nextKey = frame.keys.next(); + if (nextKey.done) { + const source = frame.changed.flag ? frame.next : frame.record; + if (mode === "post") { + const replacement = transform(source); + if (replacement !== undefined) { + frame.slot.value = replacement; + frame.changed.flag = true; + } else { + frame.slot.value = source; + } + } else { + frame.slot.value = source; + } + } else { + const key = nextKey.value; + const child = frame.record[key]; + const childSlot: Slot = { value: child }; + stack.push({ kind: "object", record: frame.record, keys: frame.keys, next: frame.next, changed: frame.changed, slot: frame.slot }); + stack.push({ kind: "object-assign", next: frame.next, changed: frame.changed, key, child, slot: childSlot }); + stack.push({ kind: "node", node: child, slot: childSlot }); + } + } else if (frame.kind === "array-assign") { + frame.next[frame.index] = frame.slot.value; + frame.changed.flag ||= frame.slot.value !== frame.child; + } else { + frame.next[frame.key] = frame.slot.value; + frame.changed.flag ||= frame.slot.value !== frame.child; + } } - return changed ? next : value; + return { value: rootSlot.value, changed: rootChanged.flag }; +} + +function rewriteForUpstream( + value: unknown, + names: ReadonlySet, + callIds: ReadonlySet, +): unknown { + return mapJsonTree(value, (node) => rewriteCustomToolNode(node, names, callIds), "pre").value; } export function rewriteRoutedCustomToolsForUpstream(body: unknown): { @@ -133,33 +266,23 @@ export function restoreRoutedCustomCalls( value: unknown, names: ReadonlySet, ): { value: unknown; changed: boolean } { - if (Array.isArray(value)) { - let changed = false; - const restored = value.map(entry => { - const result = restoreRoutedCustomCalls(entry, names); - changed ||= result.changed; - return result.value; - }); - return changed ? { value: restored, changed: true } : { value, changed: false }; - } - if (!isPlainObject(value)) return { value, changed: false }; - - let changed = false; - const restored: Record = {}; - for (const [key, entry] of Object.entries(value)) { - const result = restoreRoutedCustomCalls(entry, names); - restored[key] = result.value; - changed ||= result.changed; - } - - if (value.type === "function_call" && typeof value.name === "string" && names.has(value.name)) { - restored.type = "custom_tool_call"; - restored.id = customToolItemId(value.id); - restored.input = customToolInput(value.arguments); - delete restored.arguments; - changed = true; - } - return changed ? { value: restored, changed: true } : { value, changed: false }; + return mapJsonTree(value, (node) => { + if ( + node.type === "function_call" + && typeof node.name === "string" + && names.has(node.name) + ) { + const restored: Record = { + ...node, + type: "custom_tool_call", + id: customToolItemId(node.id), + input: customToolInput(node.arguments), + }; + delete restored.arguments; + return restored; + } + return undefined; + }, "post"); } export function restoreRoutedCustomCallsInJson( diff --git a/src/responses/replay-provenance.ts b/src/responses/replay-provenance.ts new file mode 100644 index 0000000000..3f90d2cbf7 --- /dev/null +++ b/src/responses/replay-provenance.ts @@ -0,0 +1,195 @@ +/** + * Decide whether a chained request demonstrably carries the complete stored prefix. + * + * Content equality is necessary but not sufficient: a legitimate delta may repeat the + * same message text. At least one matched Responses item must also retain a protocol + * identity that a new occurrence cannot reuse (a provider output `id` or tool + * `call_id`). User-authored message ids are deliberately not provenance. + */ + +const MAX_CANONICAL_DEPTH = 256; +const canonicalOverflow = Symbol("canonical-overflow"); +const STABLE_CALL_ID_KEYS = ["call_id", "callId"] as const; + +/** + * Canonicalize a replay item iteratively, sorting object keys and normalizing + * equivalent search shapes. Returns a sentinel when the depth bound is exceeded. + */ +function canonicalValue(value: unknown): unknown { + type Slot = { value: unknown }; + type Frame = + | { kind: "node"; node: unknown; slot: Slot; depth: number } + | { kind: "array"; array: unknown[]; index: number; next: unknown[]; slot: Slot; depth: number } + | { + kind: "object"; + keys: Generator; + record: Record; + next: Record; + slot: Slot; + depth: number; + } + | { kind: "assign"; next: unknown[] | Record; position: number | string; slot: Slot }; + + function* ownEnumerableKeys(record: Record): Generator { + for (const key in record) { + if (Object.prototype.hasOwnProperty.call(record, key)) yield key; + } + } + + let overflowed = false; + const root: Slot = { value }; + const stack: Frame[] = [{ kind: "node", node: value, slot: root, depth: 0 }]; + while (stack.length > 0) { + const frame = stack.pop()!; + if (frame.kind === "node") { + const node = frame.node; + if (Array.isArray(node)) { + if (frame.depth >= MAX_CANONICAL_DEPTH) { + overflowed = true; + frame.slot.value = canonicalOverflow; + } else { + stack.push({ + kind: "array", + array: node, + index: 0, + next: new Array(node.length), + slot: frame.slot, + depth: frame.depth, + }); + } + } else if (node && typeof node === "object") { + if (frame.depth >= MAX_CANONICAL_DEPTH) { + overflowed = true; + frame.slot.value = canonicalOverflow; + } else { + stack.push({ + kind: "object", + record: node as Record, + keys: ownEnumerableKeys(node as Record), + next: Object.create(null), + slot: frame.slot, + depth: frame.depth, + }); + } + } else { + frame.slot.value = node; + } + continue; + } + + if (frame.kind === "array") { + if (frame.index < frame.array.length) { + const child: Slot = { value: frame.array[frame.index] }; + stack.push({ + kind: "array", + array: frame.array, + index: frame.index + 1, + next: frame.next, + slot: frame.slot, + depth: frame.depth, + }); + stack.push({ kind: "assign", next: frame.next, position: frame.index, slot: child }); + stack.push({ kind: "node", node: frame.array[frame.index], slot: child, depth: frame.depth + 1 }); + } else { + frame.slot.value = frame.next; + } + continue; + } + + if (frame.kind === "object") { + const nextKey = frame.keys.next(); + if (!nextKey.done) { + const child: Slot = { value: frame.record[nextKey.value] }; + stack.push({ ...frame }); + stack.push({ kind: "assign", next: frame.next, position: nextKey.value, slot: child }); + stack.push({ kind: "node", node: child.value, slot: child, depth: frame.depth + 1 }); + continue; + } + + const normalized: Record = { ...frame.next }; + if (normalized.type === "search" && typeof normalized.query === "string") { + normalized.queries = Array.isArray(normalized.queries) + ? normalized.queries + : [normalized.query]; + } + const ordered: Record = Object.create(null); + for (const key of Object.keys(normalized).sort()) ordered[key] = normalized[key]; + frame.slot.value = ordered; + continue; + } + + if (typeof frame.position === "number") { + (frame.next as unknown[])[frame.position] = frame.slot.value; + } else { + (frame.next as Record)[frame.position] = frame.slot.value; + } + } + return overflowed ? canonicalOverflow : root.value; +} + +/** Build the content key used to compare stored and resent replay items. */ +function canonicalItemKey(item: unknown): string | undefined { + if (!item || typeof item !== "object" || Array.isArray(item)) return undefined; + const { id: _id, status: _status, sequence_number: _sequenceNumber, ...stable } = + item as Record; + const canonical = canonicalValue(stable); + return canonical === canonicalOverflow ? undefined : JSON.stringify(canonical); +} + +/** + * Check whether matching items retain a provider-issued identity that proves + * they are the same occurrence rather than coincidentally equal content. + */ +function sharesStableIdentity(stored: unknown, request: unknown): boolean { + if ( + !stored + || !request + || typeof stored !== "object" + || typeof request !== "object" + || Array.isArray(stored) + || Array.isArray(request) + ) return false; + const storedRecord = stored as Record; + const requestRecord = request as Record; + const sharesCallId = STABLE_CALL_ID_KEYS.some(key => + typeof storedRecord[key] === "string" + && storedRecord[key] !== "" + && storedRecord[key] === requestRecord[key] + ); + if (sharesCallId) return true; + // User-authored message IDs are not provider provenance: a client may legitimately + // repeat them. Provider output IDs are stable replay anchors when retained. + return storedRecord.role !== "user" + && requestRecord.role !== "user" + && typeof storedRecord.id === "string" + && storedRecord.id !== "" + && storedRecord.id === requestRecord.id; +} + +/** + * Return true only when the request contains the complete stored prefix and at + * least one provider-output item retains a stable protocol identity. + */ +export function hasProvenCompleteReplayPrefix( + stored: readonly unknown[], + requestInput: readonly unknown[], + providerOutputStart: number | undefined, +): boolean { + if ( + stored.length === 0 + || requestInput.length < stored.length + || typeof providerOutputStart !== "number" + || !Number.isSafeInteger(providerOutputStart) + || providerOutputStart < 0 + || providerOutputStart >= stored.length + ) return false; + let hasStableAnchor = false; + for (let index = 0; index < stored.length; index += 1) { + const storedKey = canonicalItemKey(stored[index]); + const requestKey = canonicalItemKey(requestInput[index]); + if (storedKey === undefined || storedKey !== requestKey) return false; + hasStableAnchor ||= index >= providerOutputStart + && sharesStableIdentity(stored[index], requestInput[index]); + } + return hasStableAnchor; +} diff --git a/src/responses/spill-store.ts b/src/responses/spill-store.ts index a35f08e5db..c53ba8f26e 100644 --- a/src/responses/spill-store.ts +++ b/src/responses/spill-store.ts @@ -35,6 +35,7 @@ export interface ResponseSpillPayload { responseId: string; createdAt: number; items: unknown[]; + providerOutputStart?: number; providers?: OcxProviderContinuationState; } @@ -260,10 +261,20 @@ function validPayload(value: unknown, responseId: string): value is ResponseSpil if (!value || typeof value !== "object" || Array.isArray(value)) return false; const payload = value as Record; const keys = Object.keys(payload); - if (keys.some(key => !["version", "responseId", "createdAt", "items", "providers"].includes(key))) return false; + if (keys.some(key => + !["version", "responseId", "createdAt", "items", "providerOutputStart", "providers"].includes(key) + )) return false; if (payload.version !== 1 || payload.responseId !== responseId) return false; if (typeof payload.createdAt !== "number" || !Number.isFinite(payload.createdAt)) return false; if (!Array.isArray(payload.items)) return false; + if ( + payload.providerOutputStart !== undefined + && ( + !Number.isSafeInteger(payload.providerOutputStart) + || (payload.providerOutputStart as number) < 0 + || (payload.providerOutputStart as number) > payload.items.length + ) + ) return false; if (payload.providers !== undefined) { if (!payload.providers || typeof payload.providers !== "object" || Array.isArray(payload.providers)) return false; for (const providerState of Object.values(payload.providers)) { @@ -285,6 +296,9 @@ export function writeResponseSpillDurably( responseId, createdAt: state.createdAt, items: state.items, + ...(state.providerOutputStart !== undefined + ? { providerOutputStart: state.providerOutputStart } + : {}), ...(state.providers ? { providers: state.providers } : {}), }; const serialized = JSON.stringify(payload); diff --git a/src/responses/state.ts b/src/responses/state.ts index a0ef3eb9dc..7601448323 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -13,6 +13,7 @@ import { type ResponseSpillRef, writeResponseSpillDurably, } from "./spill-store"; +import { hasProvenCompleteReplayPrefix } from "./replay-provenance"; const MAX_STORED_RESPONSES = 1_000; const RESPONSE_TTL_MS = 60 * 60 * 1_000; @@ -38,6 +39,8 @@ interface ResidentResponseState { kind: "resident"; createdAt: number; items: unknown[]; + /** First item sourced from provider output; absent on legacy snapshots. */ + providerOutputStart?: number; providers?: OcxProviderContinuationState; sizeBytes: number; } @@ -45,6 +48,8 @@ interface ResidentResponseState { interface SpilledResponseState { kind: "spill"; createdAt: number; + /** Mirrored from the spill payload so provenance survives demotion without a read. */ + providerOutputStart?: number; providers?: OcxProviderContinuationState; spill: ResponseSpillRef; sizeBytes: number; @@ -125,6 +130,9 @@ function measureResidentEntry(id: string, entry: ResidentInput): ResidentRespons responseId: id, createdAt: entry.createdAt, items: entry.items, + ...(entry.providerOutputStart !== undefined + ? { providerOutputStart: entry.providerOutputStart } + : {}), ...(entry.providers ? { providers: entry.providers } : {}), }); return sizeBytes === null ? null : { kind: "resident", ...entry, sizeBytes }; @@ -224,6 +232,9 @@ function swapResidentForSpill(id: string, expected: ResidentResponseState, ref: const base: Omit = { kind: "spill", createdAt: expected.createdAt, + ...(expected.providerOutputStart !== undefined + ? { providerOutputStart: expected.providerOutputStart } + : {}), ...(expected.providers ? { providers: expected.providers } : {}), spill: ref, }; @@ -245,11 +256,17 @@ function replaceSpillEntryAtomically( const ref = writeResponseSpillDurably(id, { createdAt: candidate.createdAt, items: candidate.items, + ...(candidate.providerOutputStart !== undefined + ? { providerOutputStart: candidate.providerOutputStart } + : {}), ...(candidate.providers ? { providers: candidate.providers } : {}), }); const base: Omit = { kind: "spill", createdAt: candidate.createdAt, + ...(candidate.providerOutputStart !== undefined + ? { providerOutputStart: candidate.providerOutputStart } + : {}), ...(candidate.providers ? { providers: candidate.providers } : {}), spill: ref, }; @@ -323,6 +340,9 @@ function admitOversizedCandidate( const ref = writeResponseSpillDurably(id, { createdAt: candidate.createdAt, items: candidate.items, + ...(candidate.providerOutputStart !== undefined + ? { providerOutputStart: candidate.providerOutputStart } + : {}), ...(candidate.providers ? { providers: candidate.providers } : {}), }); // Enforce the ceiling against the REAL envelope: the spill payload adds @@ -337,6 +357,9 @@ function admitOversizedCandidate( const base: Omit = { kind: "spill", createdAt: candidate.createdAt, + ...(candidate.providerOutputStart !== undefined + ? { providerOutputStart: candidate.providerOutputStart } + : {}), ...(candidate.providers ? { providers: candidate.providers } : {}), spill: ref, }; @@ -386,6 +409,7 @@ function snapshotPath(): string { interface LegacySnapshotState { createdAt?: unknown; items?: unknown; + providerOutputStart?: unknown; providers?: OcxProviderContinuationState; conversationId?: unknown; cursorCheckpointUsable?: unknown; @@ -410,6 +434,10 @@ function loadSnapshotEntry(id: string, value: unknown): void { const base: Omit = { kind: "spill", createdAt: rec.createdAt, + ...(Number.isSafeInteger(rec.providerOutputStart) + && (rec.providerOutputStart as number) >= 0 + ? { providerOutputStart: rec.providerOutputStart as number } + : {}), ...(rec.providers ? { providers: rec.providers } : {}), spill: rec.spill, }; @@ -435,6 +463,11 @@ function loadSnapshotEntry(id: string, value: unknown): void { const resident = measureResidentEntry(id, { createdAt: rec.createdAt, items: rec.items, + ...(Number.isSafeInteger(rec.providerOutputStart) + && (rec.providerOutputStart as number) >= 0 + && (rec.providerOutputStart as number) <= rec.items.length + ? { providerOutputStart: rec.providerOutputStart as number } + : {}), ...(providers ? { providers } : {}), }); if (!resident) { @@ -717,6 +750,7 @@ export async function flushResponseState(): Promise { if (persistTimer) await persistNow(pendingPersistPath ?? snapshotPath(), true); } +/** Normalize a Responses `input` field into an item array (strings become a single user item). */ function inputItems(input: unknown): unknown[] { if (input === undefined) return []; if (Array.isArray(input)) return input; @@ -748,6 +782,9 @@ function pruneResponses(at = now()): void { const ref = writeResponseSpillDurably(oldestId, { createdAt: entry.createdAt, items: entry.items, + ...(entry.providerOutputStart !== undefined + ? { providerOutputStart: entry.providerOutputStart } + : {}), ...(entry.providers ? { providers: entry.providers } : {}), }); if (swapResidentForSpill(oldestId, entry, ref)) spillCounters.writes += 1; @@ -789,6 +826,9 @@ export function evictOldestResponseContinuationForBudget(): number { const ref = writeResponseSpillDurably(id, { createdAt: entry.createdAt, items: entry.items, + ...(entry.providerOutputStart !== undefined + ? { providerOutputStart: entry.providerOutputStart } + : {}), ...(entry.providers ? { providers: entry.providers } : {}), }); if (swapResidentForSpill(id, entry, ref)) spillCounters.writes += 1; @@ -829,6 +869,9 @@ function materializeEntry( const state = measureResidentEntry(id, { createdAt: result.payload.createdAt, items: result.payload.items, + ...(result.payload.providerOutputStart !== undefined + ? { providerOutputStart: result.payload.providerOutputStart } + : {}), ...(result.payload.providers ? { providers: result.payload.providers } : {}), }); if (!state) { @@ -840,6 +883,12 @@ function materializeEntry( return { ok: true, state }; } +/** + * Expand a chained /v1/responses request's `previous_response_id` into the full stored + * history when the request carries only a delta. A full-body resend stays untouched only + * when a stable provider item id or tool call id proves the complete canonical prefix is + * replayed; ambiguous content equality stays conservative and preserves every occurrence. + */ export function expandPreviousResponseInput(body: unknown): unknown { if (!body || typeof body !== "object" || Array.isArray(body)) return body; const request = body as Record; @@ -854,11 +903,33 @@ export function expandPreviousResponseInput(body: unknown): unknown { replayFailures.set(request, materialized.failure); return body; } + const storedItems = materialized.state.items; + const requestItems = inputItems(request.input); + // A chained turn may already carry the full conversation (stateless upstreams such as + // DeepSeek force the client to resend it every turn). Prepending the stored history to a + // full-body request duplicates it, and remembering that duplicated body makes the bloat + // sticky across turns: 1x -> 2x -> 3x -> ... (observed 1,333,682 input tokens on + // 2026-08-10, ~10x the real ~127k conversation). Detect the overlap: ONLY a complete + // canonical stored-prefix match with stable protocol identity keeps the request untouched. + // Matching content alone is not proof + // of a full resend — a genuine delta can be as long as the stored history, and returning it + // unchanged would drop the required prefix. Delta turns prepend the stored history and + // append the ENTIRE request input: request items are never dropped, because a repeated + // `context_compaction` marker or an identical message is a new occurrence that must survive. + if (hasProvenCompleteReplayPrefix( + storedItems, + requestItems, + materialized.state.providerOutputStart, + )) { + const full = { ...request }; + replayedInputPrefixLengths.set(full, Math.min(storedItems.length, requestItems.length)); + return full; + } const expanded = { ...request, - input: [...materialized.state.items, ...inputItems(request.input)], + input: [...storedItems, ...requestItems], }; - replayedInputPrefixLengths.set(expanded, materialized.state.items.length); + replayedInputPrefixLengths.set(expanded, storedItems.length); return expanded; } @@ -998,9 +1069,11 @@ export function rememberResponseState( return !!item && typeof item === "object" && (item as { type?: unknown }).type === "function_call"; }); } + const storedInputItems = inputItems(request.input); setResidentEntry(response.id, { createdAt: now(), - items: [...inputItems(request.input), ...response.output], + items: [...storedInputItems, ...response.output], + providerOutputStart: storedInputItems.length, // Always preserve the Cursor conversation id so the next tool-result turn can continue the SAME // Cursor conversation (multi-turn continuation). Separately track whether Cursor's own // checkpoint/cache is safe to reuse: a turn that ended with a pending client tool call produced an diff --git a/src/server/request-decompress.ts b/src/server/request-decompress.ts index 0710470346..6011503ab7 100644 --- a/src/server/request-decompress.ts +++ b/src/server/request-decompress.ts @@ -1,5 +1,6 @@ import { gunzipSync, inflateRawSync, inflateSync, zstdDecompressSync } from "node:zlib"; -import type { TranslatorBudget } from "../lib/translator-budget"; +import { TRANSLATOR_MAX_TURN_BYTES, type TranslatorBudget } from "../lib/translator-budget"; +import { walkJsonTree } from "../lib/json-walk"; /** * Request-body decompression for the /v1/responses data plane. @@ -64,6 +65,91 @@ function cancelReaderWithoutWaiting(reader: ReadableStreamDefaultReader { + let bytes = 0; + for (let i = 0; i < text.length; i++) { + const code = text.charCodeAt(i); + // JSON.stringify escapes a string's structural characters, so the serialized + // copy is LONGER than the raw UTF-8 text. Match the serialized form: `\"` and + // `\\` are two bytes, the five short control escapes (`\b \t \n \f \r`) are two + // bytes, every other control character becomes a six-byte `\uXXXX`, and a lone + // surrogate (which has no valid pair to serialize as-is) also becomes `\uXXXX`. + if (code === 0x22 || code === 0x5c) { + bytes += 2; + } else if (code < 0x20) { + bytes += code === 0x08 || code === 0x09 || code === 0x0a || code === 0x0c || code === 0x0d ? 2 : 6; + } else if (code < 0x80) { + bytes += 1; + } else if (code < 0x800) { + bytes += 2; + } else if (code >= 0xd800 && code <= 0xdbff) { + const next = text.charCodeAt(i + 1); + if (next >= 0xdc00 && next <= 0xdfff) { + bytes += 4; // valid pair serializes as-is: two units -> 4 UTF-8 bytes + i++; + } else { + bytes += 6; // lone high surrogate -> \uXXXX + } + } else if (code >= 0xdc00 && code <= 0xdfff) { + bytes += 6; // lone low surrogate -> \uXXXX + } else { + bytes += 3; + } + if (bytes > cap) break; + } + return bytes; + }; + walkJsonTree(value, { + isDone: () => total > cap, + onValue: (current) => { + if (current === null) { + total += 4; // "null" + } else if (current === undefined) { + // JSON.stringify omits undefined values; keep walking siblings. + } else if (typeof current === "string") { + total += utf8Length(current) + 2; // surrounding quotes + } else if (typeof current === "number") { + total += String(current).length; + } else if (typeof current === "boolean") { + total += current ? 4 : 5; + } + }, + onObjectKey: (key) => { + total += utf8Length(key) + 3; // "key": + }, + onArrayStart: () => { + total += 1; // '[' + }, + onArrayEnd: () => { + total += 1; // ']' + }, + onArraySeparator: () => { + total += 1; // ',' + }, + onObjectStart: () => { + total += 1; // '{' + }, + onObjectEnd: () => { + total += 1; // '}' + }, + onObjectSeparator: () => { + total += 1; // ',' + }, + }); + return Math.min(total, cap); +} + async function readRequestBodyBytesCapped( body: ReadableStream | null, maxBytes: number, @@ -224,7 +310,11 @@ export async function readBoundedJsonRequestBody( return options.emptyBodyFallback; } const parsed = JSON.parse(text); - budget?.observeAcceptedRequestCopy(new TextEncoder().encode(JSON.stringify(parsed)).byteLength); + if (budget) { + budget.observeAcceptedRequestCopy( + boundedJsonSerializedByteLength(parsed, TRANSLATOR_MAX_TURN_BYTES), + ); + } return parsed; } finally { releaseText?.(); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 7fd2b87b03..e3ba67f933 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -50,6 +50,8 @@ import { import { isInjectionDebugEnabled } from "../../lib/debug-settings"; import { injectionDebugLog } from "../../lib/injection-debug-log"; import { resolveClientRetryAfter } from "../../lib/retry-after"; +import { charsPerToken } from "../../lib/token-estimate"; +import { candidateCapabilityEvidence } from "../../routing/capability"; import { enrichOpenCodeZenRateLimitMessage } from "../../providers/opencode-zen-rate-limit"; import { modelInList, namespacedToolName } from "../../types"; import type { AdapterEvent, OcxConfig, OcxParsedRequest, OcxProviderConfig, OcxProviderContinuationState, OcxUsage } from "../../types"; @@ -74,7 +76,7 @@ import { rotateAnthropicAccountOn429, } from "../../oauth/anthropic-routing"; import { buildWebSearchTool, planWebSearch, runWithWebSearch, shouldResolveOpenAiWebSearchSidecar } from "../../web-search"; -import { buildImageTool, buildVideoTool, planImageBridge, planVideoBridge, runWithImageBridge, clampImageMaxRounds, IMAGE_GEN_TOOL_NAME, VIDEO_GEN_TOOL_NAME } from "../../images"; +import { buildImageTool, buildVideoTool, planImageBridge, planImageBridgeSync, planVideoBridge, planVideoBridgeSync, runWithImageBridge, clampImageMaxRounds, IMAGE_GEN_TOOL_NAME, VIDEO_GEN_TOOL_NAME } from "../../images"; import { describeImagesInPlace, planVisionSidecar, resolveOpenAiVisionModel, shouldResolveOpenAiVisionSidecar, stripImagesInPlace } from "../../vision"; import { createAdapterEventQueue, preflightAdapterEvents } from "../../adapters/run-turn-queue"; import { @@ -144,6 +146,9 @@ import { applySubagentModelFallback, maybePrimeSubagentQuota, recordSubagentQuotaFailureForThreadSpawn, + resolveAgentModelFallbackForPrimary, + resolveConfiguredModelFallbackForPrimary, + subagentFallbackGuidanceText, } from "../../codex/subagent-model-fallback"; import { isNativeMainTrafficBlocked } from "../../codex/native-profile-startup"; import { @@ -200,9 +205,19 @@ import { restoreImageGenCallsInJson, } from "../responses-image-gen-repair"; import { createResponsesModelPayloadRewrite, rewriteResponsesModelJson } from "../responses-model-rewrite"; -import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; +import { MAX_SPAWN_AGENT_MODEL_OVERRIDES, type EffectiveSubagentRoster, type SpawnAgentSurface } from "../../codex/catalog"; +import { CODEX_REASONING_LEVELS } from "../../reasoning-effort"; -import { buildToolBridgeMaps, collabSurface, injectDeveloperMessage, multiAgentGuidanceText } from "./collaboration"; +import { + applyInjectionPlaceholders, + buildToolBridgeMaps, + collabSurface, + injectDeveloperMessage, + multiAgentGuidanceText, + PROACTIVE_MULTI_AGENT_MODE_TEXT, + subagentRosterText, + V2_GUIDANCE_CHAR_BUDGET, +} from "./collaboration"; import { hasUnreadableEncryptedAgentTask, looksLikeBackendCiphertext, sanitizeEncryptedContentInPlace } from "./encrypted-payload"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers"; import { classifyTransportFailureKind, transportErrorCode } from "../../lib/upstream-reachability"; @@ -232,6 +247,11 @@ import { createRoutedCustomToolRestoreBlockRewrite } from "../responses-custom-t import { createGithubCopilotResponsesBlockRewrite } from "../github-copilot-responses-repair"; import { responsesJsonToSseStream } from "../responses-json-events"; import { guardTerminalEventStream } from "./terminal-guard"; +import { + estimateAdmissionInput as estimateParsedAdmissionInput, + hardAdmissionThreshold, + shouldRejectEstimatedInput, +} from "./input-admission"; /** * Adapters whose continuation state must survive Codex's store:false requests. @@ -1529,7 +1549,7 @@ async function handleResponsesInner( ); } - let parsed; + let parsed: OcxParsedRequest; let toolBridgeMaps: ReturnType; try { parsed = parseRequest(body); @@ -1609,8 +1629,197 @@ async function handleResponsesInner( return formatErrorResponse(404, "invalid_request_error", err instanceof Error ? err.message : String(err)); } + /** + * Resolve the routed model's effective input limit. Prefer the per-model maximum + * input, then use the context window from the shared route-capability chain. + */ + const effectiveInputLimitFor = (candidateRoute: typeof route): number | undefined => { + const limit = + candidateRoute.provider.modelMaxInputTokens?.[candidateRoute.modelId] + ?? candidateCapabilityEvidence(config, candidateRoute.providerName, candidateRoute.modelId).contextWindow; + return typeof limit === "number" && limit > 0 ? limit : undefined; + }; + + /** Format the stable public 413 contract for an input rejected by admission. */ + const oversizedInputResponse = ( + modelId: string, + effectiveLimit: number, + estimatedInputTokens: number, + ): Response => Response.json( + { + error: { + type: "request_too_large", + code: "input_context_window_exceeded", + message: `input estimate (≈${estimatedInputTokens} tokens) exceeds the safe admission threshold for ${modelId} (${effectiveLimit}-token input limit); refusing to forward`, + }, + }, + { status: 413 }, + ); + + type InputAdmissionOptions = { + estimateGuidance?: boolean; + reservePendingMutations?: boolean; + }; + type InputAdmissionFailure = { + modelId: string; + effectiveLimit: number; + estimatedInputTokens: number; + }; + + /** + * Project deterministic prompt text that later request normalization or bridge + * planning will add, without performing quota, sidecar, or provider I/O. + */ + const projectedAdmissionText = ( + candidateRoute: typeof route, + candidateParsed: OcxParsedRequest, + options: InputAdmissionOptions, + ): string[] => { + const projected: string[] = []; + if (options.estimateGuidance) { + const surface = collabSurface(candidateParsed); + if ( + surface === "v1" + && (candidateParsed.options.reasoning === "max" || candidateParsed.options.reasoning === "ultra") + ) { + projected.push(`${PROACTIVE_MULTI_AGENT_MODE_TEXT}`); + } else if ( + surface === "v2" + && config.multiAgentGuidanceEnabled !== false + && ( + typeof config.injectionModel === "string" + || (config.subagentModels?.length ?? 0) > 0 + || (config.subagentModelFallback?.length ?? 0) > 0 + ) + ) { + const candidates = [ + ...(typeof config.injectionModel === "string" ? [config.injectionModel] : []), + ...(config.subagentModels ?? []), + ]; + const model = candidates.reduce( + (longest, candidate) => candidate.length > longest.length ? candidate : longest, + "", + ); + const fallback = subagentFallbackGuidanceText(config); + if (typeof config.injectionPrompt === "string" && config.injectionPrompt.length > 0) { + const allEfforts = CODEX_REASONING_LEVELS.map(level => level.effort); + const reserveModelLength = Math.max(256, model.length); + const conservativeModels = Array.from( + { length: Math.min(MAX_SPAWN_AGENT_MODEL_OVERRIDES, candidates.length) }, + (_, index) => `reserve_${index}`.padEnd(reserveModelLength, "x"), + ); + const roster = subagentRosterText( + conservativeModels.map((candidate, index) => ({ + model: candidate, + // Distinct sentinels force the longer per-model roster form while retaining + // every real effort label. This is estimation text only, never injected. + efforts: [...allEfforts, `reserve_${index}`], + })), + ); + projected.push( + `${applyInjectionPlaceholders( + config.injectionPrompt, + model.padEnd(model.length > 0 ? reserveModelLength : 0, "x"), + config.injectionEffort, + roster, + fallback, + )}`, + ); + } else { + // Default v2 guidance depends on the local Codex catalog, which is resolved only + // during final-route normalization. Reserve its documented character budget now + // so a thread-spawn request cannot cross the hard threshold only after quota I/O. + // The fallback and longest configured selector are added separately because they + // can remain after an oversized roster is dropped from the bounded default text. + projected.push(`${"x".repeat(V2_GUIDANCE_CHAR_BUDGET)}`); + if (fallback) projected.push(fallback); + if (model) projected.push(model.padEnd(256, "x")); + if (config.injectionEffort) projected.push(config.injectionEffort); + } + } + } + if (options.reservePendingMutations) { + if ( + candidateParsed._compactionRequest === true + && !isCanonicalOpenAiForwardProvider(candidateRoute.provider) + ) { + projected.push(COMPACT_PROMPT); + } + // These bridges replace hosted capabilities with small deterministic function + // definitions later. Reserve only routes where the corresponding bridge can run, + // keeping near-threshold native passthrough requests out of a false 413. + if ( + candidateParsed._webSearch !== undefined + && candidateRoute.provider.adapter !== "openai-responses" + ) { + projected.push(JSON.stringify(buildWebSearchTool())); + } + if (planImageBridgeSync(config, candidateParsed, candidateRoute.provider)) { + projected.push(JSON.stringify(buildImageTool())); + } + if (planVideoBridgeSync(config, candidateParsed, candidateRoute.provider)) { + projected.push(JSON.stringify(buildVideoTool())); + } + } + return projected; + }; + + /** Estimate one routed candidate against its hard admission scan cap. */ + const estimateInputFor = ( + candidateRoute: typeof route, + candidateParsed: OcxParsedRequest, + effectiveLimit: number, + options: InputAdmissionOptions = {}, + ): number => estimateParsedAdmissionInput( + candidateParsed, + candidateRoute.modelId, + hardAdmissionThreshold(effectiveLimit), + { extraText: projectedAdmissionText(candidateRoute, candidateParsed, options) }, + ); + + /** Return structured rejection evidence for one routed candidate, if oversized. */ + const inputAdmissionFor = ( + candidateRoute: typeof route, + candidateParsed: OcxParsedRequest = parsed, + options: InputAdmissionOptions = {}, + ): InputAdmissionFailure | undefined => { + const effectiveLimit = effectiveInputLimitFor(candidateRoute); + if (effectiveLimit === undefined) return undefined; + const estimatedInputTokens = estimateInputFor( + candidateRoute, + candidateParsed, + effectiveLimit, + options, + ); + return shouldRejectEstimatedInput(estimatedInputTokens, effectiveLimit) + ? { modelId: candidateRoute.modelId, effectiveLimit, estimatedInputTokens } + : undefined; + }; + + /** Convert a candidate's admission result into the public response contract. */ + const inputGuardFor = ( + candidateRoute: typeof route, + candidateParsed: OcxParsedRequest = parsed, + options: InputAdmissionOptions = {}, + ): Response | undefined => { + const failure = inputAdmissionFor(candidateRoute, candidateParsed, options); + return failure + ? oversizedInputResponse(failure.modelId, failure.effectiveLimit, failure.estimatedInputTokens) + : undefined; + }; + const hasUnexpandedPreviousResponse = !!parsed.previousResponseId && parsed._previousResponseInputExpanded !== true; + // Run admission before thread-spawn quota polling: an oversized request must not perform + // upstream I/O (maybePrimeSubagentQuota) before it receives the clean 413. The canonical + // OpenAI continuation-miss path is excluded so its existing 400 keeps precedence; that + // path already skips quota polling. + const initialInputGuard = hasUnexpandedPreviousResponse + && isCanonicalOpenAiForwardProvider(route.provider) + ? undefined + : inputGuardFor(route, parsed, { estimateGuidance: true, reservePendingMutations: true }); + if (initialInputGuard) return initialInputGuard; + // Exact account selectors are isolated from Pool-wide quota work. A canonical replay miss must // also fail closed without polling quota upstream. Cached fallback state can still select a // provider with native continuation support below. @@ -1632,6 +1841,85 @@ async function handleResponsesInner( const parentThreadId = req.headers.get("x-codex-parent-thread-id")?.trim() ?? null; try { + // Validate EVERY locally selectable fallback route before quota priming: the quota + // probe can change model availability, so a stricter fallback candidate may be + // selected after priming. A thread-spawn request must not run the Codex quota probe + // (upstream I/O) for an input that any candidate would reject. The initial route is + // guarded above; the final route is re-validated after fallback selection below. + if ( + threadSpawn + && !options.comboAttempt + && route.codexAccountId === undefined + && !(hasUnexpandedPreviousResponse && isCanonicalOpenAiForwardProvider(route.provider)) + ) { + const candidateChain = [ + parsed.modelId, + ...resolveConfiguredModelFallbackForPrimary(parsed.modelId, config), + ...(config.subagentModelFallback ?? []), + ...resolveAgentModelFallbackForPrimary(parsed.modelId, undefined, config.codexAccountNamespaces), + ]; + const seenCandidates = new Set(); + // Resolve every candidate once; the payload walk below is identical across + // candidates (only the chars-per-token ratio and the limit vary), so re-scanning + // per candidate would multiply a near-limit conversation scan by the candidate + // count on the request thread before any I/O. + const candidateLimits: Array<{ route: typeof route; limit: number }> = []; + for (const candidate of candidateChain) { + if (typeof candidate !== "string" || candidate.length === 0) continue; + if (seenCandidates.has(candidate)) continue; + seenCandidates.add(candidate); + if (slugsEquivalent(candidate, route.modelId)) continue; + try { + const candidateRoute = routeModel(config, candidate, evidenceFromBody(parsed._rawBody)); + const limit = effectiveInputLimitFor(candidateRoute); + if (limit !== undefined) candidateLimits.push({ route: candidateRoute, limit }); + } catch { + // A fallback candidate that cannot be routed NOW can never be selected after + // quota priming, so it carries no admission risk. A stale entry (removed or + // disabled provider, exhausted combo, unavailable policy target) must not + // fail the primary request, which is already routed and guarded above, and + // must not overwrite the primary decision recorded on that route. + continue; + } + } + const byProjection = new Map>(); + for (const entry of candidateLimits) { + const ratio = charsPerToken(entry.route.modelId); + const projected = projectedAdmissionText( + entry.route, + parsed, + { estimateGuidance: true, reservePendingMutations: true }, + ); + // Equal char ratios are not enough to share a scan: route capabilities can + // reserve different compaction or bridge mutations. + const key = JSON.stringify([ratio, projected]); + const group = byProjection.get(key) ?? []; + group.push(entry); + byProjection.set(key, group); + } + for (const group of byProjection.values()) { + const minLimit = Math.min(...group.map(entry => entry.limit)); + // Cap the walk at the strictest candidate's hard admission threshold. If the + // running estimate crosses it, that candidate rejects; otherwise the estimate + // is complete and can be compared against every candidate's own threshold. + const estimate = estimateInputFor( + group[0]!.route, + parsed, + minLimit, + { estimateGuidance: true, reservePendingMutations: true }, + ); + if (shouldRejectEstimatedInput(estimate, minLimit)) { + const strictest = group.find(entry => entry.limit === minLimit)!; + return oversizedInputResponse(strictest.route.modelId, minLimit, estimate); + } + for (const { route: candidateRoute, limit } of group) { + if (shouldRejectEstimatedInput(estimate, limit)) { + return oversizedInputResponse(candidateRoute.modelId, limit, estimate); + } + } + } + } + if ( threadSpawn && route.codexAccountId === undefined @@ -1808,6 +2096,15 @@ async function handleResponsesInner( ); } + // Input-size guard, final route: subagent fallback may have settled a different model or + // provider, so re-validate before auth, adapter construction, or upstream I/O. + const finalInputGuard = inputGuardFor( + route, + parsed, + { estimateGuidance: true, reservePendingMutations: true }, + ); + if (finalInputGuard) return finalInputGuard; + // Captured before normalization: whether the CLIENT asked for SSE. The // transport-neutral upstream-streaming policy below may force a bounded JSON // upstream for reliability (#875); the answer must then be reframed to SSE @@ -2114,6 +2411,14 @@ async function handleResponsesInner( parsed.context.messages.push({ role: "user", content: COMPACT_PROMPT, timestamp: Date.now() }); } + // The pre-auth admission pass reserves every mutation below: bounded per-image + // descriptions, COMPACT_PROMPT, and all possible bridge tool definitions. Therefore + // no 413 path remains after optional vision-sidecar I/O. planWebSearch reads the auth + // store, so compute it ONCE here and reuse it in the dispatch path below. + const wsPlan = !routedCompaction && !isPassthrough + ? planWebSearch(config, parsed, false, route.provider, route.modelId, openAiSidecar) + : undefined; + if ("passthrough" in adapter && adapter.passthrough && !routedCompaction) { let hostAdmissionLease = pendingHostAdmissionLease; pendingHostAdmissionLease = null; @@ -2846,9 +3151,6 @@ async function handleResponsesInner( // - non-runTurn: web-search wins over image when both eligible (documented priority) // - runTurn: image bridge may run (it supports runTurn); web-search is skipped so runTurn // can proceed for web-search-only turns - const wsPlan = !routedCompaction - ? planWebSearch(config, parsed, false, route.provider, route.modelId, openAiSidecar) - : undefined; const imgPlan = !routedCompaction ? await planImageBridge(config, parsed, route.provider) : undefined; const vidPlan = !routedCompaction ? await planVideoBridge(config, parsed, route.provider) : undefined; const canRunWebSearch = !!wsPlan && !adapter.runTurn; @@ -3575,6 +3877,17 @@ async function handleResponsesInner( * never sees a second hidden HTTP response or an unbounded retry loop. */ const fetchTerminalGuardContinuation = async function* (nextParsed: OcxParsedRequest): AsyncGenerator { + const admissionFailure = inputAdmissionFor(route, nextParsed); + if (admissionFailure) { + yield { + type: "error", + status: 413, + errorType: "request_too_large", + code: "input_context_window_exceeded", + message: `input estimate (≈${admissionFailure.estimatedInputTokens} tokens) exceeds the safe admission threshold for ${admissionFailure.modelId} (${admissionFailure.effectiveLimit}-token input limit); refusing terminal continuation`, + }; + return; + } let response: Response | undefined; // One-shot recovery label for the next top-of-loop continuation send after a failover rotation. let nextContinuationRecoveryKind: AttemptRecoveryKind | undefined; diff --git a/src/server/responses/input-admission.ts b/src/server/responses/input-admission.ts new file mode 100644 index 0000000000..84522a5c74 --- /dev/null +++ b/src/server/responses/input-admission.ts @@ -0,0 +1,87 @@ +import { walkJsonTree } from "../../lib/json-walk"; +import { estimateTokenFraction } from "../../lib/token-estimate"; +import type { OcxParsedRequest } from "../../types"; + +export const ADMISSION_ESTIMATE_HEADROOM_RATIO = 0.1; +// Raw base64 length is unrelated to model input tokens. This fixed reserve also +// dominates the vision sidecar's bounded per-image text replacement. +const IMAGE_TOKEN_ESTIMATE = 1_024; + +export interface InputAdmissionEstimateOptions { + extraText?: readonly string[]; +} + +/** Add the estimator uncertainty band to a model's advertised input limit. */ +export function hardAdmissionThreshold(inputLimit: number): number { + return Math.ceil(inputLimit * (1 + ADMISSION_ESTIMATE_HEADROOM_RATIO)); +} + +/** Decide whether an approximate token estimate is safely beyond the hard threshold. */ +export function shouldRejectEstimatedInput(estimatedTokens: number, inputLimit: number): boolean { + return estimatedTokens > hardAdmissionThreshold(inputLimit); +} + +/** + * Estimate prompt-bearing parsed input without materializing another serialized request. + * The caller supplies a hard scan cap, normally `hardAdmissionThreshold(limit)`. + */ +export function estimateAdmissionInput( + parsed: OcxParsedRequest, + modelId: string, + scanCap: number, + options: InputAdmissionEstimateOptions = {}, +): number { + let estimatedTokens = 0; + const isDone = () => estimatedTokens > scanCap; + const countText = (text: string) => { + estimatedTokens += estimateTokenFraction(text, modelId); + }; + const countJsonTokens = (value: unknown): void => { + walkJsonTree(value, { + isDone, + onValue: current => { + if (typeof current === "string") countText(current); + else if (typeof current === "number" || typeof current === "boolean") countText(String(current)); + }, + onObjectKey: countText, + }); + }; + + for (const text of options.extraText ?? []) { + if (text.length > 0) countText(text); + if (isDone()) return Math.ceil(estimatedTokens); + } + for (const message of parsed.context.messages) { + if (isDone()) break; + if (typeof message.content === "string") { + countText(message.content); + } else { + for (const part of message.content) { + if (part.type === "text") countText(part.text); + else if (part.type === "image") estimatedTokens += IMAGE_TOKEN_ESTIMATE; + else countJsonTokens(part); + if (isDone()) break; + } + } + for (const [key, value] of Object.entries(message)) { + if (key === "content" || value === undefined) continue; + countText(key); + countJsonTokens(value); + if (isDone()) break; + } + } + for (const prompt of parsed.context.systemPrompt ?? []) { + if (isDone()) break; + countText(prompt); + } + for (const tool of parsed.context.tools ?? []) { + if (isDone()) break; + countJsonTokens(tool); + } + if (!isDone() && parsed.options.textFormat !== undefined) countJsonTokens(parsed.options.textFormat); + if (!isDone() && parsed._webSearch !== undefined) countJsonTokens(parsed._webSearch); + if (!isDone() && parsed._imageGeneration?.originalTool !== undefined) { + countJsonTokens(parsed._imageGeneration.originalTool); + } + return Math.ceil(estimatedTokens); +} diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 1b09e8fdde..41c08fb44d 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -56,6 +56,41 @@ policy. - 다른 대안 대신 이 방식을 선택한 이유: The endpoint mismatch is reproducible from current code and upstream documentation, whereas a current-dev live canary has not established the separate terminal-delivery policy. - 장점, 단점 및 영향: Luna reaches its documented endpoint across inbound surfaces and explicit opt-out still works; any future stream workaround remains a separately reviewed compatibility decision. +### Responses input admission and replay expansion + +Chained `previous_response_id` turns expand from the local continuation store unless a complete +canonical prefix also carries stable replay provenance (a retained provider item `id` or tool +`call_id`). Content equality alone is ambiguous—a real delta may repeat an earlier item—so it stays +conservative and preserves both occurrences. Proven full-history resends remain untouched because +stateless upstreams such as DeepSeek force the client to resend the full conversation every turn. +Parsed input estimated above the routed model's effective limit plus a 10% estimator uncertainty +band is rejected with `413 request_too_large` before any upstream I/O; normal clients compact well before +this limit, so an oversized body indicates abnormal duplication (observed 1x → 2x → 3x → 4x +expansion and a Windows native crash, #314). The effective limit is the routed model's +`modelMaxInputTokens` value when configured, falling back to `modelContextWindows`. The token +estimate reuses the model-aware estimator that already drives usage and compaction: parsed message +text parts, `systemPrompt` instructions, tool names, tool descriptions, and serialized parameter +schemas are all counted, without materializing a second copy of the request body. Fractional +per-field estimates are summed before rounding, and image parts use a bounded image-token reserve +instead of treating base64 bytes as prompt text. Pre-quota checks +also count the multi-agent guidance that route normalization would inject (the v1 proactive text, +or the v2 injectionPrompt with resolved model/effort/roster/fallback placeholders), and input size +reserves bounded default guidance, compaction text, bridge tools, and vision-description expansion. +Terminal-guard continuations are checked again before their own send. Thus an initial `413` +lands before quota, sidecar, +adapter, or model-serving upstream I/O. For HTTP requests it also lands before +authentication; WebSocket frames have already passed handshake authentication and origin admission, +so the guard runs before per-turn adapter construction and upstream I/O. The thread-spawn quota +probe runs only after this admission pass. + +[Decision Log] +- 목적과 의도: Stop chained-turn replay from compounding stored history and refuse oversized Responses input before upstream I/O. +- 기존 구현 및 제약 조건: `expandPreviousResponseInput` prepended stored history unconditionally; stateless upstreams such as DeepSeek make the client resend the full conversation while still chaining `previous_response_id`, so prepending duplicated it, and recording the duplicated body made the bloat sticky across turns (1x → 2x → 3x → 4x; observed ~1.6M input tokens against a ~400k conversation). Forwarding the oversized body on Windows ballooned bun RSS and native-crashed the whole proxy (upstream Bun memory bug, #314). +- 검토한 주요 대안: Keep unconditional prepending; detect full resends by request length alone; run an exact tokenizer for admission; reject every request at or over the window; materialize and measure the whole body upfront. +- 선택한 방식: Only a complete canonical stored-prefix match carrying a retained provider item id or tool call id keeps a chained request untouched; ambiguous or partial matches preserve the stored history and every request item. Admission sums unrounded model-aware text estimates, reserves image and deterministic post-parse mutations separately, and hard-rejects only above a 10% uncertainty band. Initial and terminal-continuation checks reuse the same admission helper before their respective upstream sends. +- 다른 대안 대신 이 방식을 선택한 이유: Request length is not proof of a full resend (a genuine delta can be as long as stored history), an exact tokenizer would duplicate model-specific estimation logic and cost memory, and rejecting at the window boundary would break legitimate near-window traffic. The overlap heuristic fixes the observed compounding while staying conservative on ambiguous shapes. +- 장점, 단점 및 영향: Proven full-history chained turns stay 1x for stateless upstreams, genuine repeated delta items are never silently dropped, and abnormal duplication fails one request cleanly instead of crashing the service. A client that strips every provider/tool identity may retain duplicate history rather than risk data loss, and estimates inside the uncertainty band are forwarded for the provider's tokenizer to decide. + ### Passthrough SSE stream shapes (#314) Native passthrough SSE has TWO shapes, selected per request in diff --git a/tests/request-decompress.test.ts b/tests/request-decompress.test.ts index 81023496ae..f7fdeb6930 100644 --- a/tests/request-decompress.test.ts +++ b/tests/request-decompress.test.ts @@ -10,6 +10,7 @@ import { import { MANAGEMENT_JSON_BODY_MAX_BYTES } from "../src/server/management/body"; import { handleManagementAPI } from "../src/server/management-api"; import type { OcxConfig } from "../src/types"; +import { createTestTranslatorBudget } from "./helpers/translator-budget"; const PAYLOAD = { model: "gpt-5.5", input: "hello", stream: true }; const PAYLOAD_BYTES = new TextEncoder().encode(JSON.stringify(PAYLOAD)); @@ -349,4 +350,52 @@ describe("readJsonRequestBody", () => { }); await expect(readJsonRequestBody(req)).rejects.toBeInstanceOf(SyntaxError); }); + + test("walks deeply nested parsed bodies without overflowing the call stack", async () => { + // The bundled runtime's JSON.parse accepts ~200k nesting, while a plain recursive + // walk overflows around 100k; the frame-based budget walk must survive deeper input. + const DEPTH = 120_000; + const deepJson = `${"[".repeat(DEPTH)}0${"]".repeat(DEPTH)}`; + const req = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: deepJson, + }); + const parsed = await readJsonRequestBody(req, createTestTranslatorBudget()); + let node: unknown = parsed; + for (let i = 0; i < DEPTH; i++) node = (node as unknown[])[0]; + expect(node).toBe(0); + }); + + test("charges the serialized size including JSON.stringify escape overhead", async () => { + // The budget charge must match the copy the proxy actually retains: JSON.stringify + // re-escapes quotes, backslashes, control characters, and lone surrogates, so a raw + // UTF-8 byte count would under-charge every escape-heavy body. The text also covers + // every remaining utf8Length branch: a valid surrogate pair (emoji), a two-byte BMP + // character ("é"), and a three-byte BMP character ("中"). + const payload = { + model: "deepseek-v4-flash", + input: [ + { + role: "user", + content: [ + { + type: "input_text", + text: "line\nbreak\t\b\f\rtab \"quoted\" \\backslash\u0001\u001f lone:\ud800 é 中 😀", + }, + ], + }, + ], + }; + const req = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(payload), + }); + const budget = createTestTranslatorBudget(); + const parsed = await readJsonRequestBody(req, budget); + expect(parsed).toEqual(payload); + const serializedBytes = new TextEncoder().encode(JSON.stringify(parsed)).byteLength; + expect(budget.snapshot().currentBytes).toBe(serializedBytes); + }); }); diff --git a/tests/responses-custom-tool-repair.test.ts b/tests/responses-custom-tool-repair.test.ts index b68e9b4d9b..c2f3d9c12c 100644 --- a/tests/responses-custom-tool-repair.test.ts +++ b/tests/responses-custom-tool-repair.test.ts @@ -1,6 +1,8 @@ import { describe, expect, test } from "bun:test"; import { collectRoutedCustomToolNames, + mapJsonTree, + restoreRoutedCustomCalls, restoreRoutedCustomCallsInJson, rewriteRoutedCustomToolsForUpstream, } from "../src/responses/custom-tool-compat"; @@ -20,6 +22,17 @@ function frame(event: string, payload: Record): string { } describe("routed Responses custom-tool compatibility", () => { + test("reports a root replacement as changed in the pre-order traversal", () => { + const raw = { type: "custom", name: "exec", description: "Run JavaScript" }; + const { value, changed } = mapJsonTree( + raw, + (node) => (node.type === "custom" ? { ...node, type: "function" } : undefined), + "pre", + ); + expect(changed).toBe(true); + expect(value).toEqual({ type: "function", name: "exec", description: "Run JavaScript" }); + }); + test("rewrites exec definitions and paired history without touching apply_patch", () => { const raw = { model: "deepseek-v4-flash", @@ -67,6 +80,85 @@ describe("routed Responses custom-tool compatibility", () => { expect(body.input[3]).toEqual(raw.input[3]); }); + test("rewrites deep custom-tool payloads without overflowing the call stack", () => { + const DEPTH = 60_000; + let nested: unknown = { + type: "custom", + name: "exec", + description: "Run JavaScript", + format: { type: "grammar", syntax: "lark" }, + }; + for (let i = 0; i < DEPTH; i++) nested = { wrapper: { inner: nested } }; + const raw = { model: "deepseek-v4-flash", tools: [nested] }; + + expect(collectRoutedCustomToolNames(raw)).toEqual(new Set(["exec"])); + const rewritten = rewriteRoutedCustomToolsForUpstream(raw); + expect(rewritten.names).toEqual(new Set(["exec"])); + expect(rewritten.body).not.toBe(raw); + + let node: unknown = rewritten.body; + node = (node as { tools: unknown[] }).tools[0]; + for (let i = 0; i < DEPTH; i++) node = (node as { wrapper: { inner: unknown } }).wrapper.inner; + expect(node).toMatchObject({ + type: "function", + name: "exec", + parameters: { type: "object", properties: { input: { type: "string" } } }, + }); + }); + + test("preserves unchanged sibling subtrees by reference during the upstream rewrite", () => { + const nestedRef = { leaf: "keep me" }; + const hugeSibling = { + type: "function", + name: "unchanged_tool", + parameters: { type: "object", properties: { nested: nestedRef } }, + }; + const raw = { + model: "deepseek-v4-flash", + tools: [ + { type: "custom", name: "exec", description: "Run JavaScript", format: { type: "grammar" } }, + hugeSibling, + ], + }; + + const rewritten = rewriteRoutedCustomToolsForUpstream(raw); + expect(rewritten.names).toEqual(new Set(["exec"])); + const body = rewritten.body as typeof raw; + expect(body.tools[0]).toMatchObject({ type: "function", name: "exec" }); + // Copy-on-write: the unchanged sibling keeps its exact reference, including its + // nested subtree, so large untouched payloads are never duplicated in memory. + expect(body.tools[1]).toBe(hugeSibling); + expect( + (body.tools[1] as { parameters: { properties: { nested: unknown } } }) + .parameters.properties.nested, + ).toBe(nestedRef); + }); + + test("restores deep function-call payloads without overflowing the call stack", () => { + const DEPTH = 60_000; + let nested: unknown = { + type: "function_call", + id: "fc_exec", + call_id: "call_exec", + name: "exec", + arguments: "{\"input\":\"echo\"}", + status: "completed", + }; + for (let i = 0; i < DEPTH; i++) nested = { wrapper: { inner: nested } }; + + const restored = restoreRoutedCustomCalls(nested, new Set(["exec"])); + expect(restored.changed).toBe(true); + let node: unknown = restored.value; + for (let i = 0; i < DEPTH; i++) node = (node as { wrapper: { inner: unknown } }).wrapper.inner; + expect(node).toMatchObject({ + type: "custom_tool_call", + id: "ctc_exec", + name: "exec", + input: "echo", + }); + expect((node as Record).arguments).toBeUndefined(); + }); + test("restores non-streaming exec calls while leaving ordinary functions alone", () => { const upstream = JSON.stringify({ id: "resp_1", diff --git a/tests/responses-input-guard.test.ts b/tests/responses-input-guard.test.ts new file mode 100644 index 0000000000..9900bd0ea1 --- /dev/null +++ b/tests/responses-input-guard.test.ts @@ -0,0 +1,844 @@ +/** + * Regression coverage for the responses input-size guard: a request whose input + * exceeds the model's advertised context window must be rejected with a clean 413 + * instead of being forwarded (forwarding a ~1.6M-token duplication on Windows + * ballooned bun RSS and native-crashed the whole proxy, issue #314). + */ +import { afterEach, beforeEach, describe, expect, setDefaultTimeout, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + resetSubagentModelFallbackStateForTests, + setSubagentQuotaPrimeForTests, +} from "../src/codex/subagent-model-fallback"; +import { handleResponses } from "../src/server/responses"; +import type { OcxConfig } from "../src/types"; +import type { RequestLogContext } from "../src/server/request-log"; +import { + applyInjectionPlaceholders, + PROACTIVE_MULTI_AGENT_MODE_TEXT, + subagentRosterText, +} from "../src/server/responses/collaboration"; +import { estimateTokens } from "../src/lib/token-estimate"; +import { COMPACT_PROMPT } from "../src/responses/compaction"; +import { encodeReasoningEnvelope } from "../src/responses/reasoning-envelope"; +import { hardAdmissionThreshold } from "../src/server/responses/input-admission"; + +setDefaultTimeout(30_000); + +const originalFetch = globalThis.fetch; +let testDir: string; +let previousOpencodexHome: string | undefined; +let previousCodexHome: string | undefined; + +beforeEach(() => { + testDir = mkdtempSync(join(tmpdir(), "ocx-input-guard-")); + previousOpencodexHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = testDir; + process.env.CODEX_HOME = testDir; + resetSubagentModelFallbackStateForTests(); +}); + +afterEach(() => { + resetSubagentModelFallbackStateForTests(); + globalThis.fetch = originalFetch; + rmSync(testDir, { recursive: true, force: true }); + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; +}); + +function deepseekConfig(options: { contextWindow?: number; maxInput?: number } = {}): OcxConfig { + return { + port: 0, + defaultProvider: "deepseek", + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + ...(options.maxInput !== undefined + ? { modelMaxInputTokens: { "deepseek-v4-flash": options.maxInput } } + : {}), + ...(options.contextWindow !== undefined + ? { modelContextWindows: { "deepseek-v4-flash": options.contextWindow } } + : { modelContextWindows: { "deepseek-v4-flash": 1_000_000 } }), + }, + }, + } as OcxConfig; +} + +async function postResponses( + config: OcxConfig, + body: Record, + extraHeaders: Record = {}, +): Promise { + return handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json", ...extraHeaders }, + body: JSON.stringify(body), + }), + config, + { model: "", provider: "" } as RequestLogContext, + ); +} + +async function expectOversizedRejection(res: Response): Promise { + expect(res.status).toBe(413); + expect(await res.json()).toMatchObject({ + error: { + type: "request_too_large", + code: "input_context_window_exceeded", + }, + }); +} + +describe("responses input-size guard", () => { + test("rounds the aggregate estimate instead of every tiny text field", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_tiny_fields", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + + const res = await postResponses(deepseekConfig({ maxInput: 50, contextWindow: 1_000 }), { + model: "deepseek/deepseek-v4-flash", + input: [{ + role: "user", + content: Array.from({ length: 100 }, () => ({ type: "input_text", text: "a" })), + }], + }); + + expect(res.status).toBe(200); + expect(upstreamCalls).toBe(1); + }); + + test("fails open inside the estimator uncertainty band at the model boundary", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_near_boundary", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + + // The heuristic estimates this just over 100 tokens. It is not an exact + // tokenizer result, so a hard refusal at this boundary would be a false-positive risk. + const res = await postResponses(deepseekConfig({ maxInput: 100, contextWindow: 1_000 }), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "a".repeat(351) }] }], + }); + + expect(res.status).toBe(200); + expect(upstreamCalls).toBe(1); + }); + + test("rejects an input above the advertised context window without calling upstream", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // The shared DeepSeek estimator uses 3.5 chars/token, so this is above the 1M window. + const bigText = "a".repeat(4_200_000); + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("rejects input above the per-model maximum-input limit even when below the context window", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // OpenAI-style routes advertise 1,050,000 context but cap input at 922,000. The + // admission guard reserves a 10% estimator uncertainty band, so this fixture sits + // above that hard threshold while remaining below the advertised context window. + const res = await postResponses(deepseekConfig({ maxInput: 922_000, contextWindow: 1_050_000 }), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "a".repeat(3_600_000) }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("rejects an oversized thread-spawn request before any quota polling", async () => { + let upstreamCalls = 0; + let quotaPrimeCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => { + quotaPrimeCalls += 1; + }); + const bigText = "a".repeat(4_200_000); + const res = await postResponses( + deepseekConfig(), + { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + expect(quotaPrimeCalls).toBe(0); + }); + + test("rejects an oversized thread-spawn request before quota priming when a stricter fallback is selectable", async () => { + let upstreamCalls = 0; + let quotaPrimeCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => { + quotaPrimeCalls += 1; + }); + // The primary model fits the estimate (~600k tokens < 1M window), but the selectable + // fallback is stricter (500k max input). Every candidate must be validated before the + // quota probe runs, because priming can change which candidate is selected. + const config = { + port: 0, + defaultProvider: "deepseek", + subagentModelFallback: ["deepseek/deepseek-v4-lite"], + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash", "deepseek-v4-lite"], + modelContextWindows: { + "deepseek-v4-flash": 1_000_000, + "deepseek-v4-lite": 1_000_000, + }, + modelMaxInputTokens: { "deepseek-v4-lite": 500_000 }, + }, + }, + } as OcxConfig; + const res = await postResponses( + config, + { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "a".repeat(2_100_000) }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + expect(quotaPrimeCalls).toBe(0); + }); + + test("skips an unroutable subagent fallback instead of failing the thread-spawn request", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => {}); + // A disabled provider in the fallback chain cannot be routed; the candidate guard + // exists to size admission, not to validate routes, so a stale entry must be + // skipped and the healthy primary request must proceed normally. + const config = { + port: 0, + defaultProvider: "deepseek", + subagentModelFallback: ["ghost/deepseek-v4-lite"], + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + modelContextWindows: { "deepseek-v4-flash": 1_000_000 }, + }, + ghost: { + adapter: "openai-responses", + baseUrl: "https://api.ghost.invalid", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-lite"], + disabled: true, + }, + }, + } as OcxConfig; + const res = await postResponses( + config, + { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "hello" }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + expect(res.status).toBe(200); + expect(upstreamCalls).toBe(1); + }); + + test("rejects an oversized instructions value without calling upstream", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // The parser moves `instructions` into context.systemPrompt, which adapters forward + // upstream; a short message must not hide an oversized prompt from the guard. + const bigInstructions = "a".repeat(4_200_000); + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + instructions: bigInstructions, + input: [{ role: "user", content: [{ type: "input_text", text: "hi" }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("rejects an oversized tool schema without calling upstream", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + tools: [ + { + type: "function", + name: "big_tool", + description: "tool with an oversized schema", + parameters: { + type: "object", + properties: { + payload: { type: "string", description: "a".repeat(4_200_000) }, + }, + }, + }, + ], + input: [{ role: "user", content: [{ type: "input_text", text: "hi" }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("enforces a limit supplied only by provider-wide contextWindow metadata", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // No per-model modelContextWindows/modelMaxInputTokens maps: the guard must resolve + // the provider-wide contextWindow through the shared route-capability chain. + const config = { + port: 0, + defaultProvider: "deepseek", + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + contextWindow: 1_000_000, + }, + }, + } as OcxConfig; + const res = await postResponses(config, { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "a".repeat(4_200_000) }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("does not tokenize an inline image URL as ordinary prompt text", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // The adapter or vision sidecar prices/replaces image content separately. Treating + // base64 bytes as prompt text would reject image requests that fit the model. + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [ + { + role: "user", + content: [ + { type: "input_text", text: "look" }, + { type: "input_image", image_url: `data:image/png;base64,${"a".repeat(4_200_000)}` }, + ], + }, + ], + }); + expect(res.status).toBe(200); + expect(upstreamCalls).toBe(1); + }); + + test("counts the structured-output schema against the window", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + text: { + format: { + type: "json_schema", + name: "big", + schema: { + type: "object", + properties: { payload: { type: "string", description: "a".repeat(4_200_000) } }, + }, + }, + }, + input: [{ role: "user", content: [{ type: "input_text", text: "hi" }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("counts message-level metadata such as Kiro redacted reasoning", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // A krc-only reasoning item attaches the opaque Kiro blob to the preceding assistant + // message; the adapter replays it verbatim, so it must count against the window even + // though it is not part of the message content. + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "ok" }], + }, + { + type: "reasoning", + encrypted_content: encodeReasoningEnvelope({ krc: "a".repeat(4_200_000) }), + }, + ], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("reserves the routed compaction prompt before any upstream call", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // The compaction_trigger item is dropped from messages, but the routed compaction + // prompt is added later. Admission must reserve it before auth or sidecar work. + const LIMIT = 1_000_000; + const hardThreshold = hardAdmissionThreshold(LIMIT); + const compactPromptTokens = estimateTokens(COMPACT_PROMPT, "deepseek-v4-flash"); + expect(compactPromptTokens).toBeGreaterThan(100); + // The body alone is inside the estimator band; adding the prompt crosses the + // conservative hard threshold, so only the final guard rejects it. + const inputTokens = hardThreshold - Math.ceil(compactPromptTokens / 2); + const bigText = "a".repeat(Math.floor((inputTokens - 1) * 3.5) + 1); + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [ + { type: "compaction_trigger" }, + { + role: "user", + content: [{ type: "input_text", text: bigText }], + }, + ], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + + // Positive control: the same near-limit input WITHOUT compaction_trigger is not a + // compaction turn, gets no prompt injection, and reaches upstream normally. + const control = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }); + expect(control.status).toBe(200); + expect(upstreamCalls).toBe(1); + }); + + test("counts a deeply nested tool schema without overflowing the call stack", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // The estimate is small enough to forward, but deep enough that a recursive + // countJsonTokens would build a call frame per level; the frame-based walk must + // keep the request within the window without exhausting the JS stack. + const DEPTH = 30_000; + let parameters: unknown = {}; + for (let i = 0; i < DEPTH; i++) { + parameters = { nested: parameters }; + } + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + tools: [ + { + type: "function", + name: "deep_tool", + description: "deeply nested schema", + parameters, + }, + ], + input: [{ role: "user", content: [{ type: "input_text", text: "hi" }] }], + }); + expect(upstreamCalls).toBe(1); + expect(res.status).toBe(200); + }); + + test("forwards an input within the window", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "hello" }] }], + }); + expect(upstreamCalls).toBe(1); + expect(res.status).toBe(200); + }); + + test("counts deterministic v1 guidance against a near-limit input", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + const LIMIT = 1_000_000; + const hardThreshold = hardAdmissionThreshold(LIMIT); + const modelId = "deepseek-v4-flash"; + const guidanceText = `${PROACTIVE_MULTI_AGENT_MODE_TEXT}`; + const guidanceTokens = estimateTokens(guidanceText, modelId); + // The body alone is inside the estimator band; deterministic guidance pushes the + // complete request across the conservative hard threshold. + const inputTokens = hardThreshold - Math.ceil(guidanceTokens / 2); + const bigText = "a".repeat(Math.floor((inputTokens - 1) * 3.5) + 1); + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + reasoning: { effort: "max" }, + tools: [ + { type: "function", name: "spawn_agent", description: "" }, + { type: "function", name: "send_input", description: "" }, + ], + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + }); + + test("counts a configured injectionPrompt before any thread-spawn quota polling", async () => { + let upstreamCalls = 0; + let quotaPrimeCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => { + quotaPrimeCalls += 1; + }); + const LIMIT = 1_000_000; + const hardThreshold = hardAdmissionThreshold(LIMIT); + const modelId = "deepseek-v4-flash"; + const prompt = "a".repeat(500); + const floorText = `${applyInjectionPlaceholders(prompt, "", "", "", "")}`; + const floorTokens = estimateTokens(floorText, modelId); + // Under the hard threshold without the prompt floor, over it once the prompt is + // counted; the rejection must precede the quota probe. + const inputTokens = hardThreshold - floorTokens + 20; + const bigText = "a".repeat(Math.floor((inputTokens - 1) * 3.5) + 1); + const config = { + port: 0, + defaultProvider: "deepseek", + injectionPrompt: prompt, + injectionModel: "deepseek/deepseek-v4-lite", + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + modelContextWindows: { "deepseek-v4-flash": LIMIT }, + }, + }, + } as OcxConfig; + const res = await postResponses( + config, + { + model: "deepseek/deepseek-v4-flash", + tools: [{ type: "function", name: "spawn_agent", description: "" }], + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + expect(quotaPrimeCalls).toBe(0); + }); + + test("counts resolved roster guidance before any thread-spawn quota polling", async () => { + let upstreamCalls = 0; + let quotaPrimeCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => { + quotaPrimeCalls += 1; + }); + const LIMIT = 1_000_000; + const hardThreshold = hardAdmissionThreshold(LIMIT); + const modelId = "deepseek-v4-flash"; + const subagentModel = "deepseek/deepseek-v4-lite"; + const prompt = `${"a".repeat(100)} {{roster}}`; + // The pre-quota estimate resolves {{roster}}/{{model}} from the configured lists; + // the placeholder-free floor would be far below the limit and let a doomed request + // reach the quota probe. + const resolvedText = `${applyInjectionPlaceholders( + prompt, + subagentModel, + "", + subagentRosterText([{ model: subagentModel, efforts: [] }]), + "", + )}`; + const floorText = `${applyInjectionPlaceholders(prompt, "", "", "", "")}`; + const resolvedTokens = estimateTokens(resolvedText, modelId); + const floorTokens = estimateTokens(floorText, modelId); + const toolTokens = estimateTokens("spawn_agent", modelId); + // input + tool + floor stays below the hard threshold, while resolving the roster + // crosses it and must reject before quota polling. + const inputTokens = hardThreshold - Math.ceil((floorTokens + resolvedTokens + toolTokens) / 2); + expect(inputTokens + toolTokens + floorTokens).toBeLessThan(hardThreshold); + expect(inputTokens + toolTokens + resolvedTokens).toBeGreaterThan(hardThreshold); + const bigText = "a".repeat(Math.floor((inputTokens - 1) * 3.5) + 1); + const config = { + port: 0, + defaultProvider: "deepseek", + injectionPrompt: prompt, + subagentModels: [subagentModel], + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + modelContextWindows: { "deepseek-v4-flash": LIMIT }, + }, + }, + } as OcxConfig; + const res = await postResponses( + config, + { + model: "deepseek/deepseek-v4-flash", + tools: [{ type: "function", name: "spawn_agent", description: "" }], + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + expect(quotaPrimeCalls).toBe(0); + }); + + test("reserves bounded default v2 guidance before quota polling", async () => { + let upstreamCalls = 0; + let quotaPrimeCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + setSubagentQuotaPrimeForTests(async () => { + quotaPrimeCalls += 1; + }); + const LIMIT = 1_000_000; + const hardThreshold = hardAdmissionThreshold(LIMIT); + // No injectionPrompt on v2 means the pre-quota estimate counts NO guidance, but + // normalization still injects the default v2 guidance plus the configured fallback + // chain text. Only the post-normalization re-validation can catch this request + // (before any auth or upstream I/O). + const inputTokens = hardThreshold - 100; + const bigText = "a".repeat(Math.floor((inputTokens - 1) * 3.5) + 1); + const previousOverride = process.env.OPENCODEX_APP_SERVER_CATALOG_STATE_OVERRIDE; + process.env.OPENCODEX_APP_SERVER_CATALOG_STATE_OVERRIDE = "fresh"; + try { + const config = { + port: 0, + defaultProvider: "deepseek", + subagentModelFallback: ["deepseek/deepseek-v4-lite"], + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + modelContextWindows: { "deepseek-v4-flash": LIMIT }, + }, + }, + } as OcxConfig; + const res = await postResponses( + config, + { + model: "deepseek/deepseek-v4-flash", + tools: [{ type: "function", name: "spawn_agent", description: "" }], + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }, + { "x-openai-subagent": "collab_spawn" }, + ); + await expectOversizedRejection(res); + expect(upstreamCalls).toBe(0); + expect(quotaPrimeCalls).toBe(0); + } finally { + if (previousOverride === undefined) delete process.env.OPENCODEX_APP_SERVER_CATALOG_STATE_OVERRIDE; + else process.env.OPENCODEX_APP_SERVER_CATALOG_STATE_OVERRIDE = previousOverride; + } + }); +}); diff --git a/tests/responses-replay-overlap.test.ts b/tests/responses-replay-overlap.test.ts new file mode 100644 index 0000000000..2c4448738e --- /dev/null +++ b/tests/responses-replay-overlap.test.ts @@ -0,0 +1,425 @@ +/** + * Regression coverage for previous_response_id expansion overlapping a request + * that already carries the full conversation (stateless upstreams such as + * DeepSeek force the client to resend full history every turn). The old + * unconditional prepend compounded the stored history each turn: 1x -> 2x -> + * 3x -> ... (observed 1,333,682 input tokens, ~10x the real ~127k conversation, + * on 2026-08-10). Full-body chained turns must stay 1x, while genuine delta + * turns must still expand to the stored history plus their delta. + */ +import { afterEach, beforeEach, describe, expect, setDefaultTimeout, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + clearResponseStateForTests, + expandPreviousResponseInput, + rememberResponseState, +} from "../src/responses/state"; +import { handleResponses } from "../src/server/responses"; +import type { OcxConfig } from "../src/types"; +import type { RequestLogContext } from "../src/server/request-log"; + +setDefaultTimeout(30_000); + +const originalFetch = globalThis.fetch; +let testDir: string; +let previousOpencodexHome: string | undefined; +let previousCodexHome: string | undefined; + +beforeEach(() => { + testDir = mkdtempSync(join(tmpdir(), "ocx-replay-overlap-")); + previousOpencodexHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = testDir; + process.env.CODEX_HOME = testDir; + clearResponseStateForTests(); +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + clearResponseStateForTests(); + rmSync(testDir, { recursive: true, force: true }); + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; +}); + +const userItem = (text: string): Record => ({ + type: "message", + role: "user", + content: [{ type: "input_text", text }], +}); + +const assistantInputItem = (text: string, id?: string): Record => ({ + type: "message", + role: "assistant", + ...(id ? { id } : {}), + content: [{ type: "output_text", text }], +}); + +const assistantOutputItem = (text: string): Record => ({ + type: "message", + role: "assistant", + id: `msg_${text}`, + status: "completed", + content: [{ type: "output_text", text }], +}); + +const replayAnchor = [ + { type: "function_call", name: "anchor", call_id: "call_anchor", arguments: "{}" }, + { type: "function_call_output", call_id: "call_anchor", output: "ok" }, +]; + +const MODEL = "deepseek/deepseek-v4-flash"; + +describe("previous_response_id replay overlap", () => { + test("a legitimate delta may repeat the complete stored message prefix", () => { + const repeated = { ...userItem("repeat"), id: "client_msg_repeat" }; + rememberResponseState( + { model: MODEL, input: [repeated] }, + { id: "resp_repeat", status: "completed", output: [] }, + undefined, + { force: true }, + ); + + const request = [{ ...userItem("repeat"), id: "client_msg_repeat" }, userItem("new")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_repeat", + input: request, + }); + + expect(expanded.input).toEqual([repeated, ...request]); + }); + + test("a request-owned call_id cannot prove that repeated content is replay", () => { + const repeated = { + type: "function_call_output", + call_id: "call_client_owned", + output: "same result", + }; + rememberResponseState( + { model: MODEL, input: [repeated] }, + { id: "resp_client_call_id", status: "completed", output: [] }, + undefined, + { force: true }, + ); + + const request = [repeated, userItem("new")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_client_call_id", + input: request, + }); + + expect(expanded.input).toEqual([repeated, ...request]); + }); + + test("full-history chained turns stay 1x across four turns", () => { + const base = [...Array.from({ length: 20 }, (_, i) => userItem(`base ${i}`)), ...replayAnchor]; + const conversation = [...base, userItem("turn 1")]; + let respId = "resp_0"; + rememberResponseState( + { model: MODEL, input: conversation }, + { id: respId, status: "completed", output: [assistantOutputItem("a1")] }, + undefined, + { force: true }, + ); + conversation.push(assistantInputItem("a1", "msg_a1")); + + for (let i = 2; i <= 5; i++) { + conversation.push(userItem(`turn ${i}`)); + const next = { model: MODEL, previous_response_id: respId, input: [...conversation] }; + const expanded = expandPreviousResponseInput(next); + expect((expanded.input as unknown[]).length).toBe(conversation.length); + expect(expanded.input).toEqual(conversation); + respId = `resp_${i}`; + rememberResponseState( + expanded, + { id: respId, status: "completed", output: [assistantOutputItem(`a${i}`)] }, + undefined, + { force: true }, + ); + conversation.push(assistantInputItem(`a${i}`, `msg_a${i}`)); + } + }); + + test("a genuine delta turn still expands to stored history plus its delta", () => { + const base = Array.from({ length: 20 }, (_, i) => userItem(`base ${i}`)); + rememberResponseState( + { model: MODEL, input: base }, + { id: "resp_delta", status: "completed", output: [assistantOutputItem("a1")] }, + undefined, + { force: true }, + ); + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_delta", + input: [userItem("delta")], + }); + expect((expanded.input as unknown[]).length).toBe(base.length + 2); + expect((expanded.input as unknown[]).slice(0, base.length)).toEqual(base); + expect((expanded.input as unknown[]).at(-1)).toEqual(userItem("delta")); + }); + + test("stored output-shaped items match when the client retains provider identity", () => { + rememberResponseState( + { model: MODEL, input: [userItem("hello")] }, + { id: "resp_shape", status: "completed", output: [assistantOutputItem("hi")] }, + undefined, + { force: true }, + ); + // Status is transport metadata, while the retained provider id proves this is a replay. + const full = [userItem("hello"), { ...assistantInputItem("hi"), id: "msg_hi" }, userItem("next")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_shape", + input: full, + }); + expect(expanded.input).toEqual(full); + }); + + test("canonical keys ignore retained property order", () => { + const storedItem = { + type: "message", + role: "assistant", + id: "msg_x", + status: "completed", + content: [{ type: "output_text", text: "hi" }], + }; + const resendItem = { + role: "assistant", + id: "msg_x", + content: [{ text: "hi", type: "output_text" }], + type: "message", + }; + rememberResponseState( + { model: MODEL, input: [userItem("hello")] }, + { id: "resp_order", status: "completed", output: [storedItem] }, + undefined, + { force: true }, + ); + const full = [userItem("hello"), resendItem, userItem("next")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_order", + input: full, + }); + expect(expanded.input).toEqual(full); + }); + + test("web_search_call query/queries backfill skew still canonical-matches", () => { + // #930: history recorded before the bridge emitted both keys carries only + // `action.query`; the replay-boundary repair (backfillWebSearchQueries) adds + // `action.queries` on the outbound body. A stored item and the client resend + // therefore differ by exactly that derived field and must still count as the + // same history item — otherwise the overlap breaks and the stored history is + // prepended again, doubling the request on web-search turns. + const storedSearch: Record = { + type: "web_search_call", + id: "ws_stored", + status: "completed", + action: { type: "search", query: "opencodex context bug" }, + }; + const resendSearch: Record = { + type: "web_search_call", + id: "ws_stored", + action: { + type: "search", + query: "opencodex context bug", + queries: ["opencodex context bug"], + }, + }; + rememberResponseState( + { model: MODEL, input: [userItem("hello")] }, + { id: "resp_websearch", status: "completed", output: [storedSearch] }, + undefined, + { force: true }, + ); + const full = [userItem("hello"), resendSearch, userItem("next")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_websearch", + input: full, + }); + expect(expanded.input).toEqual(full); + }); + + test("partial prefix keeps stored history and never drops request items", () => { + const stored = [userItem("u1"), assistantInputItem("a1"), userItem("u2"), assistantInputItem("a2")]; + rememberResponseState( + { model: MODEL, input: stored }, + { id: "resp_partial", status: "completed", output: [] }, + undefined, + { force: true }, + ); + const request = [userItem("u1"), userItem("X"), userItem("D")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_partial", + input: request, + }); + const input = expanded.input as unknown[]; + expect(input.slice(0, stored.length)).toEqual(stored); + // The request's own leading item is a NEW occurrence (it can be a repeated marker or an + // identical message), so it must survive even though it also matches the stored prefix. + expect(input.slice(stored.length)).toEqual([userItem("u1"), userItem("X"), userItem("D")]); + }); + + test("a delta as long as the stored history still expands to stored plus every delta item", () => { + const stored = [userItem("u1"), assistantInputItem("a1"), userItem("u2")]; + rememberResponseState( + { model: MODEL, input: stored }, + { id: "resp_long_delta", status: "completed", output: [] }, + undefined, + { force: true }, + ); + // Four new items >= stored's three: request length must NOT be treated as a full resend. + const request = [userItem("n1"), userItem("n2"), userItem("n3"), userItem("n4")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_long_delta", + input: request, + }); + const input = expanded.input as unknown[]; + expect(input.slice(0, stored.length)).toEqual(stored); + expect(input.slice(stored.length)).toEqual(request); + }); + + test("a deeply nested chained input never throws and stays conservative", () => { + // Nest past the canonical depth budget (256 levels): the item must not produce a + // canonical key, so overlap detection degrades to "no overlap" instead of throwing + // or letting a deep resend silently match a shallow stored item. + const DEPTH = 300; + let deepItem: unknown = { + type: "message", + role: "user", + content: [{ type: "input_text", text: "deep" }], + }; + for (let i = 0; i < DEPTH; i++) deepItem = { wrapper: { inner: deepItem } }; + const stored = [deepItem, assistantInputItem("a1")]; + rememberResponseState( + { model: MODEL, input: stored }, + { id: "resp_deep", status: "completed", output: [] }, + undefined, + { force: true }, + ); + const request = [deepItem, userItem("request")]; + const expanded = expandPreviousResponseInput({ + model: MODEL, + previous_response_id: "resp_deep", + input: request, + }); + const input = expanded.input as unknown[]; + // Overlap evidence is unavailable for the deep leading item: expansion must preserve + // the stored history followed by the ENTIRE request input, never dropping items. + expect(input.slice(0, stored.length)).toEqual(stored); + expect(input.slice(stored.length)).toEqual(request); + }); +}); + +function statelessDeepseekConfig(): OcxConfig { + return { + port: 0, + defaultProvider: "deepseek", + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + statelessResponses: true, + modelContextWindows: { "deepseek-v4-flash": 1_000_000 }, + }, + }, + } as OcxConfig; +} + +async function postResponses(config: OcxConfig, body: Record): Promise { + return handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + config, + { model: "", provider: "" } as RequestLogContext, + ); +} + +describe("stateless DeepSeek end-to-end replay", () => { + test("four full-history chained turns reach upstream at 1x every time", async () => { + const upstreamBodies: unknown[][] = []; + let nextId = 1; + globalThis.fetch = (async (_url: unknown, init?: { body?: string }) => { + const body = JSON.parse(String(init?.body ?? "{}")) as { input?: unknown[] }; + upstreamBodies.push(body.input ?? []); + const n = nextId++; + return Response.json({ + id: `resp_${n}`, + object: "response", + status: "completed", + model: "deepseek-v4-flash", + output: [assistantOutputItem(`a${n}`)], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + + const config = statelessDeepseekConfig(); + const conversation = [ + ...Array.from({ length: 20 }, (_, i) => userItem(`base ${i}`)), + ...replayAnchor, + userItem("turn 1"), + ]; + let respId = ""; + for (let i = 1; i <= 4; i++) { + if (i > 1) conversation.push(userItem(`turn ${i}`)); + const res = await postResponses(config, { + model: MODEL, + ...(respId ? { previous_response_id: respId } : {}), + input: [...conversation], + }); + expect(res.status).toBe(200); + expect(upstreamBodies.at(-1)?.length).toBe(conversation.length); + conversation.push(assistantInputItem(`a${i}`, `msg_a${i}`)); + respId = `resp_${i}`; + } + expect(upstreamBodies).toHaveLength(4); + }); + + test("a delta continuation still expands to the full conversation upstream", async () => { + const upstreamBodies: unknown[][] = []; + let nextId = 1; + globalThis.fetch = (async (_url: unknown, init?: { body?: string }) => { + const body = JSON.parse(String(init?.body ?? "{}")) as { input?: unknown[] }; + upstreamBodies.push(body.input ?? []); + const n = nextId++; + return Response.json({ + id: `resp_${n}`, + object: "response", + status: "completed", + model: "deepseek-v4-flash", + output: [assistantOutputItem(`a${n}`)], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + + const config = statelessDeepseekConfig(); + const conversation = Array.from({ length: 20 }, (_, i) => userItem(`base ${i}`)); + await postResponses(config, { model: MODEL, input: [...conversation] }); + expect(upstreamBodies[0]!.length).toBe(conversation.length); + conversation.push(assistantInputItem("a1")); + + const res = await postResponses(config, { + model: MODEL, + previous_response_id: "resp_1", + input: [userItem("delta")], + }); + expect(res.status).toBe(200); + expect(upstreamBodies[1]!.length).toBe(conversation.length + 1); + }); +}); diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index 7bd3178a72..06705fe496 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -719,6 +719,35 @@ describe("Responses previous_response_id state", () => { expect(responseStateMetrics().spillStubCount).toBe(1); }); + test("durable spill retains the provider-output replay boundary", async () => { + setResponseStateByteCapForTests(1_024); + const user = { role: "user", content: "r".repeat(8_000) }; + const assistantOutput = { + type: "message", + role: "assistant", + id: "msg_spill_replay", + status: "completed", + content: [{ type: "output_text", text: "done" }], + }; + rememberResponseState( + { input: [user] }, + fixedResponse("resp_spill_replay_boundary", [assistantOutput]), + ); + await flushResponseState(); + clearResponseStateMemoryForTests(); + setResponseStateByteCapForTests(1_024); + + const assistantInput: Record = { ...assistantOutput }; + delete assistantInput.status; + const full = [user, assistantInput, { role: "user", content: "next" }]; + const expanded = expandPreviousResponseInput({ + previous_response_id: "resp_spill_replay_boundary", + input: full, + }) as { input: unknown[] }; + + expect(expanded.input).toEqual(full); + }); + test("spill references bind the expected response id and use the locked digest basename", () => { const ref = writeResponseSpillDurably("resp_identity", { createdAt: Date.now(), @@ -986,7 +1015,7 @@ describe("Responses previous_response_id state", () => { ); const items = [{ role: "user", content: "한글🙂" }, ...output]; const expected = Buffer.byteLength(JSON.stringify({ - responseId: "resp_다국어", createdAt: at, items, providers, + responseId: "resp_다국어", createdAt: at, items, providerOutputStart: 1, providers, }), "utf8"); expect(getStoredResponseBytesForTests()).toBe(expected); } finally { diff --git a/tests/terminal-guard-server.test.ts b/tests/terminal-guard-server.test.ts index ba596cb808..95fdb003a0 100644 --- a/tests/terminal-guard-server.test.ts +++ b/tests/terminal-guard-server.test.ts @@ -79,6 +79,36 @@ describe("server terminal guard integration", () => { expect(messages.at(-1)?.content?.[0]?.text).toContain("你刚才只描述了计划"); }); + test("revalidates the terminal-guard continuation before its upstream send", async () => { + const boundedConfig = { + ...config, + providers: { + "claude-se": { + adapter: "anthropic", + baseUrl: "https://example.test", + apiKey: "sk-test", + modelMaxInputTokens: { "se-claude-opus-4.8": 60 }, + }, + }, + } as unknown as OcxConfig; + + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "se-claude-opus-4.8", + input: "修复", + stream: true, + tools: [{ type: "function", name: "exec_command", parameters: {} }], + }), + }), boundedConfig, { model: "", provider: "" }); + + const text = await response.text(); + expect(response.status).toBe(200); + expect(calls).toBe(1); + expect(text).toContain("input_context_window_exceeded"); + }); + test("terminal-guard continuation 429 replays on the same key before surfacing", async () => { const retryConfig = { ...config,