FN-6030: fix workflow terminal notifications

Ensure merged workflow tasks still emit the expected terminal notifications.

- emit task:merged when PR-driven workflow merges move tasks to done
- send merged notifications for merge-backed done transitions and suppress duplicates
- add regression coverage, a published changeset, and reconcile overlapping docs/test updates

Files changed:
 .../fn-6030-workflow-terminal-notifications.md     |   5 +
 docs/storage.md                                    |   8 +-
 .../core/src/__tests__/builtin-workflows.test.ts   |   1 -
 .../__tests__/store-pr-merged-transition.test.ts   |  14 ++-
 .../src/__tests__/workflow-ir-resolver.test.ts     |   1 -
 packages/core/src/store.ts                         |  14 ++-
 .../settings/sections/ProjectModelsSection.tsx     |   3 +-
 .../engine/src/__tests__/merger-post-merge.test.ts |   9 +-
 .../src/__tests__/notification-service.test.ts     | 129 ++++++++++++++++++++-
 .../__tests__/workflow-graph-task-runner.test.ts   |  57 +++++++++
 .../src/notification/notification-service.ts       |  36 ++++--
 11 files changed, 257 insertions(+), 20 deletions(-)

Fusion-Task-Id: FN-6030

Fusion-Task-Lineage: a991f765-0986-4541-ab38-ff476cf88d16

.changeset/fn-6030-workflow-terminal-notifications.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Restore terminal task notifications for workflow/PR-backed completions that move tasks to done before emitting the canonical merged lifecycle event.

docs/storage.md

@@ -176,10 +176,10 @@ Important execution nuance:
 - Why defer now:
   - FN-5943 already landed the lower-risk fix for the observed incident: fewer rewrites, bounded merge/optimize maintenance, and threshold-triggered rebuild.
   - FN-6008 rechecked the post-FN-5943 operational evidence against the live project DB and the defer condition still holds:
-    - recent `runAuditEvents` telemetry for `target: "tasks_fts"` shows the live index staying bounded in the **tens of KB**, not MB-scale bloat;
+    - recent `runAuditEvents` telemetry for `target: "tasks_fts"` shows the live index staying bounded in the **tens to low hundreds of KB**, not MB-scale bloat;
-    - the sampled 30-event maintenance window showed **0 rebuild events** (`23` merge, `7` optimize), with `merge`/`optimize` repeatedly keeping the index small (latest samples include `44076 → 43296` bytes and `53261 → 40449` bytes);
+    - sampled maintenance windows showed **0 rebuild events**, with `merge`/`optimize` repeatedly pulling the index back down (for example `141186 → 43990` bytes, `96571 → 40693` bytes, `44076 → 43296` bytes, and `53261 → 40449` bytes);
-    - a direct `tasks_fts_data` size check during the review was only **47884 bytes** for the current project DB;
+    - direct `tasks_fts_data` size checks during review were only about **48–50 KB** for the current project DB (including **47884 bytes** in one sample and about **50 KB** for **36** live tasks in another);
-    - reviewed logs contained one general `database disk image is malformed` crash in an older merge-agent log, but no recurring post-FN-5943 live `tasks_fts` corruption pattern and no repeated FTS rebuild failures.
+    - reviewed logs showed no concrete recurring post-FN-5943 live `tasks_fts` corruption pattern or repeated FTS rebuild failures, though one older merge-agent log did contain a general `database disk image is malformed` crash.
   - The attached-file idea still improves corruption isolation, but it would trade away the current same-file trigger-maintained index for a manual two-file sync architecture with weaker crash atomicity under WAL.
 - Revisit only if post-FN-5943 production evidence shows recurring `fusion.db`-coupled FTS corruption or materially persistent live-index bloat significant enough to justify a contentless/manual-sync redesign. Until then, keep the single-file external-content design and existing maintenance path.
 

packages/core/src/__tests__/builtin-workflows.test.ts

@@ -151,7 +151,6 @@ describe("built-in workflows", () => {
       "builtin:stepwise-coding",
     ]);
   });
-
   it("builtin:coding exposes execute retries after registry lookup and parse round-trip", () => {
     const coding = getBuiltinWorkflow("builtin:coding");
     expect(coding).toBeDefined();

packages/core/src/__tests__/store-pr-merged-transition.test.ts

@@ -1,4 +1,4 @@
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import { TaskStore } from "../store.js";
 import { createTaskStoreTestHarness } from "./store-test-helpers.js";
@@ -16,7 +16,7 @@ describe("TaskStore.applyPrMergedTransition", () => {
     await harness.afterEach();
   });
 
-  it("moves in-review merged tasks to done once", async () => {
+  it("moves in-review merged tasks to done once and emits task:merged", async () => {
     const task = await store.createTask({ description: "merged task" });
     await store.moveTask(task.id, "todo");
     await store.moveTask(task.id, "in-progress");
@@ -30,9 +30,19 @@ describe("TaskStore.applyPrMergedTransition", () => {
       baseBranch: "main",
       commentCount: 0,
     });
+    const mergedListener = vi.fn();
+    store.on("task:merged", mergedListener);
 
     await expect(store.applyPrMergedTransition(task.id)).resolves.toEqual({ moved: true });
+    expect(mergedListener).toHaveBeenCalledTimes(1);
+    expect(mergedListener).toHaveBeenCalledWith(expect.objectContaining({
+      branch: "fusion/fn-1",
+      merged: true,
+      task: expect.objectContaining({ id: task.id, column: "done" }),
+    }));
+
     await expect(store.applyPrMergedTransition(task.id)).resolves.toEqual({ moved: false, skipped: "already-done" });
+    expect(mergedListener).toHaveBeenCalledTimes(1);
   });
 
   it("skips when pr status is not merged", async () => {

packages/core/src/__tests__/workflow-ir-resolver.test.ts

@@ -118,7 +118,6 @@ describe("resolveWorkflowIrById", () => {
     expect(ir).toBe(BUILTIN_CODING_WORKFLOW_IR);
     expect(store.getWorkflowDefinition).not.toHaveBeenCalled();
   });
-
   it("parses a raw-string IR from the definition", async () => {
     const raw = JSON.stringify(CUSTOM_IR);
     const store = makeStore({ defs: { "wf-raw": { ir: raw } } });

packages/core/src/store.ts

@@ -11809,13 +11809,25 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
       return { moved: false, skipped: "wrong-column" };
     }
 
-    await this.moveTask(taskId, "done", {
+    const movedTask = await this.moveTask(taskId, "done", {
       moveSource: "engine",
       preserveProgress: true,
       preserveWorktree: true,
       skipMergeBlocker: true,
     });
 
+    this.emit("task:merged", {
+      task: movedTask,
+      branch: movedTask.branch ?? movedTask.prInfo?.headBranch ?? freshTask.branch ?? freshTask.prInfo?.headBranch ?? "",
+      merged: true,
+      worktreeRemoved: false,
+      branchDeleted: false,
+      mergeConfirmed: movedTask.mergeDetails?.mergeConfirmed ?? freshTask.mergeDetails?.mergeConfirmed,
+      mergedAt: movedTask.mergeDetails?.mergedAt ?? freshTask.mergeDetails?.mergedAt,
+      mergeTargetBranch: movedTask.mergeDetails?.mergeTargetBranch ?? freshTask.mergeDetails?.mergeTargetBranch,
+      mergeTargetSource: movedTask.mergeDetails?.mergeTargetSource ?? freshTask.mergeDetails?.mergeTargetSource,
+    } satisfies MergeResult);
+
     if (ctx?.agentId && ctx?.runId) {
       this.recordRunAuditEvent({
         taskId,

packages/dashboard/app/components/settings/sections/ProjectModelsSection.tsx

@@ -13,7 +13,8 @@
  * Keys, lane labels, and conditional rendering are preserved verbatim from the
  * original inline JSX.
  */
-import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react";
+import type { ReactNode } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import { useTranslation } from "react-i18next";
 import type { ModelPreset, Settings } from "@fusion/core";
 import {

packages/engine/src/__tests__/merger-post-merge.test.ts

@@ -330,8 +330,15 @@ describe("aiMergeTask — post-merge workflow steps", () => {
     expect(postMergeAgentCall?.[0]?.cwd).toMatch(/\.worktrees\/post-merge-FN-050-[a-z0-9]+/);
     expect(postMergeAgentCall?.[0]?.cwd).not.toBe("/tmp/root");
 
-    // Task should still move to done even though post-merge step ran
+    // Task should still move to done and emit the canonical terminal event after post-merge steps run.
     expect(store.moveTask).toHaveBeenCalledWith("FN-050", "done");
+    expect(store.emit).toHaveBeenCalledWith(
+      "task:merged",
+      expect.objectContaining({ merged: true, task: expect.objectContaining({ id: "FN-050" }) }),
+    );
+    expect((store as any).getWorkflowStep.mock.invocationCallOrder[0]).toBeLessThan(
+      (store.emit as ReturnType<typeof vi.fn>).mock.invocationCallOrder[0],
+    );
   });
 
   it("uses assigned agent runtime model for post-merge prompt step when workflow step has no override", async () => {

packages/engine/src/__tests__/notification-service.test.ts

@@ -667,6 +667,107 @@ describe("NotificationService", () => {
     await second.stop();
   });
 
+  describe("terminal merged notifications", () => {
+    it("dispatches task:moved to done for PR-merged tasks to ntfy and webhook providers", async () => {
+      const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
+      const ntfySend = vi.fn(async () => ({ success: true, providerId: "mock-ntfy" }));
+      const webhookSend = vi.fn(async () => ({ success: true, providerId: "mock-webhook" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock-ntfy", isEventSupported: (event) => event === "merged", sendNotification: ntfySend });
+      service.registerProvider({ getProviderId: () => "mock-webhook", isEventSupported: (event) => event === "merged", sendNotification: webhookSend });
+      await service.start();
+
+      store.emit("task:moved", {
+        task: task({ id: "FN-301", column: "done", prInfo: { status: "merged", number: 12 } as any }),
+        from: "in-review",
+        to: "done",
+      });
+
+      await vi.waitFor(() => {
+        expect(ntfySend).toHaveBeenCalledWith("merged", expect.objectContaining({ taskId: "FN-301", event: "merged" }));
+        expect(webhookSend).toHaveBeenCalledWith("merged", expect.objectContaining({ taskId: "FN-301", event: "merged" }));
+      });
+    });
+
+    it.each([
+      ["mergeConfirmed", { mergeConfirmed: true }],
+      ["noOpMerge", { noOpMerge: true }],
+      ["mergedAt", { mergedAt: "2026-06-08T00:00:00.000Z" }],
+    ])("dispatches task:moved to done for merge-backed tasks with %s metadata", async (_name, mergeDetails) => {
+      const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
+      const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
+      await service.start();
+
+      store.emit("task:moved", {
+        task: task({ id: `FN-${Object.keys(mergeDetails).join("")}`, column: "done", mergeDetails: mergeDetails as any }),
+        from: "in-review",
+        to: "done",
+      });
+
+      await vi.waitFor(() => {
+        expect(sendNotification).toHaveBeenCalledWith(
+          "merged",
+          expect.objectContaining({ event: "merged" }),
+        );
+      });
+    });
+
+    it("does not dispatch task:moved to done for non-merge-backed tasks", async () => {
+      const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
+      const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
+      await service.start();
+
+      store.emit("task:moved", { task: task({ id: "FN-302", column: "done" }), from: "in-review", to: "done" });
+      await Promise.resolve();
+
+      expect(sendNotification).not.toHaveBeenCalledWith("merged", expect.anything());
+    });
+
+    it("deduplicates task:moved to done plus task:merged for the same merge-backed task", async () => {
+      const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
+      const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
+      await service.start();
+
+      const mergedTask = task({ id: "FN-303", 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-303",
+        merged: true,
+        worktreeRemoved: false,
+        branchDeleted: false,
+      });
+
+      await vi.waitFor(() => {
+        expect(sendNotification).toHaveBeenCalledTimes(1);
+      });
+      expect(sendNotification).toHaveBeenCalledWith("merged", expect.objectContaining({ taskId: "FN-303", event: "merged" }));
+    });
+
+    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" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock", isEventSupported: (event) => event !== "merged", sendNotification });
+      await service.start();
+
+      store.emit("task:moved", {
+        task: task({ id: "FN-304", column: "done", mergeDetails: { mergeConfirmed: true } as any }),
+        from: "in-review",
+        to: "done",
+      });
+      await Promise.resolve();
+
+      expect(sendNotification).not.toHaveBeenCalled();
+    });
+  });
+
   describe("stale-settings refresh for task lifecycle", () => {
     function createStaleLifecycleStore() {
       const listeners = new Map<string, Set<Listener>>();
@@ -752,14 +853,38 @@ describe("NotificationService", () => {
       expect(sendNotification).not.toHaveBeenCalled();
     });
 
-    it("does not notify for non-in-review moves even after stale-settings refresh", async () => {
+    it("refreshes stale disabled settings before merge-backed done move notifications", async () => {
       const store = createStaleLifecycleStore();
       const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
       const service = new NotificationService(store as any);
       service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
       await service.start();
 
-      store.emit("task:moved", { task: task({ id: "FN-104" }), from: "todo", to: "in-progress" });
+      store.emit("task:moved", {
+        task: task({ id: "FN-104", column: "done", prInfo: { status: "merged", number: 104 } as any }),
+        from: "in-review",
+        to: "done",
+      });
+
+      await vi.waitFor(() => {
+        expect(sendNotification).toHaveBeenCalledWith(
+          "merged",
+          expect.objectContaining({ taskId: "FN-104", event: "merged" }),
+        );
+      });
+      expect(schedulerLog.log).toHaveBeenCalledWith(
+        expect.stringContaining("NotificationService refreshed notification state reason=task:moved:done enabled=true"),
+      );
+    });
+
+    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" }));
+      const service = new NotificationService(store as any);
+      service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
+      await service.start();
+
+      store.emit("task:moved", { task: task({ id: "FN-106" }), from: "todo", to: "in-progress" });
       await Promise.resolve();
 
       expect(sendNotification).not.toHaveBeenCalled();

packages/engine/src/__tests__/workflow-graph-task-runner.test.ts

@@ -1,6 +1,8 @@
 import { describe, expect, it, vi } from "vitest";
+import { EventEmitter } from "node:events";
 import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr } from "@fusion/core";
 
+import { NotificationService } from "../notification/notification-service.js";
 import { WorkflowGraphTaskRunner, type WorkflowGraphRunnerStore } from "../workflow-graph-task-runner.js";
 import type { WorkflowNodeResult } from "../workflow-graph-executor.js";
 
@@ -92,6 +94,61 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
     expect(result.visitedNodeIds).toEqual(["start", "lint", "execute", "review", "merge", "notify"]);
   });
 
+  it("selected workflows reaching the merge seam produce the canonical merged notification once", async () => {
+    const emitter = new EventEmitter();
+    const graphTask = {
+      ...task,
+      title: "Workflow merge",
+      description: "Graph path",
+      column: "done",
+      mergeDetails: { mergeConfirmed: true },
+    } as TaskDetail;
+    const store = Object.assign(emitter, {
+      getSettings: vi.fn(async () => ({ ntfyEnabled: true, ntfyTopic: "topic" }) as Settings),
+      getTask: vi.fn(async (_id: string) => graphTask),
+      getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
+      getWorkflowDefinition: async () => definition(fullLifecycleIr()),
+    }) as unknown as EventEmitter & WorkflowGraphRunnerStore & {
+      getSettings: () => Promise<Settings>;
+      getTask: (id: string) => Promise<TaskDetail>;
+    };
+    const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
+    const service = new NotificationService(store as any);
+    service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification });
+    await service.start();
+
+    const runner = new WorkflowGraphTaskRunner({
+      store,
+      seams: {
+        ...recordingSeams([]),
+        merge: async () => {
+          store.emit("task:moved", { task: graphTask, from: "in-review", to: "done" });
+          store.emit("task:merged", {
+            task: graphTask,
+            branch: "fusion/fn-9001",
+            merged: true,
+            worktreeRemoved: false,
+            branchDeleted: false,
+          });
+          return { outcome: "success", value: "merged" };
+        },
+      },
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+
+    const result = await runner.run(graphTask, flagOn);
+
+    expect(result.disposition).toBe("completed");
+    await vi.waitFor(() => {
+      expect(sendNotification).toHaveBeenCalledTimes(1);
+    });
+    expect(sendNotification).toHaveBeenCalledWith(
+      "merged",
+      expect.objectContaining({ taskId: "FN-9001", taskTitle: "Workflow merge", event: "merged" }),
+    );
+    await service.stop();
+  });
+
   it("a failing seam terminates the run as failed without running later nodes", async () => {
     const calls: string[] = [];
     const runner = new WorkflowGraphTaskRunner({

packages/engine/src/notification/notification-service.ts

@@ -196,19 +196,34 @@ export class NotificationService {
   private async handleTaskMovedAsync(data: { task: Task; from: Column; to: Column }): Promise<void> {
     await this.maybeSuppressTransientFailedNotification(data.task, `moved to ${data.to}`);
 
-    if (data.to !== "in-review") {
+    if (data.to === "in-review") {
+      if (!this.notificationsEnabled) {
+        await this.refreshNotificationState("task:moved");
+        if (!this.notificationsEnabled) {
+          return;
+        }
+      }
+
+      const payload = this.createTaskPayload(data.task, "in-review");
+      this.maybeNotify(data.task.id, "in-review", payload);
       return;
     }
 
-    if (!this.notificationsEnabled) {
+    if (data.to === "done" && this.isMergeBackedTerminalTask(data.task)) {
-      await this.refreshNotificationState("task:moved");
+      // `task:merged` remains the canonical terminal merge event. This fallback
+      // preserves notification parity for PR/webhook/recovery paths that reach
+      // done through moveTask before (or without) a matching task:merged emit;
+      // maybeNotify uses the same `merged` key so a later task:merged event is
+      // suppressed instead of producing a duplicate alarm.
       if (!this.notificationsEnabled) {
-        return;
+        await this.refreshNotificationState("task:moved:done");
+        if (!this.notificationsEnabled) {
+          return;
+        }
       }
-    }
 
-    const payload = this.createTaskPayload(data.task, "in-review");
+      this.maybeNotify(data.task.id, "merged", this.createTaskPayload(data.task, "merged"));
-    this.maybeNotify(data.task.id, "in-review", payload);
+    }
   };
 
   private handleTaskUpdated = (task: Task): void => {
@@ -675,6 +690,13 @@ export class NotificationService {
     return this.pendingFailureNotifications.size;
   }
 
+  private isMergeBackedTerminalTask(task: Task): boolean {
+    return task.prInfo?.status === "merged" ||
+      task.mergeDetails?.mergeConfirmed === true ||
+      task.mergeDetails?.noOpMerge === true ||
+      typeof task.mergeDetails?.mergedAt === "string";
+  }
+
   private createTaskPayload(task: Task, event: NotificationEvent): NotificationPayload {
     return {
       taskId: task.id,