fix(responses): preserve precommit WebSocket rejection status

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.

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",

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.

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");
           }

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

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

src/server/responses/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(() => "");

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.

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",