openvibers/openvibe.actor Commits
The answer streams in as the model writes it (L55 C2) (#265)
Changes
@@@ -43,7 +43,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa|---|---|---|| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued; with `webhooks`, the answer carries `webhook_secret` once) || `GET /api/v1/tasks/:id` | `actor.task.read` | the task |--| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID` |++| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID`; `?deltas=1` adds `delta` events (`{ text, n }`: the next piece of the text the model is writing in its n-th call; no id, not stored, not replayed) || `GET /api/v1/tasks/:id/trace` | `actor.task.read` | the task's spans: every model, tool, check, judge and sub-agent call with its time and cost, and the totals || `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it || `GET /api/v1/tasks` | `actor.task.list` | `actor.task-list-result@1` |@@@ -192,6 +192,85 @@: steps.length ? steps[steps.length - 1].textContent : '';if (txt) line.textContent = txt.length > 90 ? txt.slice(0, 89) + '…' : txt;}++ // The answer as it is written (plan L55 C2): the stream's delta events ({ text, n }) build a draft under the work++ // fold, drawn from the text with a few Markdown marks (paragraphs, lists, headings, bold, code) as DOM nodes,++ // never as HTML. A new model call (n) starts a new draft; a draft the run then keeps as a note in the work list,++ // or that ended in tool calls, goes away. The finished turn replaces it with the stored, rendered answer.++ function drawInline(el, text) {++ String(text).split(/(\*\*[^*\n]+\*\*|`[^`\n]+`|\[E\d{1,3}\])/).forEach(function (part) {++ if (!part) return;++ var node;++ if (/^\*\*[^*]+\*\*$/.test(part)) { node = document.createElement('b'); node.textContent = part.slice(2, -2); }++ else if (/^\[E\d{1,3}\]$/.test(part)) { node = document.createElement('span'); node.className = 'ev'; node.textContent = part.slice(2, -1); }++ else if (/^`[^`]+`$/.test(part)) { node = document.createElement('code'); node.textContent = part.slice(1, -1); }++ else node = document.createTextNode(part);++ el.appendChild(node);++ });++ }++ function drawDraft(box, text) {++ box.textContent = '';++ var para = null, list = null, fence = null;++ String(text).split('\n').forEach(function (line) {++ if (/^\s*```/.test(line)) {++ if (fence) { fence = null; return; }++ var pre = document.createElement('pre'); fence = document.createElement('code'); pre.appendChild(fence); box.appendChild(pre);++ para = list = null;++ return;++ }++ if (fence) { fence.textContent += (fence.textContent ? '\n' : '') + line; return; }++ if (!line.trim()) { para = list = null; return; }++ var item = /^\s*(?:[-*•]|\d+[.)])\s+(.*)$/.exec(line);++ if (item) {++ var kind = /^\s*\d/.test(line) ? 'OL' : 'UL';++ if (!list || list.tagName !== kind) { list = document.createElement(kind.toLowerCase()); box.appendChild(list); }++ var li = document.createElement('li'); drawInline(li, item[1]); list.appendChild(li);++ para = null;++ return;++ }++ var head = /^\s*#{1,6}\s+(.*)$/.exec(line);++ if (head) {++ var h = document.createElement('p'); var b = document.createElement('b'); b.textContent = head[1].replace(/\*\*/g, ''); h.appendChild(b); box.appendChild(h);++ para = list = null;++ return;++ }++ list = null;++ if (!para) { para = document.createElement('p'); box.appendChild(para); } else para.appendChild(document.createTextNode(' '));++ drawInline(para, line);++ });++ }++ function draft(turn) {++ var d = turn._draft;++ var box = turn.querySelector('[data-ovt-draft]');++ if (!d || !d.text.trim()) { if (box) box.remove(); return; }++ if (!box) {++ var reply = turn.querySelector('.ovt-reply');++ if (!reply) return;++ box = document.createElement('div');++ box.className = 'ovt-answer ct-draft';++ box.setAttribute('data-ovt-draft', '');++ box.setAttribute('aria-busy', 'true');++ reply.appendChild(box);++ }++ var stay = nearBottom();++ drawDraft(box, d.text);++ if (stay) toBottom(false);++ }++ function addDelta(turn, e) {++ if (!e || typeof e.text !== 'string') return;++ var d = turn._draft;++ if (!d || d.n !== e.n) d = turn._draft = { n: e.n, text: '' };++ d.text += e.text;++ if (turn._drawing) return;++ turn._drawing = true;++ (window.requestAnimationFrame || function (fn) { return setTimeout(fn, 16); })(function () { turn._drawing = false; draft(turn); });++ }++ // A stored step that holds the draft's own text (the run kept it as a note), or a tool call: the draft is done.++ function settleDraft(turn, e) {++ var d = turn._draft;++ if (!d || !e) return;++ var said = typeof e.text === 'string' ? e.text.replace(/…$/, '').trim() : '';++ if (e.step === 'tool_call' || (e.step === 'text' && said && d.text.trim().indexOf(said) === 0)) { turn._draft = null; draft(turn); }++ }function refresh(turn, done) {var id = turn.getAttribute('data-task');return fetch('/chat/turn/' + encodeURIComponent(id), { credentials: 'same-origin' })@@@ -216,8 +295,13 @@var scr = turn.querySelector('[data-screen]'), scr2 = next.querySelector('[data-screen]');if (scr && scr2 && (scr.hidden !== scr2.hidden || scr.getAttribute('data-desk') !== scr2.getAttribute('data-desk'))) { scr.replaceWith(scr2); screen(turn); }liveLine(turn);++ draft(turn);} else {stopScreen(turn);++ // An answer the person already watched being written is not typed in again: it only takes its++ // finished form (OVThread types in a turn that was open and is not any more).++ if (turn.querySelector('[data-ovt-draft]')) turn.removeAttribute('data-open');++ turn._draft = null;if (window.OVThread) window.OVThread.replace(id, h); else turn.replaceWith(next);var dot = chat.querySelector('.ovt-list a[aria-current] .ovt-dot'); if (dot) dot.remove();if (note) note.textContent = '';@@@ -303,7 +387,7 @@screen(turn);liveLine(turn);var id = turn.getAttribute('data-task');-- var src = new EventSource('/api/v1/tasks/' + encodeURIComponent(id) + '/events?after=' + encodeURIComponent(turn.getAttribute('data-last') || '0'));++ var src = new EventSource('/api/v1/tasks/' + encodeURIComponent(id) + '/events?deltas=1&after=' + encodeURIComponent(turn.getAttribute('data-last') || '0'));streams.push(src);var timer = 0;function soon() { if (timer) return; timer = setTimeout(function () { timer = 0; refresh(turn); }, 450); }@@@ -312,9 +396,13 @@var e = JSON.parse(m.data);var line = turn.querySelector('[data-ovt-live]');if (line && e.step === 'tool_call' && e.tool) line.textContent = 'Using ' + String(e.tool).replace(/_/g, ' ') + '…';++ settleDraft(turn, e);} catch (err) { /* a malformed event is skipped */ }soon();});++ src.addEventListener('delta', function (m) {++ try { addDelta(turn, JSON.parse(m.data)); } catch (err) { /* a malformed piece is skipped */ }++ });src.addEventListener('state', soon);src.addEventListener('end', function () { src.close(); clearTimeout(timer); timer = 0; refresh(turn); });src.onerror = function () { if (src.readyState === 2) setTimeout(function () { refresh(turn); }, 3000); };@@@ -4,8 +4,9 @@* The model providers' HTTP APIs, metered per call. Two shapes:** chatClient(cfg) OpenAI-compatible /chat/completions with tools (DeepSeek, an open model on llama.cpp, OpenAI-- * for checks): chat({ messages, tools, json, maxTokens, signal }) → { message, text, toolCalls,-- * usage, cost_usd }++ * for checks): chat({ messages, tools, json, maxTokens, signal, onDelta }) → { message, text,++ * toolCalls, usage, cost_usd }; with onDelta the answer streams and each piece of its text is++ * passed to onDelta as it arrives (the result is the same)* responsesAgent(cfg) OpenAI's Responses API with its hosted web_search tool: run({ instructions, input, effort,* signal }) → { text, sources, searches, usage, cost_usd }*@@@ -20,27 +21,110 @@ class ProviderError extends Error {const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));--async function post(fetchImpl, url, apiKey, body, { signal, timeoutMs }) {++async function send(fetchImpl, url, apiKey, body, { signal, timeoutMs, accept = null }) {const timeout = AbortSignal.timeout(timeoutMs);const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;-- let res;try {-- res = await fetchImpl(url, { method: 'POST', headers: { 'content-type': 'application/json', ...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}) }, body: JSON.stringify(body), signal: sig });++ return await fetchImpl(url, { method: 'POST', headers: { 'content-type': 'application/json', ...(accept ? { accept } : {}), ...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}) }, body: JSON.stringify(body), signal: sig });} catch (err) {-- if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });-- throw new ProviderError(err && err.name === 'TimeoutError' ? 'the model did not answer in time' : 'the model provider could not be reached', { code: 'actor.provider.unreachable' });++ throw unreachable(err, signal, 'the model provider');}++}++++/** A failed or aborted request as a ProviderError: cancelled by the run, timed out, or not reachable. */++function unreachable(err, signal, who) {++ if (signal && signal.aborted) return new ProviderError('cancelled', { code: 'actor.task.cancelled' });++ return new ProviderError(err && err.name === 'TimeoutError' ? `${who === 'the model provider' ? 'the model' : who} did not answer in time` : `${who} could not be reached`, { code: 'actor.provider.unreachable' });++}++++function httpError(res, text, json) {++ const msg = (json && json.error && (json.error.message || json.error)) || clip(text, 200) || `HTTP ${res.status}`;++ return new ProviderError(`HTTP ${res.status}: ${clip(msg, 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });++}++++async function post(fetchImpl, url, apiKey, body, opts) {++ const res = await send(fetchImpl, url, apiKey, body, opts);const text = await res.text();let json = null;try { json = JSON.parse(text); } catch { /* not JSON */ }-- if (!res.ok) {-- const msg = (json && json.error && (json.error.message || json.error)) || clip(text, 200) || `HTTP ${res.status}`;-- throw new ProviderError(`HTTP ${res.status}: ${clip(msg, 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });-- }++ if (!res.ok) throw httpError(res, text, json);if (!json) throw new ProviderError('the model provider answered with something that is not JSON');return json;}++/**++ * A Chat Completions answer streamed as server-sent events (stream: true), put back together in the shape of the++ * whole answer: { choices: [{ message, finish_reason }], model, usage, x_openvibe }. onDelta(text) is called with each++ * piece of the answer's text as it arrives (never with reasoning or tool calls); a throw in it is ignored. Tool calls++ * are joined by their index. An upstream that answers with plain JSON (it cannot stream) is read whole, and its text++ * is passed to onDelta once. A failure the stream itself carries ({ error }) is a ProviderError.++ */++async function readStream(res, onDelta, { signal, who = 'the model provider' } = {}) {++ const tell = (t) => { if (t && onDelta) { try { onDelta(t); } catch { /* the listener's problem, not the call's */ } } };++ const type = res.headers && typeof res.headers.get === 'function' ? res.headers.get('content-type') : '';++ if (!/text\/event-stream/i.test(type || '') || !res.body || typeof res.body[Symbol.asyncIterator] !== 'function') {++ const text = await res.text();++ let json = null;++ try { json = JSON.parse(text); } catch { /* not JSON */ }++ if (!json) throw new ProviderError(`${who} answered with something that is not JSON`);++ const m = (json.choices && json.choices[0] && json.choices[0].message) || {};++ tell(typeof m.content === 'string' ? m.content : '');++ return json;++ }++ let content = '', reasoning = '', finish = null, model = null, usage = null, x = null, failure = null;++ const tools = [];++ const take = (data) => {++ if (!data || data === '[DONE]') return;++ let j;++ try { j = JSON.parse(data); } catch { return; }++ if (j.error) { failure = j.error; return; }++ if (j.model) model = j.model;++ if (j.usage) usage = j.usage;++ if (j.x_openvibe) x = j.x_openvibe;++ const c = Array.isArray(j.choices) ? j.choices.find((ch) => (ch.index || 0) === 0) : null;++ if (!c) return;++ const d = c.delta || {};++ if (typeof d.content === 'string' && d.content) { content += d.content; tell(d.content); }++ if (typeof d.reasoning_content === 'string') reasoning += d.reasoning_content;++ for (const tc of Array.isArray(d.tool_calls) ? d.tool_calls : []) {++ const i = Number.isInteger(tc.index) ? tc.index : tools.length;++ const t = tools[i] || (tools[i] = { id: '', type: 'function', function: { name: '', arguments: '' } });++ if (tc.id) t.id = tc.id;++ if (tc.type) t.type = tc.type;++ const fn = tc.function || {};++ // Most providers send the name once; one that repeats it whole on every piece is not doubled.++ if (fn.name) t.function.name = t.function.name && !fn.name.startsWith(t.function.name) ? t.function.name + fn.name : fn.name;++ if (typeof fn.arguments === 'string') t.function.arguments += fn.arguments;++ }++ if (c.finish_reason) finish = c.finish_reason;++ };++ const decoder = new TextDecoder();++ let buf = '';++ const lines = () => {++ let i;++ while ((i = buf.indexOf('\n')) >= 0) {++ const line = buf.slice(0, i).replace(/\r$/, '');++ buf = buf.slice(i + 1);++ if (line.startsWith('data:')) take(line.slice(5).trim());++ }++ };++ try {++ for await (const chunk of res.body) { buf += decoder.decode(chunk, { stream: true }); lines(); }++ } catch (err) {++ throw unreachable(err, signal, who);++ }++ buf += decoder.decode() + '\n';++ lines();++ if (failure) {++ const msg = typeof failure === 'string' ? failure : (failure.message || failure.code || 'error');++ const code = typeof failure === 'object' ? String(failure.code || '') : '';++ throw new ProviderError(clip(msg, 300), { code: /quota|rate|allowance/.test(code) ? 'actor.provider.rate_limited' : 'actor.provider.error' });++ }++ const calls = tools.filter(Boolean);++ const message = { role: 'assistant', content: content || (calls.length ? null : ''), ...(calls.length ? { tool_calls: calls } : {}), ...(reasoning ? { reasoning_content: reasoning } : {}) };++ return { choices: [{ index: 0, message, finish_reason: finish }], model, usage, ...(x ? { x_openvibe: x } : {}) };++}++/** USD for a usage at a rate card: { input, cached, output } tokens, calls. */function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {const r = rates || {};@@@ -54,13 +138,21 @@ function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {*/function chatClient({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.fetch, timeoutMs = 90_000, extraBody = null }) {const reasoning = /^(gpt-5|o\d)/i.test(model || '');-- return async function chat({ messages, tools = null, json = false, maxTokens = 1500, signal } = {}) {++ return async function chat({ messages, tools = null, json = false, maxTokens = 1500, signal, onDelta = null } = {}) {const body = { model, messages, ...(extraBody || {}) };// "low" is the one effort every reasoning chat model we call accepts (`minimal` is not offered by gpt-5.1).if (reasoning) { body.max_completion_tokens = maxTokens + 1024; body.reasoning_effort = 'low'; } else { body.max_tokens = maxTokens; body.temperature = 0.2; }if (tools && tools.length) { body.tools = tools; body.tool_choice = 'auto'; }if (json) body.response_format = { type: 'json_object' };-- const j = await post(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs });++ let j;++ if (onDelta) {++ // The answer streams in (plan L55 C2): each piece of its text goes to onDelta as the model writes it.++ body.stream = true;++ body.stream_options = { include_usage: true };++ const res = await send(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs, accept: 'text/event-stream' });++ if (!res.ok) { const text = await res.text(); let e = null; try { e = JSON.parse(text); } catch { /* not JSON */ } throw httpError(res, text, e); }++ j = await readStream(res, onDelta, { signal });++ } else j = await post(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs });const choice = (j.choices && j.choices[0]) || {};const message = choice.message || { role: 'assistant', content: '' };const u = j.usage || {};@@@ -118,4 +210,4 @@ function responsesAgent({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.};}--module.exports = { chatClient, responsesAgent, priceUsage, sourcesOf, citationsOf, ProviderError };++module.exports = { chatClient, responsesAgent, priceUsage, sourcesOf, citationsOf, readStream, ProviderError };@@@ -306,6 +306,8 @@ function createApi(ctx) {const after = Math.max(0, parseInt(req.get('last-event-id') || req.query.after, 10) || 0);await stream.serve(req, res, {taskId: row.id, after,++ // The answer as it is written (plan L55 C2), for a reader that asks: transient delta events.++ deltas: req.query.deltas === '1',readBacklog: () => store.eventsAfter(s, row.id, after),isEnded: async () => { const cur = await store.getTask(s, row.id); return !cur || !store.OPEN.includes(cur.state); },});@@@ -288,6 +288,11 @@ function createLoop({ config, models, registry, log = console, now = () => Date.if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;const defs = forceNoTools ? null : shortlistTools(tools, state);const startedAt = now();++ // The pieces of this call's text go to the page as they arrive (plan L55 C2), numbered by call so the page can++ // start a fresh draft when a call that ended in tool calls is followed by the next one.++ const n = (state.deltaCalls = (state.deltaCalls || 0) + 1);++ // Only while someone reads the answer as it is written: a task nobody watches calls the model as before.++ const onDelta = ctx.delta && (typeof ctx.delta.wanted !== 'function' || ctx.delta.wanted()) ? (text) => ctx.delta({ text, n }) : null;let r;try {r = await models.chat({@@@ -296,6 +301,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.tools: defs && defs.length ? defs : null,maxTokens: MAX_ANSWER_TOKENS,signal: state.signal,++ ...(onDelta ? { onDelta } : {}),});} catch (err) {await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });@@@ -840,7 +846,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.* a span; and drives on from the step it stopped at. A checkpoint whose last step already answered is delivered* without another model call.*/-- async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null }) {++ async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, delta = null }) {const source = resume && resume.state ? resume.state : (fork && fork.checkpoint && fork.checkpoint.state);const restored = restoreState(source);if (!restored) throw Object.assign(new Error('the saved step could not be read'), { code: 'actor.checkpoint.invalid' });@@@ -866,6 +872,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.trace: createTracer({ store, taskId, log, now }),readOutside: () => state.readOutside === true, notifyTask,emit: async (e) => emit(e),++ // The answer as the model writes it (plan L55 C2): delta({ text, n }), not stored; null when nobody streams.++ delta: typeof delta === 'function' ? delta : null,// The connector tools of this run (stage V2-11) and the failures that switch a server off.dynamicTools: new Map(),connectorState: { failures: new Map(), disabled: new Set() },@@@ -996,9 +1004,9 @@ function createLoop({ config, models, registry, log = console, now = () => Date.}}-- async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [] }) {++ async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [], delta = null }) {if (resume || fork) {-- return await continueFrom({ task, requester, taskId, mode, signal, emit, meter, budgetLeft, waitForPerson, requestApproval, context, personText, agentId, notifyTask, askPerson, control, resume, fork, spent, waits, waitEvent });++ return await continueFrom({ task, requester, taskId, mode, signal, emit, meter, budgetLeft, waitForPerson, requestApproval, context, personText, agentId, notifyTask, askPerson, control, resume, fork, spent, waits, waitEvent, delta });}const tier = mode === 'best' && models.resolve('best').available ? 'best' : 'smart';if (!models.resolve(tier).available) throw Object.assign(new Error(models.resolve(tier).reason || 'no model is available for the loop'), { code: 'actor.model.unavailable' });@@@ -1027,6 +1035,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.trace: createTracer({ store, taskId, log, now }),readOutside: () => state.readOutside === true, notifyTask,emit: async (e) => emit(e),++ // The answer as the model writes it (plan L55 C2): delta({ text, n }), not stored; null when nobody streams.++ delta: typeof delta === 'function' ? delta : null,// The connector tools of this run (stage V2-11) and the failures that switch a server off.dynamicTools: new Map(),connectorState: { failures: new Map(), disabled: new Set() },@@@ -31,7 +31,7 @@const { AsyncLocalStorage } = require('node:async_hooks');const { serviceAuth } = require('openvibe-contracts');--const { chatClient, priceUsage, ProviderError } = require('../agents/providers');++const { chatClient, priceUsage, readStream, ProviderError } = require('../agents/providers');class ModelError extends Error {constructor(message, { code = 'actor.model.unavailable' } = {}) { super(message); this.code = code; }@@@ -297,7 +297,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =* (x-openvibe-credential: <provider>:<subject>, the bare usr_… id AI expects — never "user:usr_…"). AI answers* with that key, or refuses: the person has no key for that provider yet, and the message says where to add it.*/-- async function chatViaAi({ choice, messages, tools, json, maxTokens, signal }) {++ async function chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta = null }) {const tokens = aiTokenClient();if (!tokens) throw new ModelError('OpenVibe.AI is not configured on this server (no Network client secret)', { code: 'actor.model.unavailable' });const base = trimSlash(ai.url);@@@ -313,6 +313,8 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =temperature: 0.2,// L37: a wish, not a pin. The provider is the one the person's own key is for, so the wish holds.prefer: { provider, ...(choice.model ? { model: choice.model } : {}) },++ // The answer streams in (plan L55 C2): AI passes the upstream's pieces on as they arrive.++ ...(onDelta ? { stream: true, stream_options: { include_usage: true } } : {}),};const timeout = AbortSignal.timeout(ai.timeoutMs || 180_000);const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;@@@ -322,7 +324,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =method: 'POST',headers: {'content-type': 'application/json',-- accept: 'application/json',++ accept: onDelta ? 'text/event-stream' : 'application/json',...(await tokens.authHeaders()),// The person's own key for this provider, and who the person is (the bare usr_… id).'x-openvibe-credential': provider + ':' + choice.subject,@@@ -334,9 +336,12 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });throw new ProviderError(err && err.name === 'TimeoutError' ? 'OpenVibe.AI did not answer in time' : 'OpenVibe.AI could not be reached', { code: 'actor.provider.unreachable' });}-- const raw = await res.text();++ let raw = '';let answer = null;-- try { answer = JSON.parse(raw); } catch { /* not JSON */ }++ if (!res.ok || !onDelta) {++ raw = await res.text();++ try { answer = JSON.parse(raw); } catch { /* not JSON */ }++ }if (!res.ok) {const err = (answer && answer.error) || {};const code = String(err.code || '');@@@ -353,6 +358,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =}throw new ProviderError('HTTP ' + res.status + ': ' + clip(msg, 300), { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });}++ if (onDelta) answer = await readStream(res, onDelta, { signal, who: 'OpenVibe.AI' });if (!answer) throw new ProviderError('OpenVibe.AI answered with something that is not JSON');const first = (answer.choices && answer.choices[0]) || {};const message = first.message || { role: 'assistant', content: '' };@@@ -371,11 +377,11 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =};}/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */-- async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null } = {}) {++ async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null } = {}) {// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it.const choice = currentChoice();if (choice && choice.provider) {-- const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal });++ const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta });return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };}await warm();@@@ -386,7 +392,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =// The check is a short structured call, so its thinking is off; the loop keeps it on.const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });-- const r = await client({ messages, tools, json, maxTokens, signal });++ const r = await client({ messages, tools, json, maxTokens, signal, onDelta });return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };}@@@ -359,6 +359,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.context: row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [],emit: (e) => emit(id, { kind: 'output', agent: agentId, ...e }),++ // The answer as it is written (plan L55 C2): to the readers that asked, never stored.++ // The loop streams a model call only while such a reader is there (wanted).++ delta: typeof stream.delta === 'function' ? Object.assign((d) => stream.delta(id, d), { wanted: () => stream.wants(id) }) : null,meter, budgetLeft, waitForPerson, requestApproval,personText, agentId: ownAgent, notifyTask,askPerson, control, resume, fork, spent: () => spent,@@@ -9,12 +9,21 @@* mark every older seq as already sent: a reader got only the `end` event, seq 9, on PostgreSQL in CI), and an `end`* that comes meanwhile closes the stream only after the backlog.* The stream is best-effort; the task record is the truth.++ *++ * The answer as it is written (plan L55 C2): a reader that asks with ?deltas=1 also gets `delta` events,++ * { text, n }: the next piece of the text the model is writing in its n-th call of the run. They are not stored and++ * carry no id (so Last-Event-ID and the backlog are unchanged), go only to a reader whose backlog is out, and are++ * gathered for DELTA_MS so a fast model sends a few events a second, not one per token. A stored event flushes the++ * gathered text first, so a reader sees them in the order they happened. The stored `output` text event, or the++ * finished task, is still the answer.*/const HEARTBEAT_MS = 15_000;++const DELTA_MS = 60;function createTaskStream({ log = console } = {}) {const subs = new Map(); // task id → Set<{ res, last }>++ const gathered = new Map(); // task id → { text, n, timer }: delta text not yet sentfunction send(sub, event) {if (event.seq <= sub.last) return;@@@ -29,10 +38,45 @@ function createTaskStream({ log = console } = {}) {}function publish(taskId, event) {++ flush(taskId);for (const sub of subs.get(taskId) || []) write(sub, event);}++ function flush(taskId) {++ const g = gathered.get(taskId);++ if (!g) return;++ gathered.delete(taskId);++ clearTimeout(g.timer);++ if (!g.text) return;++ const line = `event: delta\ndata: ${JSON.stringify({ text: g.text, n: g.n })}\n\n`;++ for (const sub of subs.get(taskId) || []) {++ if (!sub.deltas || !sub.ready) continue;++ try { sub.res.write(line); } catch (err) { log.warn('[Actor] stream write failed:', err && err.message); }++ }++ }++++ /** The next piece of the answer being written (not stored). n: the run's model call it belongs to. */++ function delta(taskId, { text, n = 0 } = {}) {++ if (!text) return;++ if (!wants(taskId)) return;++ let g = gathered.get(taskId);++ if (g && g.n !== n) { flush(taskId); g = null; }++ if (!g) {++ g = { text: '', n, timer: setTimeout(() => flush(taskId), DELTA_MS) };++ if (g.timer.unref) g.timer.unref();++ gathered.set(taskId, g);++ }++ g.text += String(text);++ }++++ /** Whether a reader of this task asked for the answer as it is written (the loop streams only then). */++ function wants(taskId) {++ return [...(subs.get(taskId) || [])].some((sub) => sub.deltas);++ }++function end(taskId) {++ const g = gathered.get(taskId);++ if (g) { clearTimeout(g.timer); gathered.delete(taskId); }const set = subs.get(taskId);subs.delete(taskId);for (const sub of set || []) {@@@ -45,11 +89,11 @@ function createTaskStream({ log = console } = {}) {* Serve one reader. readBacklog() → stored events after `after`; isEnded() → whether the task has ended (read after* the backlog, so an end that happened meanwhile closes the stream).*/-- async function serve(req, res, { taskId, after = 0, readBacklog, isEnded }) {++ async function serve(req, res, { taskId, after = 0, readBacklog, isEnded, deltas = false }) {res.status(200).set({ 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no' });res.flushHeaders();res.write(': stream open\n\n');-- const sub = { res, last: after, ready: false, pending: [], ended: false };++ const sub = { res, last: after, ready: false, pending: [], ended: false, deltas: deltas === true };let set = subs.get(taskId);if (!set) { set = new Set(); subs.set(taskId, set); }set.add(sub);@@@ -70,7 +114,7 @@ function createTaskStream({ log = console } = {}) {const status = () => ({ streams: [...subs.values()].reduce((n, s) => n + s.size, 0) });-- return { publish, end, serve, status };++ return { publish, delta, wants, end, serve, status };}module.exports = { createTaskStream };nnew file mode 100644@@@ -0,0 +1,159 @@++'use strict';++/**++ * The answer as it is written (plan L55 C2): a model call with onDelta streams (stream: true), passes each piece of++ * its text on as it arrives, and returns what the whole answer would have been; the task stream sends the pieces to the++ * readers that asked (?deltas=1) as `delta` events with no id, gathered, never stored, flushed before a stored event.++ *++ * node test/answer-stream.test.js++ */++const assert = require('assert');++const { EventEmitter } = require('events');++const { chatClient, readStream, ProviderError } = require('../server/agents/providers');++const { createTaskStream } = require('../server/tasks/stream');++++const sse = (chunks) => chunks.map((c) => `data: ${typeof c === 'string' ? c : JSON.stringify(c)}\n\n`).join('');++function streamed(text, { split = 7, status = 200 } = {}) {++ // Sent in small byte pieces, so a line is often cut in two: the reader must put lines back together.++ const bytes = new TextEncoder().encode(text);++ const body = new ReadableStream({++ start(ctl) { for (let i = 0; i < bytes.length; i += split) ctl.enqueue(bytes.slice(i, i + split)); ctl.close(); },++ });++ return new Response(body, { status, headers: { 'content-type': 'text/event-stream; charset=utf-8' } });++}++const chunk = (delta, extra = {}) => ({ id: 'c1', object: 'chat.completion.chunk', model: 'deepseek-chat', choices: [{ index: 0, delta, finish_reason: null }], ...extra });++++(async () => {++ // 1. Text, reasoning, tool calls joined by index, finish, usage and AI's x_openvibe, from pieces cut mid-line.++ {++ const pieces = [];++ const body = sse([++ chunk({ role: 'assistant', content: '' }),++ chunk({ reasoning_content: 'Think ' }),++ chunk({ reasoning_content: 'first.' }),++ chunk({ content: 'Hello' }),++ chunk({ content: ', **world**' }),++ chunk({ tool_calls: [{ index: 0, id: 'call_a', type: 'function', function: { name: 'web_', arguments: '{"q":' } }] }),++ chunk({ tool_calls: [{ index: 0, function: { name: 'search', arguments: '"cats"}' } }] }),++ chunk({ tool_calls: [{ index: 1, id: 'call_b', type: 'function', function: { name: 'read_page', arguments: '{}' } }] }),++ { id: 'c1', choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }] },++ { id: 'c1', choices: [], usage: { prompt_tokens: 120, completion_tokens: 30, prompt_cache_hit_tokens: 100 } },++ { id: 'c1', choices: [], x_openvibe: { cost_usd: 0.0012 } },++ '[DONE]',++ ]);++ const j = await readStream(streamed(body), (t) => pieces.push(t));++ assert.deepStrictEqual(pieces, ['Hello', ', **world**'], 'only the answer text goes to onDelta, piece by piece');++ const m = j.choices[0].message;++ assert.strictEqual(m.content, 'Hello, **world**');++ assert.strictEqual(m.reasoning_content, 'Think first.', 'the reasoning is kept on the message (DeepSeek wants it back)');++ assert.deepStrictEqual(m.tool_calls, [++ { id: 'call_a', type: 'function', function: { name: 'web_search', arguments: '{"q":"cats"}' } },++ { id: 'call_b', type: 'function', function: { name: 'read_page', arguments: '{}' } },++ ]);++ assert.strictEqual(j.choices[0].finish_reason, 'tool_calls');++ assert.strictEqual(j.model, 'deepseek-chat');++ assert.strictEqual(j.usage.prompt_tokens, 120);++ assert.strictEqual(j.x_openvibe.cost_usd, 0.0012);++ }++++ // 2. A provider that repeats the whole tool name on every piece is not doubled; an empty answer with tool calls++ // has content null, as the whole answer would.++ {++ const body = sse([++ chunk({ tool_calls: [{ index: 0, id: 'x', function: { name: 'fetch_url', arguments: '{"u"' } }] }),++ chunk({ tool_calls: [{ index: 0, function: { name: 'fetch_url', arguments: ':1}' } }] }),++ '[DONE]',++ ]);++ const j = await readStream(streamed(body), () => {});++ assert.strictEqual(j.choices[0].message.tool_calls[0].function.name, 'fetch_url');++ assert.strictEqual(j.choices[0].message.tool_calls[0].function.arguments, '{"u":1}');++ assert.strictEqual(j.choices[0].message.content, null);++ }++++ // 3. A failure the stream carries is a ProviderError; a listener that throws does not break the call.++ {++ const body = sse([chunk({ content: 'Par' }), { error: { message: 'upstream went away', code: 'provider.error' } }, '[DONE]']);++ await assert.rejects(readStream(streamed(body), () => { throw new Error('listener'); }), (err) => err instanceof ProviderError && /upstream went away/.test(err.message));++ }++++ // 4. An upstream that answers JSON (it cannot stream) is read whole, and its text reaches onDelta once.++ {++ const pieces = [];++ const res = new Response(JSON.stringify({ model: 'm', choices: [{ index: 0, message: { role: 'assistant', content: 'Whole.' }, finish_reason: 'stop' }], usage: { prompt_tokens: 1, completion_tokens: 1 } }), { headers: { 'content-type': 'application/json' } });++ const j = await readStream(res, (t) => pieces.push(t));++ assert.strictEqual(j.choices[0].message.content, 'Whole.');++ assert.deepStrictEqual(pieces, ['Whole.']);++ }++++ // 5. chatClient: with onDelta the request streams and asks for usage; the result has the usual shape and price.++ // Without it, nothing changes: one JSON answer.++ {++ const sent = [];++ const fetchImpl = async (url, init) => {++ const body = JSON.parse(init.body);++ sent.push({ url, body, accept: init.headers.accept || null });++ if (body.stream) return streamed(sse([chunk({ content: 'Two ' }), chunk({ content: 'words.' }), { choices: [{ index: 0, delta: {}, finish_reason: 'stop' }] }, { choices: [], usage: { prompt_tokens: 1000000, completion_tokens: 1000000, prompt_cache_hit_tokens: 0 } }, '[DONE]']));++ return new Response(JSON.stringify({ model: 'deepseek-chat', choices: [{ index: 0, message: { role: 'assistant', content: 'Two words.' }, finish_reason: 'stop' }], usage: { prompt_tokens: 1000000, completion_tokens: 1000000 } }), { headers: { 'content-type': 'application/json' } });++ };++ const chat = chatClient({ baseUrl: 'https://api.example', apiKey: 'k', model: 'deepseek-chat', rates: { input_usd_per_mtok: 0.25, output_usd_per_mtok: 1 }, fetchImpl });++ const pieces = [];++ const a = await chat({ messages: [{ role: 'user', content: 'hi' }], onDelta: (t) => pieces.push(t) });++ const b = await chat({ messages: [{ role: 'user', content: 'hi' }] });++ assert.strictEqual(sent[0].body.stream, true);++ assert.deepStrictEqual(sent[0].body.stream_options, { include_usage: true });++ assert.strictEqual(sent[0].accept, 'text/event-stream');++ assert.ok(!('stream' in sent[1].body), 'no onDelta: no stream');++ assert.deepStrictEqual(pieces, ['Two ', 'words.']);++ for (const r of [a, b]) {++ assert.strictEqual(r.text, 'Two words.');++ assert.deepStrictEqual(r.toolCalls, []);++ assert.strictEqual(r.finish, 'stop');++ assert.strictEqual(r.cost_usd, 1.25, 'priced the same way streamed or not');++ }++ }++++ // 6. A refused streamed call is the same ProviderError as a refused whole one.++ {++ const fetchImpl = async () => new Response(JSON.stringify({ error: { message: 'Too many requests' } }), { status: 429, headers: { 'content-type': 'application/json' } });++ const chat = chatClient({ baseUrl: 'https://api.example', apiKey: 'k', model: 'm', rates: {}, fetchImpl });++ await assert.rejects(chat({ messages: [], onDelta: () => {} }), (err) => err.code === 'actor.provider.rate_limited' && err.status === 429);++ }++++ // 7. The task stream: deltas go only to a reader that asked, gathered, with no id line; a stored event flushes the++ // gathered text first; a new call number flushes the old call's text; the end drops what is left.++ {++ const stream = createTaskStream({ log: { warn() {} } });++ const fakeRes = () => ({ chunks: [], ended: false, status() { return this; }, set() { return this; }, flushHeaders() {}, write(c) { this.chunks.push(c); }, end() { this.ended = true; } });++ const asked = fakeRes(), plain = fakeRes();++ await stream.serve(new EventEmitter(), asked, { taskId: 'tsk_d', after: 0, deltas: true, readBacklog: async () => [], isEnded: async () => false });++ await stream.serve(new EventEmitter(), plain, { taskId: 'tsk_d', after: 0, readBacklog: async () => [], isEnded: async () => false });++ assert.strictEqual(stream.wants('tsk_d'), true, 'a reader asked: the loop streams');++ assert.strictEqual(stream.wants('tsk_x'), false, 'nobody asked: the loop calls the model as before');++ stream.delta('tsk_d', { text: 'Hel', n: 1 });++ stream.delta('tsk_d', { text: 'lo', n: 1 });++ stream.delta('tsk_d', { text: 'Next', n: 2 }); // a new call: call 1's text goes out first++ stream.publish('tsk_d', { task_id: 'tsk_d', seq: 1, kind: 'output', step: 'text', text: 'Hello' });++ const events = (res) => res.chunks.join('').split('\n\n').filter((b) => /^(id|event):/m.test(b));++ const got = events(asked);++ assert.deepStrictEqual(got, [++ 'event: delta\ndata: {"text":"Hello","n":1}',++ 'event: delta\ndata: {"text":"Next","n":2}',++ 'id: 1\nevent: output\ndata: {"task_id":"tsk_d","seq":1,"kind":"output","step":"text","text":"Hello"}',++ ]);++ assert.deepStrictEqual(events(plain), [got[2]], 'a reader that did not ask gets only the stored events');++ stream.delta('tsk_d', { text: 'left over', n: 3 });++ stream.end('tsk_d');++ await new Promise((r) => setTimeout(r, 100));++ assert.ok(!asked.chunks.join('').includes('left over'), 'nothing is sent after the end');++ // The timer sends gathered text on its own when no stored event comes.++ const late = fakeRes();++ await stream.serve(new EventEmitter(), late, { taskId: 'tsk_e', after: 0, deltas: true, readBacklog: async () => [], isEnded: async () => false });++ stream.delta('tsk_e', { text: 'Soon', n: 1 });++ await new Promise((r) => setTimeout(r, 120));++ assert.deepStrictEqual(events(late), ['event: delta\ndata: {"text":"Soon","n":1}']);++ // Nobody asked: nothing is gathered.++ stream.delta('tsk_none', { text: 'x', n: 1 });++ stream.end('tsk_e');++ }++++ console.log('answer stream: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });@@@ -146,7 +146,12 @@ const LOC = /^\/chat\/(tsk_[0-9A-HJKMNP-TV-Z]{26})#t-\1$/;await check('in the browser (public/js/chat.js): the live stream, the turn fragment, Stop, Enter on a keyboard only', async () => {const js = require('fs').readFileSync(require('path').join(__dirname, '..', 'public', 'js', 'chat.js'), 'utf8');-- assert.match(js, /new EventSource\('\/api\/v1\/tasks\/' \+ encodeURIComponent\(id\) \+ '\/events\?after='/, 'follows the task\'s own event stream');++ assert.match(js, /new EventSource\('\/api\/v1\/tasks\/' \+ encodeURIComponent\(id\) \+ '\/events\?deltas=1&after='/, 'follows the task\'s own event stream, with the answer as it is written');++ // The answer as it is written (plan L55 C2): delta events build a draft from DOM nodes, never from HTML.++ assert.match(js, /src\.addEventListener\('delta'/, 'the draft follows the delta events');++ const drawing = js.slice(js.indexOf('function drawInline'), js.indexOf('function refresh('));++ assert.ok(drawing.length > 200 && !/innerHTML/.test(drawing), 'the draft is drawn with text nodes only');++ assert.match(js, /turn\.querySelector\('\[data-ovt-draft\]'\)\) turn\.removeAttribute\('data-open'\)/, 'an answer already watched is not typed in again');assert.match(js, /fetch\('\/chat\/turn\/' \+ encodeURIComponent\(id\)/, 'the server draws each turn; the page swaps it in');assert.match(js, /'\/tasks\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/cancel'/, 'Stop cancels the working turn');assert.match(js, /'\/chat\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/steer'/, 'what you write while it works steers it');
@@ -43,7 +43,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa
@@ -43,7 +43,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa
|---|---|---|
|---|---|---|
| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued; with `webhooks`, the answer carries `webhook_secret` once) |
| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued; with `webhooks`, the answer carries `webhook_secret` once) |
| `GET /api/v1/tasks/:id` | `actor.task.read` | the task |
| `GET /api/v1/tasks/:id` | `actor.task.read` | the task |
-| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID` |
+| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID`; `?deltas=1` adds `delta` events (`{ text, n }`: the next piece of the text the model is writing in its n-th call; no id, not stored, not replayed) |
| `GET /api/v1/tasks/:id/trace` | `actor.task.read` | the task's spans: every model, tool, check, judge and sub-agent call with its time and cost, and the totals |
| `GET /api/v1/tasks/:id/trace` | `actor.task.read` | the task's spans: every model, tool, check, judge and sub-agent call with its time and cost, and the totals |
| `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it |
| `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it |
| `GET /api/v1/tasks` | `actor.task.list` | `actor.task-list-result@1` |
| `GET /api/v1/tasks` | `actor.task.list` | `actor.task-list-result@1` |
@@ -192,6 +192,85 @@
@@ -192,6 +192,85 @@
: steps.length ? steps[steps.length - 1].textContent : '';
: steps.length ? steps[steps.length - 1].textContent : '';
if (txt) line.textContent = txt.length > 90 ? txt.slice(0, 89) + '…' : txt;
if (txt) line.textContent = txt.length > 90 ? txt.slice(0, 89) + '…' : txt;
}
}
+ // The answer as it is written (plan L55 C2): the stream's delta events ({ text, n }) build a draft under the work
+ // fold, drawn from the text with a few Markdown marks (paragraphs, lists, headings, bold, code) as DOM nodes,
+ // never as HTML. A new model call (n) starts a new draft; a draft the run then keeps as a note in the work list,
+ // or that ended in tool calls, goes away. The finished turn replaces it with the stored, rendered answer.
+ function drawInline(el, text) {
+ String(text).split(/(\*\*[^*\n]+\*\*|`[^`\n]+`|\[E\d{1,3}\])/).forEach(function (part) {
+ if (!part) return;
+ var node;
+ if (/^\*\*[^*]+\*\*$/.test(part)) { node = document.createElement('b'); node.textContent = part.slice(2, -2); }
+ else if (/^\[E\d{1,3}\]$/.test(part)) { node = document.createElement('span'); node.className = 'ev'; node.textContent = part.slice(2, -1); }
+ else if (/^`[^`]+`$/.test(part)) { node = document.createElement('code'); node.textContent = part.slice(1, -1); }
+ else node = document.createTextNode(part);
+ el.appendChild(node);
+ });
+ }
+ function drawDraft(box, text) {
+ box.textContent = '';
+ var para = null, list = null, fence = null;
+ String(text).split('\n').forEach(function (line) {
+ if (/^\s*```/.test(line)) {
+ if (fence) { fence = null; return; }
+ var pre = document.createElement('pre'); fence = document.createElement('code'); pre.appendChild(fence); box.appendChild(pre);
+ para = list = null;
+ return;
+ }
+ if (fence) { fence.textContent += (fence.textContent ? '\n' : '') + line; return; }
+ if (!line.trim()) { para = list = null; return; }
+ var item = /^\s*(?:[-*•]|\d+[.)])\s+(.*)$/.exec(line);
+ if (item) {
+ var kind = /^\s*\d/.test(line) ? 'OL' : 'UL';
+ if (!list || list.tagName !== kind) { list = document.createElement(kind.toLowerCase()); box.appendChild(list); }
+ var li = document.createElement('li'); drawInline(li, item[1]); list.appendChild(li);
+ para = null;
+ return;
+ }
+ var head = /^\s*#{1,6}\s+(.*)$/.exec(line);
+ if (head) {
+ var h = document.createElement('p'); var b = document.createElement('b'); b.textContent = head[1].replace(/\*\*/g, ''); h.appendChild(b); box.appendChild(h);
+ para = list = null;
+ return;
+ }
+ list = null;
+ if (!para) { para = document.createElement('p'); box.appendChild(para); } else para.appendChild(document.createTextNode(' '));
+ drawInline(para, line);
+ });
+ }
+ function draft(turn) {
+ var d = turn._draft;
+ var box = turn.querySelector('[data-ovt-draft]');
+ if (!d || !d.text.trim()) { if (box) box.remove(); return; }
+ if (!box) {
+ var reply = turn.querySelector('.ovt-reply');
+ if (!reply) return;
+ box = document.createElement('div');
+ box.className = 'ovt-answer ct-draft';
+ box.setAttribute('data-ovt-draft', '');
+ box.setAttribute('aria-busy', 'true');
+ reply.appendChild(box);
+ }
+ var stay = nearBottom();
+ drawDraft(box, d.text);
+ if (stay) toBottom(false);
+ }
+ function addDelta(turn, e) {
+ if (!e || typeof e.text !== 'string') return;
+ var d = turn._draft;
+ if (!d || d.n !== e.n) d = turn._draft = { n: e.n, text: '' };
+ d.text += e.text;
+ if (turn._drawing) return;
+ turn._drawing = true;
+ (window.requestAnimationFrame || function (fn) { return setTimeout(fn, 16); })(function () { turn._drawing = false; draft(turn); });
+ }
+ // A stored step that holds the draft's own text (the run kept it as a note), or a tool call: the draft is done.
+ function settleDraft(turn, e) {
+ var d = turn._draft;
+ if (!d || !e) return;
+ var said = typeof e.text === 'string' ? e.text.replace(/…$/, '').trim() : '';
+ if (e.step === 'tool_call' || (e.step === 'text' && said && d.text.trim().indexOf(said) === 0)) { turn._draft = null; draft(turn); }
+ }
function refresh(turn, done) {
function refresh(turn, done) {
var id = turn.getAttribute('data-task');
var id = turn.getAttribute('data-task');
return fetch('/chat/turn/' + encodeURIComponent(id), { credentials: 'same-origin' })
return fetch('/chat/turn/' + encodeURIComponent(id), { credentials: 'same-origin' })
@@ -216,8 +295,13 @@
@@ -216,8 +295,13 @@
var scr = turn.querySelector('[data-screen]'), scr2 = next.querySelector('[data-screen]');
var scr = turn.querySelector('[data-screen]'), scr2 = next.querySelector('[data-screen]');
if (scr && scr2 && (scr.hidden !== scr2.hidden || scr.getAttribute('data-desk') !== scr2.getAttribute('data-desk'))) { scr.replaceWith(scr2); screen(turn); }
if (scr && scr2 && (scr.hidden !== scr2.hidden || scr.getAttribute('data-desk') !== scr2.getAttribute('data-desk'))) { scr.replaceWith(scr2); screen(turn); }
liveLine(turn);
liveLine(turn);
+ draft(turn);
} else {
} else {
stopScreen(turn);
stopScreen(turn);
+ // An answer the person already watched being written is not typed in again: it only takes its
+ // finished form (OVThread types in a turn that was open and is not any more).
+ if (turn.querySelector('[data-ovt-draft]')) turn.removeAttribute('data-open');
+ turn._draft = null;
if (window.OVThread) window.OVThread.replace(id, h); else turn.replaceWith(next);
if (window.OVThread) window.OVThread.replace(id, h); else turn.replaceWith(next);
var dot = chat.querySelector('.ovt-list a[aria-current] .ovt-dot'); if (dot) dot.remove();
var dot = chat.querySelector('.ovt-list a[aria-current] .ovt-dot'); if (dot) dot.remove();
if (note) note.textContent = '';
if (note) note.textContent = '';
@@ -303,7 +387,7 @@
@@ -303,7 +387,7 @@
screen(turn);
screen(turn);
liveLine(turn);
liveLine(turn);
var id = turn.getAttribute('data-task');
var id = turn.getAttribute('data-task');
- var src = new EventSource('/api/v1/tasks/' + encodeURIComponent(id) + '/events?after=' + encodeURIComponent(turn.getAttribute('data-last') || '0'));
+ var src = new EventSource('/api/v1/tasks/' + encodeURIComponent(id) + '/events?deltas=1&after=' + encodeURIComponent(turn.getAttribute('data-last') || '0'));
streams.push(src);
streams.push(src);
var timer = 0;
var timer = 0;
function soon() { if (timer) return; timer = setTimeout(function () { timer = 0; refresh(turn); }, 450); }
function soon() { if (timer) return; timer = setTimeout(function () { timer = 0; refresh(turn); }, 450); }
@@ -312,9 +396,13 @@
@@ -312,9 +396,13 @@
var e = JSON.parse(m.data);
var e = JSON.parse(m.data);
var line = turn.querySelector('[data-ovt-live]');
var line = turn.querySelector('[data-ovt-live]');
if (line && e.step === 'tool_call' && e.tool) line.textContent = 'Using ' + String(e.tool).replace(/_/g, ' ') + '…';
if (line && e.step === 'tool_call' && e.tool) line.textContent = 'Using ' + String(e.tool).replace(/_/g, ' ') + '…';
+ settleDraft(turn, e);
} catch (err) { /* a malformed event is skipped */ }
} catch (err) { /* a malformed event is skipped */ }
soon();
soon();
});
});
+ src.addEventListener('delta', function (m) {
+ try { addDelta(turn, JSON.parse(m.data)); } catch (err) { /* a malformed piece is skipped */ }
+ });
src.addEventListener('state', soon);
src.addEventListener('state', soon);
src.addEventListener('end', function () { src.close(); clearTimeout(timer); timer = 0; refresh(turn); });
src.addEventListener('end', function () { src.close(); clearTimeout(timer); timer = 0; refresh(turn); });
src.onerror = function () { if (src.readyState === 2) setTimeout(function () { refresh(turn); }, 3000); };
src.onerror = function () { if (src.readyState === 2) setTimeout(function () { refresh(turn); }, 3000); };
@@ -4,8 +4,9 @@
@@ -4,8 +4,9 @@
* The model providers' HTTP APIs, metered per call. Two shapes:
* The model providers' HTTP APIs, metered per call. Two shapes:
*
*
* chatClient(cfg) OpenAI-compatible /chat/completions with tools (DeepSeek, an open model on llama.cpp, OpenAI
* chatClient(cfg) OpenAI-compatible /chat/completions with tools (DeepSeek, an open model on llama.cpp, OpenAI
- * for checks): chat({ messages, tools, json, maxTokens, signal }) → { message, text, toolCalls,
- * usage, cost_usd }
+ * for checks): chat({ messages, tools, json, maxTokens, signal, onDelta }) → { message, text,
+ * toolCalls, usage, cost_usd }; with onDelta the answer streams and each piece of its text is
+ * passed to onDelta as it arrives (the result is the same)
* responsesAgent(cfg) OpenAI's Responses API with its hosted web_search tool: run({ instructions, input, effort,
* responsesAgent(cfg) OpenAI's Responses API with its hosted web_search tool: run({ instructions, input, effort,
* signal }) → { text, sources, searches, usage, cost_usd }
* signal }) → { text, sources, searches, usage, cost_usd }
*
*
@@ -20,27 +21,110 @@ class ProviderError extends Error {
@@ -20,27 +21,110 @@ class ProviderError extends Error {
const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));
const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));
-async function post(fetchImpl, url, apiKey, body, { signal, timeoutMs }) {
+async function send(fetchImpl, url, apiKey, body, { signal, timeoutMs, accept = null }) {
const timeout = AbortSignal.timeout(timeoutMs);
const timeout = AbortSignal.timeout(timeoutMs);
const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;
const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;
- let res;
try {
try {
- res = await fetchImpl(url, { method: 'POST', headers: { 'content-type': 'application/json', ...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}) }, body: JSON.stringify(body), signal: sig });
+ return await fetchImpl(url, { method: 'POST', headers: { 'content-type': 'application/json', ...(accept ? { accept } : {}), ...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}) }, body: JSON.stringify(body), signal: sig });
} catch (err) {
} catch (err) {
- if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });
- throw new ProviderError(err && err.name === 'TimeoutError' ? 'the model did not answer in time' : 'the model provider could not be reached', { code: 'actor.provider.unreachable' });
+ throw unreachable(err, signal, 'the model provider');
}
}
+}
+
+/** A failed or aborted request as a ProviderError: cancelled by the run, timed out, or not reachable. */
+function unreachable(err, signal, who) {
+ if (signal && signal.aborted) return new ProviderError('cancelled', { code: 'actor.task.cancelled' });
+ return new ProviderError(err && err.name === 'TimeoutError' ? `${who === 'the model provider' ? 'the model' : who} did not answer in time` : `${who} could not be reached`, { code: 'actor.provider.unreachable' });
+}
+
+function httpError(res, text, json) {
+ const msg = (json && json.error && (json.error.message || json.error)) || clip(text, 200) || `HTTP ${res.status}`;
+ return new ProviderError(`HTTP ${res.status}: ${clip(msg, 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });
+}
+
+async function post(fetchImpl, url, apiKey, body, opts) {
+ const res = await send(fetchImpl, url, apiKey, body, opts);
const text = await res.text();
const text = await res.text();
let json = null;
let json = null;
try { json = JSON.parse(text); } catch { /* not JSON */ }
try { json = JSON.parse(text); } catch { /* not JSON */ }
- if (!res.ok) {
- const msg = (json && json.error && (json.error.message || json.error)) || clip(text, 200) || `HTTP ${res.status}`;
- throw new ProviderError(`HTTP ${res.status}: ${clip(msg, 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });
- }
+ if (!res.ok) throw httpError(res, text, json);
if (!json) throw new ProviderError('the model provider answered with something that is not JSON');
if (!json) throw new ProviderError('the model provider answered with something that is not JSON');
return json;
return json;
}
}
+/**
+ * A Chat Completions answer streamed as server-sent events (stream: true), put back together in the shape of the
+ * whole answer: { choices: [{ message, finish_reason }], model, usage, x_openvibe }. onDelta(text) is called with each
+ * piece of the answer's text as it arrives (never with reasoning or tool calls); a throw in it is ignored. Tool calls
+ * are joined by their index. An upstream that answers with plain JSON (it cannot stream) is read whole, and its text
+ * is passed to onDelta once. A failure the stream itself carries ({ error }) is a ProviderError.
+ */
+async function readStream(res, onDelta, { signal, who = 'the model provider' } = {}) {
+ const tell = (t) => { if (t && onDelta) { try { onDelta(t); } catch { /* the listener's problem, not the call's */ } } };
+ const type = res.headers && typeof res.headers.get === 'function' ? res.headers.get('content-type') : '';
+ if (!/text\/event-stream/i.test(type || '') || !res.body || typeof res.body[Symbol.asyncIterator] !== 'function') {
+ const text = await res.text();
+ let json = null;
+ try { json = JSON.parse(text); } catch { /* not JSON */ }
+ if (!json) throw new ProviderError(`${who} answered with something that is not JSON`);
+ const m = (json.choices && json.choices[0] && json.choices[0].message) || {};
+ tell(typeof m.content === 'string' ? m.content : '');
+ return json;
+ }
+ let content = '', reasoning = '', finish = null, model = null, usage = null, x = null, failure = null;
+ const tools = [];
+ const take = (data) => {
+ if (!data || data === '[DONE]') return;
+ let j;
+ try { j = JSON.parse(data); } catch { return; }
+ if (j.error) { failure = j.error; return; }
+ if (j.model) model = j.model;
+ if (j.usage) usage = j.usage;
+ if (j.x_openvibe) x = j.x_openvibe;
+ const c = Array.isArray(j.choices) ? j.choices.find((ch) => (ch.index || 0) === 0) : null;
+ if (!c) return;
+ const d = c.delta || {};
+ if (typeof d.content === 'string' && d.content) { content += d.content; tell(d.content); }
+ if (typeof d.reasoning_content === 'string') reasoning += d.reasoning_content;
+ for (const tc of Array.isArray(d.tool_calls) ? d.tool_calls : []) {
+ const i = Number.isInteger(tc.index) ? tc.index : tools.length;
+ const t = tools[i] || (tools[i] = { id: '', type: 'function', function: { name: '', arguments: '' } });
+ if (tc.id) t.id = tc.id;
+ if (tc.type) t.type = tc.type;
+ const fn = tc.function || {};
+ // Most providers send the name once; one that repeats it whole on every piece is not doubled.
+ if (fn.name) t.function.name = t.function.name && !fn.name.startsWith(t.function.name) ? t.function.name + fn.name : fn.name;
+ if (typeof fn.arguments === 'string') t.function.arguments += fn.arguments;
+ }
+ if (c.finish_reason) finish = c.finish_reason;
+ };
+ const decoder = new TextDecoder();
+ let buf = '';
+ const lines = () => {
+ let i;
+ while ((i = buf.indexOf('\n')) >= 0) {
+ const line = buf.slice(0, i).replace(/\r$/, '');
+ buf = buf.slice(i + 1);
+ if (line.startsWith('data:')) take(line.slice(5).trim());
+ }
+ };
+ try {
+ for await (const chunk of res.body) { buf += decoder.decode(chunk, { stream: true }); lines(); }
+ } catch (err) {
+ throw unreachable(err, signal, who);
+ }
+ buf += decoder.decode() + '\n';
+ lines();
+ if (failure) {
+ const msg = typeof failure === 'string' ? failure : (failure.message || failure.code || 'error');
+ const code = typeof failure === 'object' ? String(failure.code || '') : '';
+ throw new ProviderError(clip(msg, 300), { code: /quota|rate|allowance/.test(code) ? 'actor.provider.rate_limited' : 'actor.provider.error' });
+ }
+ const calls = tools.filter(Boolean);
+ const message = { role: 'assistant', content: content || (calls.length ? null : ''), ...(calls.length ? { tool_calls: calls } : {}), ...(reasoning ? { reasoning_content: reasoning } : {}) };
+ return { choices: [{ index: 0, message, finish_reason: finish }], model, usage, ...(x ? { x_openvibe: x } : {}) };
+}
+
/** USD for a usage at a rate card: { input, cached, output } tokens, calls. */
/** USD for a usage at a rate card: { input, cached, output } tokens, calls. */
function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {
function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {
const r = rates || {};
const r = rates || {};
@@ -54,13 +138,21 @@ function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {
@@ -54,13 +138,21 @@ function priceUsage(rates, { input = 0, cached = 0, output = 0, calls = 0 }) {
*/
*/
function chatClient({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.fetch, timeoutMs = 90_000, extraBody = null }) {
function chatClient({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.fetch, timeoutMs = 90_000, extraBody = null }) {
const reasoning = /^(gpt-5|o\d)/i.test(model || '');
const reasoning = /^(gpt-5|o\d)/i.test(model || '');
- return async function chat({ messages, tools = null, json = false, maxTokens = 1500, signal } = {}) {
+ return async function chat({ messages, tools = null, json = false, maxTokens = 1500, signal, onDelta = null } = {}) {
const body = { model, messages, ...(extraBody || {}) };
const body = { model, messages, ...(extraBody || {}) };
// "low" is the one effort every reasoning chat model we call accepts (`minimal` is not offered by gpt-5.1).
// "low" is the one effort every reasoning chat model we call accepts (`minimal` is not offered by gpt-5.1).
if (reasoning) { body.max_completion_tokens = maxTokens + 1024; body.reasoning_effort = 'low'; } else { body.max_tokens = maxTokens; body.temperature = 0.2; }
if (reasoning) { body.max_completion_tokens = maxTokens + 1024; body.reasoning_effort = 'low'; } else { body.max_tokens = maxTokens; body.temperature = 0.2; }
if (tools && tools.length) { body.tools = tools; body.tool_choice = 'auto'; }
if (tools && tools.length) { body.tools = tools; body.tool_choice = 'auto'; }
if (json) body.response_format = { type: 'json_object' };
if (json) body.response_format = { type: 'json_object' };
- const j = await post(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs });
+ let j;
+ if (onDelta) {
+ // The answer streams in (plan L55 C2): each piece of its text goes to onDelta as the model writes it.
+ body.stream = true;
+ body.stream_options = { include_usage: true };
+ const res = await send(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs, accept: 'text/event-stream' });
+ if (!res.ok) { const text = await res.text(); let e = null; try { e = JSON.parse(text); } catch { /* not JSON */ } throw httpError(res, text, e); }
+ j = await readStream(res, onDelta, { signal });
+ } else j = await post(fetchImpl, `${baseUrl}/chat/completions`, apiKey, body, { signal, timeoutMs });
const choice = (j.choices && j.choices[0]) || {};
const choice = (j.choices && j.choices[0]) || {};
const message = choice.message || { role: 'assistant', content: '' };
const message = choice.message || { role: 'assistant', content: '' };
const u = j.usage || {};
const u = j.usage || {};
@@ -118,4 +210,4 @@ function responsesAgent({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.
@@ -118,4 +210,4 @@ function responsesAgent({ baseUrl, apiKey, model, rates, fetchImpl = globalThis.
};
};
}
}
-module.exports = { chatClient, responsesAgent, priceUsage, sourcesOf, citationsOf, ProviderError };
+module.exports = { chatClient, responsesAgent, priceUsage, sourcesOf, citationsOf, readStream, ProviderError };
@@ -306,6 +306,8 @@ function createApi(ctx) {
@@ -306,6 +306,8 @@ function createApi(ctx) {
const after = Math.max(0, parseInt(req.get('last-event-id') || req.query.after, 10) || 0);
const after = Math.max(0, parseInt(req.get('last-event-id') || req.query.after, 10) || 0);
await stream.serve(req, res, {
await stream.serve(req, res, {
taskId: row.id, after,
taskId: row.id, after,
+ // The answer as it is written (plan L55 C2), for a reader that asks: transient delta events.
+ deltas: req.query.deltas === '1',
readBacklog: () => store.eventsAfter(s, row.id, after),
readBacklog: () => store.eventsAfter(s, row.id, after),
isEnded: async () => { const cur = await store.getTask(s, row.id); return !cur || !store.OPEN.includes(cur.state); },
isEnded: async () => { const cur = await store.getTask(s, row.id); return !cur || !store.OPEN.includes(cur.state); },
});
});
@@ -288,6 +288,11 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -288,6 +288,11 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;
if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;
const defs = forceNoTools ? null : shortlistTools(tools, state);
const defs = forceNoTools ? null : shortlistTools(tools, state);
const startedAt = now();
const startedAt = now();
+ // The pieces of this call's text go to the page as they arrive (plan L55 C2), numbered by call so the page can
+ // start a fresh draft when a call that ended in tool calls is followed by the next one.
+ const n = (state.deltaCalls = (state.deltaCalls || 0) + 1);
+ // Only while someone reads the answer as it is written: a task nobody watches calls the model as before.
+ const onDelta = ctx.delta && (typeof ctx.delta.wanted !== 'function' || ctx.delta.wanted()) ? (text) => ctx.delta({ text, n }) : null;
let r;
let r;
try {
try {
r = await models.chat({
r = await models.chat({
@@ -296,6 +301,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -296,6 +301,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
tools: defs && defs.length ? defs : null,
tools: defs && defs.length ? defs : null,
maxTokens: MAX_ANSWER_TOKENS,
maxTokens: MAX_ANSWER_TOKENS,
signal: state.signal,
signal: state.signal,
+ ...(onDelta ? { onDelta } : {}),
});
});
} catch (err) {
} catch (err) {
await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });
await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });
@@ -840,7 +846,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -840,7 +846,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
* a span; and drives on from the step it stopped at. A checkpoint whose last step already answered is delivered
* a span; and drives on from the step it stopped at. A checkpoint whose last step already answered is delivered
* without another model call.
* without another model call.
*/
*/
- async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null }) {
+ async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, delta = null }) {
const source = resume && resume.state ? resume.state : (fork && fork.checkpoint && fork.checkpoint.state);
const source = resume && resume.state ? resume.state : (fork && fork.checkpoint && fork.checkpoint.state);
const restored = restoreState(source);
const restored = restoreState(source);
if (!restored) throw Object.assign(new Error('the saved step could not be read'), { code: 'actor.checkpoint.invalid' });
if (!restored) throw Object.assign(new Error('the saved step could not be read'), { code: 'actor.checkpoint.invalid' });
@@ -866,6 +872,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -866,6 +872,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
trace: createTracer({ store, taskId, log, now }),
trace: createTracer({ store, taskId, log, now }),
readOutside: () => state.readOutside === true, notifyTask,
readOutside: () => state.readOutside === true, notifyTask,
emit: async (e) => emit(e),
emit: async (e) => emit(e),
+ // The answer as the model writes it (plan L55 C2): delta({ text, n }), not stored; null when nobody streams.
+ delta: typeof delta === 'function' ? delta : null,
// The connector tools of this run (stage V2-11) and the failures that switch a server off.
// The connector tools of this run (stage V2-11) and the failures that switch a server off.
dynamicTools: new Map(),
dynamicTools: new Map(),
connectorState: { failures: new Map(), disabled: new Set() },
connectorState: { failures: new Map(), disabled: new Set() },
@@ -996,9 +1004,9 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -996,9 +1004,9 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
}
}
}
}
- async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [] }) {
+ async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [], delta = null }) {
if (resume || fork) {
if (resume || fork) {
- return await continueFrom({ task, requester, taskId, mode, signal, emit, meter, budgetLeft, waitForPerson, requestApproval, context, personText, agentId, notifyTask, askPerson, control, resume, fork, spent, waits, waitEvent });
+ return await continueFrom({ task, requester, taskId, mode, signal, emit, meter, budgetLeft, waitForPerson, requestApproval, context, personText, agentId, notifyTask, askPerson, control, resume, fork, spent, waits, waitEvent, delta });
}
}
const tier = mode === 'best' && models.resolve('best').available ? 'best' : 'smart';
const tier = mode === 'best' && models.resolve('best').available ? 'best' : 'smart';
if (!models.resolve(tier).available) throw Object.assign(new Error(models.resolve(tier).reason || 'no model is available for the loop'), { code: 'actor.model.unavailable' });
if (!models.resolve(tier).available) throw Object.assign(new Error(models.resolve(tier).reason || 'no model is available for the loop'), { code: 'actor.model.unavailable' });
@@ -1027,6 +1035,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -1027,6 +1035,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
trace: createTracer({ store, taskId, log, now }),
trace: createTracer({ store, taskId, log, now }),
readOutside: () => state.readOutside === true, notifyTask,
readOutside: () => state.readOutside === true, notifyTask,
emit: async (e) => emit(e),
emit: async (e) => emit(e),
+ // The answer as the model writes it (plan L55 C2): delta({ text, n }), not stored; null when nobody streams.
+ delta: typeof delta === 'function' ? delta : null,
// The connector tools of this run (stage V2-11) and the failures that switch a server off.
// The connector tools of this run (stage V2-11) and the failures that switch a server off.
dynamicTools: new Map(),
dynamicTools: new Map(),
connectorState: { failures: new Map(), disabled: new Set() },
connectorState: { failures: new Map(), disabled: new Set() },
@@ -31,7 +31,7 @@
@@ -31,7 +31,7 @@
const { AsyncLocalStorage } = require('node:async_hooks');
const { AsyncLocalStorage } = require('node:async_hooks');
const { serviceAuth } = require('openvibe-contracts');
const { serviceAuth } = require('openvibe-contracts');
-const { chatClient, priceUsage, ProviderError } = require('../agents/providers');
+const { chatClient, priceUsage, readStream, ProviderError } = require('../agents/providers');
class ModelError extends Error {
class ModelError extends Error {
constructor(message, { code = 'actor.model.unavailable' } = {}) { super(message); this.code = code; }
constructor(message, { code = 'actor.model.unavailable' } = {}) { super(message); this.code = code; }
@@ -297,7 +297,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -297,7 +297,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
* (x-openvibe-credential: <provider>:<subject>, the bare usr_… id AI expects — never "user:usr_…"). AI answers
* (x-openvibe-credential: <provider>:<subject>, the bare usr_… id AI expects — never "user:usr_…"). AI answers
* with that key, or refuses: the person has no key for that provider yet, and the message says where to add it.
* with that key, or refuses: the person has no key for that provider yet, and the message says where to add it.
*/
*/
- async function chatViaAi({ choice, messages, tools, json, maxTokens, signal }) {
+ async function chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta = null }) {
const tokens = aiTokenClient();
const tokens = aiTokenClient();
if (!tokens) throw new ModelError('OpenVibe.AI is not configured on this server (no Network client secret)', { code: 'actor.model.unavailable' });
if (!tokens) throw new ModelError('OpenVibe.AI is not configured on this server (no Network client secret)', { code: 'actor.model.unavailable' });
const base = trimSlash(ai.url);
const base = trimSlash(ai.url);
@@ -313,6 +313,8 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -313,6 +313,8 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
temperature: 0.2,
temperature: 0.2,
// L37: a wish, not a pin. The provider is the one the person's own key is for, so the wish holds.
// L37: a wish, not a pin. The provider is the one the person's own key is for, so the wish holds.
prefer: { provider, ...(choice.model ? { model: choice.model } : {}) },
prefer: { provider, ...(choice.model ? { model: choice.model } : {}) },
+ // The answer streams in (plan L55 C2): AI passes the upstream's pieces on as they arrive.
+ ...(onDelta ? { stream: true, stream_options: { include_usage: true } } : {}),
};
};
const timeout = AbortSignal.timeout(ai.timeoutMs || 180_000);
const timeout = AbortSignal.timeout(ai.timeoutMs || 180_000);
const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;
const sig = signal ? AbortSignal.any([signal, timeout]) : timeout;
@@ -322,7 +324,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -322,7 +324,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
method: 'POST',
method: 'POST',
headers: {
headers: {
'content-type': 'application/json',
'content-type': 'application/json',
- accept: 'application/json',
+ accept: onDelta ? 'text/event-stream' : 'application/json',
...(await tokens.authHeaders()),
...(await tokens.authHeaders()),
// The person's own key for this provider, and who the person is (the bare usr_… id).
// The person's own key for this provider, and who the person is (the bare usr_… id).
'x-openvibe-credential': provider + ':' + choice.subject,
'x-openvibe-credential': provider + ':' + choice.subject,
@@ -334,9 +336,12 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -334,9 +336,12 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });
if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });
throw new ProviderError(err && err.name === 'TimeoutError' ? 'OpenVibe.AI did not answer in time' : 'OpenVibe.AI could not be reached', { code: 'actor.provider.unreachable' });
throw new ProviderError(err && err.name === 'TimeoutError' ? 'OpenVibe.AI did not answer in time' : 'OpenVibe.AI could not be reached', { code: 'actor.provider.unreachable' });
}
}
- const raw = await res.text();
+ let raw = '';
let answer = null;
let answer = null;
- try { answer = JSON.parse(raw); } catch { /* not JSON */ }
+ if (!res.ok || !onDelta) {
+ raw = await res.text();
+ try { answer = JSON.parse(raw); } catch { /* not JSON */ }
+ }
if (!res.ok) {
if (!res.ok) {
const err = (answer && answer.error) || {};
const err = (answer && answer.error) || {};
const code = String(err.code || '');
const code = String(err.code || '');
@@ -353,6 +358,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -353,6 +358,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
}
}
throw new ProviderError('HTTP ' + res.status + ': ' + clip(msg, 300), { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });
throw new ProviderError('HTTP ' + res.status + ': ' + clip(msg, 300), { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });
}
}
+ if (onDelta) answer = await readStream(res, onDelta, { signal, who: 'OpenVibe.AI' });
if (!answer) throw new ProviderError('OpenVibe.AI answered with something that is not JSON');
if (!answer) throw new ProviderError('OpenVibe.AI answered with something that is not JSON');
const first = (answer.choices && answer.choices[0]) || {};
const first = (answer.choices && answer.choices[0]) || {};
const message = first.message || { role: 'assistant', content: '' };
const message = first.message || { role: 'assistant', content: '' };
@@ -371,11 +377,11 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -371,11 +377,11 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
};
};
}
}
/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */
/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */
- async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null } = {}) {
+ async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null } = {}) {
// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it.
// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it.
const choice = currentChoice();
const choice = currentChoice();
if (choice && choice.provider) {
if (choice && choice.provider) {
- const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal });
+ const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta });
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
}
}
await warm();
await warm();
@@ -386,7 +392,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -386,7 +392,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
// The check is a short structured call, so its thinking is off; the loop keeps it on.
// The check is a short structured call, so its thinking is off; the loop keeps it on.
const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;
const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;
const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });
const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });
- const r = await client({ messages, tools, json, maxTokens, signal });
+ const r = await client({ messages, tools, json, maxTokens, signal, onDelta });
return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };
return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };
}
}
@@ -359,6 +359,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -359,6 +359,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.
// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.
context: row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [],
context: row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [],
emit: (e) => emit(id, { kind: 'output', agent: agentId, ...e }),
emit: (e) => emit(id, { kind: 'output', agent: agentId, ...e }),
+ // The answer as it is written (plan L55 C2): to the readers that asked, never stored.
+ // The loop streams a model call only while such a reader is there (wanted).
+ delta: typeof stream.delta === 'function' ? Object.assign((d) => stream.delta(id, d), { wanted: () => stream.wants(id) }) : null,
meter, budgetLeft, waitForPerson, requestApproval,
meter, budgetLeft, waitForPerson, requestApproval,
personText, agentId: ownAgent, notifyTask,
personText, agentId: ownAgent, notifyTask,
askPerson, control, resume, fork, spent: () => spent,
askPerson, control, resume, fork, spent: () => spent,
@@ -9,12 +9,21 @@
@@ -9,12 +9,21 @@
* mark every older seq as already sent: a reader got only the `end` event, seq 9, on PostgreSQL in CI), and an `end`
* mark every older seq as already sent: a reader got only the `end` event, seq 9, on PostgreSQL in CI), and an `end`
* that comes meanwhile closes the stream only after the backlog.
* that comes meanwhile closes the stream only after the backlog.
* The stream is best-effort; the task record is the truth.
* The stream is best-effort; the task record is the truth.
+ *
+ * The answer as it is written (plan L55 C2): a reader that asks with ?deltas=1 also gets `delta` events,
+ * { text, n }: the next piece of the text the model is writing in its n-th call of the run. They are not stored and
+ * carry no id (so Last-Event-ID and the backlog are unchanged), go only to a reader whose backlog is out, and are
+ * gathered for DELTA_MS so a fast model sends a few events a second, not one per token. A stored event flushes the
+ * gathered text first, so a reader sees them in the order they happened. The stored `output` text event, or the
+ * finished task, is still the answer.
*/
*/
const HEARTBEAT_MS = 15_000;
const HEARTBEAT_MS = 15_000;
+const DELTA_MS = 60;
function createTaskStream({ log = console } = {}) {
function createTaskStream({ log = console } = {}) {
const subs = new Map(); // task id → Set<{ res, last }>
const subs = new Map(); // task id → Set<{ res, last }>
+ const gathered = new Map(); // task id → { text, n, timer }: delta text not yet sent
function send(sub, event) {
function send(sub, event) {
if (event.seq <= sub.last) return;
if (event.seq <= sub.last) return;
@@ -29,10 +38,45 @@ function createTaskStream({ log = console } = {}) {
@@ -29,10 +38,45 @@ function createTaskStream({ log = console } = {}) {
}
}
function publish(taskId, event) {
function publish(taskId, event) {
+ flush(taskId);
for (const sub of subs.get(taskId) || []) write(sub, event);
for (const sub of subs.get(taskId) || []) write(sub, event);
}
}
+ function flush(taskId) {
+ const g = gathered.get(taskId);
+ if (!g) return;
+ gathered.delete(taskId);
+ clearTimeout(g.timer);
+ if (!g.text) return;
+ const line = `event: delta\ndata: ${JSON.stringify({ text: g.text, n: g.n })}\n\n`;
+ for (const sub of subs.get(taskId) || []) {
+ if (!sub.deltas || !sub.ready) continue;
+ try { sub.res.write(line); } catch (err) { log.warn('[Actor] stream write failed:', err && err.message); }
+ }
+ }
+
+ /** The next piece of the answer being written (not stored). n: the run's model call it belongs to. */
+ function delta(taskId, { text, n = 0 } = {}) {
+ if (!text) return;
+ if (!wants(taskId)) return;
+ let g = gathered.get(taskId);
+ if (g && g.n !== n) { flush(taskId); g = null; }
+ if (!g) {
+ g = { text: '', n, timer: setTimeout(() => flush(taskId), DELTA_MS) };
+ if (g.timer.unref) g.timer.unref();
+ gathered.set(taskId, g);
+ }
+ g.text += String(text);
+ }
+
+ /** Whether a reader of this task asked for the answer as it is written (the loop streams only then). */
+ function wants(taskId) {
+ return [...(subs.get(taskId) || [])].some((sub) => sub.deltas);
+ }
+
function end(taskId) {
function end(taskId) {
+ const g = gathered.get(taskId);
+ if (g) { clearTimeout(g.timer); gathered.delete(taskId); }
const set = subs.get(taskId);
const set = subs.get(taskId);
subs.delete(taskId);
subs.delete(taskId);
for (const sub of set || []) {
for (const sub of set || []) {
@@ -45,11 +89,11 @@ function createTaskStream({ log = console } = {}) {
@@ -45,11 +89,11 @@ function createTaskStream({ log = console } = {}) {
* Serve one reader. readBacklog() → stored events after `after`; isEnded() → whether the task has ended (read after
* Serve one reader. readBacklog() → stored events after `after`; isEnded() → whether the task has ended (read after
* the backlog, so an end that happened meanwhile closes the stream).
* the backlog, so an end that happened meanwhile closes the stream).
*/
*/
- async function serve(req, res, { taskId, after = 0, readBacklog, isEnded }) {
+ async function serve(req, res, { taskId, after = 0, readBacklog, isEnded, deltas = false }) {
res.status(200).set({ 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no' });
res.status(200).set({ 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no' });
res.flushHeaders();
res.flushHeaders();
res.write(': stream open\n\n');
res.write(': stream open\n\n');
- const sub = { res, last: after, ready: false, pending: [], ended: false };
+ const sub = { res, last: after, ready: false, pending: [], ended: false, deltas: deltas === true };
let set = subs.get(taskId);
let set = subs.get(taskId);
if (!set) { set = new Set(); subs.set(taskId, set); }
if (!set) { set = new Set(); subs.set(taskId, set); }
set.add(sub);
set.add(sub);
@@ -70,7 +114,7 @@ function createTaskStream({ log = console } = {}) {
@@ -70,7 +114,7 @@ function createTaskStream({ log = console } = {}) {
const status = () => ({ streams: [...subs.values()].reduce((n, s) => n + s.size, 0) });
const status = () => ({ streams: [...subs.values()].reduce((n, s) => n + s.size, 0) });
- return { publish, end, serve, status };
+ return { publish, delta, wants, end, serve, status };
}
}
module.exports = { createTaskStream };
module.exports = { createTaskStream };
new file mode 100644
new file mode 100644
@@ -0,0 +1,159 @@
@@ -0,0 +1,159 @@
+'use strict';
+/**
+ * The answer as it is written (plan L55 C2): a model call with onDelta streams (stream: true), passes each piece of
+ * its text on as it arrives, and returns what the whole answer would have been; the task stream sends the pieces to the
+ * readers that asked (?deltas=1) as `delta` events with no id, gathered, never stored, flushed before a stored event.
+ *
+ * node test/answer-stream.test.js
+ */
+const assert = require('assert');
+const { EventEmitter } = require('events');
+const { chatClient, readStream, ProviderError } = require('../server/agents/providers');
+const { createTaskStream } = require('../server/tasks/stream');
+
+const sse = (chunks) => chunks.map((c) => `data: ${typeof c === 'string' ? c : JSON.stringify(c)}\n\n`).join('');
+function streamed(text, { split = 7, status = 200 } = {}) {
+ // Sent in small byte pieces, so a line is often cut in two: the reader must put lines back together.
+ const bytes = new TextEncoder().encode(text);
+ const body = new ReadableStream({
+ start(ctl) { for (let i = 0; i < bytes.length; i += split) ctl.enqueue(bytes.slice(i, i + split)); ctl.close(); },
+ });
+ return new Response(body, { status, headers: { 'content-type': 'text/event-stream; charset=utf-8' } });
+}
+const chunk = (delta, extra = {}) => ({ id: 'c1', object: 'chat.completion.chunk', model: 'deepseek-chat', choices: [{ index: 0, delta, finish_reason: null }], ...extra });
+
+(async () => {
+ // 1. Text, reasoning, tool calls joined by index, finish, usage and AI's x_openvibe, from pieces cut mid-line.
+ {
+ const pieces = [];
+ const body = sse([
+ chunk({ role: 'assistant', content: '' }),
+ chunk({ reasoning_content: 'Think ' }),
+ chunk({ reasoning_content: 'first.' }),
+ chunk({ content: 'Hello' }),
+ chunk({ content: ', **world**' }),
+ chunk({ tool_calls: [{ index: 0, id: 'call_a', type: 'function', function: { name: 'web_', arguments: '{"q":' } }] }),
+ chunk({ tool_calls: [{ index: 0, function: { name: 'search', arguments: '"cats"}' } }] }),
+ chunk({ tool_calls: [{ index: 1, id: 'call_b', type: 'function', function: { name: 'read_page', arguments: '{}' } }] }),
+ { id: 'c1', choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }] },
+ { id: 'c1', choices: [], usage: { prompt_tokens: 120, completion_tokens: 30, prompt_cache_hit_tokens: 100 } },
+ { id: 'c1', choices: [], x_openvibe: { cost_usd: 0.0012 } },
+ '[DONE]',
+ ]);
+ const j = await readStream(streamed(body), (t) => pieces.push(t));
+ assert.deepStrictEqual(pieces, ['Hello', ', **world**'], 'only the answer text goes to onDelta, piece by piece');
+ const m = j.choices[0].message;
+ assert.strictEqual(m.content, 'Hello, **world**');
+ assert.strictEqual(m.reasoning_content, 'Think first.', 'the reasoning is kept on the message (DeepSeek wants it back)');
+ assert.deepStrictEqual(m.tool_calls, [
+ { id: 'call_a', type: 'function', function: { name: 'web_search', arguments: '{"q":"cats"}' } },
+ { id: 'call_b', type: 'function', function: { name: 'read_page', arguments: '{}' } },
+ ]);
+ assert.strictEqual(j.choices[0].finish_reason, 'tool_calls');
+ assert.strictEqual(j.model, 'deepseek-chat');
+ assert.strictEqual(j.usage.prompt_tokens, 120);
+ assert.strictEqual(j.x_openvibe.cost_usd, 0.0012);
+ }
+
+ // 2. A provider that repeats the whole tool name on every piece is not doubled; an empty answer with tool calls
+ // has content null, as the whole answer would.
+ {
+ const body = sse([
+ chunk({ tool_calls: [{ index: 0, id: 'x', function: { name: 'fetch_url', arguments: '{"u"' } }] }),
+ chunk({ tool_calls: [{ index: 0, function: { name: 'fetch_url', arguments: ':1}' } }] }),
+ '[DONE]',
+ ]);
+ const j = await readStream(streamed(body), () => {});
+ assert.strictEqual(j.choices[0].message.tool_calls[0].function.name, 'fetch_url');
+ assert.strictEqual(j.choices[0].message.tool_calls[0].function.arguments, '{"u":1}');
+ assert.strictEqual(j.choices[0].message.content, null);
+ }
+
+ // 3. A failure the stream carries is a ProviderError; a listener that throws does not break the call.
+ {
+ const body = sse([chunk({ content: 'Par' }), { error: { message: 'upstream went away', code: 'provider.error' } }, '[DONE]']);
+ await assert.rejects(readStream(streamed(body), () => { throw new Error('listener'); }), (err) => err instanceof ProviderError && /upstream went away/.test(err.message));
+ }
+
+ // 4. An upstream that answers JSON (it cannot stream) is read whole, and its text reaches onDelta once.
+ {
+ const pieces = [];
+ const res = new Response(JSON.stringify({ model: 'm', choices: [{ index: 0, message: { role: 'assistant', content: 'Whole.' }, finish_reason: 'stop' }], usage: { prompt_tokens: 1, completion_tokens: 1 } }), { headers: { 'content-type': 'application/json' } });
+ const j = await readStream(res, (t) => pieces.push(t));
+ assert.strictEqual(j.choices[0].message.content, 'Whole.');
+ assert.deepStrictEqual(pieces, ['Whole.']);
+ }
+
+ // 5. chatClient: with onDelta the request streams and asks for usage; the result has the usual shape and price.
+ // Without it, nothing changes: one JSON answer.
+ {
+ const sent = [];
+ const fetchImpl = async (url, init) => {
+ const body = JSON.parse(init.body);
+ sent.push({ url, body, accept: init.headers.accept || null });
+ if (body.stream) return streamed(sse([chunk({ content: 'Two ' }), chunk({ content: 'words.' }), { choices: [{ index: 0, delta: {}, finish_reason: 'stop' }] }, { choices: [], usage: { prompt_tokens: 1000000, completion_tokens: 1000000, prompt_cache_hit_tokens: 0 } }, '[DONE]']));
+ return new Response(JSON.stringify({ model: 'deepseek-chat', choices: [{ index: 0, message: { role: 'assistant', content: 'Two words.' }, finish_reason: 'stop' }], usage: { prompt_tokens: 1000000, completion_tokens: 1000000 } }), { headers: { 'content-type': 'application/json' } });
+ };
+ const chat = chatClient({ baseUrl: 'https://api.example', apiKey: 'k', model: 'deepseek-chat', rates: { input_usd_per_mtok: 0.25, output_usd_per_mtok: 1 }, fetchImpl });
+ const pieces = [];
+ const a = await chat({ messages: [{ role: 'user', content: 'hi' }], onDelta: (t) => pieces.push(t) });
+ const b = await chat({ messages: [{ role: 'user', content: 'hi' }] });
+ assert.strictEqual(sent[0].body.stream, true);
+ assert.deepStrictEqual(sent[0].body.stream_options, { include_usage: true });
+ assert.strictEqual(sent[0].accept, 'text/event-stream');
+ assert.ok(!('stream' in sent[1].body), 'no onDelta: no stream');
+ assert.deepStrictEqual(pieces, ['Two ', 'words.']);
+ for (const r of [a, b]) {
+ assert.strictEqual(r.text, 'Two words.');
+ assert.deepStrictEqual(r.toolCalls, []);
+ assert.strictEqual(r.finish, 'stop');
+ assert.strictEqual(r.cost_usd, 1.25, 'priced the same way streamed or not');
+ }
+ }
+
+ // 6. A refused streamed call is the same ProviderError as a refused whole one.
+ {
+ const fetchImpl = async () => new Response(JSON.stringify({ error: { message: 'Too many requests' } }), { status: 429, headers: { 'content-type': 'application/json' } });
+ const chat = chatClient({ baseUrl: 'https://api.example', apiKey: 'k', model: 'm', rates: {}, fetchImpl });
+ await assert.rejects(chat({ messages: [], onDelta: () => {} }), (err) => err.code === 'actor.provider.rate_limited' && err.status === 429);
+ }
+
+ // 7. The task stream: deltas go only to a reader that asked, gathered, with no id line; a stored event flushes the
+ // gathered text first; a new call number flushes the old call's text; the end drops what is left.
+ {
+ const stream = createTaskStream({ log: { warn() {} } });
+ const fakeRes = () => ({ chunks: [], ended: false, status() { return this; }, set() { return this; }, flushHeaders() {}, write(c) { this.chunks.push(c); }, end() { this.ended = true; } });
+ const asked = fakeRes(), plain = fakeRes();
+ await stream.serve(new EventEmitter(), asked, { taskId: 'tsk_d', after: 0, deltas: true, readBacklog: async () => [], isEnded: async () => false });
+ await stream.serve(new EventEmitter(), plain, { taskId: 'tsk_d', after: 0, readBacklog: async () => [], isEnded: async () => false });
+ assert.strictEqual(stream.wants('tsk_d'), true, 'a reader asked: the loop streams');
+ assert.strictEqual(stream.wants('tsk_x'), false, 'nobody asked: the loop calls the model as before');
+ stream.delta('tsk_d', { text: 'Hel', n: 1 });
+ stream.delta('tsk_d', { text: 'lo', n: 1 });
+ stream.delta('tsk_d', { text: 'Next', n: 2 }); // a new call: call 1's text goes out first
+ stream.publish('tsk_d', { task_id: 'tsk_d', seq: 1, kind: 'output', step: 'text', text: 'Hello' });
+ const events = (res) => res.chunks.join('').split('\n\n').filter((b) => /^(id|event):/m.test(b));
+ const got = events(asked);
+ assert.deepStrictEqual(got, [
+ 'event: delta\ndata: {"text":"Hello","n":1}',
+ 'event: delta\ndata: {"text":"Next","n":2}',
+ 'id: 1\nevent: output\ndata: {"task_id":"tsk_d","seq":1,"kind":"output","step":"text","text":"Hello"}',
+ ]);
+ assert.deepStrictEqual(events(plain), [got[2]], 'a reader that did not ask gets only the stored events');
+ stream.delta('tsk_d', { text: 'left over', n: 3 });
+ stream.end('tsk_d');
+ await new Promise((r) => setTimeout(r, 100));
+ assert.ok(!asked.chunks.join('').includes('left over'), 'nothing is sent after the end');
+ // The timer sends gathered text on its own when no stored event comes.
+ const late = fakeRes();
+ await stream.serve(new EventEmitter(), late, { taskId: 'tsk_e', after: 0, deltas: true, readBacklog: async () => [], isEnded: async () => false });
+ stream.delta('tsk_e', { text: 'Soon', n: 1 });
+ await new Promise((r) => setTimeout(r, 120));
+ assert.deepStrictEqual(events(late), ['event: delta\ndata: {"text":"Soon","n":1}']);
+ // Nobody asked: nothing is gathered.
+ stream.delta('tsk_none', { text: 'x', n: 1 });
+ stream.end('tsk_e');
+ }
+
+ console.log('answer stream: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -146,7 +146,12 @@ const LOC = /^\/chat\/(tsk_[0-9A-HJKMNP-TV-Z]{26})#t-\1$/;
@@ -146,7 +146,12 @@ const LOC = /^\/chat\/(tsk_[0-9A-HJKMNP-TV-Z]{26})#t-\1$/;
await check('in the browser (public/js/chat.js): the live stream, the turn fragment, Stop, Enter on a keyboard only', async () => {
await check('in the browser (public/js/chat.js): the live stream, the turn fragment, Stop, Enter on a keyboard only', async () => {
const js = require('fs').readFileSync(require('path').join(__dirname, '..', 'public', 'js', 'chat.js'), 'utf8');
const js = require('fs').readFileSync(require('path').join(__dirname, '..', 'public', 'js', 'chat.js'), 'utf8');
- assert.match(js, /new EventSource\('\/api\/v1\/tasks\/' \+ encodeURIComponent\(id\) \+ '\/events\?after='/, 'follows the task\'s own event stream');
+ assert.match(js, /new EventSource\('\/api\/v1\/tasks\/' \+ encodeURIComponent\(id\) \+ '\/events\?deltas=1&after='/, 'follows the task\'s own event stream, with the answer as it is written');
+ // The answer as it is written (plan L55 C2): delta events build a draft from DOM nodes, never from HTML.
+ assert.match(js, /src\.addEventListener\('delta'/, 'the draft follows the delta events');
+ const drawing = js.slice(js.indexOf('function drawInline'), js.indexOf('function refresh('));
+ assert.ok(drawing.length > 200 && !/innerHTML/.test(drawing), 'the draft is drawn with text nodes only');
+ assert.match(js, /turn\.querySelector\('\[data-ovt-draft\]'\)\) turn\.removeAttribute\('data-open'\)/, 'an answer already watched is not typed in again');
assert.match(js, /fetch\('\/chat\/turn\/' \+ encodeURIComponent\(id\)/, 'the server draws each turn; the page swaps it in');
assert.match(js, /fetch\('\/chat\/turn\/' \+ encodeURIComponent\(id\)/, 'the server draws each turn; the page swaps it in');
assert.match(js, /'\/tasks\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/cancel'/, 'Stop cancels the working turn');
assert.match(js, /'\/tasks\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/cancel'/, 'Stop cancels the working turn');
assert.match(js, /'\/chat\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/steer'/, 'what you write while it works steers it');
assert.match(js, /'\/chat\/' \+ encodeURIComponent\(w\.getAttribute\('data-task'\)\) \+ '\/steer'/, 'what you write while it works steers it');