// 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: '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'], 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' }), []); });