stack/packages/bus/tests/limit-stop.test.mjs

// Row 125 (#1619, REQ-LIMIT-4): the broker's end-of-turn stop. After a turn's usage end the broker
// asks the limit checks (a stub here) whether the run stops; the turn's RESULT and read receipt
// land first, then the run's next receive or state ends its claim with by='limit-check' and fails
// not-holder, which the runner treats as the session gone.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';

const shipped = name => JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
const businesses = () => ({ demo: {
  id: 'demo', human: 'jason', arbiters: { technical: 'coder', delivery: 'coder' },
  roles: {
    coder: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } },
    assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
  },
  definitions: { coder: {}, assistant: { messaging: shipped('assistant').messaging } },
  projects: { stack: { root: '/nonexistent' } },
} });
const HIT = { path: ['tokens', 'instances', 'coder'], limit: 100, count: 120, unknownTurns: 1 };

function setup(t, answers) {
  const root = mkdtempSync(join(tmpdir(), 'limit-stop-'));
  const store = new Store(root);
  t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
  const asked = [];
  const limits = { mustStop: async (business, subject) => {
    asked.push({ business, subject });
    const next = answers.shift();
    if (next instanceof Error) throw next;
    return next ?? { stop: false };
  } };
  const broker = new Broker({ store, businesses: businesses(), limits, usage: { collect: async () => ({ recorded: true }), view: () => ({}) } });
  const human = broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  return { store, broker, human, asked };
}

// One turn the way the runner takes it: receive, usage begin and end, RESULT, read receipt.
async function turn(broker, human, cap, send) {
  const { id } = send();
  broker.request(cap, { verb: 'message.receive' });
  await broker.request(cap, { verb: 'usage.begin', args: { turn: id } });
  await broker.request(cap, { verb: 'usage.end', args: { turn: id } });
  return id;
}

test('REQ-LIMIT-4 a role run at its limit stops at the end of its turn: RESULT and read land first, the claim is revoked by limit-check', async t => {
  const { store, broker, human, asked } = setup(t, [{ stop: false }, { stop: true, check: 'token-limit', hit: HIT }]);
  const coder = broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run1', harness: 'pi', provider: 'anthropic', model: 'opus' });
  broker.request(coder, { verb: 'role.claim' });
  const send = () => broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'work' } });
  // Under the limit: the next receive goes on as usual.
  await turn(broker, human, coder, send);
  assert.deepEqual(broker.request(coder, { verb: 'message.receive' }), []);
  // At the limit: the turn's reply and read receipt still land.
  const id = await turn(broker, human, coder, send);
  assert.deepEqual(asked, [1, 2].map(() => ({ business: 'demo', subject: { instance: 'coder', provider: 'anthropic', model: 'opus', account: null } })));
  const reply = broker.request(coder, { verb: 'message.send', args: { to: 'human', body: 'done', class: 'RESULT', in_reply_to: id } });
  assert.ok(reply.id);
  broker.request(coder, { verb: 'message.read', args: { id } });
  assert.ok(store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", id));
  assert.equal(store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, 'claim');
  // Then the next receive ends the claim and refuses not-holder.
  assert.throws(() => broker.request(coder, { verb: 'message.receive' }), { code: 'not-holder' });
  const row = store.get("SELECT op, by, reason, holder_run FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1");
  assert.deepEqual({ ...row }, { op: 'revoke', by: 'limit-check', holder_run: 'run1',
    reason: 'limit: token-limit tokens.instances.coder limit 100 count 120' });
  assert.match(row.reason, /^limit: /);
  assert.ok(store.get("SELECT 1 FROM events WHERE kind='action.refused' AND actor_run='run1' AND json_extract(body,'$.code')='not-holder'"));
  // A revoked role can't be claimed again by the run.
  assert.throws(() => broker.request(coder, { verb: 'role.claim' }));
  // The launcher's end record reads the limit from the claim row.
  const ended = broker.endLaunch({ business: 'demo', run: 'run1', reason: 'exited', exitCode: 22 });
  assert.deepEqual(ended, { released: false, limit: 'limit: token-limit tokens.instances.coder limit 100 count 120' });
  const event = JSON.parse(store.get("SELECT body FROM events WHERE kind='session.ended' AND subject='run1'").body);
  assert.equal(event.limit, ended.limit);
});

test('REQ-LIMIT-4 the stop also lands on state; a run with no stop ends without a limit', async t => {
  const { store, broker, human } = setup(t, [{ stop: true, check: 'token-limit', hit: HIT }]);
  const coder = broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run1', harness: 'pi' });
  broker.request(coder, { verb: 'role.claim' });
  await turn(broker, human, coder, () => broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'work' } }));
  assert.throws(() => broker.request(coder, { verb: 'state' }), { code: 'not-holder' });
  assert.equal(store.get("SELECT by FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").by, 'limit-check');
  // The stop is spent: the run is refused as any non-holder is, without a second claim row.
  const rows = store.get("SELECT count(*) n FROM role_claims WHERE role='coder'").n;
  assert.throws(() => broker.request(coder, { verb: 'message.receive' }), { code: 'not-holder' });
  assert.equal(store.get("SELECT count(*) n FROM role_claims WHERE role='coder'").n, rows);
  // Another run that was never stopped ends with no limit.
  const other = setup(t, []);
  const run2 = other.broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run2', harness: 'pi' });
  other.broker.request(run2, { verb: 'role.claim' });
  await turn(other.broker, other.human, run2, () => other.broker.request(other.human, { verb: 'message.send', args: { to: 'coder', body: 'w' } }));
  assert.deepEqual(other.broker.endLaunch({ business: 'demo', run: 'run2', reason: 'stopped', exitCode: 0 }), { released: true });
  assert.equal(JSON.parse(other.store.get("SELECT body FROM events WHERE kind='session.ended' AND subject='run2'").body).limit, undefined);
});

test('REQ-LIMIT-4 fail closed: a mustStop throw or a malformed answer stops the run, naming the bus and the code', async t => {
  for (const [answer, reason] of [
    [new Error('boom'), 'limit: bus failed check-threw'],
    [{ stop: 'yes' }, 'limit: bus failed malformed-answer'],
    [undefined, null], // no answer means { stop: false } from the stub
    [{ stop: true, check: 'token-limit', code: 'not-loaded' }, 'limit: token-limit failed not-loaded'],
    [{ stop: true, check: null, code: 'organization-unreadable' }, 'limit: bus failed organization-unreadable'],
    // Row 333 R3 (#1831) ruling 2: the no-model-key refusal stops the run by its own code.
    [{ stop: true, check: null, code: 'key-quota-no-model-key' }, 'limit: bus failed key-quota-no-model-key'],
  ]) {
    const { store, broker, human } = setup(t, answer === undefined ? [] : [answer]);
    const coder = broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run1', harness: 'pi' });
    broker.request(coder, { verb: 'role.claim' });
    await turn(broker, human, coder, () => broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'work' } }));
    if (reason === null) {
      assert.deepEqual(broker.request(coder, { verb: 'message.receive' }), []);
      continue;
    }
    assert.throws(() => broker.request(coder, { verb: 'message.receive' }), { code: 'not-holder' }, reason);
    assert.equal(store.get("SELECT reason FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").reason, reason);
  }
});

test('REQ-LIMIT-4 a thread run at its limit is released with the reason, and its end record names the limit', async t => {
  const { store, broker, human } = setup(t, [{ stop: true, check: 'token-limit', hit: { ...HIT, path: ['tokens', 'instances', 'assistant'] } }]);
  const project = broker.request(human, { verb: 'project.create', args: { name: 'stack work', repo: 'stack' } }).id;
  const thread = broker.request(human, { verb: 'thread.create', args: { project, name: 'alpha' } }).id;
  const a = broker.bindLaunch({ business: 'demo', role: 'assistant', run: 'a-run', harness: 'pi', thread });
  broker.request(a, { verb: 'role.claim' });
  await turn(broker, human, a, () => broker.request(human, { verb: 'thread.send', args: { thread, body: 'look' } }));
  assert.throws(() => broker.request(a, { verb: 'message.receive' }), { code: 'not-holder' });
  const row = store.get('SELECT op, by, reason FROM thread_claims WHERE thread=? ORDER BY seq DESC LIMIT 1', thread);
  const reason = 'limit: token-limit tokens.instances.assistant limit 100 count 120';
  assert.deepEqual({ ...row }, { op: 'release', by: 'limit-check', reason });
  assert.deepEqual(broker.endLaunch({ business: 'demo', run: 'a-run', reason: 'exited', exitCode: 22 }), { released: false, limit: reason });
  // The thread is free for a new run.
  const a2 = broker.bindLaunch({ business: 'demo', role: 'assistant', run: 'a2-run', harness: 'pi', thread });
  assert.deepEqual(broker.request(a2, { verb: 'role.claim' }), { role: 'assistant', run: 'a2-run', thread });
});

test('REQ-LIMIT-4 a broker with no limit checks configured never stops a run', async t => {
  const root = mkdtempSync(join(tmpdir(), 'limit-stop-'));
  const store = new Store(root);
  t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
  const broker = new Broker({ store, businesses: businesses(), usage: { collect: async () => ({ recorded: true }), view: () => ({}) } });
  const human = broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const coder = broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run1', harness: 'pi' });
  broker.request(coder, { verb: 'role.claim' });
  await turn(broker, human, coder, () => broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'work' } }));
  assert.deepEqual(broker.request(coder, { verb: 'message.receive' }), []);
});