diff --git a/README.md b/README.md index 8f88f3d..9daddf0 100644 --- a/README.md +++ b/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 diff --git a/index.d.ts b/index.d.ts index ed4498c..08bbabf 100644 --- a/index.d.ts +++ b/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; diff --git a/package.json b/package.json index ceb9703..e2c7880 100644 --- a/package.json +++ b/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": [ diff --git a/src/c0mpute.js b/src/c0mpute.js index d976844..73bc543 100644 --- a/src/c0mpute.js +++ b/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)); diff --git a/test/stream.test.js b/test/stream.test.js index 99c5a8a..7859e6b 100644 --- a/test/stream.test.js +++ b/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) => {