File size: 4,833 Bytes
67d18ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | /**
* OpenAI SSE stream adapter for the async path.
*
* Single persistent state machine for the whole stream — preserves message ID,
* usage, tool-call index mapping, and finish state across events. Re-using the
* same translation state is what the existing `anthropicSseToOpenaiSse` does;
* we add two extra concerns for the async bridge:
*
* 1. Preserve `: keepalive` comments (the existing translator drops them).
* Comments are the only queue-wait liveness signal; dropping them defeats
* the bridge design.
* 2. Convert Anthropic `event: error` events to OpenAI
* `data: {"error":...}` + terminal `[DONE]`. The existing translator
* silently drops Anthropic errors, which would mask permanent bridge
* failures as a clean `[DONE]`.
*
* Anti-pattern: NEVER create a fresh translator per event. State (messageId,
* toolCallIndex, finishReasonSent, usage) must persist across the whole stream
* or downstream clients see inconsistent IDs, missing tool arguments, and
* duplicate finish reasons.
*/
import { initState, parseSSEChunk, translateEvent, SSE_FRAME_SPLIT, type TranslationState, type ParsedSSE } from "../translator/sse-translator.js";
export function anthropicSseToOpenaiSseWithKeepalive(
upstream: ReadableStream<Uint8Array>,
model: string = "glm-4.6",
): ReadableStream<Uint8Array> {
const encoder = new TextEncoder();
const decoder = new TextDecoder();
const state: TranslationState = initState(model);
let doneSent = false;
let errored = false;
function emit(out: string): void {
if (errored) return;
try {
controller0.enqueue(encoder.encode(out));
} catch {
// controller closed by consumer
}
}
// Hoisted controller reference so `emit()` can be defined above the ReadableStream
// constructor without forward-let noise.
let controller0: ReadableStreamDefaultController<Uint8Array>;
return new ReadableStream<Uint8Array>({
start(controller) {
controller0 = controller;
const reader = upstream.getReader();
let buffer = "";
reader.read().then(function pump({ done, value }): Promise<unknown> | undefined {
if (done) {
// Flush trailing buffer
if (buffer.trim()) processBlock(buffer, state);
buffer = "";
emitDone();
try { controller.close(); } catch {}
return;
}
buffer += decoder.decode(value, { stream: true });
const blocks = buffer.split(SSE_FRAME_SPLIT);
buffer = blocks.pop() ?? "";
for (const block of blocks) {
processBlock(block, state);
if (errored) {
try { controller.close(); } catch {}
return;
}
}
return reader.read().then(pump);
}).catch((err) => {
if (!errored) {
const errPayload = JSON.stringify({ error: { message: `async stream error: ${(err as Error).message}`, type: "server_error" } });
emit(`data: ${errPayload}\n\n`);
emitDone();
}
try { controller.close(); } catch {}
}).finally(() => {
reader.releaseLock?.();
});
},
});
function processBlock(block: string, st: TranslationState): void {
const trimmed = block.trim();
if (trimmed === "") return;
// Pure comment frame — pass through unchanged
if (trimmed.startsWith(":")) {
emit(block + "\n\n");
return;
}
// Detect Anthropic error event before parseSSEChunk would silently drop it
// (parseSSEChunk only returns events with data; error events have data too,
// but the existing translateEvent doesn't handle `type:"error"`).
if (trimmed.startsWith("event: error") || trimmed.startsWith("event:error")) {
const dataLine = block.split("\n").find((l) => l.startsWith("data:"));
let anthropicMsg = "unknown error";
let anthropicType = "api_error";
if (dataLine) {
try {
const data = JSON.parse(dataLine.slice(5).trim());
if (data?.error?.message) anthropicMsg = String(data.error.message);
if (data?.error?.type) anthropicType = String(data.error.type);
} catch {
// leave defaults
}
}
const oaiPayload = JSON.stringify({ error: { message: anthropicMsg, type: anthropicType } });
emit(`data: ${oaiPayload}\n\n`);
emitDone();
errored = true;
return;
}
// Standard Anthropic event — parse + translate using shared state.
const parsed = parseSSEChunk(block);
for (const p of parsed) {
const out = translateEvent(st, p);
if (out) emit(out);
}
}
function emitDone(): void {
if (doneSent) return;
doneSent = true;
emit("data: [DONE]\n\n");
}
}
// Re-export for tests
export type { ParsedSSE, TranslationState };
|