fix(notifications): make pipeline.flush() drain in-flight dispatches

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<void>(r => { resolveDispatch = r })
+    const slowChannel: NotificationChannel = {
+      name: "slow",
+      enabled: () => true,
+      send: async (): Promise<NotificationResult> => {
+        await new Promise<void>(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<boolean>(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<void>(r => { resolveDispatch = r })
+    const slowChannel: NotificationChannel = {
+      name: "slow-idle",
+      enabled: () => true,
+      send: async (): Promise<NotificationResult> => {
+        await new Promise<void>(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<typeof svc.notifySessionIdle>[0]
+
+    await svc.notifySessionIdle(fakeClient, "s-idle")
+    await svc.flush()
+
+    const settled = await Promise.race([
+      dispatchSettled.then(() => true),
+      new Promise<boolean>(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<NotificationResult> => 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 ───────────────────────────

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<Promise<unknown>>()
+
+  function track<T>(p: Promise<T>): 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<string, unknown>)?.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<string, unknown>)?.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<void> {
-    // 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<typeof setTimeout> | undefined
+
+    const timeout = new Promise<void>((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 }