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

feat(FN-2340): merge fusion/fn-2340

Browse Source
This commit is contained in:
gsxdsm
2026-04-23 22:57:21 -07:00
parent 9ddb475947
commit 9bf2981b22
13 changed files with 592 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
packages/engine/src/index.ts
View File

@@ -38,7 +38,18 @@ export { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
export { withRateLimitRetry } from "./rate-limit-retry.js";
export { PrMonitor, type PrComment, type TrackedPr, type OnNewCommentsCallback } from "./pr-monitor.js";
export { PrCommentHandler } from "./pr-comment-handler.js";
export { NtfyNotifier, type NtfyNotifierOptions } from "./notifier.js";
export {
NtfyNotifier,
DEFAULT_NTFY_EVENTS,
resolveNtfyEvents,
isNtfyEventEnabled,
buildNtfyClickUrl,
sendNtfyNotification,
type NtfyNotifierOptions,
type NtfyNotificationPriority,
type NtfyNotificationConfigInput,
type SendNtfyNotificationInput,
} from "./notifier.js";
export { CronRunner, type CronRunnerOptions, type AiPromptExecutor, createAiPromptExecutor } from "./cron-runner.js";
export { RoutineRunner, type RoutineRunnerOptions } from "./routine-runner.js";
export { RoutineScheduler, type RoutineSchedulerOptions } from "./routine-scheduler.js";

30
packages/engine/src/notifier.test.ts
View File

@@ -1,7 +1,13 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { EventEmitter } from "node:events";
import type { Task, Column, MergeResult, Settings } from "@fusion/core";
import { NtfyNotifier } from "./notifier.js";
import {
NtfyNotifier,
DEFAULT_NTFY_EVENTS,
buildNtfyClickUrl,
isNtfyEventEnabled,
resolveNtfyEvents,
} from "./notifier.js";
// Mock the logger
vi.mock("./logger.js", () => ({
@@ -56,6 +62,24 @@ class MockTaskStore extends EventEmitter<MockTaskStoreEvents> {
}
}
describe("Ntfy notifier helpers", () => {
it("includes planning-awaiting-input in default events", () => {
expect(DEFAULT_NTFY_EVENTS).toContain("planning-awaiting-input");
expect(resolveNtfyEvents(undefined)).toContain("planning-awaiting-input");
});
it("checks planning-awaiting-input event enablement", () => {
expect(isNtfyEventEnabled(["planning-awaiting-input"], "planning-awaiting-input")).toBe(true);
expect(isNtfyEventEnabled(["failed"], "planning-awaiting-input")).toBe(false);
});
it("builds project dashboard root links without task id", () => {
expect(buildNtfyClickUrl({ dashboardHost: "http://localhost:4040/", projectId: "proj-1" })).toBe(
"http://localhost:4040/?project=proj-1",
);
});
});
describe("NtfyNotifier", () => {
let store: MockTaskStore;
let notifier: NtfyNotifier;
@@ -1334,7 +1358,7 @@ describe("NtfyNotifier", () => {
});
it("updates notifications when ntfyEvents changes at runtime", async () => {
store.setSettings({ ntfyEnabled: true, ntfyTopic: "test-topic", ntfyEvents: ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review"] });
store.setSettings({ ntfyEnabled: true, ntfyTopic: "test-topic", ntfyEvents: ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review", "planning-awaiting-input"] });
notifier = new NtfyNotifier(store);
await notifier.start();
@@ -1351,7 +1375,7 @@ describe("NtfyNotifier", () => {
expect(fetchMock).toHaveBeenCalledTimes(1); // No new call for in-review
// Enable in-review again
store.setSettings({ ntfyEvents: ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review"] });
store.setSettings({ ntfyEvents: ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review", "planning-awaiting-input"] });
store.triggerTaskMoved(createTask("FN-003", "Test Task 3"), "in-progress", "in-review");
await flushAsyncWork();

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

@@ -8,6 +8,46 @@ export interface NtfyNotifierOptions {
projectId?: string;
}
export type NtfyNotificationPriority = "low" | "default" | "high" | "urgent";
export const DEFAULT_NTFY_EVENTS: readonly NtfyNotificationEvent[] = [
"in-review",
"merged",
"failed",
"awaiting-approval",
"awaiting-user-review",
"planning-awaiting-input",
] as const;
export interface NtfyNotificationConfigInput {
enabled?: boolean;
topic?: string;
dashboardHost?: string;
events?: NtfyNotificationEvent[];
projectId?: string;
ntfyBaseUrl?: string;
}
export interface SendNtfyNotificationInput {
ntfyBaseUrl?: string;
topic: string;
title: string;
message: string;
priority?: NtfyNotificationPriority;
clickUrl?: string;
signal?: AbortSignal;
}
interface NtfyConfig {
enabled: boolean;
topic: string | undefined;
dashboardHost: string | undefined;
events: NtfyNotificationEvent[];
}
/** Event types for task notification deduplication */
type TaskNotificationEvent = "in-review" | "merged" | "failed" | "awaiting-approval" | "awaiting-user-review";
/**
* Format a task identifier for notifications.
* - If title exists: returns "{title}"
@@ -31,35 +71,95 @@ interface NtfyNotifierStore {
off(event: string, listener: (...args: any[]) => void): void;
}
interface NtfyConfig {
enabled: boolean;
topic: string | undefined;
dashboardHost: string | undefined;
events: NtfyNotificationEvent[];
export function resolveNtfyEvents(events?: NtfyNotificationEvent[]): NtfyNotificationEvent[] {
return events ? [...events] : [...DEFAULT_NTFY_EVENTS];
}
/** Event types for notification deduplication */
type NotificationEventType = "in-review" | "merged" | "failed" | "awaiting-approval" | "awaiting-user-review";
export function isNtfyEventEnabled(events: NtfyNotificationEvent[] | undefined, event: NtfyNotificationEvent): boolean {
return resolveNtfyEvents(events).includes(event);
}
export function buildNtfyClickUrl(options: {
dashboardHost?: string;
projectId?: string;
taskId?: string;
}): string | undefined {
const { dashboardHost, projectId, taskId } = options;
if (!dashboardHost) {
return undefined;
}
const normalizedHost = dashboardHost.replace(/\/+$/, "");
const queryParts: string[] = [];
if (projectId) {
queryParts.push(`project=${encodeURIComponent(projectId)}`);
}
if (taskId) {
queryParts.push(`task=${encodeURIComponent(taskId)}`);
}
const query = queryParts.join("&");
return query ? `${normalizedHost}/?${query}` : `${normalizedHost}/`;
}
/**
* Send a notification to ntfy.
* Errors are logged and swallowed so callers can treat delivery as best-effort.
*/
export async function sendNtfyNotification({
ntfyBaseUrl = "https://ntfy.sh",
topic,
title,
message,
priority = "default",
clickUrl,
signal,
}: SendNtfyNotificationInput): Promise<void> {
try {
const headers: Record<string, string> = {
Title: title,
Priority: priority,
"Content-Type": "text/plain",
};
if (clickUrl) {
headers.Click = clickUrl;
}
const response = await fetch(`${ntfyBaseUrl}/${topic}`, {
method: "POST",
headers,
body: message,
signal,
});
if (!response.ok) {
schedulerLog.log(`Ntfy notification failed: ${response.status} ${response.statusText}`);
}
} catch (err) {
if (err instanceof Error && err.name === "AbortError") {
return;
}
schedulerLog.log(`Failed to send ntfy notification: ${err}`);
}
}
/**
* NtfyNotifier sends push notifications via ntfy.sh when tasks complete
* or fail. It listens to TaskStore events and sends HTTP POST requests
* to the configured ntfy topic.
*
* Features:
* - Runtime reconfiguration via settings:updated events
* - Best-effort delivery (errors are logged but never thrown)
* - Duplicate prevention per event type (in-review, merged, failed, awaiting-approval)
* - Configurable notification events (hardcoded defaults)
*/
export class NtfyNotifier {
private config: NtfyConfig = { enabled: false, topic: undefined, dashboardHost: undefined, events: ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review"] };
private config: NtfyConfig = {
enabled: false,
topic: undefined,
dashboardHost: undefined,
events: [...DEFAULT_NTFY_EVENTS],
};
private ntfyBaseUrl: string;
/** Project identifier for deep links in notifications */
private projectId?: string;
/** Tracks which (taskId, eventType) pairs have been notified to prevent duplicates */
private notifiedEvents: Set<string> = new Set();
/** AbortController for in-flight requests during shutdown */
private abortController: AbortController | null = null;
constructor(
@@ -70,36 +170,20 @@ export class NtfyNotifier {
this.projectId = options.projectId;
}
/**
* Start listening to store events.
* Must be called after store is initialized.
* Returns a promise that resolves when initial config is loaded.
*/
async start(): Promise<void> {
this.abortController = new AbortController();
// Load initial config
const settings = await this.store.getSettings();
this.loadConfig(settings);
// Listen for task movements
this.store.on("task:moved", this.handleTaskMoved);
// Listen for task updates (status changes)
this.store.on("task:updated", this.handleTaskUpdated);
// Listen for merge events
this.store.on("task:merged", this.handleTaskMerged);
// Listen for settings changes for runtime reconfiguration
this.store.on("settings:updated", this.handleSettingsUpdated);
schedulerLog.log("NtfyNotifier started");
}
/**
* Stop listening to store events and abort in-flight requests.
*/
stop(): void {
if (typeof this.store.off === "function") {
this.store.off("task:moved", this.handleTaskMoved);
@@ -108,7 +192,6 @@ export class NtfyNotifier {
this.store.off("settings:updated", this.handleSettingsUpdated);
}
// Abort any in-flight requests
if (this.abortController) {
this.abortController.abort();
this.abortController = null;
@@ -122,66 +205,83 @@ export class NtfyNotifier {
const { task, to } = data;
// Notify when task moves to in-review (completed work, ready for review)
if (to === "in-review" && this.isEventEnabled("in-review")) {
const clickUrl = this.buildTaskUrl(task.id);
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.projectId,
taskId: task.id,
});
this.maybeNotify(task.id, "in-review", () =>
this.sendNotification(
this.config.topic!,
`Task ${task.id} completed`,
`Task "${formatTaskIdentifier(task)}" is ready for review`,
"default",
sendNtfyNotification({
ntfyBaseUrl: this.ntfyBaseUrl,
topic: this.config.topic!,
title: `Task ${task.id} completed`,
message: `Task "${formatTaskIdentifier(task)}" is ready for review`,
priority: "default",
clickUrl,
),
signal: this.abortController?.signal,
}),
);
}
// Note: "done" notifications come from handleTaskMerged (task:merged event)
// to avoid duplicate notifications when moveToDone is called before merge
};
private handleTaskUpdated = (task: Task): void => {
if (!this.config.enabled || !this.config.topic) return;
// Notify when task fails
if (task.status === "failed" && this.isEventEnabled("failed")) {
const clickUrl = this.buildTaskUrl(task.id);
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.projectId,
taskId: task.id,
});
this.maybeNotify(task.id, "failed", () =>
this.sendNotification(
this.config.topic!,
`Task ${task.id} failed`,
`Task "${formatTaskIdentifier(task)}" has failed and needs attention`,
"high",
sendNtfyNotification({
ntfyBaseUrl: this.ntfyBaseUrl,
topic: this.config.topic!,
title: `Task ${task.id} failed`,
message: `Task "${formatTaskIdentifier(task)}" has failed and needs attention`,
priority: "high",
clickUrl,
),
signal: this.abortController?.signal,
}),
);
}
// Notify when task requires manual plan approval
if (task.status === "awaiting-approval" && this.isEventEnabled("awaiting-approval")) {
const clickUrl = this.buildTaskUrl(task.id);
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.projectId,
taskId: task.id,
});
this.maybeNotify(task.id, "awaiting-approval", () =>
this.sendNotification(
this.config.topic!,
`Plan needs approval for ${task.id}`,
`Task "${formatTaskIdentifier(task)}" needs your approval before it can proceed`,
"high",
sendNtfyNotification({
ntfyBaseUrl: this.ntfyBaseUrl,
topic: this.config.topic!,
title: `Plan needs approval for ${task.id}`,
message: `Task "${formatTaskIdentifier(task)}" needs your approval before it can proceed`,
priority: "high",
clickUrl,
),
signal: this.abortController?.signal,
}),
);
}
// Notify when task needs human review (agent handoff to user)
if (task.status === "awaiting-user-review" && this.isEventEnabled("awaiting-user-review")) {
const clickUrl = this.buildTaskUrl(task.id);
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.projectId,
taskId: task.id,
});
this.maybeNotify(task.id, "awaiting-user-review", () =>
this.sendNotification(
this.config.topic!,
`User review needed for ${task.id}`,
`Task "${formatTaskIdentifier(task)}" needs human review before it can proceed`,
"high",
sendNtfyNotification({
ntfyBaseUrl: this.ntfyBaseUrl,
topic: this.config.topic!,
title: `User review needed for ${task.id}`,
message: `Task "${formatTaskIdentifier(task)}" needs human review before it can proceed`,
priority: "high",
clickUrl,
),
signal: this.abortController?.signal,
}),
);
}
};
@@ -189,17 +289,22 @@ export class NtfyNotifier {
private handleTaskMerged = (result: MergeResult): void => {
if (!this.config.enabled || !this.config.topic) return;
// Only notify on successful merges
if (result.merged && this.isEventEnabled("merged")) {
const clickUrl = this.buildTaskUrl(result.task.id);
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.projectId,
taskId: result.task.id,
});
this.maybeNotify(result.task.id, "merged", () =>
this.sendNotification(
this.config.topic!,
`Task ${result.task.id} merged`,
`Task "${formatTaskIdentifier(result.task)}" has been merged to main`,
"default",
sendNtfyNotification({
ntfyBaseUrl: this.ntfyBaseUrl,
topic: this.config.topic!,
title: `Task ${result.task.id} merged`,
message: `Task "${formatTaskIdentifier(result.task)}" has been merged to main`,
priority: "default",
clickUrl,
),
signal: this.abortController?.signal,
}),
);
}
};
@@ -207,7 +312,6 @@ export class NtfyNotifier {
private handleSettingsUpdated = (data: { settings: Settings; previous: Settings }): void => {
const { settings, previous } = data;
// Check if ntfy settings changed
if (settings.ntfyEnabled !== previous.ntfyEnabled ||
settings.ntfyTopic !== previous.ntfyTopic ||
settings.ntfyDashboardHost !== previous.ntfyDashboardHost ||
@@ -234,105 +338,31 @@ export class NtfyNotifier {
enabled: settings.ntfyEnabled ?? false,
topic: settings.ntfyTopic,
dashboardHost: settings.ntfyDashboardHost,
events: settings.ntfyEvents ?? ["in-review", "merged", "failed", "awaiting-approval", "awaiting-user-review"],
events: resolveNtfyEvents(settings.ntfyEvents),
};
}
/**
* Check if a notification event type is enabled based on the configured events list.
*/
private isEventEnabled(event: NotificationEventType): boolean {
return this.config.events.includes(event);
private isEventEnabled(event: TaskNotificationEvent): boolean {
return isNtfyEventEnabled(this.config.events, event);
}
/**
* Build a dashboard URL for deep linking to a task.
* Returns undefined if dashboardHost is not configured.
* Includes projectId in the URL when configured for multi-project support.
*/
private buildTaskUrl(taskId: string): string | undefined {
if (!this.config.dashboardHost) {
return undefined;
}
// Strip trailing slash from hostname if present
const host = this.config.dashboardHost.replace(/\/$/, "");
if (this.projectId) {
return `${host}/?project=${encodeURIComponent(this.projectId)}&task=${encodeURIComponent(taskId)}`;
}
return `${host}/?task=${encodeURIComponent(taskId)}`;
}
/**
* Send notification if this (taskId, eventType) pair hasn't been notified before.
* This prevents duplicate notifications for the same event type per task.
*/
private maybeNotify(
taskId: string,
eventType: NotificationEventType,
eventType: TaskNotificationEvent,
notifyFn: () => Promise<void>,
): void {
const key = `${taskId}:${eventType}`;
if (this.notifiedEvents.has(key)) {
// Already sent this notification type for this task
return;
}
this.notifiedEvents.add(key);
notifyFn().catch(() => {
// Errors are logged in sendNotification, just need to catch here
// sendNtfyNotification already logs; notifier must stay best-effort
});
}
/**
* Send a notification to ntfy.sh.
* Errors are caught and logged, never thrown.
*/
private async sendNotification(
topic: string,
title: string,
message: string,
priority: "low" | "default" | "high" | "urgent" = "default",
clickUrl?: string,
): Promise<void> {
const url = `${this.ntfyBaseUrl}/${topic}`;
const signal = this.abortController?.signal;
try {
const headers: Record<string, string> = {
"Title": title,
"Priority": priority,
"Content-Type": "text/plain",
};
// Add Click header for deep linking if URL is provided
if (clickUrl) {
headers["Click"] = clickUrl;
}
const response = await fetch(url, {
method: "POST",
headers,
body: message,
signal,
});
if (!response.ok) {
schedulerLog.log(`Ntfy notification failed: ${response.status} ${response.statusText}`);
}
} catch (err) {
// Don't throw - notifications are best-effort
if (err instanceof Error && err.name === "AbortError") {
// Expected during shutdown
return;
}
schedulerLog.log(`Failed to send ntfy notification: ${err}`);
}
}
/**
* Get current config (for testing purposes).
*/
getConfig(): NtfyConfig {
return { ...this.config, events: [...this.config.events] };
}