Actor: a stream reader never loses its backlog to an event published mid-read (#40)

ddiff --git a/server/tasks/stream.js b/server/tasks/stream.js
iindex 873c750..e5b61d8 100644
---- a/server/tasks/stream.js
++++ b/server/tasks/stream.js
@@@ -5,6 +5,9 @@
   * every stored event after its Last-Event-ID (or ?after=), then each new one as the engine publishes it; the stream
   * closes after the `end` event. Events are stored before they are published (tasks/store.js appendEvent), and a reader
   * subscribes before reading its backlog, so nothing falls between the two; a seq it already has is never sent twice.
++ * Events published while the backlog is still being read wait until the backlog is written (sending one first would
++ * 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.
   */
  
@@@ -13,12 +16,18 @@ const HEARTBEAT_MS = 15_000;
  function createTaskStream({ log = console } = {}) {
      const subs = new Map();     // task id → Set<{ res, last }>
  
--    function write(sub, event) {
++    function send(sub, event) {
          if (event.seq <= sub.last) return;
          sub.last = event.seq;
          try { sub.res.write(`id: ${event.seq}\nevent: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`); } catch (err) { log.warn('[Actor] stream write failed:', err && err.message); }
      }
  
++    // A reader still reading its backlog keeps live events aside; serve() sends them, in seq order, after the backlog.
++    function write(sub, event) {
++        if (sub.ready) send(sub, event);
++        else sub.pending.push(event);
++    }
++
      function publish(taskId, event) {
          for (const sub of subs.get(taskId) || []) write(sub, event);
      }
@@@ -26,7 +35,10 @@ function createTaskStream({ log = console } = {}) {
      function end(taskId) {
          const set = subs.get(taskId);
          subs.delete(taskId);
--        for (const sub of set || []) { try { sub.res.end(); } catch { /* gone */ } }
++        for (const sub of set || []) {
++            if (!sub.ready) { sub.ended = true; continue; }   // serve() closes it once its backlog is out
++            try { sub.res.end(); } catch { /* gone */ }
++        }
      }
  
      /**
@@@ -37,7 +49,7 @@ function createTaskStream({ log = console } = {}) {
          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 };
++        const sub = { res, last: after, ready: false, pending: [], ended: false };
          let set = subs.get(taskId);
          if (!set) { set = new Set(); subs.set(taskId, set); }
          set.add(sub);
@@@ -49,8 +61,11 @@ function createTaskStream({ log = console } = {}) {
              if (s) { s.delete(sub); if (!s.size) subs.delete(taskId); }
          };
          req.on('close', leave);
--        for (const e of await readBacklog()) write(sub, e);
--        if (await isEnded()) { leave(); res.end(); }
++        for (const e of await readBacklog()) send(sub, e);
++        sub.ready = true;
++        for (const e of sub.pending.sort((a, b) => a.seq - b.seq)) send(sub, e);
++        sub.pending = [];
++        if (sub.ended || await isEnded()) { leave(); res.end(); }
      }
  
      const status = () => ({ streams: [...subs.values()].reduce((n, s) => n + s.size, 0) });
ddiff --git a/test/stream-race.test.js b/test/stream-race.test.js
nnew file mode 100644
iindex 0000000..31ddb84
---- /dev/null
++++ b/test/stream-race.test.js
@@@ -0,0 +1,65 @@
++'use strict';
++/**
++ * A task stream reader gets every event once, in seq order, even when the engine publishes (and ends the task) while
++ * the reader's backlog is still being read. Before the fix, an `end` published during the backlog read was sent first,
++ * so every older backlog event counted as already sent: the reader saw only seq 9 (CI on PostgreSQL, 2026-10-11).
++ */
++const assert = require('assert');
++const { EventEmitter } = require('events');
++const { createTaskStream } = require('../server/tasks/stream');
++
++function fakeRes() {
++    return {
++        chunks: [], ended: false,
++        status() { return this; }, set() { return this; }, flushHeaders() {},
++        write(c) { this.chunks.push(c); }, end() { this.ended = true; },
++        seqs() { return this.chunks.join('').split('\n').filter((l) => l.startsWith('id: ')).map((l) => Number(l.slice(4))); },
++    };
++}
++const ev = (seq, kind = 'output') => ({ task_id: 'tsk_1', seq, kind });
++
++(async () => {
++    const stream = createTaskStream({ log: { warn() {} } });
++
++    // 1. The task ends while the backlog read is in flight: the late `end` waits, the backlog goes first, then the stream closes.
++    {
++        let release;
++        const gate = new Promise((r) => { release = r; });
++        const res = fakeRes();
++        const serving = stream.serve(new EventEmitter(), res, {
++            taskId: 'tsk_1', after: 0,
++            readBacklog: async () => { await gate; return [1, 2, 3, 4, 5, 6, 7, 8].map((n) => ev(n)); },
++            isEnded: async () => true,
++        });
++        stream.publish('tsk_1', ev(9, 'end'));
++        stream.end('tsk_1');
++        assert.ok(!res.ended, 'the stream stays open until its backlog is out');
++        release();
++        await serving;
++        assert.deepStrictEqual(res.seqs(), [1, 2, 3, 4, 5, 6, 7, 8, 9]);
++        assert.ok(res.ended);
++    }
++
++    // 2. Events published during the backlog read that the backlog also holds are sent once; newer ones follow in order.
++    {
++        let release;
++        const gate = new Promise((r) => { release = r; });
++        const res = fakeRes();
++        const serving = stream.serve(new EventEmitter(), res, {
++            taskId: 'tsk_2', after: 2,
++            readBacklog: async () => { await gate; return [3, 4].map((n) => ({ ...ev(n), task_id: 'tsk_2' })); },
++            isEnded: async () => false,
++        });
++        stream.publish('tsk_2', { ...ev(5), task_id: 'tsk_2' });
++        stream.publish('tsk_2', { ...ev(4), task_id: 'tsk_2' });
++        release();
++        await serving;
++        stream.publish('tsk_2', { ...ev(6, 'end'), task_id: 'tsk_2' });
++        assert.deepStrictEqual(res.seqs(), [3, 4, 5, 6], 'after Last-Event-ID 2: each seq once, in order');
++        assert.ok(!res.ended, 'not ended until the engine ends it');
++        stream.end('tsk_2');
++        assert.ok(res.ended);
++        assert.deepStrictEqual(stream.status(), { streams: 0 });
++    }
++    console.log('stream-race: ok');
++})().catch((e) => { console.error(e); process.exit(1); });