diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md
index eb856811e1..6fc7dbf9a9 100644
--- a/docs-site/src/content/docs/reference/architecture.md
+++ b/docs-site/src/content/docs/reference/architecture.md
@@ -210,3 +210,23 @@ Structured `incomplete_details.reason` and error codes are accepted without a
 message; ordinary output-limit, filtering, steering and stall incompletes do not
 cool an account. Cyber-policy classification retains precedence. The terminal is
 not replayed after output, and fixed-account request selection remains fixed.
+
+Remote compact requests can buffer their response for longer than the server's
+request-idle timeout. That listener timeout is disabled after the request body is
+accepted; client cancellation and upstream operation deadlines still apply.
+
+Buffered routed compaction treats nonempty text and reasoning deltas as progress
+without exposing partial summary text. Comments, empty deltas and gateway
+keepalives do not reset the adapter-event stall watchdog. The default stall
+timeout stays 300 seconds; encrypted compaction content is preserved unchanged.
+
+Native compact response buffering also enforces a body-byte inactivity deadline
+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.
diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json
index 67093dcd07..707518fc31 100644
--- a/scripts/test-layout/layout.json
+++ b/scripts/test-layout/layout.json
@@ -492,6 +492,7 @@
     "command-code-quota.test.ts": "providers",
     "command-code-workspace-cache.test.ts": "providers",
     "commandcode-provider.test.ts": "providers",
+    "compaction-progress.test.ts": "responses",
     "compatibility-manifest.test.ts": "codex-integration",
     "compatibility-provider-equivalence.test.ts": "routing",
     "compatibility-version.test.ts": "ci-workflows",
diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts
index 1faa9c0cbb..07c0556d5e 100644
--- a/src/adapters/openai-responses.ts
+++ b/src/adapters/openai-responses.ts
@@ -2546,6 +2546,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig):
       let snapshot = "";
       let usage: OcxUsage | undefined;
       let compactionEncryptedContent: string | undefined;
+      let completedSeen = false;
       for await (const event of decodeServerSentEvents(response.body, { translatorBudget: budget })) {
         let payload: unknown;
         try { payload = JSON.parse(event.data); } catch { continue; }
@@ -2580,6 +2581,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig):
             return;
           case "response.completed":
             {
+              completedSeen = true;
               const responsePayload = isPlainObject(payload.response) ? payload.response : undefined;
               const output = Array.isArray(responsePayload?.output) ? responsePayload.output : [];
               const compaction = output.find(item => isPlainObject(item) && item.type === "compaction");
@@ -2620,6 +2622,18 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig):
             }
             break;
         }
+        // Buffered text is still upstream progress, but gateway keepalives are not.
+        // Yield after accounting, directly to the consumer: no progress queue or content leak.
+        if (
+          !completedSeen
+          && (payload.type === "response.output_text.delta"
+            || payload.type === "response.reasoning_summary_text.delta"
+            || payload.type === "response.reasoning_text.delta")
+          && typeof payload.delta === "string"
+          && payload.delta.length > 0
+        ) {
+          yield { type: "heartbeat" };
+        }
       }
       // Gateways differ in which of these they emit; prefer the authoritative
       // completed snapshot so text is never double-counted.
diff --git a/src/server/index.ts b/src/server/index.ts
index 0391ff031d..8d46c30ab6 100644
--- a/src/server/index.ts
+++ b/src/server/index.ts
@@ -1765,7 +1765,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         return runAdmittedHttpTurn(req, policy, async turnAdmissionLease => {
           let response: Response;
           try {
-            response = await handleResponsesCompact(req, config, logCtx, turnAdmissionLease, admission);
+            response = await handleResponsesCompact(req, config, logCtx, turnAdmissionLease, admission, {
+              onRequestBodyRead: () => disableResponsesRequestTimeout(req, requestServer),
+            });
           } catch {
             response = formatErrorResponse(500, "server_error", "Unexpected compact request failure");
           }
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/compact.ts b/src/server/responses/compact.ts
index a742fad98d..c914dcb4b2 100644
--- a/src/server/responses/compact.ts
+++ b/src/server/responses/compact.ts
@@ -112,7 +112,8 @@ import type { WsData } from "../ws-bridge";
 import { codexAccountSelectionForTurn, registerTurn, trackStreamLifetime, unregisterTurn } from "../lifecycle";
 import type { AdmissionLease } from "../../lib/admission";
 import { redactSecretString } from "../../lib/redact";
-import { readBoundedResponseBody } from "../../lib/bounded-body";
+import { readBoundedResponseBytes } from "../../lib/bounded-body";
+import { resolveStallTimeoutSec } from "../../stall-timeout";
 import { isRateLimitOrQuotaFailureMessage } from "../../lib/errors";
 import { supportedLadderFor } from "../effort-policy";
 import {
@@ -212,6 +213,8 @@ function compactHandoffRoute(req: Request, previousModel: string, now = Date.now
 
 export interface HandleResponsesCompactOptions {
   nativeMainRefreshDependencies?: NativeMainRefreshDependencies;
+  /** Release the listener's idle guard only after the complete request body is accepted. */
+  onRequestBodyRead?: () => void;
 }
 
 export function compactResponseTooLargeError(): Response {
@@ -464,43 +467,45 @@ function compactResponseHeaders(upstream: Response): Headers {
   return headers;
 }
 
-export async function bufferCompactResponse(upstream: Response, signal: AbortSignal): Promise<Response> {
-  const reader = upstream.body?.getReader();
+export async function bufferCompactResponse(
+  upstream: Response,
+  signal: AbortSignal,
+  stallTimeoutSec?: number,
+): Promise<Response> {
   const headers = compactResponseHeaders(upstream);
-  if (!reader) return new Response(null, { status: upstream.status, statusText: upstream.statusText, headers });
-  const declaredLength = Number(upstream.headers.get("content-length"));
-  if (Number.isFinite(declaredLength) && declaredLength > COMPACT_RESPONSE_MAX_BYTES) {
-    await reader.cancel("compact_response_too_large").catch(() => undefined);
-    return compactResponseTooLargeError();
-  }
-  const chunks: Uint8Array[] = [];
-  let total = 0;
   try {
-    while (true) {
-      if (signal.aborted) {
-        await reader.cancel(signal.reason).catch(() => undefined);
-        return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");
-      }
-      const { done, value } = await reader.read();
-      if (done) break;
-      total += value.byteLength;
-      if (total > COMPACT_RESPONSE_MAX_BYTES) {
-        await reader.cancel("compact_response_too_large").catch(() => undefined);
-        return compactResponseTooLargeError();
-      }
-      chunks.push(value);
+    if (signal.aborted) {
+      // No reader is attached yet. Cancellation must not wait for a broken source's cleanup.
+      void upstream.body?.cancel(signal.reason).catch(() => undefined);
+      return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");
     }
-  } catch {
+    if (!upstream.body) return new Response(null, { status: upstream.status, statusText: upstream.statusText, headers });
+    const declaredLength = Number(upstream.headers.get("content-length"));
+    if (Number.isFinite(declaredLength) && declaredLength > COMPACT_RESPONSE_MAX_BYTES) {
+      void upstream.body.cancel("compact_response_too_large").catch(() => undefined);
+      return compactResponseTooLargeError();
+    }
+    // Header admission has finished; only non-empty body chunks re-arm this deadline.
+    // The raw reader preserves bytes and cancels/releases without awaiting source cleanup.
+    const result = await readBoundedResponseBytes(upstream, {
+      signal,
+      maxBytes: COMPACT_RESPONSE_MAX_BYTES,
+      inactivityTimeoutMs: resolveStallTimeoutSec(stallTimeoutSec) * 1_000,
+    });
+    if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");
+    if (result.oversized) return compactResponseTooLargeError();
+    return new Response(result.bytes, { status: upstream.status, statusText: upstream.statusText, headers });
+  } catch (error) {
     if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");
+    if (error instanceof DOMException && error.name === "TimeoutError") {
+      return Response.json({ error: {
+        message: "Compact response body stalled",
+        type: "upstream_stall_timeout",
+        code: "upstream_stall_timeout",
+      } }, { status: 504 });
+    }
     return formatErrorResponse(502, "upstream_error", "Failed to read compact response");
   }
-  const body = new Uint8Array(total);
-  let offset = 0;
-  for (const chunk of chunks) {
-    body.set(chunk, offset);
-    offset += chunk.byteLength;
-  }
-  return new Response(body, { status: upstream.status, statusText: upstream.statusText, headers });
 }
 
 
@@ -526,6 +531,7 @@ export async function handleResponsesCompact(
   if (typeof raw.model !== "string" || raw.model.length === 0) {
     return formatErrorResponse(400, "invalid_request_error", "compaction request requires a model");
   }
+  options.onRequestBodyRead?.();
   // Correct the IDENTITY before routing, or the synthetic id does not route at all. Held in
   // a local rather than written back to `raw.model`: assigning to the property widens it out
   // of the `string` narrowing the guard above just established.
@@ -1037,7 +1043,7 @@ export async function handleResponsesCompact(
       upstream.headers.get("x-codex-secondary-reset-at"),
       upstream.headers.get("x-codex-tertiary-reset-at"),
     ].filter(Boolean);
-    const buffered = await bufferCompactResponse(upstream, req.signal);
+    const buffered = await bufferCompactResponse(upstream, req.signal, config.stallTimeoutSec);
     const bufferedErrorText = buffered.ok
       ? ""
       : await buffered.clone().text().catch(() => "");
diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md
index 5196684d21..e4a4d7a016 100644
--- a/structure/04_transports-and-sidecars.md
+++ b/structure/04_transports-and-sidecars.md
@@ -1655,3 +1655,24 @@ Structured `incomplete_details.reason` and error codes are accepted without a
 message; ordinary output-limit, filtering, steering and stall incompletes do not
 cool an account. Cyber-policy classification retains precedence. The terminal is
 not replayed after output, and fixed-account request selection remains fixed.
+
+Remote compact requests release the server request-idle timeout only after a complete
+JSON object with a valid model has been read. Partial or invalid uploads retain
+the listener guard; admitted compaction then uses the upstream operation's own
+deadlines and client cancellation.
+
+Buffered routed compaction treats nonempty text and reasoning deltas as progress
+without exposing partial summary text. Comments, empty deltas and gateway
+keepalives do not reset the adapter-event stall watchdog. The default stall
+timeout stays 300 seconds; encrypted compaction content is preserved unchanged.
+
+Native compact response buffering also enforces a body-byte inactivity deadline
+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.
diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json
index 0823ba202c..f9df7f508f 100644
--- a/tests/fixtures/test-layout-expected.json
+++ b/tests/fixtures/test-layout-expected.json
@@ -327,6 +327,7 @@
   "command-code-quota.test.ts": "providers",
   "command-code-workspace-cache.test.ts": "providers",
   "commandcode-provider.test.ts": "providers",
+  "compaction-progress.test.ts": "responses",
   "compatibility-manifest.test.ts": "codex-integration",
   "compatibility-provider-equivalence.test.ts": "routing",
   "compatibility-version.test.ts": "ci-workflows",
diff --git a/tests/responses/compaction-progress.test.ts b/tests/responses/compaction-progress.test.ts
new file mode 100644
index 0000000000..3e1a1556ea
--- /dev/null
+++ b/tests/responses/compaction-progress.test.ts
@@ -0,0 +1,205 @@
+import { describe, expect, test } from "bun:test";
+import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses";
+import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge";
+import type { AdapterEvent } from "../../src/types";
+import { createTestTranslatorBudget } from "../helpers/translator-budget";
+
+const encoder = new TextEncoder();
+const provider = { adapter: "openai-responses", baseUrl: "https://gateway.example/v1", authMode: "key" as const };
+const frame = (payload: unknown) => `data: ${JSON.stringify(payload)}\n\n`;
+const completed = {
+  type: "response.completed",
+  response: {
+    id: "resp_compaction",
+    status: "completed",
+    output: [{ type: "message", role: "assistant", content: [{ type: "output_text", text: "Final summary" }] }],
+  },
+};
+
+function upstream() {
+  let controller!: ReadableStreamDefaultController<Uint8Array>;
+  let nextRead = Promise.withResolvers<void>();
+  let ended = false;
+  let pulls = 0;
+  let cancelled = false;
+  const body = new ReadableStream<Uint8Array>({
+    start(value) { controller = value; },
+    pull() { pulls++; nextRead.resolve(); },
+    cancel() { ended = true; cancelled = true; },
+  }, { highWaterMark: 0 });
+  return {
+    body,
+    get pulls() { return pulls; },
+    get cancelled() { return cancelled; },
+    waitingForRead: () => nextRead.promise,
+    send(text: string) {
+      nextRead = Promise.withResolvers<void>();
+      controller.enqueue(encoder.encode(text));
+    },
+    close() { if (!ended) { ended = true; controller.close(); } },
+  };
+}
+
+function bridged() {
+  const source = upstream();
+  const budget = createTestTranslatorBudget();
+  let beat = () => {};
+  let cleanupCalls = 0;
+  const stream = bridgeToResponsesSSE(
+    createResponsesPassthroughAdapter(provider).parseStream(new Response(source.body), budget),
+    "example-model", undefined, undefined, undefined,
+    () => { cleanupCalls++; source.close(); }, 500,
+    {
+      translatorBudget: budget, compaction: true, stallTimeoutSec: 1,
+      timers: {
+        setInterval(callback) { beat = callback; return 1; },
+        clearInterval() { beat = () => {}; },
+      },
+    },
+  );
+  const text = new Response(stream).text();
+  return {
+    source, text,
+    get cleanupCalls() { return cleanupCalls; },
+    tick: () => beat(),
+    async send(text: string) {
+      await source.waitingForRead();
+      source.send(text);
+      // The next upstream read occurs after the bridge consumes any adapter heartbeat.
+      await source.waitingForRead();
+    },
+  };
+}
+
+describe("buffered Responses compaction progress", () => {
+  // Codex oracle: openai/codex d2d5b702, codex-api/src/sse/responses.rs:367-408.
+  // Indices make these canonical reasoning fixtures; progress itself carries no content.
+  for (const delta of [
+    { type: "response.output_text.delta", delta: "Buffered progress" },
+    { type: "response.reasoning_summary_text.delta", delta: "Buffered progress", summary_index: 0 },
+    { type: "response.reasoning_text.delta", delta: "Buffered progress", content_index: 0 },
+  ]) {
+    test(`${delta.type} prevents stall before terminal without exposing partial content`, async () => {
+      const h = bridged();
+      try {
+        for (let i = 0; i < 6; i++) {
+          await h.send(frame(delta));
+          h.tick();
+          expect(h.cleanupCalls).toBe(0);
+        }
+        await h.send(frame(completed));
+        h.source.close();
+        const wire = await h.text;
+        expect(wire.match(/event: response.completed\n/g)).toHaveLength(1);
+        expect(wire.match(/event: response.output_item.done\n/g)).toHaveLength(1);
+        expect(wire).toContain('"type":"compaction"');
+        expect(wire).not.toContain("Buffered progress");
+        expect(wire).not.toContain("event: response.output_text.delta");
+        expect(wire).not.toContain("upstream_stall_timeout");
+        // The bridge invokes its upstream cleanup callback on normal terminal events too.
+        expect(h.cleanupCalls).toBe(1);
+      } finally { h.source.close(); await h.text; }
+    });
+  }
+
+  test("comments, typed keepalives and empty or malformed deltas do not reset stall", async () => {
+    const h = bridged();
+    try {
+      const noise = ": keep-alive\n\ndata: invalid-json\n\n"
+        + frame({ type: "response.heartbeat" })
+        + frame({ type: "response.output_text.delta", delta: "" })
+        + frame({ type: "response.reasoning_summary_text.delta", delta: null })
+        + frame({ type: "response.reasoning_text.delta", delta: 42 })
+        + frame({ type: "response.unknown.delta", delta: "not recognized progress" });
+      await h.send(noise);
+      h.tick();
+      await h.send(noise);
+      h.tick();
+      const wire = await h.text;
+      expect(wire).toContain("upstream_stall_timeout");
+      expect(wire).not.toContain("event: response.completed");
+      expect(wire).not.toContain('"type":"compaction"');
+      expect(h.cleanupCalls).toBe(1);
+    } finally { h.source.close(); await h.text; }
+  });
+
+  test("progress preserves snapshot precedence, usage and native ciphertext", async () => {
+    const budget = createTestTranslatorBudget();
+    const ciphertext = "gAAAAABm-native-compaction-ciphertext";
+    const usage = { input_tokens: 12, output_tokens: 4, total_tokens: 16, gateway_metadata: { cached: true } };
+    const terminal = {
+      ...completed,
+      response: { ...completed.response, usage, output: [
+        ...completed.response.output, { type: "compaction", encrypted_content: ciphertext },
+      ] },
+    };
+    const input = frame({ type: "response.output_text.delta", delta: "Partial text" })
+      + frame({ type: "response.output_text.done", text: "Done text" })
+      + frame(terminal)
+      + frame({ type: "response.output_text.delta", delta: "Late text" });
+    const events: AdapterEvent[] = [];
+    for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) {
+      events.push(event);
+    }
+    expect(events).toEqual([
+      { type: "heartbeat" },
+      { type: "text_delta", text: "Final summary" },
+      { type: "done", usage: { inputTokens: 12, outputTokens: 4, totalTokens: 16, rawUsage: usage }, compactionEncryptedContent: ciphertext },
+    ]);
+    const result = buildResponseJSON(events, "example-model", { compaction: true, translatorBudget: budget });
+    expect(result.output).toEqual([expect.objectContaining({ type: "compaction", encrypted_content: ciphertext })]);
+  });
+
+  test("reasoning progress with ciphertext-only completion does not manufacture summary text", async () => {
+    const budget = createTestTranslatorBudget();
+    const ciphertext = "gAAAAABm-ciphertext-only";
+    const input = frame({ type: "response.reasoning_text.delta", content_index: 0, delta: "Hidden reasoning" })
+      + frame({ ...completed, response: {
+        ...completed.response, output: [{ type: "compaction", encrypted_content: ciphertext }],
+      } });
+    const events: AdapterEvent[] = [];
+    for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) {
+      events.push(event);
+    }
+    expect(events).toEqual([{ type: "heartbeat" }, { type: "done", compactionEncryptedContent: ciphertext }]);
+    expect(budget.snapshot().currentBytes).toBe(encoder.encode(ciphertext).byteLength);
+    const result = buildResponseJSON(events, "example-model", { compaction: true, translatorBudget: budget });
+    expect(result.output).toEqual([expect.objectContaining({ type: "compaction", encrypted_content: ciphertext })]);
+  });
+
+  test("a suspended heartbeat does not read ahead and return cancels the reader", async () => {
+    const source = upstream();
+    const budget = createTestTranslatorBudget();
+    const iterator = createResponsesPassthroughAdapter(provider).parseStream(new Response(source.body), budget);
+    try {
+      source.send(frame({ type: "response.reasoning_text.delta", content_index: 0, delta: "Hidden reasoning" }).repeat(64));
+      expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } });
+      expect(source.pulls).toBe(0); // Only the already-enqueued chunk was consumed (HWM 0).
+      for (let i = 1; i < 64; i++) {
+        expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } });
+        expect(source.pulls).toBe(0);
+      }
+      await iterator.return(undefined);
+      expect(source.cancelled).toBe(true);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { source.close(); await iterator.return(undefined); }
+  });
+
+  for (const type of ["response.failed", "response.incomplete"]) {
+    test(`${type} after progress never flushes a successful summary`, async () => {
+      const budget = createTestTranslatorBudget();
+      const events: AdapterEvent[] = [];
+      const input = frame({ type: "response.output_text.delta", delta: "Unfinished summary" })
+        + frame({ type, response: type === "response.failed"
+          ? { error: { message: "stopped" } }
+          : { incomplete_details: { reason: "stopped" } } });
+      for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) {
+        events.push(event);
+      }
+      expect(events).toEqual([
+        { type: "heartbeat" },
+        type === "response.failed" ? { type: "error", message: "stopped" } : { type: "incomplete", reason: "stopped" },
+      ]);
+    });
+  }
+});
diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts
index 51f76ab12c..8e92f28715 100644
--- a/tests/responses/responses-compaction-routing.test.ts
+++ b/tests/responses/responses-compaction-routing.test.ts
@@ -4,7 +4,7 @@
  * contract; every other gateway has to be driven as a plain summarizer, or Codex
  * fatals on a compaction turn that came back as an ordinary message.
  */
-import { afterEach, describe, expect, spyOn, test } from "bun:test";
+import { afterEach, describe, expect, jest, spyOn, test } from "bun:test";
 import { mkdtempSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -948,6 +948,57 @@ describe("compact alternate-account attempt (#913)", () => {
     });
   }
 
+  test("native compact headers followed by a stalled body return 504 without retry and release account cleanup", async () => {
+    await withPoolEnv("ocx-compact-body-deadline-", async config => {
+      config.stallTimeoutSec = 2;
+      const readStarted = Promise.withResolvers<void>();
+      let sends = 0;
+      let cancelled = 0;
+      let acceptedBody = false;
+      const body = new ReadableStream<Uint8Array>({
+        pull() { readStarted.resolve(); },
+        cancel() { cancelled++; return new Promise<void>(() => {}); },
+      }, { highWaterMark: 0 });
+      globalThis.fetch = (async () => {
+        sends++;
+        return new Response(body, { headers: { "content-type": "application/json" } });
+      }) as typeof fetch;
+      const releaseSpy = spyOn(authContextModule, "releaseCodexAuthContextProbeLease");
+      const client = new AbortController();
+      // Same scoped, non-concurrent Bun timer control as responses/ws-upstream.test.ts.
+      jest.useFakeTimers();
+      const pending = handleResponsesCompact(
+        compactionRequest({ model: "gpt-5.5", input: [
+          { type: "message", role: "user", content: [{ type: "input_text", text: "earlier turn" }] },
+        ] }, client.signal), config, { model: "", provider: "" },
+        undefined, undefined, { onRequestBodyRead: () => { acceptedBody = true; } },
+      );
+      try {
+        await Promise.race([
+          readStarted.promise,
+          pending.then(response => { throw new Error(`compact returned ${response.status} before reading its body`); }),
+        ]);
+        expect(acceptedBody).toBe(true);
+        expect(sends).toBe(1);
+        jest.advanceTimersByTime(2_000);
+        const response = await pending;
+        expect(response.status).toBe(504);
+        expect(await response.json()).toMatchObject({ error: { code: "upstream_stall_timeout" } });
+        expect(sends).toBe(1);
+        expect(cancelled).toBe(1);
+        expect(body.locked).toBe(false);
+        expect(releaseSpy).toHaveBeenCalledWith(expect.objectContaining({ kind: "pool", accountId: "pool-a" }));
+      } finally {
+        client.abort();
+        try { await pending; } finally {
+          jest.clearAllTimers();
+          jest.useRealTimers();
+          releaseSpy.mockRestore();
+        }
+      }
+    });
+  });
+
   test("canonical trailing slashes are pinned before native compact sends pool credentials", async () => {
     await withPoolEnv("ocx-compact-canonical-url-", async config => {
       config.providers.openai!.baseUrl = "https://chatgpt.com/backend-api/codex///";
diff --git a/tests/responses/responses-compaction.test.ts b/tests/responses/responses-compaction.test.ts
index 631ff48e2b..edf6fec1bb 100644
--- a/tests/responses/responses-compaction.test.ts
+++ b/tests/responses/responses-compaction.test.ts
@@ -1,4 +1,4 @@
-import { describe, expect, test } from "bun:test";
+import { describe, expect, jest, test } from "bun:test";
 import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge";
 import { createResponsesPassthroughAdapter as createResponsesPassthroughAdapterProduction } from "../../src/adapters/openai-responses";
 import { createTranslatorBudget } from "../../src/lib/translator-budget";
@@ -15,10 +15,185 @@ import {
 } from "../../src/responses/compaction";
 import type { AdapterEvent } from "../../src/types";
 import { withTestTranslatorBudget } from "../helpers/translator-budget";
+import { bufferCompactResponse, COMPACT_RESPONSE_MAX_BYTES } from "../../src/server/responses/compact";
 
 const createResponsesPassthroughAdapter = (...args: Parameters<typeof createResponsesPassthroughAdapterProduction>) =>
   withTestTranslatorBudget(createResponsesPassthroughAdapterProduction(...args));
 
+// These non-concurrent tests scope Bun's fake timers like responses/ws-upstream.test.ts.
+// The real bounded-body reader and idleDeadline run; upstream pull acknowledgements
+// synchronize chunk consumption before advancing time, without sleeps or mocking either helper.
+async function withCompactBodyClock(run: () => Promise<void>): Promise<void> {
+  jest.useFakeTimers();
+  try {
+    await run();
+    expect(jest.getTimerCount()).toBe(0);
+  } finally {
+    jest.clearAllTimers();
+    jest.useRealTimers();
+  }
+}
+
+function compactBodySource(onCancel?: () => void) {
+  let controller!: ReadableStreamDefaultController<Uint8Array>;
+  let nextRead = Promise.withResolvers<void>();
+  const cancellationReasons: unknown[] = [];
+  let ended = false;
+  const body = new ReadableStream<Uint8Array>({
+    start(value) { controller = value; },
+    pull() { nextRead.resolve(); },
+    cancel(reason) {
+      ended = true;
+      cancellationReasons.push(reason);
+      onCancel?.();
+      // A deadline must return even if the upstream's cancellation cleanup never finishes.
+      return new Promise<void>(() => {});
+    },
+  }, { highWaterMark: 0 });
+  return {
+    body, cancellationReasons,
+    waitingForRead: () => nextRead.promise,
+    async send(bytes: Uint8Array) {
+      await nextRead.promise;
+      nextRead = Promise.withResolvers<void>();
+      controller.enqueue(bytes);
+      await nextRead.promise;
+    },
+    close() { if (!ended) { ended = true; controller.close(); } },
+  };
+}
+
+describe("native compact response body deadline", () => {
+  test("headers followed by silence expire at the default 300 seconds without waiting for cancel", () => withCompactBodyClock(async () => {
+    const source = compactBodySource();
+    const pending = bufferCompactResponse(new Response(source.body), new AbortController().signal);
+    try {
+      await source.waitingForRead();
+      jest.advanceTimersByTime(299_999);
+      expect(jest.getTimerCount()).toBe(1);
+      expect(source.cancellationReasons).toHaveLength(0);
+      jest.advanceTimersByTime(1);
+      const response = await pending;
+      expect(response.status).toBe(504);
+      expect(await response.json()).toMatchObject({ error: { type: "upstream_stall_timeout", code: "upstream_stall_timeout" } });
+      expect(source.cancellationReasons).toHaveLength(1);
+      expect(source.cancellationReasons[0]).toBeInstanceOf(DOMException);
+      expect((source.cancellationReasons[0] as DOMException).name).toBe("TimeoutError");
+      expect(source.body.locked).toBe(false);
+    } finally { source.close(); await pending; }
+  }));
+
+  test("nonempty chunks rearm the deadline and success preserves exact bytes and header hints", () => withCompactBodyClock(async () => {
+    const source = compactBodySource();
+    const expected = new Uint8Array([0, 255, 128, 195, 40]);
+    const pending = bufferCompactResponse(new Response(source.body, {
+      status: 201, statusText: "Compact ready",
+      headers: {
+        "content-type": "application/octet-stream", "content-length": "999",
+        "retry-after": "42", "x-codex-primary-reset-at": "1900000000",
+        "x-codex-secondary-reset-at": "1900000001", "x-codex-tertiary-reset-at": "1900000002",
+        location: "/compact-result", "set-cookie": "ignored=1", "transfer-encoding": "chunked",
+      },
+    }), new AbortController().signal, 2);
+    try {
+      await source.waitingForRead();
+      for (let i = 0; i < expected.length; i++) {
+        jest.advanceTimersByTime(1_500);
+        await source.send(expected.subarray(i, i + 1));
+      }
+      source.close();
+      const response = await pending;
+      expect(response.status).toBe(201);
+      expect(response.statusText).toBe("Compact ready");
+      expect(new Uint8Array(await response.arrayBuffer())).toEqual(expected);
+      expect(Object.fromEntries(response.headers)).toEqual({
+        "content-type": "application/octet-stream", "retry-after": "42",
+        "x-codex-primary-reset-at": "1900000000", "x-codex-secondary-reset-at": "1900000001",
+        "x-codex-tertiary-reset-at": "1900000002", location: "/compact-result",
+      });
+      expect(source.cancellationReasons).toHaveLength(0);
+      expect(source.body.locked).toBe(false);
+    } finally { source.close(); await pending; }
+  }));
+
+  test("empty chunks do not rearm the byte inactivity deadline", () => withCompactBodyClock(async () => {
+    const source = compactBodySource();
+    const pending = bufferCompactResponse(new Response(source.body), new AbortController().signal, 2);
+    try {
+      await source.waitingForRead();
+      jest.advanceTimersByTime(1_000);
+      await source.send(new Uint8Array(0));
+      jest.advanceTimersByTime(999);
+      expect(source.cancellationReasons).toHaveLength(0);
+      jest.advanceTimersByTime(1);
+      expect((await pending).status).toBe(504);
+      expect(source.cancellationReasons).toHaveLength(1);
+      expect(source.body.locked).toBe(false);
+    } finally { source.close(); await pending; }
+  }));
+
+  for (const idleAlsoFires of [false, true]) {
+    test(`client cancellation unblocks a pending read and wins over idle expiry (${idleAlsoFires})`, () => withCompactBodyClock(async () => {
+      const client = new AbortController();
+      // Abort during the timeout's source-cleanup callback, before the wrapper
+      // classifies its result. Advancing fake time can already flush promises.
+      const source = compactBodySource(idleAlsoFires ? () => client.abort(new Error("client stopped")) : undefined);
+      const pending = bufferCompactResponse(new Response(source.body), client.signal, 2);
+      try {
+        await source.waitingForRead();
+        if (idleAlsoFires) jest.advanceTimersByTime(2_000);
+        else client.abort(new Error("client stopped"));
+        const response = await pending;
+        expect(client.signal.aborted).toBe(true);
+        expect(response.status).toBe(499);
+        expect(await response.json()).toMatchObject({ error: { code: "client_cancelled" } });
+        expect(source.cancellationReasons).toHaveLength(1);
+        expect(source.body.locked).toBe(false);
+      } finally { source.close(); await pending; }
+    }));
+  }
+
+  test("cancellation after a completed timeout does not retroactively replace its 504", () => withCompactBodyClock(async () => {
+    const source = compactBodySource();
+    const client = new AbortController();
+    const pending = bufferCompactResponse(new Response(source.body), client.signal, 2);
+    try {
+      await source.waitingForRead();
+      jest.advanceTimersByTime(2_000);
+      const response = await pending;
+      expect(response.status).toBe(504);
+      client.abort(new Error("late cancellation"));
+      expect(response.status).toBe(504);
+      expect(source.cancellationReasons).toHaveLength(1);
+    } finally { source.close(); await pending; }
+  }));
+
+  test("declared and observed oversize bodies retain the 32 MiB limit without waiting for cancel", () => withCompactBodyClock(async () => {
+    for (const declared of [true, false]) {
+      let cancelled = 0;
+      const body = new ReadableStream<Uint8Array>({
+        pull(controller) { controller.enqueue(new Uint8Array(COMPACT_RESPONSE_MAX_BYTES + 1)); },
+        cancel() { cancelled++; return new Promise<void>(() => {}); },
+      }, { highWaterMark: 0 });
+      const response = await bufferCompactResponse(new Response(body, {
+        headers: declared ? { "content-length": String(COMPACT_RESPONSE_MAX_BYTES + 1) } : {},
+      }), new AbortController().signal, 2);
+      expect(response.status).toBe(502);
+      expect(await response.json()).toMatchObject({ error: { code: "compact_response_too_large" } });
+      expect(cancelled).toBe(1);
+      expect(body.locked).toBe(false);
+    }
+    const atLimit = new Uint8Array(COMPACT_RESPONSE_MAX_BYTES);
+    atLimit[atLimit.length - 1] = 255;
+    const response = await bufferCompactResponse(new Response(atLimit), new AbortController().signal, 2);
+    expect(response.status).toBe(200);
+    const bytes = new Uint8Array(await response.arrayBuffer());
+    expect(bytes.byteLength).toBe(COMPACT_RESPONSE_MAX_BYTES);
+    expect(bytes[0]).toBe(0);
+    expect(bytes[bytes.length - 1]).toBe(255);
+  }));
+});
+
 async function* replay(events: AdapterEvent[]): AsyncGenerator<AdapterEvent> {
   for (const event of events) yield event;
 }
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/server-auth.test.ts b/tests/server/server-auth.test.ts
index 11fddb9772..72a1297f37 100644
--- a/tests/server/server-auth.test.ts
+++ b/tests/server/server-auth.test.ts
@@ -38,7 +38,7 @@ import {
 import { clearRequestLogsForTests, getRequestLogEntries } from "../../src/server/request-log";
 import { readUsageEntries } from "../../src/usage/log";
 import { handleManagementAPI } from "../../src/server/management-api";
-import { handleResponses } from "../../src/server/responses";
+import { handleResponses, handleResponsesCompact } from "../../src/server/responses";
 import type { OcxConfig } from "../../src/types";
 import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt";
 import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home";
@@ -602,6 +602,41 @@ describe("server local API auth", () => {
     })).toBe(false);
   });
 
+  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 cfg = config();
+    cfg.defaultProvider = "fixture";
+    cfg.providers = { fixture: { ...cfg.providers.openai!, disabled: true } };
+    const request = new Request("http://localhost/v1/responses/compact", {
+      method: "POST", headers: { "content-type": "application/json" }, body,
+    });
+    let accepted = 0;
+    const result = handleResponsesCompact(request, cfg, { model: "unknown", provider: "unknown" }, undefined, undefined, {
+      onRequestBodyRead: () => { accepted++; },
+    });
+    bodyController.enqueue(new TextEncoder().encode('{"model":"fixture/gpt-test","input":['));
+    expect(accepted).toBe(0);
+    bodyController.enqueue(new TextEncoder().encode(']}'));
+    bodyController.close();
+    expect((await result).status).toBe(404);
+    expect(accepted).toBe(1);
+  });
+
+  for (const body of ["{", "[]", "{}", '{"model":0}', '{"model":""}']) {
+    test(`compact does not release idle protection for rejected body ${body}`, async () => {
+      let accepted = false;
+      const request = new Request("http://localhost/v1/responses/compact", {
+        method: "POST", headers: { "content-type": "application/json" }, body,
+      });
+      const response = await handleResponsesCompact(request, config(), { model: "unknown", provider: "unknown" }, undefined, undefined, {
+        onRequestBodyRead: () => { accepted = true; },
+      });
+      expect(response.status).toBe(400);
+      expect(accepted).toBe(false);
+    });
+  }
+
   test("responses handler keeps the request timeout until the body is fully accepted", async () => {
     let controller!: ReadableStreamDefaultController<Uint8Array>;
     const body = new ReadableStream<Uint8Array>({