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
This commit is contained in:
Fusion
2026-04-18 00:44:27 -07:00
committed by gsxdsm
parent 03f4e7fe44
commit 430f7cf39d
12 changed files with 692 additions and 7 deletions

View File

@@ -555,6 +555,31 @@ describe("ChatStore", () => {
expect(store.getMessages(session2.id)).toHaveLength(1);
expect(store.getMessages(session2.id)[0].content).toBe("Session 2");
});
it("updates the parent session's updatedAt timestamp", async () => {
const session = createTestSession(store);
store.addMessage(session.id, { role: "user", content: "Hello" });
const originalUpdatedAt = store.getSession(session.id)!.updatedAt;
await new Promise((r) => setTimeout(r, 5));
const msg = store.addMessage(session.id, { role: "assistant", content: "Reply" });
const afterAddUpdatedAt = store.getSession(session.id)!.updatedAt;
await new Promise((r) => setTimeout(r, 5));
store.deleteMessage(msg.id);
const afterDeleteUpdatedAt = store.getSession(session.id)!.updatedAt;
// The updatedAt should be newer after adding and after deleting
expect(new Date(afterAddUpdatedAt).getTime()).toBeGreaterThan(
new Date(originalUpdatedAt).getTime(),
);
expect(new Date(afterDeleteUpdatedAt).getTime()).toBeGreaterThan(
new Date(afterAddUpdatedAt).getTime(),
);
});
});
});
@@ -627,6 +652,20 @@ describe("ChatStore", () => {
expect(handler).toHaveBeenCalledWith(message.id);
});
it("deleteMessage emits chat:session:updated for the parent session", () => {
const handler = vi.fn();
store.on("chat:session:updated", handler);
const session = createTestSession(store);
const message = store.addMessage(session.id, { role: "user", content: "Hello" });
handler.mockClear();
store.deleteMessage(message.id);
expect(handler).toHaveBeenCalledTimes(1);
expect(handler.mock.calls[0][0].id).toBe(session.id);
});
it("deleteMessage does NOT emit for non-existent message", () => {
const handler = vi.fn();
store.on("chat:message:deleted", handler);
@@ -636,6 +675,15 @@ describe("ChatStore", () => {
expect(handler).not.toHaveBeenCalled();
});
it("deleteMessage does NOT emit chat:session:updated for non-existent message", () => {
const handler = vi.fn();
store.on("chat:session:updated", handler);
store.deleteMessage("msg-nonexistent");
expect(handler).not.toHaveBeenCalled();
});
it("archiveSession emits chat:session:updated", () => {
const handler = vi.fn();
store.on("chat:session:updated", handler);

View File

@@ -387,9 +387,23 @@ export class ChatStore extends EventEmitter<ChatStoreEvents> {
const existing = this.getMessage(id);
if (!existing) return false;
const sessionId = existing.sessionId;
const now = new Date().toISOString();
this.db.prepare("DELETE FROM chat_messages WHERE id = ?").run(id);
// Update the parent session's updatedAt timestamp
this.db.prepare("UPDATE chat_sessions SET updatedAt = ? WHERE id = ?").run(now, sessionId);
this.db.bumpLastModified();
this.emit("chat:message:deleted", id);
// Emit session:updated for the parent session
const updatedSession = this.getSession(sessionId);
if (updatedSession) {
this.emit("chat:session:updated", updatedSession);
}
return true;
}
}

View File

@@ -45,6 +45,18 @@ export interface ChatSession {
*/
export type ChatSessionSummary = ChatSession;
/**
* Chat session enriched with last message preview data.
* The server enriches sessions with lastMessagePreview and lastMessageAt
* by fetching the most recent message for each session.
*/
export type EnrichedChatSession = ChatSession & {
/** Preview of the last message in the session (truncated to 100 chars) */
lastMessagePreview?: string;
/** Timestamp of the last message in the session */
lastMessageAt?: string;
};
/** A parsed @ mention of an agent in a chat message */
export interface ChatMention {
agentId: string;

View File

@@ -588,6 +588,7 @@ export type {
ChatMessageRole,
ChatSession,
ChatSessionSummary,
EnrichedChatSession,
ChatMention,
ChatMessage,
ChatMessageCreateInput,

View File

@@ -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 {

View File

@@ -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

View File

@@ -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 () => {

View File

@@ -928,6 +928,21 @@ describe("Chat API Routes", () => {
expect(response.status).toBe(404);
});
it("returns 404 when deleteMessage returns false", async () => {
mockGetSession.mockReturnValue(sampleSession);
mockGetMessage.mockReturnValue(sampleMessage);
mockDeleteMessage.mockReturnValue(false);
const response = await request(
app,
"DELETE",
"/api/chat/sessions/chat-abc123/messages/msg-xyz789",
);
expect(response.status).toBe(404);
expect(mockDeleteMessage).toHaveBeenCalledWith("msg-xyz789");
});
});
// ── SSE Streaming Tests ────────────────────────────────────────────────────

View File

@@ -355,6 +355,132 @@ describe("createSSE", () => {
// ── Plugin Lifecycle Event Tests ─────────────────────────────────────────────
describe("chat store events", () => {
it("relays chat:session:created events when chatStore is provided", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res, chunks } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
const session = {
id: "chat-abc123",
agentId: "agent-001",
title: "Test Session",
status: "active",
projectId: null,
modelProvider: null,
modelId: null,
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
};
chatStore.emit("chat:session:created", session);
const sseMsg = chunks.find((c) => c.includes("event: chat:session:created"));
expect(sseMsg).toBeDefined();
expect(extractSSEPayload(sseMsg!).id).toBe("chat-abc123");
});
it("relays chat:session:updated events", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res, chunks } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
const session = {
id: "chat-abc123",
agentId: "agent-001",
title: "Updated Title",
status: "active",
projectId: null,
modelProvider: null,
modelId: null,
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-02T00:00:00.000Z",
};
chatStore.emit("chat:session:updated", session);
const sseMsg = chunks.find((c) => c.includes("event: chat:session:updated"));
expect(sseMsg).toBeDefined();
const payload = extractSSEPayload(sseMsg!);
expect(payload.id).toBe("chat-abc123");
expect(payload.title).toBe("Updated Title");
});
it("relays chat:session:deleted events with session ID", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res, chunks } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
chatStore.emit("chat:session:deleted", "chat-abc123");
const sseMsg = chunks.find((c) => c.includes("event: chat:session:deleted"));
expect(sseMsg).toBeDefined();
expect(extractSSEPayload(sseMsg!).id).toBe("chat-abc123");
});
it("relays chat:message:added events with full message", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res, chunks } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
const message = {
id: "msg-xyz789",
sessionId: "chat-abc123",
role: "user",
content: "Hello, how are you?",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
};
chatStore.emit("chat:message:added", message);
const sseMsg = chunks.find((c) => c.includes("event: chat:message:added"));
expect(sseMsg).toBeDefined();
const payload = extractSSEPayload(sseMsg!);
expect(payload.id).toBe("msg-xyz789");
expect(payload.sessionId).toBe("chat-abc123");
expect(payload.content).toBe("Hello, how are you?");
});
it("relays chat:message:deleted events with message ID", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res, chunks } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
chatStore.emit("chat:message:deleted", "msg-xyz789");
const sseMsg = chunks.find((c) => c.includes("event: chat:message:deleted"));
expect(sseMsg).toBeDefined();
expect(extractSSEPayload(sseMsg!).id).toBe("msg-xyz789");
});
it("cleans up chat store listeners on disconnect", () => {
const chatStore = createMockStore();
const req = createMockRequest();
const { res } = createMockResponse();
createSSE(store, undefined, undefined, undefined, undefined, undefined, undefined, chatStore)(req, res);
expect(chatStore.listenerCount("chat:session:created")).toBe(1);
expect(chatStore.listenerCount("chat:session:updated")).toBe(1);
expect(chatStore.listenerCount("chat:session:deleted")).toBe(1);
expect(chatStore.listenerCount("chat:message:added")).toBe(1);
expect(chatStore.listenerCount("chat:message:deleted")).toBe(1);
req.emit("close");
expect(chatStore.listenerCount("chat:session:created")).toBe(0);
expect(chatStore.listenerCount("chat:session:updated")).toBe(0);
expect(chatStore.listenerCount("chat:session:deleted")).toBe(0);
expect(chatStore.listenerCount("chat:message:added")).toBe(0);
expect(chatStore.listenerCount("chat:message:deleted")).toBe(0);
});
});
// ── Plugin Lifecycle Event Tests ─────────────────────────────────────────────
describe("plugin lifecycle events", () => {
it("emits plugin:lifecycle event for plugin:registered (installing transition)", () => {
const pluginStore = createMockStore();

View File

@@ -17,7 +17,7 @@ import { tmpdir } from "node:os";
import * as nodeFs from "node:fs";
import { promisify } from "node:util";
import type { TaskStore, Column, ScheduleType, ActivityEventType, ModelPreset, MessageType, ParticipantType, RoutineTriggerType, ProjectSettings } from "@fusion/core";
import type { TaskStore, Column, ScheduleType, ActivityEventType, ModelPreset, MessageType, ParticipantType, RoutineTriggerType, ProjectSettings, EnrichedChatSession } from "@fusion/core";
import { COLUMNS, VALID_TRANSITIONS, GLOBAL_SETTINGS_KEYS, type BatchStatusEntry, type BatchStatusResponse, type BatchStatusResult, type IssueInfo, type PrInfo, type Task, type PiExtensionEntry, type PiExtensionSettings, getCurrentRepo, isGhAuthenticated, AutomationStore, validateBackupSchedule, validateBackupRetention, validateBackupDir, syncBackupRoutine, exportSettings, importSettings, validateImportData, MessageStore, RoutineStore, isWebhookTrigger, resolveMemoryBackend, getMemoryBackendCapabilities, listMemoryBackendTypes, listProjectMemoryFiles, readProjectMemoryFile, readProjectMemoryFileContent, writeProjectMemoryFile, readMemory, writeMemory, searchProjectMemory, isQmdAvailable, installQmd, refreshQmdProjectMemoryIndex, QMD_INSTALL_COMMAND, MemoryBackendError, scheduleQmdProjectMemoryRefresh, discoverPiExtensions, updatePiExtensionDisabledIds, getFusionAgentDir, getLegacyPiAgentDir, ensureMemoryFileWithBackend } from "@fusion/core";
import type { ServerOptions } from "./server.js";
import { GitHubClient, parseBadgeUrl } from "./github.js";
@@ -8770,9 +8770,10 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
if (lastMessage) {
// Truncate content to 100 chars for preview
const content = lastMessage.content || "";
(session as any).lastMessagePreview =
const enriched: EnrichedChatSession = session;
enriched.lastMessagePreview =
content.length > 100 ? content.slice(0, 100) + "…" : content;
(session as any).lastMessageAt = lastMessage.createdAt;
enriched.lastMessageAt = lastMessage.createdAt;
}
}
}

View File

@@ -413,6 +413,9 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
app.use(express.static(clientDir));
}
// Create ChatStore for chat session management (available for SSE event forwarding)
const chatStore = options?.chatStore ?? new ChatStore(store.getFusionDir(), store.getDatabase());
// Rate limiting — stricter limit on SSE connections
app.get("/api/events", rateLimit(RATE_LIMITS.sse), async (req, res) => {
const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
@@ -432,6 +435,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
undefined,
defaultAgentStore,
defaultMessageStore,
chatStore,
)(req, res);
return;
}
@@ -468,6 +472,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
},
agentStore,
messageStore,
chatStore,
)(req, res);
} catch (err: unknown) {
sendErrorResponse(res, 500, err instanceof Error ? err.message : "Failed to open project event stream");
@@ -655,9 +660,6 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
);
}
// Create ChatStore for chat session management
const chatStore = options?.chatStore ?? new ChatStore(store.getFusionDir(), store.getDatabase());
// Create AgentStore for chat prompt enrichment (initialized lazily by ChatManager)
const chatAgentStore = new AgentStore({ rootDir: store.getFusionDir() });

View File

@@ -9,6 +9,7 @@ import type {
MessageStore,
MissionValidatorRun,
FixFeatureCreatedPayload,
ChatStore,
} from "@fusion/core";
import type { AiSessionStore } from "./ai-session-store.js";
@@ -191,6 +192,7 @@ export function createSSE(
options?: CreateSSEOptions,
agentStore?: AgentStore,
messageStore?: MessageStore,
chatStore?: ChatStore,
) {
const { projectId } = options ?? {};
@@ -389,6 +391,27 @@ export function createSSE(
send(`event: message:deleted\ndata: ${JSON.stringify({ id: messageId })}\n\n`);
};
// --- Chat store event handlers ---
const onChatSessionCreated = (session: any) => {
send(`event: chat:session:created\ndata: ${JSON.stringify(session)}\n\n`);
};
const onChatSessionUpdated = (session: any) => {
send(`event: chat:session:updated\ndata: ${JSON.stringify(session)}\n\n`);
};
const onChatSessionDeleted = (sessionId: string) => {
send(`event: chat:session:deleted\ndata: ${JSON.stringify({ id: sessionId })}\n\n`);
};
const onChatMessageAdded = (message: any) => {
send(`event: chat:message:added\ndata: ${JSON.stringify(message)}\n\n`);
};
const onChatMessageDeleted = (messageId: string) => {
send(`event: chat:message:deleted\ndata: ${JSON.stringify({ id: messageId })}\n\n`);
};
// --- Cleanup (all handlers are defined above, safe to reference) ---
let cleaned = false;
@@ -452,6 +475,13 @@ export function createSSE(
messageStore.off("message:read", onMessageRead);
messageStore.off("message:deleted", onMessageDeleted);
}
if (chatStore) {
chatStore.off("chat:session:created", onChatSessionCreated);
chatStore.off("chat:session:updated", onChatSessionUpdated);
chatStore.off("chat:session:deleted", onChatSessionDeleted);
chatStore.off("chat:message:added", onChatMessageAdded);
chatStore.off("chat:message:deleted", onChatMessageDeleted);
}
};
// --- Subscribe ---
@@ -517,6 +547,14 @@ export function createSSE(
messageStore.on("message:deleted", onMessageDeleted);
}
if (chatStore) {
chatStore.on("chat:session:created", onChatSessionCreated);
chatStore.on("chat:session:updated", onChatSessionUpdated);
chatStore.on("chat:session:deleted", onChatSessionDeleted);
chatStore.on("chat:message:added", onChatMessageAdded);
chatStore.on("chat:message:deleted", onChatMessageDeleted);
}
// 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