perf: bound the daemon's warm mirror to recent history (#167)

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/cli.ts

@@ -158,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

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

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> }>;
 }
 
 /**

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

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 () => {