fix(stream): report aborts after HTTP 200 in-band instead of closing silently
A stream that stalled or lost its upstream was closed with no terminal frame
at all, so clients saw "200 OK, a few chunks, then nothing" and could not tell
a truncated reply from a finished one. The Responses passthrough path already
synthesized response.failed; every other client format got nothing.
The watchdog now hands its reason ("stream stall timeout" or "upstream
connection lost") to onAbortTerminal, and buildStreamErrorBytes frames it per
client format: OpenAI-compatible clients get data: {"error":{...}} followed by
data: [DONE], Anthropic clients get `event: error`. The error frame always
precedes [DONE] (openai-python raises APIError on any data payload carrying an
error key), and no synthetic finish_reason is ever emitted — a truncated
stream must not look like a clean stop.
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -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]");
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user