FN-6031: add workflow notification nodes

Add workflow notify nodes with templated notification dispatch support. - add notify node support to workflow IR, engine handlers, and executor wiring - expose notify node configuration and summaries in the dashboard editor and node metadata - add regression tests and a published changeset, plus workflow/settings documentation updates Files changed: .changeset/fn-6031-notification-node.md | 5 + docs/settings-reference.md | 6 +- docs/workflow-steps.md | 10 +- packages/core/src/__tests__/workflow-ir.test.ts | 59 ++++++++++ packages/core/src/types.ts | 4 +- packages/core/src/workflow-ir-types.ts | 4 +- packages/core/src/workflow-ir.ts | 19 +++ packages/dashboard/app/components/WorkflowNodeEditor.tsx | 76 +++++++++++- packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx | 61 ++++++++++ packages/dashboard/app/components/__tests__/node-summary.test.ts | 29 +++++ packages/dashboard/app/components/__tests__/workflow-flow-mapping.test.ts | 37 ++++++ packages/dashboard/app/components/nodes/WorkflowNodeTypes.tsx | 7 +- packages/dashboard/app/components/nodes/node-summary.ts | 5 + packages/engine/src/__tests__/workflow-node-handlers-notify.test.ts | 131 +++++++++++++++++++++ packages/engine/src/executor.ts | 2 + packages/engine/src/notification/ntfy-provider.ts | 11 ++ packages/engine/src/notification/webhook-provider.ts | 6 +- packages/engine/src/workflow-graph-executor.ts | 4 + packages/engine/src/workflow-graph-task-runner.ts | 4 + packages/engine/src/workflow-node-handlers.ts | 93 ++++++++++++++- 20 files changed, 560 insertions(+), 13 deletions(-) Fusion-Task-Id: FN-6031 Fusion-Task-Lineage: ff5c91e2-3872-4264-9c18-5d9e11628f13
2026-06-09 01:57:16 -07:00
parent 59c613cf38
commit 13c6d96fe1
20 changed files with 560 additions and 13 deletions
--- a/packages/engine/src/tests/workflow-node-handlers-notify.test.ts
+++ b/packages/engine/src/tests/workflow-node-handlers-notify.test.ts
@@ -0,0 +1,131 @@
+import { describe, expect, it, vi } from "vitest";
+import type { NotificationPayload, TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core";
+
+import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
+import { createDefaultNodeHandlers, createNoopLegacySeams } from "../workflow-node-handlers.js";
+import type { WorkflowNodeExecutionContext } from "../workflow-graph-executor.js";
+
+const notifyNode: WorkflowIrNode = {
+  id: "notify",
+  kind: "notify",
+  column: "todo",
+  config: {
+    event: "workflow-notify",
+    title: "{{taskTitle}} in {{workflowName}}",
+    message: "Task {{taskId}} hit {{context:stage}} with {{context:object}}",
+  },
+};
+
+function ctx(overrides: Partial<WorkflowNodeExecutionContext> = {}): WorkflowNodeExecutionContext {
+  return {
+    task: {
+      id: "FN-6031",
+      title: "Notify task",
+      description: "Task body",
+    } as TaskDetail,
+    settings: undefined,
+    context: {
+      "workflow:id": "Custom Workflow",
+      stage: "review",
+      object: { ok: true },
+    },
+    ...overrides,
+  };
+}
+
+describe("workflow notify node handler", () => {
+  it("dispatches an interpolated notification payload", async () => {
+    const dispatch = vi.fn(async () => undefined);
+    const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, {
+      notifyDispatch: dispatch,
+    });
+
+    const result = await handlers.notify(notifyNode, ctx());
+
+    expect(result).toEqual({ outcome: "success" });
+    expect(dispatch).toHaveBeenCalledTimes(1);
+    expect(dispatch).toHaveBeenCalledWith(
+      "workflow-notify",
+      expect.objectContaining({
+        taskId: "FN-6031",
+        taskTitle: "Notify task",
+        taskDescription: "Task body",
+        event: "workflow-notify",
+        metadata: expect.objectContaining({
+          nodeId: "notify",
+          workflowName: "Custom Workflow",
+          title: "Notify task in Custom Workflow",
+          message: 'Task FN-6031 hit review with {"ok":true}',
+        }),
+      } satisfies Partial<NotificationPayload>),
+    );
+  });
+
+  it("skips successfully when dispatch is unwired", async () => {
+    const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, {});
+
+    await expect(handlers.notify(notifyNode, ctx())).resolves.toEqual({
+      outcome: "success",
+      value: "notify-skipped",
+    });
+  });
+
+  it("handles missing config gracefully", async () => {
+    const dispatch = vi.fn(async () => undefined);
+    const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, {
+      notifyDispatch: dispatch,
+    });
+
+    await expect(handlers.notify({ id: "notify", kind: "notify" } as WorkflowIrNode, ctx())).resolves.toEqual({
+      outcome: "success",
+      value: "notify-skipped",
+    });
+    expect(dispatch).not.toHaveBeenCalled();
+  });
+
+  it("does not fail the node when dispatch throws", async () => {
+    const dispatch = vi.fn(async () => {
+      throw new Error("provider down");
+    });
+    const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, {
+      notifyDispatch: dispatch,
+    });
+
+    await expect(handlers.notify(notifyNode, ctx())).resolves.toEqual({ outcome: "success" });
+  });
+
+  it("is wired into the graph executor default handlers", async () => {
+    const dispatch = vi.fn(async () => undefined);
+    const ir: WorkflowIr = {
+      version: "v2",
+      name: "Notify Workflow",
+      columns: [{ id: "todo", name: "Todo", traits: [] }],
+      nodes: [
+        { id: "start", kind: "start", column: "todo" },
+        { id: "notify", kind: "notify", column: "todo", config: { event: "custom-event", message: "{{workflowName}}" } },
+        { id: "end", kind: "end", column: "todo" },
+      ],
+      edges: [
+        { from: "start", to: "notify" },
+        { from: "notify", to: "end" },
+      ],
+    };
+    const executor = new WorkflowGraphExecutor({ notifyDispatch: dispatch });
+
+    const result = await executor.run(
+      { id: "FN-6031", description: "body" } as TaskDetail,
+      { experimentalFeatures: { workflowGraphExecutor: true } },
+      ir,
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(result.visitedNodeIds).toEqual(["start", "notify"]);
+    expect(dispatch).toHaveBeenCalledWith(
+      "custom-event",
+      expect.objectContaining({
+        taskTitle: "FN-6031",
+        metadata: expect.objectContaining({ message: "Notify Workflow" }),
+      }),
+    );
+  });
+});
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -20,6 +20,7 @@ import {
 } from "@fusion/core";
 import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js";
 import { createCodeNodeRunner } from "./code-node-runner.js";
+import { getActiveNotificationService } from "./notifier.js";
 import type { ParseStepsHandlerDeps, CodeNodeRunner } from "./workflow-node-handlers.js";
 import type { WorkflowBranchPersistence, WorkflowBranchRunState } from "./workflow-graph-branches.js";
 import type {
@@ -3786,6 +3787,7 @@ export class TaskExecutor {
        // Step-inversion (KTD-15, U14): code node runner — esbuild compile +
        // child-process execution with the harness contract.
        runCode: this.buildCodeNodeRunner(),
+        notifyDispatch: (event, payload) => getActiveNotificationService()?.dispatch(event, payload),
        // PR-entity nodes (U3): pr-create/pr-respond/pr-merge handler deps —
        // engine-owned store + CLI-injected GitHub callbacks. Absent → fail closed.
        prNodes: this.options.prNodes,
--- a/packages/engine/src/notification/ntfy-provider.ts
+++ b/packages/engine/src/notification/ntfy-provider.ts
@@ -39,6 +39,7 @@ type SupportedNtfyEvent =
  | "planning-awaiting-input"
  | "fallback-used"
  | "task-created"
+  | "workflow-notify"
  | "message:agent-to-user"
  | "message:agent-to-agent"
  | "message:room"
@@ -53,6 +54,7 @@ const SUPPORTED_EVENTS = new Set<SupportedNtfyEvent>([
  "planning-awaiting-input",
  "fallback-used",
  "task-created",
+  "workflow-notify",
  "message:agent-to-user",
  "message:agent-to-agent",
  "message:room",
@@ -235,6 +237,15 @@ export class NtfyNotificationProvider implements NotificationProvider {
        message: `${typeof payload.metadata?.agentName === "string" && payload.metadata.agentName.trim().length > 0 ? payload.metadata.agentName.trim() : "An agent"} created "${identifier}"`,
        priority: "default",
      },
+      "workflow-notify": {
+        title: typeof payload.metadata?.title === "string" && payload.metadata.title.trim().length > 0
+          ? payload.metadata.title.trim()
+          : `Workflow notification for ${taskId}`,
+        message: typeof payload.metadata?.message === "string" && payload.metadata.message.trim().length > 0
+          ? payload.metadata.message.trim()
+          : `Workflow notification for task "${identifier}"`,
+        priority: "default",
+      },
      "message:agent-to-user": {
        title: `New message from ${senderLabel}`,
        message: `${senderLabel} → you: ${preview}`,
--- a/packages/engine/src/notification/webhook-provider.ts
+++ b/packages/engine/src/notification/webhook-provider.ts
@@ -196,8 +196,10 @@ export class WebhookNotificationProvider implements NotificationProvider {
        const providerName = typeof payload.metadata?.providerName === "string" ? payload.metadata.providerName : providerId;
        return `Your ${providerName} OAuth token has expired — please re-authenticate`;
      }
-      default:
-        return `Event "${event}" for task ${identifier}`;
+      default: {
+        const message = typeof payload.metadata?.message === "string" ? payload.metadata.message.trim() : "";
+        return message || `Event "${event}" for task ${identifier}`;
+      }
    }
  }

--- a/packages/engine/src/workflow-graph-executor.ts
+++ b/packages/engine/src/workflow-graph-executor.ts
@@ -10,6 +10,7 @@ import {
  type CodeNodeRunner,
  type ForeachActiveContext,
  type ParseStepsHandlerDeps,
+  type WorkflowNotifyDispatch,
  type WorkflowCustomNodeRunner,
  type WorkflowLegacySeams,
 } from "./workflow-node-handlers.js";
@@ -74,6 +75,8 @@ export interface WorkflowGraphExecutorDeps {
  /** Step-inversion (U14, KTD-15): runner for the `code` node (esbuild compile +
   *  child-process execution). Absent → a code node fails cleanly. */
  runCode?: CodeNodeRunner;
+  /** notify node dispatch callback. Absent → notify nodes succeed with notify-skipped. */
+  notifyDispatch?: WorkflowNotifyDispatch;
  /** PR-entity nodes (U3): deps for `pr-create`/`pr-respond`/`pr-merge` (injected
   *  GitHub callbacks + store accessor). Absent → the pr-* kinds fail cleanly. */
  prNodes?: PrNodeDeps;
@@ -233,6 +236,7 @@ export class WorkflowGraphExecutor {
        primitives: deps.primitives,
        parseSteps: deps.parseStepsDeps,
        runCode: deps.runCode,
+        notifyDispatch: deps.notifyDispatch,
        prNodes: deps.prNodes,
      }),
      ...(deps.handlers ?? {}),
--- a/packages/engine/src/workflow-graph-task-runner.ts
+++ b/packages/engine/src/workflow-graph-task-runner.ts
@@ -6,6 +6,7 @@ import type {
  CodeNodeRunner,
  ForeachActiveContext,
  ParseStepsHandlerDeps,
+  WorkflowNotifyDispatch,
  WorkflowCustomNodeRunner,
  WorkflowLegacySeams,
 } from "./workflow-node-handlers.js";
@@ -72,6 +73,8 @@ export interface WorkflowGraphTaskRunnerDeps {
  /** Step-inversion (U14, KTD-15): `code` node runner. Additive; a workflow with
   *  no code node never invokes it. */
  runCode?: CodeNodeRunner;
+  /** notify node dispatch callback. Additive; absent → notify nodes are skipped. */
+  notifyDispatch?: WorkflowNotifyDispatch;
  /** PR-entity nodes (U3): deps for `pr-create`/`pr-respond`/`pr-merge`. Additive;
   *  a workflow with no pr-* node never invokes them; absent → they fail closed. */
  prNodes?: PrNodeDeps;
@@ -224,6 +227,7 @@ export class WorkflowGraphTaskRunner {
        onReworkReset: this.deps.onReworkReset,
        parseStepsDeps: this.deps.parseStepsDeps,
        runCode: this.deps.runCode,
+        notifyDispatch: this.deps.notifyDispatch,
        prNodes: this.deps.prNodes,
        // Step-inversion (KTD-11, U10): worktree isolation + parallel scheduling.
        allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
--- a/packages/engine/src/workflow-node-handlers.ts
+++ b/packages/engine/src/workflow-node-handlers.ts
@@ -1,8 +1,9 @@
 import { WorkflowIrError, getStepParser, instanceNodeId } from "@fusion/core";
-import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";
+import type { NotificationEvent, NotificationPayload, TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";

 import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";
 import { createPrNodeHandlers, createAutoMergeGateHandler, type PrNodeDeps } from "./pr-nodes.js";
+import { schedulerLog } from "./logger.js";
 import {
  primitiveNodeContext,
  type WorkflowPrimitiveContext,
@@ -763,6 +764,92 @@ export function createCodeNodeHandler(runCode?: CodeNodeRunner): WorkflowNodeHan
  };
 }

+export type WorkflowNotifyDispatch = (
+  event: NotificationEvent,
+  payload: NotificationPayload,
+) => Promise<void> | void;
+
+function stringifyTemplateValue(value: unknown): string {
+  if (value === undefined || value === null) return "";
+  if (typeof value === "string") return value;
+  if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") {
+    return String(value);
+  }
+  try {
+    return JSON.stringify(value);
+  } catch {
+    return String(value);
+  }
+}
+
+function interpolateNotifyTemplate(
+  template: string,
+  vars: {
+    taskId: string;
+    taskTitle: string;
+    workflowName: string;
+    context: Record<string, unknown>;
+  },
+): string {
+  return template.replace(/\{\{\s*([^{}]+?)\s*\}\}/g, (_match, rawName: string) => {
+    const name = rawName.trim();
+    if (name === "taskId") return vars.taskId;
+    if (name === "taskTitle") return vars.taskTitle;
+    if (name === "workflowName") return vars.workflowName;
+    if (name.startsWith("context:")) {
+      return stringifyTemplateValue(vars.context[name.slice("context:".length)]);
+    }
+    return `{{${rawName}}}`;
+  });
+}
+
+export function createNotifyHandler(notifyDispatch?: WorkflowNotifyDispatch): WorkflowNodeHandler {
+  return async (node, ctx) => {
+    const cfg = (node.config ?? {}) as { event?: unknown; message?: unknown; title?: unknown };
+    const event = typeof cfg.event === "string" ? cfg.event.trim() : "";
+    if (!event) {
+      schedulerLog.log(`Workflow notify node '${node.id}' skipped because it has no event`);
+      return { outcome: "success", value: "notify-skipped" };
+    }
+    if (!notifyDispatch) {
+      schedulerLog.log(`Workflow notify node '${node.id}' skipped because notification dispatch is unwired`);
+      return { outcome: "success", value: "notify-skipped" };
+    }
+
+    const taskTitle = typeof ctx.task.title === "string" && ctx.task.title.trim() !== ""
+      ? ctx.task.title
+      : ctx.task.id;
+    const workflowName = typeof ctx.context[WORKFLOW_ID_CONTEXT_KEY] === "string"
+      ? ctx.context[WORKFLOW_ID_CONTEXT_KEY]
+      : "unknown";
+    const vars = { taskId: ctx.task.id, taskTitle, workflowName, context: ctx.context };
+    const title = typeof cfg.title === "string" ? interpolateNotifyTemplate(cfg.title, vars) : taskTitle;
+    const message = typeof cfg.message === "string" ? interpolateNotifyTemplate(cfg.message, vars) : "";
+    const payload: NotificationPayload = {
+      taskId: ctx.task.id,
+      taskTitle,
+      taskDescription: ctx.task.description,
+      event,
+      timestamp: new Date().toISOString(),
+      metadata: {
+        nodeId: node.id,
+        workflowName,
+        title,
+        message,
+      },
+    };
+
+    try {
+      await notifyDispatch(event, payload);
+    } catch (err) {
+      const detail = err instanceof Error ? err.message : String(err);
+      schedulerLog.log(`Workflow notify node '${node.id}' dispatch failed for event=${event}: ${detail}`);
+    }
+
+    return { outcome: "success" };
+  };
+}
+
 export interface DefaultNodeHandlerDeps {
  /** Workflow-native runtime primitives. When present they replace legacy seams. */
  primitives?: WorkflowRuntimePrimitives;
@@ -770,6 +857,8 @@ export interface DefaultNodeHandlerDeps {
  parseSteps?: ParseStepsHandlerDeps;
  /** code node runner (U14). When absent, a code node fails cleanly. */
  runCode?: CodeNodeRunner;
+  /** notify node dispatch callback. When absent, notify nodes succeed with notify-skipped. */
+  notifyDispatch?: WorkflowNotifyDispatch;
  /** PR node deps (U3). When absent, the three pr-* kinds fail cleanly. */
  prNodes?: PrNodeDeps;
 }
@@ -785,6 +874,7 @@ export function createDefaultNodeHandlers(
  | "step-review"
  | "parse-steps"
  | "code"
+  | "notify"
  | "pr-create"
  | "pr-respond"
  | "pr-merge",
@@ -827,6 +917,7 @@ export function createDefaultNodeHandlers(
      : createStepReviewHandler(seams),
    "parse-steps": parseSteps,
    code: createCodeNodeHandler(deps?.runCode),
+    notify: createNotifyHandler(deps?.notifyDispatch),
    ...prNodes,
  };
 }