fix(ai): own Anthropic SSE parsing to avoid SDK JSON.parse hard-failures
Bypass the Anthropic SDK streaming parser entirely. Use client.messages.create().asResponse() and decode the SSE stream ourselves with defensive JSON parsing that repairs invalid escape sequences and control characters inside string literals. - Switch from SDK .stream() to .asResponse() + pi-owned SSE decoder - Add repairJson() / parseJsonWithRepair() to json-parse.ts - Add anthropic-sse-parsing.test.ts regression for malformed tool deltas - Update github-copilot-anthropic.test.ts mock to match new call path - Update deprecated claude-3-5-haiku-20241022 refs to claude-haiku-4-5 - Remove stale non-reasoning model test fixes #3175
This commit is contained in:
Mario Zechner
@@ -4,6 +4,7 @@ import type {
|
||||
ContentBlockParam,
|
||||
MessageCreateParamsStreaming,
|
||||
MessageParam,
|
||||
RawMessageStreamEvent,
|
||||
} from "@anthropic-ai/sdk/resources/messages.js";
|
||||
import { getEnvApiKey } from "../env-api-keys.js";
|
||||
import { calculateCost } from "../models.js";
|
||||
@@ -27,7 +28,7 @@ import type {
|
||||
} from "../types.js";
|
||||
import { AssistantMessageEventStream } from "../utils/event-stream.js";
|
||||
import { headersToRecord } from "../utils/headers.js";
|
||||
import { parseStreamingJson } from "../utils/json-parse.js";
|
||||
import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse.js";
|
||||
import { sanitizeSurrogates } from "../utils/sanitize-unicode.js";
|
||||
|
||||
import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.js";
|
||||
@@ -213,6 +214,176 @@ function mergeHeaders(...headerSources: (Record<string, string> | undefined)[]):
|
||||
return merged;
|
||||
}
|
||||
|
||||
interface ServerSentEvent {
|
||||
event: string | null;
|
||||
data: string;
|
||||
raw: string[];
|
||||
}
|
||||
|
||||
interface SseDecoderState {
|
||||
event: string | null;
|
||||
data: string[];
|
||||
raw: string[];
|
||||
}
|
||||
|
||||
function flushSseEvent(state: SseDecoderState): ServerSentEvent | null {
|
||||
if (!state.event && state.data.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const event: ServerSentEvent = {
|
||||
event: state.event,
|
||||
data: state.data.join("\n"),
|
||||
raw: [...state.raw],
|
||||
};
|
||||
state.event = null;
|
||||
state.data = [];
|
||||
state.raw = [];
|
||||
return event;
|
||||
}
|
||||
|
||||
function decodeSseLine(line: string, state: SseDecoderState): ServerSentEvent | null {
|
||||
if (line === "") {
|
||||
return flushSseEvent(state);
|
||||
}
|
||||
|
||||
state.raw.push(line);
|
||||
if (line.startsWith(":")) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const delimiterIndex = line.indexOf(":");
|
||||
const fieldName = delimiterIndex === -1 ? line : line.slice(0, delimiterIndex);
|
||||
let value = delimiterIndex === -1 ? "" : line.slice(delimiterIndex + 1);
|
||||
if (value.startsWith(" ")) {
|
||||
value = value.slice(1);
|
||||
}
|
||||
|
||||
if (fieldName === "event") {
|
||||
state.event = value;
|
||||
} else if (fieldName === "data") {
|
||||
state.data.push(value);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function nextLineBreakIndex(text: string): number {
|
||||
const carriageReturnIndex = text.indexOf("\r");
|
||||
const newlineIndex = text.indexOf("\n");
|
||||
if (carriageReturnIndex === -1) {
|
||||
return newlineIndex;
|
||||
}
|
||||
if (newlineIndex === -1) {
|
||||
return carriageReturnIndex;
|
||||
}
|
||||
return Math.min(carriageReturnIndex, newlineIndex);
|
||||
}
|
||||
|
||||
function consumeLine(text: string): { line: string; rest: string } | null {
|
||||
const lineBreakIndex = nextLineBreakIndex(text);
|
||||
if (lineBreakIndex === -1) {
|
||||
return null;
|
||||
}
|
||||
|
||||
let nextIndex = lineBreakIndex + 1;
|
||||
if (text[lineBreakIndex] === "\r" && text[nextIndex] === "\n") {
|
||||
nextIndex += 1;
|
||||
}
|
||||
|
||||
return {
|
||||
line: text.slice(0, lineBreakIndex),
|
||||
rest: text.slice(nextIndex),
|
||||
};
|
||||
}
|
||||
|
||||
async function* iterateSseMessages(
|
||||
body: ReadableStream<Uint8Array>,
|
||||
signal?: AbortSignal,
|
||||
): AsyncGenerator<ServerSentEvent> {
|
||||
const reader = body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
const state: SseDecoderState = { event: null, data: [], raw: [] };
|
||||
let buffer = "";
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) {
|
||||
throw new Error("Request was aborted");
|
||||
}
|
||||
|
||||
const { value, done } = await reader.read();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
let consumed = consumeLine(buffer);
|
||||
while (consumed) {
|
||||
buffer = consumed.rest;
|
||||
const event = decodeSseLine(consumed.line, state);
|
||||
if (event) {
|
||||
yield event;
|
||||
}
|
||||
consumed = consumeLine(buffer);
|
||||
}
|
||||
}
|
||||
|
||||
buffer += decoder.decode();
|
||||
let consumed = consumeLine(buffer);
|
||||
while (consumed) {
|
||||
buffer = consumed.rest;
|
||||
const event = decodeSseLine(consumed.line, state);
|
||||
if (event) {
|
||||
yield event;
|
||||
}
|
||||
consumed = consumeLine(buffer);
|
||||
}
|
||||
|
||||
if (buffer.length > 0) {
|
||||
const event = decodeSseLine(buffer, state);
|
||||
if (event) {
|
||||
yield event;
|
||||
}
|
||||
}
|
||||
|
||||
const trailingEvent = flushSseEvent(state);
|
||||
if (trailingEvent) {
|
||||
yield trailingEvent;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
async function* iterateAnthropicEvents(
|
||||
response: Response,
|
||||
signal?: AbortSignal,
|
||||
): AsyncGenerator<RawMessageStreamEvent> {
|
||||
if (!response.body) {
|
||||
throw new Error("Attempted to iterate over an Anthropic response with no body");
|
||||
}
|
||||
|
||||
for await (const sse of iterateSseMessages(response.body, signal)) {
|
||||
if (!sse.event || sse.event === "ping") {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (sse.event === "error") {
|
||||
throw new Error(sse.data);
|
||||
}
|
||||
|
||||
try {
|
||||
yield parseJsonWithRepair<RawMessageStreamEvent>(sse.data);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
throw new Error(
|
||||
`Could not parse Anthropic SSE event ${sse.event}: ${message}; data=${sse.data}; raw=${sse.raw.join("\\n")}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
||||
model: Model<"anthropic-messages">,
|
||||
context: Context,
|
||||
@@ -273,16 +444,16 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicOpti
|
||||
if (nextParams !== undefined) {
|
||||
params = nextParams as MessageCreateParamsStreaming;
|
||||
}
|
||||
const { data: anthropicStream, response } = await client.messages
|
||||
.stream({ ...params, stream: true }, { signal: options?.signal })
|
||||
.withResponse();
|
||||
const response = await client.messages
|
||||
.create({ ...params, stream: true }, { signal: options?.signal })
|
||||
.asResponse();
|
||||
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
|
||||
stream.push({ type: "start", partial: output });
|
||||
|
||||
type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson: string })) & { index: number };
|
||||
const blocks = output.content as Block[];
|
||||
|
||||
for await (const event of anthropicStream) {
|
||||
for await (const event of iterateAnthropicEvents(response, options?.signal)) {
|
||||
if (event.type === "message_start") {
|
||||
output.responseId = event.message.id;
|
||||
// Capture initial token usage from message_start event
|
||||
|
||||
Reference in New Issue
Block a user