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

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";
@@ -34,6 +34,7 @@ import {
   HTTP_STATUS,
   MAX_TOOLS_LIMIT,
   PROVIDER_MAX_TOKENS,
+  SSE_HEARTBEAT_INTERVAL_MS,
   STREAM_IDLE_TIMEOUT_MS,
   STREAM_READINESS_TIMEOUT_MS,
 } from "../config/constants.ts";
@@ -4273,7 +4274,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,14 +1,79 @@
 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;
 };
 
 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>({
+      transform(chunk, controller) {
+        controller.enqueue(chunk);
+      },
+    });
+  }
+
   let intervalId: ReturnType<typeof setInterval> | undefined;
   const encoder = new TextEncoder();
 
@@ -27,7 +92,9 @@ export function createSseHeartbeatTransform({
         }
 
         try {
-          controller.enqueue(encoder.encode(`: keepalive ${new Date().toISOString()}\n\n`));
+          controller.enqueue(
+            encoder.encode(buildHeartbeatPayload(shape, { chunkId, chunkModel })),
+          );
         } catch {
           stop();
         }