bus: agent state per role, and the Console shows it (#1548)
Row 59 round 2, owner Filbert, candidate manifest 9acfac19 on 3d298920.
Approved by Darkwing (27591) and Dewey (27595); Sage's gate green.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
+29
-1
@@ -125,11 +125,39 @@ an operator inspect content and delivery evidence after an uncertain result.
|
||||
| `message.read` | `{id}`; recipient holder's acknowledgement; no new holder can acknowledge old holder's delivery. |
|
||||
| `event.emit` | `{kind:'action.allowed',body:{action:'routine',target?},subject?}`; routine observations only. |
|
||||
| `launch.revoke`, `launch.restore` | No args; outside-agent human CLI only. |
|
||||
| `inbox`, `tasks`, `agents` | No args; plain JSON projections scoped to capability's business. |
|
||||
| `inbox`, `tasks` | No args; plain JSON projections scoped to capability's business. |
|
||||
| `agents` | No args; each held role's latest claim row in the capability's business. Human and reader sessions also get `state`, `working` and `inputNeeded` (Agent state, below); a role session gets the claim rows only. |
|
||||
| `trail` | `{subject}`: task ref, decision UUID or message UUID. Linked records in broker write order. |
|
||||
| `decisions` | `{list}`: `roles` or `recent`; human and reader sessions only. Open decisions routed to a role, or the last 50 closed. |
|
||||
| `credentials` | No args; human and reader sessions only. `[{service,account,role,dateKind,date,state}]` for the capability's business, never a token, file path or variable name. |
|
||||
|
||||
### Agent state (row 59, #1548)
|
||||
|
||||
For a human or reader session, `agents` derives each held role's state from
|
||||
records the bus already keeps. Nothing new is written and no schema changed.
|
||||
|
||||
- `working` is `{messages, since}` or null: messages to the role that the
|
||||
current holder received and hasn't marked read, and when the oldest was
|
||||
received. The runner receives a batch, takes one turn per message, replies
|
||||
and marks it read, so working means a turn is under way or queued.
|
||||
- `inputNeeded` lists the role's open decisions routed to the human, gated
|
||||
first and then oldest, as the Inbox orders them. Each item is
|
||||
`{id, question, class, blocking, task_ref, at}`. Seen is still open.
|
||||
- `state` is `working` when `working` is set, else `waiting` when
|
||||
`inputNeeded` is nonempty, else `idle`. Working wins over waiting, as on
|
||||
the control board.
|
||||
|
||||
Why the bus and not a seat-to-role map: the business file's `holder` is a
|
||||
display name, and the claim's holder is a run, not a seat, so no committed
|
||||
data joins a seat to a role. The bus already has the claim, the deliveries
|
||||
and the decisions.
|
||||
|
||||
Limits: a message routed but not yet received shows idle. A reply that fails
|
||||
for a reason other than a gone session leaves its message unread, so the role
|
||||
stays working; `since` shows how long. No error or offline state is derived.
|
||||
A new holder doesn't inherit the previous holder's unread deliveries;
|
||||
decisions stay with the role. Unheld and released roles aren't listed.
|
||||
|
||||
Unknown keys, roles, references, actions and verbs refuse. If writing refusal
|
||||
evidence fails, the caller receives the static `storage-unavailable` code. `human` is the reserved
|
||||
recipient. Message classes are REQUEST, ASSIGNMENT, REVIEW-REQUEST, REVIEW-RESULT,
|
||||
|
||||
@@ -849,11 +849,14 @@ export class Broker {
|
||||
s.human || s.reader ? 'human' : s.role,
|
||||
)
|
||||
.map((d) => this.#decisionView(d));
|
||||
if (verb === 'agents')
|
||||
return this.#store.all(
|
||||
if (verb === 'agents') {
|
||||
const claims = this.#store.all(
|
||||
"SELECT * FROM role_claims c WHERE business=? AND op='claim' AND seq=(SELECT max(seq) FROM role_claims WHERE business=c.business AND role=c.role) ORDER BY role",
|
||||
s.business,
|
||||
);
|
||||
// A role session gets the claims only: another role's decisions aren't its to read.
|
||||
return s.human || s.reader ? claims.map((c) => this.#agentState(c)) : claims;
|
||||
}
|
||||
if (verb === 'tasks') return this.#taskRows(s.business);
|
||||
return this.#trail(s, identifier(a.subject));
|
||||
}
|
||||
@@ -904,6 +907,28 @@ export class Broker {
|
||||
fail('unknown-verb');
|
||||
}
|
||||
}
|
||||
// Row 59 (#1548): a held role's state for the Agents view, derived from what the bus
|
||||
// already records. Working: messages the current holder received and hasn't marked read;
|
||||
// the runner marks each one read after its turn. Input needed: the role's open decisions
|
||||
// routed to the human, in Inbox order. Working wins over waiting, as on the control board.
|
||||
#agentState(c) {
|
||||
const busy = this.#store.get(
|
||||
"SELECT count(*) AS messages, min(d.at) AS since FROM messages m JOIN deliveries d ON d.message=m.id AND d.op='delivered' WHERE m.business=? AND m.to_role=? AND d.holder_run=? AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read')",
|
||||
c.business,
|
||||
c.role,
|
||||
c.holder_run,
|
||||
);
|
||||
const inputNeeded = this.#store
|
||||
.all(
|
||||
`SELECT d.* FROM decisions d WHERE business=? AND raised_by_role=? AND route_to='human' AND NOT EXISTS(SELECT 1 FROM decision_events x WHERE x.decision=d.id AND x.op IN ${CLOSED}) ORDER BY CASE class WHEN 'gated' THEN 0 ELSE 1 END,seq`,
|
||||
c.business,
|
||||
c.role,
|
||||
)
|
||||
.map((d) => ({ id: d.id, question: d.question, class: d.class, blocking: !!d.blocking, task_ref: d.task_ref, at: d.at }));
|
||||
const working = busy.messages ? { messages: busy.messages, since: busy.since } : null;
|
||||
const state = working ? 'working' : inputNeeded.length ? 'waiting' : 'idle';
|
||||
return { ...c, state, working, inputNeeded };
|
||||
}
|
||||
#claimEnd(s, h, op, decision) {
|
||||
this.#store.run(
|
||||
'INSERT INTO role_claims(at,business,role,op,holder_run,harness,address,by,decision) VALUES(?,?,?,?,?,?,?,?,?)',
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
// Row 59 (#1548): `agents` gives human and reader sessions each held role's state,
|
||||
// derived from records the bus already keeps. A role session gets the claims only.
|
||||
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';
|
||||
|
||||
const options = [
|
||||
{ key: 'yes', text: 'Allow' },
|
||||
{ key: 'no', text: 'Decline' },
|
||||
];
|
||||
const roles = {
|
||||
pm: { authority: { withinRole: ['message.send', 'task.create'], crossRole: [] } },
|
||||
cto: { authority: { withinRole: ['message.send'], crossRole: ['task.scope.change'] } },
|
||||
coder: { authority: { withinRole: ['message.send'], crossRole: ['task.scope.change'] } },
|
||||
};
|
||||
const businesses = {
|
||||
demo: { id: 'demo', human: 'jason', arbiters: { technical: 'cto', delivery: 'pm' }, roles },
|
||||
other: { id: 'other', human: 'jason', arbiters: { technical: 'cto', delivery: 'pm' }, roles },
|
||||
};
|
||||
const CLAIM = ['seq', 'at', 'business', 'role', 'op', 'holder_run', 'harness', 'address', 'by', 'reason', 'decision'];
|
||||
function setup(t) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'bus-agents-'));
|
||||
const store = new Store(root);
|
||||
const b = new Broker({ store, businesses });
|
||||
t.after(() => {
|
||||
store.close();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
});
|
||||
const agent = (role, business = 'demo', run = `${business}-${role}-run`) => {
|
||||
const cap = b.bindLaunch({ business, role, run, harness: 'pi', address: run });
|
||||
b.request(cap, { verb: 'role.claim' });
|
||||
return cap;
|
||||
};
|
||||
const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
|
||||
const reader = b.bindReader({ business: 'demo' });
|
||||
return { b, store, agent, human, reader };
|
||||
}
|
||||
const call = (b, cap, verb, args = {}) => b.request(cap, { verb, args });
|
||||
const raise = (b, cap, action, extra = {}) =>
|
||||
call(b, cap, 'decision.raise', { action, question: `${action}?`, options, recommendation: 'no', blocking: false, ...extra });
|
||||
const byRole = (rows) => Object.fromEntries(rows.map((r) => [r.role, r]));
|
||||
const states = (b, cap) => Object.fromEntries(call(b, cap, 'agents').map((r) => [r.role, r.state]));
|
||||
const delivered = (store, id) => store.get("SELECT at FROM deliveries WHERE message=? AND op='delivered'", id).at;
|
||||
|
||||
test('agents: a held role with nothing on the bus is idle; a role session gets the claim rows unchanged', (t) => {
|
||||
const { b, agent, human, reader } = setup(t);
|
||||
const pm = agent('pm');
|
||||
agent('coder');
|
||||
const rows = call(b, human, 'agents');
|
||||
assert.deepEqual(rows.map((r) => [r.role, r.state, r.working, r.inputNeeded]), [
|
||||
['coder', 'idle', null, []],
|
||||
['pm', 'idle', null, []],
|
||||
]);
|
||||
for (const r of rows) assert.deepEqual(Object.keys(r), [...CLAIM, 'state', 'working', 'inputNeeded']);
|
||||
assert.deepEqual(call(b, reader, 'agents'), rows);
|
||||
const plain = call(b, pm, 'agents');
|
||||
for (const r of plain) assert.deepEqual(Object.keys(r), CLAIM);
|
||||
assert.deepEqual(plain.map((r) => ({ ...r })), rows.map((r) => Object.fromEntries(CLAIM.map((k) => [k, r[k]]))));
|
||||
});
|
||||
|
||||
test('agents: working counts what the holder received and has not read; routed alone is not working', (t) => {
|
||||
const { b, store, agent, human } = setup(t);
|
||||
const pm = agent('pm');
|
||||
const one = call(b, human, 'message.send', { to: 'pm', body: 'first' });
|
||||
assert.equal(states(b, human).pm, 'idle', 'routed, not received');
|
||||
assert.deepEqual(call(b, pm, 'message.receive').map((m) => m.id), [one.id]);
|
||||
const two = call(b, human, 'message.send', { to: 'pm', body: 'second' });
|
||||
let row = byRole(call(b, human, 'agents')).pm;
|
||||
assert.equal(row.state, 'working');
|
||||
assert.deepEqual(row.working, { messages: 1, since: delivered(store, one.id) });
|
||||
call(b, pm, 'message.receive');
|
||||
row = byRole(call(b, human, 'agents')).pm;
|
||||
assert.deepEqual(row.working, { messages: 2, since: delivered(store, one.id) });
|
||||
call(b, pm, 'message.read', { id: one.id });
|
||||
row = byRole(call(b, human, 'agents')).pm;
|
||||
assert.deepEqual(row.working, { messages: 1, since: delivered(store, two.id) });
|
||||
call(b, pm, 'message.read', { id: two.id });
|
||||
row = byRole(call(b, human, 'agents')).pm;
|
||||
assert.deepEqual([row.state, row.working], ['idle', null]);
|
||||
});
|
||||
|
||||
test("agents: input needed is the role's open decisions routed to the human, gated first, then oldest", (t) => {
|
||||
const { b, agent, human } = setup(t);
|
||||
const cto = agent('cto');
|
||||
const coder = agent('coder');
|
||||
agent('pm');
|
||||
// The arbiter raising a cross-role decision in its own domain routes it to the human.
|
||||
const own = raise(b, cto, 'task.scope.change', { domain: 'technical', target: 'vikunja:3/1' });
|
||||
const toCto = raise(b, coder, 'task.scope.change', { domain: 'technical', target: 'vikunja:3/2' });
|
||||
const first = raise(b, cto, 'deploy', { target: 'release-1', task_ref: 'vikunja:3/3', blocking: true });
|
||||
const second = raise(b, cto, 'deploy', { target: 'release-2' });
|
||||
const replaced = raise(b, cto, 'deploy', { target: 'release-3' });
|
||||
raise(b, cto, 'deploy', { target: 'release-3b', supersedes: replaced.id, question: 'Ship 3b?' });
|
||||
const resolved = raise(b, cto, 'deploy', { target: 'release-4' });
|
||||
call(b, human, 'decision.resolve', { id: resolved.id, choice: 'no' });
|
||||
call(b, human, 'decision.seen', { id: second.id });
|
||||
assert.deepEqual([own.route_to, own.class, toCto.route_to, first.route_to, first.class], ['human', 'cross-role', 'cto', 'human', 'gated']);
|
||||
const rows = byRole(call(b, human, 'agents'));
|
||||
assert.equal(rows.cto.state, 'waiting');
|
||||
assert.equal(rows.cto.working, null);
|
||||
const latest = call(b, human, 'inbox').find((d) => d.question === 'Ship 3b?');
|
||||
assert.deepEqual(rows.cto.inputNeeded, [
|
||||
{ id: first.id, question: 'deploy?', class: 'gated', blocking: true, task_ref: 'vikunja:3/3', at: first.at },
|
||||
{ id: second.id, question: 'deploy?', class: 'gated', blocking: false, task_ref: null, at: second.at },
|
||||
{ id: latest.id, question: 'Ship 3b?', class: 'gated', blocking: false, task_ref: null, at: latest.at },
|
||||
{ id: own.id, question: 'task.scope.change?', class: 'cross-role', blocking: false, task_ref: null, at: own.at },
|
||||
]);
|
||||
// The inbox shows the same open decisions; a decision routed to a role is no one's Input needed.
|
||||
assert.deepEqual(call(b, human, 'inbox').map((d) => d.id), rows.cto.inputNeeded.map((d) => d.id));
|
||||
assert.deepEqual([rows.coder.state, rows.coder.inputNeeded, rows.pm.state], ['idle', [], 'idle']);
|
||||
for (const d of rows.cto.inputNeeded) call(b, human, 'decision.resolve', { id: d.id, choice: 'no' });
|
||||
assert.deepEqual([byRole(call(b, human, 'agents')).cto.state, byRole(call(b, human, 'agents')).cto.inputNeeded], ['idle', []]);
|
||||
});
|
||||
|
||||
test('agents: working wins over waiting, and Input needed still shows', (t) => {
|
||||
const { b, agent, human } = setup(t);
|
||||
const cto = agent('cto');
|
||||
const asked = raise(b, cto, 'deploy', { target: 'release-1' });
|
||||
call(b, human, 'message.send', { to: 'cto', body: 'status?' });
|
||||
call(b, cto, 'message.receive');
|
||||
const row = byRole(call(b, human, 'agents')).cto;
|
||||
assert.equal(row.state, 'working');
|
||||
assert.equal(row.working.messages, 1);
|
||||
assert.deepEqual(row.inputNeeded.map((d) => d.id), [asked.id]);
|
||||
});
|
||||
|
||||
test("agents: a new holder doesn't inherit what the last one left unread; decisions stay with the role", (t) => {
|
||||
const { b, agent, human } = setup(t);
|
||||
const first = agent('cto', 'demo', 'cto-one');
|
||||
const asked = raise(b, first, 'deploy', { target: 'release-1' });
|
||||
call(b, human, 'message.send', { to: 'cto', body: 'left unread' });
|
||||
call(b, first, 'message.receive');
|
||||
assert.equal(states(b, human).cto, 'working');
|
||||
call(b, first, 'role.release');
|
||||
assert.deepEqual(call(b, human, 'agents'), [], 'a released role is not listed');
|
||||
agent('cto', 'demo', 'cto-two');
|
||||
const row = byRole(call(b, human, 'agents')).cto;
|
||||
assert.deepEqual([row.holder_run, row.state, row.working], ['cto-two', 'waiting', null]);
|
||||
assert.deepEqual(row.inputNeeded.map((d) => d.id), [asked.id]);
|
||||
});
|
||||
|
||||
test('agents: state reads only its own business, even for a run name both use', (t) => {
|
||||
const { b, agent, human } = setup(t);
|
||||
// A run name is unique per business only, so the holder alone doesn't separate them.
|
||||
agent('cto', 'demo', 'cto-run');
|
||||
const away = agent('cto', 'other', 'cto-run');
|
||||
raise(b, away, 'deploy', { target: 'release-1' });
|
||||
const otherHuman = b.bindHuman({ business: 'other', human: 'jason', via: 'cli', outsideAgent: true });
|
||||
call(b, otherHuman, 'message.send', { to: 'cto', body: 'elsewhere' });
|
||||
call(b, away, 'message.receive');
|
||||
const here = byRole(call(b, human, 'agents')).cto;
|
||||
assert.deepEqual([here.state, here.working, here.inputNeeded], ['idle', null, []]);
|
||||
const there = byRole(call(b, otherHuman, 'agents')).cto;
|
||||
assert.deepEqual([there.state, there.working.messages, there.inputNeeded.length], ['working', 1, 1]);
|
||||
});
|
||||
Reference in New Issue
Block a user