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

src/backends/git.ts

@@ -244,19 +244,26 @@ export class GitBackend implements Backend, Batchable, Claimable, Snapshottable
   }
 
   /**
-   * One fetch, then every body read from local objects at that tip — the
-   * daemon's warm path. Key-by-key `get` would pay a fetch per key.
+   * One fetch, then bodies read from local objects at that tip — the daemon's
+   * warm path. Key-by-key `get` would pay a fetch per key. Listing names is
+   * cheap (`ls-tree`); reading blobs is not, so `opts.bodies` picks which
+   * ones are worth the batch (issue #167).
    */
-  async snapshot(prefix: string): Promise<Map<string, Buffer>> {
+  async snapshot(
+    prefix: string,
+    opts?: { bodies?: (key: string) => boolean },
+  ): Promise<{ keys: string[]; bodies: Map<string, Buffer> }> {
     const out = new Map<string, Buffer>();
     const tip = await this.tip();
-    if (tip === null) return out;
+    if (tip === null) return { keys: [], bodies: out };
     const full = this.k(prefix);
-    const names = (await this.git(['ls-tree', '-r', '--name-only', '-z', tip]))
+    const all = (await this.git(['ls-tree', '-r', '--name-only', '-z', tip]))
       .toString('utf8')
       .split('\0')
       .filter((p) => p.length > 0 && p.startsWith(full) && p.startsWith(this.keyPrefix));
-    if (names.length === 0) return out;
+    const keys = all.map((p) => p.slice(this.keyPrefix.length));
+    const names = opts?.bodies ? all.filter((p) => opts.bodies!(p.slice(this.keyPrefix.length))) : all;
+    if (names.length === 0) return { keys, bodies: out };
     const batch = await this.git(['cat-file', '--batch'], {
       input: Buffer.from(names.map((p) => `${tip}:${p}`).join('\n') + '\n'),
     });
@@ -272,7 +279,7 @@ export class GitBackend implements Backend, Batchable, Claimable, Snapshottable
       out.set(p.slice(this.keyPrefix.length), Buffer.from(batch.subarray(off, off + size)));
       off += size + 1; // body + trailing newline
     }
-    return out;
+    return { keys, bodies: out };
   }
 
   async list(prefix: string): Promise<string[]> {

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)));
@@ -175,6 +215,29 @@ export class Bus {
     return out;
   }
 
+  /**
+   * Clear mail that has already been read (issue #160). `peek` shows messages
+   * without consuming and `inbox` consumes what it shows — with only those
+   * two, an agent that read its mail the non-destructive way could never mark
+   * it read, so the unread count (and the stop guard behind it) stayed stuck
+   * at N forever. Acking works off the KEYS: no body is re-fetched, and ids
+   * that are not pending are reported rather than silently ignored.
+   */
+  async ack(
+    recipient: string,
+    ids: string[] | 'all',
+  ): Promise<{ acked: string[]; unknown: string[]; failed: string[] }> {
+    assertName(recipient);
+    const keys = (await this.backend.list(inboxPrefix(recipient))).filter((k) => k.endsWith('.json'));
+    const byId = new Map(keys.map((key) => [messageIdFromKey(key), key] as const));
+    const wanted = ids === 'all' ? [...byId.keys()] : ids;
+    const unknown = wanted.filter((id) => !byId.has(id));
+    const targets = wanted.filter((id) => byId.has(id));
+    const failedKeys = await this.archive(targets.map((id) => byId.get(id)!));
+    const failed = targets.filter((id) => failedKeys.includes(byId.get(id)!));
+    return { acked: targets.filter((id) => !failed.includes(id)), unknown, failed };
+  }
+
   /**
    * Archive (don't hard-delete) inbox keys under read/, preserving the audit
    * trail. Returns the keys that could NOT be archived — a raced consumer, or
@@ -264,6 +327,10 @@ function inboxKey(recipient: string, seq: string, id: string): string {
 function readKeyFromInboxKey(inboxKey: string): string {
   return 'read/' + inboxKey.slice('inbox/'.length);
 }
+/** The message id a key carries: inbox/<to>/<seq>_<id>.json — no read needed. */
+function messageIdFromKey(key: string): string {
+  return key.slice(key.lastIndexOf('_') + 1).replace(/\.json$/, '');
+}
 
 // ── sequence generation ─────────────────────────────────────────────────────
 
@@ -318,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;
 }