diff --git a/packages/cli/src/commands/pulse.ts b/packages/cli/src/commands/pulse.ts index 6641178..0200a0c 100644 --- a/packages/cli/src/commands/pulse.ts +++ b/packages/cli/src/commands/pulse.ts @@ -10,7 +10,7 @@ import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { createFlairClient } from "../utils/flair-client.js"; -import { gcMessages } from "../utils/mail.js"; +import { gcMessages, sendMessage } from "../utils/mail.js"; // --------------------------------------------------------------------------- // Types @@ -64,8 +64,11 @@ export interface PulseConfig { // Injectable runner type for testing export type SyncRunner = (cmd: string, args: string[], opts?: { encoding?: BufferEncoding; timeout?: number; env?: NodeJS.ProcessEnv }) => SpawnSyncReturns; -// Injectable mail sender for testing -export type MailSender = (to: string, body: string, agentId: string) => void; +// Injectable mail sender for testing. May return a Promise for async +// senders; sendMail handles both sync throws and async rejections, and +// races async senders against a timeout so one hung delivery cannot +// wedge the daemon. +export type MailSender = (to: string, body: string, agentId: string) => void | Promise; // Injectable Flair publisher for testing (null = disabled) export type FlairPublisher = ( @@ -162,16 +165,54 @@ export function ghApi(endpoint: string, ghAgent: string, runner: SyncRunner = sp // Mail // --------------------------------------------------------------------------- +/** Per-send timeout — defense in depth. sendMessage is synchronous and + * fast, but an injected async sender (e.g. a future bridge transport) + * could hang. One hung delivery must not stop notifications for everyone + * else. + * + * Settable at runtime via setSendTimeoutMs for tests that need a short + * timeout. */ +let SEND_TIMEOUT_MS = 5_000; + +export function setSendTimeoutMs(ms: number): void { + SEND_TIMEOUT_MS = ms; +} + export function defaultMailSender(to: string, body: string, agentId: string): void { - spawnSync("tps", ["mail", "send", to, body], { - encoding: "utf-8", - env: { ...process.env, TPS_AGENT_ID: agentId }, - }); + // Call sendMessage in-process instead of shelling out to the 'tps' PATH shim. + // The shim hangs on mail send (observed on @tpsdev-ai/cli 0.5.4) and has no + // spawnSync timeout — a single undeliverable message wedges the pulse daemon. + // In-process call eliminates both the shim dependency and the hang vector. + try { + sendMessage(to, body, agentId); + } catch (e: unknown) { + // Defense in depth: a single bad recipient must not stop the notification loop. + // Log loudly and continue. Examples: "Inbox full", disk full, invalid agent id. + console.error(`[pulse/mail] FAILED to send to ${to}: ${(e as Error).message}`); + } } function sendMail(to: string, body: string, config: PulseConfig, sender: MailSender): void { console.log(`[pulse] mail → ${to}: ${body.slice(0, 80)}…`); - sender(to, body, config.ghAgent); + try { + const result = sender(to, body, config.ghAgent); + if (result instanceof Promise) { + // Async sender — race against timeout so one hung delivery cannot + // wedge the daemon. + const timeout = new Promise((_, reject) => + setTimeout( + () => reject(new Error(`mail send to ${to} timed out after ${SEND_TIMEOUT_MS}ms`)), + SEND_TIMEOUT_MS, + ), + ); + Promise.race([result, timeout]).catch((e: unknown) => { + console.error(`[pulse/mail] FAILED to send to ${to}: ${(e as Error).message}`); + }); + } + } catch (e: unknown) { + // One bad recipient must not stop the world. Log loudly and continue. + console.error(`[pulse/mail] FAILED to send to ${to}: ${(e as Error).message}`); + } } // --------------------------------------------------------------------------- diff --git a/packages/cli/test/pulse.test.ts b/packages/cli/test/pulse.test.ts index 4f0c7fe..63086f1 100644 --- a/packages/cli/test/pulse.test.ts +++ b/packages/cli/test/pulse.test.ts @@ -7,6 +7,7 @@ import { printStatus, pruneState, startPollLoop, + setSendTimeoutMs, type PrInstance, type PrState, type PulseConfig, @@ -571,3 +572,185 @@ describe("FlairPublisher integration", () => { expect(calls).toHaveLength(0); }); }); + +// --------------------------------------------------------------------------- +// Mail send failure resilience +// +// The published PATH shim hung forever on mail send, and spawnSync had no +// timeout. One undeliverable message wedged the pulse daemon permanently. +// These tests assert that a failed or hung sender does not block subsequent +// notifications. +// --------------------------------------------------------------------------- + +describe("mail send failure resilience", () => { + // ── Ember's tests (kept — complementary coverage) ────────────────── + + test("sendMail catches sender errors and continues the loop", () => { + const config = makeConfig(); + const { calls } = trackMails(); + const instance = makeInstance({ state: "opened" }); + + let callCount = 0; + const failingSender: MailSender = (to, body, agentId) => { + callCount++; + if (callCount === 1) throw new Error("simulated send hang/failure"); + calls.push({ to, body, agentId }); + }; + + // handleTransition for opened → approved sends 1 mail to mergeAuthority. + expect(() => { + handleTransition("pr:tpsdev-ai/cli#42", instance, "approved", config, failingSender); + }).not.toThrow(); + + expect(instance.state).toBe("approved"); + + // Second transition should succeed + expect(() => { + handleTransition("pr:tpsdev-ai/cli#42", instance, "merged", config, failingSender); + }).not.toThrow(); + + expect(instance.state).toBe("merged"); + expect(calls.length).toBe(1); + expect(calls[0].to).toBe("anvil"); + }); + + test("pollOnce continues processing PRs when mail send fails for one PR", () => { + const config = makeConfig(); + const { calls } = trackMails(); + const state = makeState(); + + let callCount = 0; + const failingSender: MailSender = (to, body, agentId) => { + callCount++; + if (callCount <= 2) throw new Error("simulated send failure for first PR"); + calls.push({ to, body, agentId }); + }; + + const runner: SyncRunner = (_cmd, args) => { + const endpoint = args[2]; + if (endpoint?.includes("/pulls?")) { + return { + status: 0, + stdout: JSON.stringify([ + { number: 10, title: "PR A", state: "open", merged_at: null, user: { login: "anvil" }, requested_reviewers: [] }, + { number: 11, title: "PR B", state: "open", merged_at: null, user: { login: "anvil" }, requested_reviewers: [] }, + ]), + stderr: "", + } as ReturnType; + } + if (endpoint?.includes("/reviews")) { + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + } + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + }; + + expect(() => { + pollOnce(config, state, runner, failingSender); + }).not.toThrow(); + + expect(state.instances["pr:tpsdev-ai/cli#10"]).toBeDefined(); + expect(state.instances["pr:tpsdev-ai/cli#11"]).toBeDefined(); + expect(calls.length).toBe(2); + expect(calls[0].body).toContain("PR #11"); + expect(calls[1].body).toContain("PR #11"); + }); + + // ── Hang + timeout tests (anvil) ─────────────────────────────────── + + test("hung sender does not block subsequent notifications", () => { + const config = makeConfig(); + const state = makeState(); + + const mailLog: string[] = []; + let hangCount = 0; + + const sender: MailSender = (to, _body, _agentId) => { + if (hangCount === 0) { + hangCount++; + return new Promise(() => {}); // never resolves + } + mailLog.push(to); + }; + + const runner: SyncRunner = (_cmd, args) => { + const endpoint = args[2]; + if (endpoint?.includes("/pulls?")) { + return { + status: 0, + stdout: JSON.stringify([ + { number: 10, title: "PR A", state: "open", merged_at: null, user: { login: "anvil" }, requested_reviewers: [] }, + { number: 11, title: "PR B", state: "open", merged_at: null, user: { login: "anvil" }, requested_reviewers: [] }, + ]), + stderr: "", + } as ReturnType; + } + if (endpoint?.includes("/reviews")) { + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + } + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + }; + + pollOnce(config, state, runner, sender); + + expect(state.instances["pr:tpsdev-ai/cli#10"]).toBeDefined(); + expect(state.instances["pr:tpsdev-ai/cli#11"]).toBeDefined(); + // First PR's first mail hung; 3 mails succeed (kern for PR #10, + // sherlock + kern for PR #11). + expect(mailLog.length).toBe(3); + expect(mailLog[0]).toBe("kern"); + expect(mailLog[1]).toBe("sherlock"); + expect(mailLog[2]).toBe("kern"); + }); + + test("slow async sender is timed out, subsequent notifications still delivered", async () => { + setSendTimeoutMs(100); + const config = makeConfig(); + const state = makeState(); + + const errors: string[] = []; + const originalError = console.error; + console.error = (msg: string) => { errors.push(msg); }; + + const mailLog: string[] = []; + let slowResolved = false; + + try { + const sender: MailSender = (to, _body, _agentId) => { + if (to === "sherlock") { + return new Promise((resolve) => { + setTimeout(() => { slowResolved = true; resolve(); }, 500); + }); + } + mailLog.push(to); + }; + + const runner: SyncRunner = (_cmd, args) => { + const endpoint = args[2]; + if (endpoint?.includes("/pulls?")) { + return { + status: 0, + stdout: JSON.stringify([ + { number: 10, title: "PR A", state: "open", merged_at: null, user: { login: "anvil" }, requested_reviewers: [] }, + ]), + stderr: "", + } as ReturnType; + } + if (endpoint?.includes("/reviews")) { + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + } + return { status: 0, stdout: "[]", stderr: "" } as ReturnType; + }; + + pollOnce(config, state, runner, sender); + + await new Promise((r) => setTimeout(r, 200)); + + expect(errors.some((e) => e.includes("timed out after 100ms"))).toBe(true); + expect(slowResolved).toBe(false); + expect(mailLog).toContain("kern"); + } finally { + console.error = originalError; + setSendTimeoutMs(5_000); + } + }); +});