diff --git a/src/backends/git.ts b/src/backends/git.ts
index 410258a9..91e65bd5 100644
--- a/src/backends/git.ts
+++ b/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[]> {
diff --git a/src/bus.ts b/src/bus.ts
index 0bf444c9..e2b7f83b 100644
--- a/src/bus.ts
+++ b/src/bus.ts
@@ -175,6 +175,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 +287,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 ─────────────────────────────────────────────────────
 
diff --git a/src/cli.ts b/src/cli.ts
index a57ca04a..469ab7d4 100644
--- a/src/cli.ts
+++ b/src/cli.ts
@@ -8,7 +8,7 @@ import { parseArgs, resolveConfig, type ParsedFlags, type ResolvedConfig } from
 import type { Backend, Message } from './types.js';
 import { fileURLToPath } from 'node:url';
 import { deriveIdentity, pinnedSession, sessionHash } from './identity.js';
-import { recordAlias } from './session.js';
+import { leaseMailbox, recordAlias, releaseMailbox } from './session.js';
 import { INSTALL_COMMAND } from './update-check.js';
 import {
   EVENTS_PREFIX,
@@ -31,6 +31,7 @@ Agent quickstart (if you are an AI agent):
   agentcomm inbox --json                              # consume instructions that may be waiting
   agentcomm network                                   # who else is here, what they're doing
   agentcomm send <to> "<body>"  ·  agentcomm wait     # coordinate — reply on the sender's --thread
+  agentcomm peek  ·  agentcomm ack --all              # read without consuming, then clear what you handled
   Re-check your inbox before reporting your task done.
   Full etiquette: agentcomm conventions · bus semantics: agentcomm describe
 
@@ -62,8 +63,13 @@ Commands:
                            they're doing (active/idle + recent activity)
   send <to> [body]         Send a message (body from arg or stdin)
   broadcast [body]         Send to every registered agent except yourself
-  inbox                    Consume undelivered messages (archived under read/)
+  inbox                    Consume undelivered messages (archived under read/) —
+                           printed first, archived after, so an interrupted
+                           read costs a duplicate, never a lost message
   peek                     Show undelivered messages without consuming
+  ack <id…> | --all        Clear mail you already read some other way (peek, a
+                           digest): archives it without re-fetching. Exit 1
+                           when an id is not pending for you
   wait                     Block until a message arrives (exit 0) or timeout (exit 2)
   claim                    Atomically dequeue one message from --queue (SQL backends only)
   emit                     Record a telemetry event (--type, --name, --ref,
@@ -130,6 +136,7 @@ Flags:
   --check                  install: report drift instead of writing (exit 1 when
                            the wiring is missing or older than this CLI)
   --uninstall              install: remove the wiring this command wrote
+  --all                    ack: clear every message pending for you
   --type <text>            emit/events: the event type (skill-ran, skill-outcome, …)
   --name <text>            emit/events: what the event is about (a skill/tool name)
   --ref <text>             emit/events: correlation handle (branch, PR#, run id)
@@ -151,6 +158,10 @@ Env:
                              stderr)
   AGENTCOMM_DAEMON=1|0       Default all commands through / away from the daemon
   AGENTCOMM_POLL_MS          Daemon remote-poll interval (default 10000)
+  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,
+                             so a poll costs what is hot, not all history
   AGENTCOMM_BACKEND_PLUGINS  comma/whitespace-separated module specifiers to
                              import before resolving --backend, so a
                              third-party package can register a new URI
@@ -296,6 +307,8 @@ async function main(argv: string[]): Promise<number> {
         return await cmdInbox(bus, cfg);
       case 'peek':
         return await cmdPeek(bus, cfg);
+      case 'ack':
+        return await cmdAck(bus, cfg, flags.all, positional.slice(1));
       case 'wait':
         return await cmdWait(bus, cfg, flags.timeout ?? 30000);
       case 'claim':
@@ -313,10 +326,35 @@ async function main(argv: string[]): Promise<number> {
         return 1;
     }
   } finally {
+    if (leasedAlias) await releaseMailbox(leasedAlias).catch(() => {});
     await backend.close?.();
   }
 }
 
+let leasedAlias: string | null = null;
+
+/**
+ * Announce that this process is acting as `me`, and say so when ANOTHER live
+ * process of this session is doing the same (issue #161). A subagent inherits
+ * its parent's identity, so `agentcomm inbox` from one drains the mailbox its
+ * parent is waiting on — silently, because the alias looks right. Identity
+ * stays per-session by design (a session is a mailbox); the fix is that
+ * sharing one stops being invisible.
+ */
+async function guardMailbox(me: string, command: string, kind: 'consuming' | 'status' | 'quiet'): Promise<void> {
+  const others = await leaseMailbox(me, command).catch(() => [] as Awaited<ReturnType<typeof leaseMailbox>>);
+  leasedAlias = me;
+  if (kind === 'quiet' || others.length === 0) return;
+  const who = others.map((o) => `pid ${o.pid} (\`agentcomm ${o.command}\`, since ${o.since})`).join(', ');
+  process.stderr.write(
+    kind === 'consuming'
+      ? `agentcomm: WARNING — another live process of this session is acting as "${me}": ${who}. ` +
+          `Reads CONSUME, so this takes mail addressed to it. If you are a subagent, use a mailbox of your own: --as ${me}-<role>.\n`
+      : `agentcomm: WARNING — another live process of this session is acting as "${me}": ${who}. ` +
+          `Your status replaces its line on the shared roster; register under your own --as if you are a separate actor.\n`,
+  );
+}
+
 // ── commands ────────────────────────────────────────────────────────────────
 
 const CHANNEL_SECURITY_NOTE =
@@ -686,14 +724,18 @@ This repo has a message bus for AI agents. When working here:
   (active/idle agents, their statuses, recent activity).
 - Coordinate with other agents via \`send\`/\`wait\` (subjects: task, ack,
   done, question, status; reply on the sender's --thread).
-- Always check your inbox before reporting work done.
+- Always check your inbox before reporting work done. \`inbox\` consumes;
+  \`peek\` shows without consuming and \`agentcomm ack --all\` clears what you
+  have handled, so mail you read another way stops counting as unread.
 - Stuck? Declare it: \`agentcomm register --status "blocked: <what you
   need>"\` — other agents' digests will recruit help. If a digest shows
   someone else blocked and you KNOW the answer, send it without asking
   the user; otherwise stay on your task.
 - If your harness has subagents, prefer a background listener subagent for
   \`wait\`/inbox management (one actor per mailbox — it owns the alias or
-  uses \`--as <you>-bus\`); keep quick sends inline.
+  uses \`--as <you>-bus\`); keep quick sends inline. A subagent derives the
+  SAME alias as its parent, so a bare \`inbox\` there drains the parent's
+  mail; the CLI warns when two live processes share a mailbox — heed it.
 `;
 
 async function cmdInit(bus: Bus, cfg: ResolvedConfig, requestedHarness?: string): Promise<number> {
@@ -934,6 +976,7 @@ async function cmdRegister(
   statusAuto?: boolean,
 ): Promise<number> {
   const me = await resolveAgent(cfg);
+  await guardMailbox(me, 'register', status !== undefined ? 'status' : 'quiet');
   const record = await registerWithCollisionCheck(bus, me, status, statusAuto);
   if (cfg.json) emit(record);
   else process.stdout.write(`registered ${record.name}\n`);
@@ -1093,6 +1136,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.inbox(me, {
     deliver: (messages) => printMessages(messages, cfg, me),
     onUnarchived: (keys) =>
@@ -1105,13 +1149,53 @@ async function cmdInbox(bus: Bus, cfg: ResolvedConfig): Promise<number> {
 
 async function cmdPeek(bus: Bus, cfg: ResolvedConfig): Promise<number> {
   const me = await resolveAgent(cfg);
+  await guardMailbox(me, 'peek', 'quiet'); // reading is harmless; hold the lease so a concurrent consumer sees us
   const messages = await bus.peek(me);
   await printMessages(messages, cfg, me);
+  // peek leaves the mail unread by design; say how to clear it once acted on,
+  // or the unread count (and the stop guard) never drops (issue #160).
+  if (!cfg.json && messages.length > 0) {
+    await writeOut(`— ${messages.length} still unread; clear what you have handled: agentcomm ack --all\n`);
+  }
   return 0;
 }
 
+/**
+ * Clear mail that was read some other way — `peek`, a digest, a delivery this
+ * process could not consume. Read and clear used to be the same operation, so
+ * an agent that had read and answered every message still faced a stop guard
+ * demanding it read them (issue #160).
+ */
+async function cmdAck(bus: Bus, cfg: ResolvedConfig, all: boolean, ids: string[]): Promise<number> {
+  const me = await resolveAgent(cfg);
+  await guardMailbox(me, 'ack', 'consuming');
+  if (!all && ids.length === 0) {
+    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);
+  if (cfg.json) {
+    await writeOut(JSON.stringify({ agent: me, ...result }, null, 2) + '\n');
+    return result.unknown.length > 0 ? 1 : 0;
+  }
+  await writeOut(
+    [
+      `acked ${result.acked.length} message(s) for ${me}`,
+      ...(result.failed.length ? [`could not archive: ${result.failed.join(', ')} — still pending`] : []),
+      ...(result.unknown.length
+        ? [`not pending for ${me}: ${result.unknown.join(', ')} (already acked, or addressed to another alias)`]
+        : []),
+      '',
+    ].join('\n'),
+  );
+  return result.unknown.length > 0 ? 1 : 0;
+}
+
 async function cmdWait(bus: Bus, cfg: ResolvedConfig, timeoutMs: number): Promise<number> {
   const me = await resolveAgent(cfg);
+  // wait is non-consuming, but it HOLDS the mailbox for its whole timeout —
+  // the listener pattern. Leasing here is what lets a concurrent `inbox`
+  // report that it is about to drain someone's mail (issue #161).
+  await guardMailbox(me, 'wait', 'quiet');
   const messages = await bus.wait(me, timeoutMs);
   if (messages.length === 0) {
     if (cfg.json) emit([]);
@@ -1124,6 +1208,7 @@ async function cmdWait(bus: Bus, cfg: ResolvedConfig, timeoutMs: number): Promis
 
 async function cmdClaim(bus: Bus, cfg: ResolvedConfig, queue: string | undefined): Promise<number> {
   const me = await resolveAgent(cfg);
+  await guardMailbox(me, 'claim', 'quiet'); // a shared queue is MEANT to have many claimers
   if (!queue) {
     fail('claim requires --queue <name>');
   }
diff --git a/src/config.ts b/src/config.ts
index e2b54332..b6c73f89 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -46,6 +46,7 @@ export interface ParsedFlags {
   flush: boolean;
   check: boolean;
   uninstall: boolean;
+  all: boolean;
   events?: string;
   since?: string;
   _: string[]; // positional args
@@ -64,6 +65,7 @@ export function parseArgs(argv: string[]): ParsedFlags {
     version: false,
     check: false,
     uninstall: false,
+    all: false,
     _: [],
   };
   for (let i = 0; i < argv.length; i++) {
@@ -161,6 +163,9 @@ export function parseArgs(argv: string[]): ParsedFlags {
       case 'uninstall':
         flags.uninstall = true;
         break;
+      case 'all':
+        flags.all = true;
+        break;
       case 'events':
         flags.events = takeVal();
         break;
diff --git a/src/daemon.ts b/src/daemon.ts
index 845c553a..5a6002a2 100644
--- a/src/daemon.ts
+++ b/src/daemon.ts
@@ -180,11 +180,38 @@ export async function runDaemon(uri: string): Promise<void> {
     });
   }
 
-  // warm mirror: full key set + bodies. On Snapshottable backends every poll
-  // is one round trip (bodies included); elsewhere the poll carries keys only
-  // and bodies fill lazily through the `get` passthrough — message blobs are
-  // immutable, and agent records (which mutate: heartbeats) are invalidated
-  // each poll so the next read refetches them.
+  // warm mirror: the full key set, plus the bodies worth holding. Message
+  // blobs are immutable, so a body already held is never re-read; agent
+  // records (which mutate: heartbeats) are refreshed every poll.
+  //
+  // What is NOT held warm is history (issue #167). A bus only grows, and
+  // re-reading every archived message and telemetry batch on every poll made
+  // the daemon's cost scale with all history instead of with what is hot —
+  // forever, on a store where nothing but the last few messages is live.
+  // Archives and events older than the window keep their KEY (so list, log,
+  // purge and channel discovery see the whole store, exactly as before) and
+  // their body fills on demand through the `get` passthrough.
+  const historyMs = Math.max(0, Number(process.env.AGENTCOMM_MIRROR_HISTORY_MS ?? 7 * 24 * 3600_000));
+  const HOT_PREFIXES = ['agents/', 'inbox/'];
+  /**
+   * ms timestamp encoded in an archive/event key, or null when it has none.
+   * Both layouts put a zero-padded, monotonic ms prefix on the filename:
+   * read/<recipient>/<ms>-<counter>_<id>.json and events/<ms>-<counter>_<id>.json.
+   */
+  const keyTime = (key: string): number | null => {
+    const m = /^0*(\d+)-/.exec(key.slice(key.lastIndexOf('/') + 1));
+    return m ? Number(m[1]) : null;
+  };
+  const withinWindow = (key: string): boolean => {
+    const ts = keyTime(key);
+    return ts === null || Date.now() - ts <= historyMs;
+  };
+  /** Should the mirror hold this key's body at all? */
+  const worthWarming = (key: string): boolean => HOT_PREFIXES.some((p) => key.startsWith(p)) || withinWindow(key);
+  /** Should THIS poll read it? Not if we already hold an immutable body. */
+  const needsRead = (key: string): boolean =>
+    key.startsWith('agents/') || (worthWarming(key) && !mirror.has(key));
+
   const mirror = new Map<string, Buffer>();
   let keys = new Set<string>();
 
@@ -237,23 +264,24 @@ export async function runDaemon(uri: string): Promise<void> {
 
   async function doPoll(): Promise<void> {
     const spooled = await snapshotSpool();
+    // Bodies this poll actually read (fresh agent records + keys we do not
+    // already hold), merged onto the ones already warm.
+    const fetched = new Map<string, Buffer>();
+    let next: Set<string>;
     if (isSnapshottable(backend)) {
-      const snap = await backend.snapshot('');
-      const present = new Set(snap.keys());
-      replaySpool(spooled, snap, present);
-      for (const k of snap.keys()) if (!present.has(k)) snap.delete(k);
-      mirror.clear();
-      for (const [k, v] of snap) mirror.set(k, v);
-      keys = present;
-      return;
+      const snap = await backend.snapshot('', { bodies: needsRead });
+      next = new Set(snap.keys);
+      for (const [k, v] of snap.bodies) fetched.set(k, v);
+    } else {
+      next = new Set(await backend.list(''));
     }
-    const bodies = new Map<string, Buffer>();
-    const next = new Set(await backend.list(''));
-    replaySpool(spooled, bodies, next);
+    replaySpool(spooled, fetched, next);
     for (const k of mirror.keys()) {
-      if (!next.has(k) || k.startsWith('agents/')) mirror.delete(k);
+      // drop what the store no longer has, what a fresher read supersedes,
+      // and history that has aged out of the warm window
+      if (!next.has(k) || k.startsWith('agents/') || !worthWarming(k)) mirror.delete(k);
     }
-    for (const [k, v] of bodies) if (next.has(k)) mirror.set(k, v);
+    for (const [k, v] of fetched) if (next.has(k)) mirror.set(k, v);
     keys = next;
   }
   // Coalesce: timer ticks, `refresh`, and post-claim polls that overlap a
diff --git a/src/hook-run.ts b/src/hook-run.ts
index e4460fff..0be6717b 100644
--- a/src/hook-run.ts
+++ b/src/hook-run.ts
@@ -272,7 +272,8 @@ async function hookStopGuard(input: HookInput): Promise<void> {
       reason:
         `agentcomm delivery (working as intended — not an error): ${msgs.length} unread bus message(s) ` +
         `for ${alias} (from: ${from}). Read them with \`agentcomm inbox --json\`, act or tell the user why not, ` +
-        'then finish.',
+        'then finish. If you already read them another way (peek, a digest), clear them with ' +
+        '`agentcomm ack --all` — that is what this guard is counting.',
     }),
   );
 }
diff --git a/src/session.ts b/src/session.ts
index 74ed962e..1db324b8 100644
--- a/src/session.ts
+++ b/src/session.ts
@@ -181,3 +181,84 @@ export async function recordAlias(session: string, alias: string): Promise<strin
 function hash(seed: string): string {
   return createHash('sha1').update(seed).digest('hex').slice(0, 12);
 }
+
+// ── mailbox leases (issue #161) ─────────────────────────────────────────────
+
+/**
+ * A subagent inherits its parent's git identity and process tree, so it
+ * derives the parent's alias — and since reads consume, one `agentcomm inbox`
+ * from a subagent drains the mailbox its parent is waiting on. Identity stays
+ * per-session on purpose (a session IS a mailbox; see the sticky fingerprint
+ * above), so what is needed is not a different name but VISIBILITY: while a
+ * process is acting as an alias it leaves a lease behind, and anything
+ * destructive done to a leased mailbox by another live process says so.
+ *
+ * Leases are local files: the case they cover — two processes of one agent
+ * session on one machine — is exactly the local case.
+ */
+export interface Lease {
+  alias: string;
+  pid: number;
+  /** What that process is doing, for a message a human can act on. */
+  command: string;
+  since: string;
+}
+
+/** Leases older than this are ignored even if the pid is somehow still alive. */
+const LEASE_TTL_MS = 30 * 60_000;
+
+const leaseFile = (alias: string, pid: number): string => path.join(stateDir(), `lease-${alias}-${pid}.json`);
+
+const pidAlive = (pid: number): boolean => {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+};
+
+/**
+ * Take a lease on `alias` for this process and return the OTHER live holders.
+ * Callers decide what to say about them — a consuming read on a mailbox
+ * another process is holding is worth a loud warning; a heartbeat is not.
+ */
+export async function leaseMailbox(alias: string, command: string): Promise<Lease[]> {
+  const others: Lease[] = [];
+  const dir = stateDir();
+  let names: string[] = [];
+  try {
+    names = await fs.readdir(dir);
+  } catch {
+    /* no state yet */
+  }
+  for (const name of names) {
+    if (!name.startsWith(`lease-${alias}-`) || !name.endsWith('.json')) continue;
+    const file = path.join(dir, name);
+    try {
+      const lease = JSON.parse(await fs.readFile(file, 'utf8')) as Lease;
+      const stale = Date.now() - Date.parse(lease.since) > LEASE_TTL_MS;
+      if (lease.pid === process.pid) continue;
+      if (stale || !pidAlive(lease.pid)) {
+        await fs.rm(file, { force: true }).catch(() => {});
+        continue;
+      }
+      if (lease.alias === alias) others.push(lease);
+    } catch {
+      await fs.rm(file, { force: true }).catch(() => {});
+    }
+  }
+  const mine: Lease = { alias, pid: process.pid, command, since: new Date().toISOString() };
+  try {
+    await fs.mkdir(dir, { recursive: true });
+    await fs.writeFile(leaseFile(alias, process.pid), JSON.stringify(mine));
+  } catch {
+    /* a lease is advisory — never fail a command over it */
+  }
+  return others;
+}
+
+/** Drop this process's lease. Called when the command ends. */
+export async function releaseMailbox(alias: string): Promise<void> {
+  await fs.rm(leaseFile(alias, process.pid), { force: true }).catch(() => {});
+}
diff --git a/src/types.ts b/src/types.ts
index 0c12b1e2..8726a7e7 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -73,8 +73,19 @@ export interface Claimable {
  * clock — the bus daemon uses this to warm its mirror in one round trip.
  */
 export interface Snapshottable {
-  /** All keys under `prefix` with their bodies, read at a single consistent point. */
-  snapshot(prefix: string): Promise<Map<string, Buffer>>;
+  /**
+   * Every key under `prefix` at a single consistent point, with the bodies
+   * `opts.bodies` selects (all of them when it is absent).
+   *
+   * The split matters because a bus's history only grows: re-reading every
+   * archived body on every poll makes the daemon cost scale with all history
+   * rather than with what is hot (issue #167). Keys are cheap to list, so
+   * callers still see the whole store and fetch a cold body on demand.
+   */
+  snapshot(
+    prefix: string,
+    opts?: { bodies?: (key: string) => boolean },
+  ): Promise<{ keys: string[]; bodies: Map<string, Buffer> }>;
 }
 
 /**
diff --git a/test/bus.test.ts b/test/bus.test.ts
index 536f0477..9d792c80 100644
--- a/test/bus.test.ts
+++ b/test/bus.test.ts
@@ -306,3 +306,45 @@ describe('archive batches when the backend can (issue #159)', () => {
     expect((await inner.list('inbox/bob/')).length).toBe(0);
   });
 });
+
+/**
+ * Read and clear were the same operation (issue #160): `peek` shows without
+ * consuming, `inbox` consumes what it shows, and nothing cleared mail read
+ * any other way — so an agent that had read and answered every message still
+ * showed N unread forever.
+ */
+describe('ack clears mail already read (issue #160)', () => {
+  it('acks by id without re-fetching bodies, and reports ids that are not pending', async () => {
+    const backend = new LocalBackend(await mkTmp());
+    const bus = new Bus(backend);
+    const one = await bus.send({ from: 'alice', to: 'bob', body: 'one' });
+    const two = await bus.send({ from: 'alice', to: 'bob', body: 'two' });
+    await bus.send({ from: 'alice', to: 'bob', body: 'three' });
+
+    expect(await bus.peek('bob')).toHaveLength(3); // read, non-destructively
+
+    const acked = await bus.ack('bob', [one.id, two.id, 'not-a-real-id']);
+    expect(acked.acked.sort()).toEqual([one.id, two.id].sort());
+    expect(acked.unknown).toEqual(['not-a-real-id']);
+    expect(acked.failed).toEqual([]);
+
+    expect((await bus.peek('bob')).map((m) => m.body)).toEqual(['three']);
+    expect((await backend.list('read/bob/')).length).toBe(2); // archived, not deleted
+  });
+
+  it('--all clears the whole mailbox; acking an empty one is a no-op', async () => {
+    const bus = new Bus(new LocalBackend(await mkTmp()));
+    for (const body of ['a', 'b']) await bus.send({ from: 'alice', to: 'bob', body });
+    expect((await bus.ack('bob', 'all')).acked).toHaveLength(2);
+    expect(await bus.peek('bob')).toHaveLength(0);
+    expect(await bus.ack('bob', 'all')).toEqual({ acked: [], unknown: [], failed: [] });
+  });
+
+  it('one agent cannot ack another agent mailbox message', async () => {
+    const bus = new Bus(new LocalBackend(await mkTmp()));
+    const mine = await bus.send({ from: 'alice', to: 'bob', body: 'for bob' });
+    const result = await bus.ack('carol', [mine.id]);
+    expect(result.unknown).toEqual([mine.id]);
+    expect(await bus.peek('bob')).toHaveLength(1);
+  });
+});
diff --git a/test/cli.e2e.test.ts b/test/cli.e2e.test.ts
index 39a60904..53d85bee 100644
--- a/test/cli.e2e.test.ts
+++ b/test/cli.e2e.test.ts
@@ -74,6 +74,36 @@ describe('CLI e2e (sqlite backend)', () => {
     expect(JSON.parse(empty.stdout)).toEqual([]);
   });
 
+  it('peek → ack clears the unread count without a consuming read (issue #160)', async () => {
+    const db = `sqlite://${path.join(await mkTmp(), 'bus.db')}`;
+    const first = JSON.parse((await run(['send', 'bob', 'one', '--as', 'alice', '--backend', db, '--json'])).stdout) as {
+      id: string;
+    };
+    await run(['send', 'bob', 'two', '--as', 'alice', '--backend', db, '--json']);
+
+    // read non-destructively — the human view says how to clear it
+    const peeked = await run(['peek', '--as', 'bob', '--backend', db]);
+    expect(peeked.stdout).toMatch(/2 still unread; clear what you have handled: agentcomm ack --all/);
+
+    const one = await run(['ack', first.id, '--as', 'bob', '--backend', db, '--json']);
+    expect(one.code).toBe(0);
+    expect(JSON.parse(one.stdout)).toMatchObject({ agent: 'bob', acked: [first.id], unknown: [] });
+    expect((JSON.parse((await run(['peek', '--as', 'bob', '--backend', db, '--json'])).stdout) as unknown[]).length).toBe(1);
+
+    // an id that is not pending is an explicit non-zero, not a silent success
+    const bogus = await run(['ack', 'deadbeef', '--as', 'bob', '--backend', db]);
+    expect(bogus.code).toBe(1);
+    expect(bogus.stdout).toMatch(/not pending for bob: deadbeef/);
+
+    const all = await run(['ack', '--all', '--as', 'bob', '--backend', db]);
+    expect(all.stdout).toMatch(/acked 1 message\(s\) for bob/);
+    expect(JSON.parse((await run(['peek', '--as', 'bob', '--backend', db, '--json'])).stdout)).toEqual([]);
+
+    const bare = await run(['ack', '--as', 'bob', '--backend', db]);
+    expect(bare.code).toBe(1);
+    expect(bare.stderr).toMatch(/ack requires message ids/);
+  });
+
   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' };
diff --git a/test/daemon.e2e.test.ts b/test/daemon.e2e.test.ts
index bf09d62a..3f8cda68 100644
--- a/test/daemon.e2e.test.ts
+++ b/test/daemon.e2e.test.ts
@@ -313,6 +313,33 @@ describe('bus daemon: same semantics, immediate answers', () => {
     expect(archived).toHaveLength(3);
   });
 
+  it('keeps only recent history warm, but still serves the old (issue #167)', async () => {
+    const dir = await mkTmp();
+    // an archived message from long ago, and a fresh one — both on the store
+    const archive = path.join(dir, '.bus', 'read', 'alpha');
+    await fs.mkdir(archive, { recursive: true });
+    const oldTs = String(Date.now() - 30 * 24 * 3600_000).padStart(15, '0');
+    await fs.writeFile(
+      path.join(archive, `${oldTs}-000000_ancient.json`),
+      JSON.stringify({ id: 'ancient', from: 'beta', to: 'alpha', body: 'from a month ago', ts: new Date(0).toISOString() }),
+    );
+
+    const WINDOW = { AGENTCOMM_MIRROR_HISTORY_MS: String(7 * 24 * 3600_000) };
+    await run(['register', '--as', 'alpha', '--daemon'], dir, WINDOW);
+    await run(['send', 'alpha', 'today', '--as', 'beta', '--daemon', '--sync'], dir, WINDOW);
+
+    // the aged-out archive is still a first-class key: log finds it and reads
+    // its body through the passthrough, exactly as if it were warm
+    const log = await run(['log', '--limit', '10', '--daemon', '--json'], dir, WINDOW);
+    const bodies = (JSON.parse(log.stdout) as { body: string }[]).map((m) => m.body);
+    expect(bodies).toContain('from a month ago');
+    expect(bodies).toContain('today');
+
+    // and purge still sees it (list is unaffected by what the mirror holds)
+    const purge = await run(['purge', '--older-than', '14d', '--dry-run', '--daemon', '--json'], dir, WINDOW);
+    expect((JSON.parse(purge.stdout) as { count: number }).count).toBe(1);
+  });
+
   it('outbox survives a daemon crash: a fresh daemon delivers the leftovers', async () => {
     const dir = await mkTmp();
     const FROZEN = { AGENTCOMM_FLUSH_MS: '600000' };
diff --git a/test/git.e2e.test.ts b/test/git.e2e.test.ts
index cd724337..7ac14c7f 100644
--- a/test/git.e2e.test.ts
+++ b/test/git.e2e.test.ts
@@ -77,17 +77,36 @@ describe('GitBackend (local bare remotes — same code path as any host)', () =>
     const { uri, cache } = await bareRemote();
     const b = await open(uri, cache);
 
-    expect((await b.snapshot('')).size).toBe(0); // branch not born yet
+    expect((await b.snapshot('')).keys).toEqual([]); // branch not born yet
 
     await b.put('inbox/a/001.json', Buffer.from('one'));
     await b.put('inbox/b/001.json', Buffer.from('two'));
     await b.put('agents/a.json', Buffer.from('{"alias":"a"}'));
 
     const all = await b.snapshot('');
-    expect([...all.keys()].sort()).toEqual(['agents/a.json', 'inbox/a/001.json', 'inbox/b/001.json']);
-    for (const [k, v] of all) expect(v.equals(await b.get(k))).toBe(true); // bodies match per-key reads
+    expect([...all.keys].sort()).toEqual(['agents/a.json', 'inbox/a/001.json', 'inbox/b/001.json']);
+    for (const [k, v] of all.bodies) expect(v.equals(await b.get(k))).toBe(true); // bodies match per-key reads
 
-    expect([...(await b.snapshot('inbox/a/')).keys()]).toEqual(['inbox/a/001.json']);
+    expect((await b.snapshot('inbox/a/')).keys).toEqual(['inbox/a/001.json']);
+  }, 60000);
+
+  it('snapshot reads only the bodies asked for, but reports every key (issue #167)', async () => {
+    const { uri, cache } = await bareRemote();
+    const b = await open(uri, cache);
+    await b.put('agents/a.json', Buffer.from('{"alias":"a"}'));
+    await b.put('read/a/000000000000001-000000_old.json', Buffer.from('ancient history'));
+    await b.put('inbox/a/000000000000002-000000_new.json', Buffer.from('live mail'));
+
+    const warm = await b.snapshot('', { bodies: (key) => !key.startsWith('read/') });
+    // the archive still EXISTS as far as list/log/purge are concerned...
+    expect(warm.keys).toContain('read/a/000000000000001-000000_old.json');
+    // ...its body just was not read
+    expect([...warm.bodies.keys()].sort()).toEqual([
+      'agents/a.json',
+      'inbox/a/000000000000002-000000_new.json',
+    ]);
+    // and it is still one `get` away
+    expect((await b.get('read/a/000000000000001-000000_old.json')).toString()).toBe('ancient history');
   }, 60000);
 
   it('move is ATOMIC — one commit relocates the key', async () => {
diff --git a/test/mailbox-lease.e2e.test.ts b/test/mailbox-lease.e2e.test.ts
new file mode 100644
index 00000000..ff1f95b9
--- /dev/null
+++ b/test/mailbox-lease.e2e.test.ts
@@ -0,0 +1,123 @@
+/**
+ * e2e for mailbox leases (issue #161).
+ *
+ * A subagent inherits its parent's git identity and process tree, so it
+ * derives the parent's alias — and since reads consume, one `agentcomm inbox`
+ * from a subagent drains the mailbox its parent is waiting on. The alias
+ * stays per-session on purpose; what these cover is that sharing one is no
+ * longer silent.
+ */
+import { describe, it, expect, afterEach } from 'vitest';
+import { promises as fs } from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+import { fileURLToPath, pathToFileURL } from 'node:url';
+import { createRequire } from 'node:module';
+import { spawn } from 'node:child_process';
+
+const here = path.dirname(fileURLToPath(import.meta.url));
+const cli = path.join(here, '..', 'src', 'cli.ts');
+const tsx = pathToFileURL(createRequire(import.meta.url).resolve('tsx')).href;
+
+const tmpRoots: string[] = [];
+async function mkTmp(): Promise<string> {
+  const dir = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), 'agentcomm-lease-')));
+  tmpRoots.push(dir);
+  return dir;
+}
+afterEach(async () => {
+  for (const dir of tmpRoots.splice(0)) await fs.rm(dir, { recursive: true, force: true });
+});
+
+interface Ctx {
+  dir: string;
+  state: string;
+}
+
+function spawnCli(args: string[], ctx: Ctx): ReturnType<typeof spawn> {
+  return spawn(process.execPath, ['--import', tsx, cli, ...args], {
+    stdio: ['ignore', 'pipe', 'pipe'],
+    cwd: ctx.dir,
+    env: {
+      PATH: process.env.PATH,
+      HOME: process.env.HOME,
+      AGENTCOMM_BACKEND: `file://${path.join(ctx.dir, '.bus')}`,
+      AGENTCOMM_NO_GIT_PROBE: '1',
+      AGENTCOMM_SESSION: 'lease-test',
+      AGENTCOMM_STATE_DIR: ctx.state,
+    },
+  });
+}
+
+function run(args: string[], ctx: Ctx): Promise<{ code: number; stdout: string; stderr: string }> {
+  return new Promise((resolve, reject) => {
+    const child = spawnCli(args, ctx);
+    let stdout = '';
+    let stderr = '';
+    child.stdout!.on('data', (d) => (stdout += d.toString()));
+    child.stderr!.on('data', (d) => (stderr += d.toString()));
+    child.on('error', reject);
+    child.on('exit', (code) => resolve({ code: code ?? -1, stdout, stderr }));
+  });
+}
+
+async function ctx(): Promise<Ctx> {
+  return { dir: await mkTmp(), state: await mkTmp() };
+}
+
+describe('mailbox leases: sharing an alias stops being silent (issue #161)', () => {
+  it('a consuming read warns while another live process holds the same alias', async () => {
+    const c = await ctx();
+    // the "parent": a listener blocked on its own mailbox, holding it
+    const listener = spawnCli(['wait', '--as', 'worker', '--timeout', '8000'], c);
+    await new Promise((r) => setTimeout(r, 1200)); // let it take the lease
+
+    // the "subagent": same alias, consuming read — the mail it drains would
+    // have been the listener's
+    const drain = await run(['inbox', '--as', 'worker', '--json'], c);
+    expect(drain.stderr).toMatch(/WARNING — another live process of this session is acting as "worker"/);
+    expect(drain.stderr).toMatch(/agentcomm wait/); // says what that process is doing
+    expect(drain.stderr).toMatch(/--as worker-<role>/); // and the remedy
+
+    listener.kill('SIGKILL');
+  });
+
+  it('says nothing when nobody else holds the alias, and cleans up after itself', async () => {
+    const c = await ctx();
+    await run(['send', 'solo', 'hello', '--as', 'boss'], c);
+
+    const first = await run(['inbox', '--as', 'solo'], c);
+    expect(first.stderr).not.toMatch(/WARNING/);
+    // the lease is released at exit — the NEXT command must not see a ghost
+    const second = await run(['inbox', '--as', 'solo'], c);
+    expect(second.stderr).not.toMatch(/WARNING/);
+    expect((await fs.readdir(c.state)).filter((f) => f.startsWith('lease-'))).toEqual([]);
+  });
+
+  it('a status write onto an alias another process is holding says whose line it replaces', async () => {
+    const c = await ctx();
+    const listener = spawnCli(['wait', '--as', 'shared', '--timeout', '8000'], c);
+    await new Promise((r) => setTimeout(r, 1200));
+
+    const status = await run(['register', '--as', 'shared', '--status', 'subagent work'], c);
+    expect(status.stderr).toMatch(/WARNING — another live process of this session is acting as "shared"/);
+    expect(status.stderr).toMatch(/replaces its line on the shared roster/);
+
+    // a plain heartbeat is not a claim on the roster — no warning
+    const heartbeat = await run(['register', '--as', 'shared'], c);
+    expect(heartbeat.stderr).not.toMatch(/WARNING/);
+
+    listener.kill('SIGKILL');
+  });
+
+  it('a lease left by a dead process is ignored, not inherited', async () => {
+    const c = await ctx();
+    await fs.writeFile(
+      path.join(c.state, 'lease-ghost-999999.json'),
+      JSON.stringify({ alias: 'ghost', pid: 999999, command: 'wait', since: new Date().toISOString() }),
+    );
+    const r = await run(['inbox', '--as', 'ghost'], c);
+    expect(r.stderr).not.toMatch(/WARNING/);
+    expect(await fs.readdir(c.state)).not.toContain('lease-ghost-999999.json');
+  });
+});