From 8e0fbca420a3161ce98c5e0f3910f500daabebba Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Tue, 13 Oct 2026 11:01:50 -0500 Subject: [PATCH] bus: list messages stranded by a holder change in agents, row 80 (#1574) A holder that releases a role without reading what it was delivered leaves those messages where no later holder receives or reads them. The agents view now lists them for human and reader sessions under `stranded`, as {id, from, run, at}, oldest first, at most 50. The bus doesn't redeliver, route or mark them read; a person who still needs one sends it again. Round 2 is a mechanical rebase of round 1 onto row 93. Reviewed by Filbert (28206, approve); Sage gate record 28207. Gate on 7496251c: 14 node suites and 9 scripts green. Co-Authored-By: Claude Opus 5.5 --- packages/bus/README.md | 15 ++++-- packages/bus/src/broker.mjs | 13 ++++- packages/bus/tests/agents.test.mjs | 83 +++++++++++++++++++++++++++++- 3 files changed, 106 insertions(+), 5 deletions(-) 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, []); +});