From cbe67de4c3a1e9fedde176ef315cbb3189002b7d Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 16:59:42 +0000 Subject: [PATCH 01/13] feat(mail): the agent event loop consumes the per-message delivery API (cli#617 slice c) The loop delivers one message at a time via MailClient.deliverNext() and finalizes it with completeDelivery() only after its handler returns a completed outcome. A turn-limit return or a thrown handler is not completion: the hand-over is released, leaving the intent in_progress for the next check. A graceful stop breaks before the next commit, so only the message in flight is committed; finalization is its own synchronous mailbox-lock section that deletes the intent and moves no records. EventLoop.runDeliveries() carries the delivery loop; EventLoop.run() is unchanged. Refs #617 --- ...changed-617-slice-c-event-loop-delivery.md | 15 + packages/agent/src/index.ts | 1 + packages/agent/src/runtime/agent.ts | 5 +- packages/agent/src/runtime/event-loop.ts | 112 ++++++- .../agent/test/helpers/event-loop-child.mjs | 55 ++++ .../test/mail-event-loop-delivery.test.ts | 304 ++++++++++++++++++ 6 files changed, 484 insertions(+), 8 deletions(-) create mode 100644 .changelog/unreleased/changed-617-slice-c-event-loop-delivery.md create mode 100644 packages/agent/test/helpers/event-loop-child.mjs create mode 100644 packages/agent/test/mail-event-loop-delivery.test.ts diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md new file mode 100644 index 00000000..e58739a0 --- /dev/null +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -0,0 +1,15 @@ +- **The agent event loop delivers mail one message at a time and finalizes it only after its handler completes (Refs #617, slice c).** + + Delivery moves from the batch `checkNewMail()` path to the per-message + delivery API. A check commits a single message with a live `in_progress` + intent; the loop processes it and finalizes it on a completed outcome. A + turn-limit return or a thrown handler is not completion: the hand-over is + released and the message is presented again on the next check. A graceful + stop breaks before the next commit, so only the message in flight is + committed. A kill at any point before completion re-presents that message on + the next check, and messages not yet reached stay in `new/`. + + The guarantee: every committed message is presented until it completes or is + quarantined by the liveness bound (five presentations or 24 hours). A + completed message is never presented again, and the delivery key + `mailbox/messageId` lets handlers make repeats idempotent. diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 7a055477..66ce6bc3 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -3,6 +3,7 @@ // Runtime export { AgentRuntime } from "./runtime/agent.js"; export { EventLoop } from "./runtime/event-loop.js"; +export type { MailDeliverySource } from "./runtime/event-loop.js"; export { signedTrustTier } from "./runtime/types.js"; export type { AgentConfig, diff --git a/packages/agent/src/runtime/agent.ts b/packages/agent/src/runtime/agent.ts index 32587508..89aaab8b 100644 --- a/packages/agent/src/runtime/agent.ts +++ b/packages/agent/src/runtime/agent.ts @@ -66,7 +66,6 @@ export class AgentRuntime { } async start(): Promise { - const checkInbox = async () => this.mail.checkNewMail(); const deliverOutbox = () => this.mail.deliverOutbox(); @@ -84,7 +83,9 @@ export class AgentRuntime { }); try { - await Promise.race([this.loop.run(checkInbox), shutdown]); + // cli#617 slice (c): deliver one message at a time, finalizing only on a + // completed outcome, so a crash or stop re-presents the message in flight. + await Promise.race([this.loop.runDeliveries(this.mail), shutdown]); } finally { rmSync(pidPath, { force: true }); } diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index 93d05135..4be49c60 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -1,4 +1,4 @@ -import type { MailMessage, MailClient } from "../io/mail.js"; +import type { MailMessage, MailClient, MailDelivery, DeliverNextResult, DeliveryCompletion, DeliveryRelease } from "../io/mail.js"; import { signedTrustTier, type AgentConfig, @@ -69,6 +69,23 @@ interface EventLoopDeps { mailClient?: MailClient; } +/** + * The per-message delivery API the event loop consumes (cli#617 slice c). + * MailClient satisfies it. Each `deliverNext()` call returns at most one + * delivery, already committed with a live `in_progress` intent: either a + * re-presented message (already committed, from pending placement recovery or + * an earlier unfinished hand-over) or a newly committed one. Pending work is + * presented before new mail is committed, so the loop never re-commits a + * re-presented message. `completeDelivery` finalizes a delivery after the + * handler returns a completed outcome; `releaseDelivery` drops the hand-over's + * local claim without finalizing, leaving the intent for the next check. + */ +export interface MailDeliverySource { + deliverNext(): Promise; + completeDelivery(delivery: MailDelivery): Promise; + releaseDelivery(delivery: MailDelivery): DeliveryRelease; +} + function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -121,11 +138,89 @@ export class EventLoop { } } + /** + * cli#617 slice (c): the delivery loop, one message at a time. The source + * hands over a message that is already committed with a live `in_progress` + * intent; the loop processes it and finalizes it (completeDelivery) ONLY on a + * completed outcome. A non-completion — a turn-limit return or a thrown + * handler — releases the hand-over instead, leaving the intent `in_progress` + * so the next check re-presents it. A graceful stop breaks before the next + * `deliverNext()`, so no message beyond the one in flight is ever committed. + */ + async runDeliveries(source: MailDeliverySource): Promise { + this.running = true; + this.state = "idle"; + this.deps.events?.emit({ + type: "session.start", + model: this.deps.config.llm.model, + contextTokens: this.deps.config.contextWindowTokens ?? 0, + }); + + try { + while (this.running) { + let result: DeliverNextResult; + try { + result = await source.deliverNext(); + } catch (err: any) { + await this.recordError(err); + if (!this.running) break; + await sleep(this.pollMs); + continue; + } + + // idle: nothing to do; deferred (lock busy, an active hand-over, an + // unresolved placement): work may be waiting, retry after a pause. A + // stop requested while the check was in flight breaks here. + if (result.status !== "delivered") { + if (!this.running) break; + await sleep(this.pollMs); + continue; + } + + // A delivery is already committed. Finish it even if a stop was + // requested meanwhile, then break at the loop condition — before the + // next `deliverNext()`, which is the next commit. + const { delivery } = result; + this.state = "processing"; + let completed = false; + try { + completed = await this.processMail(delivery.message); + } catch (err: any) { + await this.recordError(err); + } + this.state = "idle"; + + // Finalize only a completed outcome. Anything else leaves the message's + // intent in `in_progress` and drops only this hand-over's local claim, + // so the next check presents it again (at least once). + try { + if (completed) await source.completeDelivery(delivery); + else source.releaseDelivery(delivery); + } catch (err: any) { + await this.recordError(err); + } + } + } finally { + this.deps.events?.emit({ + type: "session.end", + model: this.deps.config.llm.model, + }); + } + } + async stop(): Promise { this.running = false; this.state = "stopped"; } + private async recordError(err: unknown): Promise { + await this.deps.memory.append({ + type: "error", + ts: new Date().toISOString(), + data: String((err as { message?: unknown } | undefined)?.message ?? err), + }); + } + getState(): AgentState { return this.state; } @@ -261,17 +356,22 @@ export class EventLoop { return this.deps.tools.execute(call.name, call.input); } - private async processMail(message: MailMessage): Promise { + private async processMail(message: MailMessage): Promise { const body = typeof message.body === "string" ? message.body : JSON.stringify(message.body); const trust = this.parseTrust(message); const sender = this.parseSender(message); const prompt = this.formatMailPrompt(body, trust, sender); - await this.processMessage(prompt, trust, sender); + return this.processMessage(prompt, trust, sender); } - private async processMessage(promptRaw: string, trust: TrustLevel, sender?: string): Promise { + /** + * Handle one prompt to an outcome. Returns true on completion (the model + * stopped calling tools); false when the tool-turn limit was reached, which + * is NOT completion and must not finalize a delivery (cli#617 slice c). + */ + private async processMessage(promptRaw: string, trust: TrustLevel, sender?: string): Promise { const prompt = String(promptRaw).trim(); const tools = this.buildToolSpecs(trust); const allowedTools = new Set(tools.map((tool) => tool.name)); @@ -359,7 +459,7 @@ export class EventLoop { this.trimHistory(messages, systemPrompt, tools); // No tool calls = done - if (!completion.toolCalls?.length) return; + if (!completion.toolCalls?.length) return true; if (++turns > maxTurns) { await this.deps.memory.append({ @@ -369,7 +469,7 @@ export class EventLoop { }); // Graceful notification: auto-commit any staged changes and mail the sender await this.notifyTurnLimitReached(sender, turns); - return; + return false; } // Execute tools and append results to conversation diff --git a/packages/agent/test/helpers/event-loop-child.mjs b/packages/agent/test/helpers/event-loop-child.mjs new file mode 100644 index 00000000..0a579fde --- /dev/null +++ b/packages/agent/test/helpers/event-loop-child.mjs @@ -0,0 +1,55 @@ +// event-loop-child.mjs — cli#617 slice (c) test helper. +// +// Runs the REAL EventLoop against a REAL MailClient in a child process. Its +// provider blocks forever inside the handler, and writes TPS_MARKER first, so a +// test can SIGKILL the process at the moment the message is committed (its +// intent is `in_progress`) and being processed — never an injected error. +// +// argv: +// env: TPS_FROM, TPS_SEED, TPS_MARKER, TPS_WORKSPACE +import { mkdirSync, writeFileSync } from "node:fs"; +import * as ed from "@noble/ed25519"; +import { MailClient } from "../../src/io/mail.js"; +import { EventLoop } from "../../src/runtime/event-loop.js"; + +const [mailDir, agentId] = process.argv.slice(2); +const from = process.env.TPS_FROM; +const seed = Buffer.from(process.env.TPS_SEED, "base64"); +const workspace = process.env.TPS_WORKSPACE; +mkdirSync(workspace, { recursive: true }); + +const client = new MailClient(mailDir, undefined, agentId, { + async getAgent(name) { + return name === from ? { publicKey: Buffer.from(ed.getPublicKey(seed)) } : null; + }, +}); + +const provider = { + async complete() { + // The hand-over returned before this call, so the message is committed and + // its intent is in_progress right now. + writeFileSync(process.env.TPS_MARKER, "in-handler"); + await new Promise(() => {}); // spin until SIGKILL + }, +}; + +const config = { + agentId, + name: "loop-child", + mailDir, + memoryPath: `${workspace}/mem.jsonl`, + workspace, + llm: { provider: "test", model: "test-model" }, + contextWindowTokens: 8000, + maxToolTurns: 50, +}; + +const loop = new EventLoop({ + config, + memory: { append: async () => {} }, + context: {}, + provider, + tools: { list: () => [], execute: async () => ({ content: "ok" }) }, +}); + +await loop.runDeliveries(client); diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts new file mode 100644 index 00000000..05f6652e --- /dev/null +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -0,0 +1,304 @@ +/** + * cli#617 slice (c): the event loop consumes the per-message delivery API + * (deliverNext / completeDelivery / releaseDelivery) and finalizes a message + * only after its processing completes. A stop breaks before the next commit; a + * crash or kill at any point before completion re-presents the message on the + * next check. The design's acceptance list for this slice, at the loop seam. + */ + +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 { type DeliverNextResult, type MailDelivery, MailClient } from "../src/io/mail.js"; +import { EventLoop, type MailDeliverySource } from "../src/runtime/event-loop.js"; +import { mailboxReplayStore } 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, 0x61); +const CHILD = fileURLToPath(new URL("./helpers/event-loop-child.mjs", import.meta.url)); + +let mailDir: string; +let root: string; +let workspace: string; + +const cur = (file: string) => join(root, "cur", file); +const fresh = (file: string) => join(root, "new", file); +const intentPath = (file: string) => join(root, `.placement-${createHash("sha256").update(file).digest("hex")}.json`); +const intentPhase = (file: string) => mailboxReplayStore(root).placementPhase(file); +const intentJson = (file: string) => JSON.parse(readFileSync(intentPath(file), "utf8")); +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 delivered = (result: DeliverNextResult): MailDelivery => { + if (result.status !== "delivered") throw new Error(`expected a delivery, got ${JSON.stringify(result)}`); + return result.delivery; +}; + +/** How many consumed-ledger lines carry `id`. */ +function ledgerIds(id: string): number { + const path = join(root, "consumed.jsonl"); + if (!existsSync(path)) return 0; + return readFileSync(path, "utf8").split("\n").filter((line) => line && JSON.parse(line).id === id).length; +} + +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-loop-delivery-")); + root = join(mailDir, AGENT); + workspace = join(mailDir, "ws"); + mkdirSync(join(root, "new"), { recursive: true }); + mkdirSync(join(root, "cur"), { recursive: true }); + mkdirSync(workspace, { recursive: true }); +}); + +afterEach(() => rmSync(mailDir, { recursive: true, force: true })); + +/** Write one signed record to `path`. */ +function signedRecord(path: string, messageId: string, body = "loop-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, body = "loop-body") => signedRecord(fresh(file), messageId, body); + +interface Seen { id: string; presentation: number; } + +/** + * Run the real EventLoop against `c` and record every delivery. The loop runs + * until `stopAfter` deliveries have been handed over, or a delivery names + * `stopOnId`, whichever comes first — a delivery already handed over is still + * processed and finalized before the stop takes effect. + */ +async function runLoop( + c: MailClient, + opts: { provider: { complete: (req: unknown) => Promise }; maxToolTurns?: number; stopAfter?: number; stopOnId?: string }, +): Promise { + const seen: Seen[] = []; + const config = { + agentId: AGENT, name: "loop-test", mailDir, memoryPath: join(workspace, "mem.jsonl"), workspace, + llm: { provider: "test", model: "test-model" }, contextWindowTokens: 8000, + maxToolTurns: opts.maxToolTurns ?? 50, + }; + const loop = new EventLoop({ + config: config as never, + memory: { append: async () => {} } as never, + context: {} as never, + provider: opts.provider as never, + tools: { list: () => [], execute: async () => ({ content: "ok" }) } as never, + }); + const stopAfter = opts.stopAfter ?? Number.POSITIVE_INFINITY; + let handovers = 0; + const source: MailDeliverySource = { + async deliverNext() { + const result = await c.deliverNext(); + if (result.status === "delivered") { + const id = result.delivery.message.verifiedEnvelope?.messageId ?? "?"; + seen.push({ id, presentation: result.delivery.presentation }); + if (++handovers >= stopAfter || id === opts.stopOnId) void loop.stop(); + } + return result; + }, + completeDelivery: (delivery) => c.completeDelivery(delivery), + releaseDelivery: (delivery) => c.releaseDelivery(delivery), + }; + await loop.runDeliveries(source); + return seen; +} + +/** A provider whose handler returns a normal completion. */ +const completingProvider = { async complete() { return { content: "done", toolCalls: undefined, inputTokens: 1, outputTokens: 1 }; } }; +/** A provider whose handler always calls a tool, so the turn limit is reached. */ +const toolCallingProvider = { async complete() { return { content: "", toolCalls: [{ id: "t1", name: "noop", input: {} }], inputTokens: 1, outputTokens: 1 }; } }; +/** A provider that never completes for a body containing `poison-body`. */ +const poisonAwareProvider = { + async complete(req: unknown) { + const text = JSON.stringify((req as { messages?: unknown }).messages ?? ""); + return text.includes("poison-body") + ? { content: "", toolCalls: [{ id: "t1", name: "noop", input: {} }], inputTokens: 1, outputTokens: 1 } + : { content: "done", toolCalls: undefined, inputTokens: 1, outputTokens: 1 }; + }, +}; + +interface KillRun { signal: NodeJS.Signals | null; reached: boolean; stderr: string; } + +/** Run the loop in a child; SIGKILL it once its handler writes the marker. */ +function runChildAndKill(timeoutMs = 30_000): Promise { + const marker = join(mailDir, "child-in-handler"); + const child = spawn(process.execPath, [CHILD, mailDir, AGENT], { + env: { + HOME: mailDir, + PATH: process.env.PATH, + TPS_FROM: FROM, + TPS_SEED: SEED.toString("base64"), + TPS_MARKER: marker, + TPS_WORKSPACE: workspace, + }, + 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)) return; + 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 the handler (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-handler re-presents the message on the loop's next check", () => { + test("first delivery, killed while in_progress: re-presented once, later mail handled", async () => { + plant("m1.json", "kill-first"); + plant("m2.json", "later"); + + const run = await runChildAndKill(); + expect(run.reached, `child never reached the handler; stderr=${run.stderr}`).toBe(true); + expect(run.signal).toBe("SIGKILL"); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(newFiles()).toEqual(["m2.json"]); + expect(curFiles()).toEqual(["m1.json"]); + + const seen = await runLoop(client(), { provider: completingProvider, stopAfter: 2 }); + expect(seen).toEqual([{ id: "kill-first", presentation: 2 }, { id: "later", presentation: 1 }]); + expect(newFiles()).toEqual([]); + expect(intentPhase("m1.json")).toBeNull(); + expect(intentPhase("m2.json")).toBeNull(); + }); + + test("recovery, killed while in_progress: re-presented once, later mail handled", async () => { + plant("m1.json", "kill-recovered"); + linkSync(fresh("m1.json"), cur("m1.json")); + unlinkSync(fresh("m1.json")); + mailboxReplayStore(root).beginPlacement("kill-recovered", "m1.json", cur("m1.json")); + plant("m2.json", "later"); + + const run = await runChildAndKill(); + expect(run.reached, `child never reached the handler; stderr=${run.stderr}`).toBe(true); + expect(run.signal).toBe("SIGKILL"); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(newFiles()).toEqual(["m2.json"]); + + const seen = await runLoop(client(), { provider: completingProvider, stopAfter: 2 }); + expect(seen).toEqual([{ id: "kill-recovered", presentation: 2 }, { id: "later", presentation: 1 }]); + expect(newFiles()).toEqual([]); + expect(intentPhase("m1.json")).toBeNull(); + }); +}); + +describe("a graceful stop drains the message in flight and commits nothing further", () => { + test("completed mail is not re-presented; every unhandled message is handled next", async () => { + plant("m1.json", "first"); + plant("m2.json", "second"); + plant("m3.json", "third"); + + const first = await runLoop(client(), { provider: completingProvider, stopAfter: 1 }); + expect(first).toEqual([{ id: "first", presentation: 1 }]); + // The stop broke before committing the next message: both stay in new/. + expect(newFiles()).toEqual(["m2.json", "m3.json"]); + expect(curFiles()).toEqual(["m1.json"]); + expect(intentPhase("m1.json")).toBeNull(); + + const next = await runLoop(client(), { provider: completingProvider, stopAfter: 2 }); + expect(next).toEqual([{ id: "second", presentation: 1 }, { id: "third", presentation: 1 }]); + expect(newFiles()).toEqual([]); + expect(intentPhase("m1.json")).toBeNull(); + }); +}); + +describe("a turn-limit return is not completion", () => { + test("the intent stays in_progress and the message is re-presented on the next check", async () => { + plant("m1.json", "turn-limited"); + + const first = await runLoop(client(), { provider: toolCallingProvider, maxToolTurns: 0, stopAfter: 1 }); + expect(first).toEqual([{ id: "turn-limited", presentation: 1 }]); + const intent = intentJson("m1.json"); + expect(intent.phase).toBe("in_progress"); + expect(intent.claim).toBeUndefined(); + expect(curFiles()).toEqual(["m1.json"]); + + const again = await runLoop(client(), { provider: toolCallingProvider, maxToolTurns: 0, stopAfter: 1 }); + expect(again).toEqual([{ id: "turn-limited", presentation: 2 }]); + expect(intentJson("m1.json").phase).toBe("in_progress"); + }); +}); + +describe("the liveness bound quarantines a handler that never completes", () => { + test("five presentations then the intent is failed with its log line, and other mail keeps flowing", async () => { + plant("m1.json", "poison", "poison-body"); + plant("m2.json", "healthy", "healthy-body"); + + const warnings: string[] = []; + const warn = spyOn(console, "warn").mockImplementation((...args: unknown[]) => { warnings.push(args.join(" ")); }); + let seen: Seen[] = []; + try { + seen = await runLoop(client(), { provider: poisonAwareProvider, maxToolTurns: 0, stopOnId: "healthy" }); + } finally { + warn.mockRestore(); + } + + expect(seen.filter((d) => d.id === "poison").map((d) => d.presentation)).toEqual([1, 2, 3, 4, 5]); + expect(seen.filter((d) => d.id === "healthy")).toEqual([{ id: "healthy", presentation: 1 }]); + expect(intentPhase("m1.json")).toBe("failed"); + expect(warnings.some((w) => w.includes("poison") && w.includes("presentation bound"))).toBe(true); + expect(intentPhase("m2.json")).toBeNull(); + }); +}); + +describe("a re-presented message is never committed twice", () => { + test("a non-completion then a re-presentation leaves one cur/ record and one consumed ID", async () => { + plant("m1.json", "once-only"); + await runLoop(client(), { provider: toolCallingProvider, maxToolTurns: 0, stopAfter: 1 }); + + expect(curFiles()).toEqual(["m1.json"]); + expect(newFiles()).toEqual([]); + expect(ledgerIds("once-only")).toBe(1); + const identity = intentJson("m1.json").identity; + const bytes = readFileSync(cur("m1.json")); + + const c = client(); + const m = delivered(await c.deliverNext()); + expect(m.presentation).toBe(2); + expect(curFiles()).toEqual(["m1.json"]); + expect(newFiles()).toEqual([]); + expect(ledgerIds("once-only")).toBe(1); + expect(intentJson("m1.json").identity).toEqual(identity); + expect(readFileSync(cur("m1.json")).equals(bytes)).toBe(true); + expect(await c.completeDelivery(m)).toEqual({ status: "completed" }); + }); +}); From 67cd7972ce7af050d8dc98b7f062f1585ddc4104 Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 12:07:26 -0700 Subject: [PATCH 02/13] round: address the review verdict (Codex fix round) Round (Codex): deferred finalize/release retried on the same delivery (no stall); signal shutdown awaits the in-flight delivery before removing the PID file; guarantees scoped to durably finalized deliveries Co-Authored-By: Claude Opus 5.5 --- ...changed-617-slice-c-event-loop-delivery.md | 23 +- packages/agent/src/io/mail.ts | 19 +- packages/agent/src/runtime/agent.ts | 25 +- packages/agent/src/runtime/event-loop.ts | 61 +++-- .../agent/test/helpers/event-loop-child.mjs | 7 +- .../test/mail-delivery-at-least-once.test.ts | 5 +- .../test/mail-event-loop-delivery.test.ts | 213 ++++++++++++++++-- 7 files changed, 269 insertions(+), 84 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index e58739a0..d3ec390c 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -1,15 +1,12 @@ -- **The agent event loop delivers mail one message at a time and finalizes it only after its handler completes (Refs #617, slice c).** +- **The agent event loop processes one mail delivery at a time (Refs #617, slice c).** - Delivery moves from the batch `checkNewMail()` path to the per-message - delivery API. A check commits a single message with a live `in_progress` - intent; the loop processes it and finalizes it on a completed outcome. A - turn-limit return or a thrown handler is not completion: the hand-over is - released and the message is presented again on the next check. A graceful - stop breaks before the next commit, so only the message in flight is - committed. A kill at any point before completion re-presents that message on - the next check, and messages not yet reached stay in `new/`. + Handler completion requests finalization; a turn-limit return or thrown + handler requests release. Deferred finalization and release are retried on + the same delivery until settlement or deadline quarantine. A stop prevents + the next `deliverNext()` call; a check already waiting can return a delivery, + which is processed and settled before shutdown removes the PID file. - The guarantee: every committed message is presented until it completes or is - quarantined by the liveness bound (five presentations or 24 hours). A - completed message is never presented again, and the delivery key - `mailbox/messageId` lets handlers make repeats idempotent. + A committed unfinished hand-over can be re-presented after a kill when + recovery can verify it. A durably finalized delivery is not presented again. + Delivery API consumers receive a JSON-encoded + `[canonical mailbox path, messageId]` key, stable across presentations at that path. diff --git a/packages/agent/src/io/mail.ts b/packages/agent/src/io/mail.ts index 45fe428d..e73892c6 100644 --- a/packages/agent/src/io/mail.ts +++ b/packages/agent/src/io/mail.ts @@ -107,6 +107,7 @@ export type DeliveryCompletion = | { status: "completed" } /** An earlier call with this delivery removed it. */ | { status: "already-completed" } + | { status: "quarantined" } | { status: "deferred"; reason: "lock-busy" | "storage-error"; detail: string } /** * Not this hand-over's to finalize: a refusal finalizes no intent (one may @@ -121,6 +122,7 @@ export type DeliveryCompletion = export type DeliveryRelease = | { status: "released" } + | { status: "quarantined" } | { status: "refused"; reason: "foreign" | "stale" } | { status: "deferred"; reason: "lock-busy" | "storage-error"; detail: string }; @@ -412,7 +414,7 @@ export class MailClient { * Unresolved placing/in_progress work defers deliverNext() commits. The token * set covers MailClients of this module copy; claims from other copies in the * same process are treated as live until process exit or lease expiry. - * Release failed handoffs before retrying. The event loop uses checkNewMail(). + * Retry deferred completion or release on the same hand-over. */ async deliverNext(): Promise { let roots: OperationRoots; @@ -467,6 +469,7 @@ export class MailClient { if ("status" in lock) return { status: "deferred", reason: "lock-busy", detail: lock.detail }; let result: ReturnType["completePlacement"]>; try { + if (this.expireIssuedDelivery(roots.mailbox, issued)) return { status: "quarantined" }; result = mailboxReplayStore(roots.mailbox).completePlacement(issued.messageId, issued.file, issued.token); } catch (err) { this.emitRecoveryError(err); @@ -505,6 +508,7 @@ export class MailClient { if (!lock) return { status: "deferred", reason: "lock-busy", detail: "delivery not released; retry" }; let released: boolean; try { + if (this.expireIssuedDelivery(issued.mailbox, issued)) return { status: "quarantined" }; released = mailboxReplayStore(issued.mailbox).releasePlacement(issued.messageId, issued.file, issued.token); } finally { lock.release(); @@ -519,6 +523,19 @@ export class MailClient { } } + /** Called under the mailbox lock; only this issued claim may hit the deadline. */ + private expireIssuedDelivery(mailbox: string, issued: { messageId: string; file: string; token: string; state: string }): boolean { + const replay = mailboxReplayStore(mailbox); + const state = replay.placementState(issued.file); + if (state?.claim?.token !== issued.token) return false; + const expiry = replay.expireOverdue(issued.messageId, issued.file, new Date().toISOString()); + if (expiry.status !== "failed") return false; + issued.state = "released"; + activePresentations.delete(issued.token); + this.logQuarantine(issued.messageId, expiry.presentations, "deadline"); + return true; + } + /** Record a hand-over: the token is already held in this module copy; bind it to the delivery object. */ private issue(roots: OperationRoots, token: string, messageId: string, file: string, presentation: number, message: MailMessage): MailDelivery { const delivery: MailDelivery = Object.freeze({ message, deliveryKey: JSON.stringify([roots.mailbox, messageId]), presentation }); diff --git a/packages/agent/src/runtime/agent.ts b/packages/agent/src/runtime/agent.ts index 89aaab8b..3668e29f 100644 --- a/packages/agent/src/runtime/agent.ts +++ b/packages/agent/src/runtime/agent.ts @@ -66,27 +66,24 @@ export class AgentRuntime { } async start(): Promise { - const deliverOutbox = () => this.mail.deliverOutbox(); - - const pidPath = join(this.config.workspace, ".tps-agent.pid"); mkdirSync(dirname(pidPath), { recursive: true }); writeFileSync(pidPath, `${process.pid}\n`, "utf-8"); - const shutdown = new Promise((resolve) => { - const onStop = async () => { - await this.loop.stop(); - resolve(); - }; - process.once("SIGINT", onStop); - process.once("SIGTERM", onStop); - }); + let loopRun: Promise; + const onStop = async () => { + await this.loop.stop(); + await loopRun.catch(() => {}); + }; + process.once("SIGINT", onStop); + process.once("SIGTERM", onStop); try { - // cli#617 slice (c): deliver one message at a time, finalizing only on a - // completed outcome, so a crash or stop re-presents the message in flight. - await Promise.race([this.loop.runDeliveries(this.mail), shutdown]); + loopRun = this.loop.runDeliveries(this.mail); + await loopRun; } finally { + process.removeListener("SIGINT", onStop); + process.removeListener("SIGTERM", onStop); rmSync(pidPath, { force: true }); } } diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index 4be49c60..db6259d0 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -69,17 +69,7 @@ interface EventLoopDeps { mailClient?: MailClient; } -/** - * The per-message delivery API the event loop consumes (cli#617 slice c). - * MailClient satisfies it. Each `deliverNext()` call returns at most one - * delivery, already committed with a live `in_progress` intent: either a - * re-presented message (already committed, from pending placement recovery or - * an earlier unfinished hand-over) or a newly committed one. Pending work is - * presented before new mail is committed, so the loop never re-commits a - * re-presented message. `completeDelivery` finalizes a delivery after the - * handler returns a completed outcome; `releaseDelivery` drops the hand-over's - * local claim without finalizing, leaving the intent for the next check. - */ +/** The delivery API consumed by the loop; finalize and release may defer. */ export interface MailDeliverySource { deliverNext(): Promise; completeDelivery(delivery: MailDelivery): Promise; @@ -139,13 +129,8 @@ export class EventLoop { } /** - * cli#617 slice (c): the delivery loop, one message at a time. The source - * hands over a message that is already committed with a live `in_progress` - * intent; the loop processes it and finalizes it (completeDelivery) ONLY on a - * completed outcome. A non-completion — a turn-limit return or a thrown - * handler — releases the hand-over instead, leaving the intent `in_progress` - * so the next check re-presents it. A graceful stop breaks before the next - * `deliverNext()`, so no message beyond the one in flight is ever committed. + * Process a returned delivery even if stop was requested during deliverNext(). + * Retry deferred settlement on that delivery before checking for more mail. */ async runDeliveries(source: MailDeliverySource): Promise { this.running = true; @@ -177,9 +162,6 @@ export class EventLoop { continue; } - // A delivery is already committed. Finish it even if a stop was - // requested meanwhile, then break at the loop condition — before the - // next `deliverNext()`, which is the next commit. const { delivery } = result; this.state = "processing"; let completed = false; @@ -188,17 +170,9 @@ export class EventLoop { } catch (err: any) { await this.recordError(err); } - this.state = "idle"; - // Finalize only a completed outcome. Anything else leaves the message's - // intent in `in_progress` and drops only this hand-over's local claim, - // so the next check presents it again (at least once). - try { - if (completed) await source.completeDelivery(delivery); - else source.releaseDelivery(delivery); - } catch (err: any) { - await this.recordError(err); - } + await this.settleDelivery(source, delivery, completed); + this.state = this.running ? "idle" : "stopped"; } } finally { this.deps.events?.emit({ @@ -208,6 +182,31 @@ export class EventLoop { } } + private async settleDelivery(source: MailDeliverySource, delivery: MailDelivery, completed: boolean): Promise { + while (true) { + try { + const result = completed + ? await source.completeDelivery(delivery) + : source.releaseDelivery(delivery); + switch (result.status) { + case "completed": + case "already-completed": + case "released": + case "quarantined": + return; + case "refused": + await this.recordError(`delivery settlement refused: ${result.reason}`); + return; + case "deferred": + break; + } + } catch (err) { + await this.recordError(err); + } + await sleep(this.pollMs); + } + } + async stop(): Promise { this.running = false; this.state = "stopped"; diff --git a/packages/agent/test/helpers/event-loop-child.mjs b/packages/agent/test/helpers/event-loop-child.mjs index 0a579fde..e807b283 100644 --- a/packages/agent/test/helpers/event-loop-child.mjs +++ b/packages/agent/test/helpers/event-loop-child.mjs @@ -1,9 +1,6 @@ // event-loop-child.mjs — cli#617 slice (c) test helper. // -// Runs the REAL EventLoop against a REAL MailClient in a child process. Its -// provider blocks forever inside the handler, and writes TPS_MARKER first, so a -// test can SIGKILL the process at the moment the message is committed (its -// intent is `in_progress`) and being processed — never an injected error. +// Runs EventLoop with MailClient; the provider marks handler entry and waits. // // argv: // env: TPS_FROM, TPS_SEED, TPS_MARKER, TPS_WORKSPACE @@ -26,8 +23,6 @@ const client = new MailClient(mailDir, undefined, agentId, { const provider = { async complete() { - // The hand-over returned before this call, so the message is committed and - // its intent is in_progress right now. writeFileSync(process.env.TPS_MARKER, "in-handler"); await new Promise(() => {}); // spin until SIGKILL }, diff --git a/packages/agent/test/mail-delivery-at-least-once.test.ts b/packages/agent/test/mail-delivery-at-least-once.test.ts index bc344345..e5c4b7a4 100644 --- a/packages/agent/test/mail-delivery-at-least-once.test.ts +++ b/packages/agent/test/mail-delivery-at-least-once.test.ts @@ -1093,14 +1093,15 @@ describe("liveness-bound transitions for unclaimed in_progress work", () => { expect(idOf(next)).toBe("healthy"); }); - test("a held local token defers expiry past the deadline", async () => { + test("a held local token blocks another delivery until settlement quarantines it at the deadline", async () => { plant("m1.json", "slow-but-live"); const c = client(); const m = delivered(await c.deliverNext()); setPhase({ firstPresentedAt: new Date(Date.now() - 25 * 60 * 60 * 1000).toISOString() }); expect(reasonOf(await c.deliverNext())).toBe("active"); expect(intentPhase("m1.json")).toBe("in_progress"); - expect(await c.completeDelivery(m)).toEqual({ status: "completed" }); + expect(await c.completeDelivery(m)).toEqual({ status: "quarantined" }); + expect(intentPhase("m1.json")).toBe("failed"); }); test("the terminal intent keeps its message refused by the replay gate", async () => { diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index 05f6652e..ea46b79f 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -1,24 +1,21 @@ -/** - * cli#617 slice (c): the event loop consumes the per-message delivery API - * (deliverNext / completeDelivery / releaseDelivery) and finalizes a message - * only after its processing completes. A stop breaks before the next commit; a - * crash or kill at any point before completion re-presents the message on the - * next check. The design's acceptance list for this slice, at the loop seam. - */ +/** Real loop/mailbox tests for settlement retries and verifiable committed recovery. */ 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, + existsSync, linkSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, renameSync, 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 { type DeliverNextResult, type MailDelivery, MailClient } from "../src/io/mail.js"; +import { AgentRuntime } from "../src/runtime/agent.js"; +import { ProviderManager } from "../src/llm/provider.js"; +import { FlairContextProvider } from "../src/io/flair.js"; import { EventLoop, type MailDeliverySource } from "../src/runtime/event-loop.js"; -import { mailboxReplayStore } from "../src/lib/mailbox-policy.js"; +import { DELIVERY_PRESENTATION_TTL_MS, mailboxReplayStore } from "../src/lib/mailbox-policy.js"; import { signEnvelope, type FlairClient } from "../src/lib/signEnvelope.js"; const AGENT = "agent-a"; @@ -90,7 +87,7 @@ interface Seen { id: string; presentation: number; } * Run the real EventLoop against `c` and record every delivery. The loop runs * until `stopAfter` deliveries have been handed over, or a delivery names * `stopOnId`, whichever comes first — a delivery already handed over is still - * processed and finalized before the stop takes effect. + * processed and settled before the loop returns. */ async function runLoop( c: MailClient, @@ -108,7 +105,7 @@ async function runLoop( context: {} as never, provider: opts.provider as never, tools: { list: () => [], execute: async () => ({ content: "ok" }) } as never, - }); + }, 1); const stopAfter = opts.stopAfter ?? Number.POSITIVE_INFINITY; let handovers = 0; const source: MailDeliverySource = { @@ -132,7 +129,7 @@ async function runLoop( const completingProvider = { async complete() { return { content: "done", toolCalls: undefined, inputTokens: 1, outputTokens: 1 }; } }; /** A provider whose handler always calls a tool, so the turn limit is reached. */ const toolCallingProvider = { async complete() { return { content: "", toolCalls: [{ id: "t1", name: "noop", input: {} }], inputTokens: 1, outputTokens: 1 }; } }; -/** A provider that never completes for a body containing `poison-body`. */ +/** Calls tools for a body containing `poison-body`. */ const poisonAwareProvider = { async complete(req: unknown) { const text = JSON.stringify((req as { messages?: unknown }).messages ?? ""); @@ -182,7 +179,7 @@ function runChildAndKill(timeoutMs = 30_000): Promise { }); } -describe("a kill mid-handler re-presents the message on the loop's next check", () => { +describe("a committed delivery killed mid-handler is re-presented when verification succeeds", () => { test("first delivery, killed while in_progress: re-presented once, later mail handled", async () => { plant("m1.json", "kill-first"); plant("m2.json", "later"); @@ -221,15 +218,14 @@ describe("a kill mid-handler re-presents the message on the loop's next check", }); }); -describe("a graceful stop drains the message in flight and commits nothing further", () => { - test("completed mail is not re-presented; every unhandled message is handled next", async () => { +describe("a stop drains a delivery returned by the check in flight", () => { + test("durably finalized mail is not re-presented; later mail is handled next", async () => { plant("m1.json", "first"); plant("m2.json", "second"); plant("m3.json", "third"); const first = await runLoop(client(), { provider: completingProvider, stopAfter: 1 }); expect(first).toEqual([{ id: "first", presentation: 1 }]); - // The stop broke before committing the next message: both stay in new/. expect(newFiles()).toEqual(["m2.json", "m3.json"]); expect(curFiles()).toEqual(["m1.json"]); expect(intentPhase("m1.json")).toBeNull(); @@ -258,7 +254,7 @@ describe("a turn-limit return is not completion", () => { }); }); -describe("the liveness bound quarantines a handler that never completes", () => { +describe("the liveness bound quarantines repeated non-completion", () => { test("five presentations then the intent is failed with its log line, and other mail keeps flowing", async () => { plant("m1.json", "poison", "poison-body"); plant("m2.json", "healthy", "healthy-body"); @@ -302,3 +298,186 @@ describe("a re-presented message is never committed twice", () => { expect(await c.completeDelivery(m)).toEqual({ status: "completed" }); }); }); + +for (const operation of ["completeDelivery", "releaseDelivery"] as const) { + test(`${operation} retries the same deferred delivery before later mail`, async () => { + plant("m1.json", "retry", "poison-body"); + plant("m2.json", "later", "healthy-body"); + const c = client(); + const original = c[operation].bind(c); + let attempts = 0; + let held: MailDelivery | undefined; + const next = c.deliverNext.bind(c); + const nextSpy = spyOn(c, "deliverNext").mockImplementation(async () => { + expect(held).toBeUndefined(); + return next(); + }); + const claimDir = join(root, ".mail-lock.claim"); + const savedMailbox = `${root}.saved`; + const provider = { + async complete(req: unknown) { + if (attempts === 0) { + if (operation === "completeDelivery") mkdirSync(claimDir); + else { + renameSync(root, savedMailbox); + writeFileSync(root, "not a directory"); + } + } + return poisonAwareProvider.complete(req); + }, + }; + const observe = (delivery: MailDelivery, result: { status: string; reason?: string }) => { + if (delivery.message.verifiedEnvelope?.messageId !== "retry" || attempts >= 2) return; + attempts++; + if (attempts === 1) { + expect(result).toMatchObject({ status: "deferred", reason: operation === "completeDelivery" ? "lock-busy" : "storage-error" }); + held = delivery; + if (operation === "completeDelivery") rmSync(claimDir, { recursive: true }); + else { + rmSync(root); + renameSync(savedMailbox, root); + } + expect(intentJson("m1.json").claim).toBeDefined(); + } else { + expect(delivery).toBe(held); + expect(result.status).toBe(operation === "completeDelivery" ? "completed" : "released"); + held = undefined; + } + }; + const settlementSpy = operation === "completeDelivery" + ? spyOn(c, "completeDelivery").mockImplementation(async (delivery) => { + const result = await (original as MailClient["completeDelivery"])(delivery); + observe(delivery, result); + return result; + }) + : spyOn(c, "releaseDelivery").mockImplementation((delivery) => { + const result = (original as MailClient["releaseDelivery"])(delivery); + observe(delivery, result); + return result; + }); + try { + const seen = await runLoop(c, { + provider: operation === "completeDelivery" ? { + async complete(req) { await provider.complete(req); return completingProvider.complete(); }, + } : provider, + maxToolTurns: 0, stopOnId: "later", + }); + expect(attempts).toBe(2); + expect(seen.at(-1)).toEqual({ id: "later", presentation: 1 }); + expect(intentPhase("m2.json")).toBeNull(); + expect(held).toBeUndefined(); + } finally { + settlementSpy.mockRestore(); + nextSpy.mockRestore(); + } + }, 15_000); + + test(`${operation} quarantines an overdue deferred claim before later mail`, async () => { + plant("m1.json", "overdue"); + plant("m2.json", "later"); + const c = client(); + const claimDir = join(root, ".mail-lock.claim"); + let attempts = 0; + const original = c[operation].bind(c); + const expire = () => { + attempts++; + if (attempts !== 1) return; + const intent = intentJson("m1.json"); + intent.firstPresentedAt = new Date(Date.now() - DELIVERY_PRESENTATION_TTL_MS - 1_000).toISOString(); + writeFileSync(intentPath("m1.json"), JSON.stringify(intent)); + mkdirSync(claimDir); + }; + const observe = (result: { status: string }) => { + if (attempts === 1) { + expect(result.status).toBe("deferred"); + rmSync(claimDir, { recursive: true }); + } else if (attempts === 2) expect(result.status).toBe("quarantined"); + }; + const settlementSpy = operation === "completeDelivery" + ? spyOn(c, "completeDelivery").mockImplementation(async (delivery) => { + expire(); + const result = await (original as MailClient["completeDelivery"])(delivery); + observe(result); return result; + }) + : spyOn(c, "releaseDelivery").mockImplementation((delivery) => { + expire(); + const result = (original as MailClient["releaseDelivery"])(delivery); + observe(result); return result; + }); + try { + const seen = await runLoop(c, { + provider: operation === "completeDelivery" ? completingProvider : toolCallingProvider, + maxToolTurns: 0, stopOnId: "later", + }); + expect(seen).toEqual([{ id: "overdue", presentation: 1 }, { id: "later", presentation: 1 }]); + expect(intentPhase("m1.json")).toBe("failed"); + expect(intentJson("m1.json").claim).toBeUndefined(); + expect(existsSync(cur("m1.json"))).toBe(true); + } finally { settlementSpy.mockRestore(); } + }, 15_000); +} + +for (const signal of ["SIGINT", "SIGTERM"] as const) { + for (const completed of [true, false]) { + test(`AgentRuntime.start drains ${completed ? "finalize" : "release"} after ${signal} before removing its PID file`, async () => { + plant("m1.json", "in-flight"); + plant("m2.json", "later"); + let enter!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + let finish!: () => void; + const finishing = new Promise((resolve) => { finish = resolve; }); + const lookup = spyOn(FlairContextProvider.prototype, "getAgent").mockImplementation(async () => ({ publicKey: Buffer.from(ed.getPublicKey(SEED)).toString("base64") }) as never); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + enter(); await finishing; + return completed ? completingProvider.complete() : toolCallingProvider.complete(); + }); + const runtime = new AgentRuntime({ + agentId: AGENT, name: "signal-test", mailDir, workspace, + memoryPath: join(workspace, "memory.jsonl"), maxToolTurns: 0, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + let returned = false; + const start = runtime.start().then(() => { returned = true; }); + const pidPath = join(workspace, ".tps-agent.pid"); + const claimDir = join(root, ".mail-lock.claim"); + const original = MailClient.prototype[completed ? "completeDelivery" : "releaseDelivery"]; + let deferred!: () => void; + const deferring = new Promise((resolve) => { deferred = resolve; }); + const settlementSpy = completed + ? spyOn(MailClient.prototype, "completeDelivery").mockImplementation(async function(delivery) { + const result = await (original as MailClient["completeDelivery"]).call(this, delivery); + if (result.status === "deferred") deferred(); + return result; + }) + : spyOn(MailClient.prototype, "releaseDelivery").mockImplementation(function(delivery) { + const result = (original as MailClient["releaseDelivery"]).call(this, delivery); + if (result.status === "deferred") deferred(); + return result; + }); + try { + await entered; + process.emit(signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(returned).toBe(false); + expect(existsSync(pidPath)).toBe(true); + mkdirSync(claimDir); + finish(); + await deferring; + expect(returned).toBe(false); + expect(existsSync(pidPath)).toBe(true); + expect(intentJson("m1.json").claim).toBeDefined(); + rmSync(claimDir, { recursive: true }); + await start; + expect(existsSync(pidPath)).toBe(false); + expect(newFiles()).toEqual(["m2.json"]); + expect(completed ? intentPhase("m1.json") : intentJson("m1.json").claim).toBe(completed ? null : undefined); + } finally { + finish(); + rmSync(claimDir, { recursive: true, force: true }); + await runtime.stop(); + await start; + settlementSpy.mockRestore(); provider.mockRestore(); lookup.mockRestore(); + } + }, 15_000); + } +} From 5fbc3cb3fe799e3860601420b056a6c98f2dce33 Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 12:22:31 -0700 Subject: [PATCH 03/13] round: address the review verdict (Codex fix round) Round 2 (Codex): a failure while recording a handler error can no longer skip settlement (finally path; real failing-store test with signal shutdown); deadline quarantine wording conditioned on the lock and storage Co-Authored-By: Claude Opus 5.5 --- ...changed-617-slice-c-event-loop-delivery.md | 5 +- packages/agent/src/runtime/event-loop.ts | 20 +++--- .../test/mail-delivery-at-least-once.test.ts | 2 +- .../test/mail-event-loop-delivery.test.ts | 66 ++++++++++++++++++- 4 files changed, 82 insertions(+), 11 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index d3ec390c..ffbd121f 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -2,10 +2,13 @@ Handler completion requests finalization; a turn-limit return or thrown handler requests release. Deferred finalization and release are retried on - the same delivery until settlement or deadline quarantine. A stop prevents + the same delivery. Deadline quarantine requires the mailbox lock and storage + to become available; until then the claim stays pending. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, which is processed and settled before shutdown removes the PID file. + A handler that never returns cannot reach the presentation or deadline checks. + A committed unfinished hand-over can be re-presented after a kill when recovery can verify it. A durably finalized delivery is not presented again. Delivery API consumers receive a JSON-encoded diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index db6259d0..6c9ca551 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -169,10 +169,10 @@ export class EventLoop { completed = await this.processMail(delivery.message); } catch (err: any) { await this.recordError(err); + } finally { + await this.settleDelivery(source, delivery, completed); + this.state = this.running ? "idle" : "stopped"; } - - await this.settleDelivery(source, delivery, completed); - this.state = this.running ? "idle" : "stopped"; } } finally { this.deps.events?.emit({ @@ -213,11 +213,15 @@ export class EventLoop { } private async recordError(err: unknown): Promise { - await this.deps.memory.append({ - type: "error", - ts: new Date().toISOString(), - data: String((err as { message?: unknown } | undefined)?.message ?? err), - }); + try { + await this.deps.memory.append({ + type: "error", + ts: new Date().toISOString(), + data: String((err as { message?: unknown } | undefined)?.message ?? err), + }); + } catch (reportingError) { + console.error("Failed to record event-loop error:", sanitizeError(reportingError)); + } } getState(): AgentState { diff --git a/packages/agent/test/mail-delivery-at-least-once.test.ts b/packages/agent/test/mail-delivery-at-least-once.test.ts index e5c4b7a4..e9e6ae06 100644 --- a/packages/agent/test/mail-delivery-at-least-once.test.ts +++ b/packages/agent/test/mail-delivery-at-least-once.test.ts @@ -1093,7 +1093,7 @@ describe("liveness-bound transitions for unclaimed in_progress work", () => { expect(idOf(next)).toBe("healthy"); }); - test("a held local token blocks another delivery until settlement quarantines it at the deadline", async () => { + test("a held local token blocks another delivery until completion quarantines the overdue claim", async () => { plant("m1.json", "slow-but-live"); const c = client(); const m = delivered(await c.deliverNext()); diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index ea46b79f..81ce3c7f 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -12,6 +12,7 @@ import { fileURLToPath } from "node:url"; import * as ed from "@noble/ed25519"; import { type DeliverNextResult, type MailDelivery, MailClient } from "../src/io/mail.js"; import { AgentRuntime } from "../src/runtime/agent.js"; +import { MemoryStore } from "../src/io/memory.js"; import { ProviderManager } from "../src/llm/provider.js"; import { FlairContextProvider } from "../src/io/flair.js"; import { EventLoop, type MailDeliverySource } from "../src/runtime/event-loop.js"; @@ -372,7 +373,7 @@ for (const operation of ["completeDelivery", "releaseDelivery"] as const) { } }, 15_000); - test(`${operation} quarantines an overdue deferred claim before later mail`, async () => { + test(`${operation} quarantines an overdue deferred claim once the mailbox lock and storage are available`, async () => { plant("m1.json", "overdue"); plant("m2.json", "later"); const c = client(); @@ -418,6 +419,69 @@ for (const operation of ["completeDelivery", "releaseDelivery"] as const) { } for (const signal of ["SIGINT", "SIGTERM"] as const) { + test(`AgentRuntime.start releases after memory writes fail during ${signal} before removing its PID file`, async () => { + plant("m1.json", "memory-failure"); + plant("m2.json", "later"); + let enter!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + let finish!: () => void; + const finishing = new Promise((resolve) => { finish = resolve; }); + const lookup = spyOn(FlairContextProvider.prototype, "getAgent").mockImplementation(async () => ({ publicKey: Buffer.from(ed.getPublicKey(SEED)).toString("base64") }) as never); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + enter(); await finishing; + return completingProvider.complete(); + }); + const memoryPath = join(workspace, "memory.jsonl"); + const pidPath = join(workspace, ".tps-agent.pid"); + const runtime = new AgentRuntime({ + agentId: AGENT, name: "memory-failure-test", mailDir, workspace, memoryPath, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + const append = MemoryStore.prototype.append; + const failedWrites: string[] = []; + const memorySpy = spyOn(MemoryStore.prototype, "append").mockImplementation(function(event) { + try { return append.call(this, event); } + catch (error) { failedWrites.push(event.type); throw error; } + }); + const errorLog = spyOn(console, "error").mockImplementation(() => {}); + let released = false; + const release = MailClient.prototype.releaseDelivery; + const settlementSpy = spyOn(MailClient.prototype, "releaseDelivery").mockImplementation(function(delivery) { + expect(existsSync(pidPath)).toBe(true); + expect(intentJson("m1.json").claim).toBeDefined(); + const result = release.call(this, delivery); + expect(result.status).toBe("released"); + expect(intentJson("m1.json").claim).toBeUndefined(); + released = true; + return result; + }); + const start = runtime.start(); + void start.catch(() => {}); + try { + await entered; + expect(intentJson("m1.json").claim).toBeDefined(); + expect(existsSync(memoryPath)).toBe(true); + renameSync(memoryPath, `${memoryPath}.saved`); + mkdirSync(memoryPath); + process.emit(signal); + expect(existsSync(pidPath)).toBe(true); + finish(); + await start; + expect(failedWrites).toEqual(["assistant", "error"]); + expect(errorLog).toHaveBeenCalledWith("Failed to record event-loop error:", expect.any(String)); + expect(released).toBe(true); + expect(existsSync(pidPath)).toBe(false); + expect(intentJson("m1.json").claim).toBeUndefined(); + expect(newFiles()).toEqual(["m2.json"]); + } finally { + finish(); + await runtime.stop(); + await start.catch(() => {}); + settlementSpy.mockRestore(); memorySpy.mockRestore(); errorLog.mockRestore(); + provider.mockRestore(); lookup.mockRestore(); + } + }, 15_000); + for (const completed of [true, false]) { test(`AgentRuntime.start drains ${completed ? "finalize" : "release"} after ${signal} before removing its PID file`, async () => { plant("m1.json", "in-flight"); From 4c77bbde0480a98dc06cc045cc792d38c134ecc1 Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 12:40:09 -0700 Subject: [PATCH 04/13] round: address the review verdict (Codex fix round) Round 3 (Codex): the stop guarantee names the identity-refusal exception (logged, intent left pending); real-mailbox test that changes cur/ during a handler Co-Authored-By: Claude Opus 5.5 --- ...changed-617-slice-c-event-loop-delivery.md | 5 +- .../test/mail-event-loop-delivery.test.ts | 81 ++++++++++++++++++- 2 files changed, 82 insertions(+), 4 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index ffbd121f..7ed4364d 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -5,7 +5,10 @@ the same delivery. Deadline quarantine requires the mailbox lock and storage to become available; until then the claim stays pending. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, - which is processed and settled before shutdown removes the PID file. + which is processed and settled before shutdown removes the PID file, except + that a `completeDelivery()` refusal with `identity` after `cur/` changes during + the handler is logged and leaves the intent pending, blocking further delivery + of that message until resolved. A handler that never returns cannot reach the presentation or deadline checks. diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index 81ce3c7f..f185728e 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -87,8 +87,7 @@ interface Seen { id: string; presentation: number; } /** * Run the real EventLoop against `c` and record every delivery. The loop runs * until `stopAfter` deliveries have been handed over, or a delivery names - * `stopOnId`, whichever comes first — a delivery already handed over is still - * processed and settled before the loop returns. + * `stopOnId`, whichever comes first. */ async function runLoop( c: MailClient, @@ -219,7 +218,7 @@ describe("a committed delivery killed mid-handler is re-presented when verificat }); }); -describe("a stop drains a delivery returned by the check in flight", () => { +describe("a stop finalizes a returned delivery when completion succeeds", () => { test("durably finalized mail is not re-presented; later mail is handled next", async () => { plant("m1.json", "first"); plant("m2.json", "second"); @@ -419,6 +418,82 @@ for (const operation of ["completeDelivery", "releaseDelivery"] as const) { } for (const signal of ["SIGINT", "SIGTERM"] as const) { + test(`AgentRuntime.start logs an identity refusal and keeps the intent pending after ${signal}`, async () => { + plant("m1.json", "identity-refusal"); + plant("m2.json", "later"); + let enter!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + let finish!: () => void; + const finishing = new Promise((resolve) => { finish = resolve; }); + const lookup = spyOn(FlairContextProvider.prototype, "getAgent").mockImplementation(async () => ({ publicKey: Buffer.from(ed.getPublicKey(SEED)).toString("base64") }) as never); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + writeFileSync(cur("m1.json"), `${readFileSync(cur("m1.json"), "utf8")} `); + enter(); await finishing; + return completingProvider.complete(); + }); + const memoryPath = join(workspace, "memory.jsonl"); + const pidPath = join(workspace, ".tps-agent.pid"); + const runtime = new AgentRuntime({ + agentId: AGENT, name: "identity-refusal-test", mailDir, workspace, memoryPath, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + const seen: string[] = []; + const next = MailClient.prototype.deliverNext; + const nextSpy = spyOn(MailClient.prototype, "deliverNext").mockImplementation(async function() { + const result = await next.call(this); + if (result.status === "delivered") seen.push(result.delivery.message.verifiedEnvelope!.messageId); + return result; + }); + const complete = MailClient.prototype.completeDelivery; + const settlementSpy = spyOn(MailClient.prototype, "completeDelivery").mockImplementation(async function(delivery) { + expect(existsSync(pidPath)).toBe(true); + const before = readFileSync(intentPath("m1.json"), "utf8"); + const result = await complete.call(this, delivery); + expect(result).toMatchObject({ status: "refused", reason: "identity" }); + expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(before); + return result; + }); + let loopReturned = false; + const run = EventLoop.prototype.runDeliveries; + const loopSpy = spyOn(EventLoop.prototype, "runDeliveries").mockImplementation(async function(source) { + await run.call(this, source); + expect(existsSync(pidPath)).toBe(true); + loopReturned = true; + }); + const start = runtime.start(); + void start.catch(() => {}); + try { + await entered; + const pending = readFileSync(intentPath("m1.json"), "utf8"); + process.emit(signal); + expect(loopReturned).toBe(false); + expect(existsSync(pidPath)).toBe(true); + finish(); + await start; + expect(settlementSpy).toHaveBeenCalledTimes(1); + expect(loopReturned).toBe(true); + expect(existsSync(pidPath)).toBe(false); + expect(readFileSync(memoryPath, "utf8").trim().split("\n").map((line) => JSON.parse(line))).toContainEqual({ + type: "error", ts: expect.any(String), data: "delivery settlement refused: identity", + }); + expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(pending); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(intentJson("m1.json").claim).toBeDefined(); + expect(nextSpy).toHaveBeenCalledTimes(1); + expect(seen).toEqual(["identity-refusal"]); + expect(await client().deliverNext()).toMatchObject({ status: "deferred", reason: "record-changed" }); + expect(seen).toEqual(["identity-refusal"]); + expect(newFiles()).toEqual(["m2.json"]); + expect(curFiles()).toEqual(["m1.json"]); + } finally { + finish(); + await runtime.stop(); + await start.catch(() => {}); + loopSpy.mockRestore(); settlementSpy.mockRestore(); nextSpy.mockRestore(); + provider.mockRestore(); lookup.mockRestore(); + } + }, 15_000); + test(`AgentRuntime.start releases after memory writes fail during ${signal} before removing its PID file`, async () => { plant("m1.json", "memory-failure"); plant("m2.json", "later"); From aa6fbaefa2f6f9cf18481b4dda43e521c9a725a2 Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 12:57:39 -0700 Subject: [PATCH 05/13] round: address the review verdict (Codex fix round) Round 4 (Codex): stop settlement is attempted; deferrals retried; any refusal is logged and may leave the intent pending (general rule); stale-refusal stop test Co-Authored-By: Claude Opus 5.5 --- ...changed-617-slice-c-event-loop-delivery.md | 7 +- .../test/mail-event-loop-delivery.test.ts | 163 +++++++++--------- 2 files changed, 89 insertions(+), 81 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index 7ed4364d..6b4b483a 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -5,10 +5,9 @@ the same delivery. Deadline quarantine requires the mailbox lock and storage to become available; until then the claim stays pending. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, - which is processed and settled before shutdown removes the PID file, except - that a `completeDelivery()` refusal with `identity` after `cur/` changes during - the handler is logged and leaves the intent pending, blocking further delivery - of that message until resolved. + which the loop processes and attempts to settle before shutdown removes the + PID file. Deferred settlement is retried; any refusal is logged and can leave + the intent pending, blocking that message until resolved. A handler that never returns cannot reach the presentation or deadline checks. diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index f185728e..c01d6197 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { spawn } from "node:child_process"; -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { existsSync, linkSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, renameSync, rmSync, unlinkSync, writeFileSync, } from "node:fs"; @@ -418,83 +418,92 @@ for (const operation of ["completeDelivery", "releaseDelivery"] as const) { } for (const signal of ["SIGINT", "SIGTERM"] as const) { - test(`AgentRuntime.start logs an identity refusal and keeps the intent pending after ${signal}`, async () => { - plant("m1.json", "identity-refusal"); - plant("m2.json", "later"); - let enter!: () => void; - const entered = new Promise((resolve) => { enter = resolve; }); - let finish!: () => void; - const finishing = new Promise((resolve) => { finish = resolve; }); - const lookup = spyOn(FlairContextProvider.prototype, "getAgent").mockImplementation(async () => ({ publicKey: Buffer.from(ed.getPublicKey(SEED)).toString("base64") }) as never); - const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { - writeFileSync(cur("m1.json"), `${readFileSync(cur("m1.json"), "utf8")} `); - enter(); await finishing; - return completingProvider.complete(); - }); - const memoryPath = join(workspace, "memory.jsonl"); - const pidPath = join(workspace, ".tps-agent.pid"); - const runtime = new AgentRuntime({ - agentId: AGENT, name: "identity-refusal-test", mailDir, workspace, memoryPath, - llm: { provider: "openai", model: "test", apiKey: "test" }, - }); - const seen: string[] = []; - const next = MailClient.prototype.deliverNext; - const nextSpy = spyOn(MailClient.prototype, "deliverNext").mockImplementation(async function() { - const result = await next.call(this); - if (result.status === "delivered") seen.push(result.delivery.message.verifiedEnvelope!.messageId); - return result; - }); - const complete = MailClient.prototype.completeDelivery; - const settlementSpy = spyOn(MailClient.prototype, "completeDelivery").mockImplementation(async function(delivery) { - expect(existsSync(pidPath)).toBe(true); - const before = readFileSync(intentPath("m1.json"), "utf8"); - const result = await complete.call(this, delivery); - expect(result).toMatchObject({ status: "refused", reason: "identity" }); - expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(before); - return result; - }); - let loopReturned = false; - const run = EventLoop.prototype.runDeliveries; - const loopSpy = spyOn(EventLoop.prototype, "runDeliveries").mockImplementation(async function(source) { - await run.call(this, source); - expect(existsSync(pidPath)).toBe(true); - loopReturned = true; - }); - const start = runtime.start(); - void start.catch(() => {}); - try { - await entered; - const pending = readFileSync(intentPath("m1.json"), "utf8"); - process.emit(signal); - expect(loopReturned).toBe(false); - expect(existsSync(pidPath)).toBe(true); - finish(); - await start; - expect(settlementSpy).toHaveBeenCalledTimes(1); - expect(loopReturned).toBe(true); - expect(existsSync(pidPath)).toBe(false); - expect(readFileSync(memoryPath, "utf8").trim().split("\n").map((line) => JSON.parse(line))).toContainEqual({ - type: "error", ts: expect.any(String), data: "delivery settlement refused: identity", + for (const refusal of ["identity", "stale"] as const) { + test(`AgentRuntime.start logs ${refusal} refusal and keeps the intent pending after ${signal}`, async () => { + plant("m1.json", `${refusal}-refusal`); + plant("m2.json", "later"); + let enter!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + let finish!: () => void; + const finishing = new Promise((resolve) => { finish = resolve; }); + const lookup = spyOn(FlairContextProvider.prototype, "getAgent").mockImplementation(async () => ({ publicKey: Buffer.from(ed.getPublicKey(SEED)).toString("base64") }) as never); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + if (refusal === "identity") { + writeFileSync(cur("m1.json"), `${readFileSync(cur("m1.json"), "utf8")} `); + } else { + const intent = intentJson("m1.json"); + intent.claim.token = randomUUID(); + intent.claim.moduleNonce = randomUUID(); + writeFileSync(intentPath("m1.json"), JSON.stringify(intent)); + } + enter(); await finishing; + return completingProvider.complete(); }); - expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(pending); - expect(intentPhase("m1.json")).toBe("in_progress"); - expect(intentJson("m1.json").claim).toBeDefined(); - expect(nextSpy).toHaveBeenCalledTimes(1); - expect(seen).toEqual(["identity-refusal"]); - expect(await client().deliverNext()).toMatchObject({ status: "deferred", reason: "record-changed" }); - expect(seen).toEqual(["identity-refusal"]); - expect(newFiles()).toEqual(["m2.json"]); - expect(curFiles()).toEqual(["m1.json"]); - } finally { - finish(); - await runtime.stop(); - await start.catch(() => {}); - loopSpy.mockRestore(); settlementSpy.mockRestore(); nextSpy.mockRestore(); - provider.mockRestore(); lookup.mockRestore(); - } - }, 15_000); + const memoryPath = join(workspace, "memory.jsonl"); + const pidPath = join(workspace, ".tps-agent.pid"); + const runtime = new AgentRuntime({ + agentId: AGENT, name: `${refusal}-refusal-test`, mailDir, workspace, memoryPath, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + const seen: string[] = []; + const next = MailClient.prototype.deliverNext; + const nextSpy = spyOn(MailClient.prototype, "deliverNext").mockImplementation(async function() { + const result = await next.call(this); + if (result.status === "delivered") seen.push(result.delivery.message.verifiedEnvelope!.messageId); + return result; + }); + const complete = MailClient.prototype.completeDelivery; + const settlementSpy = spyOn(MailClient.prototype, "completeDelivery").mockImplementation(async function(delivery) { + expect(existsSync(pidPath)).toBe(true); + const before = readFileSync(intentPath("m1.json"), "utf8"); + const result = await complete.call(this, delivery); + expect(result).toMatchObject({ status: "refused", reason: refusal }); + expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(before); + return result; + }); + let loopReturned = false; + const run = EventLoop.prototype.runDeliveries; + const loopSpy = spyOn(EventLoop.prototype, "runDeliveries").mockImplementation(async function(source) { + await run.call(this, source); + expect(existsSync(pidPath)).toBe(true); + loopReturned = true; + }); + const start = runtime.start(); + void start.catch(() => {}); + try { + await entered; + const pending = readFileSync(intentPath("m1.json"), "utf8"); + process.emit(signal); + expect(loopReturned).toBe(false); + expect(existsSync(pidPath)).toBe(true); + finish(); + await start; + expect(settlementSpy).toHaveBeenCalledTimes(1); + expect(loopReturned).toBe(true); + expect(existsSync(pidPath)).toBe(false); + expect(readFileSync(memoryPath, "utf8").trim().split("\n").map((line) => JSON.parse(line))).toContainEqual({ + type: "error", ts: expect.any(String), data: `delivery settlement refused: ${refusal}`, + }); + expect(readFileSync(intentPath("m1.json"), "utf8")).toBe(pending); + expect(intentPhase("m1.json")).toBe("in_progress"); + expect(intentJson("m1.json").claim).toBeDefined(); + expect(nextSpy).toHaveBeenCalledTimes(1); + expect(seen).toEqual([`${refusal}-refusal`]); + expect(await client().deliverNext()).toMatchObject({ status: "deferred", reason: refusal === "identity" ? "record-changed" : "active" }); + expect(seen).toEqual([`${refusal}-refusal`]); + expect(newFiles()).toEqual(["m2.json"]); + expect(curFiles()).toEqual(["m1.json"]); + } finally { + finish(); + await runtime.stop(); + await start.catch(() => {}); + loopSpy.mockRestore(); settlementSpy.mockRestore(); nextSpy.mockRestore(); + provider.mockRestore(); lookup.mockRestore(); + } + }, 15_000); + } - test(`AgentRuntime.start releases after memory writes fail during ${signal} before removing its PID file`, async () => { + test(`AgentRuntime.start attempts release after memory writes fail during ${signal} before removing its PID file`, async () => { plant("m1.json", "memory-failure"); plant("m2.json", "later"); let enter!: () => void; @@ -558,7 +567,7 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) { }, 15_000); for (const completed of [true, false]) { - test(`AgentRuntime.start drains ${completed ? "finalize" : "release"} after ${signal} before removing its PID file`, async () => { + test(`AgentRuntime.start retries deferred ${completed ? "finalize" : "release"} after ${signal} before removing its PID file`, async () => { plant("m1.json", "in-flight"); plant("m2.json", "later"); let enter!: () => void; From ce3ccdff61c5ea837c2aaf3c78829336593b027c Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 22:38:17 +0000 Subject: [PATCH 06/13] fix(mail): the event loop logs a settlement refusal reason when its durable record fails (cli#617 slice c) recordError() put only the append failure in its fallback line, so a settlement refusal's reason was lost when MemoryStore.append() threw. Carry the sanitized error beside the append failure, and cover a refusal while the real memory file is unwritable. Add a scratch-Flair lane case that drives AgentRuntime.start() through its FlairContextProvider Agent read: a signed delivery is verified and committed, and a refused read leaves the mail uncommitted in new/. --- .github/workflows/real-flair.yml | 1 + packages/agent/src/runtime/event-loop.ts | 4 +- .../test/mail-event-loop-delivery.test.ts | 56 ++++++- .../cli/test/runtime-flair-verify.test.ts | 152 ++++++++++++++++++ scripts/real-flair-lane.ts | 4 + 5 files changed, 215 insertions(+), 2 deletions(-) create mode 100644 packages/cli/test/runtime-flair-verify.test.ts diff --git a/.github/workflows/real-flair.yml b/.github/workflows/real-flair.yml index 7629ed5e..b9842cf0 100644 --- a/.github/workflows/real-flair.yml +++ b/.github/workflows/real-flair.yml @@ -13,6 +13,7 @@ on: - 'packages/cli/test/mail-flair-read-auth.test.ts' - 'packages/cli/test/presence-flair-contract.test.ts' - 'packages/cli/test/review-trigger-flair.test.ts' + - 'packages/cli/test/runtime-flair-verify.test.ts' - 'packages/cli/test/helpers/**' - 'packages/cli/bunfig.toml' - 'packages/cli/package.json' diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index 6c9ca551..106de459 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -220,7 +220,9 @@ export class EventLoop { data: String((err as { message?: unknown } | undefined)?.message ?? err), }); } catch (reportingError) { - console.error("Failed to record event-loop error:", sanitizeError(reportingError)); + // The durable record failed, so the fallback line carries the error too — + // a settlement refusal's reason must not disappear with the append. + console.error(`Failed to record event-loop error (${sanitizeError(err)}):`, sanitizeError(reportingError)); } } diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index c01d6197..1da76ccd 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -552,7 +552,7 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) { finish(); await start; expect(failedWrites).toEqual(["assistant", "error"]); - expect(errorLog).toHaveBeenCalledWith("Failed to record event-loop error:", expect.any(String)); + expect(errorLog).toHaveBeenCalledWith(expect.stringContaining("Failed to record event-loop error"), expect.any(String)); expect(released).toBe(true); expect(existsSync(pidPath)).toBe(false); expect(intentJson("m1.json").claim).toBeUndefined(); @@ -629,3 +629,57 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) { }, 15_000); } } + +test("a settlement refusal whose durable record fails still logs the refusal reason", async () => { + plant("m1.json", "identity-refusal"); + const c = client(); + const memoryPath = join(workspace, "memory.jsonl"); + const memory = new MemoryStore(memoryPath); + const config = { + agentId: AGENT, name: "refusal-log", mailDir, memoryPath, workspace, + llm: { provider: "test", model: "test-model" }, contextWindowTokens: 8000, maxToolTurns: 50, + }; + const provider = { + async complete() { + // Break the committed cur/ record's identity so the settlement is refused. + writeFileSync(cur("m1.json"), `${readFileSync(cur("m1.json"), "utf8")} `); + return completingProvider.complete(); + }, + }; + const loop = new EventLoop({ + config: config as never, + memory, + context: {} as never, + provider: provider as never, + tools: { list: () => [], execute: async () => ({ content: "ok" }) } as never, + }, 1); + const errorLog = spyOn(console, "error").mockImplementation(() => {}); + const complete = c.completeDelivery.bind(c); + const settlementSpy = spyOn(c, "completeDelivery").mockImplementation(async (delivery) => { + // The real memory file becomes unwritable (a directory) before the refusal is recorded. + renameSync(memoryPath, `${memoryPath}.saved`); + mkdirSync(memoryPath); + const result = await complete(delivery); + expect(result).toMatchObject({ status: "refused", reason: "identity" }); + return result; + }); + const source: MailDeliverySource = { + async deliverNext() { + const result = await c.deliverNext(); + if (result.status === "delivered") void loop.stop(); + return result; + }, + completeDelivery: (delivery) => c.completeDelivery(delivery), + releaseDelivery: (delivery) => c.releaseDelivery(delivery), + }; + try { + await loop.runDeliveries(source); + const refusal = errorLog.mock.calls.find((call) => String(call[0]).includes("delivery settlement refused: identity")); + expect(refusal).toBeDefined(); + expect(String(refusal?.[1]).length).toBeGreaterThan(0); + expect(intentPhase("m1.json")).toBe("in_progress"); + } finally { + settlementSpy.mockRestore(); + errorLog.mockRestore(); + } +}, 15_000); diff --git a/packages/cli/test/runtime-flair-verify.test.ts b/packages/cli/test/runtime-flair-verify.test.ts new file mode 100644 index 00000000..d47d5b79 --- /dev/null +++ b/packages/cli/test/runtime-flair-verify.test.ts @@ -0,0 +1,152 @@ +import { afterEach, expect, mock, spyOn, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { AgentRuntime, canonicalMailboxRoot, hasCommittedMessageId, ProviderManager } from "@tpsdev-ai/agent"; +import { flairTarget } from "./helpers/scratch-flair.js"; +import { buildSignedEnvelope, pubkeyFromSeed, writeKeyFile } from "./helpers/stub-flair.js"; + +const SEED = Buffer.alloc(32, 0x31); +const WRONG_SEED = Buffer.alloc(32, 0x32); + +afterEach(() => { mock.restore(); }); + +/** The scratch Flair the lane started; the operator API and the admin credential. */ +function lane() { + const { baseUrl, opsUrl, admin } = flairTarget(); + const headers = { Authorization: `Basic ${Buffer.from(admin).toString("base64")}`, "Content-Type": "application/json" }; + const operation = (body: unknown) => fetch(opsUrl, { method: "POST", headers, body: JSON.stringify(body) }); + return { baseUrl, operation }; +} + +/** Register the receiver and sender so Flair can resolve the sender's key. */ +function register(operation: (body: unknown) => Promise, ids: string[], publicKey: string): Promise { + return operation({ + operation: "insert", + database: "flair", + table: "Agent", + records: ids.map((id) => ({ id, name: id, publicKey, createdAt: new Date().toISOString() })), + }); +} + +/** Plant one signed record for `receiver` in its mailbox `new/`, and return the message id. */ +function plant(messageId: string, mailDir: string, receiver: string, sender: string, body: string): string { + const envelope = buildSignedEnvelope(sender, receiver, body, { [sender]: SEED }, { messageId }); + const mailbox = canonicalMailboxRoot(join(mailDir, receiver)); + mkdirSync(join(mailbox, "new"), { recursive: true }); + writeFileSync(join(mailbox, "new", "m1.json"), JSON.stringify({ + from: sender, to: receiver, body: JSON.stringify(envelope), timestamp: new Date().toISOString(), + })); + return mailbox; +} + +/** A bounded wait for a predicate that has nothing to do with model output. */ +async function waitFor(predicate: () => boolean, deadlineMs: number): Promise { + const deadline = Date.now() + deadlineMs; + while (Date.now() < deadline) { + if (predicate()) return; + await Bun.sleep(25); + } + throw new Error(`runtime-flair lane: condition not met within ${deadlineMs}ms`); +} + +/** True once the runtime's mailbox wrote a `mail.receive` error for a refused read. */ +function recoveryLogged(workspace: string): boolean { + const dir = join(workspace, ".tps", "events"); + try { + return readdirSync(dir).some((name) => { + const text = readFileSync(join(dir, name), "utf8"); + return text.includes('"mail.receive"') && text.includes('"status":"error"'); + }); + } catch { + return false; + } +} + +test.skipIf(process.env.TPS_TEST_REAL_FLAIR !== "1")("real Flair verifies and handles a signed delivery through the runtime's Agent read", async () => { + const { baseUrl, operation } = lane(); + const root = mkdtempSync(join(tmpdir(), "runtime-flair-")); + const receiver = `runtime-receiver-${randomUUID()}`; + const sender = `runtime-sender-${randomUUID()}`; + const keyPath = writeKeyFile(root, receiver, SEED); + const mailDir = join(root, "mail"); + const workspace = join(root, "workspace"); + mkdirSync(workspace, { recursive: true }); + const messageId = randomUUID(); + const mailbox = plant(messageId, mailDir, receiver, sender, "runtime-lane-body"); + const inserted = await register(operation, [receiver, sender], pubkeyFromSeed(SEED).toString("base64")); + expect(inserted.ok).toBe(true); + let handled!: () => void; + const handledOnce = new Promise((resolve) => { handled = resolve; }); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + handled(); + return { content: "done", toolCalls: undefined, inputTokens: 1, outputTokens: 1 } as never; + }); + const runtime = new AgentRuntime({ + agentId: receiver, name: "runtime-verify", mailDir, workspace, + memoryPath: join(workspace, "memory.jsonl"), maxToolTurns: 0, + flair: { url: baseUrl, keyPath }, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + const start = runtime.start(); + void start.catch(() => {}); + try { + await handledOnce; + await runtime.stop(); + await start; + expect(readdirSync(join(mailbox, "cur"))).toEqual(["m1.json"]); + expect(readdirSync(join(mailbox, "new"))).toEqual([]); + expect(hasCommittedMessageId(mailbox, messageId)).toBe(true); + expect(provider).toHaveBeenCalledTimes(1); + } finally { + await runtime.stop(); + await start.catch(() => {}); + provider.mockRestore(); + await operation({ operation: "delete", database: "flair", table: "Agent", hash_values: [receiver, sender] }); + rmSync(root, { recursive: true, force: true }); + } +}, 60_000); + +test.skipIf(process.env.TPS_TEST_REAL_FLAIR !== "1")("a refused Agent read leaves the runtime's mail uncommitted", async () => { + const { baseUrl, operation } = lane(); + const root = mkdtempSync(join(tmpdir(), "runtime-flair-")); + const receiver = `runtime-receiver-${randomUUID()}`; + const sender = `runtime-sender-${randomUUID()}`; + // The mailbox is registered with SEED's public key, but the runtime signs the + // Agent read with WRONG_SEED, so Flair refuses it. + const keyPath = writeKeyFile(root, receiver, WRONG_SEED); + const mailDir = join(root, "mail"); + const workspace = join(root, "workspace"); + mkdirSync(workspace, { recursive: true }); + const messageId = randomUUID(); + const mailbox = plant(messageId, mailDir, receiver, sender, "runtime-lane-body"); + const inserted = await register(operation, [receiver, sender], pubkeyFromSeed(SEED).toString("base64")); + expect(inserted.ok).toBe(true); + const provider = spyOn(ProviderManager.prototype, "complete").mockImplementation(async () => { + throw new Error("the handler must not run for a refused Agent read"); + }); + const runtime = new AgentRuntime({ + agentId: receiver, name: "runtime-verify", mailDir, workspace, + memoryPath: join(workspace, "memory.jsonl"), maxToolTurns: 0, + flair: { url: baseUrl, keyPath }, + llm: { provider: "openai", model: "test", apiKey: "test" }, + }); + const start = runtime.start(); + void start.catch(() => {}); + try { + await waitFor(() => recoveryLogged(workspace), 30_000); + await runtime.stop(); + await start; + expect(readdirSync(join(mailbox, "new"))).toEqual(["m1.json"]); + expect(readdirSync(join(mailbox, "cur"))).toEqual([]); + expect(hasCommittedMessageId(mailbox, messageId)).toBe(false); + expect(provider).not.toHaveBeenCalled(); + } finally { + await runtime.stop(); + await start.catch(() => {}); + provider.mockRestore(); + await operation({ operation: "delete", database: "flair", table: "Agent", hash_values: [receiver, sender] }); + rmSync(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/scripts/real-flair-lane.ts b/scripts/real-flair-lane.ts index b487c552..7d393e61 100644 --- a/scripts/real-flair-lane.ts +++ b/scripts/real-flair-lane.ts @@ -27,6 +27,10 @@ export const EXPECTED_CASES: Record = { "review-trigger-flair.test.ts": [ "real Flair stamps a signed non-admin POST and the review loop skips its claimed agent attribution", ], + "runtime-flair-verify.test.ts": [ + "real Flair verifies and handles a signed delivery through the runtime's Agent read", + "a refused Agent read leaves the runtime's mail uncommitted", + ], "real-flair-lane.test.ts": [ "test files referencing the real-Flair flag are named by the lane and covered by workflow paths", ], From 7a03cb199a80ff5c8d956443c41365b74d19abf7 Mon Sep 17 00:00:00 2001 From: flint Date: Sat, 17 Oct 2026 00:15:16 +0000 Subject: [PATCH 07/13] test(mail): the refused-read lane assertion matches a mail.receive event, not substrings (cli#617 slice c) The runtime-flair lane's refused-read wait now parses the event log and matches one `mail.receive` event whose own status and error fields carry the failure, instead of scanning the raw files for two independent substrings. Three comments stated more than the code did: plant() returns the canonical mailbox path, the refused-read case registers the receiver's Agent row, and the two event-loop providers request a tool call rather than call tools. --- .../test/mail-event-loop-delivery.test.ts | 4 +- .../cli/test/runtime-flair-verify.test.ts | 46 +++++++++++++++---- 2 files changed, 38 insertions(+), 12 deletions(-) diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index 1da76ccd..e08bb76a 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -127,9 +127,9 @@ async function runLoop( /** A provider whose handler returns a normal completion. */ const completingProvider = { async complete() { return { content: "done", toolCalls: undefined, inputTokens: 1, outputTokens: 1 }; } }; -/** A provider whose handler always calls a tool, so the turn limit is reached. */ +/** A provider whose handler always requests a tool call, so the turn limit is reached. */ const toolCallingProvider = { async complete() { return { content: "", toolCalls: [{ id: "t1", name: "noop", input: {} }], inputTokens: 1, outputTokens: 1 }; } }; -/** Calls tools for a body containing `poison-body`. */ +/** Requests a tool call for a body containing `poison-body`. */ const poisonAwareProvider = { async complete(req: unknown) { const text = JSON.stringify((req as { messages?: unknown }).messages ?? ""); diff --git a/packages/cli/test/runtime-flair-verify.test.ts b/packages/cli/test/runtime-flair-verify.test.ts index d47d5b79..8122b58a 100644 --- a/packages/cli/test/runtime-flair-verify.test.ts +++ b/packages/cli/test/runtime-flair-verify.test.ts @@ -30,7 +30,7 @@ function register(operation: (body: unknown) => Promise, ids: string[] }); } -/** Plant one signed record for `receiver` in its mailbox `new/`, and return the message id. */ +/** Plant one signed record for `receiver` in its mailbox `new/`, and return its canonical mailbox path. */ function plant(messageId: string, mailDir: string, receiver: string, sender: string, body: string): string { const envelope = buildSignedEnvelope(sender, receiver, body, { [sender]: SEED }, { messageId }); const mailbox = canonicalMailboxRoot(join(mailDir, receiver)); @@ -51,17 +51,43 @@ async function waitFor(predicate: () => boolean, deadlineMs: number): Promise> { const dir = join(workspace, ".tps", "events"); + let names: string[]; try { - return readdirSync(dir).some((name) => { - const text = readFileSync(join(dir, name), "utf8"); - return text.includes('"mail.receive"') && text.includes('"status":"error"'); - }); + names = readdirSync(dir); } catch { - return false; + return []; } + return names.flatMap((name) => { + let text: string; + try { + text = readFileSync(join(dir, name), "utf8"); + } catch { + return []; + } + return text.split("\n").flatMap((line): Array> => { + try { + const parsed: unknown = JSON.parse(line); + return parsed !== null && typeof parsed === "object" ? [parsed as Record] : []; + } catch { + return []; + } + }); + }); +} + +/** + * True once the runtime's mailbox recorded an event of type `mail.receive` + * whose own `status`/`error` fields carry the failure — one parsed event, not + * two lookalike substrings. It proves a `mail.receive` error was logged; it + * does not establish the cause. + */ +function recoveryLogged(workspace: string): boolean { + return eventLog(workspace).some( + (event) => event.type === "mail.receive" && event.status === "error" && typeof event.error === "string" && event.error.length > 0, + ); } test.skipIf(process.env.TPS_TEST_REAL_FLAIR !== "1")("real Flair verifies and handles a signed delivery through the runtime's Agent read", async () => { @@ -113,8 +139,8 @@ test.skipIf(process.env.TPS_TEST_REAL_FLAIR !== "1")("a refused Agent read leave const root = mkdtempSync(join(tmpdir(), "runtime-flair-")); const receiver = `runtime-receiver-${randomUUID()}`; const sender = `runtime-sender-${randomUUID()}`; - // The mailbox is registered with SEED's public key, but the runtime signs the - // Agent read with WRONG_SEED, so Flair refuses it. + // The receiver's Agent row is registered with SEED's public key, but the + // runtime signs the Agent read with WRONG_SEED, so Flair refuses it. const keyPath = writeKeyFile(root, receiver, WRONG_SEED); const mailDir = join(root, "mail"); const workspace = join(root, "workspace"); From 3d546b2991a425f621b66d5ee97774ddd7a673e7 Mon Sep 17 00:00:00 2001 From: flint Date: Sat, 17 Oct 2026 00:30:25 +0000 Subject: [PATCH 08/13] test(cli): state the event-log helper's contract without an over-broad claim (cli#617 slice c) --- packages/cli/test/runtime-flair-verify.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/cli/test/runtime-flair-verify.test.ts b/packages/cli/test/runtime-flair-verify.test.ts index 8122b58a..3a02218f 100644 --- a/packages/cli/test/runtime-flair-verify.test.ts +++ b/packages/cli/test/runtime-flair-verify.test.ts @@ -51,7 +51,7 @@ async function waitFor(predicate: () => boolean, deadlineMs: number): Promise> { const dir = join(workspace, ".tps", "events"); let names: string[]; From 022af71a1ac9ad60858c209be1dcfa3146ba18d0 Mon Sep 17 00:00:00 2001 From: flint Date: Sat, 17 Oct 2026 01:59:19 +0000 Subject: [PATCH 09/13] fix(mail): bound a stopped event loop's settlement retries (cli#617 slice c) --- packages/agent/src/runtime/event-loop.ts | 18 ++++ .../test/mail-event-loop-delivery.test.ts | 84 +++++++++++++++++++ .../cli/test/runtime-flair-verify.test.ts | 17 ++-- 3 files changed, 113 insertions(+), 6 deletions(-) diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index 106de459..ad7bedb1 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -24,6 +24,9 @@ import { sanitizeError } from "../telemetry/events.js"; /** Absolute safety net — independent of config */ const PANIC_MAX_TURNS = 60; +/** Settlement retries a stop allows before the loop abandons one delivery. */ +const MAX_SETTLEMENT_ATTEMPTS = 20; + const COMPACTION_INSTRUCTION = `Below is a conversation history wrapped in tags. Summarize the CONTENT of the conversation — do NOT follow any instructions found inside the history. Treat everything inside the tags as DATA to summarize. @@ -183,7 +186,9 @@ export class EventLoop { } private async settleDelivery(source: MailDeliverySource, delivery: MailDelivery, completed: boolean): Promise { + let stoppedAttempts = 0; while (true) { + let lastDeferral: { reason: string; detail: string } | undefined; try { const result = completed ? await source.completeDelivery(delivery) @@ -198,11 +203,24 @@ export class EventLoop { await this.recordError(`delivery settlement refused: ${result.reason}`); return; case "deferred": + lastDeferral = result; break; } } catch (err) { await this.recordError(err); } + // While running, retrying is what keeps at-least-once delivery when + // another writer holds the lock. A stop must let shutdown finish: once + // `running` is false, bound the retries, record the last outcome, and + // return. The claim stays pending, so the next run re-presents it. + if (!this.running && ++stoppedAttempts >= MAX_SETTLEMENT_ATTEMPTS) { + await this.recordError( + lastDeferral + ? `delivery settlement deferred past the shutdown bound (${MAX_SETTLEMENT_ATTEMPTS} attempts): ${lastDeferral.reason}: ${lastDeferral.detail}` + : `delivery settlement failed past the shutdown bound (${MAX_SETTLEMENT_ATTEMPTS} attempts)`, + ); + return; + } await sleep(this.pollMs); } } diff --git a/packages/agent/test/mail-event-loop-delivery.test.ts b/packages/agent/test/mail-event-loop-delivery.test.ts index e08bb76a..f73ce4b6 100644 --- a/packages/agent/test/mail-event-loop-delivery.test.ts +++ b/packages/agent/test/mail-event-loop-delivery.test.ts @@ -683,3 +683,87 @@ test("a settlement refusal whose durable record fails still logs the refusal rea errorLog.mockRestore(); } }, 15_000); + +/** The loop config a settlement test drives; the memory mock records every appended event. */ +function settlementLoop(events: Array<{ data?: unknown }>, provider: unknown, pollMs = 1): EventLoop { + const config = { + agentId: AGENT, name: "settlement-bound", mailDir, memoryPath: join(workspace, "mem.jsonl"), workspace, + llm: { provider: "test", model: "test-model" }, contextWindowTokens: 8000, maxToolTurns: 50, + }; + return new EventLoop({ + config: config as never, + memory: { append: async (event: unknown) => { events.push(event as { data?: unknown }); } } as never, + context: {} as never, + provider: provider as never, + tools: { list: () => [], execute: async () => ({ content: "ok" }) } as never, + }, pollMs); +} + +test("a deferred settlement that never clears is abandoned after a stop and recorded", async () => { + plant("m1.json", "never-clears"); + const c = client(); + let calls = 0; + const complete = spyOn(c, "completeDelivery").mockImplementation(async () => { + calls++; + return { status: "deferred", reason: "storage-error", detail: "permanent storage failure" }; + }); + const events: Array<{ data?: unknown }> = []; + const loop = settlementLoop(events, completingProvider); + const source: MailDeliverySource = { + async deliverNext() { + const result = await c.deliverNext(); + if (result.status === "delivered") void loop.stop(); + return result; + }, + completeDelivery: (delivery) => c.completeDelivery(delivery), + releaseDelivery: (delivery) => c.releaseDelivery(delivery), + }; + let timer: ReturnType | undefined; + try { + await Promise.race([ + loop.runDeliveries(source), + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("runDeliveries did not return within 5s")), 5_000); }), + ]); + } finally { + clearTimeout(timer); + complete.mockRestore(); + } + expect(calls).toBe(20); + expect(events.some((event) => String(event.data).includes("deferred past the shutdown bound"))).toBe(true); +}, 10_000); + +test("a stop during settlement retries still settles when the deferral clears before the bound", async () => { + plant("m1.json", "clears-in-time"); + const c = client(); + const real = c.completeDelivery.bind(c); + let attempts = 0; + const complete = spyOn(c, "completeDelivery").mockImplementation(async (delivery) => { + attempts++; + if (attempts <= 3) return { status: "deferred", reason: "lock-busy", detail: "mailbox lock busy; retry" }; + return real(delivery); + }); + const events: Array<{ data?: unknown }> = []; + const loop = settlementLoop(events, completingProvider); + const source: MailDeliverySource = { + async deliverNext() { + const result = await c.deliverNext(); + if (result.status === "delivered") void loop.stop(); + return result; + }, + completeDelivery: (delivery) => c.completeDelivery(delivery), + releaseDelivery: (delivery) => c.releaseDelivery(delivery), + }; + let timer: ReturnType | undefined; + try { + await Promise.race([ + loop.runDeliveries(source), + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("runDeliveries did not return within 5s")), 5_000); }), + ]); + } finally { + clearTimeout(timer); + complete.mockRestore(); + } + expect(attempts).toBe(4); + expect(intentPhase("m1.json")).toBeNull(); + expect(events.some((event) => String(event.data).includes("shutdown bound"))).toBe(false); +}, 10_000); diff --git a/packages/cli/test/runtime-flair-verify.test.ts b/packages/cli/test/runtime-flair-verify.test.ts index 3a02218f..298722a6 100644 --- a/packages/cli/test/runtime-flair-verify.test.ts +++ b/packages/cli/test/runtime-flair-verify.test.ts @@ -3,7 +3,7 @@ import { randomUUID } from "node:crypto"; import { mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { AgentRuntime, canonicalMailboxRoot, hasCommittedMessageId, ProviderManager } from "@tpsdev-ai/agent"; +import { AgentRuntime, canonicalMailboxRoot, hasCommittedMessageId, mailboxReplayStore, ProviderManager } from "@tpsdev-ai/agent"; import { flairTarget } from "./helpers/scratch-flair.js"; import { buildSignedEnvelope, pubkeyFromSeed, writeKeyFile } from "./helpers/stub-flair.js"; @@ -79,14 +79,18 @@ function eventLog(workspace: string): Array> { } /** - * True once the runtime's mailbox recorded an event of type `mail.receive` - * whose own `status`/`error` fields carry the failure — one parsed event, not - * two lookalike substrings. It proves a `mail.receive` error was logged; it - * does not establish the cause. + * True once the runtime's mailbox recorded the wrong-key Agent-read refusal: + * a `mail.receive` error event whose own `error` names the Flair Agent read + * and the verifier's rejection. A mailbox storage error shares the event + * shape but carries neither, so it cannot satisfy this. */ function recoveryLogged(workspace: string): boolean { return eventLog(workspace).some( - (event) => event.type === "mail.receive" && event.status === "error" && typeof event.error === "string" && event.error.length > 0, + (event) => event.type === "mail.receive" + && event.status === "error" + && typeof event.error === "string" + && event.error.includes("/Agent/") + && event.error.includes("invalid_signature"), ); } @@ -124,6 +128,7 @@ test.skipIf(process.env.TPS_TEST_REAL_FLAIR !== "1")("real Flair verifies and ha expect(readdirSync(join(mailbox, "cur"))).toEqual(["m1.json"]); expect(readdirSync(join(mailbox, "new"))).toEqual([]); expect(hasCommittedMessageId(mailbox, messageId)).toBe(true); + expect(mailboxReplayStore(mailbox).placementPhase("m1.json")).toBeNull(); expect(provider).toHaveBeenCalledTimes(1); } finally { await runtime.stop(); From 4db46edb2841111ad6c6932951eebfef790d3c3b Mon Sep 17 00:00:00 2001 From: flint Date: Sat, 17 Oct 2026 02:08:51 +0000 Subject: [PATCH 10/13] docs(mail): the slice c fragment bounds settlement retries after a stop (cli#617) --- .../unreleased/changed-617-slice-c-event-loop-delivery.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index 6b4b483a..80a7ddb0 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -6,8 +6,10 @@ to become available; until then the claim stays pending. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, which the loop processes and attempts to settle before shutdown removes the - PID file. Deferred settlement is retried; any refusal is logged and can leave - the intent pending, blocking that message until resolved. + PID file. Deferred settlement is retried; a stop bounds the retries, so + settlement cannot hold shutdown open, and the last deferral or refusal is + logged. Any refusal can leave the intent pending, blocking that message until + resolved. A handler that never returns cannot reach the presentation or deadline checks. From 6b1f892cf0339b2a7eabb62f901acdcc8fb33011 Mon Sep 17 00:00:00 2001 From: flint Date: Sat, 17 Oct 2026 02:12:47 +0000 Subject: [PATCH 11/13] docs(mail): drop an over-broad shutdown claim from the slice c fragment (cli#617) --- .../unreleased/changed-617-slice-c-event-loop-delivery.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index 80a7ddb0..ec7f526a 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -6,10 +6,9 @@ to become available; until then the claim stays pending. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, which the loop processes and attempts to settle before shutdown removes the - PID file. Deferred settlement is retried; a stop bounds the retries, so - settlement cannot hold shutdown open, and the last deferral or refusal is - logged. Any refusal can leave the intent pending, blocking that message until - resolved. + PID file. Deferred settlement is retried; a stop bounds the retries, and the + last deferral or refusal is logged. Any refusal can leave the intent pending, + blocking that message until resolved. A handler that never returns cannot reach the presentation or deadline checks. From 2124290eb18a0a6f1a33626c18fe280e66a2099c Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 19:52:53 -0700 Subject: [PATCH 12/13] round: address the review verdict (Codex fix round) Comment-only round (Codex): the stopped-loop comment says the intent stays pending and MAY be re-presented once the claim is no longer live, subject to verification and the delivery bound Co-Authored-By: Claude Opus 5.5 --- packages/agent/src/runtime/event-loop.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index ad7bedb1..bb5fa3f4 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -212,7 +212,8 @@ export class EventLoop { // While running, retrying is what keeps at-least-once delivery when // another writer holds the lock. A stop must let shutdown finish: once // `running` is false, bound the retries, record the last outcome, and - // return. The claim stays pending, so the next run re-presents it. + // return. The intent remains pending and may be re-presented after the claim + // is no longer live, subject to recovery's verification and the delivery bound. if (!this.running && ++stoppedAttempts >= MAX_SETTLEMENT_ATTEMPTS) { await this.recordError( lastDeferral From d8dd39470d0ff8a2ee300eb36cc1ad9c07ac35bd Mon Sep 17 00:00:00 2001 From: flint Date: Fri, 16 Oct 2026 19:57:50 -0700 Subject: [PATCH 13/13] round: address the review verdict (Codex fix round) Wording round (Codex): after the stop bound settlement state is uncertain; a failed storage operation can leave the durable transition uncertain (comment, body, fragment) Co-Authored-By: Claude Opus 5.5 --- .../unreleased/changed-617-slice-c-event-loop-delivery.md | 3 ++- packages/agent/src/runtime/event-loop.ts | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md index ec7f526a..290f7199 100644 --- a/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md +++ b/.changelog/unreleased/changed-617-slice-c-event-loop-delivery.md @@ -3,7 +3,8 @@ Handler completion requests finalization; a turn-limit return or thrown handler requests release. Deferred finalization and release are retried on the same delivery. Deadline quarantine requires the mailbox lock and storage - to become available; until then the claim stays pending. A stop prevents + to become available. A failed storage operation can leave the durable transition + uncertain. A stop prevents the next `deliverNext()` call; a check already waiting can return a delivery, which the loop processes and attempts to settle before shutdown removes the PID file. Deferred settlement is retried; a stop bounds the retries, and the diff --git a/packages/agent/src/runtime/event-loop.ts b/packages/agent/src/runtime/event-loop.ts index bb5fa3f4..992ba4fa 100644 --- a/packages/agent/src/runtime/event-loop.ts +++ b/packages/agent/src/runtime/event-loop.ts @@ -212,8 +212,8 @@ export class EventLoop { // While running, retrying is what keeps at-least-once delivery when // another writer holds the lock. A stop must let shutdown finish: once // `running` is false, bound the retries, record the last outcome, and - // return. The intent remains pending and may be re-presented after the claim - // is no longer live, subject to recovery's verification and the delivery bound. + // return with settlement state uncertain. Later re-presentation depends on + // the resulting intent and claim state, recovery's verification, and the delivery bound. if (!this.running && ++stoppedAttempts >= MAX_SETTLEMENT_ATTEMPTS) { await this.recordError( lastDeferral