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

FN-6093: fix ntfy workflow notification delivery

Browse Source 
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
This commit is contained in:
gsxdsm
2026-06-09 09:52:38 -07:00
parent 313751962b
commit 07a5365115
6 changed files with 208 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-6093-notification-workflow-fix.md Normal file
View File

@@ -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.

9
packages/engine/src/__tests__/merger-ai.test.ts
View File

@@ -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);
});

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

@@ -87,6 +87,14 @@ function createRoomMessage(overrides: Partial<ChatRoomMessage> = {}): 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 () => {

18
packages/engine/src/merger-ai.ts
View File

@@ -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<TaskStore>).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<MergeResult> {
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);

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

@@ -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}`);
});
}
}

2
packages/engine/src/notifier.ts
View File

@@ -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 {