fix(responses): gate post-header reset recovery on SSE protocol state

docs-site/src/content/docs/ko/reference/configuration/server.md

@@ -235,3 +235,13 @@ Anthropic OAuth 사이드카는 opencodex의 기존 Claude Code OAuth fingerprin
 ## Codex 할당량 네트워크 진단
 
 메인 Codex 계정 행의 `quotaRefresh`는 할당량 조회 결과를 분류하는 진단값입니다. 남은 할당량이나 모델 접근 권한을 뜻하지 않으며, 캐시를 쓰거나 조회하지 않았다면 생략될 수 있습니다. 요청은 명령을 입력한 터미널이 아니라 실행 중인 프록시 서비스의 환경을 따릅니다. `proxy`를 지정하지 않으면 기존 환경을 유지하고, `"auto"`는 시작할 때 Windows의 정적 프록시 설정만 읽습니다. PAC/WPAD, SOCKS 전용 설정과 실행 중 변경은 자동으로 반영하지 않습니다. TUN에서 성공했다고 HTTP 프록시 경로도 정상이라는 뜻은 아닙니다. 명령과 상태값은 [네트워크 진단(영문)](/reference/configuration/server/#codex-quota-network-diagnostics)에서 확인하세요.
+
+## 프로토콜 증거 기반 Responses 스트림 복구
+
+네이티브 HTTP Responses는 downstream 본문 소비 중 protocol inspection이 `response.created`를
+파싱하고 출력이나 도구 이벤트를 하나도 관찰하지 않은 상태에서 연결 재설정 read 오류를 받으면
+스트림을 한 번 교체할 수 있습니다. 응답 헤더는 첫 SSE 이벤트를 기다리지 않습니다. 교체 요청은
+남은 전송 한도와 같은 선택 자격 증명을 사용합니다.
+`response.created` 전, 출력 이후, 교체 스트림, WebSocket, 네이티브 Chat에서 발생한 reset은
+재전송하지 않습니다. 형식이 맞지 않거나 실패 상태인 교체 응답은 버리고 원본 스트림 오류를
+유지합니다. 이 동작은 `emptyCompletionRetry`와 별개입니다.

docs-site/src/content/docs/reference/configuration/server.md

@@ -615,3 +615,13 @@ WebSocket control paths. See the canonical guide for
 [supported steering routes and settings](../../guides/codex-integration.md#steering-continuation-settings-and-public-api),
 [typed result and approval continuations](../../guides/codex-integration.md#rich-tool-results-and-explicit-approvals-after-response-completion),
 and [confirmation deadlines and retained context](../../guides/codex-integration.md#steering-confirmation-deadlines-and-retained-context).
+
+## Protocol-gated Responses stream recovery
+
+Native HTTP Responses can replace one stream after response headers when protocol inspection during
+downstream body consumption has parsed `response.created`, has observed no output or tool event,
+and then receives a connection-reset read error. Response headers do not wait for the first SSE
+event. The replacement uses the request's remaining send allowance and the same selected
+credential. A reset before `response.created`, after any output, on the replacement, over WebSocket,
+or on native Chat is never replayed. An incompatible or unsuccessful replacement is discarded and
+the original stream failure is preserved. This behavior is independent of `emptyCompletionRetry`.

src/lib/upstream-retry.ts

@@ -8,13 +8,15 @@
  * becomes a terminal, non-replayable response unless the operation is explicitly safe.
  *
  * Deliberately narrow: timeouts, aborts, ECONNREFUSED/DNS/TLS failures, and HTTP error
- * statuses (returned as Response, never thrown) are NOT retried. Mid-stream SSE resets are
- * out of scope — the response has already resolved by then.
+ * statuses (returned as Response, never thrown) are NOT retried. The Responses transport
+ * separately permits one post-header replacement only after its eager SSE preflight has
+ * observed response.created and no protocol output event.
  *
  * MUST stay a leaf module: imports nothing from server.ts or adapters (kiro-retry imports
  * the shared abort helpers from here).
  */
 import { clearableDeadline } from "./abort";
+import { redactSecretString } from "./redact";
 
 /**
  * Responses the origin may already be executing. RFC 9110 §9.2.2 forbids an intermediary
@@ -661,3 +663,47 @@ export async function fetchWithTransientRetry(
     opts.onSendsConsumed?.(sent);
   }
 }
+
+export type ProtocolSafeRefetch = (
+  recovery?: UpstreamSendRecovery,
+  signal?: AbortSignal,
+) => Promise<Response>;
+
+export interface ProtocolSafeRefetchOptions extends ResetRetryOptions {
+  /** The replacement must match the response contract already selected for the client. */
+  acceptResponse?: (response: Response) => boolean;
+}
+
+/**
+ * Attempt one caller-authorized replacement after protocol inspection proved that no output
+ * event was observed. The caller owns that proof and the physical-send budget.
+ */
+export async function refetchAfterProtocolSafeReset(
+  doFetch: ProtocolSafeRefetch,
+  err: unknown,
+  opts: ProtocolSafeRefetchOptions = {},
+): Promise<Response | null> {
+  if (!isConnectionResetError(err) || opts.abortSignal?.aborted || opts.attempts === 0) return null;
+  const label = opts.label
+    ? " (" + redactSecretString(opts.label).replace(/[\r\n\u0000-\u001f\u007f]/g, "").slice(0, 128) + ")"
+    : "";
+  let replacement: Response;
+  try {
+    replacement = await doFetch("connection-reset", opts.abortSignal);
+  } catch {
+    console.warn("[upstream-retry] protocol-safe refetch failed" + label + "; preserving original stream error");
+    return null;
+  }
+  const body = replacement.body;
+  let accepted = !opts.abortSignal?.aborted && replacement.ok && body !== null
+    && !replacement.bodyUsed && !body.locked && !isNonReplayableResponse(replacement);
+  try { if (accepted && opts.acceptResponse) accepted = opts.acceptResponse(replacement); }
+  catch { accepted = false; }
+  if (!accepted || opts.abortSignal?.aborted || body?.locked) {
+    try { void body?.cancel().catch(() => {}); } catch { /* already locked or closed */ }
+    console.warn("[upstream-retry] protocol-safe refetch rejected" + label + "; preserving original stream error");
+    return null;
+  }
+  console.warn("[upstream-retry] pre-output Responses reset" + label + "; using one replacement stream");
+  return replacement;
+}

src/server/relay.ts

@@ -880,6 +880,8 @@ export type SseInspectorHandlers = {
    * with an empty `output`.
    */
   onParsedPayload?: (payload: unknown) => void;
+  /** A complete data payload that was not parsed as a JSON event, including [DONE]. */
+  onOpaquePayload?: () => void;
   onFirstOutput?: () => void;
   /**
    * Provider-scoped compatibility: persist the completed snapshot under the
@@ -1060,6 +1062,9 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
         /* malformed SSE payloads remain best-effort/no-throw */
       }
     }
+    if (parsed === undefined && handlers.onOpaquePayload) {
+      try { handlers.onOpaquePayload(); } catch { /* inspection must never throw into the pump */ }
+    }
     if (!reported && handlers.logCtx) {
       inspectResponseLogSsePayloadParsed(handlers.logCtx, payload, parsed);
     }

src/server/responses/combo-stream-preflight.ts

@@ -108,6 +108,13 @@ export function comboStreamPayloadCommitsOutput(payload: unknown): boolean {
   if (!payload || typeof payload !== "object" || Array.isArray(payload)) return true;
   const type = (payload as { type?: unknown }).type;
   if (typeof type !== "string") return true;
+  if (type === "response.created") {
+    const response = (payload as { response?: unknown }).response;
+    if (response && typeof response === "object" && !Array.isArray(response)) {
+      const output = (response as { output?: unknown }).output;
+      if (Array.isArray(output) && output.length > 0) return true;
+    }
+  }
   return !PRE_OUTPUT_CONTROL_EVENTS.has(type) && !TERMINAL_EVENTS.has(type);
 }
 
@@ -186,11 +193,13 @@ function failedTerminalResponse(
 
 export type ComboStreamPreflightResult =
   | { kind: "accepted"; response: Response }
-  | { kind: "failed"; response: Response };
+  | { kind: "failed"; response: Response }
+  | { kind: "read-error-before-output"; response: Response; error: unknown };
 
 /**
- * Buffer a combo child's downstream SSE only until the request becomes unsafe to
- * replay or reaches a terminal. This owns exactly one body reader. The aggregate
+ * Buffer a Responses SSE only until the request becomes unsafe to replay or reaches
+ * a terminal. Combo failover and native reset recovery share this protocol boundary.
+ * This owns exactly one body reader. The aggregate
  * buffer is capped by bytes and retained chunks; hitting either cap commits the
  * current target instead of growing memory or guessing that replay is safe.
  */
@@ -211,12 +220,16 @@ export async function preflightComboStreamResponse(
   const buffered: Uint8Array[] = [];
   let bufferedBytes = 0;
   let outputCommitted = false;
+  let responseCreated = false;
   let terminalStatus: ResponsesTerminalStatus | undefined;
   let retryableTerminalPayload: Record<string, unknown> | undefined;
   const inspector = createSseInspector({
     logCtx,
+    onOpaquePayload: () => { outputCommitted = true; },
     onParsedPayload: payload => {
       if (terminalStatus !== undefined || outputCommitted || retryableTerminalPayload) return;
+      if (payload !== null && typeof payload === "object" && !Array.isArray(payload)
+        && (payload as { type?: unknown }).type === "response.created") responseCreated = true;
       const retryable = retryableTerminal(payload);
       const matchedBareError = retryable && payload !== null && typeof payload === "object"
         && !Array.isArray(payload) && (payload as { type?: unknown }).type === "error";
@@ -239,7 +252,10 @@ export async function preflightComboStreamResponse(
         // The native relay still owns post-header transport failures. Preserve
         // the bounded prefix and the errored reader; cancelling it here would
         // erase the failure before either client relay or inspection sees it.
-        return { kind: "accepted", response: replayBufferedResponse(response, reader, buffered) };
+        const replay = replayBufferedResponse(response, reader, buffered);
+        return responseCreated && !outputCommitted && terminalStatus === undefined
+          ? { kind: "read-error-before-output", response: replay, error }
+          : { kind: "accepted", response: replay };
       }
       if (next.done) {
         inspector.finish();
@@ -277,3 +293,96 @@ export async function preflightComboStreamResponse(
     inspector.dispose();
   }
 }
+
+export type ProtocolSafeResetRecovery = (error: unknown) => Promise<Response | null>;
+
+/**
+ * Defer protocol inspection until the downstream actually pulls the body. Direct
+ * passthrough must return response headers before the first SSE event arrives;
+ * combo routing is the only caller that intentionally awaits this preflight.
+ */
+export function deferProtocolSafeResetRecovery(
+  response: Response,
+  logCtx: RequestLogContext,
+  recover: ProtocolSafeResetRecovery,
+  options?: { allowMissingContentType?: boolean },
+): Response {
+  if (!response.body) return response;
+
+  let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
+  let initialization: Promise<void> | undefined;
+  let closed = false;
+
+  const cancelBody = (body: ReadableStream<Uint8Array> | null, reason?: unknown): void => {
+    try { void body?.cancel(reason).catch(() => {}); } catch { /* already locked or closed */ }
+  };
+  const initialize = async (): Promise<void> => {
+    const preflight = await preflightComboStreamResponse(
+      response,
+      logCtx,
+      () => false,
+      { allowMissingContentType: options?.allowMissingContentType === true, replayReadErrors: true },
+    );
+    let selected = preflight.response;
+    if (preflight.kind === "read-error-before-output") {
+      const replacement = await recover(preflight.error);
+      if (replacement) {
+        cancelBody(selected.body, "using protocol-safe replacement stream");
+        selected = replacement;
+      }
+    }
+    if (closed) {
+      cancelBody(selected.body, "downstream cancelled before protocol preflight completed");
+      return;
+    }
+    reader = selected.body?.getReader();
+  };
+
+  const body = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        initialization ??= initialize();
+        await initialization;
+        if (closed) return;
+        if (!reader) {
+          closed = true;
+          controller.close();
+          return;
+        }
+        const next = await reader.read();
+        if (closed) return;
+        if (next.done) {
+          closed = true;
+          try { reader.releaseLock(); } catch { /* already released */ }
+          reader = undefined;
+          controller.close();
+          return;
+        }
+        controller.enqueue(next.value);
+      } catch (error) {
+        if (closed) return;
+        closed = true;
+        try { reader?.releaseLock(); } catch { /* errored reader */ }
+        reader = undefined;
+        controller.error(error);
+      }
+    },
+    cancel(reason) {
+      if (closed) return;
+      closed = true;
+      if (reader) {
+        try { void reader.cancel(reason).catch(() => {}); } catch { /* already closed */ }
+        try { reader.releaseLock(); } catch { /* already released */ }
+        reader = undefined;
+      } else {
+        cancelBody(response.body, reason);
+      }
+    },
+  }, { highWaterMark: 0 });
+
+  return new Response(body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  });
+}

src/server/responses/fetch-helpers.ts

@@ -112,6 +112,8 @@ export function sendWithConnectionPolicy(
 }
 
 export interface ProviderFetchOptions {
+  /** A replacement HTTP body must not initiate a fresh WebSocket exchange. */
+  httpOnly?: boolean;
   nativeControl?: NativeResponseControl;
   providerName?: string;
   modelId?: string;
@@ -164,7 +166,7 @@ export function providerFetch(
   // else keeps the provider's HTTP fetch. See ws-upstream.ts for the details.
   const unpaced = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
     const upstreamWebsocket = provider.upstreamWebsocket === true;
-    if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, upstreamWebsocket)) {
+    if (!options.httpOnly && typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, upstreamWebsocket)) {
       // The fallback has to be the same HTTP fetch the non-WS branch would have
       // used, protocol pin included: a WS turn that falls back is serving the
       // request over HTTP, and dropping the provider's `upstreamHttpVersion`

src/server/responses/passthrough-dispatch.ts

@@ -112,7 +112,9 @@ import {
   fetchWithTransientRetry,
   applyUpstreamRecoveryInit,
   isNonReplayableResponse,
- prepareSameTarget429Wait,
+  refetchAfterProtocolSafeReset,
+  TRANSIENT_RETRY_MAX_ATTEMPTS,
+  prepareSameTarget429Wait,
   sleepWithAbort,
 } from "../../lib/upstream-retry";
 import { mapCodexAuthContextErrorToResponse } from "./codex-auth-error";
@@ -149,7 +151,8 @@ import {
   reasoningEffortRejectionText,
 } from "./core-opaque-recovery";
 import type { RequestLogContext } from "../request-log";
-import { preflightComboStreamResponse } from "./combo-stream-preflight";
+import { deferProtocolSafeResetRecovery, preflightComboStreamResponse } from "./combo-stream-preflight";
+import { isCodexWsUpstreamResponse } from "./ws-upstream";
 import { upstreamErrorMessageFromPayload, ENCRYPTED_FUNCTION_OUTPUT_REJECTION } from "../../lib/errors";
 import { isTransientConsoleGoUploadRejection } from "../../providers/opencode-zen-rate-limit";
 import { planReasoningEffortDowngrade } from "../../providers/reasoning-metadata";
@@ -179,6 +182,8 @@ export async function preparePassthroughExchange(
     | "genericFailoverAccountId"
     | "passiveQuotaWriterGeneration"
     | "oauthDispatch"
+    | "selectionIsCurrent"
+    | "requestBindings"
     | "resolveSelectionAdapter"
     | "isOAuth401ReplayProvider"
     | "sentOAuthSnapshot"
@@ -1544,6 +1549,79 @@ export async function preparePassthroughExchange(
         continue passthroughRecovery;
       }
     }
+
+    const streamRecoveryContentType = upstreamResponse.headers.get("content-type")?.toLowerCase() ?? "";
+    const protocolRecoveryCandidate = upstreamResponse.ok
+      && !!upstreamResponse.body
+      && !isNonReplayableResponse(upstreamResponse)
+      && !isCodexWsUpstreamResponse(upstreamResponse)
+      // A downstream WebSocket turn that fell back to HTTP must relay response.created
+      // immediately so the client can address the turn and receive explicit control
+      // refusal. The recovery preflight must retain that event until output commits, so
+      // the two contracts cannot share one body owner.
+      && !(options.nativeControl && options.inboundTransport === "websocket")
+      && remainingTransientSendBudget(TRANSIENT_RETRY_MAX_ATTEMPTS) > 0
+      && (streamRecoveryContentType.includes("text/event-stream") || (!streamRecoveryContentType && parsed.stream));
+    if (protocolRecoveryCandidate) {
+      upstreamResponse = deferProtocolSafeResetRecovery(
+        upstreamResponse,
+        { model: logCtx.model, provider: logCtx.provider },
+        error => refetchAfterProtocolSafeReset(
+          (_recovery, signal = upstream.signal) => fetchWithTransientRetry(
+            () => fetchWithHeaderTimeout(
+              request.url,
+              applyUpstreamRecoveryInit({
+                method: request.method,
+                headers: request.headers,
+                body: request.body,
+              }, "connection-reset"),
+              signal,
+              connectMs,
+              true,
+              providerFetch(route.provider, options.codexWsRuntimeIdentity, {
+                httpOnly: true,
+                providerName: route.providerName,
+                modelId: route.modelId,
+                dispatchOverride: oauthDispatch(request),
+                beforeDispatch: headers => {
+                  if (signal.aborted) throw signal.reason;
+                  if (!transportState.selectionIsCurrent(transportState.requestBindings.get(request))) {
+                    throw new Error("Credential selection changed before pre-output stream recovery");
+                  }
+                  if (isCanonicalOpenAiForwardProvider(route.provider)) {
+                    createCodexReserveDispatchGuard(
+                      admissionState.authCtx,
+                      options.codexAuthPolicy ?? config,
+                      route.modelId,
+                      options.admission,
+                      options.visionDescribeTerminal === true,
+                    )?.(headers);
+                  }
+                  transportState.noteRoutedAttemptSend(passthroughEstimate, "connection-reset");
+                },
+              }),
+              route.provider.authMode === "forward",
+            ).then(adoptObservedResponse),
+            {
+              abortSignal: signal,
+              label: safeHostLabel(request.url),
+              attempts: Math.min(1, remainingTransientSendBudget(TRANSIENT_RETRY_MAX_ATTEMPTS)),
+              onSendsConsumed: noteTransientSends,
+            },
+          ),
+          error,
+          {
+            abortSignal: upstream.signal,
+            label: safeHostLabel(request.url),
+            acceptResponse: candidate => {
+              const type = candidate.headers.get("content-type")?.toLowerCase() ?? "";
+              return type.includes("text/event-stream") || (!type && parsed.stream);
+            },
+          },
+        ),
+        { allowMissingContentType: !streamRecoveryContentType && parsed.stream },
+      );
+    }
     break;
     }