diff --git a/.changelog/unreleased/fixed-mailclient-one-mail-root.md b/.changelog/unreleased/fixed-mailclient-one-mail-root.md new file mode 100644 index 00000000..37d53610 --- /dev/null +++ b/.changelog/unreleased/fixed-mailclient-one-mail-root.md @@ -0,0 +1,12 @@ +- **MailClient resolves its mail root and mailbox directory once per operation and uses that one snapshot for every step of the operation.** + + The mailbox lock, the maildir paths and the replay state derive from the + resolved mailbox directory. Bridge-principal trust, the relay acceptance + stripe taken for promotion and for dead-lettering, and the recipient inboxes + that `deliverOutbox()` moves records into derive from the resolved mail root. + No root is kept from construction. When the symlink at the configured mail + path, or at the agent's mailbox directory, is repointed while a client is + alive, the new target is selected at the start of the next operation; an + operation already running finishes against the target it started with. A + plain rename of the configured directory is not followed: the next operation + creates the configured path again and uses that, not the renamed directory. diff --git a/packages/agent/src/io/mail.ts b/packages/agent/src/io/mail.ts index 2e116092..c2484a7e 100644 --- a/packages/agent/src/io/mail.ts +++ b/packages/agent/src/io/mail.ts @@ -7,7 +7,7 @@ import { signEnvelope, type ChainEntry, type Envelope, type FlairClient } from " import { PublicKeyFormatError } from "../lib/public-key.js"; import { agentKeyCandidates, readAgentPrivateKey } from "../lib/agent-keys.js"; import { tryAcquireRelayMoveLock } from "../lib/relay-lock.js"; -import { mailboxLockRoot } from "../lib/mail-lock-namespace.js"; +import { canonicalMailboxRoot, canonicalMailRoot } from "../lib/mail-lock-namespace.js"; import { acquireMailLock, type MailLock } from "../lib/mail-lock.js"; import { decideEnvelopeForMailbox, @@ -86,6 +86,23 @@ function resumableOnRecovery(phase: PlacementPhase | null): boolean { return phase === "placing"; } +/** + * cli#636: the roots one MailClient operation uses, resolved once when the + * operation starts and passed to every step of it. Each is the resolved real + * path (symlinks and `.`/`..` resolved) at that moment; paths beneath them + * (`new/`, `cur/`, a file) are joined to them, not resolved again. A later + * operation resolves them afresh, so when the symlink at the configured mail + * path or at the agent's mailbox directory is repointed, the new target is + * selected at the start of the next operation and an operation already running + * finishes against the target it started with. + */ +interface OperationRoots { + /** The configured mail directory: bridge-principal trust, acceptance stripes, recipient inboxes. */ + readonly mailRoot: string; + /** This agent's mailbox directory: the mailbox lock, the maildir paths and the replay state. */ + readonly mailbox: string; +} + function lstatOrAbsent(path: string): { dev: number; ino: number } | null { try { const st = lstatSync(path); @@ -102,12 +119,6 @@ function lstatOrAbsent(path: string): { dev: number; ino: number } | null { * Writes outgoing mail to mailDir/outbox/new. */ export class MailClient { - private mailboxRoot: string; - private readonly mailboxLockRoot: string; - private inboxNew: string; - private inboxCur: string; - private inboxDlq: string; - private outboxNew: string; private readonly flairClient: FlairClient; constructor( @@ -126,30 +137,40 @@ export class MailClient { ); } this.flairClient = flairClient; - this.mailboxRoot = join(mailDir, agentId); - // cli#616: promotion locks the mailbox under the resolved real path of the - // configured mail directory (see mail-lock-namespace.ts). Placement recovery - // locks `mailboxRoot` as given; both reach one filesystem lock. - this.mailboxLockRoot = mailboxLockRoot(mailDir, agentId); - this.inboxNew = join(mailDir, agentId, "new"); - this.inboxCur = join(mailDir, agentId, "cur"); - this.inboxDlq = join(mailDir, agentId, "dlq"); - this.outboxNew = join(mailDir, agentId, "outbox"); - for (const dir of [this.inboxNew, this.inboxCur, this.inboxDlq, this.outboxNew]) { - mkdirSync(dir, { recursive: true }); - } + // cli#636: create the maildir under the target configured now. Nothing is + // kept: every operation resolves its own roots (operationRoots). + this.operationRoots(); + } + + /** + * cli#636: resolve the roots for one operation (see OperationRoots) and create + * the maildir subdirectories under the mailbox. The mail root is the resolved + * real path of the configured `mailDir`; the mailbox is the resolved real path + * of `/`, so a symlink at the mailbox directory itself is + * resolved too. Both are resolved through mail-lock-namespace.ts (cli#616). + */ + private operationRoots(): OperationRoots { + const mailRoot = canonicalMailRoot(this.mailDir); + // Create the mailbox with the default mode first (canonicalMailboxRoot + // would create it 0700); resolving it then creates nothing. + mkdirSync(join(mailRoot, this.agentId), { recursive: true }); + const mailbox = canonicalMailboxRoot(join(mailRoot, this.agentId)); + for (const dir of ["new", "cur", "dlq", "outbox"]) mkdirSync(join(mailbox, dir), { recursive: true }); + return { mailRoot, mailbox }; } async checkNewMail(): Promise { const messages: MailMessage[] = []; - await this.recoverInterruptedPlacements(messages); - if (!existsSync(this.inboxNew)) return messages; + const roots = this.operationRoots(); + await this.recoverInterruptedPlacements(messages, roots); + const inboxNew = join(roots.mailbox, "new"); + if (!existsSync(inboxNew)) return messages; - const files = readdirSync(this.inboxNew).filter((f) => !f.startsWith(".") && !f.includes("/") && !f.includes("\\")); + const files = readdirSync(inboxNew).filter((f) => !f.startsWith(".") && !f.includes("/") && !f.includes("\\")); for (const file of files) { const started = Date.now(); - const srcPath = join(this.inboxNew, file); + const srcPath = join(inboxNew, file); let body: string; try { @@ -186,13 +207,13 @@ export class MailClient { } if (!verifyResult.pass) { - this.deadLetter(file, srcPath, verifyResult, started); + this.deadLetter(file, srcPath, verifyResult, started, roots); continue; } let trustTier: MailMessage["trustTier"]; try { - trustTier = verifiedMailTier(verifyResult.envelope, this.mailDir); + trustTier = verifiedMailTier(verifyResult.envelope, roots.mailRoot); } catch (err) { const detail = sanitizeError(err); this.events?.emit({ @@ -208,10 +229,10 @@ export class MailClient { } try { - const committed = await this.commitToCur(file, srcPath, body, verifyResult.envelope); + const committed = await this.commitToCur(roots, file, srcPath, body, verifyResult.envelope); if (committed === "already-delivered") continue; if (committed) { - this.deadLetter(file, srcPath, { ...committed, from: verifyResult.envelope.from }, started); + this.deadLetter(file, srcPath, { ...committed, from: verifyResult.envelope.from }, started, roots); continue; } const from = verifyResult.envelope.from; @@ -244,22 +265,23 @@ export class MailClient { /** Commit under the mailbox lock; attempt placement rollback on append failure. */ private async commitToCur( + roots: OperationRoots, file: string, srcPath: string, body: string, envelope: Envelope, ): Promise<{ pass: false; class: MailboxRejectClass; reason: string } | "already-delivered" | null> { - const lock: MailLock | null = await acquireMailLock(this.mailboxLockRoot); + const lock: MailLock | null = await acquireMailLock(roots.mailbox); if (!lock) throw new Error("mailbox lock busy; not promoted"); let relayLock: MailLock | null; - try { relayLock = tryAcquireRelayMoveLock(JSON.parse(body), this.mailDir); } + try { relayLock = tryAcquireRelayMoveLock(JSON.parse(body), roots.mailRoot); } catch (error) { lock.release(); throw error; } if (!relayLock) { lock.release(); throw new Error("relay acceptance busy; not promoted"); } try { if (readFileSync(srcPath, "utf-8") !== body) throw new Error("source changed during promotion; not promoted"); - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(roots.mailbox); if (replay.hasPendingFile(file)) throw new Error(`pending placement unresolved: ${file}`); - const dstPath = join(this.inboxCur, file); + const dstPath = join(roots.mailbox, "cur", file); if (sameFileByDeviceInode(srcPath, dstPath)) { replay.isConsumed(envelope.messageId); unlinkSync(srcPath); @@ -308,13 +330,14 @@ export class MailClient { * including an older intent without identity, leaves the intent pending and * withholds delivery; the new/ scan cannot replace that intent. */ - private async recoverInterruptedPlacements(messages: MailMessage[]): Promise { + private async recoverInterruptedPlacements(messages: MailMessage[], roots: OperationRoots): Promise { + const root = roots.mailbox; let pending: Array<{ messageId: string; file: string; phase: PlacementPhase }>; let listing: MailLock | null = null; try { - listing = await acquireMailLock(this.mailboxRoot); + listing = await acquireMailLock(root); if (!listing) return; // lock unavailable; defer recovery - pending = pendingPlacementEntries(this.mailboxRoot); + pending = pendingPlacementEntries(root); } catch (err) { this.events?.emit({ type: "mail.receive", @@ -331,7 +354,7 @@ export class MailClient { for (const { messageId, file, phase } of pending) { if (!resumableOnRecovery(phase)) continue; - const curPath = join(this.inboxCur, file); + const curPath = join(root, "cur", file); try { let bytes: Buffer; try { @@ -339,13 +362,13 @@ export class MailClient { } catch (err) { if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; // Observed absent; re-check under the lock before finishing the intent. - await this.finishUnlinkedPlacement(messageId, file); + await this.finishUnlinkedPlacement(messageId, file, root); continue; } const outcome = await this.verifyMailBody(bytes.toString("utf8")); if (!outcome.pass) continue; // withhold: the cur copy is not a verified message - const trustTier = verifiedMailTier(outcome.envelope, this.mailDir); - const delivered = await this.completeInterruptedPlacement(messageId, file, bytes, outcome.envelope, trustTier); + const trustTier = verifiedMailTier(outcome.envelope, roots.mailRoot); + const delivered = await this.completeInterruptedPlacement(messageId, file, root, bytes, outcome.envelope, trustTier); if (delivered) messages.push(delivered); } catch (err) { this.events?.emit({ @@ -361,14 +384,14 @@ export class MailClient { } /** Finish a still-pending intent only when cur/ is absent under the lock. */ - private async finishUnlinkedPlacement(messageId: string, file: string): Promise { - const lock = await acquireMailLock(this.mailboxRoot); + private async finishUnlinkedPlacement(messageId: string, file: string, root: string): Promise { + const lock = await acquireMailLock(root); if (!lock) return; try { - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(root); if (!replay.hasPendingPlacement(messageId, file)) return; if (!resumableOnRecovery(replay.placementPhase(file))) return; - if (lstatOrAbsent(join(this.inboxCur, file)) !== null) return; + if (lstatOrAbsent(join(root, "cur", file)) !== null) return; replay.finishPlacement(messageId, file); } finally { lock.release(); @@ -381,14 +404,15 @@ export class MailClient { private async completeInterruptedPlacement( messageId: string, file: string, + root: string, bytes: Buffer, envelope: Envelope, trustTier: MailMessage["trustTier"], ): Promise { - const lock = await acquireMailLock(this.mailboxRoot); + const lock = await acquireMailLock(root); if (!lock) throw new Error("mailbox lock busy; interrupted placement recovery deferred"); try { - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(root); // Match the intent, recorded device/inode and pre-verification raw bytes. if (!replay.hasPendingPlacement(messageId, file)) return null; if (!resumableOnRecovery(replay.placementPhase(file))) return null; @@ -396,7 +420,7 @@ export class MailClient { if (!replay.ownsPendingPlacement(messageId, file)) return null; let current: Buffer; try { - current = readFileSync(join(this.inboxCur, file)); + current = readFileSync(join(root, "cur", file)); } catch (err) { if ((err as NodeJS.ErrnoException).code === "ENOENT") return null; throw err; @@ -424,16 +448,18 @@ export class MailClient { srcPath: string, rejected: { class: MailboxRejectClass; reason: string; from?: string }, started: number, + roots: OperationRoots, ): void { - const dlqPath = join(this.inboxDlq, file); + const dlqDir = join(roots.mailbox, "dlq"); + const dlqPath = join(dlqDir, file); let relayLock: MailLock | null = null; try { let record: unknown; try { record = JSON.parse(readFileSync(srcPath, "utf-8")); } catch {} - relayLock = tryAcquireRelayMoveLock(record, this.mailDir); + relayLock = tryAcquireRelayMoveLock(record, roots.mailRoot); if (!relayLock) return; if (srcPath !== dlqPath) renameSync(srcPath, dlqPath); - writeRejectSidecar(this.inboxDlq, file, rejected.class, rejected.reason); + writeRejectSidecar(dlqDir, file, rejected.class, rejected.reason); } catch (err) { console.error(`[MailClient] failed to dead-letter ${file}: ${sanitizeError(err)}`); } finally { relayLock?.release(); } @@ -451,6 +477,7 @@ export class MailClient { async sendMail(to: string, body: string): Promise { const started = Date.now(); try { + const { mailbox } = this.operationRoots(); const { writeFileSync } = await import("node:fs"); // Sign the body as this agent — the v1 envelope a promote() reader // verifies. signMailBody THROWS when no key exists, so an unsigned body @@ -458,7 +485,7 @@ export class MailClient { const signed = this.signMailBody(to, body); const filename = `${Date.now()}-${Math.random().toString(36).slice(2)}.json`; writeFileSync( - join(this.outboxNew, filename), + join(mailbox, "outbox", filename), JSON.stringify({ from: this.agentId, to, body: signed, sentAt: new Date().toISOString() }, null, 2), "utf-8" ); @@ -513,17 +540,19 @@ export class MailClient { /** Deliver all messages in outbox to recipient inboxes (local relay). */ deliverOutbox(): void { - if (!existsSync(this.outboxNew)) return; - const files = readdirSync(this.outboxNew).filter(f => !f.startsWith(".")); + const roots = this.operationRoots(); + const outboxNew = join(roots.mailbox, "outbox"); + if (!existsSync(outboxNew)) return; + const files = readdirSync(outboxNew).filter(f => !f.startsWith(".")); for (const file of files) { - const srcPath = join(this.outboxNew, file); + const srcPath = join(outboxNew, file); try { const raw = readFileSync(srcPath, "utf-8"); const msg = JSON.parse(raw) as { to: string; body: string; sentAt: string }; if (!msg.to || !/^[a-zA-Z0-9_-]{1,64}$/.test(msg.to)) { continue; // skip invalid recipients } - const recipientInbox = join(this.mailDir, msg.to, "new"); + const recipientInbox = join(roots.mailRoot, msg.to, "new"); mkdirSync(recipientInbox, { recursive: true }); const destFile = `${msg.sentAt.replace(/[:.]/g, "-")}-${file}`; renameSync(srcPath, join(recipientInbox, destFile)); diff --git a/packages/agent/src/lib/mail-lock-namespace.ts b/packages/agent/src/lib/mail-lock-namespace.ts index 51c3d14d..500f7917 100644 --- a/packages/agent/src/lib/mail-lock-namespace.ts +++ b/packages/agent/src/lib/mail-lock-namespace.ts @@ -18,10 +18,13 @@ * each other: the lock is a directory created inside the locked directory, so a * symlinked path and its target already reach one filesystem lock. Within one * process the strings do matter, because the lock's re-entrancy check compares - * lock path strings. Mailbox lock - * roots come from here only for acceptance and MailClient promotion; MailClient - * placement recovery and the CLI's other mailbox lock sites lock the mailbox - * root path as given. + * lock path strings. Mailbox lock roots come from here only for acceptance and + * MailClient: acceptance derives its recipient mailbox lock roots with + * canonicalMailboxRoot, and MailClient takes every mailbox lock (promotion and + * placement recovery) at the canonicalMailboxRoot of + * `/`, resolved once per operation + * (cli#636). The CLI's other mailbox lock sites lock the mailbox root path as + * given. */ import { createHash } from "node:crypto"; import { mkdirSync, realpathSync } from "node:fs"; diff --git a/packages/agent/src/lib/relay-lock.ts b/packages/agent/src/lib/relay-lock.ts index 387c2b1e..3972a7f0 100644 --- a/packages/agent/src/lib/relay-lock.ts +++ b/packages/agent/src/lib/relay-lock.ts @@ -6,9 +6,10 @@ export { RELAY_ACCEPT_LOCK_STRIPES }; /** * The acceptance-stripe lock root for one relayed delivery. `mailRoot` defaults - * to the process mail root; MailClient passes its configured `mailDir` as - * `mailRoot` to `tryAcquireRelayMoveLock`, which derives the same root, so each - * side derives the stripe from the mail root it is configured with. + * to the process mail root; MailClient passes `tryAcquireRelayMoveLock` the mail + * root it resolved from its configured `mailDir` at the start of the operation + * (cli#636), which derives the same root, so each side derives the stripe from + * the mail root it is configured with. */ export function relayAcceptanceLockRoot(branchId: string, id: string, mailRoot = configuredMailRoot()): string { return acceptanceLockRoot(mailRoot, branchId, id); diff --git a/packages/agent/test/mail-client-repoint-root.test.ts b/packages/agent/test/mail-client-repoint-root.test.ts new file mode 100644 index 00000000..1724e8eb --- /dev/null +++ b/packages/agent/test/mail-client-repoint-root.test.ts @@ -0,0 +1,324 @@ +/** + * mail-client-repoint-root.test.ts — cli#636. + * + * MailClient resolves its mail root and its mailbox directory once at the start + * of each operation and uses that one snapshot for the whole operation: the + * mailbox lock, the maildir paths and the replay state derive from the mailbox, + * and bridge-principal trust, the acceptance stripes of promotion and rejection, + * and outbox recipient inboxes derive from the mail root. Nothing is pinned at + * construction. + * + * The first case repoints the configured mail root between operations and + * checks that the next operation uses the new target. The remaining cases + * repoint a symlink while an operation is in progress (paused in signature + * verification, or between the outbox read and the delivery move) and check + * that the operation finishes against the target it started with. Where a + * lock must exclude the client, a real second process holds it. + */ +import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; +import { spawn } from "node:child_process"; +import * as fs from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, renameSync, rmSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import * as ed from "@noble/ed25519"; +import { MailClient, type MailMessage } from "../src/io/mail.js"; +import { bridgePrincipalIds } from "../src/lib/bridge-identity.js"; +import { acceptanceLockRoot, mailboxLockRoot } from "../src/lib/mail-lock-namespace.js"; +import { hasCommittedMessageId, mailboxReplayStore } from "../src/lib/mailbox-policy.js"; +import type { FlairClient } from "../src/lib/signEnvelope.js"; +import { senderKeyResolver, signedRelayRecord } from "./helpers/signed-relay-record.js"; + +const AGENT = "mailbox"; +const BRANCH = "remote"; +// A wire-valid relayed delivery id (a UUID). +const ID = "6f1c2a3e-8d4b-4c5a-9e7f-1a2b3c4d5e6f"; + +let root: string; +let storeOld: string; +let storeNew: string; +let alias: string; +let savedEnv: Record; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "mail-client-repoint-root-")); + storeOld = join(root, "store-old"); + storeNew = join(root, "store-new"); + mkdirSync(storeOld, { recursive: true }); + mkdirSync(storeNew, { recursive: true }); + alias = join(root, "mail"); + symlinkSync(storeOld, alias, "dir"); + // The two roots name different bridge principals: in the old root the sender + // is a bridge principal (trust tier "external"); in the new root it is not. + writeBridgePrincipal(storeOld, "sender"); + writeBridgePrincipal(storeNew, "other-bridge"); + savedEnv = {}; + for (const key of ["HOME", "TPS_BRIDGE_AGENT_ID"]) savedEnv[key] = process.env[key]; + process.env.HOME = root; + delete process.env.TPS_BRIDGE_AGENT_ID; +}); + +afterEach(() => { + for (const [key, value] of Object.entries(savedEnv)) { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + rmSync(root, { recursive: true, force: true }); +}); + +function writeBridgePrincipal(store: string, id: string): void { + mkdirSync(join(store, ".bridge-principals"), { recursive: true }); + writeFileSync(join(store, ".bridge-principals", `${id}.json`), JSON.stringify({ id })); +} + +/** Point the symlink `link` at `target`. */ +function repoint(link: string, target: string): void { + unlinkSync(link); + symlinkSync(target, link, "dir"); +} + +/** Write `raw` as `file` in `//new` and return its path. */ +function seed(store: string, file: string, raw: string): string { + mkdirSync(join(store, AGENT, "new"), { recursive: true }); + const source = join(store, AGENT, "new", file); + writeFileSync(source, raw); + return source; +} + +/** + * A verifier whose first key lookup pauses until `resume()`; later lookups do + * not pause. With `wrongKey` every lookup returns a key the record was not + * signed with, so verification rejects the record. + */ +function pausedVerifier(opts: { wrongKey?: boolean } = {}) { + const resolver = senderKeyResolver(); + let resume!: () => void; + const gate = new Promise((resolve) => { resume = resolve; }); + let entered!: () => void; + const paused = new Promise((resolve) => { entered = resolve; }); + let first = true; + const client: FlairClient = { + async getAgent() { + if (first) { + first = false; + entered(); + await gate; + } + if (opts.wrongKey) return { publicKey: Buffer.from(ed.getPublicKey(Buffer.alloc(32, 2))) }; + return resolver.getAgent(); + }, + }; + return { client, paused, resume }; +} + +/** Run checkNewMail(), apply `change` while it is paused in verification, and return its result. */ +async function checkWithChangeDuringVerification( + mail: MailClient, + verifier: ReturnType, + change: () => void, +): Promise { + const pending = mail.checkNewMail(); + await Promise.race([ + verifier.paused, + pending.then(() => { throw new Error("the operation finished without reaching verification"); }), + ]); + change(); + verifier.resume(); + return pending; +} + +/** Hold the lock at `lockRoot` in a real second process and release it on stdin. */ +function holdLock(lockRoot: string): Promise<{ release: () => Promise }> { + const script = join(root, "holder.ts"); + writeFileSync( + script, + ` +const { mkdirSync } = await import("node:fs"); +const { acquireMailLockSync } = await import(${JSON.stringify(new URL("../src/lib/mail-lock.js", import.meta.url).href)}); +mkdirSync(process.env.HOLD_ROOT, { recursive: true }); +const lock = acquireMailLockSync(process.env.HOLD_ROOT); +if (!lock) throw new Error("holder could not acquire"); +process.stdout.write("ready"); +process.stdin.once("data", () => { lock.release(); process.exit(0); }); +`, + ); + const child = spawn("bun", [script], { env: { ...process.env, HOLD_ROOT: lockRoot }, stdio: ["pipe", "pipe", "inherit"] }); + const exited = new Promise((resolve) => child.once("exit", resolve)); + return new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code) => reject(new Error(`holder exited ${code} before ready`))); + child.stdout.once("data", (chunk: Buffer) => { + if (chunk.toString() === "ready") { + resolve({ release: async () => { child.stdin.end("release"); await exited; } }); + } else reject(new Error(`holder printed ${JSON.stringify(chunk.toString())}`)); + }); + }); +} + +test("a live client does not promote under the old root's lock, delivers the envelope once, and refuses its replay", async () => { + // The client is constructed while the alias points at the old store. + const mail = new MailClient(alias, undefined, AGENT, senderKeyResolver()); + + // Re-point the mail root under the live client, between operations. The new + // root is a real mailbox with one record seeded in new/. A client that + // resolves the root per operation reads and promotes here. + repoint(alias, storeNew); + const raw = signedRelayRecord(AGENT, "repointed", { branchId: BRANCH, id: ID }); + const source = seed(storeNew, "r.json", raw); + for (const dir of ["cur", "dlq", "outbox"]) mkdirSync(join(storeNew, AGENT, dir), { recursive: true }); + + // A real second process holds the new root's mailbox lock. + const holder = await holdLock(mailboxLockRoot(storeNew, AGENT)); + try { + // It must not promote under the old root's lock. + expect(await mail.checkNewMail()).toEqual([]); + expect(existsSync(source)).toBe(true); + expect(existsSync(join(storeNew, AGENT, "cur", "r.json"))).toBe(false); + } finally { + await holder.release(); + } + + // After release the one envelope is delivered exactly once. + expect((await mail.checkNewMail()).map((message) => message.filename)).toEqual(["r.json"]); + expect(existsSync(source)).toBe(false); + expect(readFileSync(join(storeNew, AGENT, "cur", "r.json"), "utf8")).toBe(raw); + + // The same signed envelope presented again under a new file name reaches the + // replay gate in the new root, which refuses it: no delivery, and the record + // is dead-lettered with a `replay` sidecar. + const again = seed(storeNew, "r-again.json", raw); + expect(await mail.checkNewMail()).toEqual([]); + expect(existsSync(again)).toBe(false); + expect(existsSync(join(storeNew, AGENT, "cur", "r-again.json"))).toBe(false); + expect(readFileSync(join(storeNew, AGENT, "dlq", "r-again.json"), "utf8")).toBe(raw); + expect(readFileSync(join(storeNew, AGENT, "dlq", "r-again.json.reason"), "utf8").split("\n")[0]).toBe("class: replay"); +}, 20_000); + +test("a mail-root repoint during verification keeps the promotion's acceptance stripe on the root the operation started with", async () => { + const verifier = pausedVerifier(); + const mail = new MailClient(alias, undefined, AGENT, verifier.client); + const source = seed(storeOld, "r.json", signedRelayRecord(AGENT, "stripe", { branchId: BRANCH, id: ID })); + + // A real second process holds the delivery's acceptance stripe on the OLD + // root; the same stripe on the new root is free. + const holder = await holdLock(acceptanceLockRoot(storeOld, BRANCH, ID)); + try { + const delivered = await checkWithChangeDuringVerification(mail, verifier, () => repoint(alias, storeNew)); + expect(delivered).toEqual([]); + expect(existsSync(source)).toBe(true); + expect(existsSync(join(storeOld, AGENT, "cur", "r.json"))).toBe(false); + expect(existsSync(join(storeNew, AGENT, "cur", "r.json"))).toBe(false); + } finally { + await holder.release(); + } + + // Control: with the stripe free, the record is deliverable from the old root. + repoint(alias, storeOld); + expect((await mail.checkNewMail()).map((message) => message.filename)).toEqual(["r.json"]); + expect(existsSync(join(storeOld, AGENT, "cur", "r.json"))).toBe(true); +}, 20_000); + +test("a mail-root repoint during verification keeps trust classification on the root the operation started with", async () => { + expect(bridgePrincipalIds(storeOld).has("sender")).toBe(true); + expect(bridgePrincipalIds(storeNew).has("sender")).toBe(false); + const verifier = pausedVerifier(); + const mail = new MailClient(alias, undefined, AGENT, verifier.client); + const raw = signedRelayRecord(AGENT, "trust", { branchId: BRANCH, id: ID }); + seed(storeOld, "r.json", raw); + + const delivered = await checkWithChangeDuringVerification(mail, verifier, () => repoint(alias, storeNew)); + expect(delivered.map((message) => [message.filename, message.trustTier])).toEqual([["r.json", "external"]]); + expect(readFileSync(join(storeOld, AGENT, "cur", "r.json"), "utf8")).toBe(raw); + expect(existsSync(join(storeNew, AGENT, "cur", "r.json"))).toBe(false); +}); + +test("a mail-root repoint during verification keeps the rejection's acceptance stripe on the root the operation started with", async () => { + const verifier = pausedVerifier({ wrongKey: true }); + const mail = new MailClient(alias, undefined, AGENT, verifier.client); + const raw = signedRelayRecord(AGENT, "rejected", { branchId: BRANCH, id: ID }); + const source = seed(storeOld, "r.json", raw); + + const holder = await holdLock(acceptanceLockRoot(storeOld, BRANCH, ID)); + try { + expect(await checkWithChangeDuringVerification(mail, verifier, () => repoint(alias, storeNew))).toEqual([]); + // The rejection did not move the record under the new root's free stripe. + expect(existsSync(source)).toBe(true); + expect(existsSync(join(storeOld, AGENT, "dlq", "r.json"))).toBe(false); + expect(existsSync(join(storeOld, AGENT, "dlq", "r.json.reason"))).toBe(false); + } finally { + await holder.release(); + } + + // Control: with the stripe free, the same record is dead-lettered as invalid. + repoint(alias, storeOld); + expect(await mail.checkNewMail()).toEqual([]); + expect(existsSync(source)).toBe(false); + expect(readFileSync(join(storeOld, AGENT, "dlq", "r.json"), "utf8")).toBe(raw); + expect(readFileSync(join(storeOld, AGENT, "dlq", "r.json.reason"), "utf8").split("\n")[0]).toBe("class: invalid"); +}, 20_000); + +test("interrupted-placement recovery classifies trust on the root the operation started with", async () => { + const verifier = pausedVerifier(); + const mail = new MailClient(alias, undefined, AGENT, verifier.client); + const mailbox = join(storeOld, AGENT); + const source = seed(storeOld, "m1.json", signedRelayRecord(AGENT, "recovered", { branchId: BRANCH, id: ID })); + // A placement interrupted after the move into cur/: recovery verifies the cur + // copy, classifies its trust and completes it. + mailboxReplayStore(mailbox).beginPlacement("recovered", "m1.json", source); + renameSync(source, join(mailbox, "cur", "m1.json")); + + const delivered = await checkWithChangeDuringVerification(mail, verifier, () => repoint(alias, storeNew)); + expect(delivered.map((message) => [message.filename, message.trustTier])).toEqual([["m1.json", "external"]]); + expect(hasCommittedMessageId(mailbox, "recovered")).toBe(true); +}); + +test("a mailbox-directory repoint during verification keeps the operation on the directory it started with", async () => { + const store = join(root, "store"); + const boxOld = join(root, "box-old"); + const boxNew = join(root, "box-new"); + for (const box of [boxOld, boxNew]) { + for (const dir of ["new", "cur", "dlq", "outbox"]) mkdirSync(join(box, dir), { recursive: true }); + } + mkdirSync(store, { recursive: true }); + const link = join(store, AGENT); + symlinkSync(boxOld, link, "dir"); + const verifier = pausedVerifier(); + const mail = new MailClient(store, undefined, AGENT, verifier.client); + const raw = signedRelayRecord(AGENT, "box", { branchId: BRANCH, id: ID }); + writeFileSync(join(boxOld, "new", "r.json"), raw); + + const delivered = await checkWithChangeDuringVerification(mail, verifier, () => repoint(link, boxNew)); + expect(delivered.map((message) => message.filename)).toEqual(["r.json"]); + expect(existsSync(join(boxOld, "new", "r.json"))).toBe(false); + expect(readFileSync(join(boxOld, "cur", "r.json"), "utf8")).toBe(raw); + expect(hasCommittedMessageId(boxOld, "box")).toBe(true); + expect(existsSync(join(boxNew, "cur", "r.json"))).toBe(false); + expect(hasCommittedMessageId(boxNew, "box")).toBe(false); +}); + +test("deliverOutbox moves a record into the recipient inbox under the root the operation started with", async () => { + const mail = new MailClient(alias, undefined, AGENT, senderKeyResolver()); + const sentAt = "2026-10-13T00:00:00.000Z"; + writeFileSync(join(storeOld, AGENT, "outbox", "o.json"), JSON.stringify({ to: "peer", body: "x", sentAt })); + const dest = "2026-10-13T00-00-00-000Z-o.json"; + + // Repoint the mail root after the outbox record is read and before it is moved. + const read = fs.readFileSync; + let repointed = false; + const spy = spyOn(fs, "readFileSync").mockImplementation(((...args: Parameters) => { + if (!repointed && String(args[0]).endsWith(join("outbox", "o.json"))) { + repointed = true; + repoint(alias, storeNew); + } + return read(...args); + }) as typeof read); + try { + mail.deliverOutbox(); + } finally { + spy.mockRestore(); + } + + expect(repointed).toBe(true); + expect(existsSync(join(storeOld, "peer", "new", dest))).toBe(true); + expect(existsSync(join(storeNew, "peer", "new", dest))).toBe(false); +}); diff --git a/packages/agent/test/mail-interrupted-recovery.test.ts b/packages/agent/test/mail-interrupted-recovery.test.ts index 048c094a..07ecbfff 100644 --- a/packages/agent/test/mail-interrupted-recovery.test.ts +++ b/packages/agent/test/mail-interrupted-recovery.test.ts @@ -10,6 +10,7 @@ import * as ed from "@noble/ed25519"; import { MailClient } from "../src/io/mail.js"; import { hasCommittedMessageId, mailboxReplayStore } from "../src/lib/mailbox-policy.js"; import { acquireMailLock } from "../src/lib/mail-lock.js"; +import { mailboxLockRoot } from "../src/lib/mail-lock-namespace.js"; import { signEnvelope, type FlairClient } from "../src/lib/signEnvelope.js"; const AGENT = "agent-a"; @@ -232,8 +233,8 @@ describe("MailClient defers unresolved placements", () => { const client = new MailClient(mailDir, undefined, AGENT, verifier()); mailboxReplayStore(mailboxRoot).beginPlacement(messageId, "m1.json", source); copyFileSync(source, curPath()); - await (client as unknown as { finishUnlinkedPlacement(id: string, file: string): Promise }) - .finishUnlinkedPlacement(messageId, "m1.json"); + await (client as unknown as { finishUnlinkedPlacement(id: string, file: string, root: string): Promise }) + .finishUnlinkedPlacement(messageId, "m1.json", mailboxLockRoot(mailDir, AGENT)); expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true); }); diff --git a/packages/agent/test/mail-promote-guard.test.ts b/packages/agent/test/mail-promote-guard.test.ts index f59e4b44..b64ffec6 100644 --- a/packages/agent/test/mail-promote-guard.test.ts +++ b/packages/agent/test/mail-promote-guard.test.ts @@ -546,7 +546,7 @@ describe("agent MailClient promotion is fail-closed (cli#380 F1)", () => { return rename(...args); }); try { - await expect(client["commitToCur"]("m1.json", join(inbox("new"), "m1.json"), + await expect(client["commitToCur"](client["operationRoots"](), "m1.json", join(inbox("new"), "m1.json"), readFileSync(join(inbox("new"), "m1.json"), "utf-8"), env)).rejects.toThrow(/append fault.*rollback fault/); expect(readFileSync(join(tmpDir, AGENT, "consumed.jsonl"), "utf-8")).toBe(""); } finally { diff --git a/packages/cli/src/utils/mail.ts b/packages/cli/src/utils/mail.ts index 2170f5a3..20cf51c6 100644 --- a/packages/cli/src/utils/mail.ts +++ b/packages/cli/src/utils/mail.ts @@ -178,7 +178,7 @@ export function relayAcceptRoot(agent: string): string { export function relayAcceptLockRoot(agent: string): string { // cli#616: the lock root is derived under the mail store's resolved real - // path, as MailClient promotion derives its mailbox lock root. + // path. try { assertValidAgentId(agent); return mailboxLockRoot(mailDirPath(), agent);