root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-4307): complete Step 1 — defer failed notifications

Browse Source 
Fusion-Task-Id: FN-4307
Fusion-Task-Lineage: 00140a5a-af93-4905-8570-32d7ecb8a874
This commit is contained in:
Fusion
2026-05-13 15:28:50 -07:00
committed by gsxdsm
parent 6b9979112e
commit 2e7e5e091c
3 changed files with 81 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/engine/src/__tests__/notification-service.test.ts
View File

@@ -117,13 +117,13 @@ describe("NotificationService", () => {
sendNotification,
};
const service = new NotificationService(store as any);
const service = new NotificationService(store as any, { failedNotificationGraceMs: 0 });
service.registerProvider(provider);
await service.start();
store.emit("task:moved", { task: task(), from: "todo", to: "in-review" });
store.emit("task:moved", { task: task(), from: "todo", to: "in-review" });
store.emit("task:updated", task({ status: "failed" }));
store.emit("task:updated", task({ status: "awaiting-approval" }));
await Promise.resolve();
expect(sendNotification).toHaveBeenCalledTimes(2);

130
packages/engine/src/notification/__tests__/notification-service.test.ts
View File

@@ -15,8 +15,6 @@ function createStore(settings: Partial<Settings> = {}) {
let currentSettings: Settings = {
ntfyEnabled: true,
ntfyTopic: "topic",
failureNotificationDelayMs: 30000,
failureNotificationMode: "sticky-only",
...settings,
} as Settings;
@@ -44,9 +42,6 @@ function createStore(settings: Partial<Settings> = {}) {
setSettings(next: Partial<Settings>) {
currentSettings = { ...currentSettings, ...next } as Settings;
},
settings() {
return currentSettings;
},
};
}
@@ -85,116 +80,61 @@ describe("NotificationService deferred failure notifications", () => {
isEventSupported: () => true,
sendNotification,
};
const service = new NotificationService(store as any);
const service = new NotificationService(store as any, { failedNotificationGraceMs: 100 });
service.registerProvider(provider);
await service.start();
return { store, service, sendNotification };
}
it("Persistent failure dispatches once after delay", async () => {
it("Failure that persists past grace dispatches exactly once", 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);
await vi.advanceTimersByTimeAsync(100);
expect(sendNotification).toHaveBeenCalledTimes(1);
expect(sendNotification).toHaveBeenCalledWith(
"failed",
expect.objectContaining({ taskId: "FN-1" }),
);
expect(sendNotification).toHaveBeenCalledWith("failed", expect.objectContaining({ taskId: "FN-1" }));
await service.stop();
});
it("Self-recovery suppresses notification (status cleared)", async () => {
it("Transient failure with Auto-recovered status clear is suppressed", 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.setTask(task({ id: "FN-1", status: "in-review", log: [{ timestamp: new Date().toISOString(), action: "Auto-recovered: merge deadlock resolved" }] }));
store.emit("task:updated", task({ id: "FN-1", status: "in-review" }));
await vi.advanceTimersByTimeAsync(30000);
await vi.advanceTimersByTimeAsync(100);
expect(sendNotification).not.toHaveBeenCalledWith("failed", expect.anything());
expect(service.getMetrics().failureNotificationSuppressedCount).toBe(1);
expect(schedulerLog.log).toHaveBeenCalledWith(expect.stringContaining("suppressed transient failed"));
await service.stop();
});
it("Recovery via task:moved to done suppresses failed notification", async () => {
const { store, service, sendNotification } = await setup();
store.setTask(task({ id: "FN-1", status: "failed", column: "in-review" }));
store.emit("task:updated", task({ id: "FN-1", status: "failed", column: "in-review" }));
store.setTask(task({ id: "FN-1", status: null, column: "done" }));
store.emit("task:moved", { task: task({ id: "FN-1", status: null, column: "done" }), from: "in-review", to: "done" });
await vi.advanceTimersByTimeAsync(100);
expect(sendNotification).not.toHaveBeenCalledWith("failed", expect.anything());
expect(service.getMetrics().failureNotificationSuppressedCount).toBe(1);
await service.stop();
});
it("stop clears pending timers without firing", 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 service.stop();
await vi.advanceTimersByTimeAsync(100);
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();
});
});

81
packages/engine/src/notification/notification-service.ts
View File

@@ -26,6 +26,8 @@ export interface NotificationServiceOptions {
chatStore?: NotificationChatStore;
/** Resolve human-readable name for an agent ID used in message notifications */
agentNameResolver?: (agentId: string) => Promise<string | null> | string | null;
/** Test hook to override failed-notification grace period (default 60_000ms). */
failedNotificationGraceMs?: number;
}
interface NotificationServiceStore {
@@ -55,17 +57,17 @@ 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 pendingFailureNotifications = new Map<string, { timer: NodeJS.Timeout; payload: NotificationPayload }>();
private readonly pendingFailureStartTimes = new Map<string, number>();
private readonly failedNotificationGraceMs: number;
private failureNotificationSuppressedCount = 0;
private failureNotificationDelayMs = 30000;
private failureNotificationMode: "sticky-only" | "all" = "sticky-only";
constructor(
private readonly store: NotificationServiceStore,
private readonly options: NotificationServiceOptions = {},
) {
this.chatStore = options.chatStore;
this.failedNotificationGraceMs = options.failedNotificationGraceMs ?? 60_000;
}
attachChatStore(chatStore: NotificationChatStore): void {
@@ -89,7 +91,6 @@ export class NotificationService {
const settings = await this.store.getSettings();
this.setNotificationsEnabledFromSettings(settings);
this.refreshFailureNotificationSettings(settings);
await this.syncNtfyProvider(settings);
await this.syncWebhookProvider(settings);
@@ -121,8 +122,8 @@ export class NotificationService {
this.detachChatStoreListener(this.chatStore);
}
for (const timeout of this.pendingFailureNotifications.values()) {
clearTimeout(timeout);
for (const pending of this.pendingFailureNotifications.values()) {
clearTimeout(pending.timer);
}
this.pendingFailureNotifications.clear();
this.pendingFailureStartTimes.clear();
@@ -134,9 +135,7 @@ 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}`);
}
void this.maybeSuppressTransientFailedNotification(data.task, `moved to ${data.to}`);
if (!this.notificationsEnabled || data.to !== "in-review") {
return;
@@ -147,20 +146,14 @@ export class NotificationService {
};
private handleTaskUpdated = (task: Task): void => {
if (task.status !== "failed") {
this.cancelPendingFailureNotification(task.id, `status=${task.status ?? "undefined"}`);
}
void this.maybeSuppressTransientFailedNotification(task, `status=${task.status ?? "undefined"}`);
if (!this.notificationsEnabled) {
return;
}
if (task.status === "failed") {
if (this.failureNotificationMode === "all" || this.failureNotificationDelayMs === 0) {
this.maybeNotify(task.id, "failed", this.createTaskPayload(task, "failed"));
} else {
this.scheduleFailureNotification(task);
}
this.scheduleFailureNotification(task);
}
if (task.status === "awaiting-approval") {
@@ -195,7 +188,6 @@ 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 ||
@@ -453,7 +445,6 @@ 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)}`);
@@ -466,43 +457,59 @@ 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(() => {
const payload = this.createTaskPayload(task, "failed");
const timer = setTimeout(() => {
void this.fireDeferredFailureNotification(task.id);
}, this.failureNotificationDelayMs);
timeout.unref?.();
this.pendingFailureNotifications.set(task.id, timeout);
}, this.failedNotificationGraceMs);
timer.unref?.();
this.pendingFailureNotifications.set(task.id, { timer, payload });
}
private cancelPendingFailureNotification(taskId: string, reason: string): void {
const timeout = this.pendingFailureNotifications.get(taskId);
if (!timeout) {
private async maybeSuppressTransientFailedNotification(task: Task, reason: string): Promise<void> {
if (!this.pendingFailureNotifications.has(task.id)) {
return;
}
clearTimeout(timeout);
const currentTask = (await this.store.getTask?.(task.id)) ?? task;
const hasAutoRecoveredLog = currentTask.log.some((entry) => /^Auto-recovered:/.test(entry.action));
const movedToDone = currentTask.column === "done";
const mergeConfirmed = currentTask.mergeDetails?.mergeConfirmed === true;
const recoveredStatus = currentTask.status !== "failed" && hasAutoRecoveredLog;
if (!movedToDone && !mergeConfirmed && !recoveredStatus) {
return;
}
this.cancelPendingFailureNotification(task.id, reason);
}
private cancelPendingFailureNotification(taskId: string, reason: string): void {
const pending = this.pendingFailureNotifications.get(taskId);
if (!pending) {
return;
}
clearTimeout(pending.timer);
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})`);
schedulerLog.log(`NotificationService.maybeNotify suppressed transient failed key=${taskId}:failed (${reason}, ${elapsedMs}ms)`);
}
private async fireDeferredFailureNotification(taskId: string): Promise<void> {
const pending = this.pendingFailureNotifications.get(taskId);
if (!pending) {
return;
}
this.pendingFailureNotifications.delete(taskId);
this.pendingFailureStartTimes.delete(taskId);
@@ -527,7 +534,7 @@ export class NotificationService {
eventType = "failed:auto-paused" as NotificationEvent;
}
this.maybeNotify(task.id, eventType, this.createTaskPayload(task, eventType));
this.maybeNotify(task.id, eventType, eventType === "failed" ? pending.payload : this.createTaskPayload(task, eventType));
}
getMetrics(): { failureNotificationSuppressedCount: number } {