stack/packages/bus/tests/task-read.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
// Row 58 (#1547): each 'tasks' row carries the task's priority and requirement, read from the
// append-only task_snapshots and task.created rows, for every session that may read tasks.
const roles = { pm: { authority: { withinRole: ['message.send', 'task.create'], crossRole: [] } } };
const businesses = {
  demo: { id: 'demo', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
  other: { id: 'other', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
};
const at = '2026-10-08T12:00:00.000Z';
// task_current keeps a poll only when it was read after the last self snapshot, whose `at` is the
// broker's clock, so a read after a write must be later than now.
const later = new Date(Date.now() + 3600e3).toISOString();
const digest = 'c'.repeat(64);
const fields = (extra = {}) => ({ bucket: 11, done: false, priority: 0, ...extra });
const poll = (task_ref, f, extra = {}) => ({ task_ref, updated: at, digest, fields: f, via: 'board', read_at: at, ...extra });
const gone = (task_ref, f) => poll(task_ref, f, { via: 'task', read_at: later });
const tasks = (b, cap) => b.request(cap, { verb: 'tasks', args: {} });
const pick = (rows) => rows.map((r) => [r.task_ref, r.priority, r.requirement]);
function open(root) {
  const store = new Store(root);
  const b = new Broker({ store, businesses });
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const agent = (business) => {
    const cap = b.bindLaunch({ business, role: 'pm', run: `${business}-pm`, harness: 'pi' });
    b.request(cap, { verb: 'role.claim' });
    return cap;
  };
  return { store, b, human, reader: b.bindReader({ business: 'demo' }), agent };
}
function setup(t) {
  const root = mkdtempSync(join(tmpdir(), 'bus-task-read-'));
  const bus = { root, ...open(root) };
  t.after(() => {
    bus.store.close();
    rmSync(root, { recursive: true, force: true });
  });
  return bus;
}
// The pm creates a task the way packages/tasks does: a self snapshot and task.created.
function create(b, cap, human, business, task_ref, requirement, f) {
  const { request } = b.request(human, { verb: 'message.send', args: { to: 'pm', body: 'please' } });
  b.recordTask({
    cap,
    business,
    snapshots: [{ task_ref, updated: at, digest, fields: f }],
    events: [{ kind: 'task.created', subject: task_ref, body: { request, requirement } }],
  });
}
// Six demo tasks: created by the stack (3/7), changed outside it since (3/8), never seen by the
// stack (3/9), created then gone (3/10), created then moved to project 4 (3/11 -> 4/11), and a
// second task.created for 4/11's id.
function seed(bus) {
  const { b, human, agent } = bus;
  const pm = agent('demo');
  create(b, pm, human, 'demo', 'vikunja:3/7', 'REQ-SLICE-1', fields({ priority: 3 }));
  create(b, pm, human, 'demo', 'vikunja:3/8', 'REQ-SLICE-2', fields({ priority: 1 }));
  create(b, pm, human, 'demo', 'vikunja:3/10', 'REQ-SLICE-3', fields({ priority: 4 }));
  create(b, pm, human, 'demo', 'vikunja:3/11', 'REQ-MOVE-1', fields({ priority: 2 }));
  b.recordTask({
    business: 'demo',
    snapshots: [
      poll('vikunja:3/8', fields({ priority: 5 }), { read_at: later }),
      poll('vikunja:3/9', fields()),
      gone('vikunja:3/10', { gone: 'no-access' }),
      gone('vikunja:3/11', { gone: 'moved', project: 4 }),
      poll('vikunja:4/11', fields({ priority: 2, bucket: 21 }), { via: 'reconcile', read_at: later }),
    ],
  });
  create(b, pm, human, 'demo', 'vikunja:4/11', 'REQ-MOVE-2', fields({ priority: 2, bucket: 21 }));
  // The same task id in another business has its own requirement.
  const { request } = b.request(b.bindHuman({ business: 'other', human: 'jason', via: 'cli', outsideAgent: true }), {
    verb: 'message.send',
    args: { to: 'pm', body: 'please' },
  });
  b.recordTask({
    cap: agent('other'),
    business: 'other',
    snapshots: [{ task_ref: 'vikunja:3/7', updated: at, digest, fields: fields({ priority: 1 }) }],
    events: [{ kind: 'task.created', subject: 'vikunja:3/7', body: { request, requirement: 'REQ-OTHER-1' } }],
  });
  return pm;
}
const EXPECTED = [
  ['vikunja:3/10', null, 'REQ-SLICE-3'],
  ['vikunja:3/11', null, 'REQ-MOVE-1'],
  ['vikunja:3/7', 3, 'REQ-SLICE-1'],
  ['vikunja:3/8', 5, 'REQ-SLICE-2'],
  ['vikunja:3/9', 0, null],
  ['vikunja:4/11', 2, 'REQ-MOVE-1'],
];

test('tasks rows carry priority and requirement for every session that reads tasks', (t) => {
  const bus = setup(t);
  const pm = seed(bus);
  for (const cap of [bus.human, bus.reader, pm]) assert.deepEqual(pick(tasks(bus.b, cap)), EXPECTED);
  // The priority is the current row's: a later poll wins over the stack's write (3/8), and Vikunja's
  // 0 stays 0, not null (3/9). A tombstone has no priority (3/10, 3/11).
  // The requirement follows the task id: 4/11 keeps the one 3/11 was created with, not the second
  // task.created, and a task the stack never created has none (3/9).
  const other = bus.b.bindReader({ business: 'other' });
  assert.deepEqual(pick(tasks(bus.b, other)), [['vikunja:3/7', 1, 'REQ-OTHER-1']]);
  // Nothing else in the row changes: it is task_current's row with two keys added.
  const current = bus.b.taskView('demo', 'current');
  assert.deepEqual(
    tasks(bus.b, bus.reader).map(({ priority, requirement, ...r }) => ({ ...r })),
    current.map((r) => ({ ...r })),
  );
  assert.equal('priority' in current[0] || 'requirement' in current[0], false);
});

test('the read model rebuilds from the ledger: a reopened bus and each trail give the same values', (t) => {
  const bus = setup(t);
  seed(bus);
  const before = tasks(bus.b, bus.reader);
  bus.store.close();
  Object.assign(bus, open(bus.root));
  const after = tasks(bus.b, bus.reader);
  assert.deepEqual(after, before);
  assert.deepEqual(pick(after), EXPECTED);
  // Each value is what the append-only rows say: the priority of the current snapshot, and the
  // requirement of the first task.created for the task's id, read through trail.
  for (const row of after) {
    const trail = bus.b.request(bus.reader, { verb: 'trail', args: { subject: row.task_ref } });
    const snap = trail.find((r) => r.table === 'task_snapshots' && r.seq === row.seq);
    assert.equal(snap.fields.priority ?? null, row.priority, row.task_ref);
    const id = row.task_ref.split('/')[1];
    const created = bus.store
      .all("SELECT subject, body FROM events WHERE business='demo' AND kind='task.created' ORDER BY seq")
      .find((e) => e.subject.split('/')[1] === id);
    assert.equal(created ? JSON.parse(created.body).requirement : null, row.requirement, row.task_ref);
  }
  // A read writes nothing.
  const count = () => bus.store.get('SELECT (SELECT count(*) FROM events) + (SELECT count(*) FROM task_snapshots) n').n;
  const n = count();
  tasks(bus.b, bus.human);
  tasks(bus.b, bus.reader);
  assert.equal(count(), n);
});

test('an empty business and one with snapshots but no task.created read cleanly', (t) => {
  const bus = setup(t);
  assert.deepEqual(tasks(bus.b, bus.reader), []);
  bus.b.recordTask({ business: 'demo', snapshots: [poll('vikunja:5/1', { bucket: 0, done: false })] });
  assert.deepEqual(pick(tasks(bus.b, bus.reader)), [['vikunja:5/1', null, null]]);
});