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

feat(FN-3484): normalize dashboard mailbox and user identity for messaging

Browse Source 
This merge normalizes dashboard user identity and mailbox messaging (FN-3484, 4 steps), adds workflow step execution for plugins (FN-3490), and updates the restart integration store mock for plugin templates (FN-3096). Core changes touch the message store and store modules with identity normalizatio

Fusion-Task-Id: FN-3484
This commit is contained in:
Fusion
2026-05-05 06:27:05 -07:00
committed by gsxdsm
parent e91e937478
commit 47ae7783b7
11 changed files with 248 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
packages/core/src/__tests__/message-store.test.ts
View File

@@ -4,6 +4,7 @@ import { join } from "node:path";
import { tmpdir } from "node:os";
import { Database } from "../db.js";
import { MessageStore } from "../message-store.js";
import { DASHBOARD_USER_ID } from "../types.js";
import type { Message, Mailbox } from "../types.js";
describe("MessageStore", () => {
@@ -123,6 +124,40 @@ describe("MessageStore", () => {
const result = store.getMessage("msg-nonexistent");
expect(result).toBeNull();
});
it.each(["dashboard", "user:dashboard", "User: user:dashboard"])(
"canonicalizes dashboard user alias '%s' when writing recipient",
(dashboardAlias) => {
const message = store.sendMessage({
fromId: "agent-1",
fromType: "agent",
toId: dashboardAlias,
toType: "user",
content: "Hello dashboard",
type: "agent-to-user",
});
expect(message.toId).toBe(DASHBOARD_USER_ID);
expect(store.getMessage(message.id)?.toId).toBe(DASHBOARD_USER_ID);
},
);
it.each(["dashboard", "user:dashboard", "User: user:dashboard"])(
"canonicalizes dashboard user alias '%s' when writing sender",
(dashboardAlias) => {
const message = store.sendMessage({
fromId: dashboardAlias,
fromType: "user",
toId: "agent-1",
toType: "agent",
content: "Reply",
type: "user-to-agent",
});
expect(message.fromId).toBe(DASHBOARD_USER_ID);
expect(store.getMessage(message.id)?.fromId).toBe(DASHBOARD_USER_ID);
},
);
});
describe("message-to-agent hook", () => {
@@ -234,6 +269,15 @@ describe("MessageStore", () => {
expect(inbox).toEqual([]);
});
it("includes legacy dashboard aliases in canonical dashboard inbox reads", () => {
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: DASHBOARD_USER_ID, toType: "user", content: "A", type: "agent-to-user" });
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: "user:dashboard", toType: "user", content: "B", type: "agent-to-user" });
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: "User: user:dashboard", toType: "user", content: "C", type: "agent-to-user" });
const inbox = store.getInbox(DASHBOARD_USER_ID, "user");
expect(inbox).toHaveLength(3);
});
it("filters by read status", () => {
const msg1 = store.sendMessage({
fromId: "agent-1",
@@ -418,6 +462,14 @@ describe("MessageStore", () => {
const count = store.markAllAsRead("user-99", "user");
expect(count).toBe(0);
});
it("marks canonical dashboard aliases as read together", () => {
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: DASHBOARD_USER_ID, toType: "user", content: "A", type: "agent-to-user" });
store.sendMessage({ fromId: "agent-2", fromType: "agent", toId: "user:dashboard", toType: "user", content: "B", type: "agent-to-user" });
const marked = store.markAllAsRead(DASHBOARD_USER_ID, "user");
expect(marked).toBe(2);
expect(store.getMailbox(DASHBOARD_USER_ID, "user").unreadCount).toBe(0);
});
});
describe("deleteMessage()", () => {
@@ -524,6 +576,31 @@ describe("MessageStore", () => {
);
expect(conversation).toEqual([]);
});
it("treats canonical dashboard identity as equivalent to legacy aliases in conversation reads", () => {
const sent = store.sendMessage({
fromId: "dashboard",
fromType: "user",
toId: "agent-1",
toType: "agent",
content: "Question",
type: "user-to-agent",
});
const reply = store.sendMessage({
fromId: "agent-1",
fromType: "agent",
toId: "user:dashboard",
toType: "user",
content: "Answer",
type: "agent-to-user",
});
const conversation = store.getConversation(
{ id: DASHBOARD_USER_ID, type: "user" },
{ id: "agent-1", type: "agent" },
);
expect(conversation.map((message) => message.id)).toEqual([sent.id, reply.id]);
});
});
describe("getMailbox()", () => {
@@ -561,6 +638,15 @@ describe("MessageStore", () => {
expect(mailbox.lastMessage).toBeUndefined();
});
it("aggregates unread count across canonical and legacy dashboard aliases", () => {
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: DASHBOARD_USER_ID, toType: "user", content: "A", type: "agent-to-user" });
store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: "User: user:dashboard", toType: "user", content: "B", type: "agent-to-user" });
const mailbox = store.getMailbox(DASHBOARD_USER_ID, "user");
expect(mailbox.unreadCount).toBe(2);
expect(mailbox.lastMessage).toBeTruthy();
});
it("counts only unread messages", () => {
const msg1 = store.sendMessage({
fromId: "agent-1",

2
packages/core/src/index.ts
View File

@@ -1,4 +1,4 @@
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeConflictStrategy, buildResearchDocumentKey } from "./types.js";
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeConflictStrategy, buildResearchDocumentKey } from "./types.js";
export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrStatus, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, TaskSource, SourceType, TaskDetail, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, MergeConflictStrategy, CanonicalMergeConflictStrategy, Settings, GlobalSettings, ProjectSettings, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, UnavailableNodePolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter } from "./types.js";
export { AGENT_VALID_TRANSITIONS } from "./types.js";
export * from "./mesh-replication-protocol.js";

97
packages/core/src/message-store.ts
View File

@@ -14,7 +14,7 @@ import { EventEmitter } from "node:events";
import { randomUUID } from "node:crypto";
import type { Database } from "./db.js";
import { fromJson, toJsonNullable } from "./db.js";
import { validateMessageMetadata, type Message, type MessageCreateInput, type MessageFilter, type MessageType, type Mailbox, type ParticipantType } from "./types.js";
import { DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, type Message, type MessageCreateInput, type MessageFilter, type MessageType, type Mailbox, type ParticipantType } from "./types.js";
// ── Event Types ─────────────────────────────────────────────────────
@@ -69,8 +69,6 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
private stmtGetById!: ReturnType<Database["prepare"]>;
private stmtUpdateRead!: ReturnType<Database["prepare"]>;
private stmtDelete!: ReturnType<Database["prepare"]>;
private stmtCountUnread!: ReturnType<Database["prepare"]>;
private stmtGetLastMessage!: ReturnType<Database["prepare"]>;
constructor(
private db: Database,
@@ -97,14 +95,6 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
this.stmtDelete = this.db.prepare(`
DELETE FROM messages WHERE id = ?
`);
this.stmtCountUnread = this.db.prepare(`
SELECT COUNT(*) as count FROM messages WHERE toId = ? AND toType = ? AND read = 0
`);
this.stmtGetLastMessage = this.db.prepare(`
SELECT * FROM messages WHERE toId = ? AND toType = ? ORDER BY createdAt DESC, rowid DESC LIMIT 1
`);
}
// ── Row-to-Object Converters ───────────────────────────────────────
@@ -141,15 +131,15 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
const now = new Date().toISOString();
const messageId = `msg-${randomUUID().slice(0, 8)}`;
const fromId = input.fromId ?? "system";
const fromType = input.fromType ?? "system";
const from = normalizeMessageParticipant(input.fromId ?? "system", input.fromType ?? "system");
const to = normalizeMessageParticipant(input.toId, input.toType);
const message: Message = {
id: messageId,
fromId,
fromType,
toId: input.toId,
toType: input.toType,
fromId: from.id,
fromType: from.type,
toId: to.id,
toType: to.type,
content: input.content,
type: input.type,
read: false,
@@ -224,6 +214,13 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
return this.queryMessagesByParticipant("from", ownerId, ownerType, filter);
}
private getParticipantIdsForLookup(ownerId: string, ownerType: ParticipantType): string[] {
if (ownerType === "user" && ownerId === DASHBOARD_USER_ID) {
return [DASHBOARD_USER_ID, "user:dashboard", "User: user:dashboard"];
}
return [ownerId];
}
private queryMessagesByParticipant(
direction: "to" | "from",
ownerId: string,
@@ -232,8 +229,12 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
): Message[] {
const idCol = direction === "to" ? "toId" : "fromId";
const typeCol = direction === "to" ? "toType" : "fromType";
const whereClauses: string[] = [`${idCol} = ?`, `${typeCol} = ?`];
const params: (string | number)[] = [ownerId, ownerType];
const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType);
const idPredicate = participantIds.length === 1
? `${idCol} = ?`
: `${idCol} IN (${participantIds.map(() => "?").join(", ")})`;
const whereClauses: string[] = [idPredicate, `${typeCol} = ?`];
const params: (string | number)[] = [...participantIds, ownerType];
if (filter?.type) {
whereClauses.push("type = ?");
@@ -294,16 +295,21 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
ownerType: ParticipantType,
): number {
const now = new Date().toISOString();
const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType);
const toIdPredicate = participantIds.length === 1
? "toId = ?"
: `toId IN (${participantIds.map(() => "?").join(", ")})`;
// Get count of unread messages before updating
const unreadRow = this.db.prepare(`
SELECT COUNT(*) as count FROM messages WHERE toId = ? AND toType = ? AND read = 0
`).get(ownerId, ownerType) as { count: number } | undefined;
SELECT COUNT(*) as count FROM messages WHERE ${toIdPredicate} AND toType = ? AND read = 0
`).get(...participantIds, ownerType) as { count: number } | undefined;
const count = unreadRow?.count ?? 0;
// Mark all as read
this.db.prepare(`
UPDATE messages SET read = 1, updatedAt = ? WHERE toId = ? AND toType = ? AND read = 0
`).run(now, ownerId, ownerType);
UPDATE messages SET read = 1, updatedAt = ? WHERE ${toIdPredicate} AND toType = ? AND read = 0
`).run(now, ...participantIds, ownerType);
this.db.bumpLastModified();
return count;
@@ -336,21 +342,39 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
participantA: { id: string; type: ParticipantType },
participantB: { id: string; type: ParticipantType },
): Message[] {
const participantAIds = this.getParticipantIdsForLookup(participantA.id, participantA.type);
const participantBIds = this.getParticipantIdsForLookup(participantB.id, participantB.type);
const participantAFromPredicate = participantAIds.length === 1
? "fromId = ?"
: `fromId IN (${participantAIds.map(() => "?").join(", ")})`;
const participantAToPredicate = participantAIds.length === 1
? "toId = ?"
: `toId IN (${participantAIds.map(() => "?").join(", ")})`;
const participantBFromPredicate = participantBIds.length === 1
? "fromId = ?"
: `fromId IN (${participantBIds.map(() => "?").join(", ")})`;
const participantBToPredicate = participantBIds.length === 1
? "toId = ?"
: `toId IN (${participantBIds.map(() => "?").join(", ")})`;
// Find messages where either participant is sender or receiver
// This captures all messages between the two participants
const rows = this.db.prepare(`
SELECT * FROM messages
WHERE (
(fromId = ? AND fromType = ? AND toId = ? AND toType = ?)
(${participantAFromPredicate} AND fromType = ? AND ${participantBToPredicate} AND toType = ?)
OR
(fromId = ? AND fromType = ? AND toId = ? AND toType = ?)
(${participantBFromPredicate} AND fromType = ? AND ${participantAToPredicate} AND toType = ?)
)
ORDER BY createdAt ASC
`).all(
participantA.id, participantA.type,
participantB.id, participantB.type,
participantB.id, participantB.type,
participantA.id, participantA.type,
...participantAIds,
participantA.type,
...participantBIds,
participantB.type,
...participantBIds,
participantB.type,
...participantAIds,
participantA.type,
);
return (rows as unknown as MessageRow[]).map((row) => this.rowToMessage(row));
@@ -366,10 +390,19 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
ownerId: string,
ownerType: ParticipantType,
): Mailbox {
const unreadRow = this.stmtCountUnread.get(ownerId, ownerType) as { count: number } | undefined;
const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType);
const toIdPredicate = participantIds.length === 1
? "toId = ?"
: `toId IN (${participantIds.map(() => "?").join(", ")})`;
const unreadRow = this.db.prepare(`
SELECT COUNT(*) as count FROM messages WHERE ${toIdPredicate} AND toType = ? AND read = 0
`).get(...participantIds, ownerType) as { count: number } | undefined;
const unreadCount = unreadRow?.count ?? 0;
const lastRow = this.stmtGetLastMessage.get(ownerId, ownerType) as unknown as MessageRow | undefined;
const lastRow = this.db.prepare(`
SELECT * FROM messages WHERE ${toIdPredicate} AND toType = ? ORDER BY createdAt DESC, rowid DESC LIMIT 1
`).get(...participantIds, ownerType) as unknown as MessageRow | undefined;
const lastMessage = lastRow ? this.rowToMessage(lastRow) : undefined;
return {

18
packages/core/src/types.ts
View File

@@ -4117,6 +4117,24 @@ export interface MigrationResult {
/** Participant types for message routing */
export type ParticipantType = "agent" | "user" | "system";
/** Canonical recipient ID for dashboard user mailbox routing. */
export const DASHBOARD_USER_ID = "dashboard";
const DASHBOARD_USER_ALIASES = new Set([DASHBOARD_USER_ID, "user:dashboard", "User: user:dashboard"]);
/** Normalize participant identity for durable mailbox routing. */
export function normalizeMessageParticipant(id: string, type: ParticipantType): { id: string; type: ParticipantType } {
if (type !== "user") {
return { id, type };
}
if (DASHBOARD_USER_ALIASES.has(id)) {
return { id: DASHBOARD_USER_ID, type };
}
return { id, type };
}
/** Message types/categories */
export type MessageType = "agent-to-agent" | "agent-to-user" | "user-to-agent" | "system";