[codex] Tune adaptive stream readiness timeouts

.env.example

@@ -1082,6 +1082,8 @@ CURSOR_USER_AGENT="Cursor/3.4"
 # STREAM_IDLE_TIMEOUT_MS=600000      # Max silence between SSE chunks (default: 600000)
 #                                     # Extended-thinking models rarely pause >90s.
 # STREAM_READINESS_TIMEOUT_MS=80000  # Time to receive the first non-ping SSE event
+# STREAM_READINESS_MAX_TIMEOUT_MS=180000 # Cap for adaptive first-event extensions
+#                                         # (large/tool-heavy/high-reasoning requests).
 
 # ── TLS client (wreq-js fingerprint proxy) ──
 # TLS_CLIENT_TIMEOUT_MS=600000       # Inherits from FETCH_TIMEOUT_MS by default

docs/reference/ENVIRONMENT.md

@@ -591,6 +591,7 @@ REQUEST_TIMEOUT_MS (global override)
 │   └── FETCH_KEEPALIVE_TIMEOUT_MS (independent, default: 4000)
 ├─→ STREAM_IDLE_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 600000)
 ├─→ STREAM_READINESS_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 80000)
+├─→ STREAM_READINESS_MAX_TIMEOUT_MS (caps adaptive readiness extensions, default: 180000)
 └─→ API_BRIDGE_PROXY_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 30000)
     ├─→ API_BRIDGE_SERVER_REQUEST_TIMEOUT_MS (derived, default: 300000)
     ├── API_BRIDGE_SERVER_HEADERS_TIMEOUT_MS (default: 60000)
@@ -604,6 +605,7 @@ REQUEST_TIMEOUT_MS (global override)
 | `FETCH_TIMEOUT_MS`                        | `600000`             | Total HTTP request timeout for upstream provider calls.                                                                                                         |
 | `STREAM_IDLE_TIMEOUT_MS`                  | `600000`             | Max silence between SSE chunks before aborting. Extended-thinking models rarely pause >90s.                                                                     |
 | `STREAM_READINESS_TIMEOUT_MS`             | `80000`              | Time to receive the first non-ping SSE event. Inherits `REQUEST_TIMEOUT_MS` when set.                                                                           |
+| `STREAM_READINESS_MAX_TIMEOUT_MS`         | `180000`             | Maximum adaptive first-event readiness window for large, tool-heavy, or high-reasoning streaming requests.                                                       |
 | `OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS` | _(off)_              | Strip non-standard `codex.*` SSE events (e.g. `codex.rate_limits`) that break the OpenAI SDK's `responses.stream()` with a 502. Set `true`/`1`/`yes` to enable. |
 | `FETCH_HEADERS_TIMEOUT_MS`                | = `FETCH_TIMEOUT_MS` | Time to receive response headers.                                                                                                                               |
 | `FETCH_BODY_TIMEOUT_MS`                   | = `FETCH_TIMEOUT_MS` | Time to receive the full response body.                                                                                                                         |

open-sse/config/constants.ts

@@ -22,6 +22,12 @@ export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs;
 // conservative for large prompts and slow first-byte reasoning providers.
 export const STREAM_READINESS_TIMEOUT_MS = upstreamTimeouts.streamReadinessTimeoutMs;
 
+// Upper bound for adaptive stream readiness extensions (large histories,
+// tool-heavy requests, high-reasoning Codex targets). Override with
+// STREAM_READINESS_MAX_TIMEOUT_MS when an operator needs longer first-event
+// windows for slow-thinking agent workloads.
+export const STREAM_READINESS_MAX_TIMEOUT_MS = upstreamTimeouts.streamReadinessMaxTimeoutMs;
+
 // Error code used when an upstream Antigravity request stalls before response
 // headers are returned. Keep it shared so executor, core normalization and
 // account fallback detection cannot drift.

open-sse/handlers/chatCore.ts

@@ -75,10 +75,7 @@ import {
   withBodyTimeout,
 } from "../utils/stream.ts";
 import { ensureStreamReadiness } from "../utils/streamReadiness.ts";
-import {
-  resolveSuppressThinkClose,
-  THINKING_MARKER_HEADER,
-} from "../utils/thinkCloseMarker.ts";
+import { resolveSuppressThinkClose, THINKING_MARKER_HEADER } from "../utils/thinkCloseMarker.ts";
 import { resolveStreamReadinessTimeout } from "../utils/streamReadinessPolicy.ts";
 import { createStreamController } from "../utils/streamHandler.ts";
 import * as streamFailure from "../utils/streamFailureFinalization.ts";
@@ -127,6 +124,7 @@ import {
   FETCH_BODY_TIMEOUT_MS,
   PROVIDER_MAX_TOKENS,
   STREAM_IDLE_TIMEOUT_MS,
+  STREAM_READINESS_MAX_TIMEOUT_MS,
   STREAM_READINESS_TIMEOUT_MS,
   ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE,
   STREAM_RECOVERY,
@@ -1144,8 +1142,7 @@ export async function handleChatCore({
       }
       // Phase 4A: unified output styles (supersedes cavemanOutputMode via the back-compat shim).
       let outputStyleResult:
-        | import("../services/compression/outputStyles/apply.ts").OutputStylesResult
-        | null = null;
+        import("../services/compression/outputStyles/apply.ts").OutputStylesResult | null = null;
       if (config.enabled) {
         try {
           const { resolveOutputStyleSelection } =
@@ -1197,8 +1194,8 @@ export async function handleChatCore({
           ? ((compressionInputBody as Record<string, unknown>).max_tokens as number)
           : null;
       let adaptiveTelemetry:
-        | import("../services/compression/adaptiveCompression/types.ts").AdaptiveTelemetry
-        | null = null;
+        import("../services/compression/adaptiveCompression/types.ts").AdaptiveTelemetry | null =
+        null;
       const compressionPlan = selectCompressionPlan(
         config,
         compressionComboKey,
@@ -2127,8 +2124,7 @@ export async function handleChatCore({
 
   let onPipelineStreamError: streamFailure.PipelineStreamErrorHandler | null = null;
   let onClientDisconnectFinalize:
-    | ((event: { reason: string; duration: number }) => boolean)
-    | null = null;
+    ((event: { reason: string; duration: number }) => boolean) | null = null;
 
   // Create stream controller for disconnect detection
   const streamController = createStreamController({
@@ -3927,6 +3923,7 @@ export async function handleChatCore({
     provider,
     model,
     body: (finalBody || translatedBody) as Record<string, unknown> | null | undefined,
+    maxTimeoutMs: STREAM_READINESS_MAX_TIMEOUT_MS,
   });
   if (streamReadinessPolicy.timeoutMs !== streamReadinessPolicy.baseTimeoutMs) {
     log?.debug?.(

open-sse/utils/streamReadinessPolicy.ts

@@ -14,7 +14,7 @@ export type StreamReadinessPolicyResult = {
   reasons: string[];
 };
 
-const DEFAULT_MAX_TIMEOUT_MS = 120_000;
+const DEFAULT_MAX_TIMEOUT_MS = 180_000;
 const LARGE_ITEM_THRESHOLD = 150;
 const VERY_LARGE_ITEM_THRESHOLD = 400;
 const TOOL_HEAVY_THRESHOLD = 15;

src/app/api/v1/chat/completions/route.ts

@@ -3,6 +3,8 @@ import { callCloudWithMachineId } from "@/shared/utils/cloud";
 import { handleChat } from "@/sse/handlers/chat";
 import { initTranslators } from "@omniroute/open-sse/translator/index.ts";
 import { createInjectionGuard } from "@/middleware/promptInjectionGuard";
+import { withEarlyStreamKeepalive } from "@omniroute/open-sse/utils/earlyStreamKeepalive";
+import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
 import { checkChatAdmission } from "@/shared/middleware/chatBodyAdmission";
 
 let initPromise = null;
@@ -80,5 +82,13 @@ export async function POST(request) {
     console.error("[SECURITY] Prompt injection guard failed:", error);
   }
 
+  const wantsStreaming = parsedBody?.stream !== false;
+  if (wantsStreaming) {
+    return await withEarlyStreamKeepalive(handleChat(request, null, parsedBody), {
+      signal: request.signal,
+      thresholdMs: resolveKeepaliveThreshold(parsedBody?.model),
+    });
+  }
+
   return await handleChat(request, null, parsedBody);
 }

src/shared/utils/runtimeTimeouts.ts

@@ -11,6 +11,7 @@ export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 600_000;
 export const MAX_TIMER_TIMEOUT_MS = 2_147_483_647;
 export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000;
 export const DEFAULT_STREAM_READINESS_TIMEOUT_MS = 80_000;
+export const DEFAULT_STREAM_READINESS_MAX_TIMEOUT_MS = 180_000;
 export const DEFAULT_FETCH_CONNECT_TIMEOUT_MS = 30_000;
 export const DEFAULT_FETCH_KEEPALIVE_TIMEOUT_MS = 4_000;
 export const DEFAULT_API_BRIDGE_PROXY_TIMEOUT_MS = 600_000;
@@ -29,6 +30,7 @@ export type UpstreamTimeoutConfig = {
   streamIdleTimeoutMs: number;
   sseHeartbeatIntervalMs: number;
   streamReadinessTimeoutMs: number;
+  streamReadinessMaxTimeoutMs: number;
   fetchHeadersTimeoutMs: number;
   fetchBodyTimeoutMs: number;
   fetchConnectTimeoutMs: number;
@@ -103,6 +105,15 @@ export function getUpstreamTimeoutConfig(
       logger,
     }
   );
+  const streamReadinessMaxTimeoutMs = readTimeoutMs(
+    env,
+    "STREAM_READINESS_MAX_TIMEOUT_MS",
+    DEFAULT_STREAM_READINESS_MAX_TIMEOUT_MS,
+    {
+      allowZero: true,
+      logger,
+    }
+  );
   const sseHeartbeatIntervalMs = readTimeoutMs(
     env,
     "SSE_HEARTBEAT_INTERVAL_MS",
@@ -117,6 +128,7 @@ export function getUpstreamTimeoutConfig(
     fetchTimeoutMs,
     streamIdleTimeoutMs,
     streamReadinessTimeoutMs,
+    streamReadinessMaxTimeoutMs,
     sseHeartbeatIntervalMs,
     fetchHeadersTimeoutMs: readTimeoutMs(env, "FETCH_HEADERS_TIMEOUT_MS", fetchTimeoutMs, {
       allowZero: true,

tests/integration/batch-e2e-rate-limit.test.ts

@@ -13,6 +13,19 @@ const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url));
 const RELAY_PORT = await getFreePort();
 const SERVER_PORT = await getFreePort();
 
+type FileUploadResponse = {
+  id?: string;
+};
+
+type BatchResponse = {
+  id?: string;
+  status?: string;
+  request_counts?: {
+    completed?: number;
+    failed?: number;
+  };
+};
+
 function getFreePort() {
   return new Promise<number>((resolve, reject) => {
     const server = net.createServer();
@@ -34,6 +47,47 @@ function sleep(ms: number) {
   return new Promise((r) => setTimeout(r, ms));
 }
 
+function summarizeText(text: string, maxLength = 800) {
+  const compact = text.replace(/\s+/g, " ").trim();
+  return compact.length > maxLength ? `${compact.slice(0, maxLength)}...` : compact;
+}
+
+function formatServerTail(proc: ReturnType<typeof createServerProcess>) {
+  return [
+    "--- stdout ---",
+    ...proc.stdoutLines.slice(-40),
+    "--- stderr ---",
+    ...proc.stderrLines.slice(-40),
+  ].join("\n");
+}
+
+async function readJsonForTest<T>(
+  response: Response,
+  label: string,
+  proc: ReturnType<typeof createServerProcess>
+): Promise<T> {
+  const text = await response.text();
+  let body: T;
+  try {
+    body = JSON.parse(text) as T;
+  } catch {
+    throw new Error(
+      [
+        `${label} returned invalid JSON (${response.status} ${response.statusText}, content-type=${response.headers.get("content-type") || "unknown"})`,
+        summarizeText(text),
+        formatServerTail(proc),
+      ].join("\n")
+    );
+  }
+
+  assert.equal(
+    response.status,
+    200,
+    `${label} failed (${response.status}): ${JSON.stringify(body)}`
+  );
+  return body;
+}
+
 /* ---------- Fake embedding relay ---------- */
 function createFakeEmbeddingRelay() {
   let requestCount = 0;
@@ -162,35 +216,38 @@ function createServerProcess() {
 
 async function waitForServer(baseUrl: string, proc: ReturnType<typeof createServerProcess>) {
   const startedAt = Date.now();
-  while (Date.now() - startedAt < 120_000) {
+  const readinessTimeoutMs = 240_000;
+  const probeTimeoutMs = 15_000;
+  let lastReadiness = "";
+  while (Date.now() - startedAt < readinessTimeoutMs) {
     if (proc.exitInfo) {
       throw new Error(
         [
           `Server exited early (code=${proc.exitInfo.code}, signal=${proc.exitInfo.signal})`,
-          "--- stdout ---",
-          ...proc.stdoutLines.slice(-40),
-          "--- stderr ---",
-          ...proc.stderrLines.slice(-40),
+          formatServerTail(proc),
         ].join("\n")
       );
     }
     try {
-      const resp = await fetch(`${baseUrl}/api/health/ping`, {
-        signal: AbortSignal.timeout(5_000),
-      });
-      if (resp.ok) return;
-    } catch {
+      for (const readinessPath of ["/api/health/ping", "/api/monitoring/health"]) {
+        const resp = await fetch(`${baseUrl}${readinessPath}`, {
+          signal: AbortSignal.timeout(probeTimeoutMs),
+        });
+        if (resp.ok) return;
+        const body = await resp.text().catch(() => "");
+        lastReadiness = `${readinessPath} -> ${resp.status}: ${summarizeText(body, 200)}`;
+      }
+    } catch (error) {
+      lastReadiness = error instanceof Error ? error.message : String(error);
       // not ready yet
     }
     await sleep(500);
   }
   throw new Error(
     [
       "Timed out waiting for server",
-      "--- stdout ---",
-      ...proc.stdoutLines.slice(-40),
-      "--- stderr ---",
-      ...proc.stderrLines.slice(-40),
+      `Last readiness probe: ${lastReadiness}`,
+      formatServerTail(proc),
     ].join("\n")
   );
 }
@@ -293,9 +350,12 @@ test("batch E2E: upload file, create batch, verify rate-limit logs appear", asyn
     method: "POST",
     body: formData,
   });
-  const uploadText = await uploadResp.text();
-  assert.equal(uploadResp.status, 200, `File upload failed (${uploadResp.status}): ${uploadText}`);
-  const uploadBody = JSON.parse(uploadText);
+  assert.match(
+    uploadResp.headers.get("content-type") || "",
+    /json/i,
+    "File upload should return JSON"
+  );
+  const uploadBody = await readJsonForTest<FileUploadResponse>(uploadResp, "File upload", app);
   const fileId = uploadBody.id;
   assert.ok(fileId, "file id missing from upload response");
 
@@ -309,22 +369,35 @@ test("batch E2E: upload file, create batch, verify rate-limit logs appear", asyn
       completion_window: "24h",
     }),
   });
-  const batchText = await batchResp.text();
-  assert.equal(batchResp.status, 200, `Batch creation failed (${batchResp.status}): ${batchText}`);
-  const batchBody = JSON.parse(batchText);
+  const batchBody = await readJsonForTest<BatchResponse>(batchResp, "Batch creation", app);
   const batchId = batchBody.id;
   assert.ok(batchId, "batch id missing from create response");
 
   // 3. Poll for batch completion
   let batchStatus = "";
   let attempts = 0;
+  let lastPollSummary = "";
   const maxAttempts = 120;
   while (attempts < maxAttempts) {
     await sleep(2_000);
     attempts++;
     const sr = await fetch(`${app.baseUrl}/api/v1/batches/${batchId}`);
-    const sb = (await sr.json()) as any;
-    batchStatus = sb.status;
+    const text = await sr.text();
+    let sb: BatchResponse;
+    try {
+      sb = JSON.parse(text);
+    } catch {
+      lastPollSummary = `poll ${attempts} returned invalid JSON (${sr.status} ${sr.statusText}, content-type=${sr.headers.get("content-type") || "unknown"}): ${summarizeText(text, 300)}`;
+      console.warn(`[poll ${attempts}] ${lastPollSummary}`);
+      continue;
+    }
+    if (!sr.ok) {
+      lastPollSummary = `poll ${attempts} failed (${sr.status} ${sr.statusText}): ${JSON.stringify(sb)}`;
+      console.warn(`[poll ${attempts}] ${lastPollSummary}`);
+      continue;
+    }
+    batchStatus = sb.status || "";
+    lastPollSummary = `poll ${attempts} status=${batchStatus}`;
     console.log(
       `[poll ${attempts}] batch ${batchId} status=${batchStatus} completed=${sb.request_counts?.completed} failed=${sb.request_counts?.failed}`
     );
@@ -334,6 +407,7 @@ test("batch E2E: upload file, create batch, verify rate-limit logs appear", asyn
     batchStatus,
     "completed",
     `Batch did not complete; final status: ${batchStatus}. ` +
+      `Last poll: ${lastPollSummary}\n` +
       `Server [BATCH] logs:\n${[...app.stdoutLines, ...app.stderrLines].filter((l) => l.includes("[BATCH]")).join("\n")}`
   );
 
@@ -360,7 +434,7 @@ test("batch E2E: upload file, create batch, verify rate-limit logs appear", asyn
 
   // 5. Verify batch results
   const finalResp = await fetch(`${app.baseUrl}/api/v1/batches/${batchId}`);
-  const finalBody = (await finalResp.json()) as any;
+  const finalBody = await readJsonForTest<BatchResponse>(finalResp, "Final batch fetch", app);
   assert.equal(
     finalBody.request_counts?.completed,
     2,