From bc023e41eeb4eee843c6dca33e30f9c238440b2e Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:51:04 +0300 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20`ack`=20=E2=80=94=20clear=20mail=20?= =?UTF-8?q?you=20already=20read=20(#160)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `peek` shows messages without consuming; `inbox` consumes what it shows. With only those two there was no way to mark mail read that arrived any other way, so an agent that had read all five messages via `peek` and replied to every one still showed five unread — and the stop guard blocked every turn for fifteen turns demanding it read them. `agentcomm ack ` / `ack --all` archives pending messages by id. It works off the KEYS (the id is in the key), so nothing is re-fetched, and ids that are not pending for you are reported with a non-zero exit instead of passing silently — acking someone else's mail is not a quiet no-op. `peek` now closes with the one-liner that clears what you handled, the stop guard says the same thing, and the guidance file teaches peek+ack as the non-destructive read path. --- src/bus.ts | 27 ++++++++++++++++++++++++ src/cli.ts | 49 ++++++++++++++++++++++++++++++++++++++++++-- src/config.ts | 5 +++++ src/hook-run.ts | 3 ++- test/bus.test.ts | 42 +++++++++++++++++++++++++++++++++++++ test/cli.e2e.test.ts | 30 +++++++++++++++++++++++++++ 6 files changed, 153 insertions(+), 3 deletions(-) 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//_.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..796b222e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -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 "" · 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 [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 | --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 emit/events: the event type (skill-ran, skill-outcome, …) --name emit/events: what the event is about (a skill/tool name) --ref emit/events: correlation handle (branch, PR#, run id) @@ -296,6 +303,8 @@ async function main(argv: string[]): Promise { 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': @@ -686,7 +695,9 @@ 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: "\` — other agents' digests will recruit help. If a digest shows someone else blocked and you KNOW the answer, send it without asking @@ -1107,9 +1118,43 @@ async function cmdPeek(bus: Bus, cfg: ResolvedConfig): Promise { const me = await resolveAgent(cfg); 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 { + const me = await resolveAgent(cfg); + if (!all && ids.length === 0) { + 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); + 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 { const me = await resolveAgent(cfg); const messages = await bus.wait(me, timeoutMs); 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/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 { 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/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' }; From ec268bf9aa0e23251264b72e2e28c9ac936617a3 Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:55:36 +0300 Subject: [PATCH 2/2] perf: bound the daemon's warm mirror to recent history (#167) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon mirrored the entire bus — every key AND every body — and refreshed it on every poll. On a snapshot-capable backend that is one round trip, but it read every blob in the store each time: pending mail, the roster, the whole 30-day archive, and every telemetry batch. The cost of a poll therefore grew with all history, forever, on a bus where nothing but the last few messages is hot. The mirror now holds what is worth holding. `agents/` is re-read every poll (records mutate); `inbox/` is always warm; `read/` and `events/` are warm only inside a window (AGENTCOMM_MIRROR_HISTORY_MS, default 7 days). Older history keeps its KEY, so list, log, purge and channel discovery see the whole store exactly as before, and a cold body loads on demand through the `get` passthrough. Message blobs are immutable, so a body already held is never re-read either: steady state is "the roster plus whatever is new". `Snapshottable.snapshot` grew a body filter and now reports the keys it saw alongside the bodies it read — listing names is cheap, reading blobs is not. --- src/backends/git.ts | 21 +++++++++----- src/cli.ts | 4 +++ src/daemon.ts | 64 +++++++++++++++++++++++++++++------------ src/types.ts | 15 ++++++++-- test/daemon.e2e.test.ts | 27 +++++++++++++++++ test/git.e2e.test.ts | 27 ++++++++++++++--- 6 files changed, 127 insertions(+), 31 deletions(-) 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> { + async snapshot( + prefix: string, + opts?: { bodies?: (key: string) => boolean }, + ): Promise<{ keys: string[]; bodies: Map }> { const out = new Map(); 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 { diff --git a/src/cli.ts b/src/cli.ts index 796b222e..31cd6aef 100644 --- a/src/cli.ts +++ b/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 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 { }); } - // 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//-_.json and events/-_.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(); let keys = new Set(); @@ -237,23 +264,24 @@ export async function runDaemon(uri: string): Promise { async function doPoll(): Promise { 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(); + let next: Set; 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(); - 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/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>; + /** + * 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 }>; } /** 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 () => {