From bd33b53805cc69f90a161383273f921f8b71d3ff Mon Sep 17 00:00:00 2001 From: gfhfyjbr Date: Sat, 11 Apr 2026 17:36:40 +0300 Subject: [PATCH] fix(stream): harden responses SSE keepalives (#1146) Integrated into release/v3.6.2 --- open-sse/handlers/responsesHandler.ts | 7 +- open-sse/utils/sseHeartbeat.ts | 51 ++++++++++++++ open-sse/utils/stream.ts | 22 +++++- tests/unit/responses-handler.test.mjs | 98 +++++++++++++++++++++++++++ tests/unit/sse-heartbeat.test.mjs | 87 ++++++++++++++++++++++++ tests/unit/stream-utils.test.mjs | 36 ++++++++++ 6 files changed, 298 insertions(+), 3 deletions(-) create mode 100644 open-sse/utils/sseHeartbeat.ts create mode 100644 tests/unit/sse-heartbeat.test.mjs diff --git a/open-sse/handlers/responsesHandler.ts b/open-sse/handlers/responsesHandler.ts index bc74532c5f..1fab9d04fb 100644 --- a/open-sse/handlers/responsesHandler.ts +++ b/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, diff --git a/open-sse/utils/sseHeartbeat.ts b/open-sse/utils/sseHeartbeat.ts new file mode 100644 index 0000000000..1ceb6820b3 --- /dev/null +++ b/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 | undefined; + const encoder = new TextEncoder(); + + const stop = () => { + if (!intervalId) return; + clearInterval(intervalId); + intervalId = undefined; + }; + + return new TransformStream({ + 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(); + }, + }); +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 83eef18e71..ccebc722af 100644 --- a/open-sse/utils/stream.ts +++ b/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(); 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); diff --git a/tests/unit/responses-handler.test.mjs b/tests/unit/responses-handler.test.mjs index fafbb2086c..36f02b62d0 100644 --- a/tests/unit/responses-handler.test.mjs +++ b/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; + } +}); diff --git a/tests/unit/sse-heartbeat.test.mjs b/tests/unit/sse-heartbeat.test.mjs new file mode 100644 index 0000000000..4c7442615a --- /dev/null +++ b/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(); + }); +}); diff --git a/tests/unit/stream-utils.test.mjs b/tests/unit/stream-utils.test.mjs index 6c0ac87482..97cd1791f4 100644 --- a/tests/unit/stream-utils.test.mjs +++ b/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\]/); +});