Base da6f1166, r1 patch applied clean with --index, +/- and context lines identical.
Bus suite 127/127, mutants 15 of 15 killed.
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
177 lines
12 KiB
Diff
177 lines
12 KiB
Diff
diff --git a/packages/bus/README.md b/packages/bus/README.md
|
|
index 793694bae..8f5eec4aa 100644
|
|
--- a/packages/bus/README.md
|
|
+++ b/packages/bus/README.md
|
|
@@ -129,7 +129,9 @@ committed. Callers must inspect its evidence before deciding what to do; they
|
|
must not resend automatically. `message.receive` records delivery when the
|
|
broker hands the batch to the connection, not when the harness acts on it.
|
|
A lost response leaves an uncertain delivery, never automatic replay to a new
|
|
-holder. `message.read` is an explicit acknowledgement. `trail(messageId)` lets
|
|
+holder. A holder that releases a role without reading what it received leaves
|
|
+those messages stranded: no later holder receives or reads them, and `agents`
|
|
+lists them under `stranded` (Agent state, below). `message.read` is an explicit acknowledgement. `trail(messageId)` lets
|
|
an operator inspect content and delivery evidence after an uncertain result.
|
|
|
|
| Verb | Arguments and result |
|
|
@@ -150,7 +152,7 @@ an operator inspect content and delivery evidence after an uncertain result.
|
|
| `launch.revoke`, `launch.restore` | No args; outside-agent human CLI only. |
|
|
| `inbox` | No args; plain JSON projection scoped to capability's business. |
|
|
| `tasks` | Optional args, scoped to capability's business (row 89, #1583). None: every row of the tasks projection (below). `{task_ref}` alone: that one row, or `task-not-found`. Any of `{done,assignee,label,requirement,limit,after}`: `{tasks,total,next}`, the short form (below). An unknown arg or bad value refuses `invalid-request`. |
|
|
-| `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. |
|
|
+| `agents` | No args; each held role's latest claim row in the capability's business. Human and reader sessions also get `state`, `working`, `inputNeeded` and `stranded` (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. |
|
|
@@ -172,6 +174,12 @@ records the bus already keeps. Nothing new is written and no schema changed.
|
|
- `state` is `working` when `working` is set, else `waiting` when
|
|
`inputNeeded` is nonempty, else `idle`. Working wins over waiting, as on
|
|
the control board.
|
|
+- `stranded` lists messages to the role that an earlier holder was delivered
|
|
+ and never read, as `{id, from, run, at}`: the sender (`human` for the CLI),
|
|
+ the run it was delivered to, and when. Oldest first, at most 50, `[]` when
|
|
+ none. The bus neither redelivers, routes nor marks these read, and the
|
|
+ released run can no longer read them, so they stay listed. A person who
|
|
+ still needs one sends it again as a new message.
|
|
|
|
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
|
|
@@ -182,7 +190,8 @@ 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.
|
|
+decisions stay with the role. Unheld and released roles aren't listed, so a
|
|
+message stranded for a role nobody holds shows once a run claims the role.
|
|
|
|
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
|
|
diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs
|
|
index 5e874ad3e..4ae20e103 100644
|
|
--- a/packages/bus/src/broker.mjs
|
|
+++ b/packages/bus/src/broker.mjs
|
|
@@ -43,6 +43,8 @@ const CLOSED = "('resolved','withdrawn','expired')";
|
|
export const RECENT = 50;
|
|
// How many replies `replies` returns, oldest first.
|
|
export const REPLIES = 50;
|
|
+// Row 80 (#1574): how many stranded messages `agents` lists for each role, oldest first.
|
|
+export const STRANDED = 50;
|
|
// Row 89 (#1583): the arguments `tasks` takes, and its page size by default and at most.
|
|
const TASK_ARGS = ['done', 'assignee', 'label', 'requirement', 'limit', 'after', 'task_ref'];
|
|
export const TASK_PAGE = 50;
|
|
@@ -1213,7 +1215,16 @@ export class Broker {
|
|
.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 };
|
|
+ // Row 80 (#1574): messages to the role that an earlier holder was delivered and never read.
|
|
+ // The run they went to no longer holds the role. A delivered row already exists, so routing
|
|
+ // and receive skip them. Listing them writes nothing.
|
|
+ const stranded = this.#store.all(
|
|
+ `SELECT m.id, m.from_role AS "from", d.holder_run AS run, d.at 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 IS NOT ? AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read') ORDER BY m.seq LIMIT ${STRANDED}`,
|
|
+ c.business,
|
|
+ c.role,
|
|
+ c.holder_run,
|
|
+ ).map((m) => ({ id: m.id, from: m.from, run: m.run, at: m.at }));
|
|
+ return { ...c, state, working, inputNeeded, stranded };
|
|
}
|
|
#claimEnd(s, h, op, decision) {
|
|
this.#store.run(
|
|
diff --git a/packages/bus/tests/agents.test.mjs b/packages/bus/tests/agents.test.mjs
|
|
index 23b5126b3..d92933bd6 100644
|
|
--- a/packages/bus/tests/agents.test.mjs
|
|
+++ b/packages/bus/tests/agents.test.mjs
|
|
@@ -55,7 +55,8 @@ test('agents: a held role with nothing on the bus is idle; a role session gets t
|
|
['coder', 'idle', null, []],
|
|
['pm', 'idle', null, []],
|
|
]);
|
|
- for (const r of rows) assert.deepEqual(Object.keys(r), [...CLAIM, 'state', 'working', 'inputNeeded']);
|
|
+ for (const r of rows) assert.deepEqual(Object.keys(r), [...CLAIM, 'state', 'working', 'inputNeeded', 'stranded']);
|
|
+ assert.deepEqual(rows.map((r) => r.stranded), [[], []]);
|
|
assert.deepEqual(call(b, reader, 'agents'), rows);
|
|
const plain = call(b, pm, 'agents');
|
|
for (const r of plain) assert.deepEqual(Object.keys(r), CLAIM);
|
|
@@ -157,3 +158,83 @@ test('agents: state reads only its own business, even for a run name both use',
|
|
const there = byRole(call(b, otherHuman, 'agents')).cto;
|
|
assert.deepEqual([there.state, there.working.messages, there.inputNeeded.length], ['working', 1, 1]);
|
|
});
|
|
+
|
|
+// Row 80 (#1574): a message delivered to a run that released the role unread is stranded for the
|
|
+// role. No later holder receives or reads it, and listing it writes no delivery row.
|
|
+test('agents: a message the holder released unread is stranded; the next holder gets nothing and cannot read it', (t) => {
|
|
+ const { b, store, agent, human } = setup(t);
|
|
+ const first = agent('pm', 'demo', 'pm-one');
|
|
+ const sent = call(b, human, 'message.send', { to: 'pm', body: 'left behind' });
|
|
+ assert.deepEqual(call(b, first, 'message.receive').map((m) => m.id), [sent.id]);
|
|
+ const before = store.all('SELECT * FROM deliveries');
|
|
+ call(b, first, 'role.release');
|
|
+ assert.deepEqual(call(b, human, 'agents'), [], 'a released role is not listed');
|
|
+ const second = agent('pm', 'demo', 'pm-two');
|
|
+ const row = byRole(call(b, human, 'agents')).pm;
|
|
+ assert.deepEqual(row.stranded, [{ id: sent.id, from: 'human', run: 'pm-one', at: delivered(store, sent.id) }]);
|
|
+ assert.deepEqual(call(b, second, 'message.receive'), []);
|
|
+ assert.throws(() => call(b, second, 'message.read', { id: sent.id }), { code: 'not-recipient' });
|
|
+ assert.throws(() => call(b, first, 'message.read', { id: sent.id }), { code: 'not-holder' });
|
|
+ assert.deepEqual(byRole(call(b, human, 'agents')).pm.stranded, row.stranded);
|
|
+ assert.deepEqual(store.all('SELECT * FROM deliveries'), before, 'the claim and the listing write no delivery row');
|
|
+});
|
|
+
|
|
+test('agents: a message the holder read is not stranded; one the current holder has not read is not listed', (t) => {
|
|
+ const { b, agent, human } = setup(t);
|
|
+ const first = agent('pm', 'demo', 'pm-one');
|
|
+ const read = call(b, human, 'message.send', { to: 'pm', body: 'read first' });
|
|
+ const unread = call(b, human, 'message.send', { to: 'pm', body: 'still held' });
|
|
+ call(b, first, 'message.receive');
|
|
+ call(b, first, 'message.read', { id: read.id });
|
|
+ assert.deepEqual(byRole(call(b, human, 'agents')).pm.stranded, [], 'the current holder still holds the unread one');
|
|
+ call(b, first, 'role.release');
|
|
+ agent('pm', 'demo', 'pm-two');
|
|
+ assert.deepEqual(byRole(call(b, human, 'agents')).pm.stranded.map((m) => m.id), [unread.id]);
|
|
+});
|
|
+
|
|
+test('agents: stranded is oldest first and capped at 50 for a role', (t) => {
|
|
+ const { b, agent, human } = setup(t);
|
|
+ const first = agent('pm', 'demo', 'pm-one');
|
|
+ const ids = Array.from({ length: 60 }, (_, i) => call(b, human, 'message.send', { to: 'pm', body: `m${i}` }).id);
|
|
+ assert.equal(call(b, first, 'message.receive').length, 60);
|
|
+ call(b, first, 'role.release');
|
|
+ agent('pm', 'demo', 'pm-two');
|
|
+ const stranded = byRole(call(b, human, 'agents')).pm.stranded;
|
|
+ assert.equal(stranded.length, 50);
|
|
+ assert.deepEqual(stranded.map((m) => m.id), ids.slice(0, 50));
|
|
+ assert.ok(stranded.every((m) => m.run === 'pm-one' && m.from === 'human'));
|
|
+});
|
|
+
|
|
+test('agents: stranded is per role and per business', (t) => {
|
|
+ const { b, agent, human } = setup(t);
|
|
+ const cto = agent('cto', 'demo', 'cto-one');
|
|
+ const pm = agent('pm', 'demo', 'pm-one');
|
|
+ const toCto = call(b, human, 'message.send', { to: 'cto', body: 'for cto' });
|
|
+ const toPm = call(b, human, 'message.send', { to: 'pm', body: 'for pm' });
|
|
+ call(b, cto, 'message.receive');
|
|
+ call(b, pm, 'message.receive');
|
|
+ call(b, cto, 'role.release');
|
|
+ call(b, pm, 'role.release');
|
|
+ agent('cto', 'demo', 'cto-two');
|
|
+ agent('pm', 'demo', 'pm-two');
|
|
+ const rows = byRole(call(b, human, 'agents'));
|
|
+ assert.deepEqual(rows.cto.stranded.map((m) => m.id), [toCto.id]);
|
|
+ assert.deepEqual(rows.pm.stranded.map((m) => m.id), [toPm.id]);
|
|
+ // The same role and run name in another business stays there.
|
|
+ const otherHuman = b.bindHuman({ business: 'other', human: 'jason', via: 'cli', outsideAgent: true });
|
|
+ const otherPm = agent('pm', 'other', 'pm-one');
|
|
+ const elsewhere = call(b, otherHuman, 'message.send', { to: 'pm', body: 'elsewhere' });
|
|
+ call(b, otherPm, 'message.receive');
|
|
+ call(b, otherPm, 'role.release');
|
|
+ agent('pm', 'other', 'pm-two');
|
|
+ assert.deepEqual(byRole(call(b, human, 'agents')).pm.stranded.map((m) => m.id), [toPm.id]);
|
|
+ assert.deepEqual(byRole(call(b, otherHuman, 'agents')).pm.stranded.map((m) => m.id), [elsewhere.id]);
|
|
+});
|
|
+
|
|
+// Row 80 (#1574), Sage ruling: a message routed to a held role and not yet received is not stranded.
|
|
+test('agents: a message routed to a held role and not yet received is not stranded', (t) => {
|
|
+ const { b, agent, human } = setup(t);
|
|
+ agent('pm', 'demo', 'pm-one');
|
|
+ call(b, human, 'message.send', { to: 'pm', body: 'routed, not received' });
|
|
+ assert.deepEqual(byRole(call(b, human, 'agents')).pm.stranded, []);
|
|
+});
|