diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 7b7634e2..6751256b 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -3,8 +3,9 @@ import { needsTranslation } from "../../translator/index.js"; import { createSSETransformStreamWithLogger, createPassthroughStreamWithLogger } from "../../utils/stream.js"; import { pipeWithDisconnect } from "../../utils/streamHandler.js"; import { PROVIDERS } from "../../config/providers.js"; -import { STREAM_STALL_TIMEOUT_MS } from "../../config/runtimeConfig.js"; +import { HTTP_STATUS, STREAM_STALL_TIMEOUT_MS } from "../../config/runtimeConfig.js"; import { buildAbortedResponsesTerminalBytes } from "../../utils/responsesStreamHelpers.js"; +import { buildStreamErrorBytes } from "../../utils/streamHelpers.js"; import { buildRequestDetail, extractRequestConfig, saveUsageStats, formatDoneLine } from "./requestDetail.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; import { SSE_HEADERS_CORS as SSE_HEADERS } from "../../utils/sseConstants.js"; @@ -81,9 +82,14 @@ export async function handleStreamingResponse({ providerResponse, provider, mode const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, customToolNames, model, connectionId, body, onStreamComplete, apiKey, credentials }); - // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event + // Terminal bytes when the stream aborts after HTTP 200 was already sent, so the + // client sees a real error instead of a silently truncated stream. + // Responses passthrough keeps its own response.failed shape; every other client + // format gets the OpenAI error frame + [DONE], or `event: error` for Claude. const isResponsesPassthrough = sourceFormat === FORMATS.OPENAI_RESPONSES && targetFormat === FORMATS.OPENAI_RESPONSES; - const onAbortTerminal = isResponsesPassthrough ? buildAbortedResponsesTerminalBytes : null; + const onAbortTerminal = isResponsesPassthrough + ? buildAbortedResponsesTerminalBytes + : (message) => buildStreamErrorBytes(HTTP_STATUS.GATEWAY_TIMEOUT, message, sourceFormat); const stallTimeoutMs = PROVIDERS[provider]?.stallTimeoutMs || STREAM_STALL_TIMEOUT_MS; const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal, stallTimeoutMs); diff --git a/open-sse/utils/streamHandler.js b/open-sse/utils/streamHandler.js index 27a623fc..fce01706 100644 --- a/open-sse/utils/streamHandler.js +++ b/open-sse/utils/streamHandler.js @@ -95,6 +95,9 @@ export function createStreamController({ onDisconnect, onError, log, provider, m * activity), not here — output of the transform stream may be silent * for long periods while raw bytes still flow (e.g. Kiro EventStream * binary frames buffering, Claude reasoning streams). + * + * @param {function} [onAbortTerminal] - Receives a human-readable abort + * message and returns terminal SSE bytes to emit downstream. */ export function createDisconnectAwareStream(transformStream, streamController, onAbortTerminal = null) { const reader = transformStream.readable.getReader(); @@ -194,6 +197,7 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont let chunkCount = 0; let totalBytes = 0; let lastChunkAt = Date.now(); + let abortMessage = "upstream connection lost"; const t0 = Date.now(); const tag = "STREAM"; const clearStall = () => { @@ -203,6 +207,7 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont clearStall(); stallTimer = setTimeout(() => { stallTimer = null; + abortMessage = "stream stall timeout"; dbg(tag, `STALL TIMEOUT ${stallTimeoutMs}ms | chunks=${chunkCount} | bytes=${totalBytes} | sinceLast=${Date.now() - lastChunkAt}ms`); streamController.handleError?.(new Error("stream stall timeout")); streamController.abort?.(); @@ -249,7 +254,7 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont return createDisconnectAwareStream( { readable: transformedBody, writable: { getWriter: () => ({ abort: () => Promise.resolve() }) } }, wrappedController, - onAbortTerminal + onAbortTerminal ? () => onAbortTerminal(abortMessage) : null ); } diff --git a/open-sse/utils/streamHelpers.js b/open-sse/utils/streamHelpers.js index a7a19180..ee4e720a 100644 --- a/open-sse/utils/streamHelpers.js +++ b/open-sse/utils/streamHelpers.js @@ -1,4 +1,8 @@ import { FORMATS } from "../translator/formats.js"; +import { buildErrorBody } from "./error.js"; +import { SSE_DONE } from "./sseConstants.js"; + +const sharedEncoder = new TextEncoder(); // Parse SSE data line export function parseSSELine(line, format = null) { @@ -120,3 +124,24 @@ export function formatSSE(data, sourceFormat) { return `data: ${JSON.stringify(data)}\n\n`; } + +// Terminal frames for a stream that aborted after HTTP 200 was already sent, so +// the status code can no longer change. OpenAI-compatible clients (openai-python +// raises APIError on any `data:` payload carrying an `error` key, checked before +// [DONE]) need the error frame first, then [DONE]; Anthropic clients need +// `event: error`. Never fabricate a successful finish_reason instead. +// +// Returns encoded bytes: onAbortTerminal callbacks are enqueued verbatim, same +// as buildAbortedResponsesTerminalBytes. +// +// NOTE: non-SSE client formats (Ollama NDJSON) get an SSE frame here — dead in +// practice because detectFormatByEndpoint never resolves to OLLAMA. +export function buildStreamErrorBytes(statusCode, message, clientFormat) { + const { error } = buildErrorBody(statusCode, message); + + const sse = clientFormat === FORMATS.CLAUDE + ? formatSSE({ type: "error", error }, FORMATS.CLAUDE) + : formatSSE({ error }, clientFormat) + SSE_DONE; + + return sharedEncoder.encode(sse); +} diff --git a/tests/unit/responses-abort-terminal.test.js b/tests/unit/responses-abort-terminal.test.js index 10af5848..e05c9d14 100644 --- a/tests/unit/responses-abort-terminal.test.js +++ b/tests/unit/responses-abort-terminal.test.js @@ -1,7 +1,9 @@ import { describe, expect, it } from "vitest"; -import { createDisconnectAwareStream } from "../../open-sse/utils/streamHandler.js"; +import { createDisconnectAwareStream, pipeWithDisconnect, createStreamController } from "../../open-sse/utils/streamHandler.js"; import { buildAbortedResponsesTerminalBytes } from "../../open-sse/utils/responsesStreamHelpers.js"; +import { buildStreamErrorBytes } from "../../open-sse/utils/streamHelpers.js"; +import { FORMATS } from "../../open-sse/translator/formats.js"; // Minimal stream controller stub function makeController() { @@ -70,3 +72,71 @@ describe("Responses abort terminal synthesis", () => { expect(text).not.toContain("[DONE]"); }); }); + +// A stream that aborts after HTTP 200 cannot change status, so the failure must +// travel in-band: structured error frame first, then [DONE]. openai-python raises +// APIError on any `data:` payload carrying an `error` key (checked before [DONE]); +// Anthropic clients need `event: error`. Never a fabricated finish_reason. +describe("buildStreamErrorBytes", () => { + const jsonOf = (sse) => JSON.parse(sse.match(/\{.*\}/s)[0]); + const textOf = (bytes) => new TextDecoder().decode(bytes); + + // onAbortTerminal callbacks are enqueued verbatim, so a string here is a + // silent no-op at runtime (createDisconnectAwareStream swallows the throw). + it("returns bytes, not a string", () => { + expect(buildStreamErrorBytes(504, "x", FORMATS.OPENAI)).toBeInstanceOf(Uint8Array); + }); + + it("emits error frame then [DONE] for OpenAI clients", () => { + const out = textOf(buildStreamErrorBytes(504, "stream stall timeout", FORMATS.OPENAI)); + + expect(out).toContain('data: {"error"'); + expect(out.indexOf("data: [DONE]")).toBeGreaterThan(out.indexOf('data: {"error"')); + + expect(jsonOf(out).error).toEqual({ + message: "stream stall timeout", + type: "server_error", + code: "gateway_timeout", + }); + }); + + it("emits event: error (no [DONE]) for Claude clients", () => { + const out = textOf(buildStreamErrorBytes(504, "stream stall timeout", FORMATS.CLAUDE)); + + expect(out).toContain("event: error\n"); + expect(out).not.toContain("[DONE]"); + expect(jsonOf(out)).toMatchObject({ type: "error", error: { message: "stream stall timeout" } }); + }); +}); + +// The wiring, not just the frame builder: the watchdog must hand its reason to +// onAbortTerminal and the bytes must reach a real consumer. +describe("stall abort through pipeWithDisconnect", () => { + it("delivers the error frame and closes the stream", async () => { + // Real controller: the stub above never fires its signal, and the abort + // must reach the upstream body for the pipe to end. + const ctrl = createStreamController({ provider: "ollama", model: "test" }); + + // Emits one chunk then goes silent; errors on abort like a real fetch body. + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: hi\n\n")); + ctrl.signal.addEventListener("abort", () => controller.error(new Error("aborted")), { once: true }); + }, + }); + + let seen = null; + const out = pipeWithDisconnect( + { body: upstream }, + new TransformStream(), + ctrl, + (message) => { seen = message; return buildStreamErrorBytes(504, message, FORMATS.OPENAI); }, + 50 + ); + + const text = await readAll(out); + expect(seen).toBe("stream stall timeout"); + expect(text).toContain('"stream stall timeout"'); + expect(text).toContain("data: [DONE]"); + }); +});