root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-3430): preserve and consume final assistant done payload

Browse Source 
This merge adds done-payload snapshot handling across the chat system (FN-3430), normalizing how final assistant messages are preserved and consumed in the dashboard hooks, plus it normalizes dashboard mailbox and user identity for inter-agent messaging (FN-3484) and introduces plugin workflow step

Fusion-Task-Id: FN-3430
This commit is contained in:
Fusion
2026-05-05 06:38:21 -07:00
committed by gsxdsm
parent 47ae7783b7
commit b8a32f8765
12 changed files with 373 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
packages/dashboard/app/api/__tests__/legacy-chat-stream.test.ts
View File

@@ -35,24 +35,46 @@ describe("streamChatResponse SSE parser", () => {
);
const textChunks: string[] = [];
const doneIds: string[] = [];
const donePayloads: Array<{ messageId: string; message?: { content: string } }> = [];
streamChatResponse("s-1", "hi", {
onText: (data) => textChunks.push(data),
onDone: (data) => doneIds.push(data.messageId),
onDone: (data) => donePayloads.push(data),
onError: vi.fn(),
});
await vi.waitFor(() => {
expect(textChunks.join("")).toBe("Hello world");
expect(doneIds).toEqual(["msg-1"]);
expect(donePayloads).toEqual([{ messageId: "msg-1" }]);
});
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("handles done events that have no data payload", async () => {
it("parses done payload assistant snapshots when present", async () => {
vi.spyOn(globalThis, "fetch").mockResolvedValue(
new Response(
createChunkedStream([
"event: done\n",
"data: {\"messageId\":\"msg-1\",\"message\":{\"id\":\"msg-1\",\"sessionId\":\"s-1\",\"role\":\"assistant\",\"content\":\"Final reply\",\"thinkingOutput\":null,\"metadata\":null,\"createdAt\":\"2026-01-01T00:00:00.000Z\"}}\n\n",
]),
{ status: 200 },
),
);
const donePayloads: Array<{ messageId: string; message?: { content: string } }> = [];
streamChatResponse("s-1", "hi", {
onDone: (data) => donePayloads.push(data),
onError: vi.fn(),
});
await vi.waitFor(() => {
expect(donePayloads).toEqual([{ messageId: "msg-1", message: { id: "msg-1", sessionId: "s-1", role: "assistant", content: "Final reply", thinkingOutput: null, metadata: null, createdAt: "2026-01-01T00:00:00.000Z" } }]);
});
});
it("handles done events that have no data payload", async () => { vi.spyOn(globalThis, "fetch").mockResolvedValue(
new Response(createChunkedStream(["event: done\n\n"]), { status: 200 }),
);

8
packages/dashboard/app/api/legacy.ts
View File

@@ -7998,7 +7998,7 @@ export function streamChatResponse(
onToolStart?: (data: { toolName: string; args?: Record<string, unknown> }) => void;
onToolEnd?: (data: { toolName: string; isError: boolean; result?: unknown }) => void;
onFallback?: (data: { primaryModel: string; fallbackModel: string; triggerPoint: "session-creation" | "prompt-time" }) => void;
onDone?: (data: { messageId: string }) => void;
onDone?: (data: { messageId: string; message?: ChatMessage }) => void;
onError?: (data: string) => void;
onConnectionStateChange?: (state: StreamConnectionState) => void;
},
@@ -8056,7 +8056,11 @@ export function streamChatResponse(
break;
case "done":
try {
handlers.onDone?.(JSON.parse(rawData));
const parsed = JSON.parse(rawData) as { messageId?: unknown; message?: unknown };
handlers.onDone?.({
messageId: typeof parsed.messageId === "string" ? parsed.messageId : "",
...(parsed.message && typeof parsed.message === "object" ? { message: parsed.message as ChatMessage } : {}),
});
} catch {
handlers.onDone?.({ messageId: "" });
}

98
packages/dashboard/app/hooks/__tests__/useChat.test.ts
View File

@@ -456,6 +456,104 @@ describe("useChat", () => {
});
});
it("uses done payload assistant snapshot when no text chunks were streamed", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
mockFetchChatMessages.mockResolvedValueOnce({ messages: [] });
let doneHandler: ((data: { messageId: string; message?: ChatMessage }) => void) | undefined;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
doneHandler = handlers.onDone as typeof doneHandler;
return { close: vi.fn(), isConnected: () => true };
});
const { result } = renderHook(() => useChat());
await waitFor(() => {
expect(result.current.sessions).toHaveLength(1);
});
act(() => {
result.current.selectSession("session-001");
});
await act(async () => {
result.current.sendMessage("Hello!");
});
act(() => {
doneHandler?.({
messageId: "msg-002",
message: {
id: "msg-002",
sessionId: "session-001",
role: "assistant",
content: "Snapshot reply",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
} as ChatMessage,
});
});
await waitFor(() => {
expect(result.current.messages.at(-1)).toEqual(expect.objectContaining({
id: "msg-002",
role: "assistant",
content: "Snapshot reply",
}));
expect(result.current.isStreaming).toBe(false);
});
});
it("prefers done payload assistant snapshot over streamed text", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
mockFetchChatMessages.mockResolvedValueOnce({ messages: [] });
let textHandler: ((data: string) => void) | undefined;
let doneHandler: ((data: { messageId: string; message?: ChatMessage }) => void) | undefined;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
textHandler = handlers.onText;
doneHandler = handlers.onDone as typeof doneHandler;
return { close: vi.fn(), isConnected: () => true };
});
const { result } = renderHook(() => useChat());
await waitFor(() => {
expect(result.current.sessions).toHaveLength(1);
});
act(() => {
result.current.selectSession("session-001");
});
act(() => {
result.current.sendMessage("Hello!");
textHandler?.("streamed text");
doneHandler?.({
messageId: "msg-003",
message: {
id: "msg-003",
sessionId: "session-001",
role: "assistant",
content: "snapshot wins",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
} as ChatMessage,
});
});
await waitFor(() => {
expect(result.current.messages.at(-1)).toEqual(expect.objectContaining({
id: "msg-003",
content: "snapshot wins",
}));
});
});
it("handles stream errors and surfaces them to the user", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });

86
packages/dashboard/app/hooks/__tests__/useQuickChat.test.ts
View File

@@ -78,6 +78,92 @@ describe("useQuickChat", () => {
await expect(sendResult).resolves.toBeUndefined();
});
it("uses done payload assistant snapshot when no text chunks were streamed", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchResumeChatSession.mockResolvedValue({ session });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
const { result } = renderHook(() => useQuickChat("proj-123"));
await act(async () => {
await result.current.switchSession("agent-001");
});
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
setTimeout(() => {
handlers.onDone?.({
messageId: "msg-001",
message: {
id: "msg-001",
sessionId: "session-001",
role: "assistant",
content: "Snapshot reply",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
} as any,
});
}, 0);
return { close: vi.fn(), isConnected: () => true };
});
await act(async () => {
await result.current.sendMessage("Hello");
});
await waitFor(() => {
expect(result.current.messages.at(-1)).toEqual(expect.objectContaining({
id: "msg-001",
role: "assistant",
content: "Snapshot reply",
}));
});
});
it("prefers done payload assistant snapshot over streamed text", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchResumeChatSession.mockResolvedValue({ session });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
let onText: ((data: string) => void) | undefined;
let onDone: ((data: { messageId: string; message?: any }) => void) | undefined;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
onText = handlers.onText;
onDone = handlers.onDone as typeof onDone;
return { close: vi.fn(), isConnected: () => true };
});
const { result } = renderHook(() => useQuickChat("proj-123"));
await act(async () => {
await result.current.switchSession("agent-001");
});
act(() => {
void result.current.sendMessage("Hello");
onText?.("streamed text");
onDone?.({
messageId: "msg-002",
message: {
id: "msg-002",
sessionId: "session-001",
role: "assistant",
content: "snapshot wins",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
},
});
});
await waitFor(() => {
expect(result.current.messages.at(-1)).toEqual(expect.objectContaining({
id: "msg-002",
content: "snapshot wins",
}));
});
});
it("startModelChat creates a KB session with provider/model override", async () => {
const { result } = renderHook(() => useQuickChat("proj-123"));

25
packages/dashboard/app/hooks/useChat.ts
View File

@@ -651,18 +651,21 @@ export function useChat(
: prev);
addToast?.(`Primary model unavailable. Switched to fallback ${data.fallbackModel}.`, "warning");
},
onDone: (data: { messageId: string }) => {
onDone: (data: { messageId: string; message?: ChatMessage }) => {
cancelStreamingFlushes();
const assistantMessage: ChatMessageInfo = {
id: data.messageId || `msg-${Date.now()}`,
sessionId: activeSession.id,
role: "assistant",
content: capturedText,
thinkingOutput: capturedThinking,
toolCalls: capturedToolCalls.length > 0 ? capturedToolCalls : undefined,
fallbackInfo: capturedFallbackInfo,
createdAt: new Date().toISOString(),
};
const finalMessage = data.message;
const assistantMessage: ChatMessageInfo = finalMessage
? mapChatMessageToInfo(finalMessage)
: {
id: data.messageId || `msg-${Date.now()}`,
sessionId: activeSession.id,
role: "assistant",
content: capturedText,
thinkingOutput: capturedThinking,
toolCalls: capturedToolCalls.length > 0 ? capturedToolCalls : undefined,
fallbackInfo: capturedFallbackInfo,
createdAt: new Date().toISOString(),
};
// Track this message ID so SSE handler skips it if event arrives first
streamingMessageIdsRef.current.add(assistantMessage.id);

25
packages/dashboard/app/hooks/useQuickChat.ts
View File

@@ -661,18 +661,21 @@ export function useQuickChat(
: prev);
addToast?.(`Primary model unavailable. Switched to fallback ${data.fallbackModel}.`, "warning");
},
onDone: (data: { messageId: string }) => {
onDone: (data: { messageId: string; message?: ChatMessage }) => {
cancelStreamingFlushes();
const assistantMessage: ChatMessageInfo = {
id: data.messageId || `msg-${Date.now()}`,
sessionId: activeSession.id,
role: "assistant",
content: capturedText,
thinkingOutput: capturedThinking || undefined,
toolCalls: capturedToolCalls.length > 0 ? capturedToolCalls : undefined,
fallbackInfo: capturedFallbackInfo,
createdAt: new Date().toISOString(),
};
const finalMessage = data.message;
const assistantMessage: ChatMessageInfo = finalMessage
? mapChatMessageToInfo(finalMessage)
: {
id: data.messageId || `msg-${Date.now()}`,
sessionId: activeSession.id,
role: "assistant",
content: capturedText,
thinkingOutput: capturedThinking || undefined,
toolCalls: capturedToolCalls.length > 0 ? capturedToolCalls : undefined,
fallbackInfo: capturedFallbackInfo,
createdAt: new Date().toISOString(),
};
// Preserve user message and add assistant message
setMessages((prev) => [...prev, assistantMessage]);

51
packages/dashboard/src/__tests__/chat-manager.test.ts
View File

@@ -349,6 +349,57 @@ describe("ChatManager.sendMessage", () => {
expect(assistantCall?.[1].content).toBe("Hello world!");
});
it("broadcasts done with persisted assistant message snapshot", async () => {
const events: Array<{ type: string; data: unknown }> = [];
const unsubscribe = chatStreamManager.subscribe("chat-001", (event) => {
events.push(event);
});
__setCreateFnAgent(async () => ({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
state: {
messages: [{ role: "assistant", content: "Final content" }],
},
},
}));
mockChatStore.addMessage.mockReturnValueOnce({ id: "msg-user", role: "user" });
mockChatStore.addMessage.mockReturnValueOnce({
id: "msg-final",
sessionId: "chat-001",
role: "assistant",
content: "Final content",
thinkingOutput: null,
metadata: null,
attachments: undefined,
createdAt: "2026-01-01T00:00:00.000Z",
});
const chatManager = createChatManager();
await chatManager.sendMessage("chat-001", "Hello");
unsubscribe();
expect(events).toContainEqual({
type: "done",
data: {
messageId: "msg-final",
message: {
id: "msg-final",
sessionId: "chat-001",
role: "assistant",
content: "Final content",
thinkingOutput: null,
metadata: null,
attachments: undefined,
createdAt: "2026-01-01T00:00:00.000Z",
},
attachments: undefined,
},
});
});
it("broadcasts tool_start and tool_end SSE events when agent calls tools", async () => {
const events: Array<{ type: string; data: unknown }> = [];

42
packages/dashboard/src/__tests__/chat-routes.test.ts
View File

@@ -1148,6 +1148,48 @@ describe("Chat API Routes", () => {
expect(output).toContain('data: {"toolName":"read","args":{"path":"/foo.ts"}}');
expect(output).toContain('data: {"toolName":"read","isError":false,"result":"file contents"}');
});
it("SSE route forwards done payload with assistant snapshot", async () => {
mockGetSession.mockReturnValue(sampleSession);
const chatModule = await import("../chat.js");
vi.mocked(chatModule.checkRateLimit).mockReturnValue(true);
mockSendMessage.mockImplementation(async (sessionId: string) => {
mockChatStreamManager.broadcast(sessionId, {
type: "done",
data: {
messageId: "msg-final",
message: {
id: "msg-final",
sessionId,
role: "assistant",
content: "Final reply",
thinkingOutput: null,
metadata: null,
createdAt: "2026-01-01T00:00:00.000Z",
},
},
});
});
const req = createSSERequest();
const { res, chunks } = createSSEResponse();
req.body = { content: "Hello" };
req.params = { id: "chat-abc123" };
req.query = {} as any;
req.headers = {} as any;
req.ip = "127.0.0.1";
req.socket = { remoteAddress: "127.0.0.1" } as any;
await invokeSSEHandler(req, res, store, mockChatStore, mockChatManager);
const output = chunks.join("");
expect(output).toContain("event: done");
expect(output).toContain('"messageId":"msg-final"');
expect(output).toContain('"content":"Final reply"');
});
});
});

36
packages/dashboard/src/chat.ts
View File

@@ -146,7 +146,23 @@ export type ChatStreamEvent =
| { type: "tool_start"; data: { toolName: string; args?: Record<string, unknown> } }
| { type: "tool_end"; data: { toolName: string; isError: boolean; result?: unknown } }
| { type: "fallback"; data: { primaryModel: string; fallbackModel: string; triggerPoint: "session-creation" | "prompt-time" } }
| { type: "done"; data: { messageId: string; attachments?: ChatAttachment[] } }
| {
type: "done";
data: {
messageId: string;
message?: {
id: string;
sessionId: string;
role: "assistant";
content: string;
thinkingOutput: string | null;
metadata: Record<string, unknown> | null;
attachments?: ChatAttachment[];
createdAt: string;
};
attachments?: ChatAttachment[];
};
}
| { type: "error"; data: string };
/** Callback function for streaming events */
@@ -987,10 +1003,24 @@ export class ChatManager {
metadata: Object.keys(assistantMetadata).length > 0 ? assistantMetadata : undefined,
});
// Broadcast done event
// Broadcast done event with persisted assistant snapshot so clients can
// render completion even when incremental text deltas were absent.
chatStreamManager.broadcast(sessionId, {
type: "done",
data: { messageId: assistantMessage.id, attachments },
data: {
messageId: assistantMessage.id,
message: {
id: assistantMessage.id,
sessionId: assistantMessage.sessionId,
role: "assistant",
content: assistantMessage.content,
thinkingOutput: assistantMessage.thinkingOutput,
metadata: assistantMessage.metadata,
attachments: assistantMessage.attachments,
createdAt: assistantMessage.createdAt,
},
attachments,
},
});
} catch (err) {
if (abortController.signal.aborted) {

2
packages/dashboard/src/routes/register-chat-routes.ts
View File

@@ -471,7 +471,7 @@ export function registerChatRoutes(ctx: ApiRoutesContext, deps: ChatRouteDeps):
* Event types:
* - thinking: AI thinking output chunks
* - text: AI response text chunks
* - done: Message sent successfully with messageId
* - done: Message sent successfully with messageId + persisted assistant message snapshot
* - error: Error message
*/
router.post("/chat/sessions/:id/messages", rateLimit(RATE_LIMITS.sse), async (req, res) => {