spb/chat-spboucher Public

1// Author: Simon-Pierre Boucher2// Contact: contact@spboucher.ai3// Project: chat.spboucher.ai45// Parses the OpenRouter SSE wire stream into normalized ChatStreamEvents.67import { orChatStream } from "./client";8import { parseStreamChunk } from "./schemas";9import { AIError, normalizeHttpError, normalizeNetworkError } from "./errors";10import type { ChatStreamEvent, GenerationRequest, ModelDefinition, ModelGateway } from "./types";11import { fetchModels } from "./models";1213/**14 * Stream a generation as normalized events. Cancellation propagates upstream15 * through request.signal — aborting truly stops the OpenRouter generation.16 */17export async function* streamGeneration(18  request: GenerationRequest,19  generationId: string20): AsyncGenerator<ChatStreamEvent> {21  const signal = request.signal ?? new AbortController().signal;2223  const body: Record<string, unknown> = {24    model: request.model,25    messages: request.messages,26  };27  if (request.temperature !== undefined) body.temperature = request.temperature;28  if (request.maxTokens !== undefined) body.max_tokens = request.maxTokens;29  if (request.routing) Object.assign(body, request.routing);3031  let res: Response;32  try {33    res = await orChatStream(body, signal);34  } catch (e) {35    const err = e instanceof AIError ? e : normalizeNetworkError(e, request.model);36    yield { type: "generation.error", message: err.message, retryable: err.retryable };37    return;38  }3940  yield { type: "generation.start", generationId, model: request.model };4142  const reader = res.body!.getReader();43  const decoder = new TextDecoder();44  let buffer = "";45  const openToolCalls = new Set<string>();4647  try {48    for (;;) {49      const { done, value } = await reader.read();50      if (done) break;51      buffer += decoder.decode(value, { stream: true });5253      // SSE frames are separated by newlines; each data line is "data: {json}" or "data: [DONE]"54      let newlineIdx: number;55      while ((newlineIdx = buffer.indexOf("\n")) !== -1) {56        const line = buffer.slice(0, newlineIdx).trim();57        buffer = buffer.slice(newlineIdx + 1);58        if (!line.startsWith("data:")) continue; // comments / keep-alives59        const payload = line.slice(5).trim();60        if (payload === "[DONE]") continue;6162        const chunk = parseStreamChunk(payload);63        if (!chunk) continue;6465        if (chunk.error) {66          const status = typeof chunk.error.code === "number" ? chunk.error.code : 500;67          const err = normalizeHttpError(status, chunk.error.message ?? "", request.model);68          yield { type: "generation.error", message: err.message, retryable: err.retryable };69          return;70        }7172        const delta = chunk.choices?.[0]?.delta;73        if (delta?.reasoning) {74          yield { type: "reasoning.delta", text: delta.reasoning };75        }76        if (delta?.content) {77          yield { type: "content.delta", text: delta.content };78        }79        if (delta?.tool_calls) {80          for (const tc of delta.tool_calls) {81            const id = tc.id ?? `tool_${tc.index ?? 0}`;82            if (tc.function?.name && !openToolCalls.has(id)) {83              openToolCalls.add(id);84              yield { type: "tool.start", toolCallId: id, name: tc.function.name };85            }86            if (tc.function?.arguments) {87              yield { type: "tool.delta", toolCallId: id, argumentsDelta: tc.function.arguments };88            }89          }90        }91        if (chunk.usage) {92          yield {93            type: "usage",94            promptTokens: chunk.usage.prompt_tokens,95            completionTokens: chunk.usage.completion_tokens,96            reasoningTokens: chunk.usage.completion_tokens_details?.reasoning_tokens,97            cachedTokens: chunk.usage.prompt_tokens_details?.cached_tokens,98            totalTokens: chunk.usage.total_tokens,99            cost: chunk.usage.cost,100          };101        }102      }103    }104    yield { type: "generation.end" };105  } catch (e) {106    if (signal.aborted) {107      // Cancellation is not an error; the caller marks the generation cancelled.108      return;109    }110    const err = normalizeNetworkError(e, request.model);111    yield { type: "generation.error", message: err.message, retryable: err.retryable };112  } finally {113    reader.cancel().catch(() => {});114  }115}116117/** The initial (and only) gateway implementation. */118export class OpenRouterGateway implements ModelGateway {119  async *stream(request: GenerationRequest): AsyncIterable<ChatStreamEvent> {120    yield* streamGeneration(request, crypto.randomUUID());121  }122  listModels(): Promise<ModelDefinition[]> {123    return fetchModels();124  }125}126