diff --git a/packages/cli/src/commands/mail-watch.ts b/packages/cli/src/commands/mail-watch.ts index 97fa78f..ae8699e 100644 --- a/packages/cli/src/commands/mail-watch.ts +++ b/packages/cli/src/commands/mail-watch.ts @@ -1,8 +1,11 @@ /** * mail-watch — watch an agent's inbox for new messages and run exec hooks. * - * OPS-121: event-driven mail watcher using fs.watch + debounce. - * Zero CPU overhead when idle — no polling. + * OPS-121: mail watcher using fs.watch + debounce for low-latency delivery, + * backed by a low-frequency poll fallback. fs.watch (FSEvents on macOS) goes + * deaf after uptime, so it can't be the sole trigger — the poll guarantees + * delivery and lets a (re)start recover any mail stranded in new/. Near-zero + * idle CPU; reliability over strictly-zero-CPU. * * Security mitigations (K&S): * - exec hooks use args[] array, no shell interpolation @@ -41,6 +44,8 @@ export interface MailWatchOptions { onMessage?: (msg: MailMessage) => void | Promise; /** Max concurrent handlers (default: 3) */ maxConcurrent?: number; + /** Polling-fallback interval in ms — guards against fs.watch going deaf (default: 15000) */ + pollMs?: number; } // --------------------------------------------------------------------------- @@ -145,21 +150,23 @@ export function watchMail(opts: MailWatchOptions): MailWatcher { const inbox = getInbox(opts.agent); - // Pre-populate seen with already-present files so we don't replay old mail - for (const f of listNewFiles(inbox.fresh)) { - seen.add(f); - } - + // NOTE: we intentionally do NOT pre-populate `seen` with files already in + // new/. new/ holds UNDELIVERED mail — the exec hook moves delivered mail to + // cur/ on ack — so anything present at (re)start must be delivered. Marking + // it seen on boot is what made a mail-watch kickstart silently strand stuck + // dispatches (and forced re-dispatches, which double-posted K&S reviews). const processNew = async () => { if (stopped) return; for (const filePath of listNewFiles(inbox.fresh)) { if (seen.has(filePath)) continue; + // Check the concurrency cap BEFORE marking seen. If we're at the limit, + // leave the file UNSEEN so the next pass (poll or fs event) retries it. + // The old order marked it seen and THEN dropped it → never delivered. + if (activeHandlers >= maxConcurrent) continue; seen.add(filePath); const msg = readMessageSafe(filePath); if (!msg) continue; // moved to cur/ before we could read — skip - - if (activeHandlers >= maxConcurrent) continue; // drop when at limit activeHandlers++; (async () => { @@ -169,7 +176,13 @@ export function watchMail(opts: MailWatchOptions): MailWatcher { try { if (opts.hook) await runHook(opts.hook, msg); } catch { /* hook errors don't crash the watcher */ } - })().finally(() => { activeHandlers--; }); + })().finally(() => { + activeHandlers--; + // A slot just freed — immediately re-check for mail that was over the + // concurrency cap on a prior pass, so a burst drains as slots cycle + // instead of waiting for the next fs event or poll. + if (!stopped) void processNew(); + }); } }; @@ -180,13 +193,24 @@ export function watchMail(opts: MailWatchOptions): MailWatcher { debounceTimer = setTimeout(() => { void processNew(); }, debounceMs); }; - // fs.watch on the new/ directory — fires on file create/rename + // fs.watch on new/ fires on file create/rename — low-latency, but fs.watch + // (FSEvents on macOS) silently stops emitting after uptime, which strands + // mail. So treat fs.watch as a latency optimization, NOT the source of truth: + // a low-frequency poll guarantees delivery even if the watcher goes deaf. + // Reliability > idle CPU for the review/mail pipeline. const watcher = fsWatch(inbox.fresh, onFsEvent); + const pollMs = opts.pollMs ?? 15_000; + const pollTimer = setInterval(() => { void processNew(); }, pollMs); + + // Deliver anything already waiting in new/ at (re)start, immediately — so a + // kickstart RECOVERS stuck mail instead of waiting for the first event/poll. + void processNew(); return { stop() { stopped = true; if (debounceTimer) clearTimeout(debounceTimer); + clearInterval(pollTimer); try { watcher.close(); } catch {} }, }; diff --git a/packages/cli/test/mail-watch.test.ts b/packages/cli/test/mail-watch.test.ts index 3da5e91..e4ad821 100644 --- a/packages/cli/test/mail-watch.test.ts +++ b/packages/cli/test/mail-watch.test.ts @@ -95,9 +95,13 @@ describe("watchMail (fs.watch)", () => { expect(received).toContain("msg-001"); }); - it("does not replay pre-existing messages present before watch starts", async () => { + it("delivers pre-existing undelivered messages waiting in new/ at startup", async () => { + // new/ holds UNDELIVERED mail — the exec hook moves delivered mail to cur/ + // on ack. So a (re)started watcher MUST deliver whatever is already waiting: + // this is the recovery path that lets a kickstart drain mail stranded by an + // fs.watch stall, instead of marking it seen and forcing a re-dispatch. const { fresh } = makeInbox(tmp, "test-agent"); - writeMsg(fresh, "old-msg", "sender", "test-agent", "pre-existing"); + writeMsg(fresh, "waiting-msg", "sender", "test-agent", "stranded before watch start"); const received: string[] = []; const watcher = watchMail({ @@ -108,7 +112,7 @@ describe("watchMail (fs.watch)", () => { await sleep(150); watcher.stop(); - expect(received).not.toContain("old-msg"); + expect(received).toContain("waiting-msg"); }); it("does not deliver the same message twice (dedup via seen set)", async () => { @@ -188,32 +192,37 @@ describe("watchMail concurrency", () => { rmSync(tmp, { recursive: true, force: true }); }); - it("respects maxConcurrent=2 — at most 2 handlers run simultaneously", async () => { + it("respects maxConcurrent=2 and still delivers every message (over-cap mail is retried)", async () => { const { fresh } = makeInbox(tmp, "conc-agent"); let active = 0; let maxSeen = 0; + const received: string[] = []; const watcher = watchMail({ agent: "conc-agent", debounceMs: 20, maxConcurrent: 2, - onMessage: async () => { + onMessage: async (msg) => { active++; maxSeen = Math.max(maxSeen, active); - await sleep(80); + await sleep(60); + received.push(msg.id); active--; }, }); await sleep(30); - // Write 5 messages — all arrive in a single debounce window + // Write 5 messages — all arrive in a single debounce window, over the cap. for (let i = 0; i < 5; i++) { writeMsg(fresh, `m${i}`, "s", "conc-agent", `body${i}`); } - await sleep(400); + await sleep(500); watcher.stop(); expect(maxSeen).toBeLessThanOrEqual(2); + // All 5 must be delivered: mail over the cap is left UNSEEN and retried as + // slots free (the old code marked it seen then dropped it → lost forever). + expect(received.sort()).toEqual(["m0", "m1", "m2", "m3", "m4"]); }); });