fix: make two processes sharing one mailbox visible (#161)

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<number> {
         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<void> {
+  const others = await leaseMailbox(me, command).catch(() => [] as Awaited<ReturnType<typeof leaseMailbox>>);
+  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}-<role>.\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 <you>-bus\`); keep quick sends inline.
+  uses \`--as <you>-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<number> {
@@ -949,6 +976,7 @@ async function cmdRegister(
   statusAuto?: boolean,
 ): Promise<number> {
   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<number> {
   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<number> {
 
 async function cmdPeek(bus: Bus, cfg: ResolvedConfig): Promise<number> {
   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<number> {
  */
 async function cmdAck(bus: Bus, cfg: ResolvedConfig, all: boolean, ids: string[]): Promise<number> {
   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 <id…> | 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<number> {
   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<number> {
   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 <name>');
   }

src/session.ts

@@ -181,3 +181,84 @@ export async function recordAlias(session: string, alias: string): Promise<strin
 function hash(seed: string): string {
   return createHash('sha1').update(seed).digest('hex').slice(0, 12);
 }
+
+// ── 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. Identity stays
+ * per-session on purpose (a session IS a mailbox; see the sticky fingerprint
+ * above), so what is needed is not a different name but VISIBILITY: while a
+ * process is acting as an alias it leaves a lease behind, and anything
+ * destructive done to a leased mailbox by another live process says so.
+ *
+ * Leases are local files: the case they cover — two processes of one agent
+ * session on one machine — is exactly the local case.
+ */
+export interface Lease {
+  alias: string;
+  pid: number;
+  /** What that process is doing, for a message a human can act on. */
+  command: string;
+  since: string;
+}
+
+/** Leases older than this are ignored even if the pid is somehow still alive. */
+const LEASE_TTL_MS = 30 * 60_000;
+
+const leaseFile = (alias: string, pid: number): string => 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<Lease[]> {
+  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<void> {
+  await fs.rm(leaseFile(alias, process.pid), { force: true }).catch(() => {});
+}

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<string> {
+  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<typeof spawn> {
+  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<Ctx> {
+  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-<role>/); // 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');
+  });
+});