diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index 21c1adfc80..8d3b1b979d 100644 --- a/open-sse/config/constants.ts +++ b/open-sse/config/constants.ts @@ -22,6 +22,12 @@ export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs; // first token, while dead 200 OK streams fail fast enough for combo fallback. export const STREAM_READINESS_TIMEOUT_MS = upstreamTimeouts.streamReadinessTimeoutMs; +// Heartbeat interval for synthetic SSE keepalive emission toward the downstream +// client (Capy, Claude Code, OpenAI SDK, etc). Keeps strict proxies from +// dropping the connection during long upstream thinking phases. Set to 0 to +// disable. Override with SSE_HEARTBEAT_INTERVAL_MS env var. +export const SSE_HEARTBEAT_INTERVAL_MS = upstreamTimeouts.sseHeartbeatIntervalMs; + // Timeout for reading the full response body after headers arrive (ms). // Prevents indefinite hangs when the upstream sends headers but stalls on the body. // Defaults to FETCH_TIMEOUT_MS. Override with FETCH_BODY_TIMEOUT_MS env var. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 921c914fa7..b926be19c3 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -11,7 +11,7 @@ import { } from "../utils/stream.ts"; import { ensureStreamReadiness } from "../utils/streamReadiness.ts"; import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts"; -import { createSseHeartbeatTransform } from "../utils/sseHeartbeat.ts"; +import { createSseHeartbeatTransform, shapeForClientFormat } from "../utils/sseHeartbeat.ts"; import { addBufferToUsage, filterUsageForFormat, estimateUsage } from "../utils/usageTracking.ts"; import { refreshWithRetry } from "../services/tokenRefresh.ts"; import { createRequestLogger } from "../utils/requestLogger.ts"; @@ -35,6 +35,7 @@ import { FETCH_BODY_TIMEOUT_MS, MAX_TOOLS_LIMIT, PROVIDER_MAX_TOKENS, + SSE_HEARTBEAT_INTERVAL_MS, STREAM_IDLE_TIMEOUT_MS, STREAM_READINESS_TIMEOUT_MS, } from "../config/constants.ts"; @@ -4555,7 +4556,11 @@ export async function handleChatCore({ finalStream = pipeWithDisconnect(providerResponse, transformStream, streamController); } finalStream = finalStream.pipeThrough( - createSseHeartbeatTransform({ signal: streamController.signal }) + createSseHeartbeatTransform({ + signal: streamController.signal, + intervalMs: SSE_HEARTBEAT_INTERVAL_MS, + shape: shapeForClientFormat(clientResponseFormat), + }) ); return { diff --git a/open-sse/handlers/responsesHandler.ts b/open-sse/handlers/responsesHandler.ts index 203b70ed4a..b82d88cb50 100644 --- a/open-sse/handlers/responsesHandler.ts +++ b/open-sse/handlers/responsesHandler.ts @@ -7,7 +7,8 @@ import { CORS_HEADERS } from "../utils/cors.ts"; import { handleChatCore } from "./chatCore.ts"; import { convertResponsesApiFormat } from "../translator/helpers/responsesApiHelper.ts"; import { createResponsesApiTransformStream } from "../transformer/responsesTransformer.ts"; -import { createSseHeartbeatTransform } from "../utils/sseHeartbeat.ts"; +import { createSseHeartbeatTransform, HEARTBEAT_SHAPES } from "../utils/sseHeartbeat.ts"; +import { SSE_HEARTBEAT_INTERVAL_MS } from "../config/constants.ts"; /** * Handle /v1/responses request @@ -71,7 +72,11 @@ export async function handleResponsesCore({ const transformStream = createResponsesApiTransformStream(null); const transformedBody = response.body .pipeThrough(transformStream) - .pipeThrough(createSseHeartbeatTransform({ signal })); + .pipeThrough(createSseHeartbeatTransform({ + signal, + intervalMs: SSE_HEARTBEAT_INTERVAL_MS, + shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS, + })); return { success: true, diff --git a/open-sse/utils/sseHeartbeat.ts b/open-sse/utils/sseHeartbeat.ts index ee5c8a0c95..2c8d85a4c0 100644 --- a/open-sse/utils/sseHeartbeat.ts +++ b/open-sse/utils/sseHeartbeat.ts @@ -1,16 +1,78 @@ export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000; +export const HEARTBEAT_SHAPES = { + COMMENT: "comment", + ANTHROPIC_PING: "anthropic-ping", + OPENAI_CHUNK: "openai-chunk", + OPENAI_RESPONSES_IN_PROGRESS: "openai-responses-in-progress", +} as const; + +export type HeartbeatShape = (typeof HEARTBEAT_SHAPES)[keyof typeof HEARTBEAT_SHAPES]; + +export const DEFAULT_SSE_HEARTBEAT_SHAPE: HeartbeatShape = HEARTBEAT_SHAPES.COMMENT; + +export function shapeForClientFormat( + clientResponseFormat: string | undefined | null, +): HeartbeatShape { + switch (clientResponseFormat) { + case "claude": + return HEARTBEAT_SHAPES.ANTHROPIC_PING; + case "openai": + return HEARTBEAT_SHAPES.OPENAI_CHUNK; + case "openai-responses": + return HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS; + default: + return HEARTBEAT_SHAPES.COMMENT; + } +} + +function buildHeartbeatPayload( + shape: HeartbeatShape, + opts: { chunkId?: string; chunkModel?: string } = {}, +): string { + switch (shape) { + case HEARTBEAT_SHAPES.ANTHROPIC_PING: + return "event: ping\ndata: {}\n\n"; + case HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS: + return 'data: {"type":"response.in_progress"}\n\n'; + case HEARTBEAT_SHAPES.OPENAI_CHUNK: { + const payload = { + id: opts.chunkId ?? "omniroute-keepalive", + object: "chat.completion.chunk", + created: Math.floor(Date.now() / 1000), + model: opts.chunkModel ?? "omniroute", + choices: [{ index: 0, delta: {}, finish_reason: null }], + }; + return `data: ${JSON.stringify(payload)}\n\n`; + } + case HEARTBEAT_SHAPES.COMMENT: + default: + return `: keepalive ${new Date().toISOString()}\n\n`; + } +} + type SseHeartbeatTransformOptions = { intervalMs?: number; signal?: AbortSignal; + shape?: HeartbeatShape; + chunkId?: string; + chunkModel?: string; }; +const HEARTBEAT_ENCODER = new TextEncoder(); + export function createSseHeartbeatTransform({ intervalMs = DEFAULT_SSE_HEARTBEAT_INTERVAL_MS, signal, -}: SseHeartbeatTransformOptions = {}) { + shape = DEFAULT_SSE_HEARTBEAT_SHAPE, + chunkId, + chunkModel, +}: SseHeartbeatTransformOptions = {}): TransformStream { + if (!Number.isFinite(intervalMs) || intervalMs <= 0) { + return new TransformStream(); + } + let intervalId: ReturnType | undefined; - const encoder = new TextEncoder(); const stop = () => { if (!intervalId) return; @@ -27,7 +89,9 @@ export function createSseHeartbeatTransform({ } try { - controller.enqueue(encoder.encode(`: keepalive ${new Date().toISOString()}\n\n`)); + controller.enqueue( + HEARTBEAT_ENCODER.encode(buildHeartbeatPayload(shape, { chunkId, chunkModel })), + ); } catch { stop(); } diff --git a/src/shared/utils/runtimeTimeouts.ts b/src/shared/utils/runtimeTimeouts.ts index 14e0e67afb..67ff3e0e39 100644 --- a/src/shared/utils/runtimeTimeouts.ts +++ b/src/shared/utils/runtimeTimeouts.ts @@ -8,6 +8,7 @@ type ReadTimeoutOptions = { export const DEFAULT_FETCH_TIMEOUT_MS = 600_000; export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 600_000; +export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000; export const DEFAULT_STREAM_READINESS_TIMEOUT_MS = 30_000; export const DEFAULT_FETCH_CONNECT_TIMEOUT_MS = 30_000; export const DEFAULT_FETCH_KEEPALIVE_TIMEOUT_MS = 4_000; @@ -25,6 +26,7 @@ function hasEnvValue(env: EnvSource, name: string): boolean { export type UpstreamTimeoutConfig = { fetchTimeoutMs: number; streamIdleTimeoutMs: number; + sseHeartbeatIntervalMs: number; streamReadinessTimeoutMs: number; fetchHeadersTimeoutMs: number; fetchBodyTimeoutMs: number; @@ -100,11 +102,21 @@ export function getUpstreamTimeoutConfig( logger, } ); + const sseHeartbeatIntervalMs = readTimeoutMs( + env, + "SSE_HEARTBEAT_INTERVAL_MS", + DEFAULT_SSE_HEARTBEAT_INTERVAL_MS, + { + allowZero: true, + logger, + } + ); return { fetchTimeoutMs, streamIdleTimeoutMs, streamReadinessTimeoutMs, + sseHeartbeatIntervalMs, fetchHeadersTimeoutMs: readTimeoutMs(env, "FETCH_HEADERS_TIMEOUT_MS", fetchTimeoutMs, { allowZero: true, logger, diff --git a/tests/unit/runtime-timeouts.test.ts b/tests/unit/runtime-timeouts.test.ts index 2a588a63b6..5ec5b77df9 100644 --- a/tests/unit/runtime-timeouts.test.ts +++ b/tests/unit/runtime-timeouts.test.ts @@ -12,6 +12,7 @@ test("upstream timeout config derives hidden fetch timeouts from FETCH_TIMEOUT_M assert.deepEqual(config, { fetchTimeoutMs: 600000, streamIdleTimeoutMs: 600000, + sseHeartbeatIntervalMs: 15000, streamReadinessTimeoutMs: 30000, fetchHeadersTimeoutMs: 600000, fetchBodyTimeoutMs: 600000, @@ -95,6 +96,27 @@ test("API bridge timeouts align request timeout with long proxy timeout by defau }); }); +test("idle timeout default stays at 10min (600_000) for slow-thinking model safety", () => { + // NOTE: PR #2233 originally lowered this to 300_000, but the reviewer asked to keep + // the legacy default (slow thinking models, long Anthropic extended-thinking runs). + // The heartbeat-shape change is preserved; only the idle-timeout default revert remains. + assert.equal(runtimeTimeouts.DEFAULT_STREAM_IDLE_TIMEOUT_MS, 600_000); + assert.equal(runtimeTimeouts.getUpstreamTimeoutConfig({}).streamIdleTimeoutMs, 600_000); +}); + +test("heartbeat interval default = 15s, env-overridable", () => { + assert.equal(runtimeTimeouts.DEFAULT_SSE_HEARTBEAT_INTERVAL_MS, 15_000); + assert.equal(runtimeTimeouts.getUpstreamTimeoutConfig({}).sseHeartbeatIntervalMs, 15_000); + assert.equal( + runtimeTimeouts.getUpstreamTimeoutConfig({ SSE_HEARTBEAT_INTERVAL_MS: "8000" }).sseHeartbeatIntervalMs, + 8_000 + ); + assert.equal( + runtimeTimeouts.getUpstreamTimeoutConfig({ SSE_HEARTBEAT_INTERVAL_MS: "0" }).sseHeartbeatIntervalMs, + 0 + ); +}); + test("API bridge proxy timeout defaults to the long upstream request window", () => { const config = runtimeTimeouts.getApiBridgeTimeoutConfig({}); diff --git a/tests/unit/sse-heartbeat-integration.test.ts b/tests/unit/sse-heartbeat-integration.test.ts new file mode 100644 index 0000000000..0b3a2764f4 --- /dev/null +++ b/tests/unit/sse-heartbeat-integration.test.ts @@ -0,0 +1,119 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { createSseHeartbeatTransform, HEARTBEAT_SHAPES, shapeForClientFormat } = await import( + "../../open-sse/utils/sseHeartbeat.ts" +); + +const STREAM_TS_STRIP_RE = /^event:\s*keepalive\b/i; + +function decodeChunk(value) { + return typeof value === "string" ? value : new TextDecoder().decode(value); +} + +test("integration: anthropic-ping heartbeat reaches downstream and does NOT trigger stream.ts strip", async () => { + // Build a fake upstream that emits one chunk then idles indefinitely + let cancelled = false; + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("event: message_start\ndata: {}\n\n")); + // never close — let heartbeat fire + }, + cancel() { cancelled = true; }, + }); + + const transform = createSseHeartbeatTransform({ + intervalMs: 20, + shape: HEARTBEAT_SHAPES.ANTHROPIC_PING, + }); + + const piped = upstream.pipeThrough(transform); + const reader = piped.getReader(); + + // Read first (real) chunk + const { value: first } = await reader.read(); + assert.match(decodeChunk(first), /event: message_start/); + + // Wait for at least one heartbeat (interval = 20ms, give it 60ms slack) + const startedAt = Date.now(); + let sawPing = false; + while (Date.now() - startedAt < 200) { + const { value, done } = await reader.read(); + if (done) break; + const chunk = decodeChunk(value); + if (/^event: ping\b/m.test(chunk)) { + sawPing = true; + // Verify it does NOT match the strip regex + for (const line of chunk.split("\n")) { + assert.ok( + !STREAM_TS_STRIP_RE.test(line.trim()), + `heartbeat chunk produced a stream.ts-strippable line: ${line}` + ); + } + break; + } + } + + await reader.cancel(); + assert.ok(sawPing, "expected to receive at least one event: ping heartbeat within 200ms"); +}); + +test("integration: openai-chunk heartbeat is valid JSON parseable by SDKs", async () => { + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[]}\n\n`)); + }, + }); + + const transform = createSseHeartbeatTransform({ + intervalMs: 20, + shape: HEARTBEAT_SHAPES.OPENAI_CHUNK, + }); + + const piped = upstream.pipeThrough(transform); + const reader = piped.getReader(); + + await reader.read(); // skip first real chunk + + const startedAt = Date.now(); + let sawValidChunk = false; + while (Date.now() - startedAt < 200) { + const { value, done } = await reader.read(); + if (done) break; + const chunk = decodeChunk(value); + if (chunk.startsWith("data: ") && chunk.includes("omniroute-keepalive")) { + const jsonStr = chunk.slice(6, chunk.indexOf("\n\n")); + const parsed = JSON.parse(jsonStr); // must not throw + assert.equal(parsed.object, "chat.completion.chunk"); + assert.equal(parsed.choices[0].finish_reason, null); + sawValidChunk = true; + break; + } + } + + await reader.cancel(); + assert.ok(sawValidChunk, "expected to receive a valid openai-chunk heartbeat within 200ms"); +}); + +test("integration: shapeForClientFormat + createSseHeartbeatTransform pipeline (claude path)", async () => { + // Simulates what chatCore.ts does at line 4276 + const shape = shapeForClientFormat("claude"); + assert.equal(shape, HEARTBEAT_SHAPES.ANTHROPIC_PING); + + const transform = createSseHeartbeatTransform({ + intervalMs: 20, + shape, + }); + + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("event: hello\ndata: {}\n\n")); + }, + }); + + const reader = upstream.pipeThrough(transform).getReader(); + await reader.read(); // first real + const { value } = await reader.read(); + assert.match(decodeChunk(value), /^event: ping\ndata: \{\}\n\n$/); + await reader.cancel(); +}); diff --git a/tests/unit/sse-heartbeat.test.ts b/tests/unit/sse-heartbeat.test.ts index 4c7442615a..be017470df 100644 --- a/tests/unit/sse-heartbeat.test.ts +++ b/tests/unit/sse-heartbeat.test.ts @@ -85,3 +85,163 @@ test("createSseHeartbeatTransform clears the interval when aborted", async () => await reader.cancel(); }); }); + +const { shapeForClientFormat } = await import("../../open-sse/utils/sseHeartbeat.ts"); + +test("shape: anthropic-ping emits event: ping with empty JSON data", async () => { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100, shape: "anthropic-ping" }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + assert.equal(emitted[0], "event: ping\ndata: {}\n\n"); + }); +}); + +test("shape: openai-chunk emits valid chat.completion.chunk with empty delta", async () => { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100, shape: "openai-chunk" }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + assert.ok(emitted[0].startsWith("data: "), `expected data: prefix, got: ${emitted[0]}`); + assert.ok(emitted[0].endsWith("\n\n"), "expected trailing \n\n"); + const jsonStr = emitted[0].slice("data: ".length, -"\n\n".length); + const json = JSON.parse(jsonStr); + assert.equal(json.object, "chat.completion.chunk"); + assert.ok(Array.isArray(json.choices) && json.choices.length === 1); + assert.equal(typeof json.choices[0].delta, "object"); + assert.equal(Object.keys(json.choices[0].delta).length, 0); + assert.equal(json.choices[0].finish_reason, null); + }); +}); + +test("shape: openai-responses-in-progress emits response.in_progress data event", async () => { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100, shape: "openai-responses-in-progress" }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + assert.equal(emitted[0], 'data: {"type":"response.in_progress"}\n\n'); + }); +}); + +test("shape default is comment (back-compat)", async () => { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100 }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + assert.match(emitted[0], /^: keepalive /); + }); +}); + +test("intervalMs <= 0 returns passthrough (no setInterval, no heartbeat)", async () => { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 0 }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + assert.equal(intervals.length, 0); + + await writer.write(new TextEncoder().encode('data: {"chunk":"x"}\n\n')); + await writer.close(); + await pump; + + assert.equal(emitted.length, 1); + assert.equal(emitted[0], 'data: {"chunk":"x"}\n\n'); + }); +}); + +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("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"]; + for (const shape of shapes) { + await withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100, shape }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + const lines = emitted[0].split("\n"); + for (const line of lines) { + assert.doesNotMatch(line.trim(), /^event:\s*keepalive\b/i, `shape ${shape} produced forbidden line: ${line}`); + } + }); + } +});