fix(responses): normalize SSE terminal tails and policy failures

docs-site/src/content/docs/ja/reference/proxy-formats.md

@@ -55,6 +55,10 @@ provider events → internal adapter events → client dialect
 
 クライアント向け Responses SSE フレームは、SSE ブロック区切りの前の生バイトで測って 1 フレームあたり 4 MiB に制限されます。HTTP では、区切りなしでこの上限を超えたアップストリーム フレームは、合成 `response.failed` イベントと続く `data: [DONE]` でフェイルクローズします。Responses WebSocket ブリッジでは、同じ条件で 502 `websocket_protocol_error` を送信し、アップストリーム リーダーをキャンセルします。完全な Responses 終端フレームがすでに到着している場合はそれが優先され、その後のサイズ超過または不正なバイトは、完了したターンをトランスポート障害に置き換えず破棄されます。
 
+:::note
+ネイティブ パススルーでは、Responses の終端イベントが優先されます。早すぎる `data: [DONE]` は、そのイベントが届くまで保留されます。通常のネイティブ パスで、解析済みの終端がないまま正常な HTTP 200 EOF に達した場合、プロキシは `incomplete_details.reason: "adapter_eof"` を持つ `response.incomplete` を 1 件、その後に `data: [DONE]` を 1 件送信します。区切りのない終端 JSON が構文的に有効なら 1 回だけ受け入れられ、不正または切り詰められた JSON は incomplete のままです。モデル単位の終端修復を有効にしたプロバイダーでは、フレーム化されていない終端らしい接尾部と EOF 時の早すぎる `data: [DONE]` は、昇格可能な完全なライフサイクル候補がなければ `missing_terminal_event` としてフェイルクローズし、候補が完全なら `response.completed` に昇格します。高信頼度の `cyber_policy` 終端は、セマンティックなログおよび課金集計上は `error.code: "cyber_policy"` を持つ `response.failed`（status 400）に正規化されますが、すでに開始済みのストリーミング HTTP 応答は 200 のままです。このコミット済みリクエストの境界では、再試行も再送も行いません。
+:::
+
 すべての端末応答使用状況オブジェクトには、プロバイダーが詳細を報告しなかった場合でも、両方の詳細オブジェクトが含まれます。
 
 ```json

docs-site/src/content/docs/ko/reference/proxy-formats.md

@@ -65,6 +65,10 @@ deltas, 그리고 정확히 하나의 종료 `response.completed`, `response.fai
 
 클라이언트로 전달되는 Responses SSE 프레임은 SSE 블록 구분자 앞의 원시 바이트 기준으로 프레임당 4 MiB로 제한됩니다. HTTP에서는 구분자 없이 이 한도를 초과한 업스트림 프레임을 합성 `response.failed` 이벤트와 이어지는 `data: [DONE]`으로 fail closed 처리합니다. Responses WebSocket 브리지에서는 같은 조건에서 502 `websocket_protocol_error`를 보내고 업스트림 reader를 취소합니다. 완전한 Responses 종료 프레임이 이미 수신된 경우에는 그 종료가 우선하며, 이후의 과도한 크기 또는 잘못된 바이트는 완료된 턴을 전송 오류로 바꾸지 않고 버립니다.
 
+:::note
+네이티브 passthrough에서는 Responses 종료 이벤트가 우선합니다. 너무 이른 `data: [DONE]`은 해당 이벤트가 도착할 때까지 보류됩니다. 일반 네이티브 경로가 파싱된 종료 이벤트 없이 정상 HTTP 200 EOF에 도달하면, 프록시는 `incomplete_details.reason: "adapter_eof"`가 있는 `response.incomplete` 하나와 `data: [DONE]` 하나를 보냅니다. 구분자 없는 종료 JSON이 문법적으로 유효하면 정확히 한 번 받아들이고, 잘못되었거나 잘린 JSON은 incomplete로 남습니다. 모델별 종료 복구를 사용하도록 설정된 공급자에서는 프레임이 없는 종료 유사 suffix와 EOF의 너무 이른 `data: [DONE]`을, 승격할 수 있는 완전한 lifecycle 후보가 없을 때 `missing_terminal_event`로 fail closed 처리하며, 완전한 후보가 있으면 `response.completed`로 승격합니다. 신뢰도가 높은 `cyber_policy` 종료 형식은 의미론적 로깅 및 집계에서 `error.code: "cyber_policy"`가 있는 `response.failed`(status 400)로 정규화되지만, 이미 시작된 스트리밍 HTTP 응답은 200을 유지합니다. 이 커밋된 요청 경계에서는 재시도하거나 재전송하지 않습니다.
+:::
+
 canonical ChatGPT forward streaming은 stable Bun 1.4.0 이상에서 Codex 업스트림 WebSocket을
 투명하게 사용할 수 있습니다. 번들 Bun 1.3.14, prerelease, 또는 검증 불가능한 런타임 identity는
 HTTP/SSE를 사용합니다. 업스트림 WS adapter는 같은 downstream SSE 계약을 유지하며, 원시 JSON

docs-site/src/content/docs/reference/proxy-formats.md

@@ -72,6 +72,22 @@ bridge, the same condition emits a 502 `websocket_protocol_error` and cancels th
 A complete Responses terminal frame is authoritative: oversized or malformed trailing bytes after
 that terminal are dropped rather than replacing the completed turn with a transport failure.
 
+:::note
+For native passthrough, a Responses terminal event is authoritative. A premature `data: [DONE]` is
+held until that event. On the ordinary native path, a clean HTTP 200 EOF without a parsed terminal
+emits one `response.incomplete` with `incomplete_details.reason: "adapter_eof"`, followed by one
+`data: [DONE]`; syntactically valid delimiter-less terminal JSON is accepted exactly once, while
+malformed or truncated JSON remains incomplete. For providers opted into model-scoped terminal
+repair, unframed terminal-like suffixes and a premature `data: [DONE]` at EOF fail closed with
+`missing_terminal_event` when no complete lifecycle candidate can be promoted; a complete candidate
+is promoted to `response.completed`. High-confidence `cyber_policy`
+terminal shapes normalize to `response.failed` with `error.code: "cyber_policy"` for semantic
+logging/accounting (status 400), while an already-started streamed HTTP response remains 200. This
+committed-request boundary does not retry or replay and does not resolve
+[#2423](https://github.com/lidge-jun/opencodex/issues/2423) or
+[#2486](https://github.com/lidge-jun/opencodex/issues/2486).
+:::
+
 For canonical ChatGPT forward streaming, stable Bun 1.4.0 or newer may transparently use
 Codex's upstream WebSocket transport. Bundled Bun 1.3.14, prereleases, and unverifiable runtime
 identities use HTTP/SSE. The upstream WS adapter keeps the same downstream SSE contract, caps both

docs-site/src/content/docs/ru/reference/proxy-formats.md

@@ -69,6 +69,10 @@ Responses. Обе формы сохраняют выбранную модель,
 
 Клиентские frame'ы Responses SSE ограничены 4 MiB на frame, считая сырые байты до разделителя SSE-блока. В HTTP незавершённый upstream-frame, превысивший этот предел, завершается fail-closed синтетическим событием `response.failed`, после которого идёт `data: [DONE]`. В мосте Responses WebSocket то же условие даёт 502 `websocket_protocol_error` и отменяет upstream-reader. Если полноценный terminal-frame Responses уже получен, он остаётся авторитетным: слишком большие или некорректные байты после него отбрасываются и не заменяют завершённый ход транспортной ошибкой.
 
+:::note
+При нативном passthrough терминальное событие Responses является авторитетным, а преждевременный `data: [DONE]` удерживается до его появления. Если обычный нативный путь достигает корректного HTTP 200 EOF без распознанного терминального события, прокси испускает один `response.incomplete` с `incomplete_details.reason: "adapter_eof"`, а затем один `data: [DONE]`. Синтаксически корректный терминальный JSON без разделителя принимается ровно один раз; некорректный или обрезанный JSON остаётся incomplete. Для провайдеров с включённым model-scoped terminal repair неоформленный terminal-like suffix и преждевременный `data: [DONE]` на EOF завершаются fail-closed с `missing_terminal_event`, если нет полного lifecycle-кандидата для повышения; полный кандидат повышается до `response.completed`. Терминальные формы `cyber_policy` с высокой уверенностью нормализуются для семантического журналирования и учёта в `response.failed` с `error.code: "cyber_policy"` (status 400), но уже начатый потоковый HTTP-ответ сохраняет статус 200. На этой границе уже отправленного запроса нет retry или replay.
+:::
+
 Каждый terminal usage-объект Responses всегда включает оба detail-объекта, даже если провайдер их
 не сообщил:
 

docs-site/src/content/docs/zh-cn/reference/proxy-formats.md

@@ -64,6 +64,10 @@ Responses 表示是这座桥的中心。原生兼容的路由可以跳过部分
 
 面向客户端的 Responses SSE 帧按 SSE 块分隔符之前的原始字节计算，每帧限制为 4 MiB。对于 HTTP，未终止的上游帧一旦超过该限制，会以合成的 `response.failed` 事件并随后发送 `data: [DONE]` 的方式 fail closed。对于 Responses WebSocket 桥，相同情况会发送 502 `websocket_protocol_error` 并取消上游 reader。已经完整到达的 Responses 终止帧具有优先权；其后的超大或格式错误字节会被丢弃，而不会把已经完成的轮次替换为传输失败。
 
+:::note
+对于原生透传，Responses 终止事件具有最高优先级；过早出现的 `data: [DONE]` 会被保留，直到该事件到达。普通原生路径在没有已解析终止事件的情况下正常到达 HTTP 200 EOF 时，代理会发送一个带有 `incomplete_details.reason: "adapter_eof"` 的 `response.incomplete`，随后发送一个 `data: [DONE]`。语法有效但缺少分隔符的终止 JSON 只会被接受一次；格式错误或被截断的 JSON 仍保持 incomplete。对于启用了按模型终止修复的提供方，未成帧但形似终止事件的后缀和 EOF 处过早出现的 `data: [DONE]`，会在没有可提升的完整生命周期候选时以 `missing_terminal_event` 的形式 fail closed；完整候选则会被提升为 `response.completed`。高置信度的 `cyber_policy` 终止形态会在语义日志和计量中规范化为带有 `error.code: "cyber_policy"` 的 `response.failed`（status 400），但已经开始的流式 HTTP 响应仍保持 200。这个已提交请求的边界不会重试或重放请求。
+:::
+
 每个终止的 Responses usage 对象都包含两个 detail 对象，即使提供方没有报告这些细节：
 
 ```json

src/bridge.ts

@@ -7,7 +7,15 @@ import type {
   OcxUsage,
 } from "./types";
 import { coerceIntegerToolArguments } from "./lib/tool-argument-integers";
-import { adapterFailureFromMessage, classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, type OcxErrorPayload } from "./lib/errors";
+import {
+  adapterFailureFromMessage,
+  classifyError,
+  cyberPolicyErrorType,
+  CYBER_POLICY_ERROR_CODE,
+  isCyberPolicyCode,
+  type OcxErrorPayload,
+} from "./lib/errors";
+import { redactSecretString } from "./lib/redact";
 import { repairFreeformToolInput } from "./responses/apply-patch-envelope";
 import { encodeCompactionSummary } from "./responses/compaction";
 import { isTruncatedStopReason, truncationReasonFor } from "./responses/truncated-stop-reason";
@@ -114,18 +122,19 @@ function toolCallArgumentsUsable(args: string): boolean {
 }
 
 function adapterFailureFromEvent(event: Extract<AdapterEvent, { type: "error" }>): { httpStatus: number; error: OcxErrorPayload } {
+  const message = redactSecretString(event.message);
   if (event.status === undefined && event.errorType === undefined && event.code === undefined) {
-    return adapterFailureFromMessage(event.message);
+    return adapterFailureFromMessage(message);
   }
-  const fallback = adapterFailureFromMessage(event.message);
+  const fallback = adapterFailureFromMessage(message);
   let httpStatus = event.status ?? fallback.httpStatus;
-  const error = classifyError(httpStatus, event.errorType ?? fallback.error.type, event.message);
+  const error = classifyError(httpStatus, event.errorType ?? fallback.error.type, message);
   if (event.errorType !== undefined) error.type = event.errorType;
   if (event.code !== undefined) error.code = event.code;
   // Codex maps cyber_policy on HTTP 400 (body) or mid-stream code; never leave it as 502.
   if (isCyberPolicyCode(error.code) || isCyberPolicyCode(event.code)) {
     error.code = CYBER_POLICY_ERROR_CODE;
-    error.type = "invalid_request_error";
+    error.type = cyberPolicyErrorType(event.errorType);
     httpStatus = 400;
   }
   return { httpStatus, error };
@@ -1296,7 +1305,9 @@ export function bridgeToResponsesSSE(
                   ...(event.usage ? { usage: responsesUsage(event.usage) } : {}),
                   error: failure.error,
                   last_error: failure.error,
-                  ...(event.retryable !== undefined ? { retryable: event.retryable } : {}),
+                  ...(isCyberPolicyCode(failure.error.code)
+                    ? { retryable: false }
+                    : event.retryable !== undefined ? { retryable: event.retryable } : {}),
                 },
               });
               reportTerminal("failed");
@@ -1320,11 +1331,17 @@ export function bridgeToResponsesSSE(
           if (currentToolCall) failCurrentToolCall();
           if (currentWebSearch) closeCurrentWebSearch("failed", []);
           releasePendingWebSources();
+          const failure = responseError(
+            500,
+            "proxy_error",
+            redactSecretString(err instanceof Error ? err.message : String(err)),
+          );
           emit("response.failed", {
             response: {
               ...responseSnapshot("failed", finishedItems),
-              error: responseError(500, "proxy_error", err instanceof Error ? err.message : String(err)),
-              last_error: responseError(500, "proxy_error", err instanceof Error ? err.message : String(err)),
+              error: failure,
+              last_error: failure,
+              ...(isCyberPolicyCode(failure.code) ? { retryable: false } : {}),
             },
           });
           reportTerminal("failed");
@@ -1952,7 +1969,9 @@ function buildResponseJSONWithBudget(
     model: modelId, output,
     ...(endTurn !== undefined ? { end_turn: endTurn } : {}),
     ...(failure ? { error: failure.error, last_error: failure.error } : {}),
-    ...(errorEvent?.retryable !== undefined ? { retryable: errorEvent.retryable } : {}),
+    ...(failure && isCyberPolicyCode(failure.error.code)
+      ? { retryable: false }
+      : errorEvent?.retryable !== undefined ? { retryable: errorEvent.retryable } : {}),
     ...(incompleteEvent ? {
       incomplete_details: {
         reason: incompleteEvent.reason,
@@ -1981,12 +2000,15 @@ export function formatErrorResponse(
   const error = classifyError(status, type, message);
   if (isCyberPolicyCode(options?.code)) {
     error.code = CYBER_POLICY_ERROR_CODE;
-    error.type = "invalid_request_error";
+    error.type = cyberPolicyErrorType(type);
   }
   const finalStatus = error.code === CYBER_POLICY_ERROR_CODE ? 400 : status;
   const headers = new Headers({ "Content-Type": "application/json" });
   const retryAfter = options?.retryAfter?.trim();
-  if (retryAfter && retryAfter.length > 0 && retryAfter.length <= 128) {
+  if (error.code !== CYBER_POLICY_ERROR_CODE
+    && retryAfter
+    && retryAfter.length > 0
+    && retryAfter.length <= 128) {
     headers.set("Retry-After", retryAfter);
   }
   return new Response(JSON.stringify({ error }), {

src/chat/outbound.ts

@@ -9,7 +9,14 @@ type Rec = Record<string, unknown>;
 
 import { decodeServerSentEvents, sseFieldValue } from "../lib/sse-decoder";
 import { isTranslatorBudgetExceededError, type TranslatorBudget } from "../lib/translator-budget";
-import { classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, isCyberPolicyMessage } from "../lib/errors";
+import {
+  classifyError,
+  cyberPolicyErrorType,
+  CYBER_POLICY_ERROR_CODE,
+  isCyberPolicyCode,
+  isCyberPolicyMessage,
+} from "../lib/errors";
+import { redactSecretString } from "../lib/redact";
 
 function isRec(v: unknown): v is Rec {
   return !!v && typeof v === "object" && !Array.isArray(v);
@@ -48,14 +55,14 @@ export function chatCompletionsUsage(usage: unknown): Rec {
 export function chatCompletionsErrorBody(
   status: number,
   message: string,
-  type = "invalid_request_error",
+  type?: string,
   code?: string | null,
 ): Rec {
   if (isCyberPolicyCode(code) || isCyberPolicyMessage(message)) {
     return {
       error: {
         message,
-        type: "invalid_request_error",
+        type: cyberPolicyErrorType(type),
         param: null,
         code: CYBER_POLICY_ERROR_CODE,
       },
@@ -64,7 +71,7 @@ export function chatCompletionsErrorBody(
   return {
     error: {
       message,
-      type,
+      type: type ?? "invalid_request_error",
       param: null,
       code: code !== undefined
         ? code
@@ -311,8 +318,9 @@ export function responsesSseToChatCompletionsSse(
         // Deliver the error frame then close the stream abnormally (no [DONE]).
         // Do not controller.error() — that can drop already-enqueued bytes from consumers
         // like response.text().
-        const statusHint = details?.status ?? streamErrorStatus(message);
-        const classified = classifyError(statusHint, details?.type ?? "upstream_error", message);
+        const safeMessage = redactSecretString(message);
+        const statusHint = details?.status ?? streamErrorStatus(safeMessage);
+        const classified = classifyError(statusHint, details?.type ?? "upstream_error", safeMessage);
         const translatorOverflow = details?.code === "translation_buffer_limit";
         if (translatorOverflow) {
           upstreamAbort.abort(new Error("upstream translation buffer exceeded the safe limit"));
@@ -324,7 +332,7 @@ export function responsesSseToChatCompletionsSse(
           classified.type = "upstream_error";
         } else if (isCyberPolicyCode(details?.code) || classified.code === CYBER_POLICY_ERROR_CODE) {
           classified.code = CYBER_POLICY_ERROR_CODE;
-          classified.type = "invalid_request_error";
+          classified.type = cyberPolicyErrorType(details?.type);
         } else if (details?.code !== undefined && details.code !== null && !classified.code) {
           classified.code = details.code;
         }

src/lab/fabric/producer-child.ts

@@ -128,7 +128,7 @@ async function main(): Promise<void> {
   writeLine({ type: "result", patch });
 }
 
-main().catch((error: unknown) => {
+await main().catch((error: unknown) => {
   writeLine({
     type: "error",
     code: "harness_failure",