diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md
index b5fe8db9ac..8e1e361a81 100644
--- a/docs-site/src/content/docs/reference/architecture.md
+++ b/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.
diff --git a/src/server/responses/agent-task-recovery.ts b/src/server/responses/agent-task-recovery.ts
index 89af59a1ef..22b7a4e66b 100644
--- a/src/server/responses/agent-task-recovery.ts
+++ b/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;
 }
diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts
index 2f41be02fa..31813756ea 100644
--- a/src/server/responses/codex-ws-exchange.ts
+++ b/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: `);
diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts
index 526010f31e..3c539c6d8e 100644
--- a/src/server/responses/core.ts
+++ b/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
diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md
index a20eda8ca6..5babf5f361 100644
--- a/structure/04_transports-and-sidecars.md
+++ b/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.
diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts
index cfb087a4bb..3ae551e63d 100644
--- a/tests/responses/ws-upstream.test.ts
+++ b/tests/responses/ws-upstream.test.ts
@@ -3,6 +3,10 @@ import { providerFetch } from "../../src/server/responses/fetch-helpers";
 import { handleResponses } from "../../src/server/responses";
 import { isEagerRelaySseResponse } from "../../src/server/relay";
 import { isWin32EagerRewrite } from "../../src/lib/bun-stream-caps";
+import { fetchWithTransientRetry } from "../../src/lib/upstream-retry";
+import { codexWsExchange } from "../../src/server/responses/codex-ws-exchange";
+import { CodexWsSession } from "../../src/server/responses/codex-ws-session";
+import { prepareCodexWsRequest } from "../../src/server/responses/codex-ws-request";
 import { CodexWsMetadata, CODEX_WS_METADATA_MAX_BYTES, CODEX_WS_METADATA_MAX_VALUE_BYTES } from "../../src/server/responses/codex-ws-metadata";
 import {
   bunSupportsBoundedCodexWsRelay,
@@ -11,6 +15,7 @@ import {
   codexWsUpstreamFetch as rawCodexWsUpstreamFetch,
   currentBunRuntimeIdentity,
   isCodexWsUpstreamResponse,
+  isCodexWsQuotaObservedResponse,
   MAX_CODEX_WS_CREATE_FRAME_BYTES,
   MAX_CODEX_WS_FRAME_BYTES,
   MAX_CODEX_WS_QUEUE_BYTES,
@@ -637,9 +642,283 @@ describe("codexWsUpstreamFetch", () => {
     expect(FakeWebSocket.instances[0].closed).toBe(true);
   });
 
+  describe("wrapped create refusals", () => {
+    const refusal = { type: "error", status_code: 429, error: {
+      type: "usage_limit_reached", message: "The usage limit has been reached", plan_type: "plus", resets_at: 1_800_000_000,
+    } };
+    const emit = (ws: FakeWebSocket, payload: Record<string, unknown>) =>
+      ws.emit("message", { data: JSON.stringify(payload, null, 2) });
+
+    async function receive(payload: Record<string, unknown>, prelude: Record<string, unknown>[] = [],
+      url = CODEX_URL, onQuota?: (headers: Headers) => void) {
+      installFake(ws => {
+        ws.emit("open", {});
+        for (const event of prelude) emit(ws, event);
+        emit(ws, payload);
+        ws.emit("close", { code: 1000, reason: "normal" });
+      });
+      let attempts = 0;
+      let fallbacks = 0;
+      const response = await fetchWithTransientRetry(() => {
+        attempts++;
+        return rawCodexWsUpstreamFetch(url, streamingInit(), (async () => {
+          fallbacks++;
+          throw new Error("a sent create must not be resent over HTTP");
+        }) as typeof fetch, BOUNDED_WS_RUNTIME, onQuota);
+      }, {});
+      const ws = FakeWebSocket.instances.at(-1)!;
+      expect(attempts).toBe(1);
+      expect(fallbacks).toBe(0);
+      expect(ws.sent).toHaveLength(1);
+      expect(ws.closed).toBe(true);
+      expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true);
+      return response;
+    }
+
+    // Independent oracle: openai/codex d2d5b702, responses_websocket.rs:1016-1064
+    // explicitly accepts numeric window-minutes as the HTTP header string "15".
+    test.each(["status", "status_code"])("returns %s 429 as bounded HTTP JSON with scalar quota headers", async field => {
+      const { status_code, ...frame } = refusal;
+      const response = await receive({ ...frame, [field]: status_code, headers: {
+        "X-Codex-Primary-Used-Percent": "100.0", "X-Codex-Primary-Window-Minutes": 15,
+        "X-Codex-Primary-Reset-At": 1_800_000_000, "X-Codex-Credits-Has-Credits": true,
+        "Retry-After": 60, "X-Request-Id": "fixture-request",
+        "x-codex-extra-secondary-used-percent": "25", "x-ratelimit-remaining-requests": 0,
+      } });
+      expect(response.status).toBe(429);
+      expect(response.headers.get("content-type")).toBe("application/json");
+      expect(response.headers.get("cache-control")).toBe("no-store");
+      expect(response.headers.get("x-codex-primary-used-percent")).toBe("100.0");
+      expect(response.headers.get("x-codex-primary-window-minutes")).toBe("15");
+      expect(response.headers.get("x-codex-primary-reset-at")).toBe("1800000000");
+      expect(response.headers.get("x-codex-credits-has-credits")).toBe("true");
+      expect(response.headers.get("retry-after")).toBe("60");
+      expect(response.headers.get("x-request-id")).toBe("fixture-request");
+      expect(response.headers.get("x-codex-extra-secondary-used-percent")).toBe("25");
+      expect(response.headers.get("x-ratelimit-remaining-requests")).toBe("0");
+      expect(isCodexWsUpstreamResponse(response)).toBe(false);
+      expect(isCodexWsQuotaObservedResponse(response)).toBe(false);
+      expect(await response.json()).toEqual({ error: refusal.error });
+    });
+
+    test.each([400, 401, 402, 403, 404, 408, 499])("preserves a precommit HTTP %i refusal", async status_code => {
+      const response = await receive({ ...refusal, status_code });
+      expect(response.status).toBe(status_code);
+      expect(await response.json()).toEqual({ error: refusal.error });
+    });
+
+    test.each([
+      { status_code: undefined }, { status_code: null }, { status_code: "429" }, { status_code: true },
+      { status_code: 429.5 }, { status_code: 399 }, { status_code: 500 }, { status_code: 502 },
+      { status_code: 503 }, { status_code: 599 }, { status_code: 429, status: 429 },
+      { status_code: 502, status: 429 }, { status_code: null, status: 401 },
+      { status_code: "bad", status: 401 }, { error: [] }, { error: "refused" },
+      { error: { code: 42 } }, { error: { message: false } }, { headers: [] }, { headers: "bad" },
+      { stream_id: "another-stream" },
+    ])("keeps an ineligible wrapper on SSE without outer retry: %j", async fields => {
+      const response = await receive({ ...refusal, ...fields });
+      expect(response.status).toBe(200);
+      expect(isCodexWsUpstreamResponse(response)).toBe(true);
+      expect(await response.text()).toContain("event: error\ndata: ");
+    });
+
+    test.each([undefined, null, {}])("handles an optional error object: %j", async error => {
+      const response = await receive({ ...refusal, error, headers: null });
+      expect(response.status).toBe(429);
+      expect(await response.json()).toEqual({ error: error ?? {
+        type: "upstream_error", message: "Upstream rejected the request",
+      } });
+    });
+
+    test("drops injection, credentials, framing and connection-nominated metadata", async () => {
+      const forbidden = ["Authorization", "Proxy-Authorization", "Cookie", "Set-Cookie", "Content-Length",
+        "Content-Encoding", "Transfer-Encoding", "Keep-Alive", "Proxy-Connection", "TE", "Trailer", "Upgrade",
+        "Content-Range", "Content-Location", "ETag", "Last-Modified", "Digest", "Content-MD5",
+        "Access-Control-Allow-Origin", "Location", "WWW-Authenticate", "x-codex-private-token"];
+      const error = { message: "refusal\r\nX-Injected: body text only" };
+      const response = await receive({ ...refusal, error, headers: {
+        ...Object.fromEntries(forbidden.map(name => [name, "must-not-leak"])),
+        "Content-Type": "text/html", "Cache-Control": "public, max-age=3600",
+        Connection: "Retry-After, X-Codex-Primary-Used-Percent, content-type, cache-control",
+        connection: "X-Request-Id", "Retry-After": "60", "X-Request-Id": "must-not-leak",
+        "x-codex-primary-used-percent": "100", "x-codex-secondary-used-percent": "99",
+        "x-ratelimit-bad name": "invalid", "x-ratelimit-crlf": "ok\r\nSet-Cookie: injected",
+        "x-ratelimit-nul": "bad\0value", "x-ratelimit-nonbyte": "漢字",
+        "x-ratelimit-array": [1], "x-ratelimit-object": { value: 1 }, "x-ratelimit-null": null,
+        "X-RateLimit-Remaining": "2", "x-ratelimit-remaining": "3",
+      } }, [{ type: "codex.response.metadata", headers: {
+        "retry-after": "10", "x-request-id": "prelude-request", "x-codex-primary-used-percent": "30",
+      } }]);
+      expect(response.status).toBe(429);
+      expect(Object.fromEntries(response.headers)).toEqual({
+        "cache-control": "no-store", "content-type": "application/json",
+        "x-codex-secondary-used-percent": "99", "x-ratelimit-remaining": "3",
+      });
+      expect(await response.json()).toEqual({ error });
+    });
+
+    test("merges prelude quota with refusal updates without replaying the observer", async () => {
+      const observations: string[] = [];
+      const response = await receive({ ...refusal, headers: { "x-codex-primary-used-percent": 100 } }, [
+        { type: "codex.rate_limits", rate_limits: {
+          primary: { used_percent: 30, window_minutes: 15, reset_at: 1_800_000_000 },
+          secondary: { used_percent: 40, window_minutes: 10080, reset_at: 1_900_000_000 },
+        } },
+        { type: "codex.response.metadata", headers: { "x-models-etag": "prelude-catalog" } },
+      ], CODEX_URL, headers => observations.push(headers.get("x-codex-primary-used-percent")!));
+      expect(response.status).toBe(429);
+      expect(response.headers.get("x-codex-primary-used-percent")).toBe("100");
+      expect(response.headers.has("x-codex-primary-window-minutes")).toBe(false);
+      expect(response.headers.has("x-codex-primary-reset-at")).toBe(false);
+      expect(response.headers.get("x-codex-secondary-used-percent")).toBe("40");
+      expect(response.headers.get("x-codex-secondary-reset-at")).toBe("1900000000");
+      expect(response.headers.get("x-models-etag")).toBe("prelude-catalog");
+      expect(observations).toEqual(["30"]);
+      expect(isCodexWsQuotaObservedResponse(response)).toBe(false);
+      expect(await response.json()).toEqual({ error: refusal.error });
+    });
+
+    const boundedHeaders = (count: number, value = "1") =>
+      Object.fromEntries(Array.from({ length: count }, (_, i) => [`x-ratelimit-fixture-${i}`, value]));
+    const quotaFamilies = (count: number) => Object.fromEntries(
+      Array.from({ length: count }, (_, i) => [`x-codex-family-${i}-primary-used-percent`, "1"]));
+    test.each([
+      ["value", { "x-models-etag": "x".repeat(4096) }, true],
+      ["value overflow", { "x-models-etag": "x".repeat(4097) }, false],
+      ["UTF-8 value", { "x-models-etag": "é".repeat(2048) }, true],
+      ["UTF-8 overflow", { "x-models-etag": "é".repeat(2049) }, false],
+      ["header count", boundedHeaders(128), true], ["header count overflow", boundedHeaders(129), false],
+      ["families", quotaFamilies(16), true], ["family overflow", quotaFamilies(17), false],
+      ["total bytes", boundedHeaders(8, "x".repeat(3990)), true],
+      ["total byte overflow", boundedHeaders(8, "x".repeat(4096)), false],
+    ] as Array<[string, Record<string, string>, boolean]>)("enforces metadata budget: %s", async (_name, headers, accepted) => {
+      const response = await receive({ ...refusal, headers });
+      if (accepted) {
+        expect(response.status).toBe(429);
+        for (const [name, value] of Object.entries(headers)) expect(response.headers.get(name)).toBe(value);
+        expect(await response.json()).toEqual({ error: refusal.error });
+      } else {
+        expect(response.status).toBe(200);
+        expect(isCodexWsUpstreamResponse(response)).toBe(true);
+        await expect(response.text()).rejects.toThrow("metadata");
+      }
+    });
+
+    test("bounds the cumulative prelude and rejection metadata even when updates replace values", async () => {
+      const response = await receive({ ...refusal, headers: boundedHeaders(5, "x".repeat(4096)) }, [
+        { type: "codex.response.metadata", headers: boundedHeaders(4, "y".repeat(4096)) },
+      ]);
+      expect(response.status).toBe(200);
+      await expect(response.text()).rejects.toThrow("metadata");
+    });
+
+    test.each([
+      ["response.created", 429], ["response.output_text.delta", 429],
+      ["response.in_progress", 429], ["response.created", 502],
+    ] as Array<[string, number]>)(
+      "does not convert or retry a refusal after %s (status %i)", async (type, status_code) => {
+        const response = await receive({ ...refusal, status_code }, [{ type, response: { id: "r1" }, delta: "output" }]);
+        expect(response.status).toBe(200);
+        const text = await response.text();
+        expect(text).toContain(`event: ${type}`);
+        expect(text).toContain("event: error");
+        expect(response.headers.has("cache-control")).toBe(false);
+      });
+
+    test.each(["websocket_connection_limit_reached", "previous_response_not_found"])(
+      "does not add native special-code reconnect for %s", async code => {
+        const response = await receive({ type: "error", error: { code } });
+        expect(response.status).toBe(200);
+        expect(await response.text()).toContain(code);
+      });
+
+    test("keeps noncanonical providers on the stream path", async () => {
+      const response = await receive(refusal, [], "https://gateway.example/v1/responses");
+      expect(response.status).toBe(200);
+      expect(await response.text()).toContain("event: error");
+    });
+
+    test.each([CODEX_URL, "https://gateway.example/v1/responses"])(
+      "settles synchronous error/send-throw/close races and detaches deadlines for %s", async url => {
+        jest.useFakeTimers();
+        const abort = new AbortController();
+        let fallbacks = 0;
+        try {
+          installFake(ws => {
+            ws.send = data => {
+              ws.sent.push(data);
+              emit(ws, refusal);
+              throw new Error("send threw after a response was received");
+            };
+            ws.emit("open", {});
+          });
+          const response = await rawCodexWsUpstreamFetch(url, { ...streamingInit(), signal: abort.signal },
+            (async () => { fallbacks++; throw new Error("unexpected fallback"); }) as typeof fetch, BOUNDED_WS_RUNTIME);
+          const ws = FakeWebSocket.instances.at(-1)!;
+          abort.abort(new Error("late abort"));
+          ws.emit("error", {});
+          emit(ws, { type: "codex.rate_limits", rate_limits: { primary: { used_percent: 10 } } });
+          ws.emit("close", {});
+          jest.advanceTimersByTime(CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS + 10_000);
+          expect(response.status).toBe(url === CODEX_URL ? 429 : 200);
+          if (url === CODEX_URL) expect(await response.json()).toEqual({ error: refusal.error });
+          else expect(await response.text()).toContain("event: error");
+          expect(ws.sent).toHaveLength(1);
+          expect(ws.closed).toBe(true);
+          expect(fallbacks).toBe(0);
+          expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true);
+        } finally { jest.useRealTimers(); }
+      });
+
+    test.each([false, true])("disposes a retained socket; correlation precedes conversion (foreign stream: %s)", async foreign => {
+      installFake(ws => {
+        ws.emit("open", {});
+        emit(ws, { type: "response.created", response: { id: "completed-first" } });
+        emit(ws, { type: "response.completed", response: { id: "completed-first", status: "completed" } });
+      });
+      const init = streamingInit();
+      const prepared = prepareCodexWsRequest(CODEX_URL, init)!;
+      const session = new CodexWsSession("wss://chatgpt.com/backend-api/codex/responses", prepared.headers, true);
+      let fallbacks = 0;
+      const options = { session, url: CODEX_URL, init, prepared, sseFallback: (async () => {
+        fallbacks++;
+        throw new Error("retained create must not fall back");
+      }) as typeof fetch };
+      try {
+        expect(session.reserve()).toBe(true);
+        await (await codexWsExchange(options)).text();
+        expect(session.reused).toBe(true);
+        expect(session.closed).toBe(false);
+        const ws = FakeWebSocket.instances.at(-1)!;
+        let terminations = 0;
+        Object.assign(ws, { terminate: () => { terminations++; } });
+        ws.send = data => { ws.sent.push(data); emit(ws, { ...refusal, ...(foreign ? { stream_id: "foreign" } : {}) }); };
+        expect(session.reserve()).toBe(true);
+        const response = await codexWsExchange(options);
+        if (foreign) {
+          expect(response.status).toBe(200);
+          await expect(response.text()).rejects.toThrow("identity mismatch");
+        } else {
+          expect(response.status).toBe(429);
+          expect(isCodexWsUpstreamResponse(response)).toBe(false);
+          expect(await response.json()).toEqual({ error: refusal.error });
+        }
+        expect(ws.sent).toHaveLength(2);
+        expect(ws.closed).toBe(true);
+        expect(terminations).toBe(1);
+        expect(session.closed).toBe(true);
+        expect(session.busy).toBe(false);
+        expect(session.hasCompleted("completed-first")).toBe(false);
+        expect(session.reserve()).toBe(false);
+        expect(fallbacks).toBe(0);
+        expect([...ws.listeners.values()].every(listeners => listeners.length === 0)).toBe(true);
+      } finally { session.dispose(); }
+    });
+  });
+
   test.each(["error", "response.completed"])("multiline upstream %s JSON remains one valid SSE data value", async type => {
     const payload = type === "error"
-      ? { type, status: 400, error: { type: "invalid_request_error", message: "fixture refusal" } }
+      ? { type, error: { type: "invalid_request_error", message: "fixture refusal" } }
       : { type, response: { id: "pretty-response", status: "completed", output: [] } };
     installFake(ws => {
       ws.emit("open", {});
diff --git a/tests/server/agent-task-recovery-cache.test.ts b/tests/server/agent-task-recovery-cache.test.ts
index 3a7324340d..2ee994f8ce 100644
--- a/tests/server/agent-task-recovery-cache.test.ts
+++ b/tests/server/agent-task-recovery-cache.test.ts
@@ -6,6 +6,17 @@ import {
   resetAgentTaskRecoveryCache,
   resolveCachedAgentTaskRecovery,
 } from "../../src/server/responses/agent-task-recovery-cache";
+import {
+  recoverEncryptedAgentTaskWithResult,
+  restoreCachedEncryptedAgentTasks,
+} from "../../src/server/responses/agent-task-recovery";
+import {
+  codexHeaders,
+  encryptedInput,
+  originalFetch,
+  recoverySse,
+  routedConfig,
+} from "../helpers/agent-task-recovery";
 
 const realDateNow = Date.now;
 
@@ -13,10 +24,163 @@ describe("agent task recovery cache", () => {
   beforeEach(() => resetAgentTaskRecoveryCache());
 
   afterEach(() => {
+    globalThis.fetch = originalFetch;
     Date.now = realDateNow;
     resetAgentTaskRecoveryCache();
   });
 
+  test("shared failure gives each waiter its own result without contaminating another key", async () => {
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let fetches = 0;
+    globalThis.fetch = (async () => {
+      const requestNumber = ++fetches;
+      await gate;
+      return requestNumber === 1
+        ? new Response("raw-failure-sentinel", { status: 503 })
+        : new Response(recoverySse("Independent assignment."));
+    }) as typeof fetch;
+    const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+    const config = routedConfig();
+    const firstInput = encryptedInput();
+    const secondInput = encryptedInput();
+    const otherInput = encryptedInput();
+    const first = recoverEncryptedAgentTaskWithResult(req, firstInput, {}, config);
+    const second = recoverEncryptedAgentTaskWithResult(req, secondInput, {}, config);
+    const other = recoverEncryptedAgentTaskWithResult(req, otherInput, {}, config, { parentThreadId: "other-parent" });
+    try {
+      expect(agentTaskRecoveryWaiterCountForTests()).toBe(3);
+      expect(fetches).toBe(2);
+      release?.();
+      const [firstResult, secondResult, otherResult] = await Promise.all([first, second, other]);
+      expect(firstResult).toEqual({ recovered: false, reason: "recovery_unavailable" });
+      expect(secondResult).toEqual({ recovered: false, reason: "recovery_unavailable" });
+      expect(firstResult).not.toBe(secondResult);
+      expect(otherResult).toEqual({ recovered: true });
+      expect(firstInput).toEqual(encryptedInput());
+      expect(secondInput).toEqual(encryptedInput());
+      expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(0);
+      expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config, { parentThreadId: "other-parent" })).toBe(1);
+      expect(fetches).toBe(2);
+    } finally {
+      release?.();
+      await Promise.all([first, second, other]);
+    }
+  });
+
+  for (const succeeds of [true, false]) {
+    test(`caller cancellation stays local when the remaining waiter ${succeeds ? "succeeds" : "fails"}`, async () => {
+      let release: (() => void) | undefined;
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      let sharedSignal: AbortSignal | null | undefined;
+      let fetches = 0;
+      globalThis.fetch = (async (_input, init) => {
+        fetches += 1;
+        sharedSignal = init?.signal;
+        await gate;
+        return succeeds ? new Response(recoverySse("Shared assignment.")) : new Response(null, { status: 503 });
+      }) as typeof fetch;
+      const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+      const config = routedConfig();
+      const controller = new AbortController();
+      const cancelledInput = encryptedInput();
+      const first = recoverEncryptedAgentTaskWithResult(req, cancelledInput, {}, config, { abortSignal: controller.signal });
+      const second = recoverEncryptedAgentTaskWithResult(req, encryptedInput(), {}, config);
+      try {
+        expect(agentTaskRecoveryWaiterCountForTests()).toBe(2);
+        controller.abort(new Error("private-cancellation-sentinel"));
+        expect(await first).toEqual({ recovered: false, reason: "caller_cancelled" });
+        expect(cancelledInput).toEqual(encryptedInput());
+        expect(sharedSignal?.aborted).toBe(false);
+        release?.();
+        expect(await second).toEqual(succeeds
+          ? { recovered: true }
+          : { recovered: false, reason: "recovery_unavailable" });
+        expect(fetches).toBe(1);
+        expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(succeeds ? 1 : 0);
+      } finally {
+        release?.();
+        await Promise.all([first, second]);
+      }
+    });
+  }
+
+  test("already cancelled callers cannot inject a positive cache hit", async () => {
+    const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+    const config = routedConfig();
+    let fetches = 0;
+    globalThis.fetch = (async () => { fetches += 1; return new Response(recoverySse("Cached assignment.")); }) as typeof fetch;
+    expect(await recoverEncryptedAgentTaskWithResult(req, encryptedInput(), {}, config)).toEqual({ recovered: true });
+    const controller = new AbortController();
+    controller.abort();
+    const input = encryptedInput();
+    expect(await recoverEncryptedAgentTaskWithResult(req, input, {}, config, { abortSignal: controller.signal }))
+      .toEqual({ recovered: false, reason: "caller_cancelled" });
+    expect(input).toEqual(encryptedInput());
+    // The existing pre-abort/null path does not discard another caller's cache entry.
+    expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(1);
+    expect(fetches).toBe(1);
+  });
+
+  test("cancellation after cache lookup retains the existing discard behavior", async () => {
+    const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+    const config = routedConfig();
+    globalThis.fetch = (async () => new Response(recoverySse("Cached assignment."))) as typeof fetch;
+    expect(await recoverEncryptedAgentTaskWithResult(req, encryptedInput(), {}, config)).toEqual({ recovered: true });
+    const controller = new AbortController();
+    const input = encryptedInput();
+    const pending = recoverEncryptedAgentTaskWithResult(req, input, {}, config, { abortSignal: controller.signal });
+    // The cache lookup returned an assignment, but the caller has not resumed to inject it.
+    controller.abort();
+    expect(await pending).toEqual({ recovered: false, reason: "caller_cancelled" });
+    expect(input).toEqual(encryptedInput());
+    expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 });
+  });
+
+  test("input replacement after admission reports input_changed and discards recovered plaintext", async () => {
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    globalThis.fetch = (async () => { await gate; return new Response(recoverySse("Do not inject.")); }) as typeof fetch;
+    const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+    const config = routedConfig();
+    const input = encryptedInput();
+    const pending = recoverEncryptedAgentTaskWithResult(req, input, {}, config);
+    try {
+      input[0] = { type: "message", role: "user", content: [] };
+      const replaced = structuredClone(input);
+      release?.();
+      expect(await pending).toEqual({ recovered: false, reason: "input_changed" });
+      expect(input).toEqual(replaced);
+      expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 });
+      expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(0);
+    } finally {
+      release?.();
+      await pending;
+    }
+  });
+
+  test("recovery_unavailable does not imply a fetch when all flight slots are occupied", async () => {
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const pending = Array.from({ length: 32 }, (_, index) => resolveCachedAgentTaskRecovery(
+      `occupied-${index}`, 200, async () => { await gate; return null; },
+    ));
+    let fetches = 0;
+    globalThis.fetch = (async () => { fetches += 1; throw new Error("must-not-fetch"); }) as typeof fetch;
+    try {
+      const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+      const input = encryptedInput();
+      expect(await recoverEncryptedAgentTaskWithResult(req, input, {}, routedConfig()))
+        .toEqual({ recovered: false, reason: "recovery_unavailable" });
+      expect(fetches).toBe(0);
+      expect(input).toEqual(encryptedInput());
+      expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 });
+    } finally {
+      release?.();
+      await Promise.all(pending);
+    }
+  });
+
   test("read-only hits retain the original expiry and exact-expiry reads release UTF-8 bytes", async () => {
     const insertedAt = 1_800_000_000_000;
     let now = insertedAt;
diff --git a/tests/server/agent-task-recovery-security.test.ts b/tests/server/agent-task-recovery-security.test.ts
index 8fd42dae0e..d44c155dfb 100644
--- a/tests/server/agent-task-recovery-security.test.ts
+++ b/tests/server/agent-task-recovery-security.test.ts
@@ -1,6 +1,13 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
-import { resetAgentTaskRecoveryState } from "../../src/server/responses/agent-task-recovery";
 import {
+  discardEncryptedAgentTaskRecovery,
+  recoverEncryptedAgentTask,
+  recoverEncryptedAgentTaskWithResult,
+  resetAgentTaskRecoveryState,
+  restoreCachedEncryptedAgentTasks,
+} from "../../src/server/responses/agent-task-recovery";
+import {
+  agentMessage,
   codexHeaders,
   encryptedInput,
   fakeChatGptJwt,
@@ -24,6 +31,55 @@ describe("agent task recovery security", () => {
     resetAgentTaskRecoveryState();
   });
 
+  test("diagnoses unsupported envelopes before admission without exposing their content", async () => {
+    const req = new Request("http://localhost/v1/responses"); // No credentials.
+    let fetches = 0;
+    globalThis.fetch = (async () => { fetches += 1; throw new Error("must-not-fetch"); }) as typeof fetch;
+    const header = { type: "input_text", text: ROUTING_ENVELOPE };
+    const encrypted = { type: "encrypted_content", encrypted_content: FERNET_TASK };
+    const inputs = [
+      agentMessage([header, encrypted, encrypted]),
+      agentMessage([header, { ...encrypted, encrypted_content: FERNET_TASK.slice(0, 50) },
+        { ...encrypted, encrypted_content: FERNET_TASK.slice(50) }]),
+      agentMessage([{ ...header, text: ROUTING_ENVELOPE.replace("NEW_TASK", "new_task") }, encrypted]),
+      encryptedInput({ ciphertext: "unsupported-ciphertext-sentinel" }),
+    ];
+    for (const input of inputs) {
+      const original = structuredClone(input);
+      expect(await recoverEncryptedAgentTaskWithResult(req, input, {}, routedConfig()))
+        .toEqual({ recovered: false, reason: "unsupported_envelope" });
+      expect(await recoverEncryptedAgentTask(req, input, {}, routedConfig())).toBe(false);
+      expect(input).toEqual(original);
+    }
+    expect(fetches).toBe(0);
+  });
+
+  test("typed admission denial cannot read or discard an authenticated cached assignment", async () => {
+    const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+    const config = routedConfig();
+    let fetches = 0;
+    globalThis.fetch = (async () => {
+      fetches += 1;
+      return new Response(recoverySse("private-assignment-sentinel"));
+    }) as typeof fetch;
+    expect(await recoverEncryptedAgentTaskWithResult(req, encryptedInput(), {}, config))
+      .toEqual({ recovered: true });
+
+    const deniedHeaders = codexHeaders();
+    deniedHeaders.set("chatgpt-account-id", "mismatched-account-sentinel");
+    const denied = new Request(req.url, { headers: deniedHeaders });
+    const input = encryptedInput();
+    const original = structuredClone(input);
+    expect(await recoverEncryptedAgentTaskWithResult(denied, input, {}, config))
+      .toEqual({ recovered: false, reason: "admission_denied" });
+    expect(await recoverEncryptedAgentTask(denied, input, {}, config)).toBe(false);
+    expect(restoreCachedEncryptedAgentTasks(denied, input, config)).toBe(0);
+    discardEncryptedAgentTaskRecovery(denied, input, config);
+    expect(input).toEqual(original);
+    expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(1);
+    expect(fetches).toBe(1);
+  });
+
   test("uses only the fixed ChatGPT endpoint and forwards only allowlisted credentials", async () => {
     const accountId = "acct-boundary";
     const token = fakeChatGptJwt(accountId);
diff --git a/tests/server/agent-task-recovery.test.ts b/tests/server/agent-task-recovery.test.ts
index 5cc96793ce..ceb1c5b6b5 100644
--- a/tests/server/agent-task-recovery.test.ts
+++ b/tests/server/agent-task-recovery.test.ts
@@ -1,7 +1,13 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
 import { createTranslatorBudget } from "../../src/lib/translator-budget";
 import { warnAgentTaskRecoveryStartup } from "../../src/server";
-import { resetAgentTaskRecoveryState } from "../../src/server/responses/agent-task-recovery";
+import {
+  discardEncryptedAgentTaskRecovery,
+  recoverEncryptedAgentTask,
+  recoverEncryptedAgentTaskWithResult,
+  resetAgentTaskRecoveryState,
+  restoreCachedEncryptedAgentTasks,
+} from "../../src/server/responses/agent-task-recovery";
 import { agentTaskRecoveryWaiterCountForTests } from "../../src/server/responses/agent-task-recovery-cache";
 import {
   agentMessage,
@@ -29,6 +35,84 @@ describe("agent task recovery (opt-in, default off)", () => {
     resetAgentTaskRecoveryState();
   });
 
+  for (const messageType of ["NEW_TASK", "MESSAGE"] as const) {
+    test(`typed ${messageType} recovery preserves boolean, replay and discard contracts`, async () => {
+      const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+      const config = routedConfig();
+      const context = { parentThreadId: "parent-diagnostics" };
+      const input = () => agentMessage([
+        { type: "input_text", text: ROUTING_ENVELOPE.replace("NEW_TASK", messageType) },
+        { type: "encrypted_content", encrypted_content: FERNET_TASK },
+      ]);
+      let fetches = 0;
+      globalThis.fetch = (async () => {
+        fetches += 1;
+        return new Response(recoverySse("Recovered diagnostic fixture."));
+      }) as typeof fetch;
+
+      const typedInput = input();
+      expect(await recoverEncryptedAgentTaskWithResult(req, typedInput, {}, config, context))
+        .toEqual({ recovered: true });
+      const booleanInput = input();
+      expect(await recoverEncryptedAgentTask(req, booleanInput, {}, config, context)).toBe(true);
+      expect(booleanInput).toEqual(typedInput);
+      expect(typedInput).toEqual([{
+        type: "message", role: "user", content: [
+          { type: "input_text", text: ROUTING_ENVELOPE.replace("NEW_TASK", messageType) },
+          { type: "input_text", text: "Recovered diagnostic fixture." },
+        ],
+      }]);
+      const replay = input();
+      expect(restoreCachedEncryptedAgentTasks(req, replay, config, context)).toBe(1);
+      expect(replay).toEqual(typedInput);
+      expect(fetches).toBe(1);
+
+      const otherType = agentMessage([
+        { type: "input_text", text: ROUTING_ENVELOPE.replace("NEW_TASK", messageType === "MESSAGE" ? "NEW_TASK" : "MESSAGE") },
+        { type: "encrypted_content", encrypted_content: FERNET_TASK },
+      ]);
+      expect(restoreCachedEncryptedAgentTasks(req, otherType, config, context)).toBe(0);
+      discardEncryptedAgentTaskRecovery(req, input(), config, context);
+      expect(restoreCachedEncryptedAgentTasks(req, input(), config, context)).toBe(0);
+      expect(fetches).toBe(1);
+    });
+  }
+
+  const failedRecoveries: Array<[string, () => Response]> = [
+    ["HTTP 503", () => new Response("raw-error-sentinel", { status: 503 })],
+    ["network exception", () => { throw new Error("raw-error-sentinel"); }],
+    ["malformed SSE", () => new Response("data: {not-json}\n\n")],
+    ["missing completion", () => new Response(recoverySse("payload-sentinel").split("data: {\"type\":\"response.completed\"")[0])],
+    ["conflicting assignment", () => new Response(recoverySse("payload-sentinel") + recoveryCompletedSse("other-payload-sentinel"))],
+    ["failed terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.failed","response":{"error":{"message":"raw-error-sentinel"}}}\n\n')],
+    ["incomplete terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.incomplete"}\n\n')],
+    ["bare error", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"error","error":{"message":"raw-error-sentinel"}}\n\n')],
+    // Exact-case events are also used by the pinned official Codex source. Recovery's
+    // additional completed-status requirement remains deliberately stricter.
+    ["mixed-case completion", () => new Response(recoverySse("payload-sentinel").replace("response.completed", "Response.Completed"))],
+    ["mixed-case status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed"', '"status":"Completed"'))],
+    ["missing status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed",', ""))],
+    ["ciphertext assignment", () => new Response(recoverySse(FERNET_TASK))],
+  ];
+  for (const [name, response] of failedRecoveries) {
+    test(`typed recovery keeps ${name} coarse and preserves false without retrying`, async () => {
+      const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() });
+      const config = routedConfig();
+      let fetches = 0;
+      globalThis.fetch = (async () => { fetches += 1; return response(); }) as typeof fetch;
+      const input = encryptedInput();
+      const original = structuredClone(input);
+      expect(await recoverEncryptedAgentTaskWithResult(req, input, {}, config))
+        .toEqual({ recovered: false, reason: "recovery_unavailable" });
+      expect(input).toEqual(original);
+      expect(fetches).toBe(1);
+      expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(0);
+      expect(await recoverEncryptedAgentTask(req, input, {}, config)).toBe(false);
+      expect(fetches).toBe(2); // One request per explicit invocation; no internal retry.
+      expect(input).toEqual(original);
+    });
+  }
+
   test("keeps the disabled fail-fast response byte-identical to the absent feature", async () => {
     const snapshot = async (config: ReturnType<typeof routedConfig>) => {
       let fetchCalls = 0;
@@ -138,10 +222,11 @@ describe("agent task recovery (opt-in, default off)", () => {
       encryptedInput(),
       codexHeaders(),
     );
-    const json = await response.json() as { error?: { code?: string } };
+    const json = await response.json() as { error?: { code?: string; recovery_reason?: string } };
 
     expect(response.status).toBe(400);
     expect(json.error?.code).toBe("unreadable_encrypted_agent_task");
+    expect(json.error?.recovery_reason).toBe("recovery_unavailable");
     expect(fetchedUrls.length).toBeGreaterThan(0);
     expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex");
   });
@@ -693,7 +778,7 @@ describe("agent task recovery (opt-in, default off)", () => {
     expect(fetchedUrls).toHaveLength(1);
     expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex/responses");
     expect(await response.json()).toMatchObject({
-      error: { code: "unreadable_encrypted_agent_task" },
+      error: { code: "unreadable_encrypted_agent_task", recovery_reason: "recovery_unavailable" },
     });
   });
 });
diff --git a/tests/server/server-auth.test.ts b/tests/server/server-auth.test.ts
index 72a1297f37..cc5eb8f1fd 100644
--- a/tests/server/server-auth.test.ts
+++ b/tests/server/server-auth.test.ts
@@ -604,7 +604,15 @@ describe("server local API auth", () => {
 
   test("compact keeps the idle guard until a valid request body is complete", async () => {
     let bodyController!: ReadableStreamDefaultController<Uint8Array>;
-    const body = new ReadableStream<Uint8Array>({ start(controller) { bodyController = controller; } });
+    const readerWaiting = Promise.withResolvers<void>();
+    let readRequests = 0;
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) { bodyController = controller; },
+      pull(controller) {
+        if (readRequests++ === 0) controller.enqueue(new TextEncoder().encode('{"model":"fixture/gpt-test","input":['));
+        else readerWaiting.resolve();
+      },
+    }, { highWaterMark: 0 });
     const cfg = config();
     cfg.defaultProvider = "fixture";
     cfg.providers = { fixture: { ...cfg.providers.openai!, disabled: true } };
@@ -615,7 +623,7 @@ describe("server local API auth", () => {
     const result = handleResponsesCompact(request, cfg, { model: "unknown", provider: "unknown" }, undefined, undefined, {
       onRequestBodyRead: () => { accepted++; },
     });
-    bodyController.enqueue(new TextEncoder().encode('{"model":"fixture/gpt-test","input":['));
+    await readerWaiting.promise;
     expect(accepted).toBe(0);
     bodyController.enqueue(new TextEncoder().encode(']}'));
     bodyController.close();