http.ts · Source

http.tspackages/ai/src/providers/http.tsraw ← providers4,950 B← Back to the app
1/**
2 * @factory/ai — provider HTTP/SSE plumbing (shared by all adapters).
3 *
4 * All adapters use the injected `fetch` (default global fetch) and one shared
5 * SSE line parser. Provider SDK types never leave the adapter.
6 */
7 
8export interface HttpOptions {
9  fetchImpl?: typeof fetch;
10  /** Default 60_000 ms for chat, 10_000 for embed (arch §2.2). */
11  timeoutMs?: number;
12  /** Stream idle timeout (arch §2.4: 30 s). Default 30_000. */
13  idleMs?: number;
14}
15 
16export class ProviderHttpError extends Error {
17  readonly status: number;
18  readonly retryable: boolean;
19  constructor(status: number, message: string, retryable: boolean) {
20    super(message);
21    this.name = "ProviderHttpError";
22    this.status = status;
23    this.retryable = retryable;
24  }
25}
26 
27/** 429 / 5xx are retryable (gateway honors retry-after); transport errors too. */
28export function isRetryableStatus(status: number): boolean {
29  return status === 429 || status >= 500;
30}
31 
32export async function httpJson<T>(
33  url: string,
34  init: RequestInit,
35  options: HttpOptions,
36): Promise<T> {
37  const controller = new AbortController();
38  const timer = setTimeout(() => controller.abort(), options.timeoutMs ?? 60_000);
39  try {
40    const fetchImpl = options.fetchImpl ?? fetch;
41    const res = await fetchImpl(url, { ...init, signal: controller.signal });
42    if (!res.ok) {
43      const body = await safeText(res);
44      throw new ProviderHttpError(
45        res.status,
46        `HTTP ${res.status}: ${truncate(body, 200)}`,
47        isRetryableStatus(res.status),
48      );
49    }
50    return (await res.json()) as T;
51  } finally {
52    clearTimeout(timer);
53  }
54}
55 
56/**
57 * Stream the response body as SSE `data:` payloads. Yields each parsed JSON
58 * payload; the `data: [DONE]` sentinel terminates the stream.
59 */
60/**
61 * TASK-0122: `signal` (caller cancellation) races each read — an abort ends
62 * the stream with an {@link abortError} WITHOUT keeping the fetch open
63 * (the reader is cancelled, not merely released).
64 */
65export async function* sseData<T>(response: Response, options: HttpOptions, signal?: AbortSignal): AsyncIterable<T> {
66  if (!response.body) {
67    throw new ProviderHttpError(502, "provider returned no body", true);
68  }
69  if (signal?.aborted) throw abortError();
70  const reader = response.body.getReader();
71  const decoder = new TextDecoder("utf8");
72  const idleMs = options.idleMs ?? 30_000;
73  let buffer = "";
74  try {
75    for (;;) {
76      const chunk = await readWithTimeout(reader, idleMs, signal);
77      if (chunk.done) break;
78      buffer += decoder.decode(chunk.value, { stream: true });
79      let newline: number;
80      while ((newline = buffer.indexOf("\n")) !== -1) {
81        const line = buffer.slice(0, newline).replace(/\r$/, "");
82        buffer = buffer.slice(newline + 1);
83        if (!line.startsWith("data:")) continue;
84        const payload = line.slice(5).trim();
85        if (payload.length === 0) continue;
86        if (payload === "[DONE]") return;
87        yield JSON.parse(payload) as T;
88      }
89    }
90  } finally {
91    // Abort: cancel the body (torn down the fetch socket) instead of merely
92    // releasing the lock. Normal completion: plain release.
93    if (signal?.aborted) {
94      await reader.cancel().catch(() => {});
95    } else {
96      reader.releaseLock();
97    }
98  }
99}
100 
101/** The canonical caller-abort error (adapters + gateway agree on its shape). */
102export function abortError(): Error {
103  const err = new Error("The operation was aborted");
104  err.name = "AbortError";
105  return err;
106}
107 
108/** Result of `reader.read()` (mirrors the stream lib's `ReadableStreamReadResult`). */
109type ReadResult<T> =
110  | { done: false; value: T }
111  | { done: true; value: T | undefined };
112 
113async function readWithTimeout(
114  reader: ReadableStreamDefaultReader<Uint8Array>,
115  timeoutMs: number,
116  signal?: AbortSignal,
117): Promise<ReadResult<Uint8Array>> {
118  let timer: ReturnType<typeof setTimeout> | undefined;
119  let onAbort: (() => void) | undefined;
120  const timeout = new Promise<"timeout">((resolve) => {
121    timer = setTimeout(() => resolve("timeout"), timeoutMs);
122  });
123  // Caller abort wins over both the idle timeout and the read.
124  const abort = new Promise<"abort">((resolve) => {
125    if (!signal) return;
126    if (signal.aborted) {
127      resolve("abort");
128      return;
129    }
130    onAbort = () => resolve("abort");
131    signal.addEventListener("abort", onAbort, { once: true });
132  });
133  try {
134    const result = await Promise.race([reader.read(), timeout, abort]);
135    if (result === "abort") throw abortError();
136    if (result === "timeout") throw new ProviderHttpError(504, "stream idle timeout", true);
137    return result;
138  } finally {
139    if (timer) clearTimeout(timer);
140    if (onAbort && signal) signal.removeEventListener("abort", onAbort);
141  }
142}
143 
144function safeText(res: Response): Promise<string> {
145  return res.text().catch(() => "");
146}
147 
148function truncate(s: string, n: number): string {
149  return s.length > n ? s.slice(0, n) + "…" : s;
150}
151