Runs may stream: POST /api/v1/runs?stream=1 answers server-sent events, delta { text } as the provider writes the answer, then run { run } (a replay or a cached run sends only its run; a refused one keeps its status) (#153)

ddiff --git a/server/api/runs.js b/server/api/runs.js
iindex 5ff1fbc..d458c6c 100644
---- a/server/api/runs.js
++++ b/server/api/runs.js
@@@ -2,7 +2,10 @@
  /**
   * Run API.
   *
-- *   POST /api/v1/runs                       create (ai.run.create); ?wait=ms waits for the result
++ *   POST /api/v1/runs                       create (ai.run.create); ?wait=ms waits for the result; ?stream=1 answers
++ *                                           server-sent events instead: `delta` { text } as an OpenAI-format upstream
++ *                                           writes the answer (each tool-calling turn's text too), then `run` { run }
++ *                                           when it ends or the wait is over (`error` { code, message } if it breaks)
   *   GET  /api/v1/runs                       the caller's runs (ai.run.read)
   *   GET  /api/v1/runs/:id                   one run (+ citations, request log metadata)
   *   run.usage.usage_sample_ids: the run's Billing readings (one per attempt per token metric), null when none
@@@ -66,11 +69,47 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
          return res.status(status).json({ run, replayed: !created.created || undefined });
      }
  
++    /**
++     * ?stream=1: the run is made first (a refused one answers its usual status and problem), then the event stream opens
++     * and the answer's text goes out as it is written. A replay or a cached run has no deltas: its `run` comes at once.
++     */
++    async function createAndStream(req, res, body, waitMs) {
++        if (req.principal.kind === 'app') body = appBody(req.principal, body, req.body);
++        checkNamespace(req, String(body.workflow || ''));
++        const idem = body.idempotency_key || req.ov.idempotencyKey || undefined;
++        let opened = false;
++        const early = [];
++        const alive = () => !res.writableEnded && !res.destroyed;
++        const send = (event, data) => { if (alive()) res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); };
++        const onDelta = (d) => {
++            const text = d && d.delta && typeof d.delta.content === 'string' ? d.delta.content : '';
++            if (!text) return;
++            if (opened) send('delta', { text }); else early.push(text);
++        };
++        const created = await runs.create({ ...body, idempotency_key: idem }, req.principal, { trace: req.ov.traceId, requestId: req.ov.requestId, onDelta });
++        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', created.run.id);
++        if (typeof res.flushHeaders === 'function') res.flushHeaders();
++        opened = true;
++        if (early.length) send('delta', { text: early.splice(0).join('') });
++        try {
++            const run = await runs.explained(await runs.wait(created, waitMs));
++            send('run', { run, replayed: !created.created || undefined });
++        } catch (err) {
++            send('error', { code: (err && err.code) || 'internal', message: String((err && err.message) || 'error').slice(0, 500) });
++        }
++        if (alive()) res.end();
++    }
++
      r.post('/api/v1/runs', create, async (req, res) => {
          try {
              const wait = Math.max(0, Math.min(Number(req.query.wait) || 0, config.runs.maxWaitMs));
++            if (req.query.stream === '1') return await createAndStream(req, res, req.body || {}, wait || config.runs.maxWaitMs);
              await createAndRespond(req, res, req.body || {}, wait);
--        } catch (err) { sendError(res, err, req.ov, log); }
++        } catch (err) { if (!res.headersSent) sendError(res, err, req.ov, log); else if (!res.writableEnded) res.end(); }
      });
  
      r.get('/api/v1/runs', read, async (req, res) => {
ddiff --git a/test/runs-stream.test.js b/test/runs-stream.test.js
nnew file mode 100644
iindex 0000000..aca2362
---- /dev/null
++++ b/test/runs-stream.test.js
@@@ -0,0 +1,55 @@
++'use strict';
++// A run whose answer streams (server/api/runs.js, POST /api/v1/runs?stream=1): server-sent `delta` events carry the
++// answer's text as the provider writes it, and `run` the finished run, whose output is the joined deltas; the same
++// idempotency key again sends only its `run` (replayed); a refused run keeps its status and problem; without ?stream
++// the answer is JSON as before.
++const assert = require('assert');
++const { boot, token, suite, ALL } = require('./helpers');
++
++const t = suite('runs-stream');
++const service = token('chat', ALL);
++let h;
++const events = async (res) => (await res.text()).split('\n\n').filter(Boolean).map((block) => {
++    const ev = /^event: (\w+)$/m.exec(block);
++    const data = /^data: (.*)$/m.exec(block);
++    return { event: ev && ev[1], data: data ? JSON.parse(data[1]) : null };
++});
++const streamRun = (body, { tok = service, key = null } = {}) => fetch(`${h.base}/api/v1/runs?stream=1&wait=30000`, {
++    method: 'POST', headers: { Authorization: `Bearer ${tok}`, 'Content-Type': 'application/json', ...(key ? { 'Idempotency-Key': key } : {}) }, body: JSON.stringify(body),
++});
++const COMPLETE = { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: 'Stream a few words please' }] } };
++
++t.test('boot', async () => { h = await boot(); });
++
++t.test('the deltas as they come, then the run; joined they are its answer', async () => {
++    const res = await streamRun(COMPLETE, { key: 'stream-key-1' });
++    assert.strictEqual(res.status, 200);
++    assert.match(res.headers.get('content-type'), /^text\/event-stream/);
++    assert.match(String(res.headers.get('x-openvibe-run-id')), /^run_/);
++    const list = await events(res);
++    const deltas = list.filter((e) => e.event === 'delta');
++    const last = list[list.length - 1];
++    assert.ok(deltas.length >= 1, 'text came before the end');
++    assert.strictEqual(last.event, 'run');
++    assert.strictEqual(last.data.run.status, 'succeeded');
++    assert.strictEqual(deltas.map((e) => e.data.text).join(''), last.data.run.output.choices[0].message.content);
++    assert.strictEqual(last.data.run.id, res.headers.get('x-openvibe-run-id'));
++});
++
++t.test('the same key again: only its run, replayed', async () => {
++    const list = await events(await streamRun(COMPLETE, { key: 'stream-key-1' }));
++    assert.deepStrictEqual(list.map((e) => e.event), ['run']);
++    assert.strictEqual(list[0].data.replayed, true);
++});
++
++t.test('a refused run keeps its status and problem; without ?stream the answer is JSON', async () => {
++    const res = await streamRun({ workflow: 'no.such.workflow', input: {} });
++    assert.ok(res.status >= 400 && res.status < 500, String(res.status));
++    assert.match(res.headers.get('content-type'), /json/);
++    const plain = await fetch(`${h.base}/api/v1/runs?wait=30000`, { method: 'POST', headers: { Authorization: `Bearer ${service}`, 'Content-Type': 'application/json' }, body: JSON.stringify(COMPLETE) });
++    assert.match(plain.headers.get('content-type'), /json/);
++    assert.strictEqual((await plain.json()).run.status, 'succeeded');
++});
++
++t.test('close', async () => { await h.stop(); });
++t.run();