feat(c0mpute): onStatus progress events (0.3.0)

README.md

@@ -47,6 +47,8 @@ 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.
 
+A peer job can wait a while: a CPU node reads a long prompt for a minute or more before its first token. Pass `onStatus` to show where it is: `{stage: 'queued', jobId, node, source, model}` once a node is picked, `{stage: 'started', node, engine}` when that node begins (it is reading the prompt until the first token), `{stage: 'writing', tokens}` per token, and `{stage: 'retry', model, reason}` before the next model is tried. Only c0mpute calls it.
+
 `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

index.d.ts

@@ -21,8 +21,16 @@ export interface TextRequest {
   feature?: string;
   timeoutMs?: number;
   signal?: AbortSignal;
+  /** c0mpute only: where the job is, for progress UIs. */
+  onStatus?: (status: C0mputeStatus) => void;
 }
 
+export type C0mputeStatus =
+  | { stage: 'queued'; jobId: string; node: string | null; source: string | null; model: string | null }
+  | { stage: 'started'; node: string | null; engine: string | null; model: string | null }
+  | { stage: 'writing'; tokens: number }
+  | { stage: 'retry'; model: string | undefined; reason: string };
+
 export interface Usage {
   inputTokens: number;
   outputTokens: number;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@profullstack/ai-providers",
-  "version": "0.2.0",
+  "version": "0.3.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": [

src/c0mpute.js

@@ -68,12 +68,26 @@ export class C0mputeProvider {
    *
    * @param {import('../index.js').TextRequest} req
    * @param {(text: string) => void} onToken
+   *
+   * `req.onStatus(status)`, when given, hears where the job is so a UI can say
+   * more than "waiting": `{stage: 'queued', jobId, node, source, model}` once a
+   * node is picked, `{stage: 'started', node, engine, model}` when the node
+   * begins (it is reading the prompt until the first token), `{stage:
+   * 'writing', tokens}` per token, `{stage: 'retry', model, reason}` before the
+   * next model is tried. A throwing listener is ignored.
    */
   stream(req, onToken) {
     return this.#call(req, onToken);
   }
 
   async #call(req, onToken) {
+    const status = (s) => {
+      try {
+        req.onStatus?.(s);
+      } catch {
+        /* a UI bug must not fail the job */
+      }
+    };
     const release = await this.slots.acquire();
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), req.timeoutMs ?? this.timeoutMs);
@@ -110,6 +124,7 @@ export class C0mputeProvider {
             status: 503,
             kind: 'capacity',
           });
+          status({ stage: 'retry', model, reason: job.error ?? 'no node serves this model' });
           continue;
         }
         if (!res.ok || !job.jobId) {
@@ -123,6 +138,13 @@ export class C0mputeProvider {
             kind: 'capacity',
           });
         }
+        status({
+          stage: 'queued',
+          jobId: job.jobId,
+          node: job.provider?.name ?? job.provider?.nodeId ?? null,
+          source: job.source ?? null,
+          model: job.model ?? model ?? null,
+        });
         const attempt = new AbortController();
         const stop = () => attempt.abort();
         controller.signal.addEventListener('abort', stop, { once: true });
@@ -158,7 +180,7 @@ export class C0mputeProvider {
               kind: 'capacity',
             });
           }
-          out = await readStream(stream.body, arm, emit);
+          out = await readStream(stream.body, arm, emit, status);
         } catch (err) {
           if (!stalled) throw err;
           if (emitted) {
@@ -173,6 +195,7 @@ export class C0mputeProvider {
             `${model ?? 'any'}: no answer for ${Math.round(this.stallMs / 1000)}s (job ${job.jobId})`,
             { status: 504, kind: 'capacity' },
           );
+          status({ stage: 'retry', model, reason: last.message });
           continue;
         } finally {
           clearTimeout(stall);
@@ -216,9 +239,10 @@ export function toMessages(req) {
  * `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) {
+export async function readStream(body, onEvent, onToken, onStatus = () => {}) {
   let text = '';
   let model;
+  let tokens = 0;
   for await (const { event, data: raw } of sseEvents(body, onEvent)) {
     if (!event || !raw) continue;
     let data;
@@ -227,10 +251,21 @@ export async function readStream(body, onEvent, onToken) {
     } catch {
       continue;
     }
-    if (event === 'meta' && typeof data.model === 'string') model = data.model;
-    else if (event === 'token' && typeof data.text === 'string') {
+    if (event === 'meta') {
+      if (typeof data.model === 'string') model = data.model;
+      onStatus({
+        stage: 'started',
+        node: data.provider_name ?? data.provider_node_id ?? null,
+        engine: data.engine ?? null,
+        model: model ?? null,
+      });
+    } else if (event === 'token' && typeof data.text === 'string') {
       text += data.text;
-      if (data.text) onToken?.(data.text);
+      if (data.text) {
+        tokens += 1;
+        onToken?.(data.text);
+        onStatus({ stage: 'writing', tokens });
+      }
     } 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));

test/stream.test.js

@@ -83,6 +83,38 @@ test('C0mputeProvider.stream emits each token and the tail a done event adds', a
   assert.equal(out.model, 'qwen');
 });
 
+test('C0mputeProvider.stream reports queued, started and writing through onStatus', async () => {
+  const fetch = async (url) =>
+    url.endsWith('/api/chat')
+      ? url.includes('nope')
+        ? Response.json({}, { status: 500 })
+        : Response.json({ jobId: 'j9', source: 'p2p', model: 'qwen', provider: { name: 'node-a', nodeId: 'n-a' } })
+      : named([
+          ['job', { status: 'assigned' }],
+          ['meta', { model: 'qwen', provider_name: 'node-a', engine: 'ollama' }],
+          ['token', { text: 'a' }],
+          ['token', { text: 'b' }],
+          ['done', { text: 'ab' }],
+        ]);
+  const seen = [];
+  await new C0mputeProvider({ fetch }).stream(
+    {
+      user: 'hi',
+      onStatus: (s) => {
+        seen.push(s);
+        if (s.stage === 'writing') throw new Error('listener bug must not fail the job');
+      },
+    },
+    () => {},
+  );
+  assert.deepEqual(seen, [
+    { stage: 'queued', jobId: 'j9', node: 'node-a', source: 'p2p', model: 'qwen' },
+    { stage: 'started', node: 'node-a', engine: 'ollama', model: 'qwen' },
+    { stage: 'writing', tokens: 1 },
+    { stage: 'writing', tokens: 2 },
+  ]);
+});
+
 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) => {