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