fix(sse): robust Anthropic /v1/messages streaming — real ping keepalive + client-disconnect guard (#5063)

Integrated into release/v3.8.38 (leva 5)

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)
 
 ---
 

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 */
       }

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) {

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);
 }
 

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<string> {
   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<Response>((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).

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);
+});