From f8bc5c8b95a92040b30dfe8e187ce83ded108cf9 Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:51:04 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20`ack`=20=E2=80=94=20clear=20mail=20you?= =?UTF-8?q?=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' };