feat: show whether a recipient is actually reading (#162)

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<void> {
+    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<Record<string, number>> {
+    const counts: Record<string, number> = {};
+    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<number> {
+    assertName(recipient);
+    return (await this.backend.list(inboxPrefix(recipient))).filter((k) => k.endsWith('.json')).length;
+  }
+
   private async tryGetAgent(name: string): Promise<AgentRecord | null> {
     try {
       return decode<AgentRecord>(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;
 }

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 <to> [body]         Send a message (body from arg or stdin)
+  send <to> [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,17 +992,29 @@ async function cmdRegister(
 async function cmdAgents(bus: Bus, cfg: ResolvedConfig): Promise<number> {
   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<string, number>);
   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 {
     for (const a of list) {
       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<number> {
   const list = await bus.agents();
   const mySession = await sessionHash();
+  const unread = await bus.unreadCounts().catch(() => ({}) as Record<string, number>);
   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<string | null> {
+  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<number> {
   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 <id…> | 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;

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([]);
+  });
+});

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' };