diff --git a/changelog.d/fixes/10127-early-sse-heartbeat.md b/changelog.d/fixes/10127-early-sse-heartbeat.md new file mode 100644 index 0000000000..4ada9f43a4 --- /dev/null +++ b/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) diff --git a/src/app/api/v1/messages/route.ts b/src/app/api/v1/messages/route.ts index 3e2172b1eb..cbbc87f202 100644 --- a/src/app/api/v1/messages/route.ts +++ b/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); diff --git a/src/app/api/v1/responses/route.ts b/src/app/api/v1/responses/route.ts index 2b1cea8c2a..6d6d5ad60f 100644 --- a/src/app/api/v1/responses/route.ts +++ b/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). diff --git a/tests/unit/early-sse-route-intent.test.ts b/tests/unit/early-sse-route-intent.test.ts new file mode 100644 index 0000000000..b8b48918fc --- /dev/null +++ b/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); + }); +}