From 90a79202a489315bbe3d57d6e60fc8327c7e8a6a Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:51:04 +0300 Subject: [PATCH 1/3] =?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 1f7a215aa898bbaea3d521bf8af6808e292a2f06 Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:55:36 +0300 Subject: [PATCH 2/3] 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 () => { From 2d796256ffae6e6b9a39d32dceea47bd62180805 Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:59:24 +0300 Subject: [PATCH 3/3] fix: make two processes sharing one mailbox visible (#161) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A subagent inherits its parent's git identity and process tree, so it derives the parent's alias. Because reads consume, one `agentcomm inbox` from a subagent drains the mailbox its parent is waiting on — silently, since the alias looks exactly right. One did; it noticed only because the CLI echoed the parent's name and the agent recognised it. The smaller version of the same thing: a subagent's `register --status` overwrote the parent's line on the shared roster. Identity stays per-session on purpose — a session IS a mailbox, and the sticky fingerprint that fixed alias drift depends on it. What was missing is visibility, so while a process acts as an alias it now leaves a local lease behind, and anything destructive another live process does to that mailbox says so: a consuming read warns that it is taking mail addressed to a live holder and names the remedy (--as -), and a status write warns whose line it is replacing. Heartbeats, peeks and queue claims lease quietly — a shared queue is meant to have many claimers. Leases are local files, because the case they cover — two processes of one agent session on one machine — is exactly the local case. A lease from a dead process is ignored and cleaned up, never inherited. --- src/cli.ts | 40 ++++++++++- src/session.ts | 81 ++++++++++++++++++++++ test/mailbox-lease.e2e.test.ts | 123 +++++++++++++++++++++++++++++++++ 3 files changed, 242 insertions(+), 2 deletions(-) create mode 100644 test/mailbox-lease.e2e.test.ts diff --git a/src/cli.ts b/src/cli.ts index 31cd6aef..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, @@ -326,10 +326,35 @@ async function main(argv: string[]): Promise { 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 { + const others = await leaseMailbox(me, command).catch(() => [] as Awaited>); + 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}-.\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 = @@ -708,7 +733,9 @@ This repo has a message bus for AI agents. When working here: 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 -bus\`); keep quick sends inline. + uses \`--as -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 { @@ -949,6 +976,7 @@ async function cmdRegister( statusAuto?: boolean, ): Promise { 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`); @@ -1108,6 +1136,7 @@ async function cmdBroadcast( */ async function cmdInbox(bus: Bus, cfg: ResolvedConfig): Promise { const me = await resolveAgent(cfg); + await guardMailbox(me, 'inbox', 'consuming'); await bus.inbox(me, { deliver: (messages) => printMessages(messages, cfg, me), onUnarchived: (keys) => @@ -1120,6 +1149,7 @@ async function cmdInbox(bus: Bus, cfg: ResolvedConfig): Promise { async function cmdPeek(bus: Bus, cfg: ResolvedConfig): Promise { 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, @@ -1138,6 +1168,7 @@ async function cmdPeek(bus: Bus, cfg: ResolvedConfig): Promise { */ async function cmdAck(bus: Bus, cfg: ResolvedConfig, all: boolean, ids: string[]): Promise { 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 | agentcomm ack --all'); } @@ -1161,6 +1192,10 @@ async function cmdAck(bus: Bus, cfg: ResolvedConfig, all: boolean, ids: string[] async function cmdWait(bus: Bus, cfg: ResolvedConfig, timeoutMs: number): Promise { 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([]); @@ -1173,6 +1208,7 @@ async function cmdWait(bus: Bus, cfg: ResolvedConfig, timeoutMs: number): Promis async function cmdClaim(bus: Bus, cfg: ResolvedConfig, queue: string | undefined): Promise { 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 '); } 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 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 { + 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 { + await fs.rm(leaseFile(alias, process.pid), { force: true }).catch(() => {}); +} 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 { + 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 { + 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 { + 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-/); // 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'); + }); +});