From 98c5fac3ad4fbb0db5bf42c1fc18fe9eeab83175 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89der=20Costa?= Date: Fri, 26 Jun 2026 11:09:12 -0300 Subject: [PATCH] =?UTF-8?q?fix(sse):=20robust=20Anthropic=20/v1/messages?= =?UTF-8?q?=20streaming=20=E2=80=94=20real=20ping=20keepalive=20+=20client?= =?UTF-8?q?-disconnect=20guard=20(#5063)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.38 (leva 5) --- CHANGELOG.md | 1 + open-sse/utils/earlyStreamKeepalive.ts | 21 +++++++++- open-sse/utils/streamHandler.ts | 28 +++++++++++++ src/app/api/v1/messages/route.ts | 28 +++++++++++++ tests/unit/early-stream-keepalive.test.ts | 31 +++++++++++++- .../stream-handler-client-disconnect.test.ts | 40 +++++++++++++++++++ 6 files changed, 146 insertions(+), 3 deletions(-) create mode 100644 tests/unit/stream-handler-client-disconnect.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 808de5ef81..4f80a616e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ _In development — bullets added per PR; finalized at release._ - **fix(translator):** preserve client `cache_control` breakpoints when routing Claude-format requests (e.g. Claude Code) to Alibaba DashScope's OpenAI-compatible providers (`alibaba` / `alibaba-cn`). The Claude→OpenAI translation previously stripped the markers from the system and message text blocks, so DashScope's explicit caching never engaged and every request was a cache miss. Cache hints now survive when preservation is requested for caching-capable OpenAI-format providers. (thanks @sacrtap) - **fix(tts):** resolve Gemini TTS models from catalog and add `gemini-3.1-flash-tts-preview` as the new default Vertex TTS model. (thanks @nguyenha935) - **fix(sse): don't cool down a healthy connection on a self-inflicted upstream timeout (504)** — when OmniRoute's own deadline elapses (surfaced as `TimeoutError`/`BodyTimeoutError` → 504), the connection is no longer disabled/failed-over, so a slow-but-healthy provider isn't penalised for our timeout. Genuine upstream 5xx/429 still trigger cooldown; antigravity keeps its own policy. (thanks @costaeder) +- **fix(sse): robust Anthropic `/v1/messages` streaming — real ping keepalive + client-disconnect guard** — slow first tokens on reasoning models could trip strict clients' idle-read watchdog; the route now keeps the stream warm with a real `event: ping` (Anthropic clients ignore SSE comments) from the very first frame, and a client disconnect (AbortError / controller-closed) no longer counts as a provider failure (no failover/cooldown). (thanks @costaeder) --- diff --git a/open-sse/utils/earlyStreamKeepalive.ts b/open-sse/utils/earlyStreamKeepalive.ts index e83a3d645f..db2b98f194 100644 --- a/open-sse/utils/earlyStreamKeepalive.ts +++ b/open-sse/utils/earlyStreamKeepalive.ts @@ -28,6 +28,12 @@ const ENCODER = new TextEncoder(); const KEEPALIVE_FRAME = ENCODER.encode(": omniroute-keepalive\n\n"); +// Anthropic Messages-format keepalive: a REAL `ping` SSE event, not a comment. +// Anthropic clients (Claude Code, the Anthropic SDK) reset their stream/first-token +// watchdog on real SSE events but ignore SSE comments (`: ...`), so on a slow first +// token the comment frame lets the client abort and retry the stream. Anthropic's own +// API emits `event: ping` for exactly this reason; the /v1/messages route mirrors it. +export const ANTHROPIC_PING_FRAME = ENCODER.encode('event: ping\ndata: {"type":"ping"}\n\n'); const ERROR_FRAME = ENCODER.encode( `event: error\ndata: ${JSON.stringify({ error: { message: "Upstream stream failed before completion.", type: "stream_error" }, @@ -41,6 +47,13 @@ export type EarlyStreamKeepaliveOptions = { intervalMs?: number; /** Client request signal — propagated so a client disconnect cancels the upstream read. */ signal?: AbortSignal | null; + /** + * Frame emitted on each keepalive tick. Defaults to an SSE comment + * (`: omniroute-keepalive`). Anthropic-format routes (/v1/messages) must pass + * `ANTHROPIC_PING_FRAME` instead, because Anthropic clients ignore SSE comments + * for their stream watchdog and only a real `event: ping` keeps them from aborting. + */ + keepaliveFrame?: Uint8Array; }; type SettledHandler = { ok: true; response: Response } | { ok: false; error: unknown }; @@ -52,6 +65,7 @@ export async function withEarlyStreamKeepalive( const thresholdMs = Math.max(0, options.thresholdMs ?? 2_000); const intervalMs = Math.max(250, options.intervalMs ?? 2_500); const signal = options.signal ?? null; + const keepaliveFrame = options.keepaliveFrame ?? KEEPALIVE_FRAME; // Settle into a tagged result so neither race branch leaves an unhandled // rejection when the threshold timer wins. @@ -88,7 +102,7 @@ export async function withEarlyStreamKeepalive( const interval = setInterval(() => { if (stopped) return; try { - controller.enqueue(KEEPALIVE_FRAME); + controller.enqueue(keepaliveFrame); } catch { stopped = true; clearInterval(interval); @@ -98,8 +112,11 @@ export async function withEarlyStreamKeepalive( interval.unref?.(); } // First keepalive immediately on commit so the client sees a byte right away. + // Use the configured frame (e.g. ANTHROPIC_PING_FRAME) — an SSE comment here + // would be ignored by Anthropic clients' watchdog on a sub-interval gap, + // defeating the keepalive for exactly the case it targets. try { - controller.enqueue(KEEPALIVE_FRAME); + controller.enqueue(keepaliveFrame); } catch { /* consumer already gone */ } diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 1e813a978f..451cb7cffa 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -143,6 +143,24 @@ function isPendingRequestClearedError(error: unknown): boolean { ); } +/** + * A client disconnect — the caller aborted the request or closed the SSE + * connection — is NOT a provider failure. It surfaces either as an + * AbortError/ResponseAborted, or, when OmniRoute then tries to enqueue another + * chunk into the now-closed response stream, as a "Controller is already closed" + * TypeError. Treating any of these as an upstream error wrongly cools down the + * account/connection, so the stream error path uses this to skip the provider + * failover/cooldown (the chatgpt-web / codex / antigravity executors already + * guard client aborts the same way). + */ +export function isClientDisconnectError(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + const name = (error as { name?: unknown }).name; + if (name === "AbortError" || name === "ResponseAborted") return true; + const message = (error as { message?: unknown }).message; + return typeof message === "string" && /Controller is already closed/i.test(message); +} + function getErrorMessage(error: unknown): string { if (error instanceof Error && error.message) return error.message; if (typeof error === "string" && error.trim().length > 0) return error; @@ -253,6 +271,16 @@ export function createStreamController({ abortTimeout = null; } + // A client disconnect is not a provider failure. If the client already went away + // (disconnected) or the error is a client abort / "Controller is already closed", + // skip the onError failover/cooldown path — otherwise one cancelled request marks + // the upstream connection unavailable. + if (disconnected || isClientDisconnectError(error)) { + clearPendingRequest(error); + logStream(disconnected ? "client_disconnect (post-abort)" : "client_disconnect"); + return; + } + const alreadyCleared = isPendingRequestClearedError(error); let handled = false; if (!alreadyCleared) { diff --git a/src/app/api/v1/messages/route.ts b/src/app/api/v1/messages/route.ts index acac8c610b..9a67080aa8 100644 --- a/src/app/api/v1/messages/route.ts +++ b/src/app/api/v1/messages/route.ts @@ -1,6 +1,11 @@ import { handleChat } from "@/sse/handlers/chat"; import { initTranslators } from "@omniroute/open-sse/translator/index.ts"; import { withInjectionGuard } from "@/middleware/promptInjectionGuard"; +import { + withEarlyStreamKeepalive, + ANTHROPIC_PING_FRAME, +} from "@omniroute/open-sse/utils/earlyStreamKeepalive"; +import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold"; let initialized = false; @@ -35,6 +40,29 @@ export async function OPTIONS() { */ async function postHandler(request: any, context: any, preParsedBody: any = null) { await ensureInitialized(); + // Streaming Anthropic clients (Claude Code, the Anthropic SDK) drop the connection + // when no bytes arrive while a large prompt is processed before the first token — a + // big context can exceed the client's stream/first-token watchdog. OmniRoute holds + // the response until the first useful upstream byte (ensureStreamReadiness), so keep + // the connection warm with early keepalives during that gap — same wrapper used by + // /v1/responses (#2544). Anthropic clients ignore SSE comments for their watchdog, so + // emit a real `event: ping` (ANTHROPIC_PING_FRAME). Non-streaming callers keep the + // verbatim path. + const accept = String(request.headers?.get?.("accept") || "").toLowerCase(); + if (accept.includes("text/event-stream")) { + let model; + try { + const body = preParsedBody ?? (await request.clone().json().catch(() => null)); + model = body?.model; + } catch { + // body unavailable / non-JSON — fall back to the default keepalive threshold + } + return await withEarlyStreamKeepalive(handleChat(request, null, preParsedBody), { + signal: request.signal, + thresholdMs: resolveKeepaliveThreshold(model), + keepaliveFrame: ANTHROPIC_PING_FRAME, + }); + } return await handleChat(request, null, preParsedBody); } diff --git a/tests/unit/early-stream-keepalive.test.ts b/tests/unit/early-stream-keepalive.test.ts index ad95b88069..a75c4c2764 100644 --- a/tests/unit/early-stream-keepalive.test.ts +++ b/tests/unit/early-stream-keepalive.test.ts @@ -1,7 +1,10 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { withEarlyStreamKeepalive } from "../../open-sse/utils/earlyStreamKeepalive.ts"; +import { + withEarlyStreamKeepalive, + ANTHROPIC_PING_FRAME, +} from "../../open-sse/utils/earlyStreamKeepalive.ts"; async function readAll(response: Response): Promise { const reader = response.body!.getReader(); @@ -56,6 +59,32 @@ test("slow handler emits early keepalive then forwards the real body (#2544)", a assert.match(body, /data: \[DONE\]/); }); +// Anthropic clients (Claude Code, the Anthropic SDK) ignore SSE comments for their +// stream/first-token watchdog and abort+retry on a slow first token. The /v1/messages +// route keeps the connection warm with a REAL `event: ping` instead of the comment frame. +test("ANTHROPIC_PING_FRAME is a real Anthropic ping event (not a comment)", () => { + const decoded = new TextDecoder().decode(ANTHROPIC_PING_FRAME); + assert.equal(decoded, 'event: ping\ndata: {"type":"ping"}\n\n'); + assert.doesNotMatch(decoded, /^:/, "must not be an SSE comment"); +}); + +test("slow handler emits the custom keepaliveFrame (Anthropic ping) before the body", async () => { + const slow = new Promise((resolve) => { + setTimeout(() => resolve(sseResponse("event: message_start\ndata: {}\n\ndata: [DONE]\n\n")), 120); + }); + + const result = await withEarlyStreamKeepalive(slow, { + thresholdMs: 25, + intervalMs: 20, + keepaliveFrame: ANTHROPIC_PING_FRAME, + }); + + const body = await readAll(result); + assert.match(body, /event: ping\ndata: {"type":"ping"}/, "should emit a real ping event"); + assert.doesNotMatch(body, /: omniroute-keepalive/, "must not fall back to the comment frame"); + assert.match(body, /event: message_start/, "should forward the real upstream body"); +}); + // #2544: a non-SSE error that arrives after we already committed to a 200 event-stream // must be framed as an in-band `event: error` (the HTTP status can no longer change), // not forwarded as raw JSON (which would be malformed SSE). diff --git a/tests/unit/stream-handler-client-disconnect.test.ts b/tests/unit/stream-handler-client-disconnect.test.ts new file mode 100644 index 0000000000..9f56f2937b --- /dev/null +++ b/tests/unit/stream-handler-client-disconnect.test.ts @@ -0,0 +1,40 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { isClientDisconnectError } = await import("../../open-sse/utils/streamHandler.ts"); + +test("client disconnect: AbortError is a client disconnect, not a provider failure", () => { + const err = new Error("aborted"); + err.name = "AbortError"; + assert.equal(isClientDisconnectError(err), true); +}); + +test("client disconnect: ResponseAborted is a client disconnect", () => { + const err = new Error("The response was aborted"); + err.name = "ResponseAborted"; + assert.equal(isClientDisconnectError(err), true); +}); + +test("client disconnect: 'Controller is already closed' TypeError is a client disconnect", () => { + // Thrown when OmniRoute enqueues into a response stream the client already closed. + // name is "TypeError", so this must be matched by message, not name. + const err = new TypeError("Invalid state: Controller is already closed"); + assert.equal(isClientDisconnectError(err), true); +}); + +test("client disconnect: a real provider 502/500 is NOT a client disconnect", () => { + const err = Object.assign(new Error("Upstream stream failed"), { statusCode: 502 }); + assert.equal(isClientDisconnectError(err), false); +}); + +test("client disconnect: a rate-limit (429) error is NOT a client disconnect", () => { + const err = Object.assign(new Error("rate_limit_error"), { statusCode: 429 }); + assert.equal(isClientDisconnectError(err), false); +}); + +test("client disconnect: null / string / plain object are NOT client disconnects", () => { + assert.equal(isClientDisconnectError(null), false); + assert.equal(isClientDisconnectError(undefined), false); + assert.equal(isClientDisconnectError("Controller is already closed"), false); + assert.equal(isClientDisconnectError({}), false); +});