diff --git a/.changelog/unreleased/changed-617-slice-b-delivery-api.md b/.changelog/unreleased/changed-617-slice-b-delivery-api.md new file mode 100644 index 00000000..43d5a6f6 --- /dev/null +++ b/.changelog/unreleased/changed-617-slice-b-delivery-api.md @@ -0,0 +1,14 @@ +- **MailClient gains a per-message delivery API; a committed message stays re-presentable until completed (Refs #617, slice b).** + + Nothing in the agent runtime calls the new API yet, so delivery is + unchanged until slice (c) wires it. The API commits one message and leaves + its placement intent in the `in_progress` phase, removed by an explicit + completion. A committed message is re-presented until then, with a liveness + bound: after five presentations without completion, or 24 hours after the + first, the intent goes terminal `failed`, the payload is kept and one log + line names the mailbox, message ID and presentation count. + + `recoverPromoted` now reads the placement intent's phase and withholds a + record that resolves to anything but `placing` with a retryable `busy` + refusal, so `tps mail ack` and `nack` and the bridge and plugin startup + sweeps do not present another consumer's in-flight delivery. diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index fa3ebb2d..580f501a 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -54,8 +54,9 @@ export { decideEnvelopeForMailbox, isTopicRecipient, mailboxReplayStore, peekConsumedForMailboxRoot, hasCommittedMessageId, pendingPlacementEntries, placeCurRecord, + MAX_DELIVERY_PRESENTATIONS, DELIVERY_PRESENTATION_TTL_MS, } from "./lib/mailbox-policy.js"; -export type { MailboxPolicyRejectClass, MailboxPolicyResult, PlacementPhase, ReplayStore, FirstDelivery } from "./lib/mailbox-policy.js"; +export type { MailboxPolicyRejectClass, MailboxPolicyResult, PlacementPhase, ReplayStore, FirstDelivery, PresentationAdvance } from "./lib/mailbox-policy.js"; export { acquireMailLock, acquireMailLockSync, MAIL_LOCK_DIR, mailLockPath, processStartToken } from "./lib/mail-lock.js"; export type { MailLock } from "./lib/mail-lock.js"; export { diff --git a/packages/agent/src/io/mail.ts b/packages/agent/src/io/mail.ts index 2e116092..e3476c94 100644 --- a/packages/agent/src/io/mail.ts +++ b/packages/agent/src/io/mail.ts @@ -79,8 +79,12 @@ function sameFileByDeviceInode(a: string, b: string): boolean { * `unknown`: an unsupported value in a recognized field) is never presented or * finished. A phase seen in the listing skips the record before its payload * is read, and the under-lock recheck catches a phase that changed after the - * payload was read and verified. Slice (b) defines the re-presentation of - * `in_progress`. + * payload was read and verified. + * + * This is the BATCH reader (`checkNewMail`), which stays at-most-once until + * slice (c) wires the per-message delivery API. Slice (b)'s `deliverNext` + * re-presents an `in_progress` intent through its own completion-gated path + * (`recoverDelivery`), not this one. */ function resumableOnRecovery(phase: PlacementPhase | null): boolean { return phase === "placing"; @@ -242,12 +246,221 @@ export class MailClient { return messages; } + /** + * cli#617 slice (b): the per-message delivery API, inert until slice (c) + * wires the event loop. It returns at most ONE message: + * + * 1. a committed message whose placement intent is live — a `placing` + * intent finished to `in_progress`, or an `in_progress` re-presentation. + * Recovery is completion-gated, and the liveness bound turns a message + * that is never completed into a terminal `failed` intent; + * 2. otherwise the next verified `new/` message, committed to `cur/` with + * its intent left in the `in_progress` phase. Later messages stay in + * `new/`. + * + * The caller finalizes with completeDelivery() after the handler + * completes; a crash or stop before that re-presents the same message on the + * next call. The delivery key is mailbox + messageId. + */ + async deliverNext(): Promise { + const recovered = await this.recoverDelivery(); + if (recovered) return recovered; + return this.commitNextForDelivery(); + } + + /** + * cli#617 slice (b): the completion-only finalize for a message handed over + * by deliverNext(). Deletes the message's `in_progress` intent in its own + * synchronous mailbox-lock section; returns whether an intent was removed. + * Idempotent: a message already finalized leaves nothing to delete. + */ + async completeDelivery(message: MailMessage): Promise { + const messageId = message.verifiedEnvelope?.messageId; + if (typeof messageId !== "string" || messageId === "") { + throw new Error("completeDelivery requires a delivered message carrying its verified envelope"); + } + const lock = await acquireMailLock(this.mailboxRoot); + if (!lock) throw new Error("mailbox lock busy; delivery not finalized"); + try { + return mailboxReplayStore(this.mailboxRoot).completePlacement(messageId, message.filename); + } finally { + lock.release(); + } + } + + /** + * cli#617 slice (b): completion-gated recovery for deliverNext(). A `placing` + * intent is finished and moved to `in_progress`; an `in_progress` intent is + * re-presented; `unknown` is skipped (never presented or finished). Returns + * the first message presented, or null. + */ + private async recoverDelivery(): Promise { + let pending: Array<{ messageId: string; file: string; phase: PlacementPhase }>; + let listing: MailLock | null = null; + try { + listing = await acquireMailLock(this.mailboxRoot); + if (!listing) return null; + pending = pendingPlacementEntries(this.mailboxRoot); + } catch (err) { + this.emitRecoveryError(err); + return null; + } finally { + listing?.release(); + } + for (const { messageId, file, phase } of pending) { + if (phase === "unknown") continue; + try { + const presented = await this.presentPendingDelivery(messageId, file); + if (presented) return presented; + } catch (err) { + this.emitRecoveryError(err); + } + } + return null; + } + + /** Re-verify a pending cur/ copy and present it under the lock, honoring the liveness bound. */ + private async presentPendingDelivery(messageId: string, file: string): Promise { + const curPath = join(this.inboxCur, file); + let bytes: Buffer; + try { + bytes = readFileSync(curPath); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; + // Same unlinked-placement cleanup as slice (a): finish only a `placing` + // intent whose cur/ copy is absent. + await this.finishUnlinkedPlacement(messageId, file); + return null; + } + const outcome = await this.verifyMailBody(bytes.toString("utf8")); + if (!outcome.pass || outcome.envelope.messageId !== messageId) return null; + const trustTier = verifiedMailTier(outcome.envelope, this.mailDir); + return this.presentUnderLock(messageId, file, bytes, outcome.envelope, trustTier); + } + + /** + * Re-read the intent, identity and raw bytes under the lock (the phase may + * have changed since the listing), advance the presentation and return the + * message. A bound hit writes the terminal `failed` phase and the named log + * line, and presents nothing. + */ + private async presentUnderLock( + messageId: string, + file: string, + bytes: Buffer, + envelope: Envelope, + trustTier: MailMessage["trustTier"], + ): Promise { + const lock = await acquireMailLock(this.mailboxRoot); + if (!lock) throw new Error("mailbox lock busy; delivery recovery deferred"); + try { + const replay = mailboxReplayStore(this.mailboxRoot); + if (!replay.hasPendingPlacement(messageId, file)) return null; + const phase = replay.placementPhase(file); + if (phase === null || phase === "unknown") return null; + if (envelope.messageId !== messageId) return null; + if (!replay.ownsPendingPlacement(messageId, file)) return null; + let current: Buffer; + try { + current = readFileSync(join(this.inboxCur, file)); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") return null; + throw err; + } + if (!current.equals(bytes)) return null; + if (!replay.isConsumed(messageId, file)) replay.recordConsumed(messageId); + const advance = replay.advancePresentation(messageId, file, new Date().toISOString()); + if (advance.status === "absent") return null; + if (advance.status === "failed") { + console.warn( + `[MailClient] placement quarantined for mailbox ${this.agentId}: message ${messageId} not completed ` + + `after ${advance.presentations} presentation(s); payload retained for the operator`, + ); + return null; + } + return { + filename: file, + body: bytes.toString("utf8"), + receivedAt: new Date(), + headers: {}, + from: envelope.from, + verifiedEnvelope: envelope, + trustTier, + }; + } finally { + lock.release(); + } + } + + /** Commit the next verifiable new/ message to cur/ with a live `in_progress` intent, or null. */ + private async commitNextForDelivery(): Promise { + if (!existsSync(this.inboxNew)) return null; + const files = readdirSync(this.inboxNew) + .filter((f) => !f.startsWith(".") && !f.includes("/") && !f.includes("\\")) + .sort(); + for (const file of files) { + const started = Date.now(); + const srcPath = join(this.inboxNew, file); + let body: string; + try { + body = readFileSync(srcPath, "utf-8"); + } catch (err) { + this.emitRecoveryError(err); + continue; + } + let verifyResult: VerifyOutcome; + try { + verifyResult = await this.verifyMailBody(body); + } catch (err) { + this.emitRecoveryError(err); // leave in new/ for a later call + continue; + } + if (!verifyResult.pass) { + this.deadLetter(file, srcPath, verifyResult, started); + continue; + } + let trustTier: MailMessage["trustTier"]; + try { + trustTier = verifiedMailTier(verifyResult.envelope, this.mailDir); + } catch (err) { + this.emitRecoveryError(err); + continue; + } + try { + const committed = await this.commitToCur(file, srcPath, body, verifyResult.envelope, true); + if (committed === "already-delivered") continue; + if (committed) { + this.deadLetter(file, srcPath, { ...committed, from: verifyResult.envelope.from }, started); + continue; + } + const from = verifyResult.envelope.from; + this.events?.emit({ + type: "mail.receive", agent: this.agentId, status: "ok", from, durationMs: Date.now() - started, + }); + return { + filename: file, body, receivedAt: new Date(), headers: {}, from, + verifiedEnvelope: verifyResult.envelope, trustTier, + }; + } catch (err) { + this.emitRecoveryError(err); + } + } + return null; + } + + private emitRecoveryError(err: unknown): void { + this.events?.emit({ + type: "mail.receive", agent: this.agentId, status: "error", from: "unknown", durationMs: 0, error: sanitizeError(err), + }); + } + /** Commit under the mailbox lock; attempt placement rollback on append failure. */ private async commitToCur( file: string, srcPath: string, body: string, envelope: Envelope, + keepInProgress = false, ): Promise<{ pass: false; class: MailboxRejectClass; reason: string } | "already-delivered" | null> { const lock: MailLock | null = await acquireMailLock(this.mailboxLockRoot); if (!lock) throw new Error("mailbox lock busy; not promoted"); @@ -294,7 +507,15 @@ export class MailClient { } throw err; } - replay.finishPlacement(envelope.messageId, file); + if (keepInProgress) { + // cli#617 slice (b): the delivery API leaves the intent live, in the + // `in_progress` phase (presentation 1), for the caller to finalize on + // completion. The batch checkNewMail() path finalizes as before. + const advance = replay.advancePresentation(envelope.messageId, file, new Date().toISOString()); + if (advance.status !== "presented") throw new Error(`placement not advanced after commit: ${advance.status}`); + } else { + replay.finishPlacement(envelope.messageId, file); + } return null; } finally { relayLock.release(); diff --git a/packages/agent/src/lib/mailbox-policy.ts b/packages/agent/src/lib/mailbox-policy.ts index 9f3f199d..a73d5aba 100644 --- a/packages/agent/src/lib/mailbox-policy.ts +++ b/packages/agent/src/lib/mailbox-policy.ts @@ -216,6 +216,14 @@ export async function decideEnvelopeForMailbox( const CONSUMED_LEDGER_FILE = "consumed.jsonl"; const CONSUMED_LEDGER_RETENTION_MS = 180 * 24 * 60 * 60 * 1000; +/** + * cli#617 slice (b): the delivery liveness bound. After this many presentations + * of one committed message without completion, or this long after the first + * presentation, the intent goes terminal `failed`. + */ +export const MAX_DELIVERY_PRESENTATIONS = 5; +export const DELIVERY_PRESENTATION_TTL_MS = 24 * 60 * 60 * 1000; + /** * The replay gate and placement intents for one mailbox. Call while holding * that mailbox's lock (acquireMailLock on the same root). @@ -247,8 +255,30 @@ export interface ReplayStore { * on failure, and the caller must then roll the move back. */ recordConsumed(messageId: string): void; + /** + * cli#617 slice (b): advance a live (`placing` or `in_progress`) intent for + * this message to its next presentation. A `placing` intent becomes + * `in_progress` at presentation 1; an `in_progress` intent's count increments. + * Returns `absent` when no live intent for this message is present (missing, + * another message, or a `failed`/`unknown` phase), `presented` with the new + * count, or `failed` when the liveness bound (MAX_DELIVERY_PRESENTATIONS or + * DELIVERY_PRESENTATION_TTL_MS) is reached — the intent is then terminal + * `failed` and the caller must not present. Call under the mailbox lock. + */ + advancePresentation(messageId: string, file: string, nowIso: string): PresentationAdvance; + /** + * cli#617 slice (b): completion-only finalize. Deletes only an `in_progress` + * intent for this message; returns whether an intent was removed. + */ + completePlacement(messageId: string, file: string): boolean; } +/** The result of advancing a placement intent to its next presentation (cli#617 slice b). */ +export type PresentationAdvance = + | { status: "absent" } + | { status: "presented"; presentations: number } + | { status: "failed"; presentations: number }; + /** The durable ReplayStore for the mailbox rooted at `root` (`/`). */ export function mailboxReplayStore(root: string): ReplayStore { return { @@ -278,6 +308,8 @@ export function mailboxReplayStore(root: string): ReplayStore { syncDirectory(root); }, recordConsumed: (messageId) => recordConsumedMessageId(root, messageId), + advancePresentation: (messageId, file, nowIso) => advancePlacementPresentation(root, messageId, file, nowIso), + completePlacement: (messageId, file) => completeInProgressPlacement(root, messageId, file), }; } @@ -554,6 +586,83 @@ function writePlacementIntent(root: string, messageId: string, file: string, sou } } +/** + * cli#617 slice (b): rewrite the live intent for `file` in place, keeping its + * messageId and recorded identity. Refuses (returns false) unless a `placing` + * or `in_progress` intent for this message is present; a `failed` or `unknown` + * intent, or another message's intent, is never replaced. The write is the + * temp-file, fsync, rename, directory-fsync sequence used everywhere else. A + * `.phase.tmp` suffix marks the phase transition for the crash-boundary + * fixture (mail-kill-child). + */ +function writePlacementPhase(root: string, messageId: string, file: string, fields: Record): boolean { + const path = placementIntentPath(root, file); + const existing = readPlacementIntent(root, file); + if (!existing || existing.messageId !== messageId) return false; + if (existing.phase !== "placing" && existing.phase !== "in_progress") return false; + const next: Record = { + messageId, + file, + ...(existing.identity ? { identity: existing.identity } : {}), + ...fields, + }; + const tmp = `${path}.${randomUUID()}.phase.tmp`; + const fd = openSync(tmp, "wx", 0o600); + try { + try { + writeFileSync(fd, JSON.stringify(next)); + fsyncSync(fd); + } finally { closeSync(fd); } + renameSync(tmp, path); + syncDirectory(root); + } finally { + rmSync(tmp, { force: true }); + } + return true; +} + +/** + * cli#617 slice (b): advance a live intent to its next presentation, or to the + * terminal `failed` phase when the liveness bound is reached. A `placing` + * intent becomes `in_progress` at presentation 1 (the first hand-over, counted + * before it); an `in_progress` intent's count increments. `failed` and + * `unknown` are not live, so they advance to `absent`. + */ +function advancePlacementPresentation(root: string, messageId: string, file: string, nowIso: string): PresentationAdvance { + const intent = readPlacementIntent(root, file); + if (!intent || intent.messageId !== messageId) return { status: "absent" }; + if (intent.phase === "placing") { + const written = writePlacementPhase(root, messageId, file, { + version: PLACEMENT_INTENT_VERSION, phase: "in_progress", presentations: 1, firstPresentedAt: nowIso, + }); + return written ? { status: "presented", presentations: 1 } : { status: "absent" }; + } + if (intent.phase !== "in_progress") return { status: "absent" }; + const previous = intent.presentations ?? 0; + const firstPresentedAt = intent.firstPresentedAt ?? nowIso; + const next = previous + 1; + const tooOld = Date.parse(nowIso) - Date.parse(firstPresentedAt) > DELIVERY_PRESENTATION_TTL_MS; + if (next > MAX_DELIVERY_PRESENTATIONS || tooOld) { + writePlacementPhase(root, messageId, file, { + version: PLACEMENT_INTENT_VERSION, phase: "failed", presentations: previous, firstPresentedAt, + }); + return { status: "failed", presentations: previous }; + } + writePlacementPhase(root, messageId, file, { + version: PLACEMENT_INTENT_VERSION, phase: "in_progress", presentations: next, firstPresentedAt, + }); + return { status: "presented", presentations: next }; +} + +/** cli#617 slice (b): completion-only finalize — delete only an `in_progress` intent for this message. */ +function completeInProgressPlacement(root: string, messageId: string, file: string): boolean { + const intent = readPlacementIntent(root, file); + if (!intent || intent.messageId !== messageId || intent.phase !== "in_progress") return false; + unlinkSync(placementIntentPath(root, file)); + syncDirectory(root); + return true; +} + function pendingPlacementIds(root: string, quarantine = false): { ids: Set; blocked: Set; retainAll: boolean } { const ids = new Set(); const blocked = new Set(); diff --git a/packages/agent/test/helpers/mail-kill-child.mjs b/packages/agent/test/helpers/mail-kill-child.mjs index 239c9e31..9588e92a 100644 --- a/packages/agent/test/helpers/mail-kill-child.mjs +++ b/packages/agent/test/helpers/mail-kill-child.mjs @@ -11,7 +11,9 @@ // // argv: // env: TPS_MAIL_MODULE file:// URL of the mail module +// TPS_DRIVE "checkNewMail" (default) or "deliverNext" // TPS_KILL_AT symbolic step boundary (see BOUNDARIES below) +// TPS_KILL_AFTER "1" pauses AFTER the matched op completes // TPS_KILL_MARKER marker file written just before blocking // TPS_MAIL_ROOT the mailbox root (/) // TPS_SRC, TPS_CUR the new/ source and its cur/ sibling @@ -25,6 +27,7 @@ const fs = require("node:fs"); const origWrite = fs.writeFileSync; const KILL_AT = process.env.TPS_KILL_AT ?? ""; +const KILL_AFTER = process.env.TPS_KILL_AFTER === "1"; const MARKER = process.env.TPS_KILL_MARKER; const ROOT = process.env.TPS_MAIL_ROOT ?? ""; const SRC = process.env.TPS_SRC ?? ""; @@ -32,6 +35,8 @@ const CUR = process.env.TPS_CUR ?? ""; const LEDGER = join(ROOT, "consumed.jsonl"); const isIntent = (p) => typeof p === "string" && basename(p).startsWith(".placement-") && p.endsWith(".json"); +// The phase transition writes the intent via a `.phase.tmp` scratch file. +const isPhaseScratch = (p) => typeof p === "string" && p.endsWith(".phase.tmp"); // fn -> predicate over the call's arguments: true means "pause HERE". const BOUNDARIES = { @@ -40,6 +45,8 @@ const BOUNDARIES = { "source-removal": (name, args) => name === "unlinkSync" && args[0] === SRC, "ledger-commit": (name, args) => name === "appendFileSync" && args[0] === LEDGER, "intent-finish": (name, args) => name === "unlinkSync" && isIntent(args[0]), + // cli#617 slice (b): the `in_progress` phase write (placing -> in_progress). + "phase-write": (name, args) => name === "renameSync" && isIntent(args[1]) && isPhaseScratch(args[0]), }; const matches = BOUNDARIES[KILL_AT]; if (!matches) { @@ -48,18 +55,24 @@ if (!matches) { } let paused = false; +function pause() { + paused = true; + if (MARKER) origWrite.call(fs, MARKER, "paused"); + const deadline = Date.now() + 60_000; + while (Date.now() < deadline) { + /* spin until killed */ + } + process.exit(3); +} for (const name of ["writeFileSync", "openSync", "linkSync", "appendFileSync", "rmSync", "unlinkSync", "renameSync"]) { const orig = fs[name]; fs[name] = function (...args) { - if (!paused && matches(name, args)) { - paused = true; - if (MARKER) origWrite.call(fs, MARKER, "paused"); - const deadline = Date.now() + 60_000; - while (Date.now() < deadline) { - /* spin until killed */ - } - process.exit(3); + if (KILL_AFTER) { + const result = orig.apply(this, args); + if (!paused && matches(name, args)) pause(); + return result; } + if (!paused && matches(name, args)) pause(); return orig.apply(this, args); }; } @@ -74,6 +87,12 @@ const client = new mod.MailClient(mailDir, undefined, agentId, { return name === from ? { publicKey: Buffer.from(ed.getPublicKey(seed)) } : null; }, }); -const messages = await client.checkNewMail(); -process.stdout.write(JSON.stringify({ count: messages.length })); -process.exit(0); +let count; +if ((process.env.TPS_DRIVE ?? "checkNewMail") === "deliverNext") { + const message = await client.deliverNext(); + count = message ? 1 : 0; +} else { + count = (await client.checkNewMail()).length; +} +process.stdout.write(JSON.stringify({ count })); +process.exit(0); \ No newline at end of file diff --git a/packages/agent/test/mail-delivery-at-least-once.test.ts b/packages/agent/test/mail-delivery-at-least-once.test.ts new file mode 100644 index 00000000..e5df04f4 --- /dev/null +++ b/packages/agent/test/mail-delivery-at-least-once.test.ts @@ -0,0 +1,333 @@ +/** + * cli#617 slice (b): the per-message delivery API — commits one message and + * leaves its intent live in the `in_progress` phase, finalized only on + * completion, with a liveness bound. Real signed records through the real + * parser and signer; nothing in the runtime calls these methods yet. + */ + +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { existsSync, linkSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import * as ed from "@noble/ed25519"; +import { MailClient } from "../src/io/mail.js"; +import { + MAX_DELIVERY_PRESENTATIONS, + hasCommittedMessageId, + mailboxReplayStore, + pendingPlacementEntries, +} from "../src/lib/mailbox-policy.js"; +import { signEnvelope, type FlairClient } from "../src/lib/signEnvelope.js"; + +const AGENT = "agent-a"; +const FROM = "agent-b"; +const SEED = Buffer.alloc(32, 0x57); +const CHILD = fileURLToPath(new URL("./helpers/mail-kill-child.mjs", import.meta.url)); +const MODULE = new URL("../src/io/mail.ts", import.meta.url).href; + +let mailDir: string; +let root: string; + +const cur = (file: string) => join(root, "cur", file); +const fresh = (file: string) => join(root, "new", file); + +function verifier(): FlairClient { + return { + async getAgent(name: string) { + return name === FROM ? { publicKey: Buffer.from(ed.getPublicKey(SEED)) } : null; + }, + }; +} + +const client = () => new MailClient(mailDir, undefined, AGENT, verifier()); + +beforeEach(() => { + mailDir = mkdtempSync(join(tmpdir(), "mail-delivery-")); + root = join(mailDir, AGENT); + mkdirSync(join(root, "new"), { recursive: true }); + mkdirSync(join(root, "cur"), { recursive: true }); +}); + +afterEach(() => rmSync(mailDir, { recursive: true, force: true })); + +/** Write one signed record to `path`. */ +function signedRecord(path: string, messageId: string, body = "delivery-body"): void { + const timestamp = new Date().toISOString(); + const envelope = signEnvelope( + { + v: 1, from: FROM, to: AGENT, body, messageId, timestamp, + delegationChain: [ + { agent: "system", kind: "human", timestamp, rationale: "originates", signature: null }, + { agent: FROM, kind: "agent", timestamp, rationale: "sends", signature: null }, + ], + }, + { [FROM]: SEED }, + ); + writeFileSync(path, JSON.stringify({ from: FROM, body: JSON.stringify(envelope) })); +} + +const plant = (file: string, messageId: string) => signedRecord(fresh(file), messageId); + +const jsonFiles = (dir: string) => (existsSync(dir) ? readdirSync(dir).filter((n) => n.endsWith(".json")).sort() : []); +const newFiles = () => jsonFiles(join(root, "new")); +const curFiles = () => jsonFiles(join(root, "cur")); + +const textOf = (path: string) => (existsSync(path) ? readFileSync(path, "utf8") : null); +const intentPath = (file: string) => join(root, `.placement-${createHash("sha256").update(file).digest("hex")}.json`); +const intentText = (file: string) => textOf(intentPath(file)); +const intentJson = (file: string) => JSON.parse(readFileSync(intentPath(file), "utf8")); +const intentPhase = (file: string) => mailboxReplayStore(root).placementPhase(file); + +/** A committed message as slice (b) leaves it, with the given intent fields. */ +function setPhase(fields: Record, file = "m1.json"): Buffer { + const path = intentPath(file); + writeFileSync(path, JSON.stringify({ ...JSON.parse(readFileSync(path, "utf8")), ...fields })); + return readFileSync(path); +} + +function committed(messageId: string, fields: Record, file = "m1.json"): Buffer { + signedRecord(fresh(file), messageId); + const replay = mailboxReplayStore(root); + replay.beginPlacement(messageId, file, fresh(file)); + linkSync(fresh(file), cur(file)); + unlinkSync(fresh(file)); + replay.recordConsumed(messageId); + return setPhase(fields, file); +} + +const quarantined = (file: string) => existsSync(`${intentPath(file)}.quarantined`); + +describe("per-message delivery commits one message at a time", () => { + test("later messages stay in new/ until the in-flight one completes", async () => { + plant("m1.json", "id-1"); + plant("m2.json", "id-2"); + plant("m3.json", "id-3"); + + const c = client(); + const first = await c.deliverNext(); + expect(first?.verifiedEnvelope?.messageId).toBe("id-1"); + expect(curFiles()).toEqual(["m1.json"]); + expect(newFiles()).toEqual(["m2.json", "m3.json"]); + + // Not finalized: the next call re-presents the same message (at least once) + // and still does not commit a later one. + const again = await c.deliverNext(); + expect(again?.verifiedEnvelope?.messageId).toBe("id-1"); + expect(newFiles()).toEqual(["m2.json", "m3.json"]); + + expect(await c.completeDelivery(again!)).toBe(true); + expect(intentPhase("m1.json")).toBeNull(); + + const second = await c.deliverNext(); + expect(second?.verifiedEnvelope?.messageId).toBe("id-2"); + expect(curFiles()).toEqual(["m1.json", "m2.json"]); + expect(newFiles()).toEqual(["m3.json"]); + }); + + test("completion is idempotent and stops re-presentation", async () => { + plant("m1.json", "solo"); + const c = client(); + const m = await c.deliverNext(); + expect(await c.completeDelivery(m!)).toBe(true); + expect(await c.completeDelivery(m!)).toBe(false); + expect(await c.deliverNext()).toBeNull(); + }); +}); + +describe("in_progress re-presentation in recovery", () => { + test("a fresh client re-presents a committed message until it is completed", async () => { + plant("m1.json", "resume-in-progress"); + const first = await client().deliverNext(); + expect(first?.verifiedEnvelope?.messageId).toBe("resume-in-progress"); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(intentJson("m1.json").presentations).toBe(1); + + const restarted = client(); + const again = await restarted.deliverNext(); + expect(again?.verifiedEnvelope?.messageId).toBe("resume-in-progress"); + expect(intentJson("m1.json").presentations).toBe(2); + + await restarted.completeDelivery(again!); + expect(await client().deliverNext()).toBeNull(); + }); + + test("a placing intent left by a crash is finished to in_progress and presented", async () => { + plant("m1.json", "resume-placing"); + const replay = mailboxReplayStore(root); + replay.beginPlacement("resume-placing", "m1.json", fresh("m1.json")); + linkSync(fresh("m1.json"), cur("m1.json")); + // No consumed id and no phase yet: a crash between the link and the phase write. + + const c = client(); + const m = await c.deliverNext(); + expect(m?.verifiedEnvelope?.messageId).toBe("resume-placing"); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(hasCommittedMessageId(root, "resume-placing")).toBe(true); + + expect((await c.deliverNext())?.verifiedEnvelope?.messageId).toBe("resume-placing"); + }); +}); + +describe("a duplicate envelope during handling is refused", () => { + test("the same signed envelope under a new name is dead-lettered as replay once committed", async () => { + plant("m1.json", "duplicate"); + const c = client(); + const m = await c.deliverNext(); + expect(await c.completeDelivery(m!)).toBe(true); + + // The same envelope (same messageId) arrives under a different filename. + writeFileSync(fresh("m1-copy.json"), readFileSync(cur("m1.json"))); + expect(await client().deliverNext()).toBeNull(); + expect(textOf(join(root, "dlq", "m1-copy.json.reason")) ?? "").toContain("class: replay"); + }); +}); + +describe("the liveness bound quarantines a message that is never completed", () => { + const FAILED_AT = `not completed after ${MAX_DELIVERY_PRESENTATIONS} presentation(s)`; + + test("after the presentation count the intent goes terminal, a line is logged, and other mail flows", async () => { + plant("m1.json", "poison"); + plant("m2.json", "healthy"); + const warnings: string[] = []; + const warn = spyOn(console, "warn").mockImplementation((...args: unknown[]) => { warnings.push(args.join(" ")); }); + let last: Awaited> = null; + try { + const c = client(); + for (let i = 0; i < MAX_DELIVERY_PRESENTATIONS; i++) { + last = await c.deliverNext(); + expect(last?.verifiedEnvelope?.messageId).toBe("poison"); + } + // The next hand-over would exceed the bound: poison is not presented, its + // intent becomes terminal `failed`, and the next message flows. + const next = await c.deliverNext(); + expect(next?.verifiedEnvelope?.messageId).toBe("healthy"); + } finally { warn.mockRestore(); } + + expect(warnings.some((w) => w.includes("poison") && w.includes(FAILED_AT))).toBe(true); + expect(intentPhase("m1.json")).toBe("failed"); + expect(textOf(cur("m1.json"))).not.toBeNull(); // payload retained + // poison is terminal, so it is not a pending entry; the message that flowed is. + expect(pendingPlacementEntries(root).map((e) => e.messageId)).toEqual(["healthy"]); + expect(mailboxReplayStore(root).hasPendingFile("m1.json")).toBe(false); + expect(hasCommittedMessageId(root, "poison")).toBe(true); + }); + + test("a message past the presentation TTL is quarantined on the next hand-over", async () => { + const stale = new Date(Date.now() - 25 * 60 * 60 * 1000).toISOString(); + committed("stale", { version: 1, phase: "in_progress", presentations: 1, firstPresentedAt: stale }); + plant("m2.json", "healthy"); + + const warnings: string[] = []; + const warn = spyOn(console, "warn").mockImplementation((...args: unknown[]) => { warnings.push(args.join(" ")); }); + let next: Awaited>; + try { next = await client().deliverNext(); } finally { warn.mockRestore(); } + + expect(next?.verifiedEnvelope?.messageId).toBe("healthy"); + expect(intentPhase("m1.json")).toBe("failed"); + expect(warnings.some((w) => w.includes("stale"))).toBe(true); + }); + + test("the terminal intent keeps its message refused by the replay gate", async () => { + const bytes = committed("poisoned", { version: 1, phase: "failed", presentations: 5, firstPresentedAt: new Date().toISOString() }); + expect(intentPhase("m1.json")).toBe("failed"); + // Same signed envelope under a fresh name: refused at commit as a replay. + writeFileSync(fresh("m1.json"), readFileSync(cur("m1.json"))); + expect(await client().deliverNext()).toBeNull(); + expect(textOf(join(root, "dlq", "m1.json.reason")) ?? "").toContain("class: replay"); + expect(intentJson("m1.json").phase).toBe("failed"); + expect(bytes.length).toBeGreaterThan(0); + }); +}); + +describe("an unknown phase is preserved", () => { + test("not presented, not finalized and not quarantined for its value; its file gate stays closed", async () => { + const bytes = committed("kept-unknown", { version: 1, phase: "awaiting-review" }); + expect(await client().deliverNext()).toBeNull(); + expect(intentText("m1.json") ?? "").toBe(bytes.toString("utf8")); + expect(quarantined("m1.json")).toBe(false); + expect(mailboxReplayStore(root).hasPendingFile("m1.json")).toBe(true); + expect(hasCommittedMessageId(root, "kept-unknown")).toBe(true); + }); +}); + +interface KillRun { signal: NodeJS.Signals | null; reached: boolean; stderr: string; } + +function runAndKillAt(killAt: string, file: string, after = false, timeoutMs = 20_000): Promise { + const marker = join(mailDir, `paused.${killAt}`); + const child = spawn(process.execPath, [CHILD, mailDir, AGENT], { + env: { + HOME: mailDir, + PATH: process.env.PATH, + TPS_MAIL_MODULE: MODULE, + TPS_DRIVE: "deliverNext", + TPS_KILL_AT: killAt, + ...(after ? { TPS_KILL_AFTER: "1" } : {}), + TPS_KILL_MARKER: marker, + TPS_MAIL_ROOT: root, + TPS_SRC: fresh(file), + TPS_CUR: cur(file), + TPS_FROM: FROM, + TPS_SEED: SEED.toString("base64"), + }, + stdio: ["ignore", "pipe", "pipe"], + }); + return new Promise((resolve, reject) => { + let stderr = ""; + let settled = false; + child.stderr.on("data", (d) => { stderr += d.toString(); }); + const poll = setInterval(() => { if (existsSync(marker)) { clearInterval(poll); child.kill("SIGKILL"); } }, 15); + const deadline = setTimeout(() => { + if (settled) return; + settled = true; clearInterval(poll); clearTimeout(deadline); + child.kill("SIGKILL"); + reject(new Error(`child never reached ${killAt} (stderr: ${stderr})`)); + }, timeoutMs); + child.once("error", (e) => { if (!settled) { settled = true; clearInterval(poll); clearTimeout(deadline); reject(e); } }); + child.once("exit", (_code, signal) => { + if (settled) return; + settled = true; clearInterval(poll); clearTimeout(deadline); + resolve({ signal, reached: existsSync(marker), stderr }); + }); + }); +} + +describe("a kill mid-delivery re-presents on the next call", () => { + for (const at of ["intent-write", "link-to-cur", "source-removal", "ledger-commit"]) { + test(`killed before ${at}: the next call presents the message, later mail waits`, async () => { + plant("m1.json", `kill-${at}`); + plant("m2.json", "later"); + + const run = await runAndKillAt(at, "m1.json"); + expect(run.reached, `child never reached ${at}; stderr=${run.stderr}`).toBe(true); + expect(run.signal).toBe("SIGKILL"); + + const c = client(); + const m = await c.deliverNext(); + expect(m?.verifiedEnvelope?.messageId).toBe(`kill-${at}`); + expect(newFiles()).toContain("m2.json"); + + expect(await c.completeDelivery(m!)).toBe(true); + const next = await c.deliverNext(); + expect(next?.verifiedEnvelope?.messageId).toBe("later"); + }); + } + + test("killed after the intent becomes in_progress: the next call re-presents it and later mail waits", async () => { + plant("m1.json", "after-in-progress"); + plant("m2.json", "later"); + + const run = await runAndKillAt("phase-write", "m1.json", true); + expect(run.reached, `child never reached the phase write; stderr=${run.stderr}`).toBe(true); + expect(run.signal).toBe("SIGKILL"); + expect(intentPhase("m1.json")).toBe("in_progress"); + + const c = client(); + const m = await c.deliverNext(); + expect(m?.verifiedEnvelope?.messageId).toBe("after-in-progress"); + expect(intentJson("m1.json").presentations).toBe(2); + expect(newFiles()).toContain("m2.json"); + }); +}); diff --git a/packages/agent/test/mail-interrupted-recovery.test.ts b/packages/agent/test/mail-interrupted-recovery.test.ts index 048c094a..3ab5f2fe 100644 --- a/packages/agent/test/mail-interrupted-recovery.test.ts +++ b/packages/agent/test/mail-interrupted-recovery.test.ts @@ -1,4 +1,11 @@ -/** Real filesystem crash-boundary and unresolved-intent regressions (cli#600). */ +/** + * Real filesystem crash-boundary and unresolved-intent regressions (cli#600). + * + * cli#617 slice (b): the recovery these tests exercise is the per-message + * delivery API's completion-gated path. A message delivered by deliverNext() + * stays re-presentable until completeDelivery() finalizes it, so the old + * at-most-once assertions become "until completed". + */ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { spawn } from "node:child_process"; @@ -76,15 +83,17 @@ interface KillRun { stderr: string; } -/** Run checkNewMail() in a child and SIGKILL it at `killAt`; resolve on exit. */ -function runAndKillAt(killAt: string, source: string, timeoutMs = 20_000): Promise { +/** Run deliverNext() in a child and SIGKILL it at `killAt`; resolve on exit. */ +function runAndKillAt(killAt: string, source: string, after = false, timeoutMs = 20_000): Promise { const marker = join(mailDir, `paused.${killAt}`); const child = spawn(process.execPath, [CHILD, mailDir, AGENT], { env: { HOME: mailDir, PATH: process.env.PATH, TPS_MAIL_MODULE: MODULE, + TPS_DRIVE: "deliverNext", TPS_KILL_AT: killAt, + ...(after ? { TPS_KILL_AFTER: "1" } : {}), TPS_KILL_MARKER: marker, TPS_MAIL_ROOT: mailboxRoot, TPS_SRC: source, @@ -130,14 +139,19 @@ function runAndKillAt(killAt: string, source: string, timeoutMs = 20_000): Promi }); } -/** Check consecutive calls from a fresh client and placement cleanup. */ -async function assertDeliveredExactlyOnce(messageId: string): Promise { +/** + * The cli#617 slice (b) contract: a delivered message is re-presented on the + * next call until it is completed, then never again; placement cleanup is done. + */ +async function assertDeliveredUntilCompleted(messageId: string): Promise { const client = new MailClient(mailDir, undefined, AGENT, verifier()); - const first = await client.checkNewMail(); - const second = await client.checkNewMail(); + const first = await client.deliverNext(); + const second = await client.deliverNext(); - expect(first.map((m) => m.verifiedEnvelope?.messageId)).toEqual([messageId]); - expect(second).toEqual([]); + expect(first?.verifiedEnvelope?.messageId).toBe(messageId); + expect(second?.verifiedEnvelope?.messageId).toBe(messageId); + expect(await client.completeDelivery(second!)).toBe(true); + expect(await client.deliverNext()).toBeNull(); expect(existsSync(newPath())).toBe(false); expect(jsonFiles(join(mailboxRoot, "cur"))).toEqual(["m1.json"]); expect(jsonFiles(join(mailboxRoot, "dlq"))).toEqual([]); @@ -147,17 +161,23 @@ async function assertDeliveredExactlyOnce(messageId: string): Promise { ).toBe(false); } -const CASES = ["intent-write", "link-to-cur", "source-removal", "ledger-commit", "intent-finish"]; +const CASES: Array<{ at: string; after?: boolean }> = [ + { at: "intent-write" }, + { at: "link-to-cur" }, + { at: "source-removal" }, + { at: "ledger-commit" }, + { at: "phase-write", after: true }, +]; describe("MailClient recovery at the tested placement boundaries (cli#600)", () => { - for (const at of CASES) { - test(`a later check returns the message once after a kill before ${at}`, async () => { + for (const { at, after } of CASES) { + test(`a later call re-presents the message until completed after a kill at ${at}`, async () => { const messageId = `kill-${at}`; const source = plant(messageId); - const run = await runAndKillAt(at, source); + const run = await runAndKillAt(at, source, after); expect(run.reached, `child never reached ${at}; stderr=${run.stderr}`).toBe(true); expect(run.signal).toBe("SIGKILL"); - await assertDeliveredExactlyOnce(messageId); + await assertDeliveredUntilCompleted(messageId); }); } }); @@ -175,8 +195,8 @@ describe("MailClient defers unresolved placements", () => { const intentPath = join(mailboxRoot, readdirSync(mailboxRoot).find((n) => n.startsWith(".placement-") && n.endsWith(".json"))!); const intent = readFileSync(intentPath); - expect(await client.checkNewMail()).toEqual([]); - expect(await client.checkNewMail()).toEqual([]); + expect(await client.deliverNext()).toBeNull(); + expect(await client.deliverNext()).toBeNull(); expect(readFileSync(intentPath).equals(intent)).toBe(true); expect(existsSync(source)).toBe(true); expect(hasCommittedMessageId(mailboxRoot, messageId)).toBe(false); @@ -209,10 +229,10 @@ describe("MailClient defers unresolved placements", () => { }, }); - expect(await client.checkNewMail()).toEqual([]); + expect(await client.deliverNext()).toBeNull(); expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true); expect(hasCommittedMessageId(mailboxRoot, messageId)).toBe(false); - await assertDeliveredExactlyOnce(messageId); + await assertDeliveredUntilCompleted(messageId); }); test("an older intent without identity with only a cur copy is withheld", async () => { @@ -221,7 +241,7 @@ describe("MailClient defers unresolved placements", () => { const client = new MailClient(mailDir, undefined, AGENT, verifier()); mailboxReplayStore(mailboxRoot).beginPlacement(messageId, "m1.json"); renameSync(source, curPath()); - expect(await client.checkNewMail()).toEqual([]); + expect(await client.deliverNext()).toBeNull(); expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true); expect(hasCommittedMessageId(mailboxRoot, messageId)).toBe(false); }); @@ -246,11 +266,11 @@ describe("MailClient defers unresolved placements", () => { const lock = await acquireMailLock(mailboxRoot); expect(lock).not.toBeNull(); try { - expect(await client.checkNewMail()).toEqual([]); + expect(await client.deliverNext()).toBeNull(); expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true); } finally { lock?.release(); } - await assertDeliveredExactlyOnce(messageId); + await assertDeliveredUntilCompleted(messageId); }); }); diff --git a/packages/cli/src/utils/mail.ts b/packages/cli/src/utils/mail.ts index 6889f467..b7593f8d 100644 --- a/packages/cli/src/utils/mail.ts +++ b/packages/cli/src/utils/mail.ts @@ -1481,6 +1481,23 @@ export async function recoverPromoted(agent: string, curPath: string, verify: Ma // includes the recipient binding, so a record carrying another mailbox's // genuine envelope cannot be presented here. (The first-delivery-only replay // gate is unnecessary: this id is already consumed.) + // cli#617 slice (b): a record whose placement intent declares a phase other + // than `placing` — `in_progress` (a delivery in flight), `failed` (terminal) + // or `unknown` — is withheld. This presenter does not present what another + // consumer has declared in flight, and the same bar now covers every direct + // recoverPromoted caller (the bridge and plugin startup sweeps, verifyMailAction + // and the plugin's internalInbound). A malformed intent withholds too, rather + // than presenting (a failed read never reads as absent). + let placementPhase: PlacementPhase | null; + try { + placementPhase = mailboxReplayStore(dirs.root).placementPhase(filename); + } catch (err) { + return { ok: false, class: "busy", reason: `placement history unavailable for ${filename}: ${err instanceof Error ? err.message : String(err)}` }; + } + if (placementPhase !== null && placementPhase !== "placing") { + return { ok: false, class: "busy", reason: `placement intent for ${filename} is ${placementPhase}; not presented` }; + } + const decision = await checkPromotedRecord(agent, msg, mailRoot, verify, dirs.root); if (!decision.ok) { const lock = acquireMailLockSync(dirs.root, { timeoutMs: 0 }); @@ -1516,10 +1533,11 @@ export async function recoverPromoted(agent: string, curPath: string, verify: Ma * phase that resolves to `in_progress` (another consumer's delivery in * flight), `failed` (terminal), or `unknown` (an unsupported value in a * recognized field) makes the sweep skip the record entirely, so this sweep - * does not present what another consumer has declared in flight. Four direct recoverPromoted callers do not - * read phases yet (slice b): the bridge and OpenClaw plugin startup sweeps, - * verifyMailAction, and the plugin's internalInbound. Throws on a malformed - * intent; the caller skips. + * does not present what another consumer has declared in flight. Slice (b) + * applies the same check inside recoverPromoted, so the four direct callers + * (the bridge and plugin startup sweeps, verifyMailAction, and the plugin's + * internalInbound) withhold it too. Throws on a malformed intent; the caller + * skips. */ function withheldByPlacementPhase(replay: ReplayStore, file: string): boolean { const phase = replay.placementPhase(file); diff --git a/packages/cli/test/mail-slice-b-presenters.test.ts b/packages/cli/test/mail-slice-b-presenters.test.ts new file mode 100644 index 00000000..9d4e1615 --- /dev/null +++ b/packages/cli/test/mail-slice-b-presenters.test.ts @@ -0,0 +1,108 @@ +/** + * cli#617 slice (b): recoverPromoted withholds a cur/ record whose placement + * intent declares a phase other than `placing`. The check lives in + * recoverPromoted, so every direct caller — the bridge and plugin startup + * sweeps, verifyMailAction and the plugin's internalInbound — gets it. + */ + +import { createPatchShared } from "./helpers/patch-shared.js"; +const patchShared = createPatchShared(); +import { afterEach, beforeEach, expect, spyOn, test, mock } from "bun:test"; +import { createHash } from "node:crypto"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import * as ed from "@noble/ed25519"; +import { getInbox, promote, recoverPromoted, sendMessage, verifyMailAction } from "../src/utils/mail.js"; +import { signOutboundBody } from "../src/utils/mail-sign.js"; +import { mailboxReplayStore } from "../../agent/src/lib/mailbox-policy.js"; + +afterEach(() => { + mock.restore(); +}); + +const seed = Buffer.alloc(32, 0x33); +let home: string; +let keys: string; +let savedEnv: Record; +const fetchSpyState = { value: undefined as unknown as ReturnType }; + +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "mail-slice-b-")); + keys = join(home, "keys"); + mkdirSync(keys); + savedEnv = {}; + for (const name of ["HOME", "TPS_HOME", "TPS_MAIL_DIR", "TPS_TEST_KEYS_DIR", "TPS_AGENT_ID", "FLAIR_URL", "FLAIR_KEY_PATH"]) { + savedEnv[name] = process.env[name]; + } + process.env.HOME = home; + process.env.TPS_HOME = join(home, ".tps"); + process.env.TPS_MAIL_DIR = join(home, "mail"); + process.env.TPS_TEST_KEYS_DIR = keys; + process.env.TPS_AGENT_ID = "local-agent"; + process.env.FLAIR_URL = "http://flair.test"; + process.env.FLAIR_KEY_PATH = join(keys, "reader.key"); + writeFileSync(process.env.FLAIR_KEY_PATH, seed); + writeFileSync(join(keys, "flint.key"), seed); + const publicKey = Buffer.from(ed.getPublicKey(seed)).toString("base64"); + patchShared(fetchSpyState, "value", spyOn(globalThis, "fetch").mockImplementation(async (input) => { + const url = new URL(String(input)); + if (url.origin !== "http://flair.test") throw new Error(`unexpected request: ${url.origin}`); + if (url.pathname === "/Health") return new Response("ok"); + const name = url.pathname.match(/^\/Agent\/(.+)$/)?.[1]; + return name && ["flint", "local-agent"].includes(name) + ? Response.json({ id: name, publicKey }) + : new Response("not found", { status: 404 }); + })); +}); + +afterEach(() => { + fetchSpyState.value.mockRestore(); + for (const [name, value] of Object.entries(savedEnv)) { + if (value === undefined) delete process.env[name]; else process.env[name] = value; + } + rmSync(home, { recursive: true, force: true }); +}); + +const intentPath = (root: string, file: string) => join(root, `.placement-${createHash("sha256").update(file).digest("hex")}.json`); + +/** Promote one signed record into kern's cur/ and write a placement intent in `fields`. */ +async function promotedRecord(fields: Record | null): Promise<{ root: string; file: string; messageId: string; path: string }> { + const sent = sendMessage("kern", signOutboundBody("flint", "kern", "original", { requireKey: true }), "flint"); + const file = sent.filePath.split("/").pop()!; + const result = await promote("kern", sent.filePath); + expect(result.ok).toBe(true); + if (!result.ok) throw new Error(result.reason); + const inbox = getInbox("kern"); + const path = join(inbox.cur, file); + if (fields !== null) { + const replay = mailboxReplayStore(inbox.root); + replay.beginPlacement(result.message.id, file, path); + const intent = intentPath(inbox.root, file); + writeFileSync(intent, JSON.stringify({ ...JSON.parse(readFileSync(intent, "utf8")), ...fields })); + } + return { root: inbox.root, file, messageId: result.message.id, path }; +} + +test("recoverPromoted presents a record with no intent, or a placing intent", async () => { + const { root, file, messageId, path } = await promotedRecord(null); + expect((await recoverPromoted("kern", path)).ok).toBe(true); + mailboxReplayStore(root).beginPlacement(messageId, file, path); + expect((await recoverPromoted("kern", path)).ok).toBe(true); +}); + +const WITHHELD: Record> = { + in_progress: { version: 1, phase: "in_progress", presentations: 1, firstPresentedAt: new Date().toISOString() }, + failed: { version: 1, phase: "failed", presentations: 5, firstPresentedAt: new Date().toISOString() }, + unknown: { version: 1, phase: "awaiting-review" }, +}; + +for (const [phase, fields] of Object.entries(WITHHELD)) { + test(`recoverPromoted withholds a ${phase} intent; verifyMailAction refuses it`, async () => { + const { path, messageId } = await promotedRecord(fields); + const recovered = await recoverPromoted("kern", path); + expect(recovered.ok).toBe(false); + if (!recovered.ok) expect(recovered.reason).toContain("not presented"); + await expect(verifyMailAction("kern", messageId)).rejects.toThrow("mail action refused"); + }); +}