feat(FN-1975): stream chat session updates through SSE

- Emit chat:session:updated from ChatStore when message deletion mutates a session and cover deleteMessage false returns - Pass ChatStore into SSE setup and forward chat session update events to connected clients - Update useChat to consume SSE updates in real time with safer EnrichedChatSession typing - Add core and dashboard tests for chat-store emissions, SSE forwarding, route wiring, and hook behavior
2026-04-18 00:44:27 -07:00
parent ddbb60e297
commit 17fcbfe6c8
12 changed files with 692 additions and 7 deletions
--- a/packages/dashboard/app/api.ts
+++ b/packages/dashboard/app/api.ts
@@ -39,6 +39,7 @@ import type {
  AgentRatingInput,
  ChatSession,
  ChatMessage,
+  EnrichedChatSession,
  Roadmap,
  RoadmapMilestone,
  RoadmapFeature,
@@ -5298,8 +5299,10 @@ export async function fetchSkillsCatalog(

 // ── Chat API ─────────────────────────────────────────────────────────────────

+// EnrichedChatSession is imported from @fusion/core above
+
 export interface ChatSessionListResponse {
-  sessions: ChatSession[];
+  sessions: EnrichedChatSession[];
 }

 export interface ChatSessionResponse {
--- a/packages/dashboard/app/hooks/tests/useChat.test.ts
+++ b/packages/dashboard/app/hooks/tests/useChat.test.ts
@@ -30,11 +30,18 @@ vi.mock("../../utils/projectStorage", () => ({
  removeScopedItem: vi.fn(),
 }));

+// Mock the SSE bus
+vi.mock("../../sse-bus", () => ({
+  subscribeSse: vi.fn(() => () => {}),
+}));
+
 import * as projectStorageModule from "../../utils/projectStorage";
+import * as sseBusModule from "../../sse-bus";

 const mockGetScopedItem = vi.mocked(projectStorageModule.getScopedItem);
 const mockSetScopedItem = vi.mocked(projectStorageModule.setScopedItem);
 const mockRemoveScopedItem = vi.mocked(projectStorageModule.removeScopedItem);
+const mockSubscribeSse = vi.mocked(sseBusModule.subscribeSse);

 const mockFetchChatSessions = vi.mocked(apiModule.fetchChatSessions);
 const mockCreateChatSession = vi.mocked(apiModule.createChatSession);
@@ -580,6 +587,307 @@ describe("useChat", () => {
    });
  });

+  describe("SSE real-time updates", () => {
+    let subscribeHandler: Record<string, (event: MessageEvent) => void> = {};
+
+    beforeEach(() => {
+      subscribeHandler = {};
+      mockSubscribeSse.mockImplementation((_url, options) => {
+        // Capture the event handlers
+        if (options?.events) {
+          subscribeHandler = options.events as typeof subscribeHandler;
+        }
+        return () => {};
+      });
+    });
+
+    afterEach(() => {
+      subscribeHandler = {};
+    });
+
+    it("subscribes to chat SSE events", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({ sessions: [] });
+
+      renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(mockSubscribeSse).toHaveBeenCalledWith(
+          "/api/events?projectId=proj-123",
+          expect.objectContaining({
+            events: expect.objectContaining({
+              "chat:session:created": expect.any(Function),
+              "chat:session:updated": expect.any(Function),
+              "chat:session:deleted": expect.any(Function),
+              "chat:message:added": expect.any(Function),
+              "chat:message:deleted": expect.any(Function),
+            }),
+          }),
+        );
+      });
+    });
+
+    it("adds new session on chat:session:created event", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      // Simulate SSE event
+      const newSession = makeSession({ id: "session-002", agentId: "agent-002", title: "New Chat" });
+      act(() => {
+        subscribeHandler["chat:session:created"]?.({
+          data: JSON.stringify(newSession),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(2);
+        expect(result.current.sessions[0]?.id).toBe("session-002");
+      });
+    });
+
+    it("avoids duplicate sessions on chat:session:created", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      // Simulate SSE event for the same session
+      const sameSession = makeSession({ id: "session-001", agentId: "agent-001" });
+      act(() => {
+        subscribeHandler["chat:session:created"]?.({
+          data: JSON.stringify(sameSession),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+    });
+
+    it("updates session on chat:session:updated event", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001", title: "Old Title" })],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+        expect(result.current.sessions[0]?.title).toBe("Old Title");
+      });
+
+      // Simulate SSE event
+      const updatedSession = makeSession({ id: "session-001", agentId: "agent-001", title: "New Title" });
+      act(() => {
+        subscribeHandler["chat:session:updated"]?.({
+          data: JSON.stringify(updatedSession),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.sessions[0]?.title).toBe("New Title");
+      });
+    });
+
+    it("removes session on chat:session:deleted event", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [
+          makeSession({ id: "session-001", agentId: "agent-001" }),
+          makeSession({ id: "session-002", agentId: "agent-002" }),
+        ],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(2);
+      });
+
+      // Simulate SSE event
+      act(() => {
+        subscribeHandler["chat:session:deleted"]?.({
+          data: JSON.stringify({ id: "session-001" }),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+        expect(result.current.sessions[0]?.id).toBe("session-002");
+      });
+    });
+
+    it("clears active session when it is deleted", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+      mockFetchChatMessages.mockResolvedValueOnce({ messages: [] });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.activeSession?.id).toBe("session-001");
+      });
+
+      // Simulate SSE event for the active session
+      act(() => {
+        subscribeHandler["chat:session:deleted"]?.({
+          data: JSON.stringify({ id: "session-001" }),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.activeSession).toBeNull();
+        expect(result.current.messages).toHaveLength(0);
+      });
+    });
+
+    it("adds message on chat:message:added event for active session", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+      mockFetchChatMessages.mockResolvedValueOnce({
+        messages: [makeMessage({ id: "msg-001", sessionId: "session-001", role: "user", content: "Hello" })],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(1);
+      });
+
+      // Simulate SSE event for a new message in the active session
+      const newMessage = makeMessage({ id: "msg-002", sessionId: "session-001", role: "assistant", content: "Hi there" });
+      act(() => {
+        subscribeHandler["chat:message:added"]?.({
+          data: JSON.stringify(newMessage),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(2);
+        expect(result.current.messages[1]?.content).toBe("Hi there");
+      });
+    });
+
+    it("does not add message on chat:message:added when streaming", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+      mockFetchChatMessages.mockResolvedValueOnce({ messages: [] });
+
+      // Track stream handlers separately from SSE handlers
+      let streamDoneHandler: ((data: { messageId: string }) => void) | undefined;
+      mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
+        // Capture the onDone handler for stream completion
+        streamDoneHandler = handlers.onDone;
+        return { close: vi.fn(), isConnected: () => true };
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(0);
+      });
+
+      // Start streaming
+      await act(async () => {
+        await result.current.sendMessage("Hello!");
+      });
+
+      await waitFor(() => {
+        expect(result.current.isStreaming).toBe(true);
+      });
+
+      // Simulate SSE event - should not add message during streaming
+      // because isStreaming is true
+      const newMessage = makeMessage({ id: "msg-002", sessionId: "session-001", role: "assistant", content: "Hi" });
+      act(() => {
+        subscribeHandler["chat:message:added"]?.({
+          data: JSON.stringify(newMessage),
+        } as MessageEvent);
+      });
+
+      // Message should not be added during streaming
+      // (the SSE handler checks isStreaming and skips adding)
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(1); // Only the optimistic user message
+      });
+    });
+
+    it("removes message on chat:message:deleted event", async () => {
+      mockFetchChatSessions.mockResolvedValueOnce({
+        sessions: [makeSession({ id: "session-001", agentId: "agent-001" })],
+      });
+      mockFetchChatMessages.mockResolvedValueOnce({
+        messages: [
+          makeMessage({ id: "msg-001", sessionId: "session-001", role: "user", content: "Hello" }),
+          makeMessage({ id: "msg-002", sessionId: "session-001", role: "assistant", content: "Hi there" }),
+        ],
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(2);
+      });
+
+      // Simulate SSE event for deleted message
+      act(() => {
+        subscribeHandler["chat:message:deleted"]?.({
+          data: JSON.stringify({ id: "msg-001" }),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(result.current.messages).toHaveLength(1);
+        expect(result.current.messages[0]?.id).toBe("msg-002");
+      });
+    });
+  });
+
  describe("active session persistence", () => {
    beforeEach(() => {
      // Default: no saved session
--- a/packages/dashboard/app/hooks/useChat.ts
+++ b/packages/dashboard/app/hooks/useChat.ts
@@ -9,6 +9,7 @@ import {
  fetchAgents,
  type ChatSessionListResponse,
 } from "../api";
+import { subscribeSse } from "../sse-bus";
 import { getScopedItem, setScopedItem, removeScopedItem } from "../utils/projectStorage";
 import type { Agent } from "@fusion/core";

@@ -99,6 +100,30 @@ export function useChat(projectId?: string): UseChatReturn {
  // Stream connection ref for cleanup
  const streamRef = useRef<{ close: () => void } | null>(null);

+  // Refs for SSE event handlers to access current state
+  const sessionsRef = useRef(sessions);
+  const activeSessionRef = useRef(activeSession);
+  const isStreamingRef = useRef(isStreaming);
+  sessionsRef.current = sessions;
+  activeSessionRef.current = activeSession;
+  isStreamingRef.current = isStreaming;
+
+  // Tracks message IDs that were added via streaming completion.
+  // Used to prevent duplicate messages when SSE event arrives before streaming state clears.
+  const streamingMessageIdsRef = useRef<Set<string>>(new Set());
+
+  // Tracks the project context version to detect stale SSE events after project switches.
+  // Incremented whenever projectId changes, invalidating any in-flight SSE handlers.
+  const projectContextVersionRef = useRef(0);
+  // Track previous projectId to detect changes
+  const previousProjectIdRef = useRef<string | undefined>(projectId);
+
+  // Detect project changes and invalidate SSE context
+  if (previousProjectIdRef.current !== projectId) {
+    previousProjectIdRef.current = projectId;
+    projectContextVersionRef.current++;
+  }
+
  // Fetch agents on mount for name resolution
  useEffect(() => {
    fetchAgents()
@@ -339,6 +364,9 @@ export function useChat(projectId?: string): UseChatReturn {
            createdAt: new Date().toISOString(),
          };

+          // Track this message ID so SSE handler skips it if event arrives first
+          streamingMessageIdsRef.current.add(assistantMessage.id);
+
          // Preserve user message and add assistant message
          setMessages((prev) => [...prev, assistantMessage]);

@@ -346,6 +374,12 @@ export function useChat(projectId?: string): UseChatReturn {
          setStreamingThinking("");
          setIsStreaming(false);
          streamRef.current = null;
+
+          // Clean up tracked ID after a short delay (SSE event should arrive quickly)
+          setTimeout(() => {
+            streamingMessageIdsRef.current.delete(assistantMessage.id);
+          }, 1000);
+
          refreshSessions();
        },
        onError: (data: string) => {
@@ -372,6 +406,89 @@ export function useChat(projectId?: string): UseChatReturn {
      )
    : sessions;

+  // SSE real-time updates
+  useEffect(() => {
+    const contextVersionAtStart = projectContextVersionRef.current;
+    const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
+
+    const isStale = () => projectContextVersionRef.current !== contextVersionAtStart;
+
+    const handleChatSessionCreated = (e: MessageEvent) => {
+      if (isStale()) return;
+      const session: ChatSessionInfo = JSON.parse(e.data);
+      // Avoid duplicates
+      setSessions((prev) => {
+        if (prev.some((s) => s.id === session.id)) return prev;
+        // Add at the top (sessions are sorted by updatedAt desc)
+        return [session, ...prev];
+      });
+    };
+
+    const handleChatSessionUpdated = (e: MessageEvent) => {
+      if (isStale()) return;
+      const updatedSession: ChatSessionInfo = JSON.parse(e.data);
+      setSessions((prev) => {
+        const updated = prev.map((s) => (s.id === updatedSession.id ? updatedSession : s));
+        return [...updated];
+      });
+      // If this is the active session, update it too
+      if (activeSessionRef.current?.id === updatedSession.id) {
+        setActiveSession(updatedSession);
+      }
+    };
+
+    const handleChatSessionDeleted = (e: MessageEvent) => {
+      if (isStale()) return;
+      const { id: sessionId }: { id: string } = JSON.parse(e.data);
+      setSessions((prev) => prev.filter((s) => s.id !== sessionId));
+      // If this was the active session, clear it
+      if (activeSessionRef.current?.id === sessionId) {
+        setActiveSession(null);
+        setMessages([]);
+      }
+    };
+
+    const handleChatMessageAdded = (e: MessageEvent) => {
+      if (isStale()) return;
+      const message: ChatMessageInfo = JSON.parse(e.data);
+
+      // Skip if this message was already added via streaming completion
+      // (SSE event may arrive before streaming state clears)
+      if (streamingMessageIdsRef.current.has(message.id)) {
+        return;
+      }
+
+      // Only add if this is the active session AND we're not streaming
+      // (during streaming, messages are managed locally to avoid duplicates)
+      // Use ref to get the current value (state may not be updated yet when handler runs)
+      if (activeSessionRef.current?.id === message.sessionId && !isStreamingRef.current) {
+        setMessages((prev) => {
+          // Avoid duplicates
+          if (prev.some((m) => m.id === message.id)) return prev;
+          return [...prev, message];
+        });
+      }
+    };
+
+    const handleChatMessageDeleted = (e: MessageEvent) => {
+      if (isStale()) return;
+      const { id: messageId }: { id: string } = JSON.parse(e.data);
+      setMessages((prev) => prev.filter((m) => m.id !== messageId));
+    };
+
+    const unsubscribe = subscribeSse(`/api/events${query}`, {
+      events: {
+        "chat:session:created": handleChatSessionCreated,
+        "chat:session:updated": handleChatSessionUpdated,
+        "chat:session:deleted": handleChatSessionDeleted,
+        "chat:message:added": handleChatMessageAdded,
+        "chat:message:deleted": handleChatMessageDeleted,
+      },
+    });
+
+    return unsubscribe;
+  }, [projectId]);
+
  // Cleanup on unmount
  useEffect(() => {
    return () => {