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

FN-6030: fix workflow terminal notifications

Browse Source 
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
This commit is contained in:
gsxdsm
2026-06-08 13:54:17 -07:00
parent ed0dc4a7b9
commit 4ffd0a2dde
11 changed files with 257 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-6030-workflow-terminal-notifications.md Normal file
View File

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

8
docs/storage.md
View File

@@ -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;
- 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);
- a direct `tasks_fts_data` size check during the review was only **47884 bytes** for the current project DB;
- 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.
- 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;
- 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);
- 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 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.

1
packages/core/src/__tests__/builtin-workflows.test.ts
View File

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

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

@@ -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 () => {

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

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

14
packages/core/src/store.ts
View File

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

3
packages/dashboard/app/components/settings/sections/ProjectModelsSection.tsx
View File

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

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

@@ -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 () => {

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

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

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

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

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

@@ -196,10 +196,7 @@ 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") {
return;
}
if (data.to === "in-review") {
if (!this.notificationsEnabled) {
await this.refreshNotificationState("task:moved");
if (!this.notificationsEnabled) {
@@ -209,6 +206,24 @@ export class NotificationService {
const payload = this.createTaskPayload(data.task, "in-review");
this.maybeNotify(data.task.id, "in-review", payload);
return;
}
if (data.to === "done" && this.isMergeBackedTerminalTask(data.task)) {
// `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) {
await this.refreshNotificationState("task:moved:done");
if (!this.notificationsEnabled) {
return;
}
}
this.maybeNotify(data.task.id, "merged", this.createTaskPayload(data.task, "merged"));
}
};
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,