From 28b26bf3e19c597778cdcd8788759a21e1aa0062 Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 11:03:33 +0300 Subject: [PATCH] feat: show whether a recipient is actually reading (#162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every `send` reported `sent → `, including the ones the recipient could never retrieve. In a session where several recipients were failing to read, the bus reported success 25 times while delivery was degraded, and no sender ever got a hint. `sent` has only ever meant QUEUED. The bus already knew enough to say more: pending inbox keys, and now a `lastRead` stamp written when an agent actually consumes its mailbox. So `send` warns when the evidence says nobody is picking the mail up — the recipient never registered under that name, or has unread stacking up with no consuming read in AGENTCOMM_STALE_READ_MS (6h). `agents` carries each agent's unread depth and how long ago it last read; `network` shows the same, plus a section for mail queued to names nobody ever registered, which a roster built from registrations alone cannot show at all. Counting is keys-only (one list, no bodies), and the read stamp rides commands agents run a handful of times a session. --- src/bus.ts | 47 +++++++++++++++++++++++++ src/cli.ts | 81 +++++++++++++++++++++++++++++++++++++++----- test/bus.test.ts | 40 ++++++++++++++++++++++ test/cli.e2e.test.ts | 39 +++++++++++++++++++++ 4 files changed, 198 insertions(+), 9 deletions(-) diff --git a/src/bus.ts b/src/bus.ts index e2b7f83b..285c2854 100644 --- a/src/bus.ts +++ b/src/bus.ts @@ -57,6 +57,7 @@ export class Bus { name, registeredAt: existing?.registeredAt ?? now, lastSeen: now, + ...(existing?.lastRead ? { lastRead: existing.lastRead } : {}), ...(session ? { session } : {}), ...(nextStatus ? { status: nextStatus, statusAuto: nextAuto, statusAt: nextStatusAt } : {}), }; @@ -82,6 +83,45 @@ export class Bus { return out; } + /** + * Record that `name` consumed its mailbox. A send reports success whether + * or not anyone is reading; this is the other half of that story, and it + * costs one write on a command agents run a handful of times a session. + * + * Only an EXISTING registration is stamped. Reading must not create one: + * registrations are never purged, so a one-off `inbox --as someone` would + * put a permanent ghost on the roster. An unregistered reader simply has + * no read history — which is exactly what `send` reports about it. + */ + async markRead(name: string): Promise { + assertName(name); + const existing = await this.tryGetAgent(name); + if (!existing) return; + const now = new Date().toISOString(); + await this.backend.put(agentKey(name), encode({ ...existing, lastSeen: now, lastRead: now })); + } + + /** + * Undelivered message count per recipient, from KEYS alone — one list, no + * bodies. Mailboxes with no registration count too: mail addressed to a + * name nobody is reading is precisely what needs surfacing. + */ + async unreadCounts(): Promise> { + const counts: Record = {}; + for (const key of await this.backend.list('inbox/')) { + if (!key.endsWith('.json')) continue; + const recipient = key.slice('inbox/'.length, key.indexOf('/', 'inbox/'.length)); + if (recipient) counts[recipient] = (counts[recipient] ?? 0) + 1; + } + return counts; + } + + /** Undelivered count for one recipient — the cheap pre-send check. */ + async unread(recipient: string): Promise { + assertName(recipient); + return (await this.backend.list(inboxPrefix(recipient))).filter((k) => k.endsWith('.json')).length; + } + private async tryGetAgent(name: string): Promise { try { return decode(await this.backend.get(agentKey(name))); @@ -345,4 +385,11 @@ export interface AgentRecord { statusAuto?: boolean; /** ISO 8601 time the status was set — bounds how long an explicit one stays sticky. */ statusAt?: string; + /** + * ISO 8601 time this agent last CONSUMED its mailbox (issue #162). `sent` + * only ever meant "queued"; this is what makes "and someone read it" + * answerable — a recipient with mail piling up and no read for hours is + * worth surfacing to whoever is sending it work. + */ + lastRead?: string; } diff --git a/src/cli.ts b/src/cli.ts index 469ab7d4..8db8a54f 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -58,10 +58,14 @@ Commands: task-status, telemetry. Stdin JSON in, hook JSON out, always exits 0 register Register/heartbeat the calling agent (--as) - agents List registered agents + agents List registered agents — with each one's unread + depth and how long ago it last consumed its mailbox network Situation report: who is on the bus and what they're doing (active/idle + recent activity) - send [body] Send a message (body from arg or stdin) + send [body] Send a message (body from arg or stdin). Warns when + the recipient is not reading: unread piling up, no + consuming read in AGENTCOMM_STALE_READ_MS (6h), or + no registration under that name at all broadcast [body] Send to every registered agent except yourself inbox Consume undelivered messages (archived under read/) — printed first, archived after, so an interrupted @@ -158,6 +162,8 @@ Env: stderr) AGENTCOMM_DAEMON=1|0 Default all commands through / away from the daemon AGENTCOMM_POLL_MS Daemon remote-poll interval (default 10000) + AGENTCOMM_STALE_READ_MS How long a recipient can go without consuming + before \`send\` flags it (default 6h) AGENTCOMM_MIRROR_HISTORY_MS How much archive/telemetry history the daemon keeps warm (default 7d). Older keys stay listable and readable; their bodies just load on demand, @@ -986,9 +992,19 @@ async function cmdRegister( async function cmdAgents(bus: Bus, cfg: ResolvedConfig): Promise { const list = await bus.agents(); const mySession = await sessionHash(); + // Who is actually CONSUMING (issue #162) — a roster of names says nothing + // about whether work routed to them is being picked up. + const unread = await bus.unreadCounts().catch(() => ({}) as Record); const isActive = (a: { lastSeen: string }) => Date.now() - Date.parse(a.lastSeen) < 10 * 60_000; if (cfg.json) { - emit(list.map((a) => ({ ...a, thisSession: a.session === mySession, active: isActive(a) }))); + emit( + list.map((a) => ({ + ...a, + unread: unread[a.name] ?? 0, + thisSession: a.session === mySession, + active: isActive(a), + })), + ); } else if (list.length === 0) { process.stdout.write('(no agents registered)\n'); } else { @@ -996,7 +1012,9 @@ async function cmdAgents(bus: Bus, cfg: ResolvedConfig): Promise { const mine = a.session === mySession ? ' (this session)' : ''; const live = isActive(a) ? ' · active' : ''; const doing = a.status ? ` — ${a.status}` : ''; - process.stdout.write(`${a.name}\tlast seen ${a.lastSeen}${live}${mine}${doing}\n`); + const waiting = unread[a.name] ? ` · ${unread[a.name]} unread` : ''; + const read = a.lastRead ? ` · read ${relTime(a.lastRead)}` : unread[a.name] ? ' · never read' : ''; + process.stdout.write(`${a.name}\tlast seen ${a.lastSeen}${live}${mine}${waiting}${read}${doing}\n`); } } return 0; @@ -1026,16 +1044,27 @@ async function cmdNetwork( ): Promise { const list = await bus.agents(); const mySession = await sessionHash(); + const unread = await bus.unreadCounts().catch(() => ({}) as Record); const isActive = (a: { lastSeen: string }) => Date.now() - Date.parse(a.lastSeen) < 10 * 60_000; const active = list.filter(isActive).sort((a, b) => b.lastSeen.localeCompare(a.lastSeen)); const idle = list.filter((a) => !isActive(a)).sort((a, b) => b.lastSeen.localeCompare(a.lastSeen)); const recent = await recentMessages(backend, 5); + // Mail addressed to a name that never registered: queued forever, and + // invisible on a roster built from registrations alone (issue #162). + const unclaimed = Object.entries(unread).filter(([name]) => !list.some((a) => a.name === name)); if (cfg.json) { + const withUnread = (a: (typeof list)[number], live: boolean) => ({ + ...a, + unread: unread[a.name] ?? 0, + thisSession: a.session === mySession, + active: live, + }); emit({ bus: cfg.backendUri, - active: active.map((a) => ({ ...a, thisSession: a.session === mySession, active: true })), - idle: idle.map((a) => ({ ...a, thisSession: a.session === mySession, active: false })), + active: active.map((a) => withUnread(a, true)), + idle: idle.map((a) => withUnread(a, false)), + unclaimed: unclaimed.map(([name, count]) => ({ name, unread: count })), recent, }); return 0; @@ -1044,7 +1073,8 @@ async function cmdNetwork( const line = (a: (typeof list)[number]): string => { const mine = a.session === mySession ? ' (you)' : ''; const doing = a.status ? a.status : '—'; - return ` ${(a.name + mine).padEnd(24)} ${doing.padEnd(42).slice(0, 42)} ${relTime(a.lastSeen)}`; + const waiting = unread[a.name] ? ` ${unread[a.name]} unread` : ''; + return ` ${(a.name + mine).padEnd(24)} ${doing.padEnd(42).slice(0, 42)} ${relTime(a.lastSeen)}${waiting}`; }; process.stdout.write(`bus ${cfg.backendUri}\n\n`); @@ -1058,6 +1088,13 @@ async function cmdNetwork( if (idle.length) { process.stdout.write(`idle (${idle.length})\n${idle.map(line).join('\n')}\n\n`); } + if (unclaimed.length) { + process.stdout.write( + `unread for nobody (no registration under that name)\n${unclaimed + .map(([name, count]) => ` ${name.padEnd(24)} ${count} message(s) queued`) + .join('\n')}\n\n`, + ); + } if (recent.length) { process.stdout.write('recent\n'); for (const m of recent) { @@ -1109,11 +1146,35 @@ async function cmdSend( } const body = rest.length > 1 ? rest.slice(1).join(' ') : await readStdin(); const msg = await bus.send({ from: me, to, body, subject, thread }); - if (cfg.json) emit(msg); - else process.stdout.write(`sent ${msg.id} → ${to}\n`); + // `sent` has only ever meant QUEUED. Say so when the evidence says nobody + // is picking it up (issue #162) — a recipient with mail stacking up and no + // read for hours is worth knowing about before you route work to it. + const delivery = await deliveryWarning(bus, to).catch(() => null); + if (cfg.json) emit({ ...msg, ...(delivery ? { warning: delivery } : {}) }); + else process.stdout.write(`sent ${msg.id} → ${to}\n${delivery ? `agentcomm: warning — ${delivery}\n` : ''}`); return 0; } +/** How stale a recipient's last read has to be before a send is worth flagging. */ +const STALE_READ_MS = Number(process.env.AGENTCOMM_STALE_READ_MS ?? 6 * 3600_000); + +/** Why this send may not reach anyone, or null when the recipient looks healthy. */ +async function deliveryWarning(bus: Bus, to: string): Promise { + const record = (await bus.agents()).find((a) => a.name === to); + const unread = await bus.unread(to); + if (!record) { + return `${to} has never registered on this bus — nothing is known to be reading that mailbox (${unread} message(s) waiting).`; + } + if (unread <= 1) return null; // the one we just sent, or an empty box: normal + const read = record.lastRead ? Date.now() - Date.parse(record.lastRead) : null; + if (read !== null && read < STALE_READ_MS) return null; + return ( + `${to} has ${unread} unread and ` + + (record.lastRead ? `has not consumed its mailbox since ${record.lastRead}` : 'has never consumed its mailbox') + + ` (last seen ${relTime(record.lastSeen)}).` + ); +} + async function cmdBroadcast( bus: Bus, cfg: ResolvedConfig, @@ -1137,6 +1198,7 @@ async function cmdBroadcast( async function cmdInbox(bus: Bus, cfg: ResolvedConfig): Promise { const me = await resolveAgent(cfg); await guardMailbox(me, 'inbox', 'consuming'); + await bus.markRead(me).catch(() => {}); // "someone is reading X" — see cmdSend (issue #162) await bus.inbox(me, { deliver: (messages) => printMessages(messages, cfg, me), onUnarchived: (keys) => @@ -1173,6 +1235,7 @@ async function cmdAck(bus: Bus, cfg: ResolvedConfig, all: boolean, ids: string[] fail('ack requires message ids (from `peek`/`inbox --json`) or --all: agentcomm ack | agentcomm ack --all'); } const result = await bus.ack(me, all ? 'all' : ids); + await bus.markRead(me).catch(() => {}); if (cfg.json) { await writeOut(JSON.stringify({ agent: me, ...result }, null, 2) + '\n'); return result.unknown.length > 0 ? 1 : 0; diff --git a/test/bus.test.ts b/test/bus.test.ts index 9d792c80..4b430aab 100644 --- a/test/bus.test.ts +++ b/test/bus.test.ts @@ -348,3 +348,43 @@ describe('ack clears mail already read (issue #160)', () => { expect(await bus.peek('bob')).toHaveLength(1); }); }); + +/** + * `sent` has only ever meant QUEUED (issue #162). These cover the two facts + * that make "and someone read it" answerable: a per-recipient unread depth, + * and when each agent last consumed its mailbox. + */ +describe('delivery visibility (issue #162)', () => { + it('counts unread per recipient, including mailboxes nobody registered', async () => { + const bus = new Bus(new LocalBackend(await mkTmp())); + await bus.register('bob'); + await bus.send({ from: 'alice', to: 'bob', body: '1' }); + await bus.send({ from: 'alice', to: 'bob', body: '2' }); + await bus.send({ from: 'alice', to: 'ghost', body: 'nobody home' }); + + expect(await bus.unreadCounts()).toEqual({ bob: 2, ghost: 1 }); + expect(await bus.unread('bob')).toBe(2); + expect(await bus.unread('nobody')).toBe(0); + }); + + it('markRead stamps lastRead, and a heartbeat preserves it', async () => { + const bus = new Bus(new LocalBackend(await mkTmp())); + await bus.register('bob', 'sess'); + expect((await bus.agents())[0]!.lastRead).toBeUndefined(); + + await bus.markRead('bob'); + const read = (await bus.agents())[0]!.lastRead; + expect(read).toBeTruthy(); + + await bus.register('bob', 'sess'); // heartbeat must not erase the read stamp + expect((await bus.agents())[0]!.lastRead).toBe(read); + }); + + it('markRead never conjures a registration — a one-off read leaves no ghost', async () => { + const bus = new Bus(new LocalBackend(await mkTmp())); + await bus.markRead('walk-in'); + // registrations are never purged; a `inbox --as someone` must not create + // a permanent roster entry + expect(await bus.agents()).toEqual([]); + }); +}); diff --git a/test/cli.e2e.test.ts b/test/cli.e2e.test.ts index 53d85bee..00a9ab8c 100644 --- a/test/cli.e2e.test.ts +++ b/test/cli.e2e.test.ts @@ -104,6 +104,45 @@ describe('CLI e2e (sqlite backend)', () => { expect(bare.stderr).toMatch(/ack requires message ids/); }); + it('send flags a recipient that is not reading; a consuming read clears the flag (issue #162)', async () => { + const db = `sqlite://${path.join(await mkTmp(), 'bus.db')}`; + + // never registered: nothing is known to be reading that mailbox + const cold = await run(['send', 'ghost', 'anyone there?', '--as', 'alice', '--backend', db]); + expect(cold.stdout).toMatch(/warning — ghost has never registered on this bus/); + + await run(['register', '--as', 'bob', '--backend', db]); + // first message to a registered agent is not suspicious on its own + const first = await run(['send', 'bob', 'one', '--as', 'alice', '--backend', db]); + expect(first.stdout).not.toMatch(/warning/); + + // mail stacking up with no consuming read ever = worth saying + const second = await run(['send', 'bob', 'two', '--as', 'alice', '--backend', db]); + expect(second.stdout).toMatch(/warning — bob has 2 unread and has never consumed its mailbox/); + + const roster = JSON.parse((await run(['agents', '--backend', db, '--json'])).stdout) as { + name: string; + unread: number; + lastRead?: string; + }[]; + expect(roster.find((a) => a.name === 'bob')).toMatchObject({ unread: 2 }); + expect(roster.find((a) => a.name === 'bob')!.lastRead).toBeUndefined(); + + // bob reads: the warning stops and the roster records the read + await run(['inbox', '--as', 'bob', '--backend', db, '--json']); + const after = await run(['send', 'bob', 'three', '--as', 'alice', '--backend', db]); + expect(after.stdout).not.toMatch(/warning/); + const read = (JSON.parse((await run(['agents', '--backend', db, '--json'])).stdout) as { name: string; lastRead?: string }[]) + .find((a) => a.name === 'bob')!.lastRead; + expect(read).toBeTruthy(); + + // and network surfaces mail queued for a name nobody registered + const net = JSON.parse((await run(['network', '--backend', db, '--json'])).stdout) as { + unclaimed: { name: string; unread: number }[]; + }; + expect(net.unclaimed).toEqual([{ name: 'ghost', unread: 1 }]); + }); + it('network reports active/idle agents, statuses, and recent activity', async () => { const db = `sqlite://${path.join(await mkTmp(), 'bus.db')}`; const env = { ...process.env, AGENTCOMM_SESSION: 'net-test', AGENTCOMM_NO_GIT_PROBE: '1' };