Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion open-sse/handlers/responsesHandler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ export async function handleResponsesCore({
createSseHeartbeatTransform({
signal,
intervalMs: SSE_HEARTBEAT_INTERVAL_MS,
shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS,
shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_KEEPALIVE,
})
);

Expand Down
16 changes: 8 additions & 8 deletions open-sse/utils/sseHeartbeat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,20 +6,20 @@
* - [2026-07-28] [Cursor Grok 4.5] - Brand-neutral default OpenAI keepalive id/model
*/
const HEARTBEAT_ENCODER = new TextEncoder();
const OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD = 'data: {"type":"response.in_progress"}\n\n';
const OPENAI_RESPONSES_KEEPALIVE_PAYLOAD = ": keepalive\n\n";

export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000;

/** Shared Responses API heartbeat frame for early and mid-stream keepalives. */
export const OPENAI_RESPONSES_IN_PROGRESS_FRAME = HEARTBEAT_ENCODER.encode(
OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD
/** Shared Responses API comment frame for early and mid-stream keepalives. */
export const OPENAI_RESPONSES_KEEPALIVE_FRAME = HEARTBEAT_ENCODER.encode(
OPENAI_RESPONSES_KEEPALIVE_PAYLOAD
);

export const HEARTBEAT_SHAPES = {
COMMENT: "comment",
ANTHROPIC_PING: "anthropic-ping",
OPENAI_CHUNK: "openai-chunk",
OPENAI_RESPONSES_IN_PROGRESS: "openai-responses-in-progress",
OPENAI_RESPONSES_KEEPALIVE: "openai-responses-keepalive",
} as const;

export type HeartbeatShape = (typeof HEARTBEAT_SHAPES)[keyof typeof HEARTBEAT_SHAPES];
Expand All @@ -35,7 +35,7 @@ export function shapeForClientFormat(
case "openai":
return HEARTBEAT_SHAPES.OPENAI_CHUNK;
case "openai-responses":
return HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS;
return HEARTBEAT_SHAPES.OPENAI_RESPONSES_KEEPALIVE;
default:
return HEARTBEAT_SHAPES.COMMENT;
}
Expand All @@ -48,8 +48,8 @@ function buildHeartbeatPayload(
switch (shape) {
case HEARTBEAT_SHAPES.ANTHROPIC_PING:
return 'event: ping\ndata: {"type":"ping"}\n\n';
case HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS:
return OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD;
case HEARTBEAT_SHAPES.OPENAI_RESPONSES_KEEPALIVE:
return OPENAI_RESPONSES_KEEPALIVE_PAYLOAD;
case HEARTBEAT_SHAPES.OPENAI_CHUNK: {
const payload = {
id: opts.chunkId ?? "chatcmpl-keepalive",
Expand Down
9 changes: 2 additions & 7 deletions src/app/api/v1/responses/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,14 @@ import {
releaseChatAdmissionWhenDone,
resolveSessionId,
} from "@/shared/middleware/chatBodyAdmission";
import { SSE_HEARTBEAT_INTERVAL_MS } from "@omniroute/open-sse/config/constants";
import { resolveStreamFlag } from "@omniroute/open-sse/utils/aiSdkCompat";
import { errorResponse } from "@omniroute/open-sse/utils/error";
import {
withEarlyStreamKeepalive,
OPENAI_RESPONSES_ERROR_FRAME,
} from "@omniroute/open-sse/utils/earlyStreamKeepalive";
import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
import { OPENAI_RESPONSES_IN_PROGRESS_FRAME } from "@omniroute/open-sse/utils/sseHeartbeat";
import { OPENAI_RESPONSES_KEEPALIVE_FRAME } from "@omniroute/open-sse/utils/sseHeartbeat";

// NOTE: We do NOT call initTranslators() here — the translator registry is
// bootstrapped at module level inside open-sse/translator/index.ts when it
Expand Down Expand Up @@ -192,11 +191,7 @@ async function postHandler(request: any) {
return await withEarlyStreamKeepalive(handlerResponse, {
signal: request.signal,
thresholdMs,
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
applicationKeepalive: {
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
intervalMs: SSE_HEARTBEAT_INTERVAL_MS,
},
startupFrame: OPENAI_RESPONSES_KEEPALIVE_FRAME,
errorFrame: OPENAI_RESPONSES_ERROR_FRAME,
correlationId,
});
Expand Down
53 changes: 13 additions & 40 deletions tests/unit/early-stream-keepalive.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@
* @file early-stream-keepalive.test.ts
* @description Unit tests for withEarlyStreamKeepalive (fast/slow path, frames, abort).
*
* @changes
* - [2026-08-16] - Assert Responses startup and recurring keepalives are neutral JSON events
*/
import test from "node:test";
import assert from "node:assert/strict";
Expand All @@ -17,7 +15,7 @@ import {
OPENAI_RESPONSES_ERROR_FRAME,
} from "../../open-sse/utils/earlyStreamKeepalive.ts";
import { takeEarlyKeepaliveBytes } from "../../open-sse/utils/earlyKeepaliveByteBuffer.ts";
import { OPENAI_RESPONSES_IN_PROGRESS_FRAME } from "../../open-sse/utils/sseHeartbeat.ts";
import { OPENAI_RESPONSES_KEEPALIVE_FRAME } from "../../open-sse/utils/sseHeartbeat.ts";

async function readAll(response: Response): Promise<string> {
const reader = response.body!.getReader();
Expand Down Expand Up @@ -174,45 +172,27 @@ test("startupFrame defaults to keepaliveFrame when omitted (no behavior change)"
);
});

// Responses clients need both frequent raw bytes and occasional parsed events while
// upstream readiness is pending. Keep those cadences separate: comments cover the
// short idle-read timeout, while sparse response.in_progress events reset parsers that
// ignore comments without flooding the application event stream.
test("slow Responses handler uses comments plus sparse in_progress events", async () => {
test("slow Responses handler emits comments without fake lifecycle events", async () => {
const slow = new Promise<Response>((resolve) => {
setTimeout(() => resolve(sseResponse('data: {"type":"response.completed"}\n\n')), 900);
});

const result = await withEarlyStreamKeepalive(slow, {
thresholdMs: 20,
intervalMs: 250,
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
applicationKeepalive: {
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
intervalMs: 500,
},
startupFrame: OPENAI_RESPONSES_KEEPALIVE_FRAME,
});

const body = await readAll(result);
const frames = body.split("\n\n").filter(Boolean);
const earlyFrames = frames.slice(0, -1);
assert.equal(earlyFrames[0], 'data: {"type":"response.in_progress"}');
assert.equal(earlyFrames[0], ": keepalive");
assert.ok(earlyFrames.length >= 3, "expected startup and recurring keepalives");
assert.ok(
earlyFrames.some((frame) => frame === ": keepalive"),
"transport ticks must remain lightweight SSE comments"
);
const applicationFrames = earlyFrames.filter((frame) => frame.startsWith("data: "));
assert.ok(applicationFrames.length >= 2, "expected startup and sparse application keepalives");
for (const frame of applicationFrames) {
assert.deepEqual(JSON.parse(frame.slice("data: ".length)), {
type: "response.in_progress",
});
assert.doesNotMatch(frame, /output_item|reasoning|✨/);
}
assert.ok(
applicationFrames.length < earlyFrames.length,
"application events must be sparser than transport heartbeats"
earlyFrames.every((frame) => frame === ": keepalive"),
"keepalives must remain transport-only comments"
);
assert.doesNotMatch(body, /response\.in_progress|output_item|reasoning|✨/);
assert.match(body, /data: {"type":"response.completed"}/, "real upstream body forwarded");
});

Expand All @@ -228,22 +208,15 @@ test("a correlationId records the startup frame and keepalive ticks, but not the
const result = await withEarlyStreamKeepalive(slow, {
thresholdMs: 25,
intervalMs: 250,
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
applicationKeepalive: {
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
intervalMs: 500,
},
startupFrame: OPENAI_RESPONSES_KEEPALIVE_FRAME,

correlationId,
});
await readAll(result);

const recorded = takeEarlyKeepaliveBytes(correlationId).join("");
assert.match(
recorded,
/data: {"type":"response\.in_progress"}/,
"startup frame must be recorded"
);
assert.match(recorded, /: keepalive/, "transport heartbeat must be recorded");
assert.doesNotMatch(recorded, /response\.in_progress/);
assert.doesNotMatch(
recorded,
/event: response\.created/,
Expand All @@ -260,8 +233,8 @@ test("omitting correlationId leaves the buffer untouched (today's behavior, unch
const result = await withEarlyStreamKeepalive(slow, {
thresholdMs: 25,
intervalMs: 20,
keepaliveFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
keepaliveFrame: OPENAI_RESPONSES_KEEPALIVE_FRAME,
startupFrame: OPENAI_RESPONSES_KEEPALIVE_FRAME,
});
await readAll(result);

Expand Down
20 changes: 4 additions & 16 deletions tests/unit/responses-route-early-keepalive-wiring.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,22 +10,10 @@ process.env.DATA_DIR = dataDir;
process.env.REQUIRE_API_KEY = "false";
after(() => fs.rmSync(dataDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }));

test("Responses route wires dual-cadence neutral keepalives", () => {
assert.match(
routeSource,
/startupFrame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME/,
"the immediate frame must not create a synthetic reasoning item"
);
assert.match(
routeSource,
/applicationKeepalive:\s*\{[\s\S]*frame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME,[\s\S]*intervalMs:\s*SSE_HEARTBEAT_INTERVAL_MS/,
"parser-visible events must use the configured mid-stream heartbeat cadence"
);
assert.doesNotMatch(
routeSource,
/keepaliveFrame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME/,
"frequent transport ticks must not all become application events"
);
test("Responses route wires transport-only keepalives", () => {
assert.match(routeSource, /startupFrame:\s*OPENAI_RESPONSES_KEEPALIVE_FRAME/);
assert.doesNotMatch(routeSource, /applicationKeepalive/);
assert.doesNotMatch(routeSource, /OPENAI_RESPONSES_IN_PROGRESS_FRAME/);
assert.doesNotMatch(routeSource, /RESPONSES_STARTUP_THINKING_FRAME/);
});

Expand Down
10 changes: 5 additions & 5 deletions tests/unit/sse-heartbeat.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -151,11 +151,11 @@ test("shape: openai-chunk emits valid chat.completion.chunk with empty delta", a
});
});

test("shape: openai-responses-in-progress emits response.in_progress data event", async () => {
test("shape: openai-responses-keepalive emits a transport comment", async () => {
await withFakeIntervals(async (intervals) => {
const transform = createSseHeartbeatTransform({
intervalMs: 100,
shape: "openai-responses-in-progress",
shape: "openai-responses-keepalive",
});
const writer = transform.writable.getWriter();
const reader = transform.readable.getReader();
Expand All @@ -172,7 +172,7 @@ test("shape: openai-responses-in-progress emits response.in_progress data event"
await writer.close();
await pump;

assert.equal(emitted[0], 'data: {"type":"response.in_progress"}\n\n');
assert.equal(emitted[0], ": keepalive\n\n");
});
});

Expand Down Expand Up @@ -226,14 +226,14 @@ test("intervalMs <= 0 returns passthrough (no setInterval, no heartbeat)", async
test("shapeForClientFormat maps formats correctly", () => {
assert.equal(shapeForClientFormat("claude"), "anthropic-ping");
assert.equal(shapeForClientFormat("openai"), "openai-chunk");
assert.equal(shapeForClientFormat("openai-responses"), "openai-responses-in-progress");
assert.equal(shapeForClientFormat("openai-responses"), "openai-responses-keepalive");
assert.equal(shapeForClientFormat("gemini"), "comment");
assert.equal(shapeForClientFormat(undefined), "comment");
assert.equal(shapeForClientFormat(null), "comment");
});

test("no shape collides with stream.ts event: keepalive strip regex", async () => {
const shapes = ["comment", "anthropic-ping", "openai-chunk", "openai-responses-in-progress"];
const shapes = ["comment", "anthropic-ping", "openai-chunk", "openai-responses-keepalive"];
for (const shape of shapes) {
await withSseCommentsOn(() => withFakeIntervals(async (intervals) => {
const transform = createSseHeartbeatTransform({
Expand Down
7 changes: 4 additions & 3 deletions tests/unit/sseHeartbeat.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ test("shapeForClientFormat maps known client formats", () => {
assert.equal(shapeForClientFormat("openai"), HEARTBEAT_SHAPES.OPENAI_CHUNK);
assert.equal(
shapeForClientFormat("openai-responses"),
HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS
HEARTBEAT_SHAPES.OPENAI_RESPONSES_KEEPALIVE
);
assert.equal(shapeForClientFormat(undefined), HEARTBEAT_SHAPES.COMMENT);
});
Expand All @@ -90,7 +90,8 @@ test("comment opt-out suppresses only COMMENT heartbeats, not data-event heartbe
const anthropic = await collectHeartbeatOutput(HEARTBEAT_SHAPES.ANTHROPIC_PING);
assert.match(anthropic, /event: ping\ndata: \{"type":"ping"\}/);

const responses = await collectHeartbeatOutput(HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS);
assert.match(responses, /data: \{"type":"response\.in_progress"\}/);
const responses = await collectHeartbeatOutput(HEARTBEAT_SHAPES.OPENAI_RESPONSES_KEEPALIVE);
assert.match(responses, /: keepalive/);
assert.doesNotMatch(responses, /data: \{"type":"response\.in_progress"\}/);
});
});
Loading