fix(streaming): honor body stream intent for early heartbeats (#10127)

changelog.d/fixes/10127-early-sse-heartbeat.md

@@ -0,0 +1 @@
+- **fix(streaming):** start early SSE heartbeats when Responses or Messages requests opt into streaming through the request body (#10127)

src/app/api/v1/messages/route.ts

@@ -7,6 +7,7 @@ import {
   ANTHROPIC_PING_FRAME,
 } from "@omniroute/open-sse/utils/earlyStreamKeepalive";
 import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
+import { resolveStreamFlag } from "@omniroute/open-sse/utils/aiSdkCompat";
 
 let initialized = false;
 
@@ -54,22 +55,27 @@ async function postHandler(request: any, context: any, preParsedBody: any = null
   // /v1/responses (#2544). Anthropic clients ignore SSE comments for their watchdog, so
   // emit a real `event: ping` (ANTHROPIC_PING_FRAME). Non-streaming callers keep the
   // verbatim path.
-  const accept = String(request.headers?.get?.("accept") || "").toLowerCase();
-  if (accept.includes("text/event-stream")) {
-    let model;
+  let body = preParsedBody;
+  if (body == null) {
     try {
-      const body = preParsedBody ?? (await request.clone().json().catch(() => null));
-      model = body?.model;
+      body = await request
+        .clone()
+        .json()
+        .catch(() => null);
     } catch {
-      // body unavailable / non-JSON — fall back to the default keepalive threshold
+      // body unavailable / non-JSON — handleChat will return its normal validation error
     }
-    return await withEarlyStreamKeepalive(handleChat(request, null, preParsedBody), {
+  }
+  const accept = String(request.headers?.get?.("accept") || "");
+  const wantsStreaming = resolveStreamFlag(body?.stream, accept, "claude");
+  if (wantsStreaming) {
+    return await withEarlyStreamKeepalive(handleChat(request, null, body), {
       signal: request.signal,
-      thresholdMs: resolveKeepaliveThreshold(model),
+      thresholdMs: resolveKeepaliveThreshold(body?.model),
       keepaliveFrame: ANTHROPIC_PING_FRAME,
     });
   }
-  return await handleChat(request, null, preParsedBody);
+  return await handleChat(request, null, body);
 }
 
 export const POST = withInjectionGuard(postHandler);

src/app/api/v1/responses/route.ts

@@ -9,6 +9,7 @@ import { resolveResponsesApiModel } from "@/app/api/internal/codex-responses-ws/
 import { getModelInfo } from "@/sse/services/model";
 import { getComboByName } from "@/lib/db/combos";
 import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
+import { resolveStreamFlag } from "@omniroute/open-sse/utils/aiSdkCompat";
 
 // NOTE: We do NOT call initTranslators() here — the translator registry is
 // bootstrapped at module level inside open-sse/translator/index.ts when it
@@ -92,8 +93,9 @@ async function postHandler(request: any, context: any, preParsedBody: any = null
     request,
     preParsedBody
   );
-  const accept = String(request.headers?.get?.("accept") || "").toLowerCase();
-  if (accept.includes("text/event-stream")) {
+  const accept = String(request.headers?.get?.("accept") || "");
+  const wantsStreaming = resolveStreamFlag(resolvedBody?.stream, accept, "openai-responses");
+  if (wantsStreaming) {
     // Adaptive threshold: web-session and anonymous-fallback providers are slower
     // to produce the first byte, so use a longer keepalive threshold (15s vs 2s).
     // Reuse resolvedBody.model — no extra clone/parse needed (#4041).

tests/unit/early-sse-route-intent.test.ts

@@ -0,0 +1,42 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import path from "node:path";
+
+import { resolveStreamFlag } from "../../open-sse/utils/aiSdkCompat.ts";
+
+const ROUTES = [
+  {
+    name: "responses",
+    sourceFormat: "openai-responses",
+    bodyExpression: "resolvedBody?.stream",
+    source: fs.readFileSync(path.join("src", "app", "api", "v1", "responses", "route.ts"), "utf8"),
+  },
+  {
+    name: "messages",
+    sourceFormat: "claude",
+    bodyExpression: "body?.stream",
+    source: fs.readFileSync(path.join("src", "app", "api", "v1", "messages", "route.ts"), "utf8"),
+  },
+] as const;
+
+for (const route of ROUTES) {
+  test(`${route.name} early-heartbeat gate uses the real stream resolver`, () => {
+    const escapedBodyExpression = route.bodyExpression.replace(/[?.]/g, "\\$&");
+    assert.match(
+      route.source,
+      new RegExp(
+        `resolveStreamFlag\\(\\s*${escapedBodyExpression},\\s*accept,\\s*"${route.sourceFormat}"\\s*\\)`
+      )
+    );
+    assert.match(route.source, /if \(wantsStreaming\) \{[\s\S]*withEarlyStreamKeepalive\(/);
+    assert.doesNotMatch(route.source, /accept\.includes\(["']text\/event-stream["']\)/);
+  });
+
+  test(`${route.name} stream intent covers body and Accept without overriding stream:false`, () => {
+    assert.equal(resolveStreamFlag(true, "application/json", route.sourceFormat), true);
+    assert.equal(resolveStreamFlag(false, "text/event-stream", route.sourceFormat), false);
+    assert.equal(resolveStreamFlag(undefined, "text/event-stream", route.sourceFormat), true);
+    assert.equal(resolveStreamFlag(undefined, "application/json", route.sourceFormat), false);
+  });
+}