From 6e1301f88e728b587925c03e623a65295b3e630c Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:59:24 +0300 Subject: [PATCH] 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'); + }); +});