From 07a536511569b55defe0d5aff091908351b340a5 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 09:52:38 -0700 Subject: [PATCH] FN-6093: fix ntfy workflow notification delivery Ensure workflow and merge-triggered ntfy notifications fire reliably. - preserve merge-backed task metadata before moving tasks to done so merged notifications retain context - treat an empty ntfy event allowlist as the documented default event set - allow failed or no-provider notification attempts to clear dedupe state and retry after settings refresh - add regression coverage for workflow notify dispatch, stale-settings refresh, and merged-task ntfy delivery Files changed: .changeset/fn-6093-notification-workflow-fix.md | 5 + packages/engine/src/__tests__/merger-ai.test.ts | 9 +- packages/engine/src/__tests__/notification-service.test.ts | 165 +++++++++++++++++++++ packages/engine/src/merger-ai.ts | 18 ++- packages/engine/src/notification/notification-service.ts | 14 +- packages/engine/src/notifier.ts | 2 +- 6 files changed, 208 insertions(+), 5 deletions(-) Fusion-Task-Id: FN-6093 Fusion-Task-Lineage: fc42cb94-f542-4a34-a325-ffb55dea02f4 --- .../fn-6093-notification-workflow-fix.md | 5 + .../engine/src/__tests__/merger-ai.test.ts | 9 +- .../__tests__/notification-service.test.ts | 165 ++++++++++++++++++ packages/engine/src/merger-ai.ts | 18 +- .../src/notification/notification-service.ts | 14 +- packages/engine/src/notifier.ts | 2 +- 6 files changed, 208 insertions(+), 5 deletions(-) create mode 100644 .changeset/fn-6093-notification-workflow-fix.md diff --git a/.changeset/fn-6093-notification-workflow-fix.md b/.changeset/fn-6093-notification-workflow-fix.md new file mode 100644 index 0000000000..a74de44f40 --- /dev/null +++ b/.changeset/fn-6093-notification-workflow-fix.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Fix workflow/AI merge ntfy notification delivery by preserving merge-backed task metadata, treating an empty ntfy event allowlist as the documented default events, and allowing failed/no-provider notification attempts to retry after settings refresh. diff --git a/packages/engine/src/__tests__/merger-ai.test.ts b/packages/engine/src/__tests__/merger-ai.test.ts index 2c0704238d..b9cdd7f77b 100644 --- a/packages/engine/src/__tests__/merger-ai.test.ts +++ b/packages/engine/src/__tests__/merger-ai.test.ts @@ -193,7 +193,14 @@ describe("runAiMerge", () => { const landedMsg = git(dir, "log -1 --pretty=%B main"); expect(landedMsg).toContain("Fusion-Task-Id: FN-1"); expect(git(dir, "log -1 --pretty=%s main")).toMatch(/^FN-1: /); - // Task moved to done + event emitted. + // Task marked merge-backed before moving to done, then event emitted. + expect(store.updateTask).toHaveBeenCalledWith( + "FN-1", + expect.objectContaining({ + status: null, + mergeDetails: expect.objectContaining({ mergeConfirmed: true }), + }), + ); expect(store.moveTask).toHaveBeenCalledWith("FN-1", "done"); expect(emitted.some((e) => e.event === "task:merged")).toBe(true); }); diff --git a/packages/engine/src/__tests__/notification-service.test.ts b/packages/engine/src/__tests__/notification-service.test.ts index f26d5663ac..c325fc1312 100644 --- a/packages/engine/src/__tests__/notification-service.test.ts +++ b/packages/engine/src/__tests__/notification-service.test.ts @@ -87,6 +87,14 @@ function createRoomMessage(overrides: Partial = {}): ChatRoomMe }; } +function mockNtfyFetch() { + return vi.spyOn(globalThis, "fetch").mockResolvedValue({ + ok: true, + status: 200, + statusText: "OK", + } as Response); +} + describe("NotificationService", () => { it("dispatches in-review event to registered provider", async () => { const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" }); @@ -262,6 +270,39 @@ describe("NotificationService", () => { initSpy.mockRestore(); }); + it("dispatches workflow notify node events through the ntfy provider when enabled", async () => { + const fetchMock = mockNtfyFetch(); + const sendSpy = vi.spyOn(NtfyNotificationProvider.prototype, "sendNotification"); + const store = createStore({ + ntfyEnabled: true, + ntfyTopic: "workflow-topic", + ntfyEvents: ["workflow-notify"] as any, + }); + const service = new NotificationService(store as any, { ntfyBaseUrl: "https://ntfy.example" }); + await service.start(); + + await service.dispatch("workflow-notify", { + taskId: "FN-306", + taskTitle: "Workflow task", + event: "workflow-notify", + metadata: { title: "Workflow ping", message: "Workflow node emitted a notification" }, + }); + + await vi.waitFor(() => { + expect(sendSpy).toHaveBeenCalledWith("workflow-notify", expect.objectContaining({ taskId: "FN-306" })); + expect(fetchMock).toHaveBeenCalledWith( + "https://ntfy.example/workflow-topic", + expect.objectContaining({ + method: "POST", + body: "Workflow node emitted a notification", + headers: expect.objectContaining({ Title: "Workflow ping" }), + }), + ); + }); + sendSpy.mockRestore(); + fetchMock.mockRestore(); + }); + it("reconfigures the ntfy provider when the access token changes without logging the token", async () => { const store = createStore({ ntfyEnabled: true, @@ -750,6 +791,35 @@ describe("NotificationService", () => { expect(sendNotification).toHaveBeenCalledWith("merged", expect.objectContaining({ taskId: "FN-303", event: "merged" })); }); + it("dispatches the full merge-backed done plus task:merged sequence through the ntfy provider once", async () => { + const fetchMock = mockNtfyFetch(); + const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic", ntfyEvents: [] as any }); + const service = new NotificationService(store as any, { ntfyBaseUrl: "https://ntfy.example" }); + await service.start(); + + const mergedTask = task({ id: "FN-305", column: "done", mergeDetails: { mergeConfirmed: true } as any }); + store.emit("task:moved", { task: mergedTask, from: "in-review", to: "done" }); + store.emit("task:merged", { + task: mergedTask, + branch: "fusion/fn-305", + merged: true, + worktreeRemoved: false, + branchDeleted: false, + }); + + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + expect(fetchMock).toHaveBeenCalledWith( + "https://ntfy.example/topic", + expect.objectContaining({ + method: "POST", + body: expect.stringContaining("has been merged to main"), + }), + ); + fetchMock.mockRestore(); + }); + it("honors provider event filtering for task:moved to done terminal notifications", async () => { const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" }); const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" })); @@ -877,6 +947,26 @@ describe("NotificationService", () => { ); }); + it("refreshes stale disabled settings before in-review move notifications through ntfy", async () => { + const fetchMock = mockNtfyFetch(); + const store = createStaleLifecycleStore(); + const service = new NotificationService(store as any, { ntfyBaseUrl: "https://ntfy.example" }); + await service.start(); + + store.emit("task:moved", { task: task({ id: "FN-108" }), from: "in-progress", to: "in-review" }); + + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledWith( + "https://ntfy.example/fusion-test", + expect.objectContaining({ + method: "POST", + body: expect.stringContaining("ready for review"), + }), + ); + }); + fetchMock.mockRestore(); + }); + it("does not notify for non-in-review/non-terminal moves even after stale-settings refresh", async () => { const store = createStaleLifecycleStore(); const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" })); @@ -918,6 +1008,81 @@ describe("NotificationService", () => { ); }); }); + + it("registers and initializes ntfy after a late settings:updated enable before task:merged", async () => { + const fetchMock = mockNtfyFetch(); + const store = createStore({ ntfyEnabled: false, ntfyTopic: "" }); + const service = new NotificationService(store as any, { ntfyBaseUrl: "https://ntfy.example" }); + await service.start(); + + store.emit("settings:updated", { + settings: { + ntfyEnabled: true, + ntfyTopic: "fusion-test", + ntfyEvents: [] as any, + ntfyDashboardHost: "http://localhost:4040", + } as Settings, + previous: { + ntfyEnabled: false, + ntfyTopic: "", + } as Settings, + }); + store.emit("task:merged", { + task: task({ id: "FN-107" }), + branch: "fusion/fn-107", + merged: true, + worktreeRemoved: true, + branchDeleted: true, + }); + + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledWith( + "https://ntfy.example/fusion-test", + expect.objectContaining({ + method: "POST", + body: expect.stringContaining("has been merged to main"), + }), + ); + }); + fetchMock.mockRestore(); + }); + + it("refreshes stale disabled settings for workflow notify dispatch", async () => { + const fetchMock = mockNtfyFetch(); + let getSettingsCallCount = 0; + const store = createStore({ ntfyEnabled: false, ntfyTopic: "" }); + store.getSettings.mockImplementation(async () => { + getSettingsCallCount += 1; + if (getSettingsCallCount === 1) { + return { ntfyEnabled: false, ntfyTopic: "" } as Settings; + } + return { + ntfyEnabled: true, + ntfyTopic: "workflow-refresh", + ntfyEvents: ["workflow-notify"] as any, + } as Settings; + }); + const service = new NotificationService(store as any, { ntfyBaseUrl: "https://ntfy.example" }); + await service.start(); + + await service.dispatch("workflow-notify", { + taskId: "FN-109", + taskTitle: "Workflow refresh", + event: "workflow-notify", + metadata: { title: "Workflow refresh", message: "settings refreshed before workflow notify" }, + }); + + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledWith( + "https://ntfy.example/workflow-refresh", + expect.objectContaining({ + method: "POST", + body: "settings refreshed before workflow notify", + }), + ); + }); + fetchMock.mockRestore(); + }); }); it("suppresses transient failed notification after Auto-recovered status clear", async () => { diff --git a/packages/engine/src/merger-ai.ts b/packages/engine/src/merger-ai.ts index 04c2260b8d..add49dbe2c 100644 --- a/packages/engine/src/merger-ai.ts +++ b/packages/engine/src/merger-ai.ts @@ -957,6 +957,8 @@ async function finalizeMerged( }; modifiedFiles = landedFiles.length > 0 ? landedFiles : undefined; await store.updateTask(taskId, { mergeDetails, modifiedFiles }); + task.mergeDetails = mergeDetails; + task.modifiedFiles = modifiedFiles; if (task.lineageId && typeof (store as Partial).upsertTaskCommitAssociation === "function") { await store.upsertTaskCommitAssociation({ taskLineageId: task.lineageId, @@ -1003,7 +1005,21 @@ async function finalizeMerged( /** Move the task to done and emit, mirroring the legacy completeTask. */ async function finalizeTask(store: TaskStore, taskId: string, result: MergeResult): Promise { - await store.updateTask(taskId, { status: null }).catch(() => undefined); + const mergedAt = new Date().toISOString(); + const mergeDetails: MergeDetails = { + ...result.task.mergeDetails, + ...(result.commitSha ? { commitSha: result.commitSha } : {}), + ...(result.rebaseBaseSha ? { rebaseBaseSha: result.rebaseBaseSha } : {}), + ...(result.landedFiles ? { landedFiles: result.landedFiles } : {}), + ...(typeof result.filesChanged === "number" ? { filesChanged: result.filesChanged } : {}), + ...(typeof result.insertions === "number" ? { insertions: result.insertions } : {}), + ...(typeof result.deletions === "number" ? { deletions: result.deletions } : {}), + ...(result.mergeCommitMessage ? { mergeCommitMessage: result.mergeCommitMessage } : {}), + mergedAt, + mergeConfirmed: result.mergeConfirmed === true, + ...(result.noOp ? { noOpMerge: true, noOpReason: result.reason } : {}), + }; + await store.updateTask(taskId, { status: null, mergeDetails }).catch(() => undefined); const task = await store.moveTask(taskId, "done"); result.task = task; store.emit("task:merged", result); diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index 4ba711beff..99770ff3f6 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/packages/engine/src/notification/notification-service.ts @@ -715,8 +715,18 @@ export class NotificationService { this.notifiedEvents.add(key); schedulerLog.log(`NotificationService.maybeNotify dispatching key=${key}`); - this.dispatcher.dispatch(eventType, payload).catch(() => { - // best effort dispatch + this.dispatcher.dispatch(eventType, payload).then((results) => { + if (results.some((result) => result.success)) { + return; + } + this.notifiedEvents.delete(key); + schedulerLog.log( + `NotificationService.maybeNotify no successful providers for key=${key} results=${JSON.stringify(results)}`, + ); + }).catch((error) => { + this.notifiedEvents.delete(key); + const message = error instanceof Error ? error.message : String(error); + schedulerLog.log(`NotificationService.maybeNotify dispatch failed key=${key} error=${message}`); }); } } diff --git a/packages/engine/src/notifier.ts b/packages/engine/src/notifier.ts index 3ecb8c6ffd..0827380475 100644 --- a/packages/engine/src/notifier.ts +++ b/packages/engine/src/notifier.ts @@ -157,7 +157,7 @@ function ntfyPriorityToInt(priority: NtfyNotificationPriority): number { } export function resolveNtfyEvents(events?: NtfyNotificationEvent[]): NtfyNotificationEvent[] { - return events ? [...events] : [...DEFAULT_NTFY_EVENTS]; + return events && events.length > 0 ? [...events] : [...DEFAULT_NTFY_EVENTS]; } export function isNtfyEventEnabled(events: NtfyNotificationEvent[] | undefined, event: NtfyNotificationEvent): boolean {