feat(FN-3959): add heartbeat timer registration repair sweep

Heartbeat timer repair sweep — adds a registration repair mechanism that detects and flags stale timers via agent updates, with tests covering the new sweep behavior. Documentation updated in agents.md to reflect the repair flow. Fusion-Task-Id: FN-3959 Fusion-Task-Lineage: 40708b6b-01e1-4673-93e1-299678de2879
2026-05-11 15:18:40 -07:00
parent 59fbd7c305
commit 9bb46f5bfa
6 changed files with 193 additions and 111 deletions
--- a/packages/dashboard/app/hooks/tests/useChatRooms.test.ts
+++ b/packages/dashboard/app/hooks/tests/useChatRooms.test.ts
@@ -209,6 +209,25 @@ describe("useChatRooms", () => {
    expect(result.current.messages.map((message) => message.id)).toEqual(["msg-user", "msg-assistant"]);
  });

+  it("refreshes persisted room messages even when room reply generation fails", async () => {
+    const active = room("room-1", "one", "2026-05-09T01:00:00.000Z");
+    mockFetchChatRooms.mockResolvedValueOnce({ rooms: [active] });
+    const { result } = renderHook(() => useChatRooms("proj-1"));
+    await waitFor(() => expect(result.current.rooms.length).toBe(1));
+
+    mockFetchChatRoomMembers.mockResolvedValueOnce({ members: [] });
+    mockFetchChatRoomMessages.mockResolvedValueOnce({ messages: [] });
+    act(() => result.current.selectRoom("room-1"));
+    await waitFor(() => expect(result.current.activeRoom?.id).toBe("room-1"));
+
+    const persistedUserMessage = roomMessage("msg-user", "room-1", "hello");
+    mockPostChatRoomMessage.mockRejectedValueOnce(new Error("No active room responders available for room room-1"));
+    mockFetchChatRoomMessages.mockResolvedValueOnce({ messages: [persistedUserMessage] });
+
+    await expect(result.current.sendRoomMessage("hello")).rejects.toThrow("No active room responders available for room room-1");
+    expect(result.current.messages.map((message) => message.id)).toEqual(["msg-user"]);
+  });
+
  it("tears down sse subscription on unmount", async () => {
    const { unmount } = renderHook(() => useChatRooms("proj-1"));
    unmount();
--- a/packages/dashboard/src/tests/chat-room-routes.test.ts
+++ b/packages/dashboard/src/tests/chat-room-routes.test.ts
@@ -312,7 +312,7 @@ describe("Chat Room API Routes", () => {
    expect(missingMessage.status).toBe(404);
  });

-  it("resolves project-scoped room services for message replies", async () => {
+  it("resolves project-scoped room services for room reads and message replies", async () => {
    const scopedRoot = mkdtempSync(join(tmpdir(), "fusion-chat-room-scoped-"));
    const scopedFusionDir = join(scopedRoot, ".fusion");
    const scopedDb = new Database(scopedFusionDir, { inMemory: true });
@@ -333,6 +333,12 @@ describe("Chat Room API Routes", () => {
      projectId: "proj-scope",
      memberAgentIds: [scopedAgent.id],
    });
+    const seededMessage = scopedChatStore.addRoomMessage(room.id, {
+      role: "user",
+      content: "seeded scoped message",
+      senderAgentId: null,
+      mentions: [],
+    });

    const defaultChatManager = {
      sendRoomMessage: async () => {
@@ -362,6 +368,14 @@ describe("Chat Room API Routes", () => {
      metadata: { roomId: room.id },
    });

+    const listRes = await request(appWithScopedEngine, "GET", "/api/chat/rooms?projectId=proj-scope");
+    expect(listRes.status).toBe(200);
+    expect((listRes.body as any).rooms.map((entry: any) => entry.id)).toEqual([room.id]);
+
+    const messagesRes = await request(appWithScopedEngine, "GET", `/api/chat/rooms/${room.id}/messages?projectId=proj-scope`);
+    expect(messagesRes.status).toBe(200);
+    expect((messagesRes.body as any).messages.map((entry: any) => entry.id)).toContain(seededMessage.id);
+
    const postRes = await request(
      appWithScopedEngine,
      "POST",
--- a/packages/dashboard/src/chat-project-services.ts
+++ b/packages/dashboard/src/chat-project-services.ts
@@ -0,0 +1,63 @@
+import { AgentStore, ChatStore, type MessageStore, type TaskStore } from "@fusion/core";
+import type { PluginRunner, ProjectEngineManager } from "@fusion/engine";
+import { ChatManager } from "./chat.js";
+import { getOrCreateProjectStore } from "./project-store-resolver.js";
+
+const scopedChatStoreCache = new Map<string, ChatStore>();
+
+function cacheKeyForStore(store: TaskStore): string {
+  return store.getFusionDir();
+}
+
+export function getOrCreateScopedChatStore(store: TaskStore, fallbackChatStore?: ChatStore): ChatStore {
+  const key = cacheKeyForStore(store);
+  const cached = scopedChatStoreCache.get(key);
+  if (cached) return cached;
+
+  const chatStore = fallbackChatStore ?? new ChatStore(store.getFusionDir(), store.getDatabase());
+  scopedChatStoreCache.set(key, chatStore);
+  return chatStore;
+}
+
+export async function resolveProjectChatContext(options: {
+  projectId?: string | null;
+  defaultStore: TaskStore;
+  defaultChatStore?: ChatStore;
+  engineManager?: ProjectEngineManager;
+}): Promise<{ store: TaskStore; chatStore: ChatStore }> {
+  const { projectId, defaultStore, defaultChatStore, engineManager } = options;
+  if (!projectId) {
+    return {
+      store: defaultStore,
+      chatStore: getOrCreateScopedChatStore(defaultStore, defaultChatStore),
+    };
+  }
+
+  const engine = engineManager?.getEngine(projectId);
+  const scopedStore = engine?.getTaskStore() ?? await getOrCreateProjectStore(projectId);
+  return {
+    store: scopedStore,
+    chatStore: getOrCreateScopedChatStore(scopedStore),
+  };
+}
+
+export async function createProjectScopedChatManager(options: {
+  store: TaskStore;
+  chatStore: ChatStore;
+  pluginRunner?: PluginRunner;
+  messageStore?: MessageStore;
+}): Promise<ChatManager> {
+  const agentStore = new AgentStore({ rootDir: options.store.getFusionDir() });
+  return new ChatManager(
+    options.chatStore,
+    options.store.getRootDir(),
+    agentStore,
+    options.pluginRunner,
+    () => options.store.getSettings(),
+    options.messageStore,
+  );
+}
+
+export function __resetScopedChatStoreCache(): void {
+  scopedChatStoreCache.clear();
+}
--- a/packages/dashboard/src/routes/register-chat-room-routes.ts
+++ b/packages/dashboard/src/routes/register-chat-room-routes.ts
@@ -1,6 +1,7 @@
-import { AgentStore, ChatStore, type ChatAttachment, type ChatRoomCreateInput, type ChatRoomStatus, type ChatRoomUpdateInput } from "@fusion/core";
+import type { ChatAttachment, ChatRoomCreateInput, ChatRoomStatus, ChatRoomUpdateInput } from "@fusion/core";
 import type { Request } from "express";
-import { ChatManager, RoomReplyGenerationError } from "../chat.js";
+import { RoomReplyGenerationError } from "../chat.js";
+import { createProjectScopedChatManager, resolveProjectChatContext } from "../chat-project-services.js";
 import { ApiError, badRequest, internalError, notFound } from "../api-error.js";
 import { rateLimit, RATE_LIMITS } from "../rate-limit.js";
 import type { ApiRoutesContext } from "./types.js";
@@ -11,65 +12,54 @@ function isSlugCollisionError(err: unknown): boolean {
 }

 export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {
-  const { router, options, chatLogger, rethrowAsApiError, getProjectContext } = ctx;
-  const scopedRoomManagers = new Map<string, { chatStore: ChatStore; chatManager: ChatManager }>();
+  const { router, options, chatLogger, rethrowAsApiError } = ctx;

-  async function resolveRoomScopedServices(req: Request, roomProjectId: string | null | undefined): Promise<{ chatStore: ChatStore; chatManager: ChatManager }> {
+  function getRequestedProjectId(req: Request): string | undefined {
+    return typeof req.query.projectId === "string"
+      ? req.query.projectId
+      : (typeof req.body?.projectId === "string" ? req.body.projectId : undefined);
+  }
+
+  async function resolveRoomScopedServices(req: Request, roomProjectId: string | null | undefined) {
    if (!roomProjectId) {
      const chatStore = options?.chatStore;
      const chatManager = options?.chatManager;
      if (!chatStore || !chatManager) {
        throw internalError("Chat store or manager not available");
      }
-      return { chatStore, chatManager };
+      return { store: ctx.store, chatStore, chatManager };
    }

-    const cached = scopedRoomManagers.get(roomProjectId);
-    if (cached) {
-      return cached;
+    const { store: scopedStore, chatStore } = await resolveProjectChatContext({
+      projectId: roomProjectId,
+      defaultStore: ctx.store,
+      defaultChatStore: options?.chatStore,
+      engineManager: options?.engineManager,
+    });
+
+    if (scopedStore === ctx.store && options?.chatStore && options?.chatManager) {
+      return { store: ctx.store, chatStore: options.chatStore, chatManager: options.chatManager };
    }

-    const scopedReq = {
-      ...req,
-      query: { ...(req.query as Record<string, unknown>), projectId: roomProjectId },
-      body: typeof req.body === "object" && req.body !== null
-        ? { ...(req.body as Record<string, unknown>), projectId: roomProjectId }
-        : { projectId: roomProjectId },
-    } as unknown as Request;
-    const { store: scopedStore, engine } = await getProjectContext(scopedReq);
-
-    const scopedChatStore = new ChatStore(scopedStore.getFusionDir(), scopedStore.getDatabase());
-    const scopedAgentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
-    const scopedChatManager = new ChatManager(
-      scopedChatStore,
-      scopedStore.getRootDir(),
-      scopedAgentStore,
-      options?.pluginRunner,
-      () => scopedStore.getSettings(),
-      engine?.getMessageStore(),
-    );
-
-    const resolved = { chatStore: scopedChatStore, chatManager: scopedChatManager };
-    scopedRoomManagers.set(roomProjectId, resolved);
-    return resolved;
+    const engine = options?.engineManager?.getEngine(roomProjectId);
+    const chatManager = await createProjectScopedChatManager({
+      store: scopedStore,
+      chatStore,
+      pluginRunner: options?.pluginRunner,
+      messageStore: engine?.getMessageStore(),
+    });
+    return { store: scopedStore, chatStore, chatManager };
  }

  router.get("/chat/rooms", rateLimit(RATE_LIMITS.api), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
-      const { projectId, status, agentId } = req.query as {
-        projectId?: string;
-        status?: string;
-        agentId?: string;
-      };
-
+      const projectId = getRequestedProjectId(req);
+      const { chatStore } = await resolveRoomScopedServices(req, projectId);
+      const { status, agentId } = req.query as { status?: string; agentId?: string };
      const statusFilter = status as ChatRoomStatus | undefined;
      const rooms = agentId
        ? chatStore.listRoomsForAgent(agentId, { projectId, status: statusFilter })
        : chatStore.listRooms({ projectId, status: statusFilter });
-
      res.json({ rooms });
    } catch (err: unknown) {
      if (err instanceof ApiError) throw err;
@@ -79,9 +69,6 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.post("/chat/rooms", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
      const { name, description, projectId, createdBy, memberAgentIds } = req.body as {
        name?: string;
        description?: string | null;
@@ -89,11 +76,11 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {
        createdBy?: string | null;
        memberAgentIds?: string[];
      };
-
      if (!name || typeof name !== "string" || !name.trim()) {
        throw badRequest("name is required and must be a non-empty string");
      }

+      const { chatStore } = await resolveRoomScopedServices(req, projectId);
      const roomInput: ChatRoomCreateInput & { memberAgentIds?: string[] } = {
        name: name.trim(),
        ...(description !== undefined ? { description } : {}),
@@ -122,13 +109,10 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.get("/chat/rooms/:id", rateLimit(RATE_LIMITS.api), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
-
      const members = chatStore.listRoomMembers(roomId);
      res.json({ room, members });
    } catch (err: unknown) {
@@ -139,16 +123,13 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.patch("/chat/rooms/:id", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
-      const roomId = String(req.params.id);
      const { name, description, status } = req.body as { name?: string; description?: string | null; status?: ChatRoomStatus };
-
      if (name === undefined && description === undefined && status === undefined) {
        throw badRequest("at least one of name, description, or status is required");
      }

+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
+      const roomId = String(req.params.id);
      const input: ChatRoomUpdateInput = {
        ...(name !== undefined ? { name: name.trim() } : {}),
        ...(description !== undefined ? { description } : {}),
@@ -175,13 +156,10 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.delete("/chat/rooms/:id", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
-
      chatStore.deleteRoom(roomId);
      res.json({ success: true });
    } catch (err: unknown) {
@@ -192,13 +170,10 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.get("/chat/rooms/:id/members", rateLimit(RATE_LIMITS.api), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
-
      const members = chatStore.listRoomMembers(roomId);
      res.json({ members });
    } catch (err: unknown) {
@@ -209,9 +184,7 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.post("/chat/rooms/:id/members", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
@@ -234,14 +207,11 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.delete("/chat/rooms/:id/members/:agentId", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const agentId = String(req.params.agentId);
      const removed = chatStore.removeRoomMember(roomId, agentId);
      if (!removed) throw notFound(`Room member ${agentId} not found in room ${roomId}`);
-
      res.json({ success: true });
    } catch (err: unknown) {
      if (err instanceof ApiError) throw err;
@@ -251,9 +221,7 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.get("/chat/rooms/:id/messages", rateLimit(RATE_LIMITS.api), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
@@ -269,7 +237,6 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {
        offset,
        ...(before ? { before } : {}),
      });
-
      res.json({ messages });
    } catch (err: unknown) {
      if (err instanceof ApiError) throw err;
@@ -279,24 +246,17 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.post("/chat/rooms/:id/messages", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const defaultChatStore = options?.chatStore;
-      if (!defaultChatStore) throw internalError("Chat store not available");
-
      const roomId = String(req.params.id);
-      const hintedProjectId = typeof req.query.projectId === "string"
-        ? req.query.projectId
-        : (typeof req.body?.projectId === "string" ? req.body.projectId : undefined);
-      const room = defaultChatStore.getRoom(roomId)
-        ?? (hintedProjectId ? (await resolveRoomScopedServices(req, hintedProjectId)).chatStore.getRoom(roomId) : undefined);
+      const requestedProjectId = getRequestedProjectId(req);
+      const { chatStore } = await resolveRoomScopedServices(req, requestedProjectId);
+      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);

      const { content, senderAgentId, attachments } = req.body as {
        content?: string;
        senderAgentId?: string | null;
-        mentions?: string[];
        attachments?: ChatAttachment[];
      };
-
      if (!content || typeof content !== "string" || !content.trim()) {
        throw badRequest("content is required and must be a non-empty string");
      }
@@ -306,7 +266,6 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

      const { chatManager } = await resolveRoomScopedServices(req, room.projectId);
      const result = await chatManager.sendRoomMessage(roomId, content.trim(), Array.isArray(attachments) ? attachments : undefined);
-
      res.status(201).json({ message: result.userMessage });
    } catch (err: unknown) {
      if (err instanceof ApiError) throw err;
@@ -319,19 +278,15 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.delete("/chat/rooms/:id/messages/:messageId", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const messageId = String(req.params.messageId);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
-
      const message = chatStore.getRoomMessage(messageId);
      if (!message || message.roomId !== roomId) {
        throw notFound(`Message ${messageId} not found`);
      }
-
      chatStore.deleteRoomMessage(messageId);
      res.json({ success: true });
    } catch (err: unknown) {
@@ -342,14 +297,11 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {

  router.post("/chat/rooms/:id/messages/:messageId/attachments", rateLimit(RATE_LIMITS.mutation), async (req, res) => {
    try {
-      const chatStore = options?.chatStore;
-      if (!chatStore) throw internalError("Chat store not available");
-
+      const { chatStore } = await resolveRoomScopedServices(req, getRequestedProjectId(req));
      const roomId = String(req.params.id);
      const messageId = String(req.params.messageId);
      const room = chatStore.getRoom(roomId);
      if (!room) throw notFound(`Chat room ${roomId} not found`);
-
      const message = chatStore.getRoomMessage(messageId);
      if (!message || message.roomId !== roomId) {
        throw notFound(`Message ${messageId} not found`);
@@ -369,20 +321,21 @@ export function registerChatRoomRoutes(ctx: ApiRoutesContext): void {
  });

  if (process.env.FUSION_DEBUG_CHAT_ROUTES === "1") {
-    const chatRoomRoutes = [
-      "GET /chat/rooms",
-      "POST /chat/rooms",
-      "GET /chat/rooms/:id",
-      "PATCH /chat/rooms/:id",
-      "DELETE /chat/rooms/:id",
-      "GET /chat/rooms/:id/members",
-      "POST /chat/rooms/:id/members",
-      "DELETE /chat/rooms/:id/members/:agentId",
-      "GET /chat/rooms/:id/messages",
-      "POST /chat/rooms/:id/messages",
-      "DELETE /chat/rooms/:id/messages/:messageId",
-      "POST /chat/rooms/:id/messages/:messageId/attachments",
-    ];
-    chatLogger.info("room routes registered", { chatRoomRoutes });
+    chatLogger.info("room routes registered", {
+      chatRoomRoutes: [
+        "GET /chat/rooms",
+        "POST /chat/rooms",
+        "GET /chat/rooms/:id",
+        "PATCH /chat/rooms/:id",
+        "DELETE /chat/rooms/:id",
+        "GET /chat/rooms/:id/members",
+        "POST /chat/rooms/:id/members",
+        "DELETE /chat/rooms/:id/members/:agentId",
+        "GET /chat/rooms/:id/messages",
+        "POST /chat/rooms/:id/messages",
+        "DELETE /chat/rooms/:id/messages/:messageId",
+        "POST /chat/rooms/:id/messages/:messageId/attachments",
+      ],
+    });
  }
 }
--- a/packages/dashboard/src/server.ts
+++ b/packages/dashboard/src/server.ts
@@ -13,6 +13,7 @@ import { createSSE, disconnectSSEClient, markSSEClientAlive } from "./sse.js";
 import { rateLimit, RATE_LIMITS } from "./rate-limit.js";
 import { ApiError, sendErrorResponse } from "./api-error.js";
 import { getOrCreateProjectStore, evictAllProjectStores, setOnProjectFirstCreated } from "./project-store-resolver.js";
+import { getOrCreateScopedChatStore } from "./chat-project-services.js";
 import { getTerminalService, STALE_SESSION_THRESHOLD_MS } from "./terminal-service.js";
 import { WebSocketServer, type WebSocket } from "ws";
 import { terminalSessionManager } from "./terminal.js";
@@ -660,15 +661,18 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
      let agentStore;
      let messageStore: MessageStore | undefined;
      let automationStore: AutomationStore | undefined;
+      let scopedChatStore = chatStore;
      if (engineManager) {
        const engine = engineManager.getEngine(projectId);
        scopedStore = engine?.getTaskStore() ?? await getOrCreateProjectStore(projectId);
+        scopedChatStore = getOrCreateScopedChatStore(scopedStore);
        // Use the engine's stores if available
        agentStore = engine?.getAgentStore();
        messageStore = engine?.getMessageStore();
        automationStore = engine?.getAutomationStore();
      } else {
        scopedStore = await getOrCreateProjectStore(projectId);
+        scopedChatStore = getOrCreateScopedChatStore(scopedStore);
      }
      // Fallback: create AgentStore if engine doesn't have one
      if (!agentStore) {
@@ -689,7 +693,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
        },
        agentStore,
        messageStore,
-        chatStore,
+        scopedChatStore,
        automationStore,
      )(req, res);
    } catch (err: unknown) {