feat(FN-4332): complete Step 2 — defer failed notifications

Fusion-Task-Id: FN-4332 Fusion-Task-Lineage: c920ac71-906b-458d-b496-1b60540b58f3
2026-05-13 12:46:05 -07:00
parent f7293abaf1
commit bbd810e28a
3 changed files with 305 additions and 2 deletions
--- a/packages/engine/src/tests/notification-service.test.ts
+++ b/packages/engine/src/tests/notification-service.test.ts
@@ -105,7 +105,11 @@ describe("NotificationService", () => {
  });

  it("deduplicates same task+event but not different event types", async () => {
-    const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
+    const store = createStore({
+      ntfyEnabled: true,
+      ntfyTopic: "topic",
+      failureNotificationMode: "all",
+    });
    const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
    const provider: NotificationProvider = {
      getProviderId: () => "mock",
--- a/packages/engine/src/notification/tests/notification-service.test.ts
+++ b/packages/engine/src/notification/tests/notification-service.test.ts
@@ -0,0 +1,200 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { NotificationProvider, Settings, Task } from "@fusion/core";
+import { NotificationService } from "../notification-service.js";
+import { schedulerLog } from "../../logger.js";
+
+vi.mock("../../logger.js", () => ({
+  schedulerLog: { log: vi.fn(), warn: vi.fn(), error: vi.fn() },
+}));
+
+type Listener = (...args: any[]) => void | Promise<void>;
+
+function createStore(settings: Partial<Settings> = {}) {
+  const listeners = new Map<string, Set<Listener>>();
+  const tasks = new Map<string, Task>();
+  let currentSettings: Settings = {
+    ntfyEnabled: true,
+    ntfyTopic: "topic",
+    failureNotificationDelayMs: 30000,
+    failureNotificationMode: "sticky-only",
+    ...settings,
+  } as Settings;
+
+  const getBucket = (event: string) => listeners.get(event) ?? new Set<Listener>();
+
+  return {
+    on(event: string, listener: Listener) {
+      const bucket = getBucket(event);
+      bucket.add(listener);
+      listeners.set(event, bucket);
+    },
+    off(event: string, listener: Listener) {
+      getBucket(event).delete(listener);
+    },
+    emit(event: string, payload: unknown) {
+      for (const listener of getBucket(event)) {
+        void listener(payload);
+      }
+    },
+    getSettings: vi.fn(async () => currentSettings),
+    getTask: vi.fn(async (id: string) => tasks.get(id)),
+    setTask(task: Task) {
+      tasks.set(task.id, task);
+    },
+    setSettings(next: Partial<Settings>) {
+      currentSettings = { ...currentSettings, ...next } as Settings;
+    },
+    settings() {
+      return currentSettings;
+    },
+  };
+}
+
+function task(overrides: Partial<Task> = {}): Task {
+  return {
+    id: "FN-1",
+    title: "Task title",
+    description: "Task desc",
+    status: "todo",
+    column: "todo",
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    dependencies: [],
+    steps: [],
+    currentStep: 0,
+    log: [],
+    ...overrides,
+  } as Task;
+}
+
+describe("NotificationService deferred failure notifications", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.clearAllMocks();
+  });
+
+  async function setup(settings: Partial<Settings> = {}) {
+    const store = createStore(settings);
+    const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
+    const provider: NotificationProvider = {
+      getProviderId: () => "mock",
+      isEventSupported: () => true,
+      sendNotification,
+    };
+    const service = new NotificationService(store as any);
+    service.registerProvider(provider);
+    await service.start();
+    return { store, service, sendNotification };
+  }
+
+  it("Persistent failure dispatches once after delay", async () => {
+    const { store, service, sendNotification } = await setup();
+    store.setTask(task({ id: "FN-1", status: "failed" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+
+    await vi.advanceTimersByTimeAsync(30000);
+
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    expect(sendNotification).toHaveBeenCalledWith(
+      "failed",
+      expect.objectContaining({ taskId: "FN-1" }),
+    );
+    await service.stop();
+  });
+
+  it("Self-recovery suppresses notification (status cleared)", async () => {
+    const { store, service, sendNotification } = await setup();
+    store.setTask(task({ id: "FN-1", status: "failed" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+    await vi.advanceTimersByTimeAsync(5000);
+
+    store.setTask(task({ id: "FN-1", status: "in-review" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "in-review" }));
+    await vi.advanceTimersByTimeAsync(30000);
+
+    expect(sendNotification).not.toHaveBeenCalled();
+    expect(service.getMetrics().failureNotificationSuppressedCount).toBe(1);
+    expect(schedulerLog.log).toHaveBeenCalledWith(expect.stringContaining("suppressed notification"));
+    await service.stop();
+  });
+
+  it("Self-recovery via column move suppresses", async () => {
+    const { store, service, sendNotification } = await setup();
+    store.setTask(task({ id: "FN-1", status: "failed" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+    store.emit("task:moved", { task: task({ id: "FN-1" }), from: "in-progress", to: "in-review" });
+
+    await vi.advanceTimersByTimeAsync(30000);
+
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    expect(sendNotification).toHaveBeenCalledWith(
+      "in-review",
+      expect.objectContaining({ event: "in-review" }),
+    );
+    expect(service.getMetrics().failureNotificationSuppressedCount).toBe(1);
+    await service.stop();
+  });
+
+  it('failureNotificationMode: "all" dispatches immediately', async () => {
+    const { store, service, sendNotification } = await setup({ failureNotificationMode: "all" });
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    await service.stop();
+  });
+
+  it("failureNotificationDelayMs: 0 dispatches immediately", async () => {
+    const { store, service, sendNotification } = await setup({ failureNotificationDelayMs: 0 });
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    await service.stop();
+  });
+
+  it("Coalescing: two rapid failed events keep one pending timer and one dispatch", async () => {
+    const { store, service, sendNotification } = await setup();
+    store.setTask(task({ id: "FN-1", status: "failed" }));
+
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+
+    expect(service.getPendingFailureCount()).toBe(1);
+
+    await vi.advanceTimersByTimeAsync(30000);
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    await service.stop();
+  });
+
+  it("Fresh re-read at fire time", async () => {
+    const { store, service, sendNotification } = await setup();
+    store.setTask(task({ id: "FN-1", status: "failed", title: "Old" }));
+    store.emit("task:updated", task({ id: "FN-1", status: "failed", title: "Old" }));
+
+    store.setTask(task({ id: "FN-1", status: "failed", title: "New Title" }));
+    await vi.advanceTimersByTimeAsync(30000);
+
+    expect(sendNotification).toHaveBeenCalledWith(
+      "failed",
+      expect.objectContaining({ taskTitle: "New Title" }),
+    );
+    await service.stop();
+  });
+
+  it("Setting change refreshes cached knobs", async () => {
+    const { store, service, sendNotification } = await setup();
+    const nextSettings = { ...store.settings(), failureNotificationMode: "all" as const };
+
+    store.emit("settings:updated", { settings: nextSettings, previous: store.settings() });
+    store.setSettings({ failureNotificationMode: "all" });
+    await Promise.resolve();
+
+    store.emit("task:updated", task({ id: "FN-1", status: "failed" }));
+
+    expect(sendNotification).toHaveBeenCalledTimes(1);
+    await service.stop();
+  });
+});
--- a/packages/engine/src/notification/notification-service.ts
+++ b/packages/engine/src/notification/notification-service.ts
@@ -30,6 +30,7 @@ export interface NotificationServiceOptions {

 interface NotificationServiceStore {
  getSettings(): Promise<Settings> | Settings;
+  getTask?(id: string): Promise<Task | undefined> | Task | undefined;
  on(event: string, listener: (...args: any[]) => void): void;
  off(event: string, listener: (...args: any[]) => void): void;
 }
@@ -54,6 +55,11 @@ export class NotificationService {
  private ntfyProvider?: NtfyNotificationProvider;
  private webhookProvider?: WebhookNotificationProvider;
  private refreshInFlight: Promise<void> | null = null;
+  private readonly pendingFailureNotifications = new Map<string, NodeJS.Timeout>();
+  private readonly pendingFailureStartTimes = new Map<string, number>();
+  private failureNotificationSuppressedCount = 0;
+  private failureNotificationDelayMs = 30000;
+  private failureNotificationMode: "sticky-only" | "all" = "sticky-only";

  constructor(
    private readonly store: NotificationServiceStore,
@@ -83,6 +89,7 @@ export class NotificationService {

    const settings = await this.store.getSettings();
    this.setNotificationsEnabledFromSettings(settings);
+    this.refreshFailureNotificationSettings(settings);
    await this.syncNtfyProvider(settings);
    await this.syncWebhookProvider(settings);

@@ -114,6 +121,12 @@ export class NotificationService {
      this.detachChatStoreListener(this.chatStore);
    }

+    for (const timeout of this.pendingFailureNotifications.values()) {
+      clearTimeout(timeout);
+    }
+    this.pendingFailureNotifications.clear();
+    this.pendingFailureStartTimes.clear();
+
    await this.dispatcher.shutdownAll();
    this.started = false;

@@ -121,6 +134,10 @@ export class NotificationService {
  }

  private handleTaskMoved = (data: { task: Task; from: Column; to: Column }): void => {
+    if (["in-review", "done", "archived"].includes(data.to)) {
+      this.cancelPendingFailureNotification(data.task.id, `moved to ${data.to}`);
+    }
+
    if (!this.notificationsEnabled || data.to !== "in-review") {
      return;
    }
@@ -130,12 +147,20 @@ export class NotificationService {
  };

  private handleTaskUpdated = (task: Task): void => {
+    if (task.status !== "failed") {
+      this.cancelPendingFailureNotification(task.id, `status=${task.status ?? "undefined"}`);
+    }
+
    if (!this.notificationsEnabled) {
      return;
    }

    if (task.status === "failed") {
-      this.maybeNotify(task.id, "failed", this.createTaskPayload(task, "failed"));
+      if (this.failureNotificationMode === "all" || this.failureNotificationDelayMs === 0) {
+        this.maybeNotify(task.id, "failed", this.createTaskPayload(task, "failed"));
+      } else {
+        this.scheduleFailureNotification(task);
+      }
    }

    if (task.status === "awaiting-approval") {
@@ -170,6 +195,7 @@ export class NotificationService {
  private handleSettingsUpdated = async (data: { settings: Settings; previous: Settings }): Promise<void> => {
    const { settings, previous } = data;
    this.setNotificationsEnabledFromSettings(settings);
+    this.refreshFailureNotificationSettings(settings);

    if (
      settings.ntfyEnabled !== previous.ntfyEnabled ||
@@ -427,6 +453,7 @@ export class NotificationService {
    this.refreshInFlight = (async () => {
      const settings = await this.store.getSettings();
      this.setNotificationsEnabledFromSettings(settings);
+      this.refreshFailureNotificationSettings(settings);
      await this.syncNtfyProvider(settings);
      await this.syncWebhookProvider(settings);
      schedulerLog.log(`NotificationService refreshed notification state reason=${reason} enabled=${String(this.notificationsEnabled)}`);
@@ -439,6 +466,78 @@ export class NotificationService {
    }
  }

+  private refreshFailureNotificationSettings(settings: Settings): void {
+    this.failureNotificationDelayMs =
+      typeof settings.failureNotificationDelayMs === "number" && settings.failureNotificationDelayMs >= 0
+        ? settings.failureNotificationDelayMs
+        : 30000;
+    this.failureNotificationMode = settings.failureNotificationMode ?? "sticky-only";
+  }
+
+  private scheduleFailureNotification(task: Task): void {
+    if (this.pendingFailureNotifications.has(task.id)) {
+      return;
+    }
+
+    this.pendingFailureStartTimes.set(task.id, Date.now());
+    const timeout = setTimeout(() => {
+      void this.fireDeferredFailureNotification(task.id);
+    }, this.failureNotificationDelayMs);
+    timeout.unref?.();
+    this.pendingFailureNotifications.set(task.id, timeout);
+  }
+
+  private cancelPendingFailureNotification(taskId: string, reason: string): void {
+    const timeout = this.pendingFailureNotifications.get(taskId);
+    if (!timeout) {
+      return;
+    }
+
+    clearTimeout(timeout);
+    this.pendingFailureNotifications.delete(taskId);
+    const startedAt = this.pendingFailureStartTimes.get(taskId);
+    this.pendingFailureStartTimes.delete(taskId);
+    const elapsedMs = typeof startedAt === "number" ? Math.max(0, Date.now() - startedAt) : 0;
+    this.failureNotificationSuppressedCount += 1;
+    schedulerLog.log(`[notify] ${taskId} failed-state cleared within ${elapsedMs}ms — suppressed notification (${reason})`);
+  }
+
+  private async fireDeferredFailureNotification(taskId: string): Promise<void> {
+    this.pendingFailureNotifications.delete(taskId);
+    this.pendingFailureStartTimes.delete(taskId);
+
+    const task = await this.store.getTask?.(taskId);
+    if (!task) {
+      return;
+    }
+
+    if (task.status !== "failed") {
+      this.failureNotificationSuppressedCount += 1;
+      schedulerLog.log(`[notify] ${taskId} no longer failed at dispatch time — suppressed notification`);
+      return;
+    }
+
+    const pausedTask = task as Task & { pausedReason?: string };
+    let eventType: NotificationEvent = "failed";
+    if (
+      pausedTask.paused === true &&
+      pausedTask.pausedReason === "dispatch-storm" &&
+      DEFAULT_NTFY_EVENTS.includes("failed:auto-paused" as (typeof DEFAULT_NTFY_EVENTS)[number])
+    ) {
+      eventType = "failed:auto-paused" as NotificationEvent;
+    }
+
+    this.maybeNotify(task.id, eventType, this.createTaskPayload(task, eventType));
+  }
+
+  getMetrics(): { failureNotificationSuppressedCount: number } {
+    return { failureNotificationSuppressedCount: this.failureNotificationSuppressedCount };
+  }
+
+  getPendingFailureCount(): number {
+    return this.pendingFailureNotifications.size;
+  }
+
  private createTaskPayload(task: Task, event: NotificationEvent): NotificationPayload {
    return {
      taskId: task.id,