From 50a64ea6e97f0bbd88bec3fb81f7f09ae4c6dc8e Mon Sep 17 00:00:00 2001 From: ratchaw Date: Tue, 12 May 2026 10:20:39 +0100 Subject: [PATCH] fix(anthropic): emit proper SSE error events on streaming requests When /v1/messages received an upstream provider error (for example an Anthropic 400 for a tool_use / tool_result mismatch) on a streaming request, the gateway ended the stream with only `event: message_stop`, which breaks the Anthropic streaming contract and hides the real error. The streaming handler now tracks SSE event types from the internal chat completions stream, detects error chunks (by event type or payload shape), and emits Anthropic-style `event: error` payloads followed by `event: message_stop`. The `!response.ok` path now returns SSE when the client asked for streaming instead of a JSON error body. --- apps/gateway/src/anthropic/anthropic.ts | 69 ++++++++- .../streaming-error-translation.spec.ts | 141 ++++++++++++++++++ .../anthropic/streaming-error-translation.ts | 85 +++++++++++ 3 files changed, 294 insertions(+), 1 deletion(-) create mode 100644 apps/gateway/src/anthropic/streaming-error-translation.spec.ts create mode 100644 apps/gateway/src/anthropic/streaming-error-translation.ts diff --git a/apps/gateway/src/anthropic/anthropic.ts b/apps/gateway/src/anthropic/anthropic.ts index 83fa47d1ef..a5d10eff19 100644 --- a/apps/gateway/src/anthropic/anthropic.ts +++ b/apps/gateway/src/anthropic/anthropic.ts @@ -6,6 +6,8 @@ import { app } from "@/app.js"; import { logger, toError } from "@llmgateway/logger"; +import { buildAnthropicErrorEvent } from "./streaming-error-translation.js"; + import type { ServerTypes } from "@/vars.js"; export const anthropic = new OpenAPIHono(); @@ -521,6 +523,34 @@ anthropic.openapi(messages, async (c) => { statusText: response.statusText, }); const errorData = await response.text(); + + if (anthropicRequest.stream) { + let parsedError: unknown = null; + try { + parsedError = JSON.parse(errorData); + } catch { + parsedError = null; + } + const errorEvent = buildAnthropicErrorEvent( + parsedError ?? { + error: { + message: errorData || response.statusText, + type: "api_error", + }, + }, + ); + return streamSSE(c, async (stream) => { + await stream.writeSSE({ + data: JSON.stringify(errorEvent), + event: "error", + }); + await stream.writeSSE({ + data: JSON.stringify({ type: "message_stop" }), + event: "message_stop", + }); + }); + } + return c.json( { error: true, @@ -570,6 +600,7 @@ anthropic.openapi(messages, async (c) => { }; let currentTextBlockIndex: number | null = null; const toolCallBlockIndex = new Map(); + let currentEventType: string | null = null; try { while (true) { @@ -582,7 +613,21 @@ anthropic.openapi(messages, async (c) => { const lines = buffer.split("\n"); buffer = lines.pop() ?? ""; - for (const line of lines) { + for (const rawLine of lines) { + const line = rawLine.endsWith("\r") + ? rawLine.slice(0, -1) + : rawLine; + + if (line === "") { + currentEventType = null; + continue; + } + + if (line.startsWith("event: ")) { + currentEventType = line.slice(7).trim(); + continue; + } + if (line.startsWith("data: ")) { const data = line.slice(6).trim(); if (data === "[DONE]") { @@ -609,6 +654,28 @@ anthropic.openapi(messages, async (c) => { continue; } + const looksLikeError = + currentEventType === "error" || + (chunk && + typeof chunk === "object" && + (chunk.type === "error" || + (chunk.error && + typeof chunk.error === "object" && + !chunk.choices && + !chunk.id))); + + if (looksLikeError) { + await stream.writeSSE({ + data: JSON.stringify(buildAnthropicErrorEvent(chunk)), + event: "error", + }); + await stream.writeSSE({ + data: JSON.stringify({ type: "message_stop" }), + event: "message_stop", + }); + return; + } + if (!messageId && chunk.id) { messageId = chunk.id; model = chunk.model ?? anthropicRequest.model; diff --git a/apps/gateway/src/anthropic/streaming-error-translation.spec.ts b/apps/gateway/src/anthropic/streaming-error-translation.spec.ts new file mode 100644 index 0000000000..449d12965a --- /dev/null +++ b/apps/gateway/src/anthropic/streaming-error-translation.spec.ts @@ -0,0 +1,141 @@ +import { describe, expect, it } from "vitest"; + +import { + buildAnthropicErrorEvent, + mapInternalErrorTypeToAnthropic, +} from "./streaming-error-translation.js"; + +describe("mapInternalErrorTypeToAnthropic", () => { + it("maps internal-only finish reasons to Anthropic types", () => { + expect(mapInternalErrorTypeToAnthropic("client_error")).toBe( + "invalid_request_error", + ); + expect(mapInternalErrorTypeToAnthropic("gateway_error")).toBe( + "authentication_error", + ); + expect(mapInternalErrorTypeToAnthropic("upstream_error")).toBe("api_error"); + }); + + it("preserves canonical Anthropic types (including ones not exposed internally)", () => { + for (const type of [ + "invalid_request_error", + "authentication_error", + "billing_error", + "permission_error", + "not_found_error", + "request_too_large", + "rate_limit_error", + "api_error", + "timeout_error", + "overloaded_error", + ]) { + expect(mapInternalErrorTypeToAnthropic(type)).toBe(type); + } + }); + + it("falls back to api_error for unknown / missing types", () => { + expect(mapInternalErrorTypeToAnthropic(undefined)).toBe("api_error"); + expect(mapInternalErrorTypeToAnthropic(null)).toBe("api_error"); + expect(mapInternalErrorTypeToAnthropic("totally_made_up")).toBe( + "api_error", + ); + expect(mapInternalErrorTypeToAnthropic(42)).toBe("api_error"); + }); +}); + +describe("buildAnthropicErrorEvent", () => { + it("translates the user's exact repro: passthrough Anthropic shape with invalid_request_error", () => { + const chunk = { + type: "error", + error: { + type: "invalid_request_error", + message: + "messages.2: `tool_use` ids were found without `tool_result` blocks immediately after: toolu_01TEST123", + }, + }; + expect(buildAnthropicErrorEvent(chunk)).toEqual({ + type: "error", + error: { + type: "invalid_request_error", + message: + "messages.2: `tool_use` ids were found without `tool_result` blocks immediately after: toolu_01TEST123", + }, + }); + }); + + it("preserves unknown-but-valid Anthropic types on passthrough (future-proofing)", () => { + const chunk = { + type: "error", + error: { type: "billing_error", message: "Add a payment method" }, + }; + const out = buildAnthropicErrorEvent(chunk); + expect(out.error.type).toBe("billing_error"); + expect(out.error.message).toBe("Add a payment method"); + }); + + it("translates the wrapped internal client_error shape from chat.ts", () => { + const chunk = { + error: { + message: "Error from provider anthropic: 400 Bad Request ...", + type: "client_error", + param: null, + code: "client_error", + responseText: "...", + }, + }; + expect(buildAnthropicErrorEvent(chunk)).toEqual({ + type: "error", + error: { + type: "invalid_request_error", + message: "Error from provider anthropic: 400 Bad Request ...", + }, + }); + }); + + it("translates the wrapped internal upstream_error shape", () => { + const chunk = { + error: { + message: "Provider returned 503", + type: "upstream_error", + code: "upstream_error", + }, + }; + expect(buildAnthropicErrorEvent(chunk).error.type).toBe("api_error"); + }); + + it("translates the wrapped internal gateway_error shape", () => { + const chunk = { + error: { message: "Bad API key", type: "gateway_error" }, + }; + expect(buildAnthropicErrorEvent(chunk).error.type).toBe( + "authentication_error", + ); + }); + + it("falls back to api_error + JSON-stringified body for unparseable shapes", () => { + const chunk = { random: "thing" }; + const out = buildAnthropicErrorEvent(chunk); + expect(out.type).toBe("error"); + expect(out.error.type).toBe("api_error"); + expect(out.error.message).toBe(JSON.stringify(chunk)); + }); + + it("uses string chunks verbatim as the message", () => { + expect(buildAnthropicErrorEvent("plain error text")).toEqual({ + type: "error", + error: { type: "api_error", message: "plain error text" }, + }); + }); + + it("falls back to JSON-stringified inner error when inner.message is non-string", () => { + const chunk = { + error: { + type: "client_error", + message: { nested: "object" }, + }, + }; + const out = buildAnthropicErrorEvent(chunk); + expect(out.error.type).toBe("invalid_request_error"); + expect(out.error.message).toBe(JSON.stringify(chunk.error)); + }); +}); diff --git a/apps/gateway/src/anthropic/streaming-error-translation.ts b/apps/gateway/src/anthropic/streaming-error-translation.ts new file mode 100644 index 0000000000..a4a44b805d --- /dev/null +++ b/apps/gateway/src/anthropic/streaming-error-translation.ts @@ -0,0 +1,85 @@ +// Canonical Anthropic `error.type` values per +// https://platform.claude.com/docs/en/api/errors. The docs explicitly note this +// enum will grow over time, so unknown strings on the passthrough path are +// preserved verbatim rather than downgraded to api_error. +const ANTHROPIC_ERROR_TYPES = new Set([ + "invalid_request_error", + "authentication_error", + "billing_error", + "permission_error", + "not_found_error", + "request_too_large", + "rate_limit_error", + "api_error", + "timeout_error", + "overloaded_error", +]); + +export function mapInternalErrorTypeToAnthropic( + internalType?: unknown, +): string { + if (typeof internalType !== "string") { + return "api_error"; + } + switch (internalType) { + case "client_error": + return "invalid_request_error"; + case "gateway_error": + return "authentication_error"; + case "upstream_error": + return "api_error"; + } + if (ANTHROPIC_ERROR_TYPES.has(internalType)) { + return internalType; + } + return "api_error"; +} + +export function buildAnthropicErrorEvent(chunk: unknown): { + type: "error"; + error: { type: string; message: string }; +} { + if (chunk && typeof chunk === "object") { + const obj = chunk as Record; + // Passthrough: upstream already produced a valid Anthropic error shape. + // Preserve `inner.type` as-is (the Anthropic enum is documented to grow), + // only falling back to api_error when the field is missing/non-string. + if (obj.type === "error" && obj.error && typeof obj.error === "object") { + const inner = obj.error as Record; + return { + type: "error", + error: { + type: + typeof inner.type === "string" && inner.type + ? inner.type + : "api_error", + message: + typeof inner.message === "string" + ? inner.message + : JSON.stringify(inner), + }, + }; + } + // Wrapped internal shape from /v1/chat/completions error path. + if (obj.error && typeof obj.error === "object") { + const inner = obj.error as Record; + return { + type: "error", + error: { + type: mapInternalErrorTypeToAnthropic(inner.type ?? inner.code), + message: + typeof inner.message === "string" + ? inner.message + : JSON.stringify(inner), + }, + }; + } + } + return { + type: "error", + error: { + type: "api_error", + message: typeof chunk === "string" ? chunk : JSON.stringify(chunk), + }, + }; +}