AI: stream: true streams the upstream's deltas as they arrive (no more one-delta emulation) (#78)

ddiff --git a/README.md b/README.md
iindex a97df2b..4bbfcd4 100644
---- a/README.md
++++ b/README.md
@@@ -153,9 +153,17 @@ coding agents, Codes' harnesses) points its base URL at `https://ai.openvibe.ser
    upstream for `default.chat` by price and health, and a request with `tools` goes only to a provider that declares
    `tools`. The answer's `model` names the model that ran.
  - **Answer:** `chat.completion` with the upstream's `choices` and `usage`, plus `x_openvibe` (`run_id`, `provider`,
--  `fallback_used`, `cost_usd`) and the `X-OpenVibe-Run-Id` header. `stream: true` answers server-sent events in the chunk
--  format, **emulated**: the whole answer arrives as one delta, then the finish reason (and `usage` with
--  `stream_options.include_usage`), then `[DONE]`.
++  `fallback_used`, `cost_usd`) and the `X-OpenVibe-Run-Id` header.
++- **Streaming:** `stream: true` answers server-sent events in the chunk format **as the upstream streams them**.
++  - An OpenAI-format upstream (the router's, or the project's own) is asked with `stream: true` and
++    `stream_options.include_usage`, and each delta goes out when it arrives: text, and a tool call's id, name and
++    argument pieces.
++  - Then come the finish reason, `usage` (with `stream_options.include_usage`), `x_openvibe`, and `[DONE]`.
++  - The run itself ends with the whole answer put back together, the upstream's usage (estimated when it sends none)
++    and its cost, as without streaming.
++  - Once a delta has gone out, a failure is final: an `error` event, then `[DONE]`. It is never retried or handed to a
++    fallback whose words would follow the first one's.
++  - An answer that came back in one piece (a replayed idempotent run) is sent as one delta.
  - **Each request is a run** of workflow `ai.complete` (step kind `complete`): the run's quota, free allowance, cost,
    audit and explain apply, and the prompt is not kept. Errors are OpenAI's shape (`{ error: { message, type, code } }`).
  - **Upstreams:** OpenAI-format providers (`kind: openai`) pass the request through (`complete()`; a reasoning model
ddiff --git a/server/api/openai.js b/server/api/openai.js
iindex e8482c2..a582591 100644
---- a/server/api/openai.js
++++ b/server/api/openai.js
@@@ -20,7 +20,10 @@
   *
   * Each request is a run of workflow ai.complete, so it has the run's quota, free allowance, cost, audit and explain
   * (x_openvibe in the answer); the prompt is not kept. stream: true answers server-sent events in the Chat Completions
-- * chunk format, emulated: the whole answer arrives as one delta, then finish_reason, then [DONE]. Errors are OpenAI's
++ * chunk format as the upstream streams them (an OpenAI-format upstream, the router's or the project's own): each delta
++ * goes out when it arrives, then usage (stream_options.include_usage), x_openvibe and [DONE]. The run still ends with
++ * the whole answer, its usage and cost. An answer that came back in one piece (a replayed idempotent run, an upstream
++ * that cannot stream) is sent as one delta. A failure after the stream began is an `error` event, then [DONE]. Errors are OpenAI's
   * shape ({ error: { message, type, code } }) with the HTTP status the run's problem had.
   */
  const express = require('express');
@@@ -34,6 +37,54 @@ const FIELDS = ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'resp
  // request is OpenAI's 400, not 422.
  const gatewaySafe = (status) => (status === 502 || status === 504 ? 503 : status === 422 ? 400 : status);
  
++/**
++ * The event stream of one streamed completion. delta(d) sends a chunk as it arrives (or keeps it until open(run) has
++ * sent the headers, which happens once the run exists); write(o) sends any chunk; fail() sends OpenAI's error object as
++ * an event; done() ends with [DONE]. A closed connection drops what is left, quietly.
++ */
++function createLiveStream(res) {
++    let run = null;
++    let opened = false;
++    let any = false;
++    const waiting = [];
++    const finishedIdx = new Set();
++    const created = Math.floor(Date.now() / 1000);
++    const alive = () => !res.writableEnded && !res.destroyed;
++    const write = (o) => { if (alive()) res.write(`data: ${JSON.stringify(o)}\n\n`); };
++    const chunk = (d) => ({ id: `chatcmpl-${run.id}`, object: 'chat.completion.chunk', created, model: d.model || 'openvibe', system_fingerprint: null,
++        choices: [{ index: d.index || 0, delta: d.delta || {}, finish_reason: d.finish_reason || null }] });
++    const out = {
++        get opened() { return opened; },
++        delta(d) {
++            any = true;
++            if (d.finish_reason) finishedIdx.add(d.index || 0);
++            if (!opened) { waiting.push(d); return; }
++            write(chunk(d));
++        },
++        open(r) {
++            run = r;
++            res.status(200);
++            res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
++            res.setHeader('Cache-Control', 'no-store');
++            res.setHeader('X-Accel-Buffering', 'no');
++            res.setHeader('X-OpenVibe-Run-Id', r.id);
++            if (typeof res.flushHeaders === 'function') res.flushHeaders();
++            opened = true;
++            for (const d of waiting.splice(0)) write(chunk(d));
++        },
++        write,
++        sentAny: () => any,
++        finished: (i) => finishedIdx.has(i),
++        fail(message, code) {
++            const status = /^(input|validation)\./.test(code || '') ? 400 : /quota|allowance|budget/.test(code || '') ? 429 : 503;
++            write({ error: { message: String(message || 'error').slice(0, 1000), type: status === 400 ? 'invalid_request_error' : status === 429 ? 'rate_limit_error' : 'api_error', code: code || null, param: null } });
++            return out.done();
++        },
++        done() { if (alive()) { res.write('data: [DONE]\n\n'); res.end(); } },
++    };
++    return out;
++}
++
  function openaiError(res, status, message, code, type) {
      res.status(status).json({ error: { message: String(message || 'error').slice(0, 1000), type: type || (status === 401 || status === 403 ? 'authentication_error' : status === 429 ? 'rate_limit_error' : status >= 500 ? 'api_error' : 'invalid_request_error'), code: code || null, param: null } });
  }
@@@ -77,15 +128,24 @@ function openaiRouter({ runs, auth, config, log = console }) {
          const own = b.x_openvibe && typeof b.x_openvibe === 'object' ? b.x_openvibe.provider : undefined;
          if (own !== undefined) body = { ...body, project_provider: own };
          let run;
++        // stream: true. The run is made first (a refused one keeps its status code); then the event stream opens and every
++        // delta the upstream sends goes out as it arrives. A delta that came before the stream opened waits for it.
++        const live = b.stream === true ? createLiveStream(res) : null;
          try {
--            const created = await runs.create(body, req.principal, { trace: req.ov && req.ov.traceId, requestId: req.ov && req.ov.requestId });
++            const created = await runs.create(body, req.principal, { trace: req.ov && req.ov.traceId, requestId: req.ov && req.ov.requestId, ...(live ? { onDelta: live.delta } : {}) });
++            if (live) live.open(created.run);
              run = await runs.explained(await runs.wait(created, Math.max(config.runs.maxWaitMs, 120000)));
          } catch (err) {
++            if (live && live.opened) return live.fail(err && (err.detail || err.message), err && err.code);
              const status = gatewaySafe((err && err.status) || 500);
              if (status === 429 && err.retryAfter) res.setHeader('Retry-After', String(err.retryAfter));
              if (status >= 500) log.warn(`[ai] /v1/chat/completions: ${err && err.message}`);
              return openaiError(res, status, err && (err.detail || err.message), err && err.code);
          }
++        if (live && (run.status === 'failed' || run.status === 'cancelled' || !run.output || !Array.isArray(run.output.choices))) {
++            const code = (run.error && run.error.code) || (['failed', 'cancelled'].includes(run.status) ? 'run.failed' : 'run.timeout');
++            return live.fail((run.error && run.error.detail) || (code === 'run.timeout' ? `the completion is still running; GET /api/v1/runs/${run.id}` : code), code);
++        }
          if (run.status === 'failed' || run.status === 'cancelled') {
              const code = (run.error && run.error.code) || 'run.failed';
              const status = /^(input|validation)\./.test(code) ? 400 : /quota|allowance|budget/.test(code) ? 429
@@@ -100,14 +160,14 @@ function openaiRouter({ runs, auth, config, log = console }) {
          const model = run.output.model || (run.provenance && run.provenance.model) || 'openvibe';
          const usage = { prompt_tokens: run.usage.tokens_in || 0, completion_tokens: run.usage.tokens_out || 0, total_tokens: (run.usage.tokens_in || 0) + (run.usage.tokens_out || 0) };
          const x = { run_id: run.id, provider: run.provenance && run.provenance.provider, fallback_used: Boolean(run.provenance && run.provenance.fallback_used), cost_usd: run.usage.cost_usd, synthetic: Boolean(run.synthetic) };
--        res.setHeader('X-OpenVibe-Run-Id', run.id);
--        if (b.stream === true) {
--            res.status(200);
--            res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
--            res.setHeader('Cache-Control', 'no-store');
--            const send = (o) => res.write(`data: ${JSON.stringify(o)}\n\n`);
++        if (!live) res.setHeader('X-OpenVibe-Run-Id', run.id);   // a stream sent it with its headers
++        if (live) {
++            const send = (o) => live.write(o);
              const base = { id, object: 'chat.completion.chunk', created, model, system_fingerprint: run.output.system_fingerprint || null };
--            for (const c of run.output.choices) {
++            // The deltas already went out as they arrived; close any choice whose finish the upstream did not stream.
++            if (live.sentAny()) {
++                for (const c of run.output.choices) if (!live.finished(c.index || 0)) send({ ...base, choices: [{ index: c.index || 0, delta: {}, finish_reason: c.finish_reason || 'stop' }] });
++            } else for (const c of run.output.choices) {
                  const m = c.message || {};
                  const delta = { role: 'assistant', content: m.content == null ? null : m.content };
                  if (Array.isArray(m.tool_calls) && m.tool_calls.length) delta.tool_calls = m.tool_calls.map((t, i) => ({ index: i, ...t }));
@@@ -116,8 +176,7 @@ function openaiRouter({ runs, auth, config, log = console }) {
              }
              if (b.stream_options && b.stream_options.include_usage) send({ ...base, choices: [], usage });
              send({ ...base, choices: [], x_openvibe: x });
--            res.write('data: [DONE]\n\n');
--            return res.end();
++            return live.done();
          }
          return res.status(200).json({ id, object: 'chat.completion', created, model, choices: run.output.choices, usage, system_fingerprint: run.output.system_fingerprint || null, x_openvibe: x });
      });
ddiff --git a/server/home.js b/server/home.js
iindex 61ec70b..4204444 100644
---- a/server/home.js
++++ b/server/home.js
@@@ -156,7 +156,7 @@ function llmsTxt({ siteUrl }) {
          '',
          '- Token: POST https://openvibe.network/oauth/token with grant_type=client_credentials, your app\'s client id and secret, audience=openvibe.ai (five minutes).',
          `- Operations: POST ${siteUrl}/api/v1/chat, /generate, /summarize, /classify, /extract, /embed; GET ${siteUrl}/api/v1/runs/:id for your own runs.`,
--        `- OpenAI-compatible: POST ${siteUrl}/v1/chat/completions (Chat Completions with tools; stream is emulated) and GET ${siteUrl}/v1/models, with the same token. Point an OpenAI client's base URL at ${siteUrl}/v1.`,
++        `- OpenAI-compatible: POST ${siteUrl}/v1/chat/completions (Chat Completions with tools; stream: true streams the upstream's deltas) and GET ${siteUrl}/v1/models, with the same token. Point an OpenAI client's base URL at ${siteUrl}/v1.`,
          `- An app's own provider: add x_openvibe.provider { base_url, model, secret } to name the project's own OpenAI-format API; secret is a project secret granted to ai for reveal to that origin (ADR-056), revealed per run and never stored.`,
          '- Every run response carries explain: the objective, the model selected and every candidate with its estimated cost, latency and excluded reason.',
          '- Sandbox apps run on free and local capacity only; runs, usage and the free allowance are per project; a person\'s own provider key is never used for an app.',
ddiff --git a/server/providers/guarded-fetch.js b/server/providers/guarded-fetch.js
iindex c0f8112..ccac779 100644
---- a/server/providers/guarded-fetch.js
++++ b/server/providers/guarded-fetch.js
@@@ -33,6 +33,22 @@ function guardedFetch(input, init = {}) {
          if (signal) { if (signal.aborted) return onAbort(); signal.addEventListener('abort', onAbort, { once: true }); }
          req.on('error', (err) => { if (signal) signal.removeEventListener('abort', onAbort); reject(err); });
          req.on('response', (res) => {
++            if (init.stream === true && res.statusCode >= 200 && res.statusCode < 300) {
++                // A streamed answer: hand the body over as it arrives, still capped at MAX_BYTES.
++                if (signal) signal.removeEventListener('abort', onAbort);
++                let seen = 0;
++                const body = new ReadableStream({
++                    start(controller) {
++                        res.on('data', (c) => { seen += c.length; if (seen > MAX_BYTES) { res.destroy(); controller.error(new Error('response too large')); } else controller.enqueue(new Uint8Array(c)); });
++                        res.on('end', () => { try { controller.close(); } catch { /* already errored */ } });
++                        res.on('error', (e) => { try { controller.error(e); } catch { /* already closed */ } });
++                    },
++                    cancel() { res.destroy(); req.destroy(); },
++                });
++                const h = new Headers();
++                for (const [k, v] of Object.entries(res.headers)) if (v != null) h.set(k, Array.isArray(v) ? v.join(', ') : String(v));
++                return resolve(new Response(body, { status: res.statusCode, headers: h }));
++            }
              const chunks = []; let bytes = 0;
              res.on('data', (c) => { bytes += c.length; if (bytes > MAX_BYTES) { res.destroy(); req.destroy(); reject(new Error('response too large')); } else chunks.push(c); });
              res.on('error', reject);
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex 7a29593..e73cc92 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -383,12 +383,14 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
          const base = { operation, provider_key: cred.providerKey, model_key: model, route_key: ctx.routeKey || null, route_version: ctx.routeVersion || null, fallback: 0, prompt_hash: ctx.promptHash || null };
          const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || 60000, 120000));
          const call = completing ? a.complete : a.chat;
++        let streamed = false;
++        const onDelta = completing && typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
          let lastErr = null;
          for (let attempt = 0; attempt <= config.providerRetries; attempt++) {
              if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
              const t0 = Date.now();
              try {
--                const result = await call({ ...req, model, signal: ctx.signal, timeoutMs });
++                const result = await call({ ...req, ...(onDelta ? { onDelta } : {}), model, signal: ctx.signal, timeoutMs });
                  const usage = result.usage || { input: 0, output: 0, cached: 0 };
                  const cost = rates.costOfUsage(rates.buildCards({ providerKey: cred.providerKey, model: result.model || model, pricing: config.pricing, flat: false }), usage);
                  const latency = Date.now() - t0;
@@@ -403,6 +405,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
                  if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
                  if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
                  if (err && err.code === 'EADDRNOTPUBLIC') throw new AiError(422, 'credential.endpoint_refused', err.message);
++                if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`);
                  if (attempt < config.providerRetries && retryable(err)) { await sleep(config.retryDelayMs, ctx.signal).catch(() => {}); continue; }
                  break;
              }
@@@ -418,6 +421,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
          if (req.openai && Array.isArray(req.openai.tools) && req.openai.tools.length) features.push('tools');
          const tried = [];
          let unreadable = null;   // a candidate could not read the caller's media (whisper's ffmpeg): no other will either
++        // A streamed answer (/v1/chat/completions, stream: true) is the caller's once a delta has gone out: a failure
++        // after that is final, never retried or handed to a fallback whose words would follow the first one's.
++        let streamed = false;
++        const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
          const { order: list, explain } = await candidates(route, features, operation, ctx);
          for (let i = 0; i < list.length; i++) {
              const c = list[i];
@@@ -447,7 +454,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
                      const fn = a[operation];
                      if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });
                      statFor(c.provider).calls += p.kind === 'stub' ? 0 : 1;   // stub counts its own calls
--                    const result = await fn({ ...req, model, signal: ctx.signal, timeoutMs });
++                    const result = await fn({ ...req, ...(onDelta ? { onDelta } : {}), model, signal: ctx.signal, timeoutMs });
                      if (operation === 'embed' && (!Array.isArray(result.vectors) || result.vectors.length !== req.input.length
                          || result.vectors.some(v => !Array.isArray(v) || !v.length || v.some(n => typeof n !== 'number' || !Number.isFinite(n)))))
                          throw new ProviderError('embedding response has invalid vectors', { code: 'provider.output' });
@@@ -477,6 +484,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
                      }
                      statFor(c.provider).failures++;
                      if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
++                    if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`, { tried: [...tried, { provider: c.provider, error: String(err && err.message || err).slice(0, 200) }], explain });
                      // A timed-out image may still be generated and billed upstream: never pay for it twice.
                      if (attempt < retries && retryable(err) && !(operation === 'image' && err.code === 'provider.timeout')) { await sleep(config.retryDelayMs + Math.floor(Math.random() * Math.min(700, config.retryDelayMs)), ctx.signal).catch(() => {}); continue; }
                      break;
ddiff --git a/server/providers/openai.js b/server/providers/openai.js
iindex 2823b19..80f3002 100644
---- a/server/providers/openai.js
++++ b/server/providers/openai.js
@@@ -14,7 +14,8 @@
   */
  const fs = require('fs');
  const path = require('path');
--const { ProviderError, postJson, deadline, describeAbort } = require('./common');
++const { ProviderError, postJson, deadline, describeAbort, approxTokens } = require('./common');
++const { sseEvents, createAccumulator } = require('./stream');
  
  const DEFAULT_BASE_URL = 'https://api.openai.com/v1';
  const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
@@@ -143,6 +144,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
              if (req.maxTokens) body.max_tokens = req.maxTokens;
              if (req.temperature != null) body.temperature = req.temperature;
          }
++        if (typeof req.onDelta === 'function') return await completeStreaming(body, req);
          const j = await postJson(`${baseUrl}/chat/completions`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
          if (!Array.isArray(j.choices) || !j.choices.length) throw new ProviderError('completion has no choices', { code: 'provider.output' });
          const msg = j.choices[0].message || {};
@@@ -155,6 +157,64 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
          };
      }
  
++    /**
++     * complete with stream: true: each upstream chunk's choices go to req.onDelta as they arrive, and the answer is put
++     * back together (content joined, tool calls assembled) so the run ends with the same completion as without
++     * streaming. Usage comes from the final chunk (stream_options.include_usage); an upstream that sends none is
++     * estimated from the text.
++     */
++    async function completeStreaming(body, req) {
++        let res;
++        try {
++            res = await fetchImpl(`${baseUrl}/chat/completions`, {
++                method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream', ...headers() },
++                body: JSON.stringify({ ...body, stream: true, stream_options: { include_usage: true } }),
++                signal: deadline(req.signal, req.timeoutMs), stream: true,
++            });
++        } catch (err) {
++            throw describeAbort(err, req.signal, req.timeoutMs) || new ProviderError(`connection failed: ${err.message}`, { code: 'provider.unreachable' });
++        }
++        if (!res.ok) {
++            const text = await res.text().catch(() => '');
++            let j = {};
++            try { j = JSON.parse(text); } catch { /* not JSON */ }
++            const msg = (j.error && (j.error.message || j.error)) || `HTTP ${res.status}: ${String(text).slice(0, 200)}`;
++            throw new ProviderError(typeof msg === 'string' ? msg : JSON.stringify(msg), { status: res.status, code: res.status === 401 || res.status === 403 ? 'provider.auth' : 'provider.http', body: j });
++        }
++        const acc = createAccumulator();
++        let usage = null;
++        let model = req.model;
++        let fingerprint = null;
++        try {
++            for await (const chunk of sseEvents(res.body)) {
++                if (chunk.error) throw new ProviderError(String(chunk.error.message || chunk.error).slice(0, 300), { code: 'provider.http' });
++                if (chunk.model) model = chunk.model;
++                if (chunk.system_fingerprint) fingerprint = chunk.system_fingerprint;
++                if (chunk.usage) usage = chunk.usage;
++                for (const c of chunk.choices || []) {
++                    acc.add(c);
++                    req.onDelta({ index: Number(c.index) || 0, delta: c.delta || {}, finish_reason: c.finish_reason || null, model });
++                }
++            }
++        } catch (err) {
++            if (err instanceof ProviderError) throw err;
++            throw describeAbort(err, req.signal, req.timeoutMs) || new ProviderError(`the stream broke: ${err.message}`, { code: 'provider.unreachable' });
++        }
++        const choices = acc.result();
++        if (!choices.length) throw new ProviderError('completion has no choices', { code: 'provider.output' });
++        const first = choices[0].message;
++        const u = usage || {};
++        const estimated = !usage;
++        const inTok = estimated ? (body.messages || []).reduce((n, m) => n + approxTokens(typeof m.content === 'string' ? m.content : JSON.stringify(m.content || '')), 0) : (u.prompt_tokens || 0);
++        const outTok = estimated ? approxTokens(`${first.content || ''}${JSON.stringify(first.tool_calls || '')}`) : (u.completion_tokens || 0);
++        return {
++            completion: { choices, system_fingerprint: fingerprint },
++            text: typeof first.content === 'string' ? first.content : '',
++            model,
++            usage: { input: inTok, output: outTok, cached: (u.prompt_tokens_details && u.prompt_tokens_details.cached_tokens) || 0, ...(estimated ? { estimated: true } : {}) },
++        };
++    }
++
      const adapter = {
          key: record.key,
          kind: 'openai',
ddiff --git a/server/providers/stream.js b/server/providers/stream.js
nnew file mode 100644
iindex 0000000..2661ec2
---- /dev/null
++++ b/server/providers/stream.js
@@@ -0,0 +1,65 @@
++'use strict';
++/**
++ * Streamed Chat Completions (POST /v1/chat/completions with stream: true): read an upstream's server-sent events and
++ * put the answer back together, so a streamed run ends with the same completion, usage and record as one that was not.
++ *
++ *   sseEvents(body)       async iterator of the JSON objects in `data:` lines, until `data: [DONE]` (a web
++ *                         ReadableStream or any async iterable of bytes)
++ *   createAccumulator()   add(choiceDelta) for each chunk's choices; result() → the choices as a non-streamed answer
++ *                         (content concatenated, tool calls assembled by index with their arguments joined)
++ */
++
++async function* sseEvents(body) {
++    const dec = new TextDecoder();
++    let buf = '';
++    for await (const part of body) {
++        buf += dec.decode(part, { stream: true });
++        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:')) continue;
++            const data = line.slice(5).trim();
++            if (data === '[DONE]') return;
++            if (!data) continue;
++            let obj;
++            try { obj = JSON.parse(data); } catch { continue; }
++            yield obj;
++        }
++    }
++}
++
++function createAccumulator() {
++    const choices = new Map();
++    return {
++        add(c) {
++            const i = Number(c.index) || 0;
++            const cur = choices.get(i) || { index: i, message: { role: 'assistant', content: null }, finish_reason: null, tools: [] };
++            const d = c.delta || {};
++            if (d.role) cur.message.role = d.role;
++            if (typeof d.content === 'string') cur.message.content = (cur.message.content || '') + d.content;
++            if (Array.isArray(d.tool_calls)) {
++                for (const t of d.tool_calls) {
++                    const k = Number(t.index) || 0;
++                    const slot = cur.tools[k] || (cur.tools[k] = { id: '', type: 'function', function: { name: '', arguments: '' } });
++                    if (t.id) slot.id = t.id;
++                    if (t.type) slot.type = t.type;
++                    if (t.function) {
++                        if (t.function.name) slot.function.name += t.function.name;
++                        if (t.function.arguments) slot.function.arguments += t.function.arguments;
++                    }
++                }
++            }
++            if (c.finish_reason) cur.finish_reason = c.finish_reason;
++            choices.set(i, cur);
++        },
++        result() {
++            return [...choices.values()].sort((a, b) => a.index - b.index).map((c) => {
++                const tools = c.tools.filter(Boolean);
++                return { index: c.index, message: { ...c.message, ...(tools.length ? { tool_calls: tools } : {}) }, finish_reason: c.finish_reason || (tools.length ? 'tool_calls' : 'stop') };
++            });
++        },
++    };
++}
++
++module.exports = { sseEvents, createAccumulator };
ddiff --git a/server/providers/stub.js b/server/providers/stub.js
iindex 15a2232..2b0fe47 100644
---- a/server/providers/stub.js
++++ b/server/providers/stub.js
@@@ -158,12 +158,24 @@ function createStubProvider(record = { key: 'stub', metadata: {} }, { stats = nu
          const rnd = seeded(`complete:${JSON.stringify(msgs).slice(-2000)}`);
          const inTok = msgs.reduce((n, m) => n + approxTokens(typeof m.content === 'string' ? m.content : JSON.stringify(m.content || '')), 0);
          const tool = Array.isArray(o.tools) && o.tools.length && last.role !== 'tool' && o.tool_choice !== 'none' ? o.tools[0] : null;
++        const stream = typeof req.onDelta === 'function' ? req.onDelta : null;
          if (tool && tool.function) {
              const args = tool.function.parameters ? synthesize(tool.function.parameters, rnd, 0, 'arguments') : {};
              const call = { id: `call_${crypto.createHash('sha256').update(JSON.stringify(msgs)).digest('hex').slice(0, 24)}`, type: 'function', function: { name: tool.function.name, arguments: JSON.stringify(args) } };
++            if (stream) {
++                // As OpenAI streams a call: the id and name first, then the arguments in pieces, then the finish.
++                stream({ index: 0, delta: { role: 'assistant', content: null, tool_calls: [{ index: 0, id: call.id, type: 'function', function: { name: call.function.name, arguments: '' } }] }, finish_reason: null, model: req.model || 'stub-1' });
++                for (const piece of call.function.arguments.match(/.{1,8}/gs) || []) stream({ index: 0, delta: { tool_calls: [{ index: 0, function: { arguments: piece } }] }, finish_reason: null, model: req.model || 'stub-1' });
++                stream({ index: 0, delta: {}, finish_reason: 'tool_calls', model: req.model || 'stub-1' });
++            }
              return { completion: { choices: [{ index: 0, message: { role: 'assistant', content: null, tool_calls: [call] }, finish_reason: 'tool_calls' }] }, text: '', model: req.model || 'stub-1', synthetic: true, usage: { input: inTok, output: 20, cached: 0, estimated: true } };
          }
          const text = last.role === 'tool' ? `(synthetic) the tool answered: ${String(last.content || '').slice(0, 80)}` : `(synthetic) ${phrase(rnd, 8)}`;
++        if (stream) {
++            const words = text.split(/(?<= )/);
++            words.forEach((w, i) => stream({ index: 0, delta: i === 0 ? { role: 'assistant', content: w } : { content: w }, finish_reason: null, model: req.model || 'stub-1' }));
++            stream({ index: 0, delta: {}, finish_reason: 'stop', model: req.model || 'stub-1' });
++        }
          return { completion: { choices: [{ index: 0, message: { role: 'assistant', content: text }, finish_reason: 'stop' }] }, text, model: req.model || 'stub-1', synthetic: true, usage: { input: inTok, output: approxTokens(text), cached: 0, estimated: true } };
      }
  
ddiff --git a/server/runs.js b/server/runs.js
iindex b71d5c7..a6a4e10 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -154,7 +154,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
       * principal: { sub, subject: {type, id} }
       * Returns { run, created, promise? }.
       */
--    async function create(body, principal, { trace = null, requestId = null, retryOf = null } = {}) {
++    async function create(body, principal, { trace = null, requestId = null, retryOf = null, onDelta = null } = {}) {
          if (!body || typeof body !== 'object') throw new AiError(400, 'input.invalid', 'JSON body required');
          const wfKey = String(body.workflow || '');
          const wf = body.version ? await registry.getWorkflow(wfKey, Number(body.version)) : await registry.activeWorkflow(wfKey);
@@@ -206,6 +206,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              requesterType: requester.type, requesterId: requester.id,
              actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,
              ...(app ? { appRun: true, projectId: app.projectId, sandbox: app.sandbox } : {}),
++            // POST /v1/chat/completions with stream: true: the upstream's deltas, as they arrive (kept in memory only).
++            ...(typeof onDelta === 'function' ? { onDelta } : {}),
          };
          const now = iso(clock.now());
          const id = newRunId();
@@@ -356,7 +358,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  // limited to free and local capacity when it is a sandbox run or when no tier budget governs its
                  // project (AI_GOVERN_TIERS off): metered capacity is only ever spent under a project budget.
                  appRun: Boolean(ctx.appRun), freeOnly: Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled))),
--                ...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}) });
++                ...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),
++                ...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}) });
              if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
              // The status, its citations and the ai.run.succeeded event commit together.
              const done = await db.tx(async () => {
ddiff --git a/server/workflows/engine.js b/server/workflows/engine.js
iindex f613850..67b0803 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -199,6 +199,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
              maxTokens: Math.max(1, Math.round(input.max_tokens || route.max_output_tokens || 1024)),
              temperature: input.temperature == null ? null : input.temperature,
              timeoutMs: Math.max(route.timeout_ms || 0, 120000),
++            ...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}),
          };
          const exec = await pool.execute(route, 'complete', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });
          const c = exec.result.completion;
ddiff --git a/test/openai-compat.test.js b/test/openai-compat.test.js
iindex b99f90b..25f9315 100644
---- a/test/openai-compat.test.js
++++ b/test/openai-compat.test.js
@@@ -52,19 +52,42 @@ t.test('tools: the model calls one; the tool result goes back and it answers', a
      assert.match(second.body.choices[0].message.content, /19 C and sunny/);
  });
  
--t.test('stream: chunks with the delta, then finish_reason, usage on request, then [DONE]', async () => {
--    const res = await fetch(`${h.base}/v1/chat/completions`, { method: 'POST', headers: { Authorization: `Bearer ${service}`, 'Content-Type': 'application/json' },
--        body: JSON.stringify({ model: 'openvibe', stream: true, stream_options: { include_usage: true }, messages: [{ role: 'user', content: 'Stream please' }] }) });
++const sse = async (res) => {
++    const lines = (await res.text()).split('\n').filter((l) => l.startsWith('data: ')).map((l) => l.slice(6));
++    return { done: lines[lines.length - 1] === '[DONE]', chunks: lines.filter((l) => l !== '[DONE]').map((l) => JSON.parse(l)) };
++};
++const streamPost = (body, tok = service) => fetch(`${h.base}/v1/chat/completions`, { method: 'POST', headers: { Authorization: `Bearer ${tok}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ stream: true, ...body }) });
++
++t.test('stream: the deltas as they come, joined they are the answer; then finish, usage on request, x_openvibe, [DONE]', async () => {
++    const res = await streamPost({ model: 'openvibe', stream_options: { include_usage: true }, messages: [{ role: 'user', content: 'Stream please' }] });
      assert.strictEqual(res.status, 200);
      assert.match(res.headers.get('content-type'), /^text\/event-stream/);
--    const lines = (await res.text()).split('\n').filter((l) => l.startsWith('data: ')).map((l) => l.slice(6));
--    assert.strictEqual(lines[lines.length - 1], '[DONE]');
--    const chunks = lines.slice(0, -1).map((l) => JSON.parse(l));
--    assert.ok(chunks.every((c) => c.object === 'chat.completion.chunk'));
--    assert.strictEqual(chunks[0].choices[0].delta.role, 'assistant');
--    assert.strictEqual(typeof chunks[0].choices[0].delta.content, 'string');
--    assert.strictEqual(chunks[1].choices[0].finish_reason, 'stop');
++    const runId = res.headers.get('x-openvibe-run-id');
++    assert.match(String(runId), /^run_/);
++    const { done, chunks } = await sse(res);
++    assert.ok(done, '[DONE] last');
++    assert.ok(chunks.every((c) => c.object === 'chat.completion.chunk' && c.id === `chatcmpl-${runId}`));
++    const deltas = chunks.filter((c) => c.choices.length && c.choices[0].delta && typeof c.choices[0].delta.content === 'string');
++    assert.ok(deltas.length > 2, `streamed in pieces (${deltas.length})`);
++    assert.strictEqual(deltas[0].choices[0].delta.role, 'assistant');
++    const joined = deltas.map((c) => c.choices[0].delta.content).join('');
++    const finish = chunks.findIndex((c) => c.choices.length && c.choices[0].finish_reason);
++    assert.strictEqual(chunks[finish].choices[0].finish_reason, 'stop');
++    assert.ok(finish > chunks.indexOf(deltas[deltas.length - 1]), 'the finish comes after the last delta');
      assert.ok(chunks.some((c) => c.usage && c.usage.total_tokens > 0), 'usage when asked');
++    assert.strictEqual(chunks.find((c) => c.x_openvibe).x_openvibe.run_id, runId);
++    const run = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: service });
++    assert.strictEqual(run.body.run.output.choices[0].message.content, joined, 'the run kept the whole answer the stream sent');
++});
++
++t.test('stream: a tool call arrives as its id and name, then its arguments in pieces, and they join into valid JSON', async () => {
++    const { chunks } = await sse(await streamPost({ messages: [{ role: 'user', content: 'Weather in Lisbon?' }], tools: [WEATHER] }));
++    const calls = chunks.flatMap((c) => (c.choices[0] && c.choices[0].delta && c.choices[0].delta.tool_calls) || []);
++    assert.ok(calls.length > 2, 'in pieces');
++    assert.strictEqual(calls[0].function.name, 'get_weather');
++    assert.match(calls[0].id, /^call_/);
++    assert.ok(JSON.parse(calls.map((t) => (t.function && t.function.arguments) || '').join('')), 'the arguments join into JSON');
++    assert.strictEqual(chunks.find((c) => c.choices[0] && c.choices[0].finish_reason).choices[0].finish_reason, 'tool_calls');
  });
  
  t.test('an app token works, and its run is its project\'s', async () => {
@@@ -124,6 +147,17 @@ t.test('production\'s shape: an OpenAI-format provider declaring chat (not compl
          req.on('data', (d) => { b += d; });
          req.on('end', () => {
              calls.push({ path: req.url, body: JSON.parse(b) });
++            const body = JSON.parse(b);
++            if (body.stream) {
++                res.writeHead(200, { 'content-type': 'text/event-stream' });
++                const chunk = (o) => res.write(`data: ${JSON.stringify({ id: 'chatcmpl-up', object: 'chat.completion.chunk', model: 'gpt-up', ...o })}\n\n`);
++                if (body.messages[0].content === 'break') { chunk({ choices: [{ index: 0, delta: { role: 'assistant', content: 'half ' }, finish_reason: null }] }); return setTimeout(() => res.destroy(), 50); }
++                ['from ', 'the ', 'stream'].forEach((w, i) => chunk({ choices: [{ index: 0, delta: i ? { content: w } : { role: 'assistant', content: w }, finish_reason: null }] }));
++                chunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }] });
++                chunk({ choices: [], usage: { prompt_tokens: 9, completion_tokens: 3, total_tokens: 12 } });
++                res.write('data: [DONE]\n\n');
++                return res.end();
++            }
              res.writeHead(200, { 'content-type': 'application/json' });
              res.end(JSON.stringify({ id: 'chatcmpl-up', object: 'chat.completion', created: 1, model: 'gpt-up', choices: [{ index: 0, message: { role: 'assistant', content: 'from upstream' }, finish_reason: 'stop' }], usage: { prompt_tokens: 3, completion_tokens: 2, total_tokens: 5 } }));
          });
@@@ -140,10 +174,27 @@ t.test('production\'s shape: an OpenAI-format provider declaring chat (not compl
          assert.deepStrictEqual([calls.length, calls[0].path, calls[0].body.model], [1, '/v1/chat/completions', 'gpt-up']);
          // Tools need a provider that declares them: none does here, so the answer is OpenAI's JSON error with 503 (never a
          // 502, which Cloudflare would replace with its own page).
++        // stream: true reaches the upstream as stream: true (include_usage asked), and its deltas are passed on in order.
++        const streamed = await fetch(`${h2.base}/v1/chat/completions`, { method: 'POST', headers: { Authorization: `Bearer ${service}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ stream: true, messages: [{ role: 'user', content: 'stream it' }] }) });
++        const sent = await sse(streamed);
++        assert.ok(sent.done);
++        assert.deepStrictEqual(sent.chunks.filter((c) => c.choices.length && c.choices[0].delta.content).map((c) => c.choices[0].delta.content), ['from ', 'the ', 'stream'], 'the upstream deltas, in order');
++        assert.deepStrictEqual([calls[calls.length - 1].body.stream, calls[calls.length - 1].body.stream_options], [true, { include_usage: true }]);
++        const srun = await request(h2.base, 'GET', `/api/v1/runs/${sent.chunks.find((c) => c.x_openvibe).x_openvibe.run_id}`, { tok: service });
++        assert.deepStrictEqual([srun.body.run.output.choices[0].message.content, srun.body.run.usage.tokens_in, srun.body.run.usage.tokens_out], ['from the stream', 9, 3], 'the run kept the joined answer and the upstream usage');
++        // An upstream that breaks off after a delta: an error event, then [DONE]; nothing is retried or handed on.
++        const before = calls.length;
++        const broken = await sse(await fetch(`${h2.base}/v1/chat/completions`, { method: 'POST', headers: { Authorization: `Bearer ${service}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ stream: true, messages: [{ role: 'user', content: 'break' }] }) }));
++        assert.ok(broken.done);
++        assert.strictEqual(broken.chunks[0].choices[0].delta.content, 'half ');
++        const err = broken.chunks.find((c) => c.error);
++        assert.ok(err && err.error.code === 'provider.stream_broken', JSON.stringify(broken.chunks).slice(0, 300));
++        assert.strictEqual(calls.length, before + 1, 'not retried');
++        const beforeTools = calls.length;
          const tools = await request(h2.base, 'POST', '/v1/chat/completions', { tok: service, body: { messages: [{ role: 'user', content: 'Weather?' }], tools: [WEATHER] } });
          assert.strictEqual(tools.status, 503, tools.text);
          assert.strictEqual(tools.body.error.code, 'provider.unavailable');
--        assert.strictEqual(calls.length, 1, 'the provider without tools was not asked');
++        assert.strictEqual(calls.length, beforeTools, 'the provider without tools was not asked');
      } finally {
          await h2.stop();
          await new Promise((r) => upstream.close(r));