fix(stream): harden responses SSE keepalives (#1146)

Integrated into release/v3.6.2

open-sse/handlers/responsesHandler.ts

@@ -7,6 +7,7 @@ import { getCorsOrigin } 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";
 
 /**
  * Handle /v1/responses request
@@ -19,6 +20,7 @@ import { createResponsesApiTransformStream } from "../transformer/responsesTrans
  * @param {function} options.onRequestSuccess - Callback when request succeeds
  * @param {function} options.onDisconnect - Callback when client disconnects
  * @param {string} options.connectionId - Connection ID for usage tracking
+ * @param {AbortSignal} [options.signal] - Abort signal for request/disconnect cleanup
  * @returns {Promise<{success: boolean, response?: Response, status?: number, error?: string}>}
  */
 export async function handleResponsesCore({
@@ -30,6 +32,7 @@ export async function handleResponsesCore({
   onRequestSuccess,
   onDisconnect,
   connectionId,
+  signal,
 }) {
   // Convert Responses API format to Chat Completions format
   const convertedBody = convertResponsesApiFormat(body);
@@ -66,7 +69,9 @@ export async function handleResponsesCore({
 
   // Transform SSE stream to Responses API format (no logging in worker)
   const transformStream = createResponsesApiTransformStream(null);
-  const transformedBody = response.body.pipeThrough(transformStream);
+  const transformedBody = response.body
+    .pipeThrough(transformStream)
+    .pipeThrough(createSseHeartbeatTransform({ signal }));
 
   return {
     success: true,

open-sse/utils/sseHeartbeat.ts

@@ -0,0 +1,51 @@
+const DEFAULT_INTERVAL_MS = 15_000;
+
+type SseHeartbeatTransformOptions = {
+  intervalMs?: number;
+  signal?: AbortSignal;
+};
+
+export function createSseHeartbeatTransform({
+  intervalMs = DEFAULT_INTERVAL_MS,
+  signal,
+}: SseHeartbeatTransformOptions = {}) {
+  let intervalId: ReturnType<typeof setInterval> | undefined;
+  const encoder = new TextEncoder();
+
+  const stop = () => {
+    if (!intervalId) return;
+    clearInterval(intervalId);
+    intervalId = undefined;
+  };
+
+  return new TransformStream<Uint8Array, Uint8Array>({
+    start(controller) {
+      intervalId = setInterval(() => {
+        if (signal?.aborted) {
+          stop();
+          return;
+        }
+
+        try {
+          controller.enqueue(encoder.encode(`: keepalive ${new Date().toISOString()}\n\n`));
+        } catch {
+          stop();
+        }
+      }, intervalMs);
+
+      if (intervalId && typeof intervalId === "object" && "unref" in intervalId) {
+        intervalId.unref?.();
+      }
+
+      signal?.addEventListener("abort", stop, { once: true });
+    },
+
+    transform(chunk, controller) {
+      controller.enqueue(chunk);
+    },
+
+    flush() {
+      stop();
+    },
+  });
+}

open-sse/utils/stream.ts

@@ -158,6 +158,7 @@ export function createSSEStream(options: StreamOptions = {}) {
   /** Passthrough: accumulate tool_calls deltas for call log responseBody */
   const passthroughToolCalls = new Map<string, ToolCall>();
   let passthroughToolCallSeq = 0;
+  let skipPassthroughEvent = false;
 
   // State for translate mode (accumulatedContent for call log response body)
   const state: TranslateState | null =
@@ -240,6 +241,20 @@ export function createSSEStream(options: StreamOptions = {}) {
             let injectedUsage = false;
             let clientPayload: unknown = null;
 
+            if (skipPassthroughEvent) {
+              if (!trimmed) {
+                skipPassthroughEvent = false;
+              }
+              continue;
+            }
+
+            // Drop whole keepalive event blocks — strict OpenAI-compatible SDKs
+            // try to JSON.parse empty keepalive payloads and crash.
+            if (/^event:\s*keepalive\b/i.test(trimmed)) {
+              skipPassthroughEvent = true;
+              continue;
+            }
+
             if (trimmed.startsWith("data:")) {
               const providerPayload = parseSSELine(trimmed);
               if (providerPayload) {
@@ -672,12 +687,15 @@ export function createSSEStream(options: StreamOptions = {}) {
           if (remaining) buffer += remaining;
 
           if (mode === STREAM_MODE.PASSTHROUGH) {
-            if (buffer) {
+            const bufferedLine = buffer.trim();
+            if (skipPassthroughEvent || /^event:\s*keepalive\b/i.test(bufferedLine)) {
+              skipPassthroughEvent = false;
+            } else if (buffer) {
               let output = buffer;
               if (buffer.startsWith("data:") && !buffer.startsWith("data: ")) {
                 output = "data: " + buffer.slice(5);
               }
-              const bufferedPayload = parseSSELine(buffer.trim());
+              const bufferedPayload = parseSSELine(bufferedLine);
               if (bufferedPayload) {
                 providerPayloadCollector.push(bufferedPayload);
                 clientPayloadCollector.push(bufferedPayload);

tests/unit/responses-handler.test.mjs

@@ -74,6 +74,7 @@ async function invokeResponsesCore({
   model = "gpt-4o-mini",
   credentials,
   responseFactory,
+  signal,
 } = {}) {
   const calls = [];
 
@@ -101,6 +102,7 @@ async function invokeResponsesCore({
       onRequestSuccess: null,
       onDisconnect: null,
       connectionId: null,
+      signal,
     });
 
     return { result, calls, call: calls.at(-1) };
@@ -233,3 +235,99 @@ test("handleResponsesCore rejects invalid Responses API input that cannot be tra
       error.message.includes("web_search_preview tool type is not supported")
   );
 });
+
+test("handleResponsesCore injects SSE keepalive comments for Responses streams", async () => {
+  const originalSetInterval = globalThis.setInterval;
+  const originalClearInterval = globalThis.clearInterval;
+  const intervals = [];
+  let nextId = 0;
+
+  globalThis.setInterval = (callback, delay = 0, ...args) => {
+    const interval = {
+      id: ++nextId,
+      callback,
+      delay,
+      args,
+      cleared: false,
+    };
+    intervals.push(interval);
+    return interval;
+  };
+
+  globalThis.clearInterval = (interval) => {
+    if (interval && typeof interval === "object") {
+      interval.cleared = true;
+    }
+  };
+
+  try {
+    const { result } = await invokeResponsesCore({
+      body: {
+        model: "gpt-4o-mini",
+        input: "hello",
+      },
+    });
+
+    assert.equal(result.success, true);
+    const heartbeatInterval = intervals.find((interval) => interval.delay === 15000);
+    assert.ok(heartbeatInterval, "expected a 15s heartbeat interval");
+
+    await heartbeatInterval.callback(...heartbeatInterval.args);
+    const sse = await result.response.text();
+
+    assert.match(sse, /^: keepalive .*$/m);
+    assert.match(sse, /event: response\.created/);
+    assert.match(sse, /data: \[DONE\]/);
+    assert.equal(heartbeatInterval.cleared, true);
+  } finally {
+    globalThis.setInterval = originalSetInterval;
+    globalThis.clearInterval = originalClearInterval;
+  }
+});
+
+test("handleResponsesCore clears heartbeat timers immediately when the request signal aborts", async () => {
+  const originalSetInterval = globalThis.setInterval;
+  const originalClearInterval = globalThis.clearInterval;
+  const intervals = [];
+  let nextId = 0;
+
+  globalThis.setInterval = (callback, delay = 0, ...args) => {
+    const interval = {
+      id: ++nextId,
+      callback,
+      delay,
+      args,
+      cleared: false,
+    };
+    intervals.push(interval);
+    return interval;
+  };
+
+  globalThis.clearInterval = (interval) => {
+    if (interval && typeof interval === "object") {
+      interval.cleared = true;
+    }
+  };
+
+  try {
+    const controller = new AbortController();
+    const { result } = await invokeResponsesCore({
+      body: {
+        model: "gpt-4o-mini",
+        input: "hello",
+      },
+      signal: controller.signal,
+    });
+
+    assert.equal(result.success, true);
+    const heartbeatInterval = intervals.find((interval) => interval.delay === 15000);
+    assert.ok(heartbeatInterval, "expected a 15s heartbeat interval");
+
+    controller.abort();
+    assert.equal(heartbeatInterval.cleared, true);
+    await result.response.body?.cancel();
+  } finally {
+    globalThis.setInterval = originalSetInterval;
+    globalThis.clearInterval = originalClearInterval;
+  }
+});

tests/unit/sse-heartbeat.test.mjs

@@ -0,0 +1,87 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+const { createSseHeartbeatTransform } = await import("../../open-sse/utils/sseHeartbeat.ts");
+
+function withFakeIntervals(fn) {
+  const originalSetInterval = globalThis.setInterval;
+  const originalClearInterval = globalThis.clearInterval;
+  const intervals = [];
+  let nextId = 0;
+
+  globalThis.setInterval = (callback, delay = 0, ...args) => {
+    const interval = {
+      id: ++nextId,
+      callback,
+      delay,
+      args,
+      cleared: false,
+    };
+    intervals.push(interval);
+    return interval;
+  };
+
+  globalThis.clearInterval = (interval) => {
+    if (interval && typeof interval === "object") {
+      interval.cleared = true;
+    }
+  };
+
+  return Promise.resolve()
+    .then(() => fn(intervals))
+    .finally(() => {
+      globalThis.setInterval = originalSetInterval;
+      globalThis.clearInterval = originalClearInterval;
+    });
+}
+
+function decodeChunk(value) {
+  return typeof value === "string" ? value : new TextDecoder().decode(value);
+}
+
+test("createSseHeartbeatTransform emits SSE comments while preserving stream output", async () => {
+  await withFakeIntervals(async (intervals) => {
+    const transform = createSseHeartbeatTransform({ intervalMs: 250 });
+    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 writer.write(new TextEncoder().encode('data: {"chunk":"one"}\n\n'));
+
+    assert.equal(intervals.length, 1);
+    assert.equal(intervals[0].delay, 250);
+
+    await intervals[0].callback(...intervals[0].args);
+    await writer.close();
+    await pump;
+
+    assert.equal(emitted[0], 'data: {"chunk":"one"}\n\n');
+    assert.match(emitted[1], /^: keepalive /);
+    assert.equal(intervals[0].cleared, true);
+  });
+});
+
+test("createSseHeartbeatTransform clears the interval when aborted", async () => {
+  await withFakeIntervals(async (intervals) => {
+    const controller = new AbortController();
+    const transform = createSseHeartbeatTransform({ signal: controller.signal });
+    const reader = transform.readable.getReader();
+    const writer = transform.writable.getWriter();
+
+    assert.equal(intervals.length, 1);
+    assert.equal(intervals[0].cleared, false);
+
+    controller.abort();
+    assert.equal(intervals[0].cleared, true);
+
+    await writer.close();
+    await reader.cancel();
+  });
+});

tests/unit/stream-utils.test.mjs

@@ -704,3 +704,39 @@ test("createStructuredSSECollector drops excess events and compactStructuredStre
     },
   });
 });
+
+test("createSSEStream passthrough drops keepalive event blocks without losing Responses deltas", async () => {
+  const text = await readTransformed(
+    [
+      "event: keepalive\ndata:\n\n",
+      `data: ${JSON.stringify({
+        type: "response.output_text.delta",
+        delta: "Hello keepalive-safe",
+      })}\n\n`,
+      `data: ${JSON.stringify({
+        type: "response.completed",
+        response: {
+          id: "resp_keepalive",
+          object: "response",
+          model: "gpt-4.1-mini",
+          status: "completed",
+          usage: { input_tokens: 2, output_tokens: 1, total_tokens: 3 },
+        },
+      })}\n\n`,
+      "data: [DONE]\n\n",
+    ],
+    {
+      mode: "passthrough",
+      sourceFormat: FORMATS.OPENAI_RESPONSES,
+      provider: "openai",
+      model: "gpt-4.1-mini",
+      body: { input: "hello" },
+    }
+  );
+
+  assert.equal(text.includes("event: keepalive"), false);
+  assert.equal(text.includes("data:\n\n"), false);
+  assert.match(text, /response\.output_text\.delta/);
+  assert.match(text, /Hello keepalive-safe/);
+  assert.match(text, /data: \[DONE\]/);
+});