From 4ffd0a2ddef2effbb9ca839c3136540d182eebe6 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 13:54:17 -0700 Subject: [PATCH] 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 --- ...fn-6030-workflow-terminal-notifications.md | 5 + docs/storage.md | 8 +- .../src/__tests__/builtin-workflows.test.ts | 1 - .../store-pr-merged-transition.test.ts | 14 +- .../__tests__/workflow-ir-resolver.test.ts | 1 - packages/core/src/store.ts | 14 +- .../sections/ProjectModelsSection.tsx | 3 +- .../src/__tests__/merger-post-merge.test.ts | 9 +- .../__tests__/notification-service.test.ts | 129 +++++++++++++++++- .../workflow-graph-task-runner.test.ts | 57 ++++++++ .../src/notification/notification-service.ts | 36 ++++- 11 files changed, 257 insertions(+), 20 deletions(-) create mode 100644 .changeset/fn-6030-workflow-terminal-notifications.md diff --git a/.changeset/fn-6030-workflow-terminal-notifications.md b/.changeset/fn-6030-workflow-terminal-notifications.md new file mode 100644 index 0000000000..c939d9a578 --- /dev/null +++ b/.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. diff --git a/docs/storage.md b/docs/storage.md index 63099763a2..6b19d533e1 100644 --- a/docs/storage.md +++ b/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; - - 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. diff --git a/packages/core/src/__tests__/builtin-workflows.test.ts b/packages/core/src/__tests__/builtin-workflows.test.ts index 28cfde1541..b5a77d8a1d 100644 --- a/packages/core/src/__tests__/builtin-workflows.test.ts +++ b/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(); diff --git a/packages/core/src/__tests__/store-pr-merged-transition.test.ts b/packages/core/src/__tests__/store-pr-merged-transition.test.ts index 4870b04bcb..3576d98c91 100644 --- a/packages/core/src/__tests__/store-pr-merged-transition.test.ts +++ b/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 () => { diff --git a/packages/core/src/__tests__/workflow-ir-resolver.test.ts b/packages/core/src/__tests__/workflow-ir-resolver.test.ts index cc429652c4..5d8ed61c55 100644 --- a/packages/core/src/__tests__/workflow-ir-resolver.test.ts +++ b/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 } } }); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index e9e4ed5ab0..a5166dcc71 100644 --- a/packages/core/src/store.ts +++ b/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, diff --git a/packages/dashboard/app/components/settings/sections/ProjectModelsSection.tsx b/packages/dashboard/app/components/settings/sections/ProjectModelsSection.tsx index 322c5c99d5..56462b56a8 100644 --- a/packages/dashboard/app/components/settings/sections/ProjectModelsSection.tsx +++ b/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 { diff --git a/packages/engine/src/__tests__/merger-post-merge.test.ts b/packages/engine/src/__tests__/merger-post-merge.test.ts index b55f4d598f..44d6931a8d 100644 --- a/packages/engine/src/__tests__/merger-post-merge.test.ts +++ b/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).mock.invocationCallOrder[0], + ); }); it("uses assigned agent runtime model for post-merge prompt step when workflow step has no override", async () => { diff --git a/packages/engine/src/__tests__/notification-service.test.ts b/packages/engine/src/__tests__/notification-service.test.ts index a517b293c8..f26d5663ac 100644 --- a/packages/engine/src/__tests__/notification-service.test.ts +++ b/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>(); @@ -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(); diff --git a/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts b/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts index b28177a3e4..acfef78c1a 100644 --- a/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts +++ b/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; + getTask: (id: string) => Promise; + }; + 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({ diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index 3bdf478402..4ba711beff 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/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 { 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) { - await this.refreshNotificationState("task:moved"); + 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) { - return; + await this.refreshNotificationState("task:moved:done"); + if (!this.notificationsEnabled) { + return; + } } - } - const payload = this.createTaskPayload(data.task, "in-review"); - this.maybeNotify(data.task.id, "in-review", payload); + 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,