stack/packages/bus/tests/processing.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 { Socket } from 'node:net';
import { Broker } from '../src/broker.mjs';
import { Store } from '../src/store.mjs';
import { serve } from '../src/server.mjs';
import { Client } from '../src/client.mjs';
const state = await import('../../processing-control/src/state.mjs').catch(e => {
  if (e.code === 'ERR_MODULE_NOT_FOUND') return {};
  throw e;
});
const businesses = { demo: { id: 'demo', human: 'jason', vars: { 'bus.tester': true },
  arbiters: { technical: 'chief', delivery: 'chief' }, roles: {chief: { definition: 'chief-of-staff', authority: { withinRole: ['decision.resolve.technical', 'decision.resolve.delivery'], crossRole: [] } },
    coder: { definition: 'coder', authority: { withinRole: ['message.send', 'record.write'], crossRole: [] } },
  } } };
function fixture(t, extra = {}) {
  assert.equal(typeof state.initializeControl, 'function');
  const root = mkdtempSync(join(tmpdir(), 'processing-bus-'));
  state.initializeControl(root);
  const store = new Store(root);
  const processing = {
    read: () => state.readControl(root),
    async halt(actor) { return state.transition(root, { ...actor, halted: true }); },
    async resume(actor) { return state.transition(root, { ...actor, halted: false }); },
  };
  const b = new Broker({ store, businesses, processing, ...extra });
  const cap = b.bindLaunch({ business: 'demo', role: 'coder', run: 'run', harness: 'pi' });
  b.request(cap, { verb: 'role.claim' });
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const reader = b.bindReader({ business: 'demo' });
  const tester = b.bindTester({ business: 'demo' });
  t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
  return { root, b, store, cap, human, reader, tester, processing,
    call: (who, verb, args = {}) => b.request(who, { verb, args }) };
}

// Dropping #human before either effect would allow a role/tester to clear the switch.
for (const verb of ['processing.halt', 'processing.resume']) test(`${verb} refuses agent reader and tester before any transition`, async t => {
  const f = fixture(t), before = state.readControl(f.root);
  for (const cap of [f.cap, f.reader, f.tester]) {
    await assert.rejects(async () => f.call(cap, verb, { generation: before.generation }), e => ['human-required', 'read-only', 'tester-refused'].includes(e.code));
  }
  assert.deepEqual(state.readControl(f.root), before);
});

test('human halt and clear each have one event with verified actor; launch.restore cannot clear', async t => {
  const f = fixture(t);
  const halted = await f.call(f.human, 'processing.halt');
  assert.equal(halted.halted, true);
  await f.call(f.human, 'processing.halt');
  f.call(f.human, 'launch.restore');
  assert.equal(f.call(f.reader, 'processing.status').halted, true);
  const resumed = await f.call(f.human, 'processing.resume', { generation: halted.generation });
  assert.equal(resumed.halted, false);
  const events = f.store.all("SELECT body FROM events WHERE kind='action.allowed' AND json_extract(body,'$.action') IN ('processing.halt','processing.resume')").map(r => JSON.parse(r.body));
  assert.deepEqual(events.map(e => [e.action, e.human]), [['processing.halt', 'jason'], ['processing.resume', 'jason']]);
  assert.equal(new Set(events.map(e => e.operation)).size, 2);
});

test('all ordinary work verbs refuse before dispatch while pure reads remain available', async t => {
  const f = fixture(t); await f.call(f.human, 'processing.halt');
  for (const verb of ['role.claim', 'role.revoke', 'decision.raise', 'decision.resolve', 'message.send', 'message.receive', 'message.read', 'staff.request', 'review.request', 'review.verdict', 'event.emit', 'organization.set', 'record.write', 'usage.begin', 'unknown.future.work']) {
    await assert.rejects(async () => f.call(f.cap, verb), { code: 'processing-halted' }, verb);
  }
  assert.equal(f.call(f.cap, 'state').launch.state, 'allowed');
  assert.ok(Array.isArray(f.call(f.reader, 'agents')));
  assert.equal(f.call(f.reader, 'processing.status').halted, true);
  // A human can still queue intent and manage settings under their existing permissions.
  assert.doesNotThrow(() => f.call(f.human, 'message.send', { to: 'coder', body: 'wait for resume' }));
  assert.doesNotThrow(() => f.call(f.cap, 'role.release'));
});

test('task web authorization and bind paths refuse before external work', async t => {
  const f = fixture(t); await f.call(f.human, 'processing.halt');
  let effects = 0;
  await assert.rejects(f.b.requestTask(f.cap, { verb: 'task.get', args: {} }, () => { effects++; }), { code: 'processing-halted' });
  await assert.rejects(f.b.requestWeb(f.cap, { verb: 'web.fetch', args: { url: 'https://example.com' } }, { handle() { effects++; } }), { code: 'processing-halted' });
  assert.throws(() => f.b.authorize(f.cap, 'role.launch', { target: 'coder' }), { code: 'processing-halted' });
  assert.throws(() => f.b.bindLaunch({ business: 'demo', role: 'coder', run: 'later', harness: 'pi' }), { code: 'processing-halted' });
  assert.equal(effects, 0);
});

for (const action of ['processing.halt', 'processing.resume', 'processing.stop']) test(`recordEvent cannot forge ${action}`, t => {
  const f = fixture(t);
  assert.throws(() => f.b.recordEvent(f.cap, { kind: 'action.allowed', body: { action, human: 'jason' } }), { code: 'reserved-event' });
  assert.equal(f.store.get("SELECT count(*) n FROM events WHERE kind='action.allowed'").n, 0);
});

for (const own of [false, true]) test(`recordTask ${own ? 'self' : 'poll'} cannot forge a processing action through its kinds filter`, t => {
  const f = fixture(t);
  assert.throws(() => f.b.recordTask({ business: 'demo', ...(own ? { cap: f.cap } : {}), events: [
    { kind: 'action.allowed', body: { action: 'processing.halt', human: 'jason' } },
  ] }), { code: 'reserved-event' });
});

test('spec read remains a read but cannot release new write work after an await', async t => {
  let finish;
  const ready = new Promise(resolve => { finish = resolve; });
  const f = fixture(t, { project: { async writeRecord() { await ready; return { path: 'docs/a.md', commit: 'a'.repeat(40) }; } } });
  const writing = f.call(f.cap, 'record.write', { project: 'stack', path: 'docs/a.md', record: '# A' });
  await f.call(f.human, 'processing.halt'); finish();
  // An already committed external effect must remain outcome-unknown, not a fake rollback.
  await assert.rejects(writing, { code: 'outcome-unknown' });
});

test('the real socket awaits halt and resume beyond the ordinary idle deadline', async t => {
  const f = fixture(t), path = join(f.root, 'control.sock');
  // Only this owned server socket uses a manual idle clock. Client sockets and
  // data/authentication/dispatch stay real; host scheduling cannot spend this clock.
  let now = 0;
  const deadlines = new Map(), originalTimeout = Socket.prototype.setTimeout;
  t.mock.method(Socket.prototype, 'setTimeout', function(ms, callback) {
    if (this.server?.address() !== path) return originalTimeout.call(this, ms, callback);
    if (callback) this.on('timeout', callback);
    if (ms) deadlines.set(this, now + ms); else deadlines.delete(this);
    this.once('close', () => deadlines.delete(this));
    return this;
  });
  const advance = ms => {
    now += ms;
    for (const [socket, deadline] of deadlines) if (deadline <= now && !socket.destroyed) socket.emit('timeout');
  };
  const hold = method => {
    let entered, release;
    const ready = new Promise(resolve => { entered = resolve; });
    const resumed = new Promise(resolve => { release = resolve; });
    const original = f.processing[method];
    f.processing[method] = async args => { entered(); await resumed; return original(args); };
    return { ready, release };
  };
  const halt = hold('halt'), resume = hold('resume');
  let peers = 0;
  const server = await serve({ broker: f.b, path, timeout: 5,
    // Row 196 added asynchronous kernel-peer discovery before the dispatch timeout.
    // This test isolates the post-auth await; peer proof is exercised in its owning files.
    peerReader: { close: async () => {}, read: async () => { peers++; return { pid: process.pid, uid: process.getuid(), gid: process.getgid() }; } },
    authenticateHuman: () => f.b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }) });
  try {
    const client = new Client({ path, human: { fixture: true }, timeout: 1000 });
    const pendingHalt = client.call('processing.halt');
    await halt.ready;
    advance(30);
    halt.release();
    const halted = await pendingHalt;
    assert.equal(halted.halted, true);
    const pendingResume = client.call('processing.resume', { generation: halted.generation });
    await resume.ready;
    advance(30);
    resume.release();
    assert.equal((await pendingResume).halted, false);
    assert.equal((await new Client({ path, cap: f.reader }).call('processing.status')).halted, false);
    assert.equal(peers, 2);
  } finally { halt.release(); resume.release(); await server.close(); }
});

test('a malformed processing reader never silently admits agent work', t => {
  const f = fixture(t);
  for (const answer of [null, {}, { halted: false }, { version: 1, generation: 'bad', halted: false }, { halted: 'false' }]) {
    f.processing.read = () => answer;
    assert.throws(() => f.call(f.cap, 'message.send', { to: 'human', body: 'no' }), { code: 'processing-state-invalid' });
  }
});

test('halt refuses the newly landed host executor path before credentials or effects', async t => {
  const f = fixture(t); await f.call(f.human, 'processing.halt');
  let called = false;
  await assert.rejects(f.b.requestHost(f.cap, { verb: 'host.run', args: {} }, {
    busy: () => { called = true; return false; }, run: async () => { called = true; },
  }), e => e.code === 'processing-halted');
  assert.equal(called, false);
});