feat(ai): add packages/ai provider layer (Chunk #8)
Self-contained SSE-based provider layer for OpenAI-compatible endpoints (Ollama/llama.cpp/LM Studio/vLLM), Anthropic, Gemini, and OpenRouter. - budget.ts: re-export core token-budget engine (countTokens, allocateBudget, DEFAULT_BUDGET) - client.ts: streamCompletion/collectStream/complete with budget enforcement via enforceBudget + measureBudget - discover.ts: auto-detect local providers by probing common /v1/models - adapters.ts: per-provider param mapping + streaming adapters + factory - thinking.ts: thinking/reasoning-param normalization per provider - http.ts: SSE parsing + 429 retry - errors.ts: BudgetExceededError 35 unit/integration tests pass; typecheck and lint clean.
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
/**
|
||||
* Low-level HTTP + SSE helpers for the provider layer.
|
||||
*
|
||||
* This module provides:
|
||||
* - `readSSEEvents`: parse a `text/event-stream` `ReadableStream` into
|
||||
* concatenated `data:` payloads (one per event).
|
||||
* - `fetchWithRetry`: a `fetch` wrapper that retries on 429/5xx with
|
||||
* exponential backoff and surfaces auth errors immediately.
|
||||
*
|
||||
* It is intentionally dependency-free and browser-safe (uses the global
|
||||
* `fetch`/`TextDecoder`/`ReadableStream`). See CHUNK#8.md §3.
|
||||
*/
|
||||
|
||||
import {
|
||||
ProviderAuthError,
|
||||
ProviderError,
|
||||
ProviderRateLimitError,
|
||||
} from "./errors.js";
|
||||
|
||||
/** Read a `text/event-stream` body and yield each event's `data:` payload. */
|
||||
export async function* readSSEEvents(
|
||||
stream: ReadableStream<Uint8Array<ArrayBuffer>>,
|
||||
): AsyncGenerator<string> {
|
||||
const reader = stream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
const dataLines: string[] = [];
|
||||
|
||||
const flush = (): string | null => {
|
||||
if (dataLines.length === 0) {
|
||||
return null;
|
||||
}
|
||||
const value = dataLines.join("\n");
|
||||
dataLines.length = 0;
|
||||
return value;
|
||||
};
|
||||
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (value) {
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
}
|
||||
|
||||
let newlineIndex: number;
|
||||
while ((newlineIndex = buffer.indexOf("\n")) >= 0) {
|
||||
let line = buffer.slice(0, newlineIndex);
|
||||
buffer = buffer.slice(newlineIndex + 1);
|
||||
if (line.endsWith("\r")) {
|
||||
line = line.slice(0, -1);
|
||||
}
|
||||
|
||||
if (line === "") {
|
||||
// Blank line terminates an SSE event.
|
||||
const payload = flush();
|
||||
if (payload !== null) {
|
||||
yield payload;
|
||||
}
|
||||
} else if (line.startsWith("data:")) {
|
||||
dataLines.push(line.slice(5).replace(/^ /, ""));
|
||||
}
|
||||
// Ignore `event:`, `id:`, `retry:` lines.
|
||||
}
|
||||
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Flush any trailing data without a terminating blank line.
|
||||
if (buffer.length > 0) {
|
||||
if (buffer.startsWith("data:")) {
|
||||
dataLines.push(buffer.slice(5).replace(/^ /, ""));
|
||||
}
|
||||
const payload = flush();
|
||||
if (payload !== null) {
|
||||
yield payload;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Parse a JSON SSE event, returning `null` for malformed input. */
|
||||
export function parseSSEEvent<T = unknown>(event: string): T | null {
|
||||
try {
|
||||
return JSON.parse(event) as T;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Sleep for `ms` milliseconds. */
|
||||
export function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
/** Build the retry backoff delay for a given attempt (0-based). */
|
||||
function backoffDelay(attempt: number, maxMs: number): number {
|
||||
const base = 250 * 2 ** attempt;
|
||||
return Math.min(base, maxMs);
|
||||
}
|
||||
|
||||
/**
|
||||
* `fetch` wrapper that retries on 429/5xx with exponential backoff. Auth
|
||||
* errors (401/403) are thrown immediately; other non-retryable errors are
|
||||
* surfaced as-is.
|
||||
*/
|
||||
export async function fetchWithRetry(
|
||||
url: string,
|
||||
init: RequestInit,
|
||||
options: { retries?: number; maxDelayMs?: number; signal?: AbortSignal } = {},
|
||||
): Promise<Response> {
|
||||
const retries = options.retries ?? 2;
|
||||
const maxDelayMs = options.maxDelayMs ?? 5000;
|
||||
let lastError: unknown;
|
||||
|
||||
for (let attempt = 0; attempt <= retries; attempt++) {
|
||||
if (attempt > 0) {
|
||||
await sleep(backoffDelay(attempt - 1, maxDelayMs));
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await fetch(url, init);
|
||||
|
||||
if (response.ok) {
|
||||
return response;
|
||||
}
|
||||
|
||||
const detail = await safeText(response);
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new ProviderAuthError(
|
||||
`Provider rejected credentials (${response.status}).`,
|
||||
);
|
||||
}
|
||||
if (response.status === 429 || response.status >= 500) {
|
||||
lastError = new ProviderRateLimitError(
|
||||
`Provider rate-limited or overloaded (${response.status}).`,
|
||||
);
|
||||
if (attempt < retries) {
|
||||
continue;
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
throw new ProviderError(
|
||||
`Provider request failed (${response.status}): ${detail}`,
|
||||
);
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
throw error;
|
||||
}
|
||||
lastError = error;
|
||||
if (attempt >= retries) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Unreachable, but satisfies the type checker.
|
||||
throw lastError instanceof Error
|
||||
? lastError
|
||||
: new ProviderError("Request failed after retries.");
|
||||
}
|
||||
|
||||
/** Read a response body as text, tolerating errors. */
|
||||
async function safeText(response: Response): Promise<string> {
|
||||
try {
|
||||
return await response.text();
|
||||
} catch {
|
||||
return "(no body)";
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user