feat(FN-1976): forward mailbox message events end to end

- Wire MessageStore into the in-process executor runtime and project engine
- Forward MessageStore events through dashboard SSE infrastructure for mailbox updates
- Close mailbox pipeline gaps across API routes, server wiring, and mailbox UI components
- Add regression coverage for messaging routes, SSE forwarding, and agent tool behavior
- Document the MessageStore SSE wiring pattern in .fusion/memory.md

.fusion/memory.md

@@ -18,6 +18,11 @@
   - `useAgents` hook excludes ephemeral agents from `activeAgents` by default
   - Server SSE endpoint resolves `AgentStore` from engine via `getAgentStore()` for project-scoped streams
 
+- **`FN-1976 Message SSE + Route Store Cohesion`**:
+  - `MessageStore` is an `EventEmitter`; SSE listeners must attach to the SAME `MessageStore` instance used by message-writing routes.
+  - Reusing the same SQLite database is not enough for realtime updates — two `MessageStore` instances on one DB do not share in-memory events.
+  - In dashboard routes, prefer `engine.getMessageStore()` (or `options.engine?.getMessageStore()` for default scope) before creating a fallback `new MessageStore(db)`.
+
 - **`FN-1736 Multi-Project Scoping Audit`**: Comprehensive audit of project-scoping across the Fusion stack found:
   - SSE/WebSocket endpoints (`/api/tasks/:id/logs/stream`, `/api/events`, `/api/ws`) already use `resolveProjectScopedStore()` or `getProjectContext()` correctly
   - Badge WebSocket (`setupBadgeWebSocket`) properly scopes per-project with listeners on scoped stores

packages/dashboard/app/components/MailboxModal.tsx

@@ -161,6 +161,42 @@ export function MailboxModal({
     if (isOpen) refreshUnreadCount();
   }, [isOpen, refreshUnreadCount]);
 
+  // Subscribe to mailbox SSE events while the modal is open.
+  useEffect(() => {
+    if (!isOpen || typeof EventSource === "undefined") {
+      return;
+    }
+
+    const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
+    const eventSource = new EventSource(`/api/events${query}`);
+
+    const onMailboxUpdate = () => {
+      void refreshUnreadCount();
+      if (activeTab === "inbox") {
+        void loadInbox();
+      } else if (activeTab === "outbox") {
+        void loadOutbox();
+      }
+
+      if (selectedAgentId) {
+        void loadAgentMailbox(selectedAgentId);
+      }
+    };
+
+    eventSource.addEventListener("message:sent", onMailboxUpdate);
+    eventSource.addEventListener("message:received", onMailboxUpdate);
+    eventSource.addEventListener("message:read", onMailboxUpdate);
+    eventSource.addEventListener("message:deleted", onMailboxUpdate);
+
+    return () => {
+      eventSource.removeEventListener("message:sent", onMailboxUpdate);
+      eventSource.removeEventListener("message:received", onMailboxUpdate);
+      eventSource.removeEventListener("message:read", onMailboxUpdate);
+      eventSource.removeEventListener("message:deleted", onMailboxUpdate);
+      eventSource.close();
+    };
+  }, [isOpen, projectId, activeTab, selectedAgentId, refreshUnreadCount, loadInbox, loadOutbox, loadAgentMailbox]);
+
   // ── Actions ───────────────────────────────────────────────────────────
 
   const handleOpenMessage = useCallback(async (message: Message) => {

packages/dashboard/app/components/MailboxView.tsx

@@ -225,6 +225,42 @@ export function MailboxView({
     loadAgents();
   }, [loadAgents]);
 
+  // Subscribe to mailbox SSE events for near-real-time refresh.
+  useEffect(() => {
+    if (typeof EventSource === "undefined") {
+      return;
+    }
+
+    const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
+    const eventSource = new EventSource(`/api/events${query}`);
+
+    const onMailboxUpdate = () => {
+      void refreshUnreadCount();
+      if (activeTab === "inbox") {
+        void loadInbox();
+      } else if (activeTab === "outbox") {
+        void loadOutbox();
+      }
+
+      if (selectedAgentId) {
+        void loadAgentMailbox(selectedAgentId);
+      }
+    };
+
+    eventSource.addEventListener("message:sent", onMailboxUpdate);
+    eventSource.addEventListener("message:received", onMailboxUpdate);
+    eventSource.addEventListener("message:read", onMailboxUpdate);
+    eventSource.addEventListener("message:deleted", onMailboxUpdate);
+
+    return () => {
+      eventSource.removeEventListener("message:sent", onMailboxUpdate);
+      eventSource.removeEventListener("message:received", onMailboxUpdate);
+      eventSource.removeEventListener("message:read", onMailboxUpdate);
+      eventSource.removeEventListener("message:deleted", onMailboxUpdate);
+      eventSource.close();
+    };
+  }, [projectId, activeTab, selectedAgentId, refreshUnreadCount, loadInbox, loadOutbox, loadAgentMailbox]);
+
   // ── Actions ───────────────────────────────────────────────────────────
 
   const handleOpenMessage = useCallback(async (message: Message) => {

packages/dashboard/app/components/MissionManager.tsx

@@ -568,6 +568,10 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
   const [activeTab, setActiveTab] = useState<"structure" | "activity">("structure");
   const [missionEvents, setMissionEvents] = useState<MissionEvent[]>([]);
   const missionEventsRef = useRef<MissionEvent[]>([]);
+  const missionsRef = useRef<MissionWithSummary[]>([]);
+  const selectedMissionRef = useRef<MissionWithHierarchy | null>(null);
+  const activeTabRef = useRef<"structure" | "activity">("structure");
+  const eventsFilterRef = useRef<"all" | "errors" | "state_changes" | "tasks" | "slices" | "autopilot">("all");
   const [eventsLoading, setEventsLoading] = useState(false);
   const [eventsTotal, setEventsTotal] = useState(0);
   const [eventsFilter, setEventsFilter] = useState<
@@ -578,6 +582,12 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
   const activityEventsContainerRef = useRef<HTMLDivElement>(null);
   const activityEventsEndRef = useRef<HTMLDivElement>(null);
 
+  // Keep latest state available to long-lived SSE handlers without reconnect churn.
+  missionsRef.current = missions;
+  selectedMissionRef.current = selectedMission;
+  activeTabRef.current = activeTab;
+  eventsFilterRef.current = eventsFilter;
+
   const scrollActivityToLatest = useCallback((behavior: ScrollBehavior = "auto") => {
     const endNode = activityEventsEndRef.current;
     if (endNode && typeof endNode.scrollIntoView === "function") {
@@ -766,7 +776,7 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
   }, [activeTab, isActive, loadMissionEvents, selectedMission, eventsFilter]);
 
   useEffect(() => {
-    if (!isActive || missions.length === 0 || typeof EventSource === "undefined") {
+    if (!isActive || typeof EventSource === "undefined") {
       return;
     }
 
@@ -778,7 +788,7 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
     const eventSource = new EventSource(eventUrl);
 
     const refreshHealth = () => {
-      void loadMissionHealth(missions);
+      void loadMissionHealth(missionsRef.current);
     };
 
     const handleMissionUpdated = (rawEvent: Event) => {
@@ -802,32 +812,32 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
       }
 
       // Reload the selected mission detail to reflect updated mission state (autopilot, status, etc.)
-      if (selectedMission) {
+      if (selectedMissionRef.current) {
-        void loadMissionDetail(selectedMission.id);
+        void loadMissionDetail(selectedMissionRef.current.id);
       }
     };
 
     const handleSliceUpdated = (rawEvent: Event) => {
       refreshHealth();
       // Reload the selected mission detail to reflect updated slice status
-      if (selectedMission) {
+      if (selectedMissionRef.current) {
-        void loadMissionDetail(selectedMission.id);
+        void loadMissionDetail(selectedMissionRef.current.id);
       }
     };
 
     const handleFeatureUpdated = () => {
       refreshHealth();
       // Reload the selected mission detail to reflect updated feature status
-      if (selectedMission) {
+      if (selectedMissionRef.current) {
-        void loadMissionDetail(selectedMission.id);
+        void loadMissionDetail(selectedMissionRef.current.id);
       }
     };
 
     const handleMilestoneUpdated = (_rawEvent: Event) => {
       refreshHealth();
       // Reload the selected mission detail to reflect updated milestone status
-      if (selectedMission) {
+      if (selectedMissionRef.current) {
-        void loadMissionDetail(selectedMission.id);
+        void loadMissionDetail(selectedMissionRef.current.id);
       }
     };
 
@@ -860,8 +870,8 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
           // Refresh validation runs
           void loadValidationRuns(payload.featureId);
           // Refresh mission detail to update feature status
-          if (selectedMission) {
+          if (selectedMissionRef.current) {
-            void loadMissionDetail(selectedMission.id);
+            void loadMissionDetail(selectedMissionRef.current.id);
           }
         }
       } catch {
@@ -908,8 +918,8 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
           // Refresh feature loop state for the source feature
           void loadFeatureLoopState(payload.sourceFeatureId);
           // Refresh mission detail to show the new fix feature in the list
-          if (selectedMission) {
+          if (selectedMissionRef.current) {
-            void loadMissionDetail(selectedMission.id);
+            void loadMissionDetail(selectedMissionRef.current.id);
           }
         }
       } catch {
@@ -920,7 +930,8 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
     const handleMissionEvent = (rawEvent: Event) => {
       refreshHealth();
 
-      if (!selectedMission || activeTab !== "activity") {
+      const currentSelectedMission = selectedMissionRef.current;
+      if (!currentSelectedMission || activeTabRef.current !== "activity") {
         return;
       }
 
@@ -935,10 +946,10 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
         if (!isMissionEvent(payload)) {
           return;
         }
-        if (payload.missionId !== selectedMission.id) {
+        if (payload.missionId !== currentSelectedMission.id) {
           return;
         }
-        if (!matchesEventFilter(payload.eventType, eventsFilter)) {
+        if (!matchesEventFilter(payload.eventType, eventsFilterRef.current)) {
           return;
         }
 
@@ -995,16 +1006,11 @@ export function MissionManager({ isOpen, isInline = false, onClose, addToast, pr
       eventSource.close();
     };
   }, [
-    activeTab,
-    eventsFilter,
     isActive,
     isActivityScrolledNearBottom,
     loadMissionDetail,
     loadMissionHealth,
-    missions,
     projectId,
-    scrollActivityToLatest,
-    selectedMission,
   ]);
 
   // Mission handlers

packages/dashboard/src/__tests__/sse.test.ts

@@ -73,6 +73,30 @@ function createMockPlugin(overrides: Partial<{
   };
 }
 
+function createMockMessage(overrides: Partial<{
+  id: string;
+  fromId: string;
+  fromType: string;
+  toId: string;
+  toType: string;
+  content: string;
+  type: string;
+  read: boolean;
+}> = {}) {
+  return {
+    id: overrides.id ?? "msg-123",
+    fromId: overrides.fromId ?? "dashboard",
+    fromType: overrides.fromType ?? "user",
+    toId: overrides.toId ?? "agent-1",
+    toType: overrides.toType ?? "agent",
+    content: overrides.content ?? "hello",
+    type: overrides.type ?? "user-to-agent",
+    read: overrides.read ?? false,
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  };
+}
+
 describe("createSSE", () => {
   let store: ReturnType<typeof createMockStore>;
 
@@ -280,6 +304,55 @@ describe("createSSE", () => {
     expect(getActiveSSEConnections()).toBe(initial);
   });
 
+  describe("message events", () => {
+    it("relays message lifecycle events when messageStore is provided", () => {
+      const messageStore = createMockStore();
+      const req = createMockRequest();
+      const { res, chunks } = createMockResponse();
+      createSSE(store, undefined, undefined, undefined, undefined, undefined, messageStore)(req, res);
+
+      const sentMessage = createMockMessage();
+      messageStore.emit("message:sent", sentMessage);
+      messageStore.emit("message:received", sentMessage);
+      messageStore.emit("message:read", { ...sentMessage, read: true });
+      messageStore.emit("message:deleted", sentMessage.id);
+
+      const sentEvent = chunks.find((c) => c.includes("event: message:sent"));
+      const receivedEvent = chunks.find((c) => c.includes("event: message:received"));
+      const readEvent = chunks.find((c) => c.includes("event: message:read"));
+      const deletedEvent = chunks.find((c) => c.includes("event: message:deleted"));
+
+      expect(sentEvent).toBeDefined();
+      expect(receivedEvent).toBeDefined();
+      expect(readEvent).toBeDefined();
+      expect(deletedEvent).toBeDefined();
+
+      expect(extractSSEPayload(sentEvent!).id).toBe(sentMessage.id);
+      expect(extractSSEPayload(receivedEvent!).id).toBe(sentMessage.id);
+      expect(extractSSEPayload(readEvent!).read).toBe(true);
+      expect(extractSSEPayload(deletedEvent!).id).toBe(sentMessage.id);
+    });
+
+    it("cleans up message listeners on disconnect", () => {
+      const messageStore = createMockStore();
+      const req = createMockRequest();
+      const { res } = createMockResponse();
+      createSSE(store, undefined, undefined, undefined, undefined, undefined, messageStore)(req, res);
+
+      expect(messageStore.listenerCount("message:sent")).toBe(1);
+      expect(messageStore.listenerCount("message:received")).toBe(1);
+      expect(messageStore.listenerCount("message:read")).toBe(1);
+      expect(messageStore.listenerCount("message:deleted")).toBe(1);
+
+      req.emit("close");
+
+      expect(messageStore.listenerCount("message:sent")).toBe(0);
+      expect(messageStore.listenerCount("message:received")).toBe(0);
+      expect(messageStore.listenerCount("message:read")).toBe(0);
+      expect(messageStore.listenerCount("message:deleted")).toBe(0);
+    });
+  });
+
   // ── Plugin Lifecycle Event Tests ─────────────────────────────────────────────
 
   describe("plugin lifecycle events", () => {

packages/dashboard/src/routes.test.ts

@@ -15282,5 +15282,328 @@ describe("POST /api/ai/refine-text with projectId scoping", () => {
   });
 });
 
+describe("Messaging Routes", () => {
+  let rootDir: string;
+  let store: TaskStore;
+  let app: express.Express;
+  let messageStore: import("@fusion/core").MessageStore;
+
+  beforeEach(async () => {
+    rootDir = mkdtempSync(join(tmpdir(), "kb-message-routes-"));
+    const { TaskStore, MessageStore } = await import("@fusion/core");
+
+    store = new TaskStore(rootDir);
+    await store.init();
+    messageStore = new MessageStore(store.getDatabase());
+
+    app = express();
+    app.use(express.json());
+    app.use("/api", createApiRoutes(store));
+  });
+
+  afterEach(() => {
+    rmSync(rootDir, { recursive: true, force: true });
+  });
+
+  it("uses the engine MessageStore when available", async () => {
+    const message = {
+      id: "msg-runtime-1",
+      fromId: "dashboard",
+      fromType: "user",
+      toId: "agent-1",
+      toType: "agent",
+      content: "runtime store message",
+      type: "user-to-agent",
+      read: false,
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    };
+
+    const runtimeMessageStore = {
+      sendMessage: vi.fn().mockReturnValue(message),
+    };
+    const runtimeEngine = {
+      getMessageStore: vi.fn().mockReturnValue(runtimeMessageStore),
+    };
+
+    const runtimeApp = express();
+    runtimeApp.use(express.json());
+    runtimeApp.use("/api", createApiRoutes(store, { engine: runtimeEngine as any }));
+
+    const res = await REQUEST(
+      runtimeApp,
+      "POST",
+      "/api/messages",
+      JSON.stringify({
+        toId: "agent-1",
+        toType: "agent",
+        content: "runtime store message",
+        type: "user-to-agent",
+      }),
+      { "Content-Type": "application/json" },
+    );
+
+    expect(res.status).toBe(201);
+    expect(runtimeEngine.getMessageStore).toHaveBeenCalled();
+    expect(runtimeMessageStore.sendMessage).toHaveBeenCalledWith({
+      fromId: "dashboard",
+      fromType: "user",
+      toId: "agent-1",
+      toType: "agent",
+      content: "runtime store message",
+      type: "user-to-agent",
+      metadata: undefined,
+    });
+    expect(res.body.id).toBe("msg-runtime-1");
+  });
+
+  it("GET /api/messages/inbox returns dashboard inbox messages", async () => {
+    const inboxMessage = messageStore.sendMessage({
+      fromId: "agent-1",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Hello dashboard",
+      type: "agent-to-user",
+    });
+    messageStore.sendMessage({
+      fromId: "agent-2",
+      fromType: "agent",
+      toId: "someone-else",
+      toType: "user",
+      content: "not for dashboard",
+      type: "agent-to-user",
+    });
+
+    const res = await GET(app, "/api/messages/inbox");
+
+    expect(res.status).toBe(200);
+    expect(res.body.messages).toHaveLength(1);
+    expect(res.body.messages[0].id).toBe(inboxMessage.id);
+    expect(res.body.unreadCount).toBe(1);
+  });
+
+  it("GET /api/messages/outbox returns dashboard sent messages", async () => {
+    const sent = await REQUEST(
+      app,
+      "POST",
+      "/api/messages",
+      JSON.stringify({
+        toId: "agent-7",
+        toType: "agent",
+        content: "Can you review this?",
+        type: "user-to-agent",
+      }),
+      { "Content-Type": "application/json" },
+    );
+    expect(sent.status).toBe(201);
+
+    const res = await GET(app, "/api/messages/outbox");
+
+    expect(res.status).toBe(200);
+    expect(res.body.messages).toHaveLength(1);
+    expect(res.body.messages[0].id).toBe(sent.body.id);
+    expect(res.body.messages[0].fromId).toBe("dashboard");
+  });
+
+  it("GET /api/messages/unread-count returns the unread count", async () => {
+    const unread = messageStore.sendMessage({
+      fromId: "agent-1",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Unread",
+      type: "agent-to-user",
+    });
+    const read = messageStore.sendMessage({
+      fromId: "agent-2",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Read",
+      type: "agent-to-user",
+    });
+    messageStore.markAsRead(read.id);
+
+    const res = await GET(app, "/api/messages/unread-count");
+
+    expect(unread).toBeDefined();
+    expect(res.status).toBe(200);
+    expect(res.body.unreadCount).toBe(1);
+  });
+
+  it("POST /api/messages validates required fields and creates messages", async () => {
+    const created = await REQUEST(
+      app,
+      "POST",
+      "/api/messages",
+      JSON.stringify({
+        toId: "agent-3",
+        toType: "agent",
+        content: "Need your help",
+        type: "user-to-agent",
+      }),
+      { "Content-Type": "application/json" },
+    );
+
+    expect(created.status).toBe(201);
+    expect(created.body.toId).toBe("agent-3");
+    expect(created.body.fromId).toBe("dashboard");
+
+    const invalidCases = [
+      { body: { toType: "agent", content: "x", type: "user-to-agent" }, message: "toId is required" },
+      { body: { toId: "agent-1", content: "x", type: "user-to-agent", toType: "bad" }, message: "toType must be one of" },
+      { body: { toId: "agent-1", toType: "agent", content: "", type: "user-to-agent" }, message: "content is required" },
+      { body: { toId: "agent-1", toType: "agent", content: "a".repeat(2001), type: "user-to-agent" }, message: "content is required" },
+      { body: { toId: "agent-1", toType: "agent", content: "x", type: "bad-type" }, message: "type must be one of" },
+    ];
+
+    for (const testCase of invalidCases) {
+      const res = await REQUEST(
+        app,
+        "POST",
+        "/api/messages",
+        JSON.stringify(testCase.body),
+        { "Content-Type": "application/json" },
+      );
+      expect(res.status).toBe(400);
+      expect(String(res.body.error)).toContain(testCase.message);
+    }
+  });
+
+  it("GET /api/messages/:id returns a message and 404 when missing", async () => {
+    const msg = messageStore.sendMessage({
+      fromId: "agent-3",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Lookup me",
+      type: "agent-to-user",
+    });
+
+    const found = await GET(app, `/api/messages/${msg.id}`);
+    const missing = await GET(app, "/api/messages/msg-missing");
+
+    expect(found.status).toBe(200);
+    expect(found.body.id).toBe(msg.id);
+    expect(missing.status).toBe(404);
+  });
+
+  it("POST /api/messages/:id/read marks the message as read", async () => {
+    const msg = messageStore.sendMessage({
+      fromId: "agent-4",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Mark me as read",
+      type: "agent-to-user",
+    });
+
+    const res = await REQUEST(app, "POST", `/api/messages/${msg.id}/read`);
+
+    expect(res.status).toBe(200);
+    expect(res.body.read).toBe(true);
+    expect(messageStore.getMessage(msg.id)?.read).toBe(true);
+  });
+
+  it("DELETE /api/messages/:id deletes the message", async () => {
+    const msg = messageStore.sendMessage({
+      fromId: "agent-5",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Delete me",
+      type: "agent-to-user",
+    });
+
+    const res = await REQUEST(app, "DELETE", `/api/messages/${msg.id}`);
+
+    expect(res.status).toBe(204);
+    expect(messageStore.getMessage(msg.id)).toBeNull();
+  });
+
+  it("POST /api/messages/read-all marks all dashboard inbox messages as read", async () => {
+    messageStore.sendMessage({
+      fromId: "agent-1",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "1",
+      type: "agent-to-user",
+    });
+    messageStore.sendMessage({
+      fromId: "agent-2",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "2",
+      type: "agent-to-user",
+    });
+
+    const res = await REQUEST(app, "POST", "/api/messages/read-all");
+    const unread = await GET(app, "/api/messages/unread-count");
+
+    expect(res.status).toBe(200);
+    expect(res.body.markedAsRead).toBe(2);
+    expect(unread.body.unreadCount).toBe(0);
+  });
+
+  it("GET /api/messages/conversation/:participantType/:participantId returns the conversation thread", async () => {
+    const outbound = messageStore.sendMessage({
+      fromId: "dashboard",
+      fromType: "user",
+      toId: "agent-convo",
+      toType: "agent",
+      content: "Question",
+      type: "user-to-agent",
+    });
+    const inbound = messageStore.sendMessage({
+      fromId: "agent-convo",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Answer",
+      type: "agent-to-user",
+    });
+    messageStore.sendMessage({
+      fromId: "agent-other",
+      fromType: "agent",
+      toId: "dashboard",
+      toType: "user",
+      content: "Other thread",
+      type: "agent-to-user",
+    });
+
+    const res = await GET(app, "/api/messages/conversation/agent/agent-convo");
+
+    expect(res.status).toBe(200);
+    const ids = res.body.map((m: { id: string }) => m.id);
+    expect(ids).toContain(outbound.id);
+    expect(ids).toContain(inbound.id);
+    expect(ids).toHaveLength(2);
+  });
+
+  it("GET /api/agents/:id/mailbox returns mailbox summary and inbox messages", async () => {
+    const agentId = "agent-mailbox";
+    const msg = messageStore.sendMessage({
+      fromId: "dashboard",
+      fromType: "user",
+      toId: agentId,
+      toType: "agent",
+      content: "Ping",
+      type: "user-to-agent",
+    });
+
+    const res = await GET(app, `/api/agents/${agentId}/mailbox`);
+
+    expect(res.status).toBe(200);
+    expect(res.body.ownerId).toBe(agentId);
+    expect(res.body.ownerType).toBe("agent");
+    expect(res.body.unreadCount).toBe(1);
+    expect(res.body.messages).toHaveLength(1);
+    expect(res.body.messages[0].id).toBe(msg.id);
+  });
+});
+
 // Note: Project pause/resume route tests are in src/__tests__/project-pause-resume-routes.test.ts
 // to avoid test isolation issues with vi.restoreAllMocks() from other tests in routes.test.ts

packages/dashboard/src/routes.ts

@@ -16076,8 +16076,18 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
   const messageStoreCache = new Map<string, MessageStore>();
 
   async function getMessageStore(req: Request): Promise<MessageStore> {
-    const { store: scopedStore } = await getProjectContext(req);
+    const { store: scopedStore, engine, projectId } = await getProjectContext(req);
     const rootDir = scopedStore.getRootDir();
+
+    // Prefer the runtime's MessageStore when available so routes and SSE share
+    // the same EventEmitter instance (required for live mailbox updates).
+    const runtimeMessageStore = engine?.getMessageStore()
+      ?? (!projectId ? options?.engine?.getMessageStore() : undefined);
+    if (runtimeMessageStore) {
+      messageStoreCache.set(rootDir, runtimeMessageStore);
+      return runtimeMessageStore;
+    }
+
     let msgStore = messageStoreCache.get(rootDir);
     if (!msgStore) {
       const db = scopedStore.getDatabase();

packages/dashboard/src/server.ts

@@ -3,7 +3,7 @@ import { randomUUID } from "node:crypto";
 import { join, dirname } from "node:path";
 import { existsSync } from "node:fs";
 import { fileURLToPath } from "node:url";
-import type { Task, TaskStore, MergeResult, AutomationStore, RoutineStore, CentralCore } from "@fusion/core";
+import type { Task, TaskStore, MergeResult, AutomationStore, RoutineStore, CentralCore, MessageStore } from "@fusion/core";
 import { ChatStore } from "@fusion/core";
 import type { AuthStorageLike, ModelRegistryLike } from "./routes.js";
 import { createApiRoutes } from "./routes.js";
@@ -367,7 +367,16 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
       const { AgentStore: AgentStoreClass } = await import("@fusion/core");
       const defaultAgentStore = new AgentStoreClass({ rootDir: store.getFusionDir() });
       await defaultAgentStore.init();
-      createSSE(store, store.getMissionStore(), aiSessionStore, store.getPluginStore(), undefined, defaultAgentStore)(req, res);
+      const defaultMessageStore = options?.engine?.getMessageStore();
+      createSSE(
+        store,
+        store.getMissionStore(),
+        aiSessionStore,
+        store.getPluginStore(),
+        undefined,
+        defaultAgentStore,
+        defaultMessageStore,
+      )(req, res);
       return;
     }
 
@@ -377,11 +386,13 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
       // rather than a separate store created by getOrCreateProjectStore.
       let scopedStore: TaskStore;
       let agentStore;
+      let messageStore: MessageStore | undefined;
       if (engineManager) {
         const engine = engineManager.getEngine(projectId);
         scopedStore = engine?.getTaskStore() ?? await getOrCreateProjectStore(projectId);
-        // Use the engine's AgentStore if available
+        // Use the engine's stores if available
         agentStore = engine?.getAgentStore();
+        messageStore = engine?.getMessageStore();
       } else {
         scopedStore = await getOrCreateProjectStore(projectId);
       }
@@ -391,9 +402,17 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
         agentStore = new AgentStoreClass({ rootDir: scopedStore.getFusionDir() });
         await agentStore.init();
       }
-      createSSE(scopedStore, scopedStore.getMissionStore(), aiSessionStore, scopedStore.getPluginStore(), {
+      createSSE(
-        projectId,
+        scopedStore,
-      }, agentStore)(req, res);
+        scopedStore.getMissionStore(),
+        aiSessionStore,
+        scopedStore.getPluginStore(),
+        {
+          projectId,
+        },
+        agentStore,
+        messageStore,
+      )(req, res);
     } catch (err: unknown) {
       sendErrorResponse(res, 500, err instanceof Error ? err.message : "Failed to open project event stream");
     }

packages/dashboard/src/sse.ts

@@ -1,5 +1,5 @@
 import type { Request, Response } from "express";
-import type { TaskStore, MissionStore, PluginStore, PluginInstallation, PluginState, AgentStore } from "@fusion/core";
+import type { TaskStore, MissionStore, PluginStore, PluginInstallation, PluginState, AgentStore, MessageStore } from "@fusion/core";
 import type { AiSessionStore } from "./ai-session-store.js";
 
 let activeConnections = 0;
@@ -70,6 +70,13 @@ export type PluginLifecycleTransition =
   | "uninstalled"
   | "settings-updated";
 
+/** Message event types forwarded through the SSE stream. */
+export type MessageSseEventType =
+  | "message:sent"
+  | "message:received"
+  | "message:read"
+  | "message:deleted";
+
 /**
  * Normalized plugin lifecycle payload emitted via SSE.
  * This is the stable contract the UI can reconcile.
@@ -173,6 +180,7 @@ export function createSSE(
   pluginStore?: PluginStore,
   options?: CreateSSEOptions,
   agentStore?: AgentStore,
+  messageStore?: MessageStore,
 ) {
   const { projectId } = options ?? {};
 
@@ -342,6 +350,23 @@ export function createSSE(
       send(`event: agent:stateChanged\ndata: ${JSON.stringify({ id: agentId, from: fromState, to: toState })}\n\n`);
     };
 
+    // --- Message event handlers ---
+    const onMessageSent = (message: unknown) => {
+      send(`event: message:sent\ndata: ${JSON.stringify(message)}\n\n`);
+    };
+
+    const onMessageReceived = (message: unknown) => {
+      send(`event: message:received\ndata: ${JSON.stringify(message)}\n\n`);
+    };
+
+    const onMessageRead = (message: unknown) => {
+      send(`event: message:read\ndata: ${JSON.stringify(message)}\n\n`);
+    };
+
+    const onMessageDeleted = (messageId: string) => {
+      send(`event: message:deleted\ndata: ${JSON.stringify({ id: messageId })}\n\n`);
+    };
+
     // --- Cleanup (all handlers are defined above, safe to reference) ---
 
     let cleaned = false;
@@ -396,6 +421,12 @@ export function createSSE(
         agentStore.off("agent:deleted", onAgentDeleted);
         agentStore.off("agent:stateChanged", onAgentStateChanged);
       }
+      if (messageStore) {
+        messageStore.off("message:sent", onMessageSent);
+        messageStore.off("message:received", onMessageReceived);
+        messageStore.off("message:read", onMessageRead);
+        messageStore.off("message:deleted", onMessageDeleted);
+      }
     };
 
     // --- Subscribe ---
@@ -451,6 +482,13 @@ export function createSSE(
       agentStore.on("agent:stateChanged", onAgentStateChanged);
     }
 
+    if (messageStore) {
+      messageStore.on("message:sent", onMessageSent);
+      messageStore.on("message:received", onMessageReceived);
+      messageStore.on("message:read", onMessageRead);
+      messageStore.on("message:deleted", onMessageDeleted);
+    }
+
     // Heartbeat every 30s to keep connection alive.
     // Sent as a named event so the client's EventSource can detect it
     // (SSE comments starting with ":" are silently consumed and never

packages/engine/src/agent-tools.test.ts

@@ -104,8 +104,8 @@ describe("createSendMessageTool", () => {
     );
   });
 
-  it("uses provided type when specified", async () => {
+  it("uses provided type when specified and maps recipient type for agent-to-user", async () => {
-    const mockMessage = createMessage();
+    const mockMessage = createMessage({ toType: "user", type: "agent-to-user" });
     vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
 
     await executeTool(tool, {
@@ -115,7 +115,22 @@ describe("createSendMessageTool", () => {
     });
 
     expect(messageStore.sendMessage).toHaveBeenCalledWith(
-      expect.objectContaining({ type: "agent-to-user" })
+      expect.objectContaining({ type: "agent-to-user", toType: "user" })
+    );
+  });
+
+  it("maps recipient type to agent for agent-to-agent messages", async () => {
+    const mockMessage = createMessage({ toType: "agent", type: "agent-to-agent" });
+    vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
+
+    await executeTool(tool, {
+      to_id: "agent-2",
+      content: "Test",
+      type: "agent-to-agent",
+    });
+
+    expect(messageStore.sendMessage).toHaveBeenCalledWith(
+      expect.objectContaining({ type: "agent-to-agent", toType: "agent" })
     );
   });
 

packages/engine/src/agent-tools.ts

@@ -68,7 +68,7 @@ export const delegateTaskParams = Type.Object({
 });
 
 export const sendMessageParams = Type.Object({
-  to_id: Type.String({ description: "Recipient agent ID (e.g. 'agent-abc123')" }),
+  to_id: Type.String({ description: "Recipient ID (agent ID or user ID, depending on message type)" }),
   content: Type.String({ description: "Message body (1-2000 characters)" }),
   type: Type.Optional(Type.Union([
     Type.Literal("agent-to-agent"),
@@ -474,13 +474,16 @@ export function createSendMessageTool(messageStore: MessageStore, fromAgentId: s
       }
 
       try {
+        const messageType = params.type ?? "agent-to-agent";
+        const recipientType = messageType === "agent-to-user" ? "user" : "agent";
+
         const message = messageStore.sendMessage({
           fromId: fromAgentId,
           fromType: "agent",
           toId: params.to_id,
-          toType: "agent",
+          toType: recipientType,
           content,
-          type: params.type ?? "agent-to-agent",
+          type: messageType,
         });
 
         return {

packages/engine/src/project-engine.ts

@@ -268,6 +268,11 @@ export class ProjectEngine {
     return this.runtime.getAgentStore();
   }
 
+  /** Get the MessageStore (if initialized). Returns undefined before start(). */
+  getMessageStore(): import("@fusion/core").MessageStore | undefined {
+    return this.runtime.getMessageStore();
+  }
+
   /** Get the HeartbeatMonitor (if initialized). */
   getHeartbeatMonitor() {
     return this.runtime.getHeartbeatMonitor();

packages/engine/src/runtimes/in-process-runtime.ts

@@ -131,7 +131,12 @@ export class InProcessRuntime
 
     try {
       // 1. Initialize TaskStore (use external if provided, otherwise create new)
-      const { TaskStore, PluginStore: PluginStoreClass, PluginLoader: PluginLoaderClass } = await import("@fusion/core");
+      const {
+        TaskStore,
+        PluginStore: PluginStoreClass,
+        PluginLoader: PluginLoaderClass,
+        MessageStore: MessageStoreClass,
+      } = await import("@fusion/core");
       if (this.config.externalTaskStore) {
         this.taskStore = this.config.externalTaskStore;
         runtimeLog.log(`TaskStore provided externally for project ${this.config.projectId}`);
@@ -141,6 +146,9 @@ export class InProcessRuntime
         runtimeLog.log(`TaskStore initialized for project ${this.config.projectId}`);
       }
 
+      // Initialize MessageStore early so TaskExecutor receives send_message capability.
+      this.messageStore = new MessageStoreClass(this.taskStore.getDatabase());
+
       // 2. Initialize Plugin system (PluginStore + PluginLoader + PluginRunner)
       this.pluginStore = new PluginStoreClass(this.taskStore.getFusionDir());
       await this.pluginStore.init();
@@ -263,6 +271,7 @@ export class InProcessRuntime
         usageLimitPauser: this.usageLimitPauser,
         stuckTaskDetector: this.stuckTaskDetector,
         pluginRunner: this.pluginRunner,
+        messageStore: this.messageStore,
         missionStore,
         onSliceComplete: (slice) => {
           void this.scheduler.onSliceComplete(slice);
@@ -353,13 +362,10 @@ export class InProcessRuntime
 
       // 6. Initialize AgentStore and HeartbeatMonitor
       try {
-        const { AgentStore: AgentStoreClass, MessageStore: MessageStoreClass } = await import("@fusion/core");
+        const { AgentStore: AgentStoreClass } = await import("@fusion/core");
         this.agentStore = new AgentStoreClass({ rootDir: this.taskStore.getFusionDir() });
         await this.agentStore.init();
 
-        // Initialize MessageStore for wake-on-message behavior
-        this.messageStore = new MessageStoreClass(this.taskStore.getDatabase());
-
         this.heartbeatMonitor = new HeartbeatMonitor({
           store: this.agentStore,
           agentStore: this.agentStore, // enables per-agent config resolution
@@ -718,6 +724,14 @@ export class InProcessRuntime
     return this.agentStore;
   }
 
+  /**
+   * Get the MessageStore instance (if initialized).
+   * Returns undefined before start() or if initialization fails.
+   */
+  getMessageStore(): import("@fusion/core").MessageStore | undefined {
+    return this.messageStore;
+  }
+
   /**
    * Get the project's Scheduler instance.
    * @throws Error if runtime has not been started