fix(FN-758): fix project-scoped real-time dashboard updates

- Add project-store-resolver to correctly scope SSE and WebSocket connections to the active project's TaskStore
- Fix shared resolver so multi-project setups deliver task events only to the right dashboard clients
- Add regression tests for project store resolver (routing, missing project, SSE integration)
- Remove unused import and clean up stale changeset
- Add changeset for published package patch bump

packages/dashboard/src/__tests__/project-store-resolver.test.ts

@@ -0,0 +1,225 @@
+/**
+ * Regression tests for project-scoped real-time SSE delivery (FN-758).
+ *
+ * Validates that:
+ * 1. The shared project-store resolver reuses the same TaskStore instance
+ * 2. No duplicate watch() calls or listeners are stacked
+ * 3. Eviction properly cleans up
+ * 4. SSE streams receive live events from the same store instance used for mutations
+ */
+
+import { EventEmitter } from "node:events";
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import {
+  getOrCreateProjectStore,
+  evictProjectStore,
+  evictAllProjectStores,
+} from "../project-store-resolver.js";
+import type { TaskStore } from "@fusion/core";
+
+// Mock @fusion/core to control TaskStore.getOrCreateForProject
+const createdStores: Array<{
+  projectId: string;
+  store: TaskStore;
+  watchMock: ReturnType<typeof vi.fn>;
+  closeMock: ReturnType<typeof vi.fn>;
+  stopWatchingMock: ReturnType<typeof vi.fn>;
+}> = [];
+
+vi.mock("@fusion/core", async () => {
+  const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
+  return {
+    ...actual,
+    TaskStore: {
+      ...actual.TaskStore,
+      getOrCreateForProject: vi.fn(async (projectId: string): Promise<TaskStore> => {
+        const store = Object.create(EventEmitter.prototype) as TaskStore;
+        EventEmitter.call(store as any);
+        (store as any).setMaxListeners(100);
+
+        const watchMock = vi.fn(async () => {});
+        const closeMock = vi.fn(() => {});
+        const stopWatchingMock = vi.fn(() => {});
+
+        const entry = { projectId, store, watchMock, closeMock, stopWatchingMock };
+
+        // Minimal store interface
+        (store as any).init = vi.fn(async () => {});
+        (store as any).watch = watchMock;
+        (store as any).close = closeMock;
+        (store as any).stopWatching = stopWatchingMock;
+        (store as any).getMissionStore = vi.fn(() => ({
+          on: vi.fn(),
+          off: vi.fn(),
+        }));
+
+        createdStores.push(entry);
+        return store;
+      }),
+    },
+  };
+});
+
+describe("project-store-resolver", () => {
+  beforeEach(() => {
+    evictAllProjectStores();
+    createdStores.length = 0;
+  });
+
+  afterEach(() => {
+    evictAllProjectStores();
+    createdStores.length = 0;
+  });
+
+  it("creates a new store on first call for a project", async () => {
+    const store = await getOrCreateProjectStore("proj_abc");
+    expect(store).toBeDefined();
+    expect(createdStores).toHaveLength(1);
+    expect(createdStores[0].projectId).toBe("proj_abc");
+  });
+
+  it("reuses the same TaskStore instance for repeated calls with the same projectId", async () => {
+    const store1 = await getOrCreateProjectStore("proj_abc");
+    const store2 = await getOrCreateProjectStore("proj_abc");
+    const store3 = await getOrCreateProjectStore("proj_abc");
+
+    // Same instance returned every time
+    expect(store1).toBe(store2);
+    expect(store2).toBe(store3);
+
+    // Only one store was created
+    expect(createdStores).toHaveLength(1);
+  });
+
+  it("creates separate stores for different projectIds", async () => {
+    const storeA = await getOrCreateProjectStore("proj_alpha");
+    const storeB = await getOrCreateProjectStore("proj_beta");
+
+    expect(storeA).not.toBe(storeB);
+    expect(createdStores).toHaveLength(2);
+    expect(createdStores[0].projectId).toBe("proj_alpha");
+    expect(createdStores[1].projectId).toBe("proj_beta");
+  });
+
+  it("calls watch() exactly once per project store", async () => {
+    await getOrCreateProjectStore("proj_once");
+    await getOrCreateProjectStore("proj_once");
+    await getOrCreateProjectStore("proj_once");
+
+    expect(createdStores).toHaveLength(1);
+    expect(createdStores[0].watchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not stack duplicate watch() calls on repeated lookups", async () => {
+    for (let i = 0; i < 10; i++) {
+      await getOrCreateProjectStore("proj_repeat");
+    }
+
+    expect(createdStores).toHaveLength(1);
+    expect(createdStores[0].watchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("evictProjectStore stops watching and closes the store", async () => {
+    await getOrCreateProjectStore("proj_evict");
+    expect(createdStores).toHaveLength(1);
+
+    evictProjectStore("proj_evict");
+
+    expect(createdStores[0].stopWatchingMock).toHaveBeenCalledTimes(1);
+    expect(createdStores[0].closeMock).toHaveBeenCalledTimes(1);
+
+    // Next call should create a fresh store
+    createdStores.length = 0;
+    await getOrCreateProjectStore("proj_evict");
+    expect(createdStores).toHaveLength(1);
+  });
+
+  it("evictAllProjectStores cleans up all cached stores", async () => {
+    await getOrCreateProjectStore("proj_a");
+    await getOrCreateProjectStore("proj_b");
+    await getOrCreateProjectStore("proj_c");
+
+    expect(createdStores).toHaveLength(3);
+
+    evictAllProjectStores();
+
+    for (const entry of createdStores) {
+      expect(entry.stopWatchingMock).toHaveBeenCalledTimes(1);
+      expect(entry.closeMock).toHaveBeenCalledTimes(1);
+    }
+  });
+
+  it("evictProjectStore is a no-op for unknown projectIds", () => {
+    expect(() => evictProjectStore("nonexistent")).not.toThrow();
+  });
+
+  it("shared store receives events from both SSE and API route context", async () => {
+    const store = await getOrCreateProjectStore("proj_events");
+
+    // Simulate an SSE listener attaching to the store's EventEmitter
+    const events: string[] = [];
+    store.on("task:created", (task: any) => {
+      events.push(`created:${task.id}`);
+    });
+    store.on("task:updated", (task: any) => {
+      events.push(`updated:${task.id}`);
+    });
+
+    // Simulate the same store being used for a mutation (like getScopedStore)
+    const sameStore = await getOrCreateProjectStore("proj_events");
+    expect(sameStore).toBe(store);
+
+    // Emit events from "mutation path" — same store instance
+    sameStore.emit("task:created", { id: "FN-001" });
+    sameStore.emit("task:updated", { id: "FN-001", title: "Updated" });
+
+    // SSE listener should have received both events through the shared EventEmitter
+    expect(events).toEqual(["created:FN-001", "updated:FN-001"]);
+  });
+
+  it("SSE stream receives live events via shared store (integration)", async () => {
+    // This test simulates the full server-side path:
+    // 1. SSE handler calls getOrCreateProjectStore("proj_sse") → store A
+    // 2. Task mutation route calls getOrCreateProjectStore("proj_sse") → same store A
+    // 3. Task mutation emits event on store A
+    // 4. SSE listener (attached to store A) receives the event
+
+    const sseStore = await getOrCreateProjectStore("proj_sse");
+
+    // Collect SSE messages (simulates createSSE attaching listeners)
+    const sseMessages: string[] = [];
+    sseStore.on("task:created", (task: any) => {
+      sseMessages.push(`event: task:created\ndata: ${JSON.stringify(task)}\n\n`);
+    });
+    sseStore.on("task:updated", (task: any) => {
+      sseMessages.push(`event: task:updated\ndata: ${JSON.stringify(task)}\n\n`);
+    });
+    sseStore.on("task:moved", (data: any) => {
+      sseMessages.push(`event: task:moved\ndata: ${JSON.stringify(data)}\n\n`);
+    });
+
+    // Simulate API route handler using the same resolver
+    const apiStore = await getOrCreateProjectStore("proj_sse");
+    expect(apiStore).toBe(sseStore);
+
+    // Simulate task creation via API (this is what routes.ts does)
+    const newTask = { id: "FN-100", description: "Integration test task" };
+    apiStore.emit("task:created", newTask);
+
+    // Simulate task update
+    const updatedTask = { ...newTask, title: "Updated title" };
+    apiStore.emit("task:updated", updatedTask);
+
+    // Simulate task move
+    apiStore.emit("task:moved", { task: updatedTask, from: "triage", to: "todo" });
+
+    // Assert SSE listener received all events in order
+    expect(sseMessages).toHaveLength(3);
+    expect(sseMessages[0]).toContain("task:created");
+    expect(sseMessages[0]).toContain("FN-100");
+    expect(sseMessages[1]).toContain("task:updated");
+    expect(sseMessages[1]).toContain("Updated title");
+    expect(sseMessages[2]).toContain("task:moved");
+    expect(sseMessages[2]).toContain('"to":"todo"');
+  });
+});

packages/dashboard/src/mission-routes.ts

@@ -15,6 +15,7 @@
 import { Router, type Request, type Response, type NextFunction } from "express";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { TaskStore } from "@fusion/core";
+import { getOrCreateProjectStore } from "./project-store-resolver.js";
 import type {
   Mission,
   Milestone,
@@ -170,7 +171,7 @@ export function createMissionRouter(store: TaskStore): Router {
   router.use(async (req, _res, next) => {
     try {
       const projectId = getProjectIdFromRequest(req);
-      const scopedStore = projectId ? await TaskStore.getOrCreateForProject(projectId) : store;
+      const scopedStore = projectId ? await getOrCreateProjectStore(projectId) : store;
       requestContext.run(scopedStore.getMissionStore(), next);
     } catch (error) {
       next(error);

packages/dashboard/src/project-store-resolver.ts

@@ -0,0 +1,86 @@
+/**
+ * Project-scoped TaskStore resolver for the dashboard server.
+ *
+ * Caches TaskStore instances by projectId so that SSE subscriptions
+ * and API route handlers for the same project share a single in-memory
+ * EventEmitter. Without this cache, every call to
+ * `TaskStore.getOrCreateForProject()` creates an independent TaskStore
+ * with its own EventEmitter — mutations on one instance would never
+ * reach SSE listeners on another, breaking real-time dashboard updates
+ * for project-scoped views.
+ *
+ * Usage:
+ *   import { getOrCreateProjectStore } from "./project-store-resolver.js";
+ *   const store = await getOrCreateProjectStore(projectId);
+ */
+
+import type { TaskStore } from "@fusion/core";
+
+/**
+ * Internal cache: projectId → TaskStore instance.
+ * Keyed by projectId (not project path) because the dashboard server
+ * routes identify projects by their central-registry ID.
+ */
+const storeCache = new Map<string, TaskStore>();
+
+/**
+ * Track which stores have been fully initialized for real-time operation
+ * (watcher started). This prevents duplicate watch() calls on repeated
+ * lookups.
+ */
+const initializedProjects = new Set<string>();
+
+/**
+ * Get or create a cached TaskStore for the given projectId.
+ *
+ * - First call for a projectId: creates, inits, and caches the store.
+ *   Also starts the SQLite polling watcher so external changes (CLI,
+ *   engine agents) are detected and emitted as events.
+ * - Subsequent calls: returns the cached instance immediately.
+ *
+ * @param projectId - The central-registry project ID
+ * @returns A shared TaskStore instance for this project
+ */
+export async function getOrCreateProjectStore(projectId: string): Promise<TaskStore> {
+  const cached = storeCache.get(projectId);
+  if (cached) {
+    return cached;
+  }
+
+  const { TaskStore: TaskStoreClass } = await import("@fusion/core");
+  const store = await TaskStoreClass.getOrCreateForProject(projectId);
+
+  // Start watching for external changes (CLI, engine agents, etc.)
+  // so SSE listeners receive live events even when mutations happen
+  // outside this process.
+  if (!initializedProjects.has(projectId)) {
+    initializedProjects.add(projectId);
+    await store.watch();
+  }
+
+  storeCache.set(projectId, store);
+  return store;
+}
+
+/**
+ * Remove a cached store and stop its watcher.
+ * Useful for cleanup on project removal or server shutdown.
+ */
+export function evictProjectStore(projectId: string): void {
+  const store = storeCache.get(projectId);
+  if (store) {
+    store.stopWatching();
+    store.close();
+    storeCache.delete(projectId);
+    initializedProjects.delete(projectId);
+  }
+}
+
+/**
+ * Evict all cached stores. Used during server shutdown.
+ */
+export function evictAllProjectStores(): void {
+  for (const projectId of storeCache.keys()) {
+    evictProjectStore(projectId);
+  }
+}

packages/dashboard/src/routes.ts

@@ -24,6 +24,7 @@ import {
   type BadgeUrlComponents,
 } from "./github-webhooks.js";
 import { createMissionRouter } from "./mission-routes.js";
+import { getOrCreateProjectStore } from "./project-store-resolver.js";
 
 /**
  * Minimal interface matching pi-coding-agent's ModelRegistry API surface
@@ -1090,8 +1091,9 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
     const projectId = getProjectIdFromRequest(req);
     if (!projectId) return store;
 
-    const { TaskStore: TaskStoreClass } = await import("@fusion/core");
-    return TaskStoreClass.getOrCreateForProject(projectId);
+    // Use the shared project-store resolver so mutations emit events
+    // on the same EventEmitter that SSE listeners are attached to.
+    return getOrCreateProjectStore(projectId);
   }
 
   if (process.env.FUSION_DEBUG_PLANNING_ROUTES === "1") {

packages/dashboard/src/server.ts

@@ -8,6 +8,7 @@ import type { AuthStorageLike, ModelRegistryLike } from "./routes.js";
 import { createApiRoutes } from "./routes.js";
 import { createSSE } from "./sse.js";
 import { rateLimit, RATE_LIMITS } from "./rate-limit.js";
+import { getOrCreateProjectStore, evictAllProjectStores } from "./project-store-resolver.js";
 import { getTerminalService, type TerminalSession } from "./terminal-service.js";
 import { WebSocketServer, type WebSocket } from "ws";
 import { terminalSessionManager } from "./terminal.js";
@@ -117,8 +118,9 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
     }
 
     try {
-      const { TaskStore: TaskStoreClass } = await import("@fusion/core");
-      const scopedStore = await TaskStoreClass.getOrCreateForProject(projectId);
+      // Use the shared project-store resolver so SSE listeners attach to
+      // the same EventEmitter used by project-scoped task API routes.
+      const scopedStore = await getOrCreateProjectStore(projectId);
       createSSE(scopedStore, scopedStore.getMissionStore())(req, res);
     } catch (err: any) {
       res.status(500).json({ error: err.message ?? "Failed to open project event stream" });
@@ -143,8 +145,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
 
     void (async () => {
       if (projectId) {
-        const { TaskStore: TaskStoreClass } = await import("@fusion/core");
-        activeStore = await TaskStoreClass.getOrCreateForProject(projectId);
+        activeStore = await getOrCreateProjectStore(projectId);
       }
 
       const onAgentLog = (entry: { taskId: string; text: string; type: string; timestamp: string }) => {
@@ -631,6 +632,8 @@ export function setupBadgeWebSocket(
     wsManager.dispose();
     void badgePubSub.dispose();
     wss.close();
+    // Clean up cached project-scoped stores (stop watchers, close DB connections)
+    evictAllProjectStores();
     dashboardApp.terminalWsServer = null;
     dashboardApp.badgeWsServer = null;
     dashboardApp.badgeWsManager = null;