Use OpenAI chunks for early chat keepalives (#7136)

* Use OpenAI chunks for early chat keepalives

* Update keepalive assertion to match chat completion chunk format

---------

Co-authored-by: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com>

open-sse/utils/earlyStreamKeepalive.ts

@@ -28,6 +28,12 @@
 
 const ENCODER = new TextEncoder();
 const KEEPALIVE_FRAME = ENCODER.encode(": omniroute-keepalive\n\n");
+// OpenAI-compatible keepalive: a syntactically valid empty streaming chunk.
+// Some OpenAI-compatible clients parse every non-empty SSE line as JSON and
+// reject legal SSE comments before their first provider chunk arrives.
+export const OPENAI_KEEPALIVE_FRAME = ENCODER.encode(
+  'data: {"id":"omniroute-keepalive","object":"chat.completion.chunk","created":0,"model":"omniroute","choices":[{"index":0,"delta":{},"finish_reason":null}]}\n\n'
+);
 // Anthropic Messages-format keepalive: a REAL `ping` SSE event, not a comment.
 // Anthropic clients (Claude Code, the Anthropic SDK) reset their stream/first-token
 // watchdog on real SSE events but ignore SSE comments (`: ...`), so on a slow first

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

@@ -5,7 +5,10 @@ import { generateRequestId } from "@/shared/utils/requestId";
 import { initTranslators } from "@omniroute/open-sse/translator/index.ts";
 import { createInjectionGuard } from "@/middleware/promptInjectionGuard";
 import { acceptHeaderForcesStream } from "@omniroute/open-sse/utils/aiSdkCompat.ts";
-import { withEarlyStreamKeepalive } from "@omniroute/open-sse/utils/earlyStreamKeepalive";
+import {
+  OPENAI_KEEPALIVE_FRAME,
+  withEarlyStreamKeepalive,
+} from "@omniroute/open-sse/utils/earlyStreamKeepalive";
 import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
 import { checkChatAdmission } from "@/shared/middleware/chatBodyAdmission";
 import {
@@ -132,6 +135,7 @@ export async function POST(request) {
       {
         signal: request.signal,
         thresholdMs: resolveKeepaliveThreshold(parsedBody?.model),
+        keepaliveFrame: OPENAI_KEEPALIVE_FRAME,
         extraHeaders: { "X-Correlation-Id": reqId },
       }
     );

tests/unit/chat-combo-live-test.test.ts

@@ -274,7 +274,10 @@ test("chat completions route emits early keepalive while waiting for stream read
   assert.match(response.headers.get("content-type") || "", /text\/event-stream/);
 
   const body = await readAll(response);
-  assert.match(body, /: omniroute-keepalive/);
+  assert.match(
+    body,
+    /data: \{"id":"omniroute-keepalive","object":"chat\.completion\.chunk"/
+  );
   assert.match(body, /OK/);
   assert.match(body, /\[DONE\]/);
 });

tests/unit/early-stream-keepalive.test.ts

@@ -4,6 +4,7 @@ import assert from "node:assert/strict";
 import {
   withEarlyStreamKeepalive,
   ANTHROPIC_PING_FRAME,
+  OPENAI_KEEPALIVE_FRAME,
 } from "../../open-sse/utils/earlyStreamKeepalive.ts";
 
 async function readAll(response: Response): Promise<string> {
@@ -68,9 +69,40 @@ test("ANTHROPIC_PING_FRAME is a real Anthropic ping event (not a comment)", () =
   assert.doesNotMatch(decoded, /^:/, "must not be an SSE comment");
 });
 
+test("OPENAI_KEEPALIVE_FRAME is a JSON-parseable OpenAI streaming chunk", () => {
+  const decoded = new TextDecoder().decode(OPENAI_KEEPALIVE_FRAME);
+  assert.match(decoded, /^data: /);
+  assert.doesNotMatch(decoded, /^:/, "must not be an SSE comment");
+
+  const payload = JSON.parse(decoded.slice("data: ".length).trim());
+  assert.equal(payload.object, "chat.completion.chunk");
+  assert.deepEqual(payload.choices, [{ index: 0, delta: {}, finish_reason: null }]);
+});
+
+test("slow handler emits the custom OpenAI keepalive chunk before the body", async () => {
+  const slow = new Promise<Response>((resolve) => {
+    setTimeout(() => resolve(sseResponse("data: [DONE]\n\n")), 120);
+  });
+
+  const result = await withEarlyStreamKeepalive(slow, {
+    thresholdMs: 25,
+    intervalMs: 20,
+    keepaliveFrame: OPENAI_KEEPALIVE_FRAME,
+  });
+
+  const body = await readAll(result);
+  assert.doesNotMatch(body, /: omniroute-keepalive/);
+  const firstFrame = body.split("\n\n")[0];
+  assert.doesNotThrow(() => JSON.parse(firstFrame.slice("data: ".length)));
+  assert.match(body, /data: \[DONE\]/);
+});
+
 test("slow handler emits the custom keepaliveFrame (Anthropic ping) before the body", async () => {
   const slow = new Promise<Response>((resolve) => {
-    setTimeout(() => resolve(sseResponse("event: message_start\ndata: {}\n\ndata: [DONE]\n\n")), 120);
+    setTimeout(
+      () => resolve(sseResponse("event: message_start\ndata: {}\n\ndata: [DONE]\n\n")),
+      120
+    );
   });
 
   const result = await withEarlyStreamKeepalive(slow, {