diff --git a/src/notifications/pipeline.test.ts b/src/notifications/pipeline.test.ts index 9156fa8..7d8f387 100644 --- a/src/notifications/pipeline.test.ts +++ b/src/notifications/pipeline.test.ts @@ -1,6 +1,7 @@ // Tests for createNotificationService (pipeline fan-out) // Covers: notifyPermissionPending return value, SSE-reachability gating, -// fire-and-forget channel dispatch, flush no-op, emit/emitPilot wiring. +// fire-and-forget channel dispatch, in-flight tracking + flush drain/timeout, +// emit/emitPilot wiring. import { describe, expect, test } from "bun:test" import { createNotificationService } from "./pipeline" import type { NotificationServiceDeps } from "./pipeline" @@ -331,12 +332,99 @@ describe("emit and emitPilot", () => { // ─── flush ──────────────────────────────────────────────────────────────────── describe("flush", () => { - test("flush resolves without error (currently a no-op)", async () => { + test("flush resolves immediately with nothing in flight", async () => { const deps = makeDeps() const svc = createNotificationService(deps) + // Fast path: no tracked promises, should resolve without delay await expect(svc.flush()).resolves.toBeUndefined() }) + + test("flush awaits an in-flight slow channel dispatch from notifyPermissionPending", async () => { + // A slow channel that resolves after a short delay + let resolveDispatch!: () => void + const dispatchSettled = new Promise(r => { resolveDispatch = r }) + const slowChannel: NotificationChannel = { + name: "slow", + enabled: () => true, + send: async (): Promise => { + await new Promise(r => setTimeout(r, 30)) + resolveDispatch() + return { ok: true } + }, + } + const deps = makeDeps({ channels: [slowChannel] }) + const svc = createNotificationService(deps) + + // Fire permission pending — starts the slow dispatch in the background + await svc.notifyPermissionPending("p-slow", "title", "sess", "execute") + // The dispatch is still running (30ms). flush() should wait for it. + const flushDone = svc.flush() + await flushDone + // By the time flush() resolves, the dispatch must have settled + const settled = await Promise.race([ + dispatchSettled.then(() => true), + new Promise(r => setTimeout(() => r(false), 5)), + ]) + expect(settled).toBe(true) + }) + + test("flush awaits an in-flight slow channel dispatch from notifySessionIdle", async () => { + let resolveDispatch!: () => void + const dispatchSettled = new Promise(r => { resolveDispatch = r }) + const slowChannel: NotificationChannel = { + name: "slow-idle", + enabled: () => true, + send: async (): Promise => { + await new Promise(r => setTimeout(r, 30)) + resolveDispatch() + return { ok: true } + }, + } + const deps = makeDeps({ channels: [slowChannel] }) + const svc = createNotificationService(deps) + + const fakeClient = { + session: { get: async () => ({ data: { title: "S" } }) }, + } as unknown as Parameters[0] + + await svc.notifySessionIdle(fakeClient, "s-idle") + await svc.flush() + + const settled = await Promise.race([ + dispatchSettled.then(() => true), + new Promise(r => setTimeout(() => r(false), 5)), + ]) + expect(settled).toBe(true) + }) + + test("flush resolves within bound and audits notifications.flush_timeout when a dispatch hangs", async () => { + // A channel whose send() never resolves — simulates a wedged channel + const hangingChannel: NotificationChannel = { + name: "hanging", + enabled: () => true, + send: (): Promise => new Promise(() => {}), // never resolves + } + const auditEntries: AuditEntry[] = [] + const audit = makeAudit(auditEntries) + const deps = makeDeps({ channels: [hangingChannel], audit }) + // Use a very short flush timeout so the test doesn't take 5 seconds + const svc = createNotificationService(deps, { flushTimeoutMs: 50 }) + + await svc.notifyPermissionPending("p-hang", "title", "sess", "execute") + + // flush() must resolve within the short bound (well under 200ms) + const start = Date.now() + await svc.flush() + const elapsed = Date.now() - start + expect(elapsed).toBeLessThan(200) + + // The drop must be observable via audit (AGENTS.md §3 "No silent failures") + const timeoutEntry = auditEntries.find(e => e.action === "notifications.flush_timeout") + expect(timeoutEntry).toBeDefined() + expect(typeof timeoutEntry?.details.pending).toBe("number") + expect((timeoutEntry?.details.pending as number)).toBeGreaterThan(0) + }) }) // ─── notifySessionIdle — fan-out to extra channels ─────────────────────────── diff --git a/src/notifications/pipeline.ts b/src/notifications/pipeline.ts index 19421c6..a282353 100644 --- a/src/notifications/pipeline.ts +++ b/src/notifications/pipeline.ts @@ -50,8 +50,40 @@ export interface NotificationServiceDeps { channels?: NotificationChannel[] } -export function createNotificationService(deps: NotificationServiceDeps): NotificationService { +/** + * Maximum ms flush() will wait for in-flight fire-and-forget dispatches. + * 5 s is generous enough for any real channel (HTTP + Telegram RTT) while + * ensuring shutdown can't hang indefinitely on a wedged/slow channel. + * The value is exported so server/index.ts can document it and tests can + * override it via the second parameter of createNotificationService. + */ +export const FLUSH_TIMEOUT_MS = 5_000 + +export interface NotificationServiceOptions { + /** + * Override the flush timeout for testing. + * Defaults to FLUSH_TIMEOUT_MS (5 000 ms). + */ + flushTimeoutMs?: number +} + +export function createNotificationService( + deps: NotificationServiceDeps, + options: NotificationServiceOptions = {}, +): NotificationService { const { eventBus, telegram, audit, push, channels = [] } = deps + const flushTimeoutMs = options.flushTimeoutMs ?? FLUSH_TIMEOUT_MS + + // ─── In-flight tracking ─────────────────────────────────────────────────── + // Keeps a live set of every fire-and-forget dispatch currently outstanding. + // We ONLY track the post-.catch() promise (already-caught, never rejects) so + // flush() can await them without risking an unhandled rejection. + const inFlight = new Set>() + + function track(p: Promise): void { + inFlight.add(p) + p.finally(() => inFlight.delete(p)) + } function emit(event: BusEvent): void { eventBus.emit(event) @@ -74,23 +106,29 @@ export function createNotificationService(deps: NotificationServiceDeps): Notifi const pushReachable = push.isEnabled() && push.count() > 0 const sseReachable = eventBus.hasClients() - telegram - .sendPermissionRequest(permissionID, title, sessionID) - .catch((err) => audit.log("telegram.send_failed", { error: String(err) })) + // Fire-and-forget — keep the existing .catch(audit) intact; + // track the non-rejecting post-.catch() promise so flush() can drain it. + track( + telegram + .sendPermissionRequest(permissionID, title, sessionID) + .catch((err) => audit.log("telegram.send_failed", { error: String(err) })), + ) if (push.isEnabled()) { - push - .broadcast({ - title: "Permission request", - body: title, - data: { - kind: "permission", - id: permissionID, - sessionID, - url: "/", - }, - }) - .catch((err) => audit.log("push.send_failed", { error: String(err) })) + track( + push + .broadcast({ + title: "Permission request", + body: title, + data: { + kind: "permission", + id: permissionID, + sessionID, + url: "/", + }, + }) + .catch((err) => audit.log("push.send_failed", { error: String(err) })), + ) } // Fan-out to additional channels (e.g. future Slack, Discord, webhook channels). @@ -102,8 +140,10 @@ export function createNotificationService(deps: NotificationServiceDeps): Notifi } for (const ch of channels) { if (!ch.enabled()) continue - ch.send(channelEvent).catch((err) => - audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + track( + ch.send(channelEvent).catch((err) => + audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + ), ) } @@ -135,19 +175,23 @@ export function createNotificationService(deps: NotificationServiceDeps): Notifi try { const session = await client.session.get({ path: { id: sessionID } }) title = ((session.data as Record)?.title as string) ?? "Untitled" + // telegram.sendSessionIdle is await-ed inside the try/catch — NOT fire-and-forget; + // leave it as-is. await telegram.sendSessionIdle(sessionID, title) } catch (err) { audit.log("telegram.send_failed", { error: String(err), kind: "session_idle" }) } - // Fan-out to additional channels + // Fan-out to additional channels — fire-and-forget; track each dispatch. const channelEvent: NotificationEvent = { kind: "session.idle", payload: { sessionID, title }, } for (const ch of channels) { if (!ch.enabled()) continue - ch.send(channelEvent).catch((err) => - audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + track( + ch.send(channelEvent).catch((err) => + audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + ), ) } } @@ -161,26 +205,52 @@ export function createNotificationService(deps: NotificationServiceDeps): Notifi try { const session = await client.session.get({ path: { id: sessionID } }) title = ((session.data as Record)?.title as string) ?? "Untitled" + // telegram.sendSessionError is await-ed inside the try/catch — NOT fire-and-forget; + // leave it as-is. await telegram.sendSessionError(sessionID, title, error) } catch (err) { audit.log("telegram.send_failed", { error: String(err), kind: "session_error" }) } - // Fan-out to additional channels + // Fan-out to additional channels — fire-and-forget; track each dispatch. const channelEvent: NotificationEvent = { kind: "session.error", payload: { sessionID, title, error }, } for (const ch of channels) { if (!ch.enabled()) continue - ch.send(channelEvent).catch((err) => - audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + track( + ch.send(channelEvent).catch((err) => + audit.log("channel.send_failed", { channel: ch.name, error: String(err) }), + ), ) } } async function flush(): Promise { - // No in-flight tracking currently — no-op placeholder. - // Called during shutdown to drain any queued notifications before process exit. + // Fast path: nothing to wait for. + if (inFlight.size === 0) return + + // Snapshot the current set so additions after flush() starts don't extend + // the wait indefinitely (new dispatches would need a subsequent flush). + const snapshot = [...inFlight] + + let timer: ReturnType | undefined + + const timeout = new Promise((resolve) => { + timer = setTimeout(() => { + // Drop is observable — never silent (AGENTS.md §3 "No silent failures"). + const stillPending = snapshot.filter(p => inFlight.has(p)).length + audit.log("notifications.flush_timeout", { pending: stillPending }) + resolve() + }, flushTimeoutMs) + }) + + // Race: drain wins if all dispatches settle within the bound; timeout wins + // and audits the drop if any channel is still wedged after flushTimeoutMs. + await Promise.race([Promise.allSettled(snapshot), timeout]) + + // Always clear the timer — no leaked handles whether we drained or timed out. + clearTimeout(timer) } return { emit, emitPilot, notifyPermissionPending, notifySessionIdle, notifySessionError, flush }