diff --git a/docs-site/src/content/docs/fr/reference/configuration/server.md b/docs-site/src/content/docs/fr/reference/configuration/server.md index d957d9b50f..cde1b2c5f6 100644 --- a/docs-site/src/content/docs/fr/reference/configuration/server.md +++ b/docs-site/src/content/docs/fr/reference/configuration/server.md @@ -14,7 +14,7 @@ exécute des fonctionnalités d'assistance autour des demandes du fournisseur. | `hostname?` | `string` | `"127.0.0.1"` | Adresse de liaison. Les liaisons hors bouclage nécessitent `OPENCODEX_API_AUTH_TOKEN`. | | `proxy?` | `string` | — | URL du proxy HTTP(S) sortant ou `${ENV_VAR}`. Appliquée à `HTTP_PROXY` / `HTTPS_PROXY` uniquement lorsque ces variables ne sont pas définies ; le bouclage reste dans `NO_PROXY`. | | `emptyCompletionRetry?` | `boolean` | `false` | Active une nouvelle tentative Responses identique lorsqu’une réponse ne contient ni texte ni appel d’outil. Cette tentative peut être facturée. `OCX_EMPTY_COMPLETION_RETRY=0` la désactive sans modifier la configuration ; les combinaisons et les tours de compactage routés restent exclus. | -| `stallTimeoutSec?` | `number` | `300` | Nombre de secondes sans données en amont avant `response.incomplete`. Minimum : 1. | +| `stallTimeoutSec?` | `number` | `300` | Secondes sans progression utile en amont, pour Responses et le Chat natif. Minimum : 1. | | `connectTimeoutMs?` | `number` | `200000` | Délai maximal par tentative pour DNS/TCP/TLS et les en-têtes finaux ; il prend fin avant la génération du corps. | | `shutdownTimeoutMs?` | `number` | `5000` | Délai de vidange gracieux avant l’annulation des tours actifs. | | `websockets?` | `boolean` | `false` | Annonce et autorise la route WebSocket Responses destinée aux clients. La valeur false maintient les clients sur HTTP/SSE ; elle ne désactive pas une optimisation WebSocket canonique admissible vers ChatGPT en amont. | @@ -34,6 +34,10 @@ Si une ancienne version de développement a modifié les métadonnées de l'hist `ocx recover-history --legacy-openai --yes` pour forcer la récupération du fournisseur natif. La commande réétiquette chaque ligne `opencodex` contenant un message utilisateur, y compris l'historique légitime d'un fournisseur dédié ; consultez l'avertissement sur la portée complète dans la référence du cycle de vie avant de l'exécuter. +### Délais et fin de réponse du Chat natif + +Le Chat natif utilise aussi `stallTimeoutSec` pendant l’attente de la sortie amont. Le texte non vide, le raisonnement, le refus, les mises à jour d’outils et les événements de fin renouvellent ce délai ; les commentaires de maintien de connexion, le rôle seul et les statistiques seules ne le renouvellent pas. L’attente d’un client lent suspend le décompte. Un blocage produit `upstream_stall_timeout` : un événement d’erreur en streaming, ou HTTP 502 sans streaming. Une annulation avant le résultat terminal renvoie une erreur d’annulation plutôt qu’une réponse partielle réussie. Le Chat sans streaming accepte les délimiteurs SSE LF et CRLF et les champs data multilignes. + ## Accès à distance La liaison par défaut à `127.0.0.1` est limitée au bouclage. Une adresse hors bouclage telle que `0.0.0.0` diff --git a/docs-site/src/content/docs/ja/reference/configuration/server.md b/docs-site/src/content/docs/ja/reference/configuration/server.md index 0dd9cf59e4..d33946ab30 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/server.md +++ b/docs-site/src/content/docs/ja/reference/configuration/server.md @@ -13,7 +13,7 @@ description: リスナー、リモート アクセス、アドミッション | `hostname?` | `string` | `"127.0.0.1"` |バインドアドレス。非ループバック バインドには `OPENCODEX_API_AUTH_TOKEN` が必要です。 | | `proxy?` | `string` | — |送信 HTTP(S) プロキシ URL または `${ENV_VAR}`。これらの変数が設定されていない場合にのみ、`HTTP_PROXY` / `HTTPS_PROXY` に適用されます。ループバックは `NO_PROXY` に残ります。 | | `emptyCompletionRetry?` | `boolean` | `false` | テキストもツール呼び出しもない Responses ターンを、ターミナルイベント前にストリームが終了した場合も含め、同一リクエストで 1 回再試行するよう明示的に有効化します。再試行は課金対象になる場合があります。`OCX_EMPTY_COMPLETION_RETRY=0` で設定を変更せず無効化できます。combo と routed-compaction turn は対象外です。 | -| `stallTimeoutSec?` | `number` | `300` | `response.incomplete` より前にアップストリーム データがない秒数。最小 1。 +| `stallTimeoutSec?` | `number` | `300` | Responses とネイティブ Chat の有効な上流進捗がない秒数。最小 1 秒。 | | `connectTimeoutMs?` | `number` | `200000` |試行ごとの DNS/TCP/TLS/最終ヘッダーの期限。本体が生成される前に終了します。 | | `shutdownTimeoutMs?` | `number` | `5000` |アクティブなターンが中止される前の正常な排出期限。 | | `websockets?` | `boolean` | `false` | クライアント向け Responses WebSocket パスを広告して許可します。false の場合クライアントは HTTP/SSE を使いますが、対象となる canonical ChatGPT upstream WS 最適化は無効にしません。 | @@ -32,6 +32,10 @@ description: リスナー、リモート アクセス、アドミッション バックアップ サポートが存在する前に古い開発ビルドで再開履歴メタデータが変更された場合は、`ocx recover-history --legacy-openai --yes` を実行してネイティブ プロバイダーの回復を強制します。 このコマンドは、正当な専用プロバイダー履歴を含む、ユーザーメッセージを持つすべての `opencodex` 行を再ラベル付けします。実行前にライフサイクル リファレンスの全範囲に関する警告を確認してください。 +### ネイティブ Chat のタイムアウトと完了 + +ネイティブ Chat も上流出力の待機に `stallTimeoutSec` を使用します。空でないテキスト、推論、拒否内容、ツール更新、完了イベントは待機時間を更新しますが、キープアライブのコメント、ロールのみのイベント、使用量のみのイベントは更新しません。低速クライアントの読み取り待ちは計時を停止します。タイムアウト時は `upstream_stall_timeout` が返り、ストリーミングではエラーイベント、非ストリーミングでは HTTP 502 になります。終端結果より前のキャンセルは成功した部分回答ではなくキャンセルエラーになります。非ストリーミング Chat は LF、CRLF、複数行 data の SSE に対応します。 + ## リモートアクセス デフォルトの `127.0.0.1` バインドはループバックのみです。 `0.0.0.0` などの非ループバック アドレスには、`/api/*` とデータ プレーンの両方でトークン認証が必要です。開始する前にトークンをエクスポートします。 diff --git a/docs-site/src/content/docs/ko/reference/configuration/server.md b/docs-site/src/content/docs/ko/reference/configuration/server.md index 577379fed9..ff51305c1f 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/server.md +++ b/docs-site/src/content/docs/ko/reference/configuration/server.md @@ -13,7 +13,7 @@ description: 리스너, 원격 접근, admission 키, 타임아웃, 저장소, | `hostname?` | `string` | `"127.0.0.1"` | 바인드 주소입니다. 루프백이 아닌 바인드에는 데이터 admission 토큰이 필요하며, `OPENCODEX_API_AUTH_TOKEN` → `OCX_API_TOKEN_FILE` → 설치된 owner-only `service-api-token` 순서로 결정됩니다. 손으로 내보낼 값은 없습니다. [Remote access](#remote-access)를 보세요. | | `proxy?` | `string` | — | 송신용 HTTP(S) 프록시 URL 또는 `${ENV_VAR}`입니다. 해당 변수가 비어 있을 때만 `HTTP_PROXY` / `HTTPS_PROXY`에 적용되며, 루프백은 `NO_PROXY`에 그대로 남습니다. | | `emptyCompletionRetry?` | `boolean` | `false` | 텍스트나 도구 호출이 없는 Responses 턴을, 터미널 이벤트 전에 스트림이 종료된 경우를 포함해 동일한 요청으로 한 번 재시도하도록 선택합니다. 재시도에는 비용이 발생할 수 있습니다. `OCX_EMPTY_COMPLETION_RETRY=0`은 설정을 바꾸지 않고 비활성화하며, combo 및 routed-compaction turn은 제외됩니다. | -| `stallTimeoutSec?` | `number` | `300` | 업스트림 데이터가 없을 때 `response.incomplete`가 되기까지의 초 수입니다. 최소 1입니다. | +| `stallTimeoutSec?` | `number` | `300` | Responses 및 네이티브 Chat에서 유효한 업스트림 진행이 없는 시간(초). 최소 1초. | | `connectTimeoutMs?` | `number` | `200000` | 시도별 DNS/TCP/TLS/최종 헤더 기한입니다. 본문 생성 전에 끝납니다. | | `shutdownTimeoutMs?` | `number` | `5000` | 진행 중인 turn을 중단하기 전에 허용하는 정상 종료 드레인 기한입니다. | | `websockets?` | `boolean` | `false` | 클라이언트용 Responses WebSocket 경로를 광고하고 허용합니다. `false`이면 클라이언트는 HTTP/SSE를 사용하며, 적격 canonical ChatGPT 업스트림 WS 최적화는 비활성화하지 않습니다. | @@ -32,6 +32,10 @@ description: 리스너, 원격 접근, admission 키, 타임아웃, 저장소, 오래된 개발 빌드가 백업 지원이 생기기 전에 resume-history 메타데이터를 바꿨다면, native-provider 복구를 강제로 수행하려면 `ocx recover-history --legacy-openai --yes`를 실행합니다. 이 명령은 정상적인 dedicated-provider history를 포함해 사용자 메시지가 있는 모든 `opencodex` row를 재태깅합니다. 실행하기 전에 lifecycle reference의 전체 범위 경고를 확인하세요. +### 네이티브 Chat 시간 초과와 완료 + +네이티브 Chat도 업스트림 출력을 기다릴 때 `stallTimeoutSec`를 사용합니다. 비어 있지 않은 텍스트, 추론, 거부 내용, 도구 업데이트 및 완료 이벤트는 대기 시간을 갱신하지만 연결 유지 주석, 역할만 있는 이벤트, 사용량만 있는 이벤트는 갱신하지 않습니다. 느린 클라이언트의 읽기를 기다리는 동안에는 시간이 차감되지 않습니다. 시간 초과 시 `upstream_stall_timeout`이 발생하며 스트리밍 요청은 오류 이벤트를, 비스트리밍 요청은 HTTP 502를 받습니다. 종료 결과 전에 취소하면 부분 답변을 성공으로 반환하지 않고 취소 오류를 반환합니다. 비스트리밍 Chat은 LF, CRLF 및 여러 줄 data SSE 형식을 지원합니다. + ## Remote access 기본 `127.0.0.1` 바인드는 루프백 전용입니다. `0.0.0.0`이나 tailnet IP처럼 루프백이 아닌 주소는 `/api/*`와 데이터 플레인 모두에서 토큰 인증이 필요합니다. diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 108d2e0925..a1d65a3d7f 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -15,7 +15,7 @@ runs helper features around provider requests. | `proxy?` | `string` | — | Outbound HTTP(S) proxy URL, `${ENV_VAR}`, or `"auto"`. Applied to `HTTP_PROXY` / `HTTPS_PROXY` only when those variables are unset; loopback remains in `NO_PROXY`. `"auto"` reads the Windows system proxy (WinINET `ProxyEnable`/`ProxyServer`, `https=` then `http=` entry) once at process start and logs the host it chose. On other platforms, or when the system proxy is off, SOCKS-only, or unreadable, it uses direct egress and says so. PAC/WPAD and live proxy changes are not followed; restart the service after changing the system proxy. | | `noProxy?` | `string \| string[]` | — | Hosts that bypass `proxy`, merged with inherited `NO_PROXY` and loopback entries. A string may use comma-separated `NO_PROXY` syntax or `${ENV_VAR}`. | | `emptyCompletionRetry?` | `boolean` | `false` | Opt in to one identical Responses retry when a turn has no text or tool call, including a stream that ends before a terminal event. The retry may be billable. `OCX_EMPTY_COMPLETION_RETRY=0` disables it without changing config; combo and routed-compaction turns remain excluded. | -| `stallTimeoutSec?` | `number` | `300` | Seconds without upstream data before `response.incomplete`. Minimum 1. | +| `stallTimeoutSec?` | `number` | `300` | Seconds without meaningful upstream progress (Responses and native Chat). Minimum 1. | | `oauthOpenBrowser?` | `boolean` | `true` | Whether a login may open a browser on the machine running the proxy. Absent and `true` both open, so an existing install is unchanged; only an explicit `false` declines. Decline when you need the authorization link in a different browser profile, or when the dashboard is not on the proxy's machine — the login still starts and the URL is still returned and displayed. `POST /api/oauth/login` and `POST /api/codex-auth/login` accept a per-request `openBrowser` boolean that overrides this, and the dashboard exposes the same choice beside the login button. Device-code flows never open a browser either way. | | `connectTimeoutMs?` | `number` | `200000` | Per-attempt DNS/TCP/TLS/final-header deadline; it ends before body generation. | | `shutdownTimeoutMs?` | `number` | `5000` | Graceful drain deadline before active turns are aborted. | @@ -65,6 +65,10 @@ If an older development build changed resume-history metadata before backup supp It force-relabels every user-message `opencodex` row, including legitimate dedicated-provider history; review the full-scope warning in the lifecycle reference before running it. +### Native Chat timeouts and completion + +Native Chat also uses `stallTimeoutSec` while waiting for upstream output. Nonempty text, reasoning, refusal, tool updates, and finish frames renew the allowance; keepalive comments, role-only frames, and usage alone do not. Waiting for a slow client to read pauses the allowance. A stall produces `upstream_stall_timeout`: an error frame for streaming clients, or HTTP 502 for non-streaming clients. Cancellation before a terminal result returns a cancellation error instead of a successful partial answer. Buffered Chat results accept both LF and CRLF SSE framing, including multiline data. + ## Codex quota network diagnostics The main Codex account row may include `quotaRefresh` when a quota fetch was diff --git a/docs-site/src/content/docs/ru/reference/configuration/server.md b/docs-site/src/content/docs/ru/reference/configuration/server.md index 84b3caf183..73e975ce30 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/server.md +++ b/docs-site/src/content/docs/ru/reference/configuration/server.md @@ -14,7 +14,7 @@ description: Listener, удалённый доступ, admission key, тайм | `hostname?` | `string` | `"127.0.0.1"` | Адрес bind'а. Не-loopback bind требует `OPENCODEX_API_AUTH_TOKEN`. | | `proxy?` | `string` | — | URL исходящего HTTP(S)-прокси или `${ENV_VAR}`. Применяется к `HTTP_PROXY` / `HTTPS_PROXY` только когда эти переменные не заданы; loopback всегда остаётся в `NO_PROXY`. | | `emptyCompletionRetry?` | `boolean` | `false` | Явно включает один идентичный повтор Responses, если в turn нет ни текста, ни tool call, включая случай, когда stream завершается до terminal event. Повтор может тарифицироваться. `OCX_EMPTY_COMPLETION_RETRY=0` отключает его без изменения config; combo и routed-compaction turn исключены. | -| `stallTimeoutSec?` | `number` | `300` | Секунды без upstream-данных до `response.incomplete`. Минимум 1. | +| `stallTimeoutSec?` | `number` | `300` | Секунды без полезного прогресса upstream для Responses и нативного Chat. Минимум 1. | | `connectTimeoutMs?` | `number` | `200000` | Дедлайн одной попытки DNS/TCP/TLS/final-header; он завершается до генерации тела ответа. | | `shutdownTimeoutMs?` | `number` | `5000` | Дедлайн graceful-drain до принудительного прерывания активных turn'ов. | | `websockets?` | `boolean` | `false` | Объявляет и разрешает клиентский WebSocket-путь Responses. При false клиенты используют HTTP/SSE; это не отключает подходящую upstream WS-оптимизацию canonical ChatGPT. | @@ -35,6 +35,10 @@ backup'а, выполните `ocx recover-history --legacy-openai --yes`, чт native-provider history. Команда переименовывает все строки `opencodex` с пользовательским сообщением, включая корректную историю выделенного провайдера; перед запуском прочитайте предупреждение о полном охвате в справочнике lifecycle. +### Тайм-ауты и завершение нативного Chat + +Нативный Chat также использует `stallTimeoutSec` при ожидании вывода upstream. Непустой текст, рассуждения, отказ, обновления инструментов и события завершения обновляют время ожидания; комментарии keepalive, только роль и только статистика использования его не обновляют. Ожидание чтения медленным клиентом приостанавливает отсчёт. При зависании возникает `upstream_stall_timeout`: событие ошибки для потокового клиента или HTTP 502 без потоковой передачи. Отмена до конечного результата возвращает ошибку отмены вместо успешного частичного ответа. Непотоковый Chat поддерживает SSE с LF, CRLF и многострочными полями data. + ## Удалённый доступ По умолчанию bind `127.0.0.1` доступен только на loopback. Не-loopback-адрес, например diff --git a/docs-site/src/content/docs/tr/reference/configuration/server.md b/docs-site/src/content/docs/tr/reference/configuration/server.md index 439cd70606..880af54edb 100644 --- a/docs-site/src/content/docs/tr/reference/configuration/server.md +++ b/docs-site/src/content/docs/tr/reference/configuration/server.md @@ -15,7 +15,7 @@ yardımcı özellikleri nasıl çalıştıracağını kontrol eder. | `hostname?` | `string` | `"127.0.0.1"` | Bağlama adresi. Geri döngü olmayan bağlamalar `OPENCODEX_API_AUTH_TOKEN` gerektirir. | | `proxy?` | `string` | — | Giden HTTP(S) proxy URL'si veya `${ENV_VAR}`. Yalnızca bu değişkenler ayarlanmadığında `HTTP_PROXY` / `HTTPS_PROXY`'ye uygulanır; geri döngü `NO_PROXY` içinde kalır. | | `emptyCompletionRetry?` | `boolean` | `false` | Metin veya araç çağrısı içermeyen bir Responses tamamlamasını aynı istekle bir kez yeniden denemeyi açıkça etkinleştirir. Yeniden deneme ücretlendirilebilir. `OCX_EMPTY_COMPLETION_RETRY=0`, yapılandırmayı değiştirmeden devre dışı bırakır; combo ve routed-compaction turları hariçtir. | -| `stallTimeoutSec?` | `number` | `300` | `response.incomplete` öncesinde yukarı akış verisi olmadan geçen saniye. Minimum 1. | +| `stallTimeoutSec?` | `number` | `300` | Responses ve yerel Chat için anlamlı üst sunucu ilerlemesi olmadan geçen saniye. En az 1. | | `connectTimeoutMs?` | `number` | `200000` | Deneme başına DNS/TCP/TLS/nihai başlık son tarihi; gövde üretiminden önce biter. | | `shutdownTimeoutMs?` | `number` | `5000` | Aktif turlar iptal edilmeden önce zarif boşaltma süresi sınırı. | | `websockets?` | `boolean` | `false` | Responses WebSocket yolu için `supports_websockets` bildirin. False, HTTP/SSE'yi tutar. | @@ -36,6 +36,10 @@ geçmişi meta verilerini değiştirdiyse yerel sağlayıcı kurtarmasını zorl `ocx recover-history --legacy-openai --yes` çalıştırın. Komut, geçerli dedicated-provider geçmişi de dahil olmak üzere kullanıcı iletisi bulunan tüm `opencodex` satırlarını yeniden etiketler; çalıştırmadan önce lifecycle başvurusundaki tam kapsam uyarısını okuyun. +### Yerel Chat zaman aşımı ve tamamlanma + +Yerel Chat de üst sunucu çıktısını beklerken `stallTimeoutSec` kullanır. Boş olmayan metin, akıl yürütme, ret içeriği, araç güncellemeleri ve bitiş olayları süreyi yeniler; bağlantıyı canlı tutan yorumlar, yalnızca rol ve yalnızca kullanım bilgileri yenilemez. Yavaş istemcinin okumasını beklemek süreyi duraklatır. Zaman aşımı `upstream_stall_timeout` üretir: akış istemcileri hata olayı, akışsız istemciler HTTP 502 alır. Sonuç tamamlanmadan iptal edilen istek, başarılı bir kısmi yanıt yerine iptal hatası döndürür. Akışsız Chat, LF ve CRLF ayraçlarını ve çok satırlı data alanlarını destekler. + ## Uzaktan erişim Varsayılan `127.0.0.1` bağlaması yalnızca geri döngüdür. `0.0.0.0` gibi geri diff --git a/docs-site/src/content/docs/zh-cn/reference/configuration/server.md b/docs-site/src/content/docs/zh-cn/reference/configuration/server.md index 4031f9ff19..6a11e7155f 100644 --- a/docs-site/src/content/docs/zh-cn/reference/configuration/server.md +++ b/docs-site/src/content/docs/zh-cn/reference/configuration/server.md @@ -14,7 +14,7 @@ description: 监听、远程访问、准入密钥、超时、存储、侧车、 | `hostname?` | `string` | `"127.0.0.1"` | 绑定地址。非回环绑定需要 `OPENCODEX_API_AUTH_TOKEN`。 | | `proxy?` | `string` | — | 出站 HTTP(S) 代理 URL，或 `${ENV_VAR}`。仅当 `HTTP_PROXY` / `HTTPS_PROXY` 未设置时才会应用；回环地址始终保留在 `NO_PROXY` 中。 | | `emptyCompletionRetry?` | `boolean` | `false` | 显式启用：当 Responses turn 既无文本也无工具调用时，使用相同请求重试一次，包括流在终止事件之前结束的情况。重试可能产生费用。`OCX_EMPTY_COMPLETION_RETRY=0` 可在不修改配置的情况下禁用；combo 与 routed-compaction turn 不参与。 | -| `stallTimeoutSec?` | `number` | `300` | 在上游没有数据之前可等待的秒数，超过后返回 `response.incomplete`。最小值为 1。 | +| `stallTimeoutSec?` | `number` | `300` | 上游无有效进展的秒数，适用于 Responses 和原生 Chat；最小 1 秒。 | | `connectTimeoutMs?` | `number` | `200000` | 每次尝试的 DNS/TCP/TLS/最终响应头截止时间；它在正文生成之前结束。 | | `shutdownTimeoutMs?` | `number` | `5000` | 优雅停机截止时间，超过后会中止仍在进行中的请求。 | | `websockets?` | `boolean` | `false` | 声明并允许面向客户端的 Responses WebSocket 路径。设为 false 时客户端使用 HTTP/SSE；它不会禁用符合条件的 canonical ChatGPT 上游 WS 优化。 | @@ -34,6 +34,10 @@ description: 监听、远程访问、准入密钥、超时、存储、侧车、 `ocx recover-history --legacy-openai --yes` 强制使用原生提供方恢复。 此命令会重标所有包含用户消息的 `opencodex` 行，其中包括正常的专用提供方历史记录；执行前请查看生命周期参考中的完整范围警告。 +### 原生 Chat 的超时与完成状态 + +原生 Chat 等待上游输出时也使用 `stallTimeoutSec`。非空文本、推理、拒绝内容、工具更新和完成事件会重置等待额度；保活注释、仅角色事件和单独的用量信息不会。等待慢客户端读取期间暂停计时。超时产生 `upstream_stall_timeout`：流式请求收到错误事件，非流式请求返回 HTTP 502。在终态结果到达前取消请求会返回取消错误，而不会把部分答案当作成功。非流式 Chat 支持 LF、CRLF 及多行 data 的 SSE 格式。 + ## 远程访问 默认的 `127.0.0.1` 绑定仅限回环地址。像 `0.0.0.0` 这样的非回环地址需要 diff --git a/docs-site/src/content/docs/zh-tw/reference/configuration/server.md b/docs-site/src/content/docs/zh-tw/reference/configuration/server.md index 402015f060..87307ffe0b 100644 --- a/docs-site/src/content/docs/zh-tw/reference/configuration/server.md +++ b/docs-site/src/content/docs/zh-tw/reference/configuration/server.md @@ -13,7 +13,7 @@ description: 監聽器、遠端存取、許可金鑰、逾時、儲存、sidecar | `hostname?` | `string` | `"127.0.0.1"` | 綁定位址。非回送綁定需要 `OPENCODEX_API_AUTH_TOKEN`。 | | `proxy?` | `string` | — | 對外 HTTP(S) 代理 URL 或 `${ENV_VAR}`。僅在那些變數未設定時套用至 `HTTP_PROXY` / `HTTPS_PROXY`；回送保留在 `NO_PROXY` 中。 | | `emptyCompletionRetry?` | `boolean` | `false` | 明確啟用：當 Responses 完成時沒有文字或工具呼叫，以相同請求重試一次。重試可能產生費用。`OCX_EMPTY_COMPLETION_RETRY=0` 可在不變更設定的情況下停用；combo 與 routed-compaction turn 不適用。 | -| `stallTimeoutSec?` | `number` | `300` | 在 `response.incomplete` 前無上游資料的秒數。最小 1。 | +| `stallTimeoutSec?` | `number` | `300` | 上游無有效進展的秒數，適用於 Responses 與原生 Chat；最小 1 秒。 | | `connectTimeoutMs?` | `number` | `200000` | 每次嘗試的 DNS/TCP/TLS/final-header 截止時間；它在 body 生成前結束。 | | `shutdownTimeoutMs?` | `number` | `5000` | 在中止活躍回合前的優雅排空截止時間。 | | `websockets?` | `boolean` | `false` | 廣告並允許面向 client 的 Responses WebSocket 路徑。False 時 client 使用 HTTP/SSE；不會停用符合條件的 canonical ChatGPT upstream WS 最佳化。 | @@ -32,6 +32,10 @@ description: 監聽器、遠端存取、許可金鑰、逾時、儲存、sidecar 若較舊的開發組建在備份支援存在前變更了 resume-history 中繼資料，請執行 `ocx recover-history --legacy-openai --yes` 以強制原生供應商復原。 此命令會重新標記所有含有使用者訊息的 `opencodex` row，其中也包含正常的專用 provider 歷史；執行前請查看 lifecycle reference 中的完整範圍警告。 +### 原生 Chat 的逾時與完成狀態 + +原生 Chat 等待上游輸出時也使用 `stallTimeoutSec`。非空文字、推理、拒絕內容、工具更新及完成事件會重設等待額度；保活註解、僅角色事件及單獨的用量資訊不會。等待慢速用戶端讀取時暫停計時。逾時產生 `upstream_stall_timeout`：串流請求收到錯誤事件，非串流請求回傳 HTTP 502。終態結果到達前取消請求會回傳取消錯誤，不會將部分答案當成成功。非串流 Chat 支援 LF、CRLF 與多行 data 的 SSE 格式。 + ## 遠端存取 預設的 `127.0.0.1` 綁定僅限回送。如 `0.0.0.0` 的非回送位址需要在 `/api/*` 與 data plane 上都進行 token 認證。在啟動前匯出 token： diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index cc3dfebf3f..e85cf64418 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -1320,7 +1320,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti }]; } const json = parsed; - const responseBytes = new TextEncoder().encode(JSON.stringify(json)).byteLength; + const responseBytes = Buffer.byteLength(JSON.stringify(json), "utf8"); budget.chargeRetained(responseBytes, { kind: "retained_collectors" }); try { const events: AdapterEvent[] = []; diff --git a/src/adapters/google.ts b/src/adapters/google.ts index 7fcc88ba59..e2286d217d 100644 --- a/src/adapters/google.ts +++ b/src/adapters/google.ts @@ -1311,7 +1311,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte bytesReservation.commitRetained(); budget.releaseRetained(total, { kind: "retained_collectors" }); rawText = new TextDecoder().decode(bytes); - rawTextBytes = new TextEncoder().encode(rawText).byteLength; + rawTextBytes = Buffer.byteLength(rawText, "utf8"); const textReservation = budget.reserveTransient(rawTextBytes, { kind: "retained_collectors" }); textReservation.commitRetained(); budget.releaseRetained(total, { kind: "retained_collectors" }); @@ -1333,7 +1333,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte return [{ type: "error", message: `google response was not a JSON object (${valueType})` }]; } raw = parsedRaw; - rawBytes = new TextEncoder().encode(JSON.stringify(raw)).byteLength; + rawBytes = Buffer.byteLength(JSON.stringify(raw), "utf8"); const rawReservation = budget.reserveTransient(rawBytes, { kind: "retained_collectors" }); rawReservation.commitRetained(); budget.releaseRetained(rawTextBytes, { kind: "retained_collectors" }); diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index e338d845ad..a5d4126e15 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -204,7 +204,7 @@ export function buildOpenAIChatPassthroughRequest( messageCount: Array.isArray(body.messages) ? body.messages.length : 0, toolCount: Array.isArray(body.tools) ? body.tools.length : 0, hasCredential, - bodyBytes: new TextEncoder().encode(bodyJson).length, + bodyBytes: Buffer.byteLength(bodyJson, "utf8"), }); } @@ -1650,7 +1650,7 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd messageCount: Array.isArray(messages) ? messages.length : 0, toolCount: Array.isArray(tools) ? tools.length : 0, hasCredential, - bodyBytes: new TextEncoder().encode(bodyJson).length, + bodyBytes: Buffer.byteLength(bodyJson, "utf8"), }); } @@ -2080,7 +2080,7 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd if (Object.hasOwn(json, "service_tier")) { tierMetadata?.observeResponseServiceTier(json.service_tier); } - const responseBytes = new TextEncoder().encode(JSON.stringify(json)).byteLength; + const responseBytes = Buffer.byteLength(JSON.stringify(json), "utf8"); budget.chargeRetained(responseBytes, { kind: "retained_collectors" }); try { const payload = unwrapChatCompletionPayload(json); diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 92de305f6d..1beb08d297 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -3,6 +3,7 @@ import { stripBracketedModelSuffix } from "./openai-chat"; import { normalizeOpenCodeGoAdditionalTools } from "./opencode-go-additional-tools"; import { isXaiResponsesDestination } from "../providers/xai-transport"; import { createHash } from "node:crypto"; +import { Buffer } from "node:buffer"; import type { IncomingMeta, ProviderAdapter } from "./base"; import { namespacedToolName, type AdapterEvent, type OcxParsedRequest, type OcxProviderConfig, type OcxUsage, type TierDecision } from "../types"; import { catalogModelSupportsReasoningSummaries } from "../codex/catalog"; @@ -2308,6 +2309,15 @@ function responsesErrorMessage(payload: unknown): string { return "upstream compaction failed"; } +/** Count an append without rescanning accumulated text, including split surrogate pairs. */ +function appendedUtf8Bytes(previousBytes: number, lastCodeUnit: number, fragment: string): number { + const first = fragment.charCodeAt(0); + // Separate lone surrogates each count as a three-byte replacement character; together + // they encode as one four-byte scalar. Empty fragments produce NaN and never pair. + const joinsSurrogatePair = lastCodeUnit >= 0xd800 && lastCodeUnit <= 0xdbff && first >= 0xdc00 && first <= 0xdfff; + return previousBytes + Buffer.byteLength(fragment, "utf8") - (joinsSurrogatePair ? 2 : 0); +} + export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ProviderAdapter & { passthrough: true } { return { name: "openai-responses", @@ -2569,7 +2579,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ); const releaseBodyObservation = translatorBudget.observeExternallyCapped( "passthrough_serialization", - new TextEncoder().encode(body).byteLength, + Buffer.byteLength(body, "utf8"), ); return { url, @@ -2594,12 +2604,18 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): yield { type: "error", message: "passthrough adapter received no response body" }; return; } - const budgetEncoder = new TextEncoder(); let deltas = ""; + let deltasBytes = 0; + let deltasLastCodeUnit = 0; let doneText = ""; + let doneTextBytes = 0; + let doneTextLastCodeUnit = 0; let snapshot = ""; + let snapshotBytes = 0; let usage: OcxUsage | undefined; + let usageRawBytes = 0; let compactionEncryptedContent: string | undefined; + let compactionEncryptedContentBytes = 0; let completedSeen = false; for await (const event of decodeServerSentEvents(response.body, { translatorBudget: budget })) { let payload: unknown; @@ -2609,21 +2625,25 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): case "response.output_text.delta": if (typeof payload.delta === "string") { const next = deltas + payload.delta; - const previousBytes = budgetEncoder.encode(deltas).byteLength; - const reservation = budget.reserveTransient(budgetEncoder.encode(next).byteLength, { kind: "retained_collectors" }); + const nextBytes = appendedUtf8Bytes(deltasBytes, deltasLastCodeUnit, payload.delta); + const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" }); deltas = next; reservation.commitRetained(); - budget.releaseRetained(previousBytes, { kind: "retained_collectors" }); + budget.releaseRetained(deltasBytes, { kind: "retained_collectors" }); + deltasBytes = nextBytes; + if (payload.delta.length > 0) deltasLastCodeUnit = payload.delta.charCodeAt(payload.delta.length - 1); } break; case "response.output_text.done": if (typeof payload.text === "string") { const next = doneText + payload.text; - const previousBytes = budgetEncoder.encode(doneText).byteLength; - const reservation = budget.reserveTransient(budgetEncoder.encode(next).byteLength, { kind: "retained_collectors" }); + const nextBytes = appendedUtf8Bytes(doneTextBytes, doneTextLastCodeUnit, payload.text); + const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" }); doneText = next; reservation.commitRetained(); - budget.releaseRetained(previousBytes, { kind: "retained_collectors" }); + budget.releaseRetained(doneTextBytes, { kind: "retained_collectors" }); + doneTextBytes = nextBytes; + if (payload.text.length > 0) doneTextLastCodeUnit = payload.text.charCodeAt(payload.text.length - 1); } break; case "response.failed": @@ -2641,28 +2661,28 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): const compaction = output.find(item => isPlainObject(item) && item.type === "compaction"); if (isPlainObject(compaction) && typeof compaction.encrypted_content === "string") { const nextEncryptedContent = compaction.encrypted_content; - const previousBytes = budgetEncoder.encode(compactionEncryptedContent ?? "").byteLength; - const reservation = budget.reserveTransient(budgetEncoder.encode(nextEncryptedContent).byteLength, { kind: "retained_collectors" }); + const nextEncryptedContentBytes = Buffer.byteLength(nextEncryptedContent, "utf8"); + const reservation = budget.reserveTransient(nextEncryptedContentBytes, { kind: "retained_collectors" }); compactionEncryptedContent = nextEncryptedContent; reservation.commitRetained(); - budget.releaseRetained(previousBytes, { kind: "retained_collectors" }); + budget.releaseRetained(compactionEncryptedContentBytes, { kind: "retained_collectors" }); + compactionEncryptedContentBytes = nextEncryptedContentBytes; } const next = responsesPayloadText(payload.response); - const previousBytes = budgetEncoder.encode(snapshot).byteLength; - const reservation = budget.reserveTransient(budgetEncoder.encode(next).byteLength, { kind: "retained_collectors" }); + const nextBytes = Buffer.byteLength(next, "utf8"); + const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" }); snapshot = next; reservation.commitRetained(); - budget.releaseRetained(previousBytes, { kind: "retained_collectors" }); + budget.releaseRetained(snapshotBytes, { kind: "retained_collectors" }); + snapshotBytes = nextBytes; } { const nextUsage = usageFromResponsesPayload(payload.response); // The attached raw usage object can be event-sized (unknown keys carry arbitrary // values); it stays reachable until the terminal yields, so charge it like the // adjacent retained collectors or it would defeat the per-request memory cap. - const previousRawBytes = usage?.rawUsage === undefined ? 0 - : budgetEncoder.encode(JSON.stringify(usage.rawUsage)).byteLength; const nextRawBytes = nextUsage?.rawUsage === undefined ? 0 - : budgetEncoder.encode(JSON.stringify(nextUsage.rawUsage)).byteLength; + : Buffer.byteLength(JSON.stringify(nextUsage.rawUsage), "utf8"); if (nextRawBytes > 0) { const reservation = budget.reserveTransient(nextRawBytes, { kind: "retained_collectors" }); usage = nextUsage; @@ -2670,9 +2690,10 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): } else { usage = nextUsage; } - if (previousRawBytes > 0) { - budget.releaseRetained(previousRawBytes, { kind: "retained_collectors" }); + if (usageRawBytes > 0) { + budget.releaseRetained(usageRawBytes, { kind: "retained_collectors" }); } + usageRawBytes = nextRawBytes; } break; } @@ -2694,10 +2715,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): const text = snapshot || doneText || deltas; if (text) yield { type: "text_delta", text }; budget.releaseRetained( - budgetEncoder.encode(deltas).byteLength - + budgetEncoder.encode(doneText).byteLength - + budgetEncoder.encode(snapshot).byteLength - + (usage?.rawUsage === undefined ? 0 : budgetEncoder.encode(JSON.stringify(usage.rawUsage)).byteLength), + deltasBytes + doneTextBytes + snapshotBytes + usageRawBytes, { kind: "retained_collectors" }, ); yield { @@ -2712,7 +2730,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): try { payload = await response.json(); } catch { return [{ type: "error", message: "malformed upstream compaction response" }]; } - budget.chargeRetained(new TextEncoder().encode(JSON.stringify(payload)).byteLength, { kind: "retained_collectors" }); + budget.chargeRetained(Buffer.byteLength(JSON.stringify(payload), "utf8"), { kind: "retained_collectors" }); if (!isPlainObject(payload)) { return [{ type: "error", message: "malformed upstream compaction response" }]; } diff --git a/src/chat/outbound.ts b/src/chat/outbound.ts index e69d03f499..a51e8ff36c 100644 --- a/src/chat/outbound.ts +++ b/src/chat/outbound.ts @@ -7,7 +7,7 @@ */ type Rec = Record; -import { decodeServerSentEvents, sseFieldValue } from "../lib/sse-decoder"; +import { decodeServerSentEvents } from "../lib/sse-decoder"; import { isTranslatorBudgetExceededError, type TranslatorBudget, @@ -21,6 +21,7 @@ import { isCyberPolicyMessage, } from "../lib/errors"; import { redactSecretString } from "../lib/redact"; +import { createSseBlockBuffer, sseDataPayload } from "../server/sse-payload-rewrite"; function isRec(v: unknown): v is Rec { return !!v && typeof v === "object" && !Array.isArray(v); @@ -880,129 +881,123 @@ export async function collectChatCompletion( model: string, translatorBudget: TranslatorBudget, ): Promise { + const reader = stream.getReader(); const decoder = new TextDecoder(); - let buffer = ""; + const buffer = createSseBlockBuffer(translatorBudget); let content = ""; let refusal: string | null = null; let reasoning = ""; + const retainedBytes = { content: 0, refusal: 0, reasoning: 0 }; const toolCalls = new Map(); // Per-call budget scopes (2 MiB/call enforced by the budget): the map key is the // wire index, which is stable across deltas and present before the call id. const callScope = (index: number) => `chat_collect_${index}`; let finishReason = "stop"; let usage: unknown; - const replaceRetained = (previous: string, next: string, kind: "live_transient" | "retained_collectors") => { - const reservation = translatorBudget.reserveTransient(Buffer.byteLength(next), { kind }); - reservation.commitRetained(); - translatorBudget.releaseRetained(Buffer.byteLength(previous), { kind }); - return next; + const appendRetained = (key: keyof typeof retainedBytes, previous: string, fragment: string): string => { + const previousBytes = retainedBytes[key]; + const nextBytes = appendedUtf8Bytes(previous, previousBytes, fragment); + const reservation = translatorBudget.reserveTransient(nextBytes, { kind: "retained_collectors" }); + try { + const next = previous + fragment; + reservation.commitRetained(); + translatorBudget.releaseRetained(previousBytes, { kind: "retained_collectors" }); + retainedBytes[key] = nextBytes; + return next; + } catch (error) { + reservation.release(); + throw error; + } + }; + const releaseCollectors = () => { + translatorBudget.releaseRetained(retainedBytes.content + retainedBytes.refusal + retainedBytes.reasoning, + { kind: "retained_collectors" }); }; - const reader = stream.getReader(); try { + // Share native relay framing, while retaining the collector's existing admission + // order: release each consumed input frame before accumulating its output fields. + // CRLF and multiline data obey the same contract as the streaming response. for (;;) { - let done = false; - let value: Uint8Array | undefined; - try { - ({ done, value } = await reader.read()); - } catch (err) { - if (isChatCompletionsStreamError(err)) throw err; - if (isTranslatorBudgetExceededError(err)) { - // Provider-controlled overflow is an upstream failure, not a client - // request error: match the adapter/bridge contract (502 upstream_error). - throw new ChatCompletionsStreamError(err.message, { - status: 502, - type: "upstream_error", - code: err.code, - }); - } - throw new ChatCompletionsStreamError(err instanceof Error ? err.message : String(err)); - } + const { done, value } = await reader.read(); if (done) break; if (!value) continue; - buffer = replaceRetained(buffer, buffer + decoder.decode(value, { stream: true }), "live_transient"); - let sep: number; - while ((sep = buffer.indexOf("\n\n")) !== -1) { - const rawFrame = buffer.slice(0, sep); - buffer = replaceRetained(buffer, buffer.slice(sep + 2), "live_transient"); - for (const line of rawFrame.split("\n")) { - const rawData = sseFieldValue(line, "data"); - if (rawData === null) continue; - const data = rawData.trim(); - if (!data || data === "[DONE]") continue; - let parsed: unknown; - try { parsed = JSON.parse(data); } catch { continue; } - if (!isRec(parsed)) continue; - if (isRec(parsed.error)) { - const message = typeof parsed.error.message === "string" - ? parsed.error.message - : "upstream request failed"; - const type = typeof parsed.error.type === "string" ? parsed.error.type : "server_error"; - const code = typeof parsed.error.code === "string" ? parsed.error.code : null; - const status = code === "translation_buffer_limit" - ? 502 - : code === CYBER_POLICY_ERROR_CODE || isCyberPolicyMessage(message) - ? 400 - : streamErrorStatus(message); - const streamError = new ChatCompletionsStreamError(message, { - status, - type: code === "translation_buffer_limit" ? "upstream_error" : type, - code, - }); - throw streamError; - } - if (parsed.usage) usage = parsed.usage; - const choices = Array.isArray(parsed.choices) ? parsed.choices : []; - const choice = isRec(choices[0]) ? choices[0] : null; - if (!choice) continue; - if (typeof choice.finish_reason === "string") finishReason = choice.finish_reason; - const delta = isRec(choice.delta) ? choice.delta : null; - if (!delta) continue; - if (typeof delta.content === "string") content = replaceRetained(content, content + delta.content, "retained_collectors"); - if (delta.refusal !== undefined && delta.refusal !== null) { - if (typeof delta.refusal !== "string") throw refusalTranslationError(); - refusal = replaceRetained(refusal ?? "", (refusal ?? "") + delta.refusal, "retained_collectors"); - } - if (typeof delta.reasoning_content === "string") reasoning = replaceRetained(reasoning, reasoning + delta.reasoning_content, "retained_collectors"); - if (Array.isArray(delta.tool_calls)) { - for (const tc of delta.tool_calls) { - if (!isRec(tc)) continue; - const index = typeof tc.index === "number" ? tc.index : 0; - let current = toolCalls.get(index); - if (!current) { - current = { id: "", name: "", arguments: "", argumentBytes: 0 }; - toolCalls.set(index, current); - translatorBudget.openCall(callScope(index)); - } - if (typeof tc.id === "string") current.id = tc.id; - const fn = isRec(tc.function) ? tc.function : {}; - // Done-frame final arguments are authoritative last-write-wins snapshots. - if (typeof fn.name === "string" && fn.name.length > 0) current.name = fn.name; - if (typeof fn.arguments === "string") { - const replace = fn.arguments.startsWith("{") || fn.arguments.startsWith("[") || current.arguments.length === 0; - const nextBytes = replace - ? Buffer.byteLength(fn.arguments) - : appendedUtf8Bytes(current.arguments, current.argumentBytes, fn.arguments); - const reservation = translatorBudget.reserveTransient(nextBytes, { kind: "tool_args", callId: callScope(index) }); - try { - current.arguments = replace ? fn.arguments : current.arguments + fn.arguments; - reservation.commitRetained(); - translatorBudget.releaseRetained(current.argumentBytes, { kind: "tool_args", callId: callScope(index) }); - current.argumentBytes = nextBytes; - } catch (error) { - reservation.release(); - throw error; - } + buffer.append(decoder.decode(value, { stream: true })); + for (let frame = buffer.next(); frame; frame = buffer.next()) { + const data = sseDataPayload(frame.block)?.trim(); + if (!data || data === "[DONE]") continue; + let parsed: unknown; + try { parsed = JSON.parse(data); } catch { continue; } + if (!isRec(parsed)) continue; + if (isRec(parsed.error)) { + const message = typeof parsed.error.message === "string" + ? parsed.error.message + : "upstream request failed"; + const type = typeof parsed.error.type === "string" ? parsed.error.type : "server_error"; + const code = typeof parsed.error.code === "string" ? parsed.error.code : null; + const status = code === "translation_buffer_limit" + ? 502 + : code === CYBER_POLICY_ERROR_CODE || isCyberPolicyMessage(message) + ? 400 + : streamErrorStatus(message); + const streamError = new ChatCompletionsStreamError(message, { + status, + type: code === "translation_buffer_limit" ? "upstream_error" : type, + code, + }); + throw streamError; + } + if (parsed.usage) usage = parsed.usage; + const choices = Array.isArray(parsed.choices) ? parsed.choices : []; + const choice = isRec(choices[0]) ? choices[0] : null; + if (!choice) continue; + if (typeof choice.finish_reason === "string") finishReason = choice.finish_reason; + const delta = isRec(choice.delta) ? choice.delta : null; + if (!delta) continue; + if (typeof delta.content === "string") content = appendRetained("content", content, delta.content); + if (delta.refusal !== undefined && delta.refusal !== null) { + if (typeof delta.refusal !== "string") throw refusalTranslationError(); + refusal = appendRetained("refusal", refusal ?? "", delta.refusal); + } + if (typeof delta.reasoning_content === "string") reasoning = appendRetained("reasoning", reasoning, delta.reasoning_content); + if (Array.isArray(delta.tool_calls)) { + for (const tc of delta.tool_calls) { + if (!isRec(tc)) continue; + const index = typeof tc.index === "number" ? tc.index : 0; + let current = toolCalls.get(index); + if (!current) { + current = { id: "", name: "", arguments: "", argumentBytes: 0 }; + toolCalls.set(index, current); + translatorBudget.openCall(callScope(index)); + } + if (typeof tc.id === "string") current.id = tc.id; + const fn = isRec(tc.function) ? tc.function : {}; + // Done-frame final arguments are authoritative last-write-wins snapshots. + if (typeof fn.name === "string" && fn.name.length > 0) current.name = fn.name; + if (typeof fn.arguments === "string") { + const replace = fn.arguments.startsWith("{") || fn.arguments.startsWith("[") || current.arguments.length === 0; + const nextBytes = replace + ? Buffer.byteLength(fn.arguments) + : appendedUtf8Bytes(current.arguments, current.argumentBytes, fn.arguments); + const reservation = translatorBudget.reserveTransient(nextBytes, { kind: "tool_args", callId: callScope(index) }); + try { + current.arguments = replace ? fn.arguments : current.arguments + fn.arguments; + reservation.commitRetained(); + translatorBudget.releaseRetained(current.argumentBytes, { kind: "tool_args", callId: callScope(index) }); + current.argumentBytes = nextBytes; + } catch (error) { + reservation.release(); + throw error; } } } } } + buffer.compact(); } } catch (error) { - // Processing may fail between reads; cancel while we still own the lock so the - // upstream translator releases its maps and stops any pending provider read. + // Cancel while we still own the reader so a failed collection releases its upstream. try { await reader.cancel(error); } catch { /* preserve the original failure */ } - translatorBudget.releaseRetained(Buffer.byteLength(refusal ?? ""), { kind: "retained_collectors" }); + releaseCollectors(); // Never leak an open call scope on the error path; the turn budget's // dispose is a backstop, not the owner of this transfer. for (const index of toolCalls.keys()) translatorBudget.closeCall(callScope(index)); @@ -1015,8 +1010,12 @@ export async function collectChatCompletion( code: error.code, }); } - throw error; + if (isChatCompletionsStreamError(error)) throw error; + throw new ChatCompletionsStreamError(error instanceof Error ? error.message : String(error)); } finally { + // Preserve the previous EOF contract: only delimiter-terminated events are collected. + // A partial final frame is discarded, with its retained input ownership released. + buffer.clear(); reader.releaseLock(); } @@ -1048,6 +1047,7 @@ export async function collectChatCompletion( return copy; }); } catch (error) { + releaseCollectors(); for (const copyBytes of chargedCopies) { translatorBudget.releaseRetained(copyBytes, { kind: "retained_collectors" }); } diff --git a/src/lib/admission.ts b/src/lib/admission.ts index 7160603b6c..9965b781ea 100644 --- a/src/lib/admission.ts +++ b/src/lib/admission.ts @@ -58,20 +58,26 @@ export function createAdmissionGate(name: string, limit: number): { } export function retainedUtf8Bytes(value: string): number { - return new TextEncoder().encode(value).byteLength; + // Keep TextEncoder's runtime coercion for legacy callers outside the string-typed contract. + // Template coercion rejects Symbols; String(value) would silently accept them. + return Buffer.byteLength(typeof value === "string" ? value : value === undefined ? "" : `${value}`, "utf8"); } function utf8Prefix(value: string, maxBytes: number): string { if (maxBytes <= 0) return ""; let bytes = 0; - let result = ""; - for (const character of value) { - const size = retainedUtf8Bytes(character); + let end = 0; + while (end < value.length) { + const code = value.charCodeAt(end); + const next = value.charCodeAt(end + 1); + const pair = code >= 0xd800 && code <= 0xdbff && next >= 0xdc00 && next <= 0xdfff; + // An unpaired surrogate encodes as a three-byte replacement, like TextEncoder. + const size = code <= 0x7f ? 1 : code <= 0x7ff ? 2 : pair ? 4 : 3; if (bytes + size > maxBytes) break; - result += character; bytes += size; + end += pair ? 2 : 1; } - return result; + return value.slice(0, end); } export function truncateRetainedUtf8(value: string, maxBytes: number): string { diff --git a/src/lib/translator-budget.ts b/src/lib/translator-budget.ts index 18400eebf7..d1914894f7 100644 --- a/src/lib/translator-budget.ts +++ b/src/lib/translator-budget.ts @@ -112,14 +112,15 @@ export function retainTranslatedEvent( */ export function retainTranslatedEventBatch(events: T[], budget: TranslatorBudget): void { if (events.length === 0) return; - const serialized = events.map(event => JSON.stringify(event)); - const totalBytes = Buffer.byteLength(`[${serialized.join(",")}]`); + // Preserve atomic batch admission without retaining serialized strings or joining a second copy. + const eventBytes = events.map(event => Buffer.byteLength(JSON.stringify(event))); + const totalBytes = eventBytes.reduce((total, bytes) => total + bytes, events.length + 1); budget.chargeRetained(totalBytes, { kind: "retained_collectors" }); for (let index = 0; index < events.length; index++) { const delimiterBytes = index === events.length - 1 ? 2 : 1; retainedEventOwnership.set(events[index]!, { budget, - bytes: Buffer.byteLength(serialized[index]!) + delimiterBytes, + bytes: eventBytes[index]! + delimiterBytes, }); } } diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 3234dfed00..b50af22cb5 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -288,7 +288,7 @@ async function handleChatCompletionsWithBudget( try { internalBodyJson = JSON.stringify(internalBody); translatorBudget.chargeRetained( - new TextEncoder().encode(internalBodyJson).byteLength, + Buffer.byteLength(internalBodyJson, "utf8"), { kind: "request_copies" }, ); } catch (err) { diff --git a/src/server/chat-native-sse.ts b/src/server/chat-native-sse.ts index 0d80733057..af9806f9a8 100644 --- a/src/server/chat-native-sse.ts +++ b/src/server/chat-native-sse.ts @@ -7,7 +7,8 @@ import { type TranslatorBudget, } from "../lib/translator-budget"; import type { OcxUsage } from "../types"; -import { nextSseBlock, replaceSseDataPayload, sseDataPayload } from "./sse-payload-rewrite"; +import { resolveStallTimeoutSec } from "../stall-timeout"; +import { createSseBlockBuffer, replaceSseDataPayload, sseDataPayload } from "./sse-payload-rewrite"; type Rec = Record; @@ -121,6 +122,7 @@ interface NativeChatSseOptions { requestedModel: string; translatorBudget: TranslatorBudget; signal: AbortSignal; + stallTimeoutSec?: number; onFirstOutput?: () => void; onUsage: (usage: OcxUsage) => void; onTerminal?: (status: number, message?: string) => void; @@ -135,8 +137,11 @@ export function nativeChatSse( const decoder = new TextDecoder(); const encoder = new TextEncoder(); const scope = { kind: "live_transient" as const }; - let buffer = ""; - let bufferBytes = 0; + const buffer = createSseBlockBuffer(options.translatorBudget, nextBytes => { + if (nextBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) { + throw new Error("upstream SSE event exceeded the safe limit", { cause: { code: "translation_buffer_limit" } }); + } + }); let queuedBytes = 0; let sawFinish = false; let sawDone = false; @@ -144,50 +149,31 @@ export function nativeChatSse( let settled = false; let cancelled = false; let cancelledBySignal = false; + const stallMs = resolveStallTimeoutSec(options.stallTimeoutSec) * 1_000; + // Count only active upstream consumption; downstream backpressure must not + // consume the provider's inactivity allowance. Comments and empty deltas do. + let remainingStallMs = stallMs; + let pullDeadline = 0; + const stalled = new Error("upstream stream stalled without meaningful progress"); const releaseQueued = () => { if (queuedBytes === 0) return; options.translatorBudget.releaseRetained(queuedBytes, scope); queuedBytes = 0; }; - const replaceBuffer = (next: string) => { - const nextBytes = encoder.encode(next).byteLength; - const reservation = options.translatorBudget.reserveTransient(nextBytes, scope); - reservation.commitRetained(); - options.translatorBudget.releaseRetained(bufferBytes, scope); - buffer = next; - bufferBytes = nextBytes; - }; - const appendBuffer = (fragment: string) => { - if (!fragment) return; - const fragmentBytes = encoder.encode(fragment).byteLength; - const nextBytes = bufferBytes + fragmentBytes; - if (nextBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) { - throw new Error("upstream SSE event exceeded the safe limit", { cause: { code: "translation_buffer_limit" } }); - } - const reservation = options.translatorBudget.reserveTransient(nextBytes, scope); + const enqueue = (controller: ReadableStreamDefaultController, text: string) => { + const byteLength = Buffer.byteLength(text); + const reservation = options.translatorBudget.reserveTransient(byteLength, scope); try { - buffer += fragment; + controller.enqueue(encoder.encode(text)); reservation.commitRetained(); - options.translatorBudget.releaseRetained(bufferBytes, scope); - bufferBytes = nextBytes; + queuedBytes += byteLength; } catch (error) { reservation.release(); throw error; } }; - const enqueue = (controller: ReadableStreamDefaultController, text: string) => { - const bytes = encoder.encode(text); - const reservation = options.translatorBudget.reserveTransient(bytes.byteLength, scope); - controller.enqueue(bytes); - reservation.commitRetained(); - queuedBytes += bytes.byteLength; - }; - const releaseBuffer = () => { - options.translatorBudget.releaseRetained(bufferBytes, scope); - buffer = ""; - bufferBytes = 0; - }; + const releaseBuffer = () => buffer.clear(); const settle = (status: number, message?: string) => { if (settled) return; settled = true; @@ -219,13 +205,20 @@ export function nativeChatSse( ): void => { const payload = sseDataPayload(block); if (payload === null) { + if (performance.now() >= pullDeadline) throw stalled; enqueue(controller, block + delimiter); return; } const trimmed = payload.trim(); + if (trimmed === "") { + if (performance.now() >= pullDeadline) throw stalled; + enqueue(controller, block + delimiter); + return; + } if (trimmed === "[DONE]") { sawDone = true; enqueue(controller, replaceSseDataPayload(block, "[DONE]") + delimiter); + releaseBuffer(); settle(200); try { void reader.cancel().catch(() => {}); } catch { /* already closed */ } controller.close(); @@ -254,6 +247,7 @@ export function nativeChatSse( } const safe = chatCompletionsErrorBody(status, classified.message, classified.type, classified.code); enqueue(controller, replaceSseDataPayload(block, JSON.stringify(safe)) + delimiter); + releaseBuffer(); settle(isCyberPolicyCode(classified.code) ? 400 : status, classified.message); try { void reader.cancel(new Error(classified.message)).catch(() => {}); } catch { /* already closed */ } controller.close(); @@ -262,16 +256,28 @@ export function nativeChatSse( const usage = usageFromChat(parsed.usage); if (usage) options.onUsage(usage); const choices = Array.isArray(parsed.choices) ? parsed.choices : []; - if (choices.some(choice => isRec(choice) && typeof choice.finish_reason === "string" && choice.finish_reason.length > 0)) { - sawFinish = true; - } - if (!firstOutput && choices.some(choice => { + const finished = choices.some(choice => isRec(choice) && typeof choice.finish_reason === "string" && choice.finish_reason.length > 0); + if (finished) sawFinish = true; + const hasProgress = choices.some(choice => { if (!isRec(choice) || !isRec(choice.delta)) return false; const delta = choice.delta; return (typeof delta.content === "string" && delta.content.length > 0) || (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) - || (Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0); - })) { + || (typeof delta.reasoning === "string" && delta.reasoning.length > 0) + || (Array.isArray(delta.reasoning_details) && delta.reasoning_details.some(detail => + isRec(detail) && typeof detail.text === "string" && detail.text.length > 0)) + || (typeof delta.refusal === "string" && delta.refusal.length > 0) + || (Array.isArray(delta.tool_calls) && delta.tool_calls.some(tool => { + if (!isRec(tool)) return false; + const fn = isRec(tool.function) ? tool.function : {}; + return (typeof tool.id === "string" && tool.id.length > 0) + || (typeof fn.name === "string" && fn.name.length > 0) + || (typeof fn.arguments === "string" && fn.arguments.length > 0); + })); + }); + if (hasProgress || finished) pullDeadline = performance.now() + stallMs; + else if (performance.now() >= pullDeadline) throw stalled; + if (!firstOutput && hasProgress) { firstOutput = true; options.onFirstOutput?.(); } @@ -280,6 +286,7 @@ export function nativeChatSse( function onAbort() { cancelledBySignal = true; + releaseBuffer(); if (!settled) { settled = true; options.onCancel?.(); @@ -292,15 +299,39 @@ export function nativeChatSse( return new ReadableStream({ async pull(controller) { releaseQueued(); + pullDeadline = performance.now() + remainingStallMs; try { for (;;) { - const next = nextSseBlock(buffer); + if (cancelledBySignal) { + releaseBuffer(); + controller.close(); + return; + } + const next = buffer.next(); if (next) { - replaceBuffer(next.rest); + buffer.compact(); processBlock(controller, next.block, next.delimiter); return; } - const { done, value } = await reader.read(); + if (performance.now() >= pullDeadline) throw stalled; + let timeout: ReturnType | undefined; + let read: Awaited>; + try { + read = await Promise.race([ + reader.read(), + new Promise((_, reject) => { + const check = () => { + const remaining = pullDeadline - performance.now(); + if (remaining <= 0) reject(stalled); + else timeout = setTimeout(check, Math.min(remaining, 2_147_483_647)); + }; + check(); + }), + ]); + } finally { + if (timeout !== undefined) clearTimeout(timeout); + } + const { done, value } = read; if (cancelled) return; if (cancelledBySignal) { releaseBuffer(); @@ -308,11 +339,11 @@ export function nativeChatSse( return; } if (!done) { - appendBuffer(decoder.decode(value, { stream: true })); + buffer.append(decoder.decode(value, { stream: true })); continue; } - appendBuffer(decoder.decode()); - if (buffer.trim().length > 0) { + buffer.append(decoder.decode()); + if (buffer.tail().trim().length > 0) { fail(controller, "upstream SSE ended with an unterminated event", "upstream_sse_unterminated"); return; } @@ -327,13 +358,20 @@ export function nativeChatSse( return; } } catch (error) { + if (cancelled || cancelledBySignal) { + releaseBuffer(); + try { controller.close(); } catch { /* already closed */ } + return; + } const overflow = isTranslatorBudgetExceededError(error) || (error instanceof Error && (error.cause as { code?: unknown } | undefined)?.code === "translation_buffer_limit"); fail( controller, overflow ? "upstream SSE event exceeded the safe limit" : error instanceof Error ? error.message : String(error), - overflow ? "translation_buffer_limit" : "upstream_sse_error", + overflow ? "translation_buffer_limit" : error === stalled ? "upstream_stall_timeout" : "upstream_sse_error", ); + } finally { + remainingStallMs = Math.max(0, pullDeadline - performance.now()); } }, cancel(reason) { diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 9fb24e143a..8cc633665c 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -262,7 +262,7 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio retainedRequestBytes = 0; }; const retainRequest = (request: AdapterRequest) => { - const bytes = new TextEncoder().encode(request.body).byteLength; + const bytes = Buffer.byteLength(request.body); translatorBudget.chargeRetained(bytes, { kind: "request_copies" }); retainedRequestBytes = bytes; }; @@ -501,24 +501,29 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; if (contentType.includes("text/event-stream") && response.body) { if (requestedStream) transferTurnToStream(); + let terminalStatus: number | undefined; const stream = nativeChatSse(response.body, { requestedModel, translatorBudget, signal: upstream.signal, + stallTimeoutSec: config.stallTimeoutSec, onFirstOutput: logIds ? () => recordFirstOutput(logCtx, logIds.start) : undefined, onUsage: usage => { logCtx.usage = usage; attempt.usage = usage; }, + onTerminal: (status: number, message?: string) => { + terminalStatus = status; + if (!requestedStream) return; + try { + cleanupAbort(); + finishLog(status, message, "terminal"); + if (status >= 400) upstream.abort(); + } finally { + releaseStreamTurn(); + } + }, ...(requestedStream ? { - onTerminal: (status: number, message?: string) => { - try { - cleanupAbort(); - finishLog(status, message, "terminal"); - } finally { - releaseStreamTurn(); - } - }, onCancel: () => { try { cleanupAbort(); @@ -543,11 +548,20 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio try { const completion = await collectChatCompletion(stream, requestedModel, translatorBudget); cleanupAbort(); + // A cancelled native relay closes its downstream body. EOF alone must not + // promote the buffered prefix to a successful Chat completion. A terminal + // already accepted by the relay retains precedence over a later abort. + if (req.signal.aborted && terminalStatus === undefined) { + return fail(499, "Client cancelled request", "client_cancelled"); + } finishLog(200); return Response.json(completion); } catch (error) { cleanupAbort(); upstream.abort(); + if (req.signal.aborted && terminalStatus === undefined) { + return fail(499, "Client cancelled request", "client_cancelled"); + } if (isChatCompletionsStreamError(error)) { return fail(error.status, error.message, error.type, error.code); } diff --git a/src/server/request-decompress.ts b/src/server/request-decompress.ts index 1939479429..f2d2d71ff7 100644 --- a/src/server/request-decompress.ts +++ b/src/server/request-decompress.ts @@ -1,3 +1,4 @@ +import { Buffer } from "node:buffer"; import { gunzipSync, inflateRawSync, inflateSync, zstdDecompressSync } from "node:zlib"; import type { TranslatorBudget } from "../lib/translator-budget"; @@ -29,7 +30,7 @@ export const MAX_DECOMPRESSED_BODY_BYTES = 256 * 1024 * 1024; * shrink and is stuck. An UNBOUNDED inbound cap is not an acceptable answer: this admission * limit is the only thing standing between one request and the process heap, and * `readBoundedJsonRequestBody` materializes the body several times over (retained wire bytes, - * decoded bytes, the decoded string, the re-encoded measurement copies, and the parsed object + * decoded bytes, the decoded string, the serialized measurement string, and the parsed object * graph), so peak RSS is a MULTIPLE of whatever is admitted here. 512 MiB is the largest value * that keeps that multiple survivable on an ordinary machine, and it is what #3573 asked for. */ @@ -324,12 +325,14 @@ export async function readBoundedJsonRequestBody( const decoded = decodeRequestBody(raw, encoding, maxBytes); releaseDecoded = decoded === raw ? undefined : budget?.observeAcceptedRequestCopy(decoded.byteLength); const text = new TextDecoder().decode(decoded); - releaseText = budget?.observeAcceptedRequestCopy(new TextEncoder().encode(text).byteLength); + // Count UTF-8 without allocating another request-sized byte array for diagnostics. + releaseText = budget?.observeAcceptedRequestCopy(Buffer.byteLength(text, "utf8")); if (options && "emptyBodyFallback" in options && text.trim() === "") { return options.emptyBodyFallback; } const parsed = JSON.parse(text); - budget?.observeAcceptedRequestCopy(new TextEncoder().encode(JSON.stringify(parsed)).byteLength); + // Keep the serialized-size contract: normalization can expand numeric literals. + budget?.observeAcceptedRequestCopy(Buffer.byteLength(JSON.stringify(parsed), "utf8")); return parsed; } finally { releaseText?.(); diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index f9fb62065d..1c2f38b510 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -1,4 +1,5 @@ import type { TranslatorBudget } from "../lib/translator-budget"; +import { Buffer } from "node:buffer"; /** * Shared client-facing SSE payload rewrite shell. @@ -37,18 +38,21 @@ export function payloadRewriteAsBlockRewrite(rewrite: SsePayloadRewrite): SseBlo export function composeSseBlockRewrites(...rewrites: SseBlockRewrite[]): SseBlockRewrite { const active = rewrites.filter(Boolean); if (active.length === 0) return Object.assign((block: string) => [block], {}); + let disposed = false; const composed: SseBlockRewrite = (block: string) => { let blocks: readonly string[] = [block]; for (const rewrite of active) { const next: string[] = []; - for (const current of blocks) next.push(...rewrite(current)); + for (const current of blocks) { + if (disposed) return []; + next.push(...rewrite(current)); + } blocks = next; } return blocks; }; // Child disposal is part of the contract: one idempotent disposer for the // whole chain, so relay teardown never leaks a nested collector. - let disposed = false; composed.dispose = () => { if (disposed) return; disposed = true; @@ -70,10 +74,116 @@ export function nextSseBlock(buffer: string): { block: string; delimiter: string }; } +/** + * Incremental form of nextSseBlock for bounded relays. The scan cursor visits only new text and + * consuming a block subtracts its byte length instead of recounting the remaining suffix. + * Old/new buffer overlap still requires admission before either append or consumption commits. + * Call compact before yielding to stop retaining an already-consumed prefix across pulls. + */ +export function createSseBlockBuffer( + budget: TranslatorBudget, + assertAppendSize?: (bytes: number) => void, +): { + append(fragment: string): void; + next(): { block: string; delimiter: string } | null; + tail(): string; + compact(): void; + clear(): void; +} { + const scope = { kind: "live_transient" as const }; + let buffer = ""; + let offset = 0; + let scanOffset = 0; + let bufferBytes = 0; + + const compact = (): void => { + if (offset === 0) return; + buffer = buffer.slice(offset); + scanOffset -= offset; + offset = 0; + }; + + return { + append(fragment) { + if (!fragment) return; + let fragmentBytes = Buffer.byteLength(fragment, "utf8"); + // Decoder output never splits a surrogate pair, but keep the helper exact for string callers. + const last = buffer.charCodeAt(buffer.length - 1); + const first = fragment.charCodeAt(0); + if (offset < buffer.length && last >= 0xd800 && last <= 0xdbff && first >= 0xdc00 && first <= 0xdfff) { + fragmentBytes -= 2; + } + const nextBytes = bufferBytes + fragmentBytes; + assertAppendSize?.(nextBytes); + const reservation = budget.reserveTransient(nextBytes, scope); + try { + compact(); + buffer += fragment; + reservation.commitRetained(); + budget.releaseRetained(bufferBytes, scope); + bufferBytes = nextBytes; + } catch (error) { + reservation.release(); + throw error; + } + }, + next() { + for (;;) { + const newline = buffer.indexOf("\n", scanOffset); + if (newline < 0) { + scanOffset = buffer.length; + return null; + } + let end = newline + 1; + if (buffer[end] === "\r") end += 1; + if (end === buffer.length) { + // Keep the candidate first newline until its possible blank-line delimiter arrives. + scanOffset = newline; + return null; + } + if (buffer[end] !== "\n") { + scanOffset = newline + 1; + continue; + } + end += 1; + const start = newline > offset && buffer[newline - 1] === "\r" ? newline - 1 : newline; + const block = buffer.slice(offset, start); + const delimiter = buffer.slice(start, end); + const nextBytes = bufferBytes - Buffer.byteLength(block, "utf8") - delimiter.length; + const reservation = budget.reserveTransient(nextBytes, scope); + reservation.commitRetained(); + budget.releaseRetained(bufferBytes, scope); + bufferBytes = nextBytes; + offset = end; + scanOffset = end; + if (offset === buffer.length) { + buffer = ""; + offset = 0; + scanOffset = 0; + } + return { block, delimiter }; + } + }, + tail: () => buffer.slice(offset), + compact, + clear() { + budget.releaseRetained(bufferBytes, scope); + buffer = ""; + offset = 0; + scanOffset = 0; + bufferBytes = 0; + }, + }; +} + /** Join all data lines from one SSE event according to the event-stream field rules. */ export function sseDataPayload(block: string): string | null { const data: string[] = []; for (const line of block.split(/\r?\n/)) { + if (line === "data") { + data.push(""); + continue; + } if (!line.startsWith("data:")) continue; const value = line.slice(5); data.push(value.startsWith(" ") ? value.slice(1) : value); @@ -88,7 +198,7 @@ export function replaceSseDataPayload(block: string, payload: string): string { const rewritten: string[] = []; let replaced = false; for (const line of lines) { - if (!line.startsWith("data:")) { + if (line !== "data" && !line.startsWith("data:")) { rewritten.push(line); continue; } @@ -137,8 +247,7 @@ export function relaySseWithBlockRewrite( const reader = body.getReader(); const decoder = new TextDecoder(); const encoder = new TextEncoder(); - let buffer = ""; - let bufferBytes = 0; + const buffer = createSseBlockBuffer(translatorBudget); // Relays have several independent teardown paths; disposal is exactly once. let disposed = false; let cancelled = false; @@ -148,40 +257,16 @@ export function relaySseWithBlockRewrite( try { rewrite.dispose?.(); } catch { /* teardown must not throw */ } }; - const appendBuffer = (fragment: string): void => { - if (!fragment) return; - const nextBytes = bufferBytes + encoder.encode(fragment).byteLength; - const reservation = translatorBudget.reserveTransient(nextBytes, { kind: "live_transient" }); - try { - buffer += fragment; - reservation.commitRetained(); - translatorBudget.releaseRetained(bufferBytes, { kind: "live_transient" }); - bufferBytes = nextBytes; - } catch (error) { - reservation.release(); - throw error; - } - }; - - const replaceBuffer = (next: string): void => { - const nextBytes = encoder.encode(next).byteLength; - const reservation = translatorBudget.reserveTransient(nextBytes, { kind: "live_transient" }); - reservation.commitRetained(); - buffer = next; - translatorBudget.releaseRetained(bufferBytes, { kind: "live_transient" }); - bufferBytes = nextBytes; - }; - const enqueueText = ( controller: ReadableStreamDefaultController, text: string, ): void => { - const bytes = encoder.encode(text).byteLength; + const bytes = Buffer.byteLength(text, "utf8"); const reservation = translatorBudget.reserveTransient(bytes, { kind: "live_transient" }); try { const encoded = encoder.encode(text); - reservation.commitRetained(); controller.enqueue(encoded); + reservation.commitRetained(); translatorBudget.releaseRetained(bytes, { kind: "live_transient" }); } catch (error) { reservation.release(); @@ -189,35 +274,33 @@ export function relaySseWithBlockRewrite( } }; - const releaseBuffer = (): void => { - translatorBudget.releaseRetained(bufferBytes, { kind: "live_transient" }); - buffer = ""; - bufferBytes = 0; - }; - const emitProcessedBlocks = ( controller: ReadableStreamDefaultController, flushFinal = false, ): number => { let emitted = 0; - let next: { block: string; delimiter: string; rest: string } | null; - while ((next = nextSseBlock(buffer))) { - replaceBuffer(next.rest); - for (const outBlock of rewrite(next.block)) { + let next: { block: string; delimiter: string } | null; + while (!cancelled && (next = buffer.next())) { + const outBlocks = rewrite(next.block); + if (cancelled) return emitted; + for (const outBlock of outBlocks) { enqueueText(controller, outBlock + next.delimiter); emitted += 1; } } - if (flushFinal && buffer.length > 0) { - const tailBlocks = rewrite(buffer); + buffer.compact(); + const tail = flushFinal ? buffer.tail() : ""; + if (tail.length > 0) { + const tailBlocks = rewrite(tail); + if (cancelled) return emitted; // A trailing fragment has no delimiter of its own; multiple emitted // blocks must still be framed as separate events (#893 review). - const tailDelimiter = buffer.includes("\r\n") ? "\r\n\r\n" : "\n\n"; + const tailDelimiter = tail.includes("\r\n") ? "\r\n\r\n" : "\n\n"; for (let i = 0; i < tailBlocks.length; i++) { enqueueText(controller, tailBlocks[i]! + (i < tailBlocks.length - 1 ? tailDelimiter : "")); emitted += 1; } - releaseBuffer(); + buffer.clear(); } return emitted; }; @@ -236,18 +319,20 @@ export function relaySseWithBlockRewrite( // after its disposal (#893 review). if (cancelled) return; if (done) { - appendBuffer(decoder.decode()); + buffer.append(decoder.decode()); emitProcessedBlocks(controller, true); - releaseBuffer(); + if (cancelled) return; + buffer.clear(); disposeRewrite(); controller.close(); return; } - appendBuffer(decoder.decode(value, { stream: true })); - if (emitProcessedBlocks(controller) > 0) return; + buffer.append(decoder.decode(value, { stream: true })); + const emitted = emitProcessedBlocks(controller); + if (cancelled || emitted > 0) return; } } catch (error) { - releaseBuffer(); + buffer.clear(); disposeRewrite(); // Cancelling one tee branch waits for its sibling. Surface the failure // now so downstream can abort upstream and release the inspection branch. @@ -257,7 +342,7 @@ export function relaySseWithBlockRewrite( }, cancel(reason) { cancelled = true; - releaseBuffer(); + buffer.clear(); disposeRewrite(); reader.cancel(reason).catch(() => {}); }, diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index aa0bc914cb..7109f58b89 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -1,5 +1,7 @@ # Adapter Registry Authority +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Decision Runtime adapter construction has one authority: `src/adapters/registry.ts`. diff --git a/structure/catalog.md b/structure/catalog.md index eda6f26e80..4e9fc9bfff 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -1,5 +1,7 @@ # Model Catalog +Shared parsing and streaming follow the [request-copy](transports/responses.md#request-copy-accounting) and [stream-buffer accounting](transports/responses.md#stream-buffer-accounting) contracts. + ## Shared catalog `src/codex/catalog.ts` builds a shared Codex-shaped catalog for CLI, TUI, App, and SDK. It: diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 54689b36a1..b8cc81a381 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -1,5 +1,7 @@ # Claude Desktop Integration +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Connected Claude Desktop profiles Connected `ocx claude desktop apply` reads the hub's Desktop snapshot and writes the hub origin diff --git a/structure/clients/integrations.md b/structure/clients/integrations.md index 33d877b5e5..81e93f764e 100644 --- a/structure/clients/integrations.md +++ b/structure/clients/integrations.md @@ -1,5 +1,7 @@ # Client Integrations +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + The client-integration subsystem writes one generated OpenCodex provider contribution into a third-party client's existing config without taking ownership of the rest of that file. Its core promise is reversibility: apply snapshots first, writes atomically, records exactly what it owns, diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index d1d0193048..5620a5b95c 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -1,5 +1,7 @@ # Images Data Plane +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Standalone Images Codex's local `image_gen.imagegen` tool makes a second Images request after the model calls it: diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 0593da7e6a..37180ee423 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -1,5 +1,7 @@ # Inbound Compatibility Surfaces +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Chat Completions inbound native path `POST /v1/chat/completions` sends eligible `openai-chat` routes directly to the provider's Chat @@ -35,6 +37,19 @@ request-signal cancellation contracts as routed Responses transport. Because call before it binds the adapter; the pick remains inert unless a strategy is configured and the committed key is cooling. See [`responses.md`](../transports/responses.md). +### Native Chat completion lifecycle + +`src/server/chat-native-sse.ts` applies the resolved `stallTimeoutSec` while waiting for upstream +progress. Nonempty text, reasoning, refusal, tool identity/arguments, and finish frames renew the +allowance; comments, role-only frames, empty deltas, and usage alone do not. Downstream backpressure +pauses this wait budget. A stall emits a Chat error with `upstream_stall_timeout` and logs 502; +the non-streaming endpoint returns HTTP 502 rather than a successful partial result. + +`src/chat/outbound.ts` collects LF/CRLF, multiline data, and split UTF-8 through the shared SSE +block buffer and tracks appended output bytes incrementally. A caller cancellation before a native +terminal returns 499 / `client_cancelled`; an already accepted terminal keeps its result. Reader, +timer, turn, and translator ownership are released through the existing lifecycle. + ## Chat conversation identity forwarding `src/server/chat-completions.ts` preserves caller `prompt_cache_key` on the Chat-to-Responses diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 82d0ef5d20..6ec22d212b 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -1,5 +1,7 @@ # GUI And Management API +Shared parsing and streaming follow the [request-copy](transports/responses.md#request-copy-accounting) and [stream-buffer accounting](transports/responses.md#stream-buffer-accounting) contracts. + ## Dashboard serving The bundled React dashboard is built into `gui/dist` and served by the same Bun proxy. `ocx gui` diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md index 8e31434039..71984d7a64 100644 --- a/structure/ops/docs-and-release.md +++ b/structure/ops/docs-and-release.md @@ -1,5 +1,7 @@ # Docs And Release +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Public docs The public documentation site lives in `docs-site/` and is built with Astro + Starlight. English is diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index c8ef069110..eec5389e80 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -1,5 +1,7 @@ # Background Service And Sidecars +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Background service command selection A bare `ocx service` is an idempotent install-or-repair command. Argument validation happens before diff --git a/structure/overview.md b/structure/overview.md index d5d1a2207a..b82d9353e3 100644 --- a/structure/overview.md +++ b/structure/overview.md @@ -1,5 +1,7 @@ # Overview +Shared parsing and streaming follow the [request-copy](transports/responses.md#request-copy-accounting) and [stream-buffer accounting](transports/responses.md#stream-buffer-accounting) contracts. + ## Product boundary opencodex is a local proxy for Codex. It does not patch Codex binaries. It changes local Codex diff --git a/structure/providers/chat-compat.md b/structure/providers/chat-compat.md index 7cf84ef7ec..16f023a32f 100644 --- a/structure/providers/chat-compat.md +++ b/structure/providers/chat-compat.md @@ -1,5 +1,7 @@ # Chat Provider Compatibility +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Reasoning and tool-result compatibility Kiro groups only consecutive original-message tool results whose raw call ID exactly matches diff --git a/structure/providers/cursor.md b/structure/providers/cursor.md index be42793e0b..381a523659 100644 --- a/structure/providers/cursor.md +++ b/structure/providers/cursor.md @@ -1,5 +1,7 @@ # Cursor Provider +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## Cursor Native Exec Cursor's experimental live transport can receive server-driven local read/write/delete/ls/grep, diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index 5f4217b093..58bb0ecca3 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -1,5 +1,7 @@ # xAI Grok Provider +Shared parsing and streaming follow the [request-copy](../transports/responses.md#request-copy-accounting) and [stream-buffer accounting](../transports/responses.md#stream-buffer-accounting) contracts. + ## xAI Grok hardening (official Grok Build contract parity) Grounded in the open-sourced official client (xai-org/grok-build); unit + evidence: diff --git a/structure/runtime.md b/structure/runtime.md index ac2eb378e1..a5870e03a9 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -1,5 +1,7 @@ # Runtime +Shared parsing and streaming follow the [request-copy](transports/responses.md#request-copy-accounting) and [stream-buffer accounting](transports/responses.md#stream-buffer-accounting) contracts. + ## Entrypoints | Path | Responsibility | diff --git a/structure/subagents.md b/structure/subagents.md index a80af4d990..12b88f04ae 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -1,5 +1,7 @@ # Subagents And Multi-Agent Surface +Shared parsing and streaming follow the [request-copy](transports/responses.md#request-copy-accounting) and [stream-buffer accounting](transports/responses.md#stream-buffer-accounting) contracts. + ## Multi-agent surface mode (3-state) `OcxConfig.multiAgentMode` controls the `multi_agent_version` field stamped on catalog entries: diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index 7a7d2f9b4d..935b16fdfe 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -1,5 +1,7 @@ # Transport Inventory +Shared parsing and streaming follow the [request-copy](responses.md#request-copy-accounting) and [stream-buffer accounting](responses.md#stream-buffer-accounting) contracts. + ## Transport inventory The sections above cover the transports with load-bearing invariants. The rest of the transport diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 6d206195a3..f693b689b8 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -6,6 +6,36 @@ provider, lets the selected adapter speak the upstream protocol, then bridges adapter events back to Responses-compatible streaming output. +### Request-copy accounting + +`src/server/request-decompress.ts` observes the UTF-8 sizes of decoded text and reserialized JSON +without allocating encoded byte arrays solely to count them. Parsed-body accounting still uses +`JSON.stringify(parsed)`: numeric normalization can make it larger than the input text. These +observations retain the existing ownership and release lifecycle and do not consume the translator's +hard byte cap. Admission limits, parsing, compression, and error envelopes are unchanged. +`tests/usage/request-decompress.test.ts` covers exact accounting across codecs and Unicode/numeric +normalization, UTF-8 counting without encoded copies, and release after malformed or optional empty input. + +### Stream-buffer accounting + +`src/server/sse-payload-rewrite.ts` shares an incremental block buffer with native Chat. It scans +only new input, counts consumed blocks rather than remaining suffixes, and preserves LF/CRLF, +partial-event, injection/drop, and EOF behavior. Output admission precedes its single UTF-8 encoding; +failed enqueue and cancellation release the reservation without re-entering a disposed rewrite. +Old/new buffer overlap remains charged against the same translator cap. + +`src/adapters/openai-responses.ts` counts new compaction fragments, including surrogate pairs formed +across deltas, while retaining snapshot/done/delta precedence and existing terminal ownership. +Serialized request and buffered-response observations use byte counts without measurement arrays. +The same rule applies to Anthropic, Google, and Chat response accounting; serialization itself is +preserved where the existing metric is the serialized JSON size. + +`src/lib/translator-budget.ts` admits an event batch atomically from per-event serialized byte sizes +plus exact separators, without joining a second full JSON array. `src/lib/admission.ts` counts and +truncates diagnostic text at UTF-8 code-point boundaries without allocating arrays per character; +byte sizing retains TextEncoder's coercion behavior for legacy non-string runtime callers. +These optimizations do not add request queues, retry policies, or RSS-based admission gates. + ### Credential-bearing HTTP redirects Credential/body-bearing HTTP sends use `redirect: "manual"` at the final executor boundary, diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index 6a5574c6d2..e40fa788fd 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -1,7 +1,13 @@ # Streaming Health And WebSocket +Shared parsing and streaming follow the [request-copy](responses.md#request-copy-accounting) and [stream-buffer accounting](responses.md#stream-buffer-accounting) contracts. + ## Heartbeat and stall deadline +Native Chat uses the same resolved `stallTimeoutSec` with a pending-upstream-read allowance that +pauses under downstream backpressure. Its Chat error and cancellation contract is documented in +[native Chat completion lifecycle](../data-planes/inbound-compat.md#native-chat-completion-lifecycle). + The HTTP/SSE bridge emits an SSE comment-line keep-alive (`: opencodex heartbeat`) during upstream silence to re-arm Codex's idle timer (Codex's default `stream_idle_timeout` is 300 s and ANY SSE bytes re-arm it). A comment line is discarded by every eventsource parser without producing an event, diff --git a/tests/adapters/translator-budget.test.ts b/tests/adapters/translator-budget.test.ts index 8e9800899c..a15103c4bc 100644 --- a/tests/adapters/translator-budget.test.ts +++ b/tests/adapters/translator-budget.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; import { createAnthropicAdapter } from "../../src/adapters/anthropic"; import { createGoogleAdapter } from "../../src/adapters/google"; @@ -9,6 +9,7 @@ import { createTranslatorBudget, releaseTranslatedEvent, retainTranslatedEvent, + retainTranslatedEventBatch, translatorObservedBufferSnapshot, } from "../../src/lib/translator-budget"; import type { AdapterEvent } from "../../src/types"; @@ -22,6 +23,81 @@ async function textWithin(stream: ReadableStream, timeoutMs = 2_000) } describe("translator budget", () => { + for (const kind of ["anthropic", "google", "openai-chat"] as const) { + test(`${kind} buffered response sizing avoids encoded measurement copies`, async () => { + const text = "中文😀\ud800".repeat(1024); + const provider = { adapter: kind, apiKey: "fixture", baseUrl: "https://example.test/v1" }; + const adapter = kind === "anthropic" ? createAnthropicAdapter(provider) + : kind === "google" ? createGoogleAdapter(provider) + : createOpenAIChatAdapter(provider); + const payload = kind === "anthropic" + ? { content: [{ type: "text", text }], stop_reason: "end_turn" } + : kind === "google" + ? { candidates: [{ content: { parts: [{ text }] }, finishReason: "STOP" }] } + : { choices: [{ message: { content: text }, finish_reason: "stop" }] }; + const response = new Response(Buffer.from(JSON.stringify(payload))); + const budget = createTranslatorBudget(); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const events = await adapter.parseResponse(response, budget); + expect(events).toContainEqual({ type: "text_delta", text }); + expect(encode).not.toHaveBeenCalled(); + for (const event of events) releaseTranslatedEvent(event, budget); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + encode.mockRestore(); + budget.dispose(); + } + }); + } + + test("batch retention counts each event once without constructing a serialized batch", () => { + const events = [ + { type: "text_delta", text: "中文😀\ud800".repeat(1024) }, + { type: "done", usage: { inputTokens: 1e20, outputTokens: -0 } }, + ]; + const eventBytes = events.map(event => Buffer.byteLength(JSON.stringify(event))); + const total = Buffer.byteLength(JSON.stringify(events)); + const budget = createTranslatorBudget({ maxTurnBytes: total }); + const count = spyOn(Buffer, "byteLength"); + try { + retainTranslatedEventBatch(events, budget); + expect(count).toHaveBeenCalledTimes(events.length); + expect(budget.snapshot()).toMatchObject({ currentBytes: total, highWaterBytes: total, overflows: 0 }); + releaseTranslatedEvent(events[0]!, budget); + expect(budget.snapshot().currentBytes).toBe(eventBytes[1]! + 2); + releaseTranslatedEvent(events[1]!, budget); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + count.mockRestore(); + budget.dispose(); + } + }); + + test("batch overflow and serialization failure acquire no partial event ownership", () => { + const events = [{ type: "text_delta", text: "first" }, { type: "done" }]; + const bytes = Buffer.byteLength(JSON.stringify(events)); + const budget = createTranslatorBudget({ maxTurnBytes: bytes - 1 }); + try { + expect(() => retainTranslatedEventBatch(events, budget)).toThrow(/translator/); + expect(budget.snapshot().currentBytes).toBe(0); + for (const event of events) releaseTranslatedEvent(event, budget); + expect(budget.snapshot().currentBytes).toBe(0); + retainTranslatedEvent(events[0]!, budget); + releaseTranslatedEvent(events[0]!, budget); + expect(budget.snapshot().currentBytes).toBe(0); + + const invalid = { toJSON() { throw new Error("invalid event"); } }; + expect(() => retainTranslatedEventBatch([events[0]!, invalid], budget)).toThrow("invalid event"); + expect(budget.snapshot().currentBytes).toBe(0); + retainTranslatedEvent(events[0]!, budget); + releaseTranslatedEvent(events[0]!, budget); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + budget.dispose(); + } + }); + test("incremental event retention transfers array-tail ownership during in-order release", () => { const budget = createTranslatorBudget({ maxTurnBytes: 4_096 }); const first = { type: "text_delta", text: "first" }; diff --git a/tests/lib/debug.test.ts b/tests/lib/debug.test.ts index 6ce7359d2d..b80f7db41f 100644 --- a/tests/lib/debug.test.ts +++ b/tests/lib/debug.test.ts @@ -1,11 +1,78 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { appendDebugLogLine, debugBufferMetrics, getDebugLogEntries, resetDebugLogBufferForTests, subscribeDebugLogEntries } from "../../src/lib/debug-log-buffer"; -import { ResourceAdmissionError, RETAINED_TRUNCATION_MARKER, retainedUtf8Bytes } from "../../src/lib/admission"; +import { ResourceAdmissionError, RETAINED_TRUNCATION_MARKER, retainedUtf8Bytes, truncateRetainedUtf8 } from "../../src/lib/admission"; import { getInjectionDebugLogEntries, injectionDebugLog, resetInjectionDebugLogBufferForTests } from "../../src/lib/injection-debug-log"; import { markActivity, activityBreadcrumb } from "../../src/lib/sidecar-tracker"; import { debugDroppedFrame, debugProviderDiagnostic } from "../../src/lib/debug"; import { resetDebugSettingsForTests, setDebugSettings } from "../../src/lib/debug-settings"; +describe("retained UTF-8 sizing", () => { + test("preserves TextEncoder coercion for non-string runtime inputs", () => { + const encoder = new TextEncoder(); + const inputs: unknown[] = [ + undefined, null, true, 0, -0, 1e20, NaN, Infinity, 42n, + {}, ["中文", "\ud800"], new String("😀\udc00"), + new Uint8Array([1, 2]), Buffer.from([0xff]), + { [Symbol.toPrimitive](hint: string) { return hint === "string" ? "中\ud800" : 7; } }, + ]; + for (const value of inputs) { + const expected = Reflect.apply(encoder.encode, encoder, [value]).byteLength; + expect(retainedUtf8Bytes(value as string)).toBe(expected); + } + }); + + test("preserves TextEncoder rejection of Symbols and failed string coercion", () => { + const encoder = new TextEncoder(); + for (const value of [Symbol("input"), Object(Symbol("input")), Object.create(null)]) { + expect(() => Reflect.apply(encoder.encode, encoder, [value])).toThrow(TypeError); + expect(() => retainedUtf8Bytes(value as string)).toThrow(TypeError); + } + const failure = new Error("string conversion failed"); + const value = { toString() { throw failure; } }; + expect(() => Reflect.apply(encoder.encode, encoder, [value])).toThrow(failure); + expect(() => retainedUtf8Bytes(value as unknown as string)).toThrow(failure); + }); + + test("keeps UTF-8 and truncation boundaries for multibyte and unpaired surrogate text", () => { + const encoder = new TextEncoder(); + const markerBytes = encoder.encode(RETAINED_TRUNCATION_MARKER).byteLength; + const samples = ["", "plain", "é中😀", "\ud800x\udc00", "😀\ud800中éx".repeat(12)]; + for (const value of samples) { + const bytes = encoder.encode(value).byteLength; + expect(retainedUtf8Bytes(value)).toBe(bytes); + for (const cap of [0, 1, 2, 3, 4, markerBytes - 1, markerBytes, markerBytes + 1, markerBytes + 4, markerBytes + 7, bytes]) { + const prefix = (text: string, limit: number) => { + const points = Array.from(text); + let end = 0; + let size = 0; + while (end < points.length && size + encoder.encode(points[end]!).byteLength <= limit) { + size += encoder.encode(points[end]!).byteLength; + end += 1; + } + return points.slice(0, end).join(""); + }; + const expected = bytes <= cap ? value + : cap < markerBytes ? prefix(RETAINED_TRUNCATION_MARKER, cap) + : prefix(value, cap - markerBytes) + RETAINED_TRUNCATION_MARKER; + expect(truncateRetainedUtf8(value, cap)).toBe(expected); + } + } + }); + + test("truncates large diagnostics without per-character encoded arrays", () => { + const value = "x".repeat(1024 * 1024); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const result = truncateRetainedUtf8(value, 16 * 1024); + expect(result.endsWith(RETAINED_TRUNCATION_MARKER)).toBe(true); + expect(Buffer.byteLength(result)).toBe(16 * 1024); + expect(encode).not.toHaveBeenCalled(); + } finally { + encode.mockRestore(); + } + }); +}); + describe("debug frame logging", () => { const previous = process.env.OCX_DEBUG; diff --git a/tests/responses/chat-completions-endpoint.test.ts b/tests/responses/chat-completions-endpoint.test.ts index b622027a97..482575bb98 100644 --- a/tests/responses/chat-completions-endpoint.test.ts +++ b/tests/responses/chat-completions-endpoint.test.ts @@ -1,5 +1,4 @@ -import { waitForNativeMainStartupGate } from "../../src/codex/native-profile-startup"; -import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -1758,6 +1757,224 @@ test("chat-native request abort releases its active-turn lease and logs 499", as } }); +test("chat-native cancelled non-streaming SSE returns 499 instead of partial success", async () => { + const { handleChatCompletions } = await import("../../src/server/chat-completions"); + const clientAbort = new AbortController(); + let readStarted!: () => void; + const started = new Promise(resolve => { readStarted = resolve; }); + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"partial"}}]}\n\n')); + readStarted(); + }, + }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + const result = handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", signal: clientAbort.signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }), mockConfig("https://provider.example/v1"), {} as Parameters[2]); + await started; + await Bun.sleep(10); + clientAbort.abort("client done"); + const response = await result; + expect(response.status).toBe(499); + expect(await response.json()).toMatchObject({ error: { type: "client_cancelled" } }); +}); + +test("chat-native SSE enforces configured stall timeout despite non-progress frames", async () => { + const { handleChatCompletions } = await import("../../src/server/chat-completions"); + const { clearRequestLogsForTests, getRequestLogEntries } = await import("../../src/server/request-log"); + clearRequestLogsForTests(); + const clientAbort = new AbortController(); + let timer: ReturnType | undefined; + let cancels = 0; + const before = getActiveTurnCount(); + const releaseMisses = activeRegistryMetrics().activeTurns.releaseMisses; + const lease = tryAdmitTurn(); + if (!lease) throw new Error("failed to admit native Chat stall test turn"); + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + const wire = new TextEncoder().encode(': heartbeat\n\ndata\n\ndata:\n\ndata: {"choices":[{"delta":{"role":"assistant","content":"","tool_calls":[{"index":0}]}}],"usage":{"prompt_tokens":1,"completion_tokens":0}}\n\n'); + controller.enqueue(wire); + timer = setInterval(() => controller.enqueue(wire), 100); + }, + cancel() { cancels += 1; clearInterval(timer); }, + }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + try { + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", signal: clientAbort.signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: true, messages: [{ role: "user", content: "hi" }] }), + }), { ...mockConfig("https://provider.example/v1"), stallTimeoutSec: 1 }, {} as Parameters[2], + { requestId: "chat-stall-clock", start: Date.now(), turnAdmissionLease: lease }); + const outcome = await Promise.race([ + response.text(), Bun.sleep(1_600).then(() => "STILL_PENDING"), + ]); + expect(outcome).toContain('"code":"upstream_stall_timeout"'); + expect(outcome).not.toContain("[DONE]"); + clientAbort.abort("late cancellation"); + expect(cancels).toBe(1); + expect(getActiveTurnCount()).toBe(before); + expect(activeRegistryMetrics().activeTurns.releaseMisses).toBe(releaseMisses); + expect(getRequestLogEntries().filter(entry => entry.requestId === "chat-stall-clock")).toHaveLength(1); + expect(getRequestLogEntries().at(-1)?.status).toBe(502); + } finally { clientAbort.abort(); clearInterval(timer); lease.release(); } +}); + +test("chat-native meaningful reasoning keeps a long stream alive and pauses the stall clock under backpressure", async () => { + const { nativeChatSse } = await import("../../src/server/chat-native-sse"); + const budget = createTestTranslatorBudget(); + const abort = new AbortController(); + const encoder = new TextEncoder(); + let source!: ReadableStreamDefaultController; + const body = new ReadableStream({ start(controller) { source = controller; } }); + const stream = nativeChatSse(body, { + requestedModel: "mock/test-model", translatorBudget: budget, signal: abort.signal, + stallTimeoutSec: 1, onUsage() {}, + }); + const reasoning = 'data: {"choices":[{"delta":{"reasoning_content":"thinking"}}]}\n\n'; + source.enqueue(encoder.encode(reasoning)); + // One queued output applies backpressure. That interval is not upstream silence. + await Bun.sleep(1_100); + const output = new Response(stream).text(); + try { + for (let index = 0; index < 3; index++) { + await Bun.sleep(450); + const delta = index === 0 ? { reasoning: "thinking" } + : index === 1 ? { reasoning_details: [{ text: "thinking" }] } + : { reasoning_content: "thinking" }; + source.enqueue(encoder.encode(`data: ${JSON.stringify({ choices: [{ delta }] })}\n\n`)); + } + source.enqueue(encoder.encode('data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n')); + const text = await output; + expect(text.match(/thinking/g)).toHaveLength(4); + expect(text).toContain("[DONE]"); + expect(text).not.toContain("upstream_stall_timeout"); + } finally { abort.abort(); budget.dispose(); } +}); + +test("chat-native caller abort while the stall clock is pending wins over the later deadline", async () => { + const { nativeChatSse } = await import("../../src/server/chat-native-sse"); + const budget = createTestTranslatorBudget(); + const abort = new AbortController(); + const terminals: Array<[number, string | undefined]> = []; + let cancels = 0; + let upstreamCancels = 0; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"partial"}}]}\n\n')); + }, + cancel() { upstreamCancels += 1; }, + }); + const stream = nativeChatSse(body, { + requestedModel: "mock/test-model", translatorBudget: budget, signal: abort.signal, + stallTimeoutSec: 1, onUsage() {}, + onTerminal(status, message) { terminals.push([status, message]); }, + onCancel() { cancels += 1; }, + }); + const reader = stream.getReader(); + try { + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toContain("partial"); + // Upstream now stays silent, so the next pull waits on the one-second stall clock. + // A caller abort during that wait must be the only reported outcome, even once the + // deadline it was racing has elapsed. + const pending = reader.read(); + await Bun.sleep(300); + abort.abort("client gone"); + const next = await pending; + expect(next.done).toBe(true); + await Bun.sleep(1_000); + expect(cancels).toBe(1); + expect(upstreamCancels).toBe(1); + expect(terminals).toEqual([]); + } finally { abort.abort(); reader.releaseLock(); budget.dispose(); } +}); + +test("chat-native valid terminal retains precedence over a late non-streaming abort", async () => { + const { handleChatCompletions } = await import("../../src/server/chat-completions"); + const clientAbort = new AbortController(); + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"complete"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n')); + }, + cancel() { clientAbort.abort("late cancellation after terminal"); }, + }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", signal: clientAbort.signal, headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }), mockConfig("https://provider.example/v1"), {} as Parameters[2]); + expect(clientAbort.signal.aborted).toBe(true); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ choices: [{ message: { content: "complete" } }] }); +}); + +test("chat-native non-streaming SSE reports a typed stall failure instead of partial success", async () => { + const { handleChatCompletions } = await import("../../src/server/chat-completions"); + let cancels = 0; + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"partial"}}]}\n\n')); + }, + cancel() { cancels += 1; }, + }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }), { ...mockConfig("https://provider.example/v1"), stallTimeoutSec: 1 }, {} as Parameters[2]); + expect(response.status).toBe(502); + expect(await response.json()).toMatchObject({ error: { type: "upstream_error", code: "upstream_stall_timeout" } }); + expect(cancels).toBe(1); +}); + +test("chat-native SSE only encodes outbound frames and releases buffered tail on cancellation", async () => { + const { nativeChatSse } = await import("../../src/server/chat-native-sse"); + const frames = Array.from({ length: 64 }, () => 'data: {"choices":[{"delta":{"content":"中文😀"}}]}\n\n'); + const wire = new TextEncoder().encode(frames.join("")); + const budget = createTestTranslatorBudget(); + const abort = new AbortController(); + const body = new ReadableStream({ start(controller) { controller.enqueue(wire); } }); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const stream = nativeChatSse(body, { + requestedModel: "mock/test-model", translatorBudget: budget, signal: abort.signal, onUsage() {}, + }); + const reader = stream.getReader(); + for (let index = 0; index < 8; index++) expect((await reader.read()).done).toBe(false); + await reader.cancel("done inspecting"); + // No append/suffix-size encoding: every actual encoding is a normalized output frame. + expect(encode.mock.calls.length).toBeGreaterThanOrEqual(8); + for (const [text] of encode.mock.calls) expect(text).toContain('"object":"chat.completion.chunk"'); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { encode.mockRestore(); abort.abort(); budget.dispose(); } +}); + +test("chat-native non-streaming SSE collects CRLF multiline and split UTF-8 frames", async () => { + const { handleChatCompletions } = await import("../../src/server/chat-completions"); + const frames = [ + 'data\r\n\r\ndata:\r\n\r\n', + 'data: {"choices":\r\ndata\r\ndata: [{"delta":{"content":"中文😀"}}]}\r\n\r\n', + 'data: {"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":7,"completion_tokens":3}}\r\n\r\n', + 'data: [DONE]\r\n\r\n', + ]; + const wire = new TextEncoder().encode(frames.join("")); + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + for (let offset = 0; offset < wire.byteLength; offset += 7) controller.enqueue(wire.slice(offset, offset + 7)); + controller.close(); + }, + }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }), mockConfig("https://provider.example/v1"), {} as Parameters[2]); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + choices: [{ message: { content: "中文😀" }, finish_reason: "stop" }], + usage: { prompt_tokens: 7, completion_tokens: 3 }, + }); +}); + test("chat-native direct streaming without an admission lease does not record a release miss", async () => { const { handleChatCompletions } = await import("../../src/server/chat-completions"); const releaseMisses = activeRegistryMetrics().activeTurns.releaseMisses; @@ -2336,6 +2553,55 @@ test("collectChatCompletion releases every call scope after the final owner is c ); }); +test("collectChatCompletion accounts split surrogate content incrementally and releases it on failure", async () => { + const { collectChatCompletion } = await import("../../src/chat/outbound"); + for (const fail of [false, true]) { + const budget = createTestTranslatorBudget(); + const frames = [ + { choices: [{ delta: { content: "\ud83d", reasoning_content: "\ud83d", refusal: "\ud83d" } }] }, + { choices: [{ delta: { content: "\ude00", reasoning_content: "\ude00", refusal: "\ude00" } }] }, + ...(fail ? [{ error: { message: "upstream request failed", type: "upstream_error", code: "upstream_failure" } }] + : [{ choices: [{ delta: {}, finish_reason: "stop" }] }]), + ]; + const stream = new ReadableStream({ + start(controller) { + for (const frame of frames) controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(frame)}\n\n`)); + controller.close(); + }, + }); + try { + if (fail) { + await expect(collectChatCompletion(stream, "mock/test-model", budget)).rejects.toMatchObject({ code: "upstream_failure" }); + expect(budget.snapshot().currentBytes).toBe(0); + } else { + expect(await collectChatCompletion(stream, "mock/test-model", budget)).toMatchObject({ + choices: [{ message: { content: "😀", reasoning_content: "😀", refusal: "😀" } }], + }); + expect(budget.snapshot().currentBytes).toBe(12); + } + } finally { budget.dispose(); } + } +}); + +test("collectChatCompletion preserves admission for a 20 MiB frame and discards an incomplete EOF frame", async () => { + const { collectChatCompletion } = await import("../../src/chat/outbound"); + const text = "x".repeat(20 * 1024 * 1024); + const completeFrame = `data: ${JSON.stringify({ choices: [{ delta: { content: text }, finish_reason: "stop" }] })}\n\n`; + const incompleteFrame = 'data: {"choices":[{"delta":{"content":"discard me"}}]}'; + const budget = createTestTranslatorBudget(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(completeFrame)); + controller.enqueue(new TextEncoder().encode(incompleteFrame)); + controller.close(); + }, + }); + try { + expect(await collectChatCompletion(stream, "mock/test-model", budget)).toMatchObject({ choices: [{ message: { content: text } }] }); + expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(text)); + } finally { budget.dispose(); } +}); + test("collectChatCompletion final-copy overflow cleans up scopes and charges", async () => { const module = await import("../../src/chat/outbound"); // Args (100 bytes) fit; args + serialized copy exceed the turn cap, so the diff --git a/tests/responses/openai-responses-passthrough.test.ts b/tests/responses/openai-responses-passthrough.test.ts index d5a49fd39e..d99fdb9285 100644 --- a/tests/responses/openai-responses-passthrough.test.ts +++ b/tests/responses/openai-responses-passthrough.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; +import { Buffer } from "node:buffer"; import { createOpenAIChatAdapter } from "../../src/adapters/openai-chat"; import { createResponsesPassthroughAdapter as createResponsesPassthroughAdapterProduction } from "../../src/adapters/openai-responses"; import { openaiResponsesUrl } from "../../src/adapters/openai-responses-url"; @@ -19,7 +20,7 @@ import { SUMMARY_PREFIX, } from "../../src/responses/compaction"; import { createTranslatorBudget } from "../../src/lib/translator-budget"; -import type { OcxConfig } from "../../src/types"; +import type { AdapterEvent, OcxConfig } from "../../src/types"; import { withTestTranslatorBudget } from "../helpers/translator-budget"; import { restoreRoutedNamespaceCalls } from "../../src/responses/namespace-tool-compat"; import { restoreRoutedCustomCalls } from "../../src/responses/custom-tool-compat"; @@ -33,6 +34,195 @@ const provider = { authMode: "forward" as const, }; +describe("Responses request and compaction byte accounting", () => { + const encoder = new TextEncoder(); + const frame = (payload: unknown) => `data: ${JSON.stringify(payload)}\n\n`; + const adapter = () => createResponsesPassthroughAdapterProduction(provider); + + test("outbound accounting measures serialized UTF-8 without an encoded copy", () => { + const budget = createTranslatorBudget({ maxTurnBytes: 1 }); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const request = adapter().buildRequest({ + modelId: "example-model", context: { messages: [] }, stream: true, options: {}, + _rawBody: { model: "example-model", input: "中文😀é\ud800x\udc00", temperature: 1e20 }, + }, { headers: new Headers(), translatorBudget: budget }); + expect(encode).not.toHaveBeenCalled(); + encode.mockRestore(); + expect(budget.snapshot()).toMatchObject({ + currentBytes: encoder.encode(request.body).byteLength, overflows: 0, + }); + request.releaseBodyObservation?.(); + request.releaseBodyObservation?.(); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { encode.mockRestore(); budget.dispose(); } + }); + + test("buffered compaction preserves the serialized payload cap without an encoded copy", async () => { + const payload = { + output: [{ type: "message", content: [{ type: "output_text", text: "中文😀\ud800" }] }], + usage: { input_tokens: 1e20, output_tokens: 1, metadata: "\udc00" }, + }; + const wire = encoder.encode(JSON.stringify(payload)); + const budget = createTranslatorBudget({ maxTurnBytes: wire.byteLength }); + const response = new Response(wire); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const events = await adapter().parseResponse!(response, budget); + expect(events[0]).toEqual({ type: "text_delta", text: "中文😀\ud800" }); + expect(encode).not.toHaveBeenCalled(); + expect(budget.snapshot()).toMatchObject({ currentBytes: wire.byteLength, overflows: 0 }); + } finally { encode.mockRestore(); budget.dispose(); } + const limited = createTranslatorBudget({ maxTurnBytes: wire.byteLength - 1 }); + try { + await expect(adapter().parseResponse!(new Response(wire), limited)).rejects.toMatchObject({ + code: "translation_buffer_limit", + }); + } finally { limited.dispose(); } + }); + + test("compaction counts only new fragments without request-sized encoded arrays", async () => { + const count = 256; + const fragment = "x".repeat(1024); + const wire = encoder.encode(frame({ type: "response.output_text.delta", delta: fragment }).repeat(count) + + frame({ type: "response.completed", response: { output: [] } })); + const response = new Response(wire); + const budget = createTranslatorBudget(); + const originalByteLength = Buffer.byteLength; + let countedCodeUnits = 0; + const byteLength = spyOn(Buffer, "byteLength").mockImplementation((value, encoding) => { + if (typeof value === "string") countedCodeUnits += value.length; + return originalByteLength(value, encoding); + }); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(response, budget)) events.push(event); + expect(events.filter(event => event.type === "heartbeat")).toHaveLength(count); + expect(events.at(-2)).toEqual({ type: "text_delta", text: fragment.repeat(count) }); + expect(events.at(-1)).toEqual({ type: "done" }); + expect(budget.snapshot()).toMatchObject({ currentBytes: 0, overflows: 0 }); + // Replacing encode with byteLength alone still recounts all preceding deltas. + expect(countedCodeUnits).toBeLessThanOrEqual(count * fragment.length + 1024); + expect(encode).not.toHaveBeenCalled(); + } finally { byteLength.mockRestore(); encode.mockRestore(); budget.dispose(); } + }); + + test.each(["response.output_text.delta", "response.output_text.done"])( + "%s counts surrogate pairs joined across fragments exactly", async type => { + const fragments = ["中\ud83d", "", "\ude00", "\ud800", "x\udc00", "é"]; + let combined = ""; + const expectedBytes = fragments.map(fragment => encoder.encode(combined += fragment).byteLength); + const wire = encoder.encode(fragments.map(fragment => frame({ + type, [type.endsWith("delta") ? "delta" : "text"]: fragment, + })).join("")); + const budget = createTranslatorBudget(); + const reserve = spyOn(budget, "reserveTransient"); + try { + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(new Response(wire), budget)) events.push(event); + expect(reserve.mock.calls.filter(([, scope]) => scope.kind === "retained_collectors") + .map(([bytes]) => bytes)).toEqual(expectedBytes); + expect(events.at(-2)).toEqual({ type: "text_delta", text: combined }); + expect(events.at(-1)).toEqual({ type: "done" }); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { reserve.mockRestore(); budget.dispose(); } + }, + ); + + test("compaction replacement still admits the full old and new text overlap", async () => { + const fragment = "x".repeat(10_000); + const bytes = encoder.encode(frame({ type: "response.output_text.delta", delta: fragment })); + const budget = createTranslatorBudget({ maxTurnBytes: 35_000 }); + let sent = 0; + let cancelled = false; + const response = new Response(new ReadableStream({ + pull(controller) { + if (sent++ < 2) controller.enqueue(bytes); + else controller.close(); + }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 })); + const iterator = adapter().parseStream(response, budget); + try { + expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } }); + await expect(iterator.next()).rejects.toMatchObject({ code: "translation_buffer_limit" }); + expect(cancelled).toBe(true); + expect(budget.snapshot()).toMatchObject({ currentBytes: fragment.length, overflows: 1 }); + } finally { await iterator.return(undefined); budget.dispose(); } + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("terminal replacement releases old snapshot and usage while retaining the current ciphertext", async () => { + const oldUsage = { input_tokens: 1, output_tokens: 2, metadata: "中文\ud800" }; + const newUsage = { input_tokens: 3, output_tokens: 4, metadata: "😀\udc00" }; + const oldCiphertext = "old-中文"; + const ciphertext = "new-😀"; + const terminal = (text: string, encrypted: string, usage?: unknown) => ({ + type: "response.completed", response: { + output: [ + { type: "message", content: [{ type: "output_text", text }] }, + { type: "compaction", encrypted_content: encrypted }, + ], ...(usage === undefined ? {} : { usage }), + }, + }); + const wire = encoder.encode(frame({ type: "response.output_text.delta", delta: "partial" }) + + frame({ type: "response.output_text.done", text: "fallback" }) + + frame(terminal("old snapshot", oldCiphertext, oldUsage)) + + frame(terminal("new snapshot", ciphertext, newUsage)) + + frame(terminal("", ciphertext))); + const budget = createTranslatorBudget(); + const reserve = spyOn(budget, "reserveTransient"); + try { + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(new Response(wire), budget)) events.push(event); + expect(events).toEqual([ + { type: "heartbeat" }, { type: "text_delta", text: "fallback" }, + { type: "done", compactionEncryptedContent: ciphertext }, + ]); + const expected = [ + "partial", "fallback", oldCiphertext, "old snapshot", JSON.stringify(oldUsage), + ciphertext, "new snapshot", JSON.stringify(newUsage), ciphertext, "", + ].map(text => encoder.encode(text).byteLength); + expect(reserve.mock.calls.filter(([, scope]) => scope.kind === "retained_collectors") + .map(([bytes]) => bytes)).toEqual(expected); + expect(budget.snapshot().currentBytes).toBe(encoder.encode(ciphertext).byteLength); + } finally { reserve.mockRestore(); budget.dispose(); } + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test.each(["response.failed", "response.incomplete", "consumer return"])( + "%s leaves collector cleanup with the owning budget and cancels upstream", async ending => { + const partial = "中\ud800"; + const endingPayload = ending === "response.failed" + ? { type: ending, response: { error: { message: "stopped" } } } + : { type: ending, response: { incomplete_details: { reason: "stopped" } } }; + const bytes = encoder.encode(frame({ type: "response.output_text.delta", delta: partial }) + + (ending === "consumer return" ? "" : frame(endingPayload))); + const budget = createTranslatorBudget(); + let cancelled = false; + const response = new Response(new ReadableStream({ + start(controller) { controller.enqueue(bytes); }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 })); + const iterator = adapter().parseStream(response, budget); + try { + expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } }); + if (ending !== "consumer return") { + expect(await iterator.next()).toEqual({ done: false, value: ending === "response.failed" + ? { type: "error", message: "stopped" } : { type: "incomplete", reason: "stopped" } }); + expect(await iterator.next()).toEqual({ done: true, value: undefined }); + } else { + await iterator.return(undefined); + } + expect(cancelled).toBe(true); + expect(budget.snapshot().currentBytes).toBe(encoder.encode(partial).byteLength); + } finally { await iterator.return(undefined); budget.dispose(); } + expect(budget.snapshot().currentBytes).toBe(0); + }, + ); +}); + describe("native routed code-mode result visibility", () => { const routed = { adapter: "openai-responses", baseUrl: "https://api.x.ai/v1", authMode: "key" as const }; const exec = { type: "custom", name: "exec", description: "Run JavaScript in a V8 isolate." }; diff --git a/tests/responses/sse-payload-rewrite.test.ts b/tests/responses/sse-payload-rewrite.test.ts index 773665a054..8fbfa0e957 100644 --- a/tests/responses/sse-payload-rewrite.test.ts +++ b/tests/responses/sse-payload-rewrite.test.ts @@ -1,13 +1,19 @@ /** * Single-pass composition of client-facing SSE payload rewrites (#588 follow-up). */ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; +import { Buffer } from "node:buffer"; import { createImageGenCallRestoreRewrite } from "../../src/server/responses-image-gen-repair"; import { createResponsesItemIdPayloadRewrite } from "../../src/server/responses-item-id-repair"; import { composeSsePayloadRewrites, + composeSseBlockRewrites, + createSseBlockBuffer, + nextSseBlock, + replaceSseDataPayload, relaySseWithBlockRewrite, relaySseWithPayloadRewrite, + sseDataPayload, } from "../../src/server/sse-payload-rewrite"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; import { relaySseWithFailedTail } from "../../src/server/relay"; @@ -55,6 +61,274 @@ async function readAll(stream: ReadableStream): Promise { } describe("SSE payload rewrite composition", () => { + test.each(["\n", "\r\n"])("reads and replaces colonless data fields with %j lines", newline => { + const block = ["event: update", "data", 'data: {"text":"hello"}', "data", "database: unchanged"].join(newline); + expect(sseDataPayload(block)).toBe('\n{"text":"hello"}\n'); + expect(replaceSseDataPayload(block, '{"text":"changed"}')).toBe( + ["event: update", 'data: {"text":"changed"}', "database: unchanged"].join(newline), + ); + expect(sseDataPayload("data")).toBe(""); + expect(sseDataPayload("database")).toBeNull(); + }); + + test("encodes only delivered blocks and counts coalesced input in linear space", async () => { + const text = Array.from({ length: 256 }, (_, index) => `data: ${index} 中文😀${"x".repeat(128)}\n\n`).join(""); + const upstream = streamFromText(text); + const budget = createTestTranslatorBudget(); + const originalEncode = TextEncoder.prototype.encode; + const originalByteLength = Buffer.byteLength; + let encodedBytes = 0; + let countedCodeUnits = 0; + const encode = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (this: TextEncoder, input) { + const encoded = originalEncode.call(this, input); + encodedBytes += encoded.byteLength; + return encoded; + }); + const byteLength = spyOn(Buffer, "byteLength").mockImplementation((value, encoding) => { + if (typeof value === "string") countedCodeUnits += value.length; + return originalByteLength(value, encoding); + }); + try { + expect(await readAll(relaySseWithBlockRewrite(upstream, block => [block], budget))).toBe(text); + expect(encodedBytes).toBe(originalByteLength(text, "utf8")); + expect(countedCodeUnits).toBeLessThanOrEqual(text.length * 3); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + encode.mockRestore(); + byteLength.mockRestore(); + } + }); + + test("does not rescan an unterminated prefix after each transport fragment", async () => { + const fragments = Array.from({ length: 256 }, () => "x".repeat(128)); + fragments[0] = "data: " + fragments[0]; + fragments.push("\r\n\r\n"); + const originalMatch = String.prototype.match; + const originalIndexOf = String.prototype.indexOf; + let scannedCodeUnits = 0; + const match = spyOn(String.prototype, "match").mockImplementation(function (this: string, regexp) { + if (regexp instanceof RegExp && regexp.source === "\\r?\\n\\r?\\n") scannedCodeUnits += this.length; + return originalMatch.call(this, regexp); + }); + const indexOf = spyOn(String.prototype, "indexOf").mockImplementation(function (this: string, search, position) { + const found = originalIndexOf.call(this, search, position); + if (search === "\n") scannedCodeUnits += (found < 0 ? this.length : found + 1) - (position ?? 0); + return found; + }); + try { + const text = fragments.join(""); + expect(await readAll(relaySseWithBlockRewrite(streamFromTexts(fragments), block => [block], createTestTranslatorBudget()))).toBe(text); + expect(scannedCodeUnits).toBeLessThanOrEqual(text.length * 2); + } finally { + match.mockRestore(); + indexOf.mockRestore(); + } + }); + + test("rejects an oversized replacement before encoding or delivering it", async () => { + const upstream = streamFromText("data: small\n\n"); + const budget = createTestTranslatorBudget({ maxTurnBytes: 64 }); + const encode = spyOn(TextEncoder.prototype, "encode"); + let disposeCalls = 0; + const rewrite = Object.assign(() => [`data: ${"界".repeat(32)}`], { + dispose() { disposeCalls += 1; }, + }); + try { + await expect(readAll(relaySseWithBlockRewrite(upstream, rewrite, budget))).rejects.toMatchObject({ code: "translation_buffer_limit" }); + expect(encode).not.toHaveBeenCalled(); + expect(disposeCalls).toBe(1); + expect(budget.snapshot()).toMatchObject({ currentBytes: 0, overflows: 1 }); + } finally { + encode.mockRestore(); + } + }); + + test.each(["\n\n", "\r\n\r\n", "\r\n\n", "\n\r\n"])( + "preserves delimiter %j through byte-split Unicode, drops, injection, and EOF", + async delimiter => { + const first = "event: keep\r\ndata: 中文😀"; + const tail = "event: tail\r\ndata: é"; + const text = `${first}${delimiter}data: drop${delimiter}data: inject${delimiter}${tail}`; + const bytes = new TextEncoder().encode(text); + let index = 0; + let disposeCalls = 0; + const source = new ReadableStream({ + pull(controller) { + if (index === bytes.length) controller.close(); + else controller.enqueue(bytes.subarray(index, ++index)); + }, + }); + const rewrite = Object.assign((block: string) => { + if (block === "data: drop") return []; + if (block === "data: inject") return ["data: one", "data: two"]; + if (block === tail) return [tail, "data: end"]; + return [block]; + }, { dispose() { disposeCalls += 1; } }); + const budget = createTestTranslatorBudget(); + expect(await readAll(relaySseWithBlockRewrite(source, rewrite, budget))).toBe( + `${first}${delimiter}data: one${delimiter}data: two${delimiter}${tail}\r\n\r\ndata: end`, + ); + expect(budget.snapshot().currentBytes).toBe(0); + expect(disposeCalls).toBe(1); + }, + ); + + test("offset framing preserves exact suffix and overlap accounting through compaction", () => { + const budget = createTestTranslatorBudget(); + const buffer = createSseBlockBuffer(budget); + let reference = ""; + let peak = 0; + for (const fragment of ["data: 中文😀\r\n\r\ndata: é\n\npartial", "界\r", "\n", "\r", "\nlast\n\n", "\ud800", "\udc00\n\n"]) { + const previousBytes = Buffer.byteLength(reference, "utf8"); + reference += fragment; + peak = Math.max(peak, previousBytes + Buffer.byteLength(reference, "utf8")); + buffer.append(fragment); + for (;;) { + const expected = nextSseBlock(reference); + const beforeBytes = Buffer.byteLength(reference, "utf8"); + const actual = buffer.next(); + if (!expected) { + expect(actual).toBeNull(); + break; + } + expect(actual).toEqual({ block: expected.block, delimiter: expected.delimiter }); + reference = expected.rest; + peak = Math.max(peak, beforeBytes + Buffer.byteLength(reference, "utf8")); + // Native Chat yields after one event; HTTP rewriting drains the whole batch. + buffer.compact(); + expect(buffer.tail()).toBe(reference); + expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(reference, "utf8")); + } + expect(buffer.tail()).toBe(reference); + expect(budget.snapshot()).toMatchObject({ + currentBytes: Buffer.byteLength(reference, "utf8"), + highWaterBytes: peak, + overflows: 0, + }); + } + buffer.clear(); + buffer.clear(); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test.each(["append", "consume"] as const)("keeps old/new %s overlap admission atomic", operation => { + const initial = operation === "append" ? "data: partial" : "data: first\n\ndata: second\n\n"; + const budget = createTestTranslatorBudget({ maxTurnBytes: Buffer.byteLength(initial) + 1 }); + const buffer = createSseBlockBuffer(budget); + buffer.append(initial); + expect(() => operation === "append" ? buffer.append("x") : buffer.next()).toThrow("buffer exceeded"); + expect(buffer.tail()).toBe(initial); + expect(budget.snapshot()).toMatchObject({ currentBytes: Buffer.byteLength(initial), overflows: 1 }); + buffer.clear(); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("cancel during a pending read releases partial input and disposes once", async () => { + const waiting = Promise.withResolvers(); + const bytes = new TextEncoder().encode("data: partial 中文"); + let sent = false; + let cancelCalls = 0; + let disposeCalls = 0; + let rewriteCalls = 0; + const source = new ReadableStream({ + pull(controller) { + if (!sent) { + sent = true; + controller.enqueue(bytes); + } else waiting.resolve(); + }, + cancel() { cancelCalls += 1; }, + }); + const rewrite = Object.assign((block: string) => { + rewriteCalls += 1; + return [block]; + }, { dispose() { disposeCalls += 1; } }); + const budget = createTestTranslatorBudget(); + const reader = relaySseWithBlockRewrite(source, rewrite, budget).getReader(); + const pending = reader.read(); + await waiting.promise; + await reader.cancel("client left"); + expect(await pending).toMatchObject({ done: true }); + expect(rewriteCalls).toBe(0); + expect(cancelCalls).toBe(1); + expect(disposeCalls).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test.each([false, true])("cancel inside rewrite stops queued blocks and EOF rewrites (tail: %s)", async tail => { + const text = tail ? "data: unfinished" : "data: first\n\ndata: second\n\n"; + let disposeCalls = 0; + let rewriteCalls = 0; + let cancellation: Promise | undefined; + let reader: ReadableStreamDefaultReader; + const rewrite = Object.assign((block: string) => { + rewriteCalls += 1; + cancellation = reader.cancel("cancel during rewrite"); + return [block, "data: injected"]; + }, { dispose() { disposeCalls += 1; } }); + const budget = createTestTranslatorBudget(); + const encode = spyOn(TextEncoder.prototype, "encode"); + const upstream = streamFromText(text); + encode.mockClear(); + try { + reader = relaySseWithBlockRewrite(upstream, rewrite, budget).getReader(); + expect(await reader.read()).toMatchObject({ done: true }); + await cancellation; + expect(encode).not.toHaveBeenCalled(); + expect(rewriteCalls).toBe(1); + expect(disposeCalls).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + encode.mockRestore(); + } + }); + + test("releases the output reservation when enqueue fails after allocation", async () => { + const upstream = streamFromText("data: first\n\n"); + const budget = createTestTranslatorBudget(); + const originalEncode = TextEncoder.prototype.encode; + let reader: ReadableStreamDefaultReader; + let cancellation: Promise | undefined; + const encode = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (this: TextEncoder, text) { + const encoded = originalEncode.call(this, text); + // Close the downstream immediately before enqueue to exercise reservation cleanup. + cancellation = reader.cancel("closed before enqueue"); + return encoded; + }); + try { + reader = relaySseWithBlockRewrite(upstream, block => [block], budget).getReader(); + expect(await reader.read()).toMatchObject({ done: true }); + await cancellation; + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + encode.mockRestore(); + } + }); + + test("cancellation in one composed stage never invokes a disposed later stage", async () => { + let reader: ReadableStreamDefaultReader; + let cancellation: Promise | undefined; + let laterCalls = 0; + let disposeCalls = 0; + const rewrite = composeSseBlockRewrites( + block => { + cancellation = reader.cancel("cancel during first stage"); + return [block]; + }, + Object.assign((block: string) => { + laterCalls += 1; + return [block]; + }, { dispose() { disposeCalls += 1; } }), + ); + const budget = createTestTranslatorBudget(); + reader = relaySseWithBlockRewrite(streamFromText("data: event\n\n"), rewrite, budget).getReader(); + expect(await reader.read()).toMatchObject({ done: true }); + await cancellation; + expect(laterCalls).toBe(0); + expect(disposeCalls).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + }); + test("applies image-gen restore and item-id repair in one relay pass", async () => { const upstream = [ 'event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","id":"msg_0","role":"assistant"}}\n\n', diff --git a/tests/usage/request-decompress.test.ts b/tests/usage/request-decompress.test.ts index cb88dd29e1..e6c0fcfc56 100644 --- a/tests/usage/request-decompress.test.ts +++ b/tests/usage/request-decompress.test.ts @@ -1,5 +1,6 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { deflateRawSync, deflateSync } from "node:zlib"; +import { createTranslatorBudget } from "../../src/lib/translator-budget"; import { DecompressedBodyTooLargeError, decodeRequestBody, @@ -315,6 +316,77 @@ describe("configurable inbound body limit (Issue #3573)", () => { }); describe("readJsonRequestBody", () => { + const accountingBodies = [ + ["Unicode and escaped surrogates", new TextEncoder().encode('{"text":"中文😀é","escaped":"\\ud800\\udc00\\ud800x\\udc00"}')], + ["numeric normalization and duplicate keys", new TextEncoder().encode(' { "n": 1e20, "small": 1e-7, "zero": -0, "dup": 1, "dup": 2 } ')], + ["replacement decoding and BOM", Uint8Array.from([0xef, 0xbb, 0xbf, 0x22, 0xff, 0x22])], + ] as const; + + for (const encoding of ["identity", "zstd", "gzip", "deflate"] as const) { + for (const [label, decoded] of accountingBodies) { + test(`keeps exact request-copy accounting for ${encoding}: ${label}`, async () => { + const wire = encoding === "identity" ? decoded + : encoding === "zstd" ? Bun.zstdCompressSync(decoded) + : encoding === "gzip" ? Bun.gzipSync(decoded) + : deflateSync(decoded); + const text = new TextDecoder().decode(decoded); + const expected = JSON.parse(text); + const textBytes = new TextEncoder().encode(text).byteLength; + const parsedBytes = new TextEncoder().encode(JSON.stringify(expected)).byteLength; + // Accepted request copies remain observed even when they exceed the translator cap. + const budget = createTranslatorBudget({ maxTurnBytes: 1 }); + try { + const request = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-encoding": encoding, "content-length": String(wire.byteLength) }, + body: wire, + }); + expect(await readJsonRequestBody(request, budget)).toEqual(expected); + expect(budget.snapshot()).toMatchObject({ + currentBytes: parsedBytes, + highWaterBytes: wire.byteLength + (encoding === "identity" ? 0 : decoded.byteLength) + + textBytes + parsedBytes, + overflows: 0, + }); + } finally { + budget.dispose(); + } + expect(budget.snapshot().currentBytes).toBe(0); + }); + } + } + + test("request-copy accounting avoids allocating UTF-8 copies of the body", async () => { + const text = JSON.stringify({ input: "x".repeat(256 * 1024) }); + const wire = new TextEncoder().encode(text); + const request = new Request("http://localhost/v1/responses", { method: "POST", body: wire }); + const budget = createTranslatorBudget(); + const encode = spyOn(TextEncoder.prototype, "encode"); + try { + expect(await readJsonRequestBody(request, budget)).toEqual(JSON.parse(text)); + expect(budget.snapshot().currentBytes).toBe(wire.byteLength); + expect(encode).not.toHaveBeenCalled(); + } finally { + encode.mockRestore(); + budget.dispose(); + } + }); + + test("releases observed request copies after malformed JSON and empty-body fallback", async () => { + for (const text of ['{"input":', " \n"]) { + const budget = createTranslatorBudget(); + const request = new Request("http://localhost/v1/responses", { method: "POST", body: text }); + try { + const pending = readBoundedJsonRequestBody(request, 1024, budget, { emptyBodyFallback: null }); + if (text.trim() === "") expect(await pending).toBeNull(); + else await expect(pending).rejects.toBeInstanceOf(SyntaxError); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { + budget.dispose(); + } + } + }); + test("reports a compressed declaration without reading or echoing request metadata", async () => { const { body, stats } = trackedBodyStream([Bun.gzipSync(PAYLOAD_BYTES)]); const req = new Request("http://localhost/v1/responses/compact?private-query", {