fix(responses): expose bounded recovery refusal reasons

docs-site/src/content/docs/reference/architecture.md

@@ -225,3 +225,15 @@ using `stallTimeoutSec` (300 seconds by default). Nonempty chunks reset that
 deadline; a stalled body returns HTTP 504, client cancellation retains HTTP 499,
 and cleanup does not wait for a stuck upstream cancellation promise. The 32 MiB
 response ceiling and the original body bytes are preserved.
+
+A canonical upstream WebSocket refused-create error can become an HTTP 4xx only
+before the response is committed and after stream correlation checks. Permitted
+quota headers are bounded and rebuilt without upstream framing headers; the JSON
+response is not cacheable. Post-commit and 5xx errors keep the no-resend path.
+
+When encrypted agent-task recovery refuses a routed task, its existing 400 error
+can include a bounded `recovery_reason`: `unsupported_envelope`,
+`admission_denied`, `recovery_unavailable`, `caller_cancelled`, or `input_changed`.
+The field is omitted when no classified recovery result exists.
+`recovery_unavailable` includes cache/singleflight capacity and does not prove an
+upstream request was attempted. No retry or broader envelope acceptance is enabled.

src/server/responses/agent-task-recovery.ts

@@ -41,6 +41,18 @@ export interface AgentTaskRecoveryOptions {
   cacheEntries?: number;
 }
 
+export type AgentTaskRecoveryFailureReason =
+  | "unsupported_envelope"
+  | "admission_denied"
+  // Includes cache capacity rejection; does not imply an upstream request was attempted.
+  | "recovery_unavailable"
+  | "caller_cancelled"
+  | "input_changed";
+
+export type AgentTaskRecoveryResult =
+  | { readonly recovered: true }
+  | { readonly recovered: false; readonly reason: AgentTaskRecoveryFailureReason };
+
 export function agentTaskRecoveryConfig(config: OcxConfig): AgentTaskRecoveryOptions | null {
   const raw = config.agentTaskRecovery;
   if (!raw || raw.enabled !== true) return null;
@@ -275,16 +287,20 @@ interface AdmittedRecovery {
   cacheKey: string;
 }
 
+type RecoveryAdmissionResult =
+  | { admitted: true; recovery: AdmittedRecovery }
+  | { admitted: false; reason: "unsupported_envelope" | "admission_denied" };
+
 function admittedRecovery(
   req: Request,
   input: unknown,
   config: OcxConfig,
   parentThreadId?: string | null,
-): AdmittedRecovery | null {
+): RecoveryAdmissionResult {
   const envelope = findEnvelope(input);
-  if (!envelope) return null;
+  if (!envelope) return { admitted: false, reason: "unsupported_envelope" };
   const admission = recoveryAdmission(req, config);
-  if (!admission) return null;
+  if (!admission) return { admitted: false, reason: "admission_denied" };
   const cacheKey = createHash("sha256")
     .update(admission.cacheScope)
     .update("\0")
@@ -298,7 +314,7 @@ function admittedRecovery(
     .update("\0")
     .update(envelope.ciphertext)
     .digest("hex");
-  return { envelope, admission, cacheKey };
+  return { admitted: true, recovery: { envelope, admission, cacheKey } };
 }
 
 function recoveryPayload(envelope: AgentEnvelope, model: string): string {
@@ -465,23 +481,43 @@ export async function recoverEncryptedAgentTask(
   config: OcxConfig,
   context: { parentThreadId?: string | null; abortSignal?: AbortSignal } = {},
 ): Promise<boolean> {
+  return (await recoverEncryptedAgentTaskWithResult(req, input, options, config, context)).recovered;
+}
+
+/** Returns only bounded, caller-local diagnostics; no native error or payload content. */
+export async function recoverEncryptedAgentTaskWithResult(
+  req: Request,
+  input: unknown,
+  options: AgentTaskRecoveryOptions,
+  config: OcxConfig,
+  context: { parentThreadId?: string | null; abortSignal?: AbortSignal } = {},
+): Promise<AgentTaskRecoveryResult> {
   // Admission is deliberately checked before cache access. A cache hit must not
   // turn this process into a plaintext oracle for an unauthenticated caller.
   const admitted = admittedRecovery(req, input, config, context.parentThreadId);
-  if (!admitted) return false;
-  const { admission, cacheKey, envelope } = admitted;
+  if (!admitted.admitted) return { recovered: false, reason: admitted.reason };
+  const { admission, cacheKey, envelope } = admitted.recovery;
   const assignment = await resolveCachedAgentTaskRecovery(
     cacheKey,
     options.cacheEntries ?? 200,
     signal => requestRecovery(admission, envelope, options, signal),
     context.abortSignal,
   );
-  if (!assignment) return false;
-  if (context.abortSignal?.aborted || !injectAssignment(input, envelope, assignment)) {
+  if (!assignment) {
+    return {
+      recovered: false,
+      reason: context.abortSignal?.aborted ? "caller_cancelled" : "recovery_unavailable",
+    };
+  }
+  if (context.abortSignal?.aborted) {
     discardCachedAgentTaskRecovery(cacheKey);
-    return false;
+    return { recovered: false, reason: "caller_cancelled" };
   }
-  return true;
+  if (!injectAssignment(input, envelope, assignment)) {
+    discardCachedAgentTaskRecovery(cacheKey);
+    return { recovered: false, reason: "input_changed" };
+  }
+  return { recovered: true };
 }
 
 export function discardEncryptedAgentTaskRecovery(
@@ -491,7 +527,7 @@ export function discardEncryptedAgentTaskRecovery(
   context: { parentThreadId?: string | null } = {},
 ): void {
   const admitted = admittedRecovery(req, input, config, context.parentThreadId);
-  if (admitted) discardCachedAgentTaskRecovery(admitted.cacheKey);
+  if (admitted.admitted) discardCachedAgentTaskRecovery(admitted.recovery.cacheKey);
 }
 
 export function resetAgentTaskRecoveryState(): void {
@@ -510,9 +546,9 @@ export function restoreCachedEncryptedAgentTasks(
     const single = [item];
     // Revalidates caller credentials and the exact supported agent envelope before cache access.
     const admitted = admittedRecovery(req, single, config, context.parentThreadId);
-    if (!admitted) continue;
-    const assignment = cachedAgentTaskRecovery(admitted.cacheKey);
-    if (assignment && injectAssignment(single, admitted.envelope, assignment)) restored += 1;
+    if (!admitted.admitted) continue;
+    const assignment = cachedAgentTaskRecovery(admitted.recovery.cacheKey);
+    if (assignment && injectAssignment(single, admitted.recovery.envelope, assignment)) restored += 1;
   }
   return restored;
 }

src/server/responses/codex-ws-exchange.ts

@@ -1,4 +1,5 @@
 import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer";
+import { isSafeResponseHeader } from "../safe-response-headers";
 import { CodexWsMetadata, type CodexWsQuotaObserver } from "./codex-ws-metadata";
 import { CODEX_RESPONSES_HTTP_URL, type PreparedCodexWsRequest } from "./codex-ws-request";
 import { CodexWsCorrelation } from "./codex-ws-correlation";
@@ -16,6 +17,69 @@ interface ExchangeOptions {
   beforeDispatch?: (headers: Headers) => void;
 }
 
+const HTTP_HEADER_TOKEN = /^[!#$%&'*+.^_`|~0-9a-z-]+$/i;
+
+function record(value: unknown): value is Record<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
+}
+
+/** Rebuild only permitted metadata: upstream framing describes a different body. */
+function rejectionHeaders(source: Record<string, unknown>, prelude: Headers): Headers {
+  const connectionHeaders = new Set<string>();
+  for (const [name, value] of Object.entries(source)) {
+    if (name.toLowerCase() !== "connection" || typeof value !== "string") continue;
+    for (const token of value.split(",")) {
+      const lower = token.trim().toLowerCase();
+      if (HTTP_HEADER_TOKEN.test(lower)) connectionHeaders.add(lower);
+    }
+  }
+  // Reuse the metadata owner's count/value/family budgets and window freshness
+  // rules, without publishing quota twice. The unmarked HTTP response owns it.
+  const projected = new CodexWsMetadata();
+  try {
+    for (const values of [Object.fromEntries(prelude), source]) {
+      const headers = Object.fromEntries(Object.entries(values).filter(([name, value]) => {
+        if (!HTTP_HEADER_TOKEN.test(name) || !isSafeResponseHeader(name)
+          || connectionHeaders.has(name.toLowerCase())) return false;
+        if (typeof value !== "string" && typeof value !== "number" && typeof value !== "boolean") return false;
+        return !(typeof value === "number" && !Number.isFinite(value)) && !/[\r\n\0]/.test(String(value));
+      }));
+      if (Object.keys(headers).length === 0) continue;
+      const event = { type: "codex.response.metadata", headers };
+      // Bound the combined serialized seed and updates, even for replacements.
+      projected.consume(event, Buffer.byteLength(JSON.stringify(event)));
+    }
+    const headers = projected.snapshot();
+    headers.set("content-type", "application/json");
+    headers.set("cache-control", "no-store");
+    return headers;
+  } finally {
+    projected.finish();
+  }
+}
+
+/**
+ * Carry #3740's refused-create status back to the HTTP recovery path. Codex's
+ * responses_websocket.rs accepts status/status_code and scalar header values;
+ * unlike its native client, this relay converts only precommit 4xx. Returning a
+ * post-send 5xx or fetch rejection could cause the outer retry wrapper to resend.
+ */
+function wrappedRejectionResponse(payload: Record<string, unknown>, prelude: Headers): Response | null {
+  if (payload.type !== "error" || payload.stream_id !== undefined) return null;
+  // The native typed wrapper has one aliased field, not two competing statuses.
+  if (Object.hasOwn(payload, "status_code") && Object.hasOwn(payload, "status")) return null;
+  const status = Object.hasOwn(payload, "status_code") ? payload.status_code : payload.status;
+  if (typeof status !== "number" || !Number.isInteger(status) || status < 400 || status > 499) return null;
+  const error = payload.error;
+  if (error != null && (!record(error)
+    || [error.code, error.message].some(value => value != null && typeof value !== "string"))) return null;
+  if (payload.headers != null && !record(payload.headers)) return null;
+  const headers = rejectionHeaders(record(payload.headers) ? payload.headers : {}, prelude);
+  return new Response(JSON.stringify({
+    error: error ?? { type: "upstream_error", message: "Upstream rejected the request" },
+  }), { status, headers });
+}
+
 /** The sole SSE exchange state machine for both one-shot and retained sockets. */
 export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
   const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch } = options;
@@ -193,6 +257,21 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
       if (!controlFrame && !type.startsWith("response.") && type !== "error") return;
       if (!controlFrame) {
         try { correlation?.accept(normalized.payload); } catch (error) { failStream(error); return; }
+        // Correlation must run first: a reused socket's foreign-stream error
+        // must not become an HTTP refusal that could authorize account replay.
+        if (metadata && sent && !responseCommitted && type === "error") {
+          let rejection: Response | null;
+          try { rejection = wrappedRejectionResponse(normalized.payload, metadata.snapshot()); }
+          catch (error) { failStream(error); return; }
+          if (rejection) {
+            terminal = true;
+            cleanup();
+            try { controller.close(); } catch { /* unused stream already closed */ }
+            session.dispose();
+            resolve(rejection);
+            return;
+          }
+        }
         commitResponse();
       }
       const prefix = encoder.encode(`event: ${type}\ndata: `);

src/server/responses/core.ts

@@ -321,8 +321,9 @@ import {
 import {
   agentTaskRecoveryConfig,
   discardEncryptedAgentTaskRecovery,
-  recoverEncryptedAgentTask,
+  recoverEncryptedAgentTaskWithResult,
   restoreCachedEncryptedAgentTasks,
+  type AgentTaskRecoveryFailureReason,
 } from "./agent-task-recovery";
 import { relaySseEagerBounded } from "../relay-eager";
 import {
@@ -1933,13 +1934,14 @@ export const UPSTREAM_JSON_BODY_READ_OPTIONS = {
   firstByteTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS,
 };
 
-function unreadableEncryptedAgentTaskResponse(): Response {
+function unreadableEncryptedAgentTaskResponse(reason?: AgentTaskRecoveryFailureReason): Response {
   return new Response(
     JSON.stringify({
       error: {
         message: UNREADABLE_ENCRYPTED_AGENT_TASK_MESSAGE,
         type: "invalid_request_error",
         code: "unreadable_encrypted_agent_task",
+        ...(reason === undefined ? {} : { recovery_reason: reason }),
       },
     }),
     { status: 400, headers: { "Content-Type": "application/json" } },
@@ -2532,6 +2534,7 @@ export async function handleComboResponses(
   const payloadEligible = (target: (typeof combo.targets)[number]): boolean =>
     comboPayloadReadable || !unreadableEncryptedAgentTask || canDecryptUnreadableAgentTask(target);
   let encryptedTaskRecoveryAttempted = false;
+  let recoveryFailureReason: AgentTaskRecoveryFailureReason | undefined;
   let storedPool401ReplayDispatched = false;
   const recoverUnreadableEncryptedTask = async (): Promise<boolean> => {
     if (encryptedTaskRecoveryAttempted) return false;
@@ -2553,15 +2556,18 @@ export async function handleComboResponses(
     }
     let recovered = false;
     try {
-      recovered = await recoverEncryptedAgentTask(
+      const result = await recoverEncryptedAgentTaskWithResult(
         req,
         (body as { input?: unknown } | undefined)?.input,
         recovery,
         config,
         { parentThreadId: inboundClientThreadId, abortSignal: options.abortSignal },
       );
+      recovered = result.recovered;
+      recoveryFailureReason = result.recovered ? undefined : result.reason;
     } catch {
       recovered = false;
+      recoveryFailureReason = undefined;
     }
     // Recovery has the same in-place input mutation contract as the direct routed path.
     if (
@@ -2611,7 +2617,7 @@ export async function handleComboResponses(
     if (!(await recoverUnreadableEncryptedTask())) {
       return options.abortSignal?.aborted
         ? clientCancelledResponse()
-        : unreadableEncryptedAgentTaskResponse();
+        : unreadableEncryptedAgentTaskResponse(recoveryFailureReason);
     }
   }
 
@@ -3415,6 +3421,7 @@ async function handleResponsesInner(
     previewSelectionAdmission?.release();
   }
 
+  let recoveryFailureReason: AgentTaskRecoveryFailureReason | undefined;
   // Native fallback and explicitly trusted direct Responses routes can consume ciphertext,
   // so recover only after final route selection.
   if (
@@ -3433,15 +3440,18 @@ async function handleResponsesInner(
       (body as { input?: unknown } | undefined)?.input,
     );
     if (unreadableEncryptedAgentTask) try {
-      recovered = await recoverEncryptedAgentTask(
+      const result = await recoverEncryptedAgentTaskWithResult(
         req,
         (body as { input?: unknown } | undefined)?.input,
         agentTaskRecovery,
         config,
         { parentThreadId, abortSignal: options.abortSignal },
       );
+      recovered = result.recovered;
+      recoveryFailureReason = result.recovered ? undefined : result.reason;
     } catch {
       recovered = false;
+      recoveryFailureReason = undefined;
     }
     if (recovered) {
       unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask(
@@ -3565,7 +3575,7 @@ async function handleResponsesInner(
     && !finalRouteCanPassThroughEncryptedTask
     && unreadableEncryptedAgentTask
   ) {
-    return unreadableEncryptedAgentTaskResponse();
+    return unreadableEncryptedAgentTaskResponse(recoveryFailureReason);
   }
 
   // The canonical ChatGPT backend rejects previous_response_id, so a local replay miss leaves no

structure/04_transports-and-sidecars.md

@@ -1671,3 +1671,15 @@ using `stallTimeoutSec` (300 seconds by default). Nonempty chunks reset that
 deadline; a stalled body returns HTTP 504, client cancellation retains HTTP 499,
 and cleanup does not wait for a stuck upstream cancellation promise. The 32 MiB
 response ceiling and the original body bytes are preserved.
+
+A canonical upstream WebSocket refused-create error can become an HTTP 4xx only
+before the response is committed and after stream correlation checks. Permitted
+quota headers are bounded and rebuilt without upstream framing headers; the JSON
+response is not cacheable. Post-commit and 5xx errors keep the no-resend path.
+
+When encrypted agent-task recovery refuses a routed task, its existing 400 error
+can include a bounded `recovery_reason`: `unsupported_envelope`,
+`admission_denied`, `recovery_unavailable`, `caller_cancelled`, or `input_changed`.
+The field is omitted when no classified recovery result exists.
+`recovery_unavailable` includes cache/singleflight capacity and does not prove an
+upstream request was attempted. No retry or broader envelope acceptance is enabled.