fix(sse-heartbeat): shape-aware keepalives keep streams alive through stricter proxies (#2233)

Integrated into release/v3.8.0 with idle timeout default reverted to 600s

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.

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 {

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,

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<Uint8Array, Uint8Array> {
+  if (!Number.isFinite(intervalMs) || intervalMs <= 0) {
+    return new TransformStream<Uint8Array, Uint8Array>();
+  }
+
   let intervalId: ReturnType<typeof setInterval> | 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();
         }

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,

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({});
 

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();
+});

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}`);
+      }
+    });
+  }
+});