diff --git a/README.md b/README.md
index c61e416..8f88f3d 100644
--- a/README.md
+++ b/README.md
@@ -35,7 +35,19 @@ const { object } = await generateObject({
 });
 ```
 
-`generateText` and `generateObject` share one router per process (`getAi()`), so rotation and parking see every call. Build your own with `createAiProviders({ env, c0mpute, openai, anthropic, router })`.
+### Streaming
+
+```js
+import { streamText } from '@profullstack/ai-providers';
+
+const s = streamText({ system: 'Translate to French.', user: post.body, maxTokens: 2000 });
+for await (const piece of s.textStream) res.write(piece);
+const { provider, model, text } = await s.result;
+```
+
+c0mpute streams its job's `token` events; OpenAI and Anthropic stream with `stream: true`. Fallback works as for `generateText` until the first piece of text has gone out. After that a failure is final (the error has `partial: true`), because the caller has already shown that text. `onToken` in the request is an alternative to iterating, and `getAi().streamText(req)` returns just the final result.
+
+`generateText`, `generateObject` and `streamText` share one router per process (`getAi()`), so rotation and parking see every call. Build your own with `createAiProviders({ env, c0mpute, openai, anthropic, router })`.
 
 ### Keep your own provider code
 
@@ -75,6 +87,7 @@ A thrown error with `status` 401/402/403, or a 429 or 400 that mentions credits
 npx @profullstack/ai-providers status
 npx @profullstack/ai-providers ask "one line about c0mpute"
 npx @profullstack/ai-providers ask --json 'Return {"ok": true}'
+npx @profullstack/ai-providers ask --stream "a haiku about peer nodes"
 ```
 
 ## What c0mpute does not do
diff --git a/ROLLOUT.md b/ROLLOUT.md
index fb1fbbc..e35ce9b 100644
--- a/ROLLOUT.md
+++ b/ROLLOUT.md
@@ -41,7 +41,7 @@ Not conversions:
 
 | Need | Who | Plan |
 |---|---|---|
-| Streaming text | rssamplifier translate, parklookup trip generator | `streamText()` over the same SSE reader (0.2.0) |
+| Streaming text | rssamplifier translate, parklookup trip generator | **Done in 0.2.0:** `streamText()` over the same SSE reader, paid fallbacks stream too |
 | Tool calling | c0upons agent loop | JSON-schema tool emulation on c0mpute, real tools on paid fallbacks |
 | Vision | aiornot, imageroster, mediaanalyzer, linkedin-networking-bot | stays on vendor until a c0mpute vision model is a house model |
 | OpenAI-compatible endpoint | vu1nz (Python) and anything non-JS | Infernet `/v1/chat/completions` shim over `/api/chat` |
diff --git a/index.d.ts b/index.d.ts
index 2461e7a..ed4498c 100644
--- a/index.d.ts
+++ b/index.d.ts
@@ -39,6 +39,22 @@ export interface ProviderOutput {
 export interface TextProvider {
   readonly name: ProviderName;
   generate(req: TextRequest): Promise<ProviderOutput>;
+  /** Optional. Without it, streamText sends the whole generate() answer as one token. */
+  stream?(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
+}
+
+export interface StreamRequest extends TextRequest {
+  /** Called with each piece of text as it arrives. */
+  onToken?: (text: string) => void;
+  /** Only these providers, by name. */
+  only?: string[];
+}
+
+export interface TextStream extends AsyncIterable<string> {
+  /** Each piece of text; throws if the call fails. */
+  textStream: AsyncGenerator<string, void, unknown>;
+  /** The same result generateText returns, once the stream is complete. */
+  result: Promise<TextResult>;
 }
 
 export interface Attempt {
@@ -116,6 +132,8 @@ export declare class AiRouter {
     opts?: { feature?: string; only?: string[] },
   ): Promise<{ provider: ProviderName; value: T; attempts: Attempt[] }>;
   generateText(req: TextRequest): Promise<TextResult>;
+  /** After the first token a failure is final and carries `partial: true`. */
+  streamText(req: StreamRequest): Promise<TextResult>;
   generateObject<T = unknown>(req: ObjectRequest<T>): Promise<ObjectResult<T>>;
 }
 
@@ -132,6 +150,8 @@ export declare class ProviderError extends Error {
 }
 export declare class AllProvidersFailedError extends Error {
   readonly attempts: Attempt[];
+  /** true when streamText had already sent text before the failure. */
+  readonly partial?: boolean;
   readonly skipped: { name: string; waitMs: number; reason: string }[];
 }
 export declare function classify(status: number, message?: string): FailureKind;
@@ -155,6 +175,7 @@ export declare class C0mputeProvider implements TextProvider {
   constructor(opts?: C0mputeOptions);
   readonly name: 'c0mpute';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 
 export interface PaidOptions {
@@ -170,11 +191,13 @@ export declare class OpenAIProvider implements TextProvider {
   constructor(opts?: PaidOptions);
   readonly name: 'openai';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 export declare class AnthropicProvider implements TextProvider {
   constructor(opts?: PaidOptions);
   readonly name: 'anthropic';
   generate(req: TextRequest): Promise<ProviderOutput>;
+  stream(req: TextRequest, onToken: (text: string) => void): Promise<ProviderOutput>;
 }
 
 export declare function extractJson(text: string): unknown;
@@ -194,3 +217,8 @@ export declare function createAiProviders(opts?: CreateOptions): AiRouter;
 export declare function getAi(): AiRouter;
 export declare function generateText(req: TextRequest): Promise<TextResult>;
 export declare function generateObject<T = unknown>(req: ObjectRequest<T>): Promise<ObjectResult<T>>;
+export declare function streamText(req: StreamRequest, router?: AiRouter): TextStream;
+export declare function sseEvents(
+  body: ReadableStream<Uint8Array>,
+  onChunk?: () => void,
+): AsyncGenerator<{ event?: string; data: string }, void, unknown>;
diff --git a/package.json b/package.json
index c239e1c..ceb9703 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@profullstack/ai-providers",
-  "version": "0.1.2",
+  "version": "0.2.0",
   "type": "module",
   "description": "One text-model call for every Profullstack app: c0mpute first, OpenAI and Anthropic as alternating fallbacks, dead keys parked. Zero dependencies.",
   "keywords": [
diff --git a/src/c0mpute.js b/src/c0mpute.js
index 12184ff..d976844 100644
--- a/src/c0mpute.js
+++ b/src/c0mpute.js
@@ -16,6 +16,7 @@
  * so "any node" is off unless `anyModel` is set.
  */
 import { ProviderError, toProviderError } from './errors.js';
+import { sseEvents } from './sse.js';
 
 export const DEFAULT_C0MPUTE_URL = 'https://infernetprotocol.com';
 export const DEFAULT_C0MPUTE_MODELS = [
@@ -56,7 +57,23 @@ export class C0mputeProvider {
   }
 
   /** @param {import('../index.js').TextRequest} req */
-  async generate(req) {
+  generate(req) {
+    return this.#call(req);
+  }
+
+  /**
+   * Same as generate, calling `onToken` with each piece of text as the node
+   * writes it. Once a token has gone out, a stall or a dropped stream is an
+   * error rather than a quiet retry on the next model.
+   *
+   * @param {import('../index.js').TextRequest} req
+   * @param {(text: string) => void} onToken
+   */
+  stream(req, onToken) {
+    return this.#call(req, onToken);
+  }
+
+  async #call(req, onToken) {
     const release = await this.slots.acquire();
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), req.timeoutMs ?? this.timeoutMs);
@@ -119,6 +136,13 @@ export class C0mputeProvider {
           }, this.stallMs);
         };
         let out;
+        let emitted = false;
+        const emit = onToken
+          ? (t) => {
+              emitted = true;
+              onToken(t);
+            }
+          : undefined;
         try {
           arm();
           const stream = await this.fetch(`${this.baseUrl}/api/chat/stream/${job.jobId}`, {
@@ -134,9 +158,16 @@ export class C0mputeProvider {
               kind: 'capacity',
             });
           }
-          out = await readStream(stream.body, arm);
+          out = await readStream(stream.body, arm, emit);
         } catch (err) {
           if (!stalled) throw err;
+          if (emitted) {
+            throw new ProviderError(
+              'c0mpute',
+              `${model ?? 'any'}: stopped streaming for ${Math.round(this.stallMs / 1000)}s mid-answer (job ${job.jobId})`,
+              { status: 504, kind: 'capacity' },
+            );
+          }
           last = new ProviderError(
             'c0mpute',
             `${model ?? 'any'}: no answer for ${Math.round(this.stallMs / 1000)}s (job ${job.jobId})`,
@@ -180,49 +211,36 @@ export function toMessages(req) {
   ];
 }
 
-/** Reads `meta`, `token` and `done` events until `done` or `error`. */
-export async function readStream(body, onEvent) {
-  const reader = body.getReader();
-  const decoder = new TextDecoder();
-  let buffer = '';
+/**
+ * Reads `meta`, `token` and `done` events until `done` or `error`.
+ * `onToken` gets each token as it arrives; if `done` carries a fuller text than
+ * the tokens added up to, the missing tail is emitted too.
+ */
+export async function readStream(body, onEvent, onToken) {
   let text = '';
   let model;
-  for (;;) {
-    const { value, done } = await reader.read();
-    if (value?.length) onEvent?.();
-    if (value) buffer += decoder.decode(value, { stream: true });
-    buffer = buffer.replace(/\r\n/g, '\n');
-    let cut;
-    while ((cut = buffer.indexOf('\n\n')) >= 0) {
-      const block = buffer.slice(0, cut);
-      buffer = buffer.slice(cut + 2);
-      const event = /^event:\s*(.+)$/m.exec(block)?.[1]?.trim();
-      const raw = block
-        .split('\n')
-        .filter((l) => l.startsWith('data:'))
-        .map((l) => l.slice(5).trimStart())
-        .join('\n');
-      if (!event || !raw) continue;
-      let data;
-      try {
-        data = JSON.parse(raw);
-      } catch {
-        continue;
-      }
-      if (event === 'meta' && typeof data.model === 'string') model = data.model;
-      else if (event === 'token' && typeof data.text === 'string') text += data.text;
-      else if (event === 'done') {
-        await reader.cancel().catch(() => {});
-        return { text: typeof data.text === 'string' && data.text ? data.text : text, model };
-      } else if (event === 'error') {
-        await reader.cancel().catch(() => {});
-        throw new ProviderError('c0mpute', String(data.message ?? data.error ?? 'job failed'), {
-          status: 502,
-          kind: 'capacity',
-        });
-      }
+  for await (const { event, data: raw } of sseEvents(body, onEvent)) {
+    if (!event || !raw) continue;
+    let data;
+    try {
+      data = JSON.parse(raw);
+    } catch {
+      continue;
+    }
+    if (event === 'meta' && typeof data.model === 'string') model = data.model;
+    else if (event === 'token' && typeof data.text === 'string') {
+      text += data.text;
+      if (data.text) onToken?.(data.text);
+    } else if (event === 'done') {
+      const final = typeof data.text === 'string' && data.text ? data.text : text;
+      if (onToken && final.length > text.length && final.startsWith(text)) onToken(final.slice(text.length));
+      return { text: final, model };
+    } else if (event === 'error') {
+      throw new ProviderError('c0mpute', String(data.message ?? data.error ?? 'job failed'), {
+        status: 502,
+        kind: 'capacity',
+      });
     }
-    if (done) break;
   }
   // No done event: the node or the connection died mid-answer.
   throw new ProviderError(
diff --git a/src/cli.js b/src/cli.js
index 434a034..488ea80 100755
--- a/src/cli.js
+++ b/src/cli.js
@@ -3,6 +3,7 @@
  * ai-providers status            which providers are configured, in call order
  * ai-providers ask "prompt"      one call through the router; prints who answered
  * ai-providers ask --json "..."  same, asking for a JSON object
+ * ai-providers ask --stream "..." same, printing the text as it arrives
  */
 import { createAiProviders } from './index.js';
 
@@ -14,23 +15,27 @@ if (cmd === 'status' || !cmd) {
   console.log(JSON.stringify({ order: order.map((p) => p.name), providers: ai.status() }, null, 2));
 } else if (cmd === 'ask') {
   const json = rest[0] === '--json';
-  const prompt = (json ? rest.slice(1) : rest).join(' ');
+  const stream = rest[0] === '--stream';
+  const prompt = (json || stream ? rest.slice(1) : rest).join(' ');
   if (!prompt) {
-    console.error('usage: ai-providers ask [--json] "prompt"');
+    console.error('usage: ai-providers ask [--json|--stream] "prompt"');
     process.exit(2);
   }
   try {
     const r = json
       ? await ai.generateObject({ user: prompt, maxTokens: 800 })
-      : await ai.generateText({ user: prompt, maxTokens: 800 });
+      : stream
+        ? await ai.streamText({ user: prompt, maxTokens: 800, onToken: (t) => process.stdout.write(t) })
+        : await ai.generateText({ user: prompt, maxTokens: 800 });
     console.error(`# ${r.provider} ${r.model} in=${r.usage.inputTokens} out=${r.usage.outputTokens}`);
     for (const a of r.attempts) console.error(`# skipped: ${a.error.message}`);
-    console.log(json ? JSON.stringify(r.object, null, 2) : r.text);
+    if (stream) process.stdout.write('\n');
+    else console.log(json ? JSON.stringify(r.object, null, 2) : r.text);
   } catch (err) {
     console.error(err.message);
     process.exit(1);
   }
 } else {
-  console.error('usage: ai-providers status | ask [--json] "prompt"');
+  console.error('usage: ai-providers status | ask [--json|--stream] "prompt"');
   process.exit(2);
 }
diff --git a/src/index.js b/src/index.js
index c8d65fb..2e6436a 100644
--- a/src/index.js
+++ b/src/index.js
@@ -14,6 +14,7 @@ export * from './c0mpute.js';
 export * from './paid.js';
 export * from './router.js';
 export * from './json.js';
+export * from './sse.js';
 
 const OFF = /^(0|off|false|no|none|disabled?)$/i;
 const list = (v) =>
@@ -99,3 +100,75 @@ export function getAi() {
 export const generateText = (req) => getAi().generateText(req);
 /** Shorthand for getAi().generateObject(req). */
 export const generateObject = (req) => getAi().generateObject(req);
+
+/**
+ * Streams text through the shared router (or `router`).
+ *
+ *   const s = streamText({ user: 'Translate ...' });
+ *   for await (const piece of s.textStream) res.write(piece);
+ *   const { provider, text } = await s.result;
+ *
+ * `textStream` (also the object's own async iterator) yields each piece of
+ * text and throws if the call fails. `result` resolves to the generateText
+ * result. `req.onToken`, if given, is called too.
+ *
+ * @param {import('../index.js').StreamRequest} req
+ * @param {AiRouter} [router]
+ */
+export function streamText(req, router = getAi()) {
+  const queue = [];
+  let wake;
+  let finished = false;
+  let failure;
+  const nudge = () => {
+    const w = wake;
+    wake = undefined;
+    w?.();
+  };
+  const result = router
+    .streamText({
+      ...req,
+      onToken: (t) => {
+        req.onToken?.(t);
+        queue.push(t);
+        nudge();
+      },
+    })
+    .then(
+      (r) => {
+        finished = true;
+        nudge();
+        return r;
+      },
+      (err) => {
+        failure = err;
+        finished = true;
+        nudge();
+        throw err;
+      },
+    );
+  // The failure also surfaces through textStream; a caller that only iterates
+  // must not get an unhandled rejection.
+  result.catch(() => {});
+  async function* pieces() {
+    for (;;) {
+      if (queue.length) {
+        yield queue.shift();
+        continue;
+      }
+      if (finished) {
+        if (failure) throw failure;
+        return;
+      }
+      await new Promise((resolve) => (wake = resolve));
+    }
+  }
+  const textStream = pieces();
+  return {
+    textStream,
+    result,
+    [Symbol.asyncIterator]() {
+      return textStream;
+    },
+  };
+}
diff --git a/src/paid.js b/src/paid.js
index 096c92c..16cae6c 100644
--- a/src/paid.js
+++ b/src/paid.js
@@ -5,6 +5,7 @@
  */
 import { ProviderError, toProviderError } from './errors.js';
 import { toMessages } from './c0mpute.js';
+import { sseEvents } from './sse.js';
 
 export const DEFAULT_OPENAI_MODEL = 'gpt-5-mini';
 export const DEFAULT_ANTHROPIC_MODEL = 'claude-haiku-5-5';
@@ -43,7 +44,16 @@ export class OpenAIProvider {
     this.fetch = opts.fetch ?? globalThis.fetch;
   }
 
-  async generate(req) {
+  generate(req) {
+    return this.#call(req);
+  }
+
+  /** Chat Completions with `stream: true`; `onToken` gets each delta. */
+  stream(req, onToken) {
+    return this.#call(req, onToken);
+  }
+
+  async #call(req, onToken) {
     const t = withTimeout(req, this.timeoutMs);
     try {
       const model = req.models?.openai || this.model;
@@ -55,10 +65,12 @@ export class OpenAIProvider {
           messages: toMessages(req),
           max_completion_tokens: req.maxTokens ?? 1024,
           ...(req.json ? { response_format: { type: 'json_object' } } : {}),
+          ...(onToken ? { stream: true, stream_options: { include_usage: true } } : {}),
         }),
         signal: t.signal,
       });
       if (!res.ok) throw await failure('openai', res);
+      if (onToken) return await readOpenAIStream(res, model, onToken);
       const j = await res.json();
       return {
         text: j.choices?.[0]?.message?.content ?? '',
@@ -87,7 +99,16 @@ export class AnthropicProvider {
     this.fetch = opts.fetch ?? globalThis.fetch;
   }
 
-  async generate(req) {
+  generate(req) {
+    return this.#call(req);
+  }
+
+  /** Messages with `stream: true`; `onToken` gets each text delta. */
+  stream(req, onToken) {
+    return this.#call(req, onToken);
+  }
+
+  async #call(req, onToken) {
     const t = withTimeout(req, this.timeoutMs);
     try {
       const model = req.models?.anthropic || this.model;
@@ -106,10 +127,12 @@ export class AnthropicProvider {
           max_tokens: req.maxTokens ?? 1024,
           ...(system ? { system } : {}),
           messages,
+          ...(onToken ? { stream: true } : {}),
         }),
         signal: t.signal,
       });
       if (!res.ok) throw await failure('anthropic', res);
+      if (onToken) return await readAnthropicStream(res, model, onToken);
       const j = await res.json();
       const text = (j.content ?? [])
         .filter((b) => b.type === 'text')
@@ -131,3 +154,86 @@ export class AnthropicProvider {
     }
   }
 }
+
+const parse = (raw) => {
+  try {
+    return JSON.parse(raw);
+  } catch {
+    return undefined;
+  }
+};
+
+async function readOpenAIStream(res, model, onToken) {
+  if (!res.body) throw new ProviderError('openai', 'no response body to stream', { status: 502 });
+  let text = '';
+  let usage;
+  let finished = false;
+  for await (const { data: raw } of sseEvents(res.body)) {
+    if (raw === '[DONE]') {
+      finished = true;
+      break;
+    }
+    const j = parse(raw);
+    if (!j) continue;
+    if (j.error) throw new ProviderError('openai', String(j.error.message ?? 'stream error'), { status: 502 });
+    if (j.model) model = j.model;
+    if (j.usage) usage = j.usage;
+    const piece = j.choices?.[0]?.delta?.content;
+    if (typeof piece === 'string' && piece) {
+      text += piece;
+      onToken(piece);
+    }
+  }
+  if (!finished) {
+    throw new ProviderError('openai', 'the stream ended before the answer finished', {
+      status: 502,
+      kind: 'capacity',
+    });
+  }
+  return {
+    text,
+    model,
+    usage: {
+      inputTokens: usage?.prompt_tokens ?? 0,
+      outputTokens: usage?.completion_tokens ?? Math.ceil(text.length / 4),
+      estimated: !usage,
+    },
+  };
+}
+
+async function readAnthropicStream(res, model, onToken) {
+  if (!res.body) throw new ProviderError('anthropic', 'no response body to stream', { status: 502 });
+  let text = '';
+  let inputTokens = 0;
+  let outputTokens = 0;
+  let finished = false;
+  for await (const { event, data: raw } of sseEvents(res.body)) {
+    const j = parse(raw);
+    if (!j) continue;
+    const type = j.type ?? event;
+    if (type === 'message_start') {
+      if (j.message?.model) model = j.message.model;
+      inputTokens = j.message?.usage?.input_tokens ?? inputTokens;
+      outputTokens = j.message?.usage?.output_tokens ?? outputTokens;
+    } else if (type === 'content_block_delta' && j.delta?.type === 'text_delta' && j.delta.text) {
+      text += j.delta.text;
+      onToken(j.delta.text);
+    } else if (type === 'message_delta') {
+      outputTokens = j.usage?.output_tokens ?? outputTokens;
+    } else if (type === 'message_stop') {
+      finished = true;
+      break;
+    } else if (type === 'error') {
+      const message = String(j.error?.message ?? 'stream error');
+      const status = j.error?.type === 'overloaded_error' ? 529 : 502;
+      throw new ProviderError('anthropic', message, { status });
+    }
+  }
+  if (!finished) {
+    throw new ProviderError('anthropic', 'the stream ended before the answer finished', {
+      status: 502,
+      kind: 'capacity',
+    });
+  }
+  return { text, model, usage: { inputTokens, outputTokens, estimated: false } };
+}
diff --git a/src/router.js b/src/router.js
index 6bb3738..aec535e 100644
--- a/src/router.js
+++ b/src/router.js
@@ -104,7 +104,9 @@ export class AiRouter {
    *
    * @template T
    * @param {(provider: any) => Promise<T>} task
-   * @param {{ feature?: string, only?: string[] }} [opts]
+   * @param {{ feature?: string, only?: string[], final?: () => boolean }} [opts]
+   *   `final`, checked after a failure: true stops here instead of moving on
+   *   (streamText sets it once text has gone out).
    * @returns {Promise<{ provider: string, value: T, attempts: object[] }>}
    */
   async run(task, opts = {}) {
@@ -131,6 +133,11 @@ export class AiRouter {
           kind: error.kind,
           error: error.message,
         });
+        if (opts.final?.()) {
+          const err = new AllProvidersFailedError(attempts, skipped);
+          err.partial = true;
+          throw err;
+        }
       }
     }
     throw new AllProvidersFailedError(attempts, skipped);
@@ -148,6 +155,40 @@ export class AiRouter {
     return { ...value, provider, attempts };
   }
 
+  /**
+   * Streams text: `req.onToken` gets each piece as it arrives, and the promise
+   * resolves to the same result generateText returns. Providers are tried in
+   * the usual order, but only until one has streamed something: after the
+   * first token a failure is final (`error.partial === true`), since the
+   * caller has already shown that text. A provider with no `stream` method is
+   * called with `generate` and its whole answer sent as one token.
+   *
+   * @param {import('../index.js').StreamRequest} req
+   */
+  async streamText(req) {
+    let emitted = false;
+    const send = (t) => {
+      if (!t) return;
+      emitted = true;
+      req.onToken?.(t);
+    };
+    const { provider, value, attempts } = await this.run(
+      async (p) => {
+        const out =
+          typeof p.stream === 'function'
+            ? await p.stream(req, send)
+            : await p.generate(req).then((o) => {
+                if (o.text?.trim()) send(o.text);
+                return o;
+              });
+        if (!out.text?.trim()) throw new ProviderError(p.name, 'empty answer', { kind: 'output' });
+        return out;
+      },
+      { feature: req.feature, only: req.only, final: () => emitted },
+    );
+    return { ...value, provider, attempts };
+  }
+
   /**
    * Asks for JSON, extracts it, validates it against `schema`. An unusable
    * answer gets one repair attempt on the same provider before moving on.
diff --git a/src/sse.js b/src/sse.js
new file mode 100644
index 0000000..1ff7b85
--- /dev/null
+++ b/src/sse.js
@@ -0,0 +1,42 @@
+/**
+ * Server-sent events from a fetch body, one `{event, data}` per block.
+ * Shared by c0mpute's job stream and the paid providers' streaming APIs.
+ * Leaving the loop early (return/throw) cancels the body.
+ *
+ * @param {ReadableStream<Uint8Array>} body
+ * @param {() => void} [onChunk] called for every non-empty chunk read
+ */
+export async function* sseEvents(body, onChunk) {
+  const reader = body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = '';
+  let finished = false;
+  try {
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (value?.length) onChunk?.();
+      if (value) buffer += decoder.decode(value, { stream: true });
+      if (done) buffer += '\n\n';
+      buffer = buffer.replace(/\r\n/g, '\n');
+      let cut;
+      while ((cut = buffer.indexOf('\n\n')) >= 0) {
+        const block = buffer.slice(0, cut);
+        buffer = buffer.slice(cut + 2);
+        if (!block.trim()) continue;
+        const event = /^event:\s*(.+)$/m.exec(block)?.[1]?.trim();
+        const data = block
+          .split('\n')
+          .filter((l) => l.startsWith('data:'))
+          .map((l) => l.slice(5).replace(/^ /, ''))
+          .join('\n');
+        yield { event, data };
+      }
+      if (done) {
+        finished = true;
+        return;
+      }
+    }
+  } finally {
+    if (!finished) await reader.cancel().catch(() => {});
+  }
+}
diff --git a/test/stream.test.js b/test/stream.test.js
new file mode 100644
index 0000000..99c5a8a
--- /dev/null
+++ b/test/stream.test.js
@@ -0,0 +1,232 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import {
+  AiRouter,
+  ProviderError,
+  streamText,
+  sseEvents,
+  C0mputeProvider,
+  OpenAIProvider,
+  AnthropicProvider,
+} from '../src/index.js';
+
+const enc = new TextEncoder();
+/** A body that delivers each string as its own chunk, so blocks split across reads. */
+const chunked = (chunks) =>
+  new Response(
+    new ReadableStream({
+      start(c) {
+        for (const s of chunks) c.enqueue(enc.encode(s));
+        c.close();
+      },
+    }),
+    { headers: { 'content-type': 'text/event-stream' } },
+  );
+const named = (events) => chunked(events.map(([e, d]) => `event: ${e}\ndata: ${JSON.stringify(d)}\n\n`));
+const dataOnly = (items) => chunked(items.map((d) => `data: ${typeof d === 'string' ? d : JSON.stringify(d)}\n\n`));
+
+const streamer = (name, pieces, { failAfter } = {}) => ({
+  name,
+  calls: 0,
+  async generate() {
+    throw new Error('generate should not be called');
+  },
+  async stream(_req, onToken) {
+    this.calls++;
+    let text = '';
+    for (const [i, p] of pieces.entries()) {
+      if (failAfter === i) throw new ProviderError(name, 'node died', { status: 502 });
+      text += p;
+      onToken(p);
+    }
+    return { text, model: `${name}-m`, usage: { inputTokens: 1, outputTokens: 1, estimated: true } };
+  },
+});
+const down = (name, status = 503) => ({
+  name,
+  calls: 0,
+  async generate() {
+    this.calls++;
+    throw new ProviderError(name, 'no node', { status });
+  },
+  async stream() {
+    this.calls++;
+    throw new ProviderError(name, 'no node', { status });
+  },
+});
+
+test('sseEvents splits blocks across chunk boundaries and handles CRLF and a missing final blank line', async () => {
+  const body = chunked(['event: a\r\nda', 'ta: {"x":1}\r\n\r\nevent: b\n', 'data: 2\n\ndata: [DONE]']).body;
+  const out = [];
+  for await (const e of sseEvents(body)) out.push(e);
+  assert.deepEqual(out, [
+    { event: 'a', data: '{"x":1}' },
+    { event: 'b', data: '2' },
+    { event: undefined, data: '[DONE]' },
+  ]);
+});
+
+test('C0mputeProvider.stream emits each token and the tail a done event adds', async () => {
+  const fetch = async (url) =>
+    url.endsWith('/api/chat')
+      ? Response.json({ jobId: 'j1', source: 'p2p' })
+      : named([
+          ['meta', { model: 'qwen' }],
+          ['token', { text: 'Bon' }],
+          ['token', { text: 'jour' }],
+          ['done', { text: 'Bonjour !' }],
+        ]);
+  const pieces = [];
+  const out = await new C0mputeProvider({ fetch }).stream({ user: 'hi' }, (t) => pieces.push(t));
+  assert.deepEqual(pieces, ['Bon', 'jour', ' !']);
+  assert.equal(out.text, 'Bonjour !');
+  assert.equal(out.model, 'qwen');
+});
+
+test('C0mputeProvider.stream: a stall after tokens is an error, not a retry on the next model', async () => {
+  let jobs = 0;
+  const fetch = async (url, init) => {
+    if (url.endsWith('/api/chat')) {
+      jobs++;
+      return Response.json({ jobId: `j${jobs}`, source: 'p2p' });
+    }
+    return new Response(
+      new ReadableStream({
+        start(c) {
+          c.enqueue(enc.encode('event: token\ndata: {"text":"Hal"}\n\n'));
+          init.signal.addEventListener('abort', () => c.error(new DOMException('aborted', 'AbortError')));
+        },
+      }),
+    );
+  };
+  const p = new C0mputeProvider({ fetch, models: ['a', 'b'], stallMs: 30 });
+  await assert.rejects(p.stream({ user: 'x' }, () => {}), /mid-answer/);
+  assert.equal(jobs, 1);
+});
+
+test('OpenAIProvider.stream reads deltas and usage', async () => {
+  let body;
+  const fetch = async (_url, init) => {
+    body = JSON.parse(init.body);
+    return dataOnly([
+      { model: 'gpt-x', choices: [{ delta: { role: 'assistant' } }] },
+      { choices: [{ delta: { content: 'Hel' } }] },
+      { choices: [{ delta: { content: 'lo' } }] },
+      { choices: [], usage: { prompt_tokens: 5, completion_tokens: 2 } },
+      '[DONE]',
+    ]);
+  };
+  const pieces = [];
+  const out = await new OpenAIProvider({ apiKey: 'k', fetch }).stream({ user: 'u' }, (t) => pieces.push(t));
+  assert.equal(body.stream, true);
+  assert.deepEqual(body.stream_options, { include_usage: true });
+  assert.deepEqual(pieces, ['Hel', 'lo']);
+  assert.equal(out.text, 'Hello');
+  assert.equal(out.model, 'gpt-x');
+  assert.deepEqual(out.usage, { inputTokens: 5, outputTokens: 2, estimated: false });
+});
+
+test('OpenAIProvider.stream: a stream cut before [DONE] is a capacity error', async () => {
+  const fetch = async () => dataOnly([{ choices: [{ delta: { content: 'Hel' } }] }]);
+  await assert.rejects(
+    new OpenAIProvider({ apiKey: 'k', fetch }).stream({ user: 'u' }, () => {}),
+    (e) => e.kind === 'capacity',
+  );
+});
+
+test('AnthropicProvider.stream reads text deltas and usage', async () => {
+  let body;
+  const fetch = async (_url, init) => {
+    body = JSON.parse(init.body);
+    return named([
+      ['message_start', { type: 'message_start', message: { model: 'claude-x', usage: { input_tokens: 7, output_tokens: 1 } } }],
+      ['content_block_start', { type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } }],
+      ['ping', { type: 'ping' }],
+      ['content_block_delta', { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'Hi' } }],
+      ['content_block_delta', { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: ' there' } }],
+      ['content_block_stop', { type: 'content_block_stop', index: 0 }],
+      ['message_delta', { type: 'message_delta', delta: { stop_reason: 'end_turn' }, usage: { output_tokens: 3 } }],
+      ['message_stop', { type: 'message_stop' }],
+    ]);
+  };
+  const pieces = [];
+  const out = await new AnthropicProvider({ apiKey: 'k', fetch }).stream({ system: 's', user: 'u' }, (t) =>
+    pieces.push(t),
+  );
+  assert.equal(body.stream, true);
+  assert.equal(body.system, 's');
+  assert.deepEqual(pieces, ['Hi', ' there']);
+  assert.equal(out.text, 'Hi there');
+  assert.equal(out.model, 'claude-x');
+  assert.deepEqual(out.usage, { inputTokens: 7, outputTokens: 3, estimated: false });
+});
+
+test('paid streams fail as billing on the same dead-key bodies as generate', async () => {
+  const openai = new OpenAIProvider({
+    apiKey: 'k',
+    fetch: async () => Response.json({ error: { message: 'Incorrect API key provided' } }, { status: 401 }),
+  });
+  await assert.rejects(openai.stream({ user: 'u' }, () => {}), (e) => e.kind === 'billing');
+  const anthropic = new AnthropicProvider({
+    apiKey: 'k',
+    fetch: async () =>
+      Response.json(
+        { type: 'error', error: { message: 'Your credit balance is too low to access the Anthropic API.' } },
+        { status: 400 },
+      ),
+  });
+  await assert.rejects(anthropic.stream({ user: 'u' }, () => {}), (e) => e.kind === 'billing');
+});
+
+test('router.streamText falls back before the first token and reports who answered', async () => {
+  const c = down('c0mpute');
+  const o = streamer('openai', ['a', 'b']);
+  const pieces = [];
+  const r = await new AiRouter({ primary: [c], fallback: [o] }).streamText({
+    user: 'x',
+    onToken: (t) => pieces.push(t),
+  });
+  assert.equal(r.provider, 'openai');
+  assert.equal(r.text, 'ab');
+  assert.deepEqual(pieces, ['a', 'b']);
+  assert.equal(r.attempts.length, 1);
+});
+
+test('router.streamText: after a token, a failure is final and partial', async () => {
+  const c = streamer('c0mpute', ['a', 'b'], { failAfter: 1 });
+  const o = streamer('openai', ['z']);
+  await assert.rejects(
+    new AiRouter({ primary: [c], fallback: [o] }).streamText({ user: 'x' }),
+    (e) => e.partial === true && /node died/.test(e.message),
+  );
+  assert.equal(o.calls, 0);
+});
+
+test('router.streamText uses generate for a provider without stream', async () => {
+  const plain = {
+    name: 'custom',
+    async generate() {
+      return { text: 'whole', model: 'm', usage: { inputTokens: 0, outputTokens: 0, estimated: true } };
+    },
+  };
+  const pieces = [];
+  const r = await new AiRouter({ primary: [plain] }).streamText({ user: 'x', onToken: (t) => pieces.push(t) });
+  assert.deepEqual(pieces, ['whole']);
+  assert.equal(r.provider, 'custom');
+});
+
+test('streamText() yields pieces and resolves result; failures throw from the iterator', async () => {
+  const s = streamText({ user: 'x' }, new AiRouter({ primary: [down('c0mpute')], fallback: [streamer('anthropic', ['x', 'y', 'z'])] }));
+  const got = [];
+  for await (const p of s) got.push(p);
+  assert.deepEqual(got, ['x', 'y', 'z']);
+  const r = await s.result;
+  assert.equal(r.provider, 'anthropic');
+  assert.equal(r.text, 'xyz');
+
+  const bad = streamText({ user: 'x' }, new AiRouter({ primary: [down('c0mpute')] }));
+  await assert.rejects(async () => {
+    for await (const _ of bad.textStream);
+  }, /AI generation failed/);
+  await assert.rejects(bad.result, /AI generation failed/);
+});