feat(FN-4178): add room message notification delivery

This merge implements room message notifications across the system, adding a core room event type, wiring the notification dispatcher to room activity, and delivering notifications via ntfy and webhook providers with updated settings UI and API routes.

Fusion-Task-Id: FN-4178
This commit is contained in:
Fusion
2026-05-12 13:25:29 -07:00
committed by gsxdsm
parent f27ac8bf59
commit 2a26b0d344
20 changed files with 531 additions and 53 deletions

View File

@@ -1,6 +1,6 @@
import { describe, expect, it, vi } from "vitest";
import { EventEmitter } from "node:events";
import type { Message, NotificationProvider, Settings, Task } from "@fusion/core";
import type { ChatRoomMessage, Message, NotificationProvider, Settings, Task } from "@fusion/core";
import { NotificationService } from "../notification/notification-service.js";
import { NtfyNotificationProvider } from "../notification/ntfy-provider.js";
import { schedulerLog } from "../logger.js";
@@ -65,6 +65,22 @@ function createMessage(overrides: Partial<Message> = {}): Message {
} as Message;
}
function createRoomMessage(overrides: Partial<ChatRoomMessage> = {}): ChatRoomMessage {
return {
id: "rmsg-1",
roomId: "room-1",
role: "assistant",
content: "hello from room agent",
thinkingOutput: null,
metadata: null,
attachments: [],
senderAgentId: "agent-1",
mentions: [],
createdAt: new Date().toISOString(),
...overrides,
};
}
describe("NotificationService", () => {
it("dispatches in-review event to registered provider", async () => {
const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
@@ -263,6 +279,98 @@ describe("NotificationService", () => {
);
});
it("dispatches message:room from chat:room:message:added", async () => {
const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
const chatStore = new EventEmitter() as EventEmitter & {
getRoom: (id: string) => { id: string; name: string } | undefined;
};
chatStore.getRoom = (id: string) => (id === "room-1" ? { id, name: "Incident Room" } : undefined);
const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
const provider: NotificationProvider = {
getProviderId: () => "mock",
isEventSupported: () => true,
sendNotification,
};
const service = new NotificationService(store as any, {
chatStore: chatStore as any,
agentNameResolver: (agentId) => (agentId === "agent-1" ? "Triage Bot" : null),
});
service.registerProvider(provider);
await service.start();
chatStore.emit("chat:room:message:added", createRoomMessage());
await vi.waitFor(() => {
expect(sendNotification).toHaveBeenCalled();
});
expect(sendNotification).toHaveBeenCalledWith(
"message:room",
expect.objectContaining({
event: "message:room",
metadata: expect.objectContaining({
messageId: "rmsg-1",
roomId: "room-1",
roomName: "Incident Room",
senderAgentId: "agent-1",
senderName: "Triage Bot",
preview: "hello from room agent",
type: "room-assistant",
}),
}),
);
});
it("dispatches room notifications when chat store attaches after start", async () => {
const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
const chatStore = new EventEmitter() as EventEmitter & {
getRoom: (id: string) => { id: string; name: string } | undefined;
};
chatStore.getRoom = (id: string) => (id === "room-1" ? { id, name: "Incident Room" } : undefined);
const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
const provider: NotificationProvider = {
getProviderId: () => "mock",
isEventSupported: () => true,
sendNotification,
};
const service = new NotificationService(store as any, {
agentNameResolver: () => "Triage Bot",
});
service.registerProvider(provider);
await service.start();
service.attachChatStore(chatStore as any);
chatStore.emit("chat:room:message:added", createRoomMessage());
await vi.waitFor(() => {
expect(sendNotification).toHaveBeenCalledWith(
"message:room",
expect.objectContaining({ event: "message:room" }),
);
});
});
it("ignores non-agent or non-assistant room messages", async () => {
const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
const chatStore = new EventEmitter();
const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" }));
const provider: NotificationProvider = {
getProviderId: () => "mock",
isEventSupported: () => true,
sendNotification,
};
const service = new NotificationService(store as any, { chatStore: chatStore as any });
service.registerProvider(provider);
await service.start();
chatStore.emit("chat:room:message:added", createRoomMessage({ role: "user" }));
chatStore.emit("chat:room:message:added", createRoomMessage({ id: "rmsg-2", senderAgentId: null }));
await Promise.resolve();
expect(sendNotification).not.toHaveBeenCalled();
});
it("dispatches even when agent name resolution fails", async () => {
const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" });
const messageStore = new EventEmitter();

View File

@@ -73,6 +73,7 @@ describe("Ntfy notifier helpers", () => {
expect(DEFAULT_NTFY_EVENTS).toContain("fallback-used");
expect(DEFAULT_NTFY_EVENTS).toContain("message:agent-to-user");
expect(DEFAULT_NTFY_EVENTS).toContain("message:agent-to-agent");
expect(DEFAULT_NTFY_EVENTS).toContain("message:room");
});
it("checks planning-awaiting-input event enablement", () => {
@@ -106,6 +107,18 @@ describe("Ntfy notifier helpers", () => {
}),
).toBe("http://localhost:4040/?project=proj-1&view=mailbox&mailbox-message=msg-1#message-msg-1");
});
it("builds room message deep links", () => {
expect(
buildNtfyClickUrl({
dashboardHost: "http://localhost:4040/",
projectId: "proj-1",
roomId: "room-1",
messageId: "msg-1",
view: "rooms",
}),
).toBe("http://localhost:4040/?project=proj-1&view=rooms&room=room-1#message-msg-1");
});
});
describe("sendNtfyNotificationWithResult", () => {

View File

@@ -46,6 +46,7 @@ describe("NtfyNotificationProvider", () => {
["fallback-used", "Fallback model used for FN-1", "switched from", "high"],
["message:agent-to-user", "New message from Triage Bot", "Triage Bot → you: preview text", "high"],
["message:agent-to-agent", "Triage Bot → Executor Bot", "Triage Bot messaged Executor Bot: preview text", "default"],
["message:room", "#Incident Room — Triage Bot", "Triage Bot in #Incident Room: preview text", "default"],
])("maps %s event correctly", async (event, expectedTitle, messagePart, priority) => {
await provider.sendNotification(event as any, {
taskId: "FN-1",
@@ -56,8 +57,12 @@ describe("NtfyNotificationProvider", () => {
toId: "agent-2",
fromName: "Triage Bot",
toName: "Executor Bot",
senderAgentId: "agent-1",
senderName: "Triage Bot",
preview: "preview text",
messageId: "msg-1",
roomId: "room-1",
roomName: "Incident Room",
},
});
@@ -87,6 +92,7 @@ describe("NtfyNotificationProvider", () => {
expect(provider.isEventSupported("fallback-used" as any)).toBe(true);
expect(provider.isEventSupported("message:agent-to-user" as any)).toBe(true);
expect(provider.isEventSupported("message:agent-to-agent" as any)).toBe(true);
expect(provider.isEventSupported("message:room" as any)).toBe(true);
expect(provider.isEventSupported("custom-event" as any)).toBe(false);
});
@@ -124,6 +130,28 @@ describe("NtfyNotificationProvider", () => {
});
});
it("uses room deep link for room notifications", async () => {
await provider.sendNotification("message:room" as any, {
event: "message:room" as any,
metadata: {
messageId: "msg-room",
roomId: "room-1",
roomName: "Incident Room",
senderAgentId: "agent-1",
senderName: "Triage Bot",
preview: "hello",
},
});
expect(mocks.buildNtfyClickUrl).toHaveBeenCalledWith(
expect.objectContaining({
roomId: "room-1",
messageId: "msg-room",
view: "rooms",
}),
);
});
it("uses mailbox deep link when message is not task-bound", async () => {
await provider.sendNotification("message:agent-to-agent" as any, {
event: "message:agent-to-agent" as any,

View File

@@ -167,6 +167,7 @@ describe("WebhookNotificationProvider", () => {
["fallback-used", "Fusion recovered by switching from"],
["message:agent-to-user", "From: Triage Bot → You: hello"],
["message:agent-to-agent", "From: Triage Bot → To: Executor Bot: hello"],
["message:room", "In #Incident Room: Triage Bot: hello"],
["unknown-event", 'Event "unknown-event" for task My Task'],
])("message formatting for %s", async (event, expectedPart) => {
fetchMock.mockResolvedValue({ ok: true, status: 200, statusText: "OK" });
@@ -176,7 +177,17 @@ describe("WebhookNotificationProvider", () => {
taskId: "FN-1",
taskTitle: "My Task",
event,
metadata: { fromId: "agent-1", toId: "agent-2", fromName: "Triage Bot", toName: "Executor Bot", preview: "hello" },
metadata: {
fromId: "agent-1",
toId: "agent-2",
fromName: "Triage Bot",
toName: "Executor Bot",
senderAgentId: "agent-1",
senderName: "Triage Bot",
roomId: "room-1",
roomName: "Incident Room",
preview: "hello",
},
});
const [, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
@@ -184,6 +195,40 @@ describe("WebhookNotificationProvider", () => {
expect(body.text).toContain(expectedPart);
});
it("includes room metadata and room deep link for room notifications", async () => {
fetchMock.mockResolvedValue({ ok: true, status: 200, statusText: "OK" });
await provider.initialize({
webhookUrl: "https://example.com/hook",
webhookFormat: "generic",
dashboardHost: "http://dash",
projectId: "p1",
});
await provider.sendNotification("message:room", {
event: "message:room",
metadata: {
messageId: "msg-room",
roomId: "room-1",
roomName: "Incident Room",
senderAgentId: "agent-1",
senderName: "Triage Bot",
preview: "hello",
},
});
const [, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
const payload = JSON.parse(String(requestInit.body));
expect(payload.metadata).toEqual(
expect.objectContaining({
roomId: "room-1",
roomName: "Incident Room",
senderAgentId: "agent-1",
senderName: "Triage Bot",
}),
);
expect(payload.clickUrl).toBe("http://dash/?project=p1&view=rooms&room=room-1#message-msg-room");
});
it("title fallback uses taskId and truncated description snippet", async () => {
fetchMock.mockResolvedValue({ ok: true, status: 200, statusText: "OK" });
await provider.initialize({ webhookUrl: "https://example.com/hook", webhookFormat: "slack" });

View File

@@ -1,4 +1,5 @@
import type {
ChatRoomMessage,
Column,
MergeResult,
Message,
@@ -21,6 +22,8 @@ export interface NotificationServiceOptions {
ntfyBaseUrl?: string;
/** Optional message store for mailbox message notifications */
messageStore?: NotificationMessageStore;
/** Optional chat store for room message notifications */
chatStore?: NotificationChatStore;
/** Resolve human-readable name for an agent ID used in message notifications */
agentNameResolver?: (agentId: string) => Promise<string | null> | string | null;
}
@@ -36,10 +39,17 @@ interface NotificationMessageStore {
off?(event: "message:sent", listener: (message: Message) => void): void;
}
export interface NotificationChatStore {
on(event: "chat:room:message:added", listener: (message: ChatRoomMessage) => void): void;
off?(event: "chat:room:message:added", listener: (message: ChatRoomMessage) => void): void;
getRoom?(id: string): { id: string; name: string } | undefined;
}
export class NotificationService {
private readonly dispatcher = new NotificationDispatcher();
private readonly notifiedEvents = new Set<string>();
private started = false;
private chatStore: NotificationChatStore | undefined;
private notificationsEnabled = false;
private ntfyProvider?: NtfyNotificationProvider;
private webhookProvider?: WebhookNotificationProvider;
@@ -48,7 +58,19 @@ export class NotificationService {
constructor(
private readonly store: NotificationServiceStore,
private readonly options: NotificationServiceOptions = {},
) {}
) {
this.chatStore = options.chatStore;
}
attachChatStore(chatStore: NotificationChatStore): void {
if (this.chatStore && this.chatStore !== chatStore) {
this.detachChatStoreListener(this.chatStore);
}
this.chatStore = chatStore;
if (this.started) {
this.chatStore.on("chat:room:message:added", this.handleRoomMessageAdded);
}
}
registerProvider(provider: NotificationProvider): void {
this.dispatcher.registerProvider(provider);
@@ -71,8 +93,8 @@ export class NotificationService {
this.store.on("task:merged", this.handleTaskMerged);
this.store.on("settings:updated", this.handleSettingsUpdated);
this.options.messageStore?.on("message:sent", this.handleMessageSent);
this.started = true;
this.chatStore?.on("chat:room:message:added", this.handleRoomMessageAdded);
schedulerLog.log("NotificationService started");
}
@@ -89,6 +111,7 @@ export class NotificationService {
if (typeof this.options.messageStore?.off === "function") {
this.options.messageStore.off("message:sent", this.handleMessageSent);
}
this.detachChatStoreListener(this.chatStore);
}
await this.dispatcher.shutdownAll();
@@ -246,6 +269,10 @@ export class NotificationService {
void this.handleMessageSentAsync(message);
};
private handleRoomMessageAdded = (message: ChatRoomMessage): void => {
void this.handleRoomMessageAddedAsync(message);
};
private async handleMessageSentAsync(message: Message): Promise<void> {
schedulerLog.log(
`NotificationService.handleMessageSent messageId=${message.id} type=${message.type} notificationsEnabled=${String(this.notificationsEnabled)} hasNtfyProvider=${String(Boolean(this.ntfyProvider))}`,
@@ -267,9 +294,7 @@ export class NotificationService {
return;
}
const preview = message.content.length > 100
? `${message.content.slice(0, 100)}`
: message.content;
const preview = this.createPreview(message.content);
const taskId = typeof message.metadata?.taskId === "string" ? message.metadata.taskId : undefined;
@@ -299,6 +324,44 @@ export class NotificationService {
);
}
private async handleRoomMessageAddedAsync(message: ChatRoomMessage): Promise<void> {
schedulerLog.log(
`NotificationService.handleRoomMessageAdded messageId=${message.id} roomId=${message.roomId} role=${message.role} notificationsEnabled=${String(this.notificationsEnabled)}`,
);
if (message.role !== "assistant" || message.senderAgentId == null) {
return;
}
if (!this.notificationsEnabled) {
await this.refreshNotificationState("chat:room:message:added");
if (!this.notificationsEnabled) {
return;
}
}
const senderName = await this.resolveAgentName("agent", message.senderAgentId, "from");
const roomName = this.chatStore?.getRoom?.(message.roomId)?.name;
const preview = this.createPreview(message.content);
this.maybeNotify(message.id, "message:room", {
event: "message:room",
metadata: {
messageId: message.id,
roomId: message.roomId,
...(roomName ? { roomName } : {}),
senderAgentId: message.senderAgentId,
...(senderName ? { senderName } : {}),
preview,
type: "room-assistant",
},
});
schedulerLog.log(
`NotificationService.handleRoomMessageAdded scheduled eventType=message:room messageId=${message.id}`,
);
}
private async resolveAgentName(
participantType: Message["fromType"],
participantId: string,
@@ -326,6 +389,16 @@ export class NotificationService {
}
}
private createPreview(content: string): string {
return content.length > 100 ? `${content.slice(0, 100)}` : content;
}
private detachChatStoreListener(chatStore: NotificationChatStore | undefined): void {
if (typeof chatStore?.off === "function") {
chatStore.off("chat:room:message:added", this.handleRoomMessageAdded);
}
}
private setNotificationsEnabledFromSettings(settings: Settings): void {
this.notificationsEnabled = Boolean(
(settings.ntfyEnabled && settings.ntfyTopic) ||

View File

@@ -39,7 +39,8 @@ type SupportedNtfyEvent =
| "planning-awaiting-input"
| "fallback-used"
| "message:agent-to-user"
| "message:agent-to-agent";
| "message:agent-to-agent"
| "message:room";
const SUPPORTED_EVENTS = new Set<SupportedNtfyEvent>([
"in-review",
@@ -51,6 +52,7 @@ const SUPPORTED_EVENTS = new Set<SupportedNtfyEvent>([
"fallback-used",
"message:agent-to-user",
"message:agent-to-agent",
"message:room",
]);
export function resolveParticipantLabel(
@@ -67,6 +69,24 @@ export function resolveParticipantLabel(
return id.length > 0 ? id : kind === "from" ? "agent" : "recipient";
}
function resolveRoomSenderLabel(metadata: NotificationPayload["metadata"] | undefined): string {
const senderName = typeof metadata?.senderName === "string" ? metadata.senderName.trim() : "";
if (senderName.length > 0) {
return senderName;
}
const senderAgentId = typeof metadata?.senderAgentId === "string" ? metadata.senderAgentId.trim() : "";
return senderAgentId.length > 0 ? senderAgentId : "agent";
}
function resolveRoomLabel(metadata: NotificationPayload["metadata"] | undefined): string {
const roomName = typeof metadata?.roomName === "string" ? metadata.roomName.trim() : "";
if (roomName.length > 0) {
return roomName;
}
const roomId = typeof metadata?.roomId === "string" ? metadata.roomId.trim() : "";
return roomId.length > 0 ? roomId : "room";
}
export class NtfyNotificationProvider implements NotificationProvider {
private config?: NtfyProviderConfig;
private abortController: AbortController | null = null;
@@ -137,14 +157,25 @@ export class NtfyNotificationProvider implements NotificationProvider {
const replyToMessageId = typeof payload.metadata?.replyToMessageId === "string"
? payload.metadata.replyToMessageId
: undefined;
const roomSenderLabel = resolveRoomSenderLabel(payload.metadata);
const roomLabel = resolveRoomLabel(payload.metadata);
const roomId = typeof payload.metadata?.roomId === "string" ? payload.metadata.roomId : undefined;
const clickUrl = buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
taskId: payload.taskId,
messageId,
view: "mailbox",
});
const clickUrl = event === "message:room"
? buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
roomId,
messageId,
view: "rooms",
})
: buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
taskId: payload.taskId,
messageId,
view: "mailbox",
});
const contentByEvent: Record<SupportedNtfyEvent, { title: string; message: string; priority: "default" | "high" }> = {
"in-review": {
@@ -192,6 +223,11 @@ export class NtfyNotificationProvider implements NotificationProvider {
message: `${senderLabel} messaged ${recipientLabel}: ${preview}`,
priority: "default",
},
"message:room": {
title: `#${roomLabel}${roomSenderLabel}`,
message: `${roomSenderLabel} in #${roomLabel}: ${preview}`,
priority: "default",
},
};
const content = contentByEvent[event as SupportedNtfyEvent];

View File

@@ -34,6 +34,24 @@ function resolveParticipantLabel(
return id.length > 0 ? id : kind === "from" ? "agent" : "recipient";
}
function resolveRoomSenderLabel(metadata: NotificationPayload["metadata"] | undefined): string {
const senderName = typeof metadata?.senderName === "string" ? metadata.senderName.trim() : "";
if (senderName.length > 0) {
return senderName;
}
const senderAgentId = typeof metadata?.senderAgentId === "string" ? metadata.senderAgentId.trim() : "";
return senderAgentId.length > 0 ? senderAgentId : "agent";
}
function resolveRoomLabel(metadata: NotificationPayload["metadata"] | undefined): string {
const roomName = typeof metadata?.roomName === "string" ? metadata.roomName.trim() : "";
if (roomName.length > 0) {
return roomName;
}
const roomId = typeof metadata?.roomId === "string" ? metadata.roomId.trim() : "";
return roomId.length > 0 ? roomId : "room";
}
export class WebhookNotificationProvider implements NotificationProvider {
private config: WebhookProviderConfig | null = null;
private abortController: AbortController | null = null;
@@ -167,6 +185,12 @@ export class WebhookNotificationProvider implements NotificationProvider {
const preview = typeof payload.metadata?.preview === "string" ? payload.metadata.preview : "(no preview)";
return `From: ${from} → To: ${to}: ${preview}`;
}
case "message:room": {
const roomName = resolveRoomLabel(payload.metadata);
const senderLabel = resolveRoomSenderLabel(payload.metadata);
const preview = typeof payload.metadata?.preview === "string" ? payload.metadata.preview : "(no preview)";
return `In #${roomName}: ${senderLabel}: ${preview}`;
}
default:
return `Event "${event}" for task ${identifier}`;
}
@@ -196,9 +220,12 @@ export class WebhookNotificationProvider implements NotificationProvider {
}
const messageId = typeof payload.metadata?.messageId === "string" ? payload.metadata.messageId : undefined;
const roomId = typeof payload.metadata?.roomId === "string" ? payload.metadata.roomId : undefined;
const fromLabel = resolveParticipantLabel(payload.metadata, "from");
const toLabel = resolveParticipantLabel(payload.metadata, "to");
const roomLabel = resolveRoomLabel(payload.metadata);
const roomSenderLabel = resolveRoomSenderLabel(payload.metadata);
return {
event: payload.event,
@@ -215,14 +242,28 @@ export class WebhookNotificationProvider implements NotificationProvider {
toName: typeof payload.metadata?.toName === "string" ? payload.metadata.toName : toLabel,
}
: {}),
...(payload.event === "message:room"
? {
roomName: typeof payload.metadata?.roomName === "string" ? payload.metadata.roomName : roomLabel,
senderName: typeof payload.metadata?.senderName === "string" ? payload.metadata.senderName : roomSenderLabel,
}
: {}),
},
clickUrl: buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
taskId: payload.taskId,
messageId,
view: "mailbox",
}),
clickUrl: payload.event === "message:room"
? buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
roomId,
messageId,
view: "rooms",
})
: buildNtfyClickUrl({
dashboardHost: this.config.dashboardHost,
projectId: this.config.projectId,
taskId: payload.taskId,
messageId,
view: "mailbox",
}),
};
}
}

View File

@@ -31,6 +31,7 @@ export const DEFAULT_NTFY_EVENTS: readonly NtfyNotificationEvent[] = [
"fallback-used",
"message:agent-to-user",
"message:agent-to-agent",
"message:room",
] as const;
export interface NtfyNotificationConfigInput {
@@ -150,9 +151,10 @@ export function buildNtfyClickUrl(options: {
projectId?: string;
taskId?: string;
messageId?: string;
roomId?: string;
view?: string;
}): string | undefined {
const { dashboardHost, projectId, taskId, messageId, view } = options;
const { dashboardHost, projectId, taskId, messageId, roomId, view } = options;
if (!dashboardHost) {
return undefined;
}
@@ -165,6 +167,9 @@ export function buildNtfyClickUrl(options: {
}
if (taskId) {
queryParts.push(`task=${encodeURIComponent(taskId)}`);
} else if (roomId) {
queryParts.push(`view=${encodeURIComponent(view ?? "rooms")}`);
queryParts.push(`room=${encodeURIComponent(roomId)}`);
} else if (messageId) {
queryParts.push(`view=${encodeURIComponent(view ?? "mailbox")}`);
queryParts.push(`mailbox-message=${encodeURIComponent(messageId)}`);

View File

@@ -19,6 +19,7 @@ import { PrMonitor } from "./pr-monitor.js";
import { PrCommentHandler } from "./pr-comment-handler.js";
import { NtfyNotifier } from "./notifier.js";
import { NotificationService } from "./notification/index.js";
import type { NotificationChatStore } from "./notification/notification-service.js";
import { GridlockDetector } from "./gridlock-detector.js";
import { CronRunner, createAiPromptExecutor } from "./cron-runner.js";
import type { RoutineRunner } from "./routine-runner.js";
@@ -601,6 +602,10 @@ export class ProjectEngine {
return this.runtime.getMessageStore();
}
attachChatStore(chatStore: NotificationChatStore): void {
this.notificationService?.attachChatStore(chatStore);
}
/** Get the HeartbeatMonitor (if initialized). */
getHeartbeatMonitor() {
return this.runtime.getHeartbeatMonitor();