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

FN-016: preserve interrupted chat responses

Browse Source 
Persist partial assistant output when chat streaming is stopped so users can resume from an honest interrupted transcript.

- Snapshot streamed text, thinking, and tool calls before cancellation.
- Persist and reconcile interrupted responses for main chat and planner chat surfaces.
- Add cancellation routes, manager handling, changeset metadata, and desktop/mobile regression coverage.

Files changed:
 .changeset/fn-016-chat-cancel-partial.md           |   7 +
 packages/dashboard/app/api/chat/chat.ts            |  13 +-
 .../app/components/TaskPlannerChatTab.tsx          |  91 ++++++++-
 .../__tests__/ChatView.streaming-thread.test.tsx   |  54 ++++++
 .../__tests__/TaskPlannerChatTab.test.tsx          |  53 +++++-
 .../dashboard/app/hooks/__tests__/useChat.test.ts  |  84 ++++++++
 packages/dashboard/app/hooks/useChat.ts            | 131 +++++++++++--
 .../dashboard/src/__tests__/chat-manager.test.ts   |  80 ++++++--
 packages/dashboard/src/chat.ts                     | 208 ++++++++++++++-----
 .../dashboard/src/routes/register-chat-routes.ts   |   7 +-
 10 files changed, 641 insertions(+), 87 deletions(-)

Fusion-Task-Id: FN-016

Fusion-Task-Lineage: 3e14b9b7-967b-4f80-b941-2216bfa9c69e

Co-authored-by: Fusion <noreply@runfusion.ai>
This commit is contained in:
Fusion Agent
2026-08-18 22:28:33 +00:00
parent 6a28811d1a
commit 0899d49e5c
10 changed files with 641 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-016-chat-cancel-partial.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Preserve partially generated chat replies when an operator stops generation.
category: fix
dev: Direct Chat and task Planner Chat persist interrupted assistant prefixes before cancellation completes.

13
packages/dashboard/app/api/chat/chat.ts
View File

@@ -433,12 +433,17 @@ export function clearChatRoomMessages(
* Do not add streamChatRoomResponse until FN-3810 introduces AI invocation/streaming.
*/
/** Cancel an in-flight chat generation. */
/**
* Cancel an in-flight chat generation and await its durable interrupted-message result.
* FNXC:ChatCancellation 2026-08-18-21:55:
* Stop callers need the persisted assistant prefix before they reconcile the thread or
* release a queued follow-up; the server response is the cancellation barrier.
*/
export function cancelChatResponse(
sessionId: string,
projectId?: string,
): Promise<{ success: boolean }> {
return api<{ success: boolean }>(
): Promise<{ success: boolean; interrupted: boolean; message?: ChatMessage }> {
return api<{ success: boolean; interrupted: boolean; message?: ChatMessage }>(
withProjectId(`/chat/sessions/${encodeURIComponent(sessionId)}/cancel`, projectId),
{
method: "POST",
@@ -547,7 +552,7 @@ export interface ChatStreamHandlers {
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; message?: ChatMessage }) => void;
onDone?: (data: { messageId: string; message?: ChatMessage; interrupted?: boolean }) => void;
onError?: (data: string | ChatFailureInfo, meta?: ChatStreamErrorMeta) => void;
onConnectionStateChange?: (state: StreamConnectionState) => void;
}

91
packages/dashboard/app/components/TaskPlannerChatTab.tsx
View File

@@ -8,7 +8,7 @@ import type { ToastType } from "../hooks/useToast";
import { useComposerDictation } from "../hooks/useComposerDictation";
import { MicButton } from "./MicButton";
import type { ChatMessageInfo, ToolCallInfo } from "../hooks/chatTypes";
import { attachChatStream, editChatMessage, ensureTaskPlannerChatSession, fetchChatMessages, fetchChatSession, fetchTaskDetail, fetchTaskPlannerChatSession, streamChatResponse, type ChatFailureInfo, type ChatStreamErrorMeta } from "../api";
import { attachChatStream, cancelChatResponse, editChatMessage, ensureTaskPlannerChatSession, fetchChatMessages, fetchChatSession, fetchTaskDetail, fetchTaskPlannerChatSession, streamChatResponse, type ChatFailureInfo, type ChatStreamErrorMeta } from "../api";
import { parseQuestionToolCall, type ParsedQuestionToolCall } from "../utils/parseQuestionToolCall";
import { ChatQuestionResponse } from "./ChatQuestionResponse";
import { ProviderIcon } from "./ProviderIcon";
@@ -328,6 +328,14 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
const [historyLoaded, setHistoryLoaded] = useState(false);
const [error, setError] = useState<string | null>(null);
const streamRef = useRef<{ close: () => void } | null>(null);
const streamSnapshotRef = useRef<{
requestId: number;
sessionId: string;
text: string;
thinking: string;
toolCalls: ToolCallInfo[];
} | null>(null);
const cancellationInProgressRef = useRef<Promise<void> | null>(null);
const transcriptRef = useRef<HTMLDivElement | null>(null);
const [isTranscriptAtBottom, setIsTranscriptAtBottom] = useState(true);
const isTranscriptAtBottomRef = useRef(true);
@@ -428,6 +436,16 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
let accumulated = inFlightSnapshot?.streamingText ?? "";
let accumulatedThinking = inFlightSnapshot?.streamingThinking ?? "";
const streamingToolCalls = cloneToolCalls(inFlightSnapshot?.toolCalls);
const updateStreamSnapshot = (): void => {
streamSnapshotRef.current = {
requestId,
sessionId: resolvedSessionId,
text: accumulated,
thinking: accumulatedThinking,
toolCalls: cloneToolCalls(streamingToolCalls),
};
};
updateStreamSnapshot();
/*
* FNXC:TaskDetailPlannerChat 2026-07-15-00:00:
@@ -453,16 +471,19 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
onText: (delta: string) => {
if (!isCurrentStreamRequest()) return;
accumulated += delta;
updateStreamSnapshot();
applyStreamingSnapshot(resolvedSessionId, accumulated, accumulatedThinking, streamingToolCalls);
},
onThinking: (delta: string) => {
if (!isCurrentStreamRequest()) return;
accumulatedThinking += delta;
updateStreamSnapshot();
applyStreamingSnapshot(resolvedSessionId, accumulated, accumulatedThinking, streamingToolCalls);
},
onToolStart: ({ toolName, args }: { toolName: string; args?: Record<string, unknown> }) => {
if (!isCurrentStreamRequest()) return;
streamingToolCalls.push({ toolName, args, isError: false, status: "running" });
updateStreamSnapshot();
applyStreamingSnapshot(resolvedSessionId, accumulated, accumulatedThinking, streamingToolCalls);
},
onToolEnd: ({ toolName, isError, result }: { toolName: string; isError: boolean; result?: unknown }) => {
@@ -481,6 +502,7 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
if (steeringText) {
void refreshTaskAfterSteering();
}
updateStreamSnapshot();
applyStreamingSnapshot(resolvedSessionId, accumulated, accumulatedThinking, streamingToolCalls);
},
onDone: (data: { messageId: string; message?: ChatMessage }) => {
@@ -488,6 +510,7 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
composerStateRef.current = "idle";
setComposerState("idle");
setStreamingThinking("");
streamSnapshotRef.current = null;
streamRef.current = null;
if (data.message) {
setMessages((current) => {
@@ -505,6 +528,7 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
composerStateRef.current = "idle";
setComposerState("idle");
setStreamingThinking("");
streamSnapshotRef.current = null;
streamRef.current = null;
setMessages((current) => {
const withoutStreaming = current.filter((candidate) => candidate.id !== "streaming-assistant");
@@ -846,14 +870,75 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan
}, []);
const stopPlannerStreaming = useCallback(() => {
if (cancellationInProgressRef.current) return;
const snapshot = streamSnapshotRef.current;
if (!snapshot) return;
streamRequestRef.current += 1;
streamRef.current?.close();
streamRef.current = null;
composerStateRef.current = "idle";
setComposerState("idle");
setStreamingThinking("");
setMessages((current) => current.filter((message) => message.id !== "streaming-assistant"));
}, []);
const interruptedLocalId = `interrupted-${snapshot.requestId}`;
const hasInterruptedOutput = Boolean(snapshot.text || snapshot.thinking || snapshot.toolCalls.length > 0);
if (hasInterruptedOutput) {
// FNXC:ChatCancellation 2026-08-18-21:55:
// Planner Stop keeps its displayed prefix as a normal transcript bubble until the
// scoped cancellation response confirms the durable interrupted assistant message.
setMessages((current) => [
...current.filter((message) => message.id !== "streaming-assistant" && message.id !== interruptedLocalId),
{
id: interruptedLocalId,
sessionId: snapshot.sessionId,
role: "assistant",
content: snapshot.text,
thinkingOutput: snapshot.thinking || null,
metadata: snapshot.toolCalls.length > 0 ? { toolCalls: snapshot.toolCalls, interrupted: true } : { interrupted: true },
createdAt: new Date().toISOString(),
},
]);
} else {
setMessages((current) => current.filter((message) => message.id !== "streaming-assistant"));
}
const cancellation = Promise.resolve(cancelChatResponse(snapshot.sessionId, projectId))
.then(async (result) => {
const cancellationResult = result ?? { success: true, interrupted: false };
if (!cancellationResult.success) {
throw new Error("Planner chat cancellation did not complete");
}
let refreshed: ChatMessage[] | null = null;
try {
refreshed = (await fetchChatMessages(snapshot.sessionId, { order: "asc" }, projectId)).messages;
} catch {
// Keep the local interrupted bubble if the history read is temporarily unavailable.
}
const persisted = cancellationResult.message ? [cancellationResult.message] : [];
if (refreshed || persisted.length > 0) {
const reconciled = [
...(refreshed ?? []),
...persisted.filter((message) => !(refreshed ?? []).some((candidate) => candidate.id === message.id)),
];
setMessages((current) => mergePlannerTranscriptWithOptimistic(
current.filter((message) => message.id !== interruptedLocalId),
reconciled,
));
}
})
.catch((cancelError) => {
addToastRef.current(getErrorMessage(cancelError) || t("taskDetail.plannerChat.cancelFailed", "Failed to save the interrupted planner response"), "error");
})
.finally(() => {
streamSnapshotRef.current = null;
if (cancellationInProgressRef.current === cancellation) {
cancellationInProgressRef.current = null;
}
});
cancellationInProgressRef.current = cancellation;
}, [projectId, t]);
const handleKeyDown = useCallback((event: React.KeyboardEvent<HTMLTextAreaElement>) => {
if (showCommandMenu && event.key === "ArrowDown") {

54
packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx
View File

@@ -72,6 +72,8 @@ import * as useChatRoomsModule from "../../hooks/useChatRooms";
const mockFetchChatSessions = vi.mocked(apiModule.fetchChatSessions);
const mockFetchChatSession = vi.mocked(apiModule.fetchChatSession);
const mockFetchChatMessages = vi.mocked(apiModule.fetchChatMessages);
const mockStreamChatResponse = vi.mocked(apiModule.streamChatResponse);
const mockCancelChatResponse = vi.mocked(apiModule.cancelChatResponse);
const mockAttachChatStream = vi.mocked(apiModule.attachChatStream);
const mockGetScopedItem = vi.mocked(projectStorageModule.getScopedItem);
const mockSubscribeSse = vi.mocked(sseBusModule.subscribeSse);
@@ -152,6 +154,8 @@ describe("FN-6599 ChatView streaming prior thread", () => {
mockGetScopedItem.mockReturnValue(undefined);
mockSubscribeSse.mockReturnValue(() => {});
mockFetchChatSession.mockResolvedValue({ session: makeSession({ id: "session-001", agentId: "agent-001" }) });
mockStreamChatResponse.mockReturnValue({ close: vi.fn(), isConnected: () => true });
mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: false });
mockAttachChatStream.mockReturnValue({ close: vi.fn(), isConnected: () => true });
});
@@ -251,6 +255,56 @@ describe("FN-6599 ChatView streaming prior thread", () => {
expect(mockAttachChatStream).toHaveBeenCalledTimes(2);
});
it.each([
["desktop", 1280],
["mobile", 390],
])("FN-016 keeps a direct partial reply after rendered Stop on %s", async (_label, width) => {
Object.defineProperty(window, "innerWidth", { configurable: true, value: width });
window.dispatchEvent(new Event("resize"));
const session = makeSession({ id: "session-stop", agentId: "agent-001" });
const interrupted = makeMessage({
id: "assistant-interrupted",
sessionId: session.id,
role: "assistant",
content: "Distinct direct stopped prefix",
metadata: { interrupted: true },
createdAt: "2026-08-18T21:55:00.000Z",
});
mockGetScopedItem.mockImplementation((key) => key === "kb-chat-active-session" ? session.id : undefined);
mockFetchChatSessions.mockResolvedValue({ sessions: [session] });
mockFetchChatSession.mockResolvedValue({ session });
mockFetchChatMessages
.mockResolvedValueOnce({ messages: [] })
.mockResolvedValue({ messages: [
makeMessage({ id: "user-stop", sessionId: session.id, role: "user", content: "Keep this" }),
interrupted,
] });
mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: true, message: interrupted });
let streamHandlers: any;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
streamHandlers = handlers;
return { close: vi.fn(), isConnected: () => true };
});
const rendered = render(<ChatView projectId="proj-123" addToast={vi.fn()} />);
const input = await screen.findByTestId("chat-input");
fireEvent.change(input, { target: { value: "Keep this" } });
fireEvent.click(await screen.findByTestId("chat-send-btn"));
await waitFor(() => expect(mockStreamChatResponse).toHaveBeenCalledTimes(1));
act(() => streamHandlers?.onText?.("Distinct direct stopped prefix"));
await waitFor(() => expect(screen.getByText("Distinct direct stopped prefix")).toBeInTheDocument());
fireEvent.click(screen.getByTestId("chat-stop-btn"));
await waitFor(() => expect(mockCancelChatResponse).toHaveBeenCalledWith(session.id, "proj-123"));
await waitFor(() => expect(screen.getAllByText("Distinct direct stopped prefix")).toHaveLength(1));
expect(screen.getByTestId("chat-send-btn")).toBeInTheDocument();
expect(screen.queryByTestId("chat-stop-btn")).not.toBeInTheDocument();
rendered.unmount();
render(<ChatView projectId="proj-123" addToast={vi.fn()} />);
expect(await screen.findByText("Distinct direct stopped prefix")).toBeInTheDocument();
});
it.each([
["desktop", 1280],
["mobile", 390],

53
packages/dashboard/app/components/__tests__/TaskPlannerChatTab.test.tsx
View File

@@ -12,7 +12,7 @@ const originalScrollTopDescriptor = Object.getOwnPropertyDescriptor(HTMLElement.
const originalScrollHeightDescriptor = Object.getOwnPropertyDescriptor(HTMLElement.prototype, "scrollHeight");
const originalClientHeightDescriptor = Object.getOwnPropertyDescriptor(HTMLElement.prototype, "clientHeight");
const { mockEnsureTaskPlannerChatSession, mockFetchTaskPlannerChatSession, mockFetchChatSession, mockFetchChatMessages, mockFetchTaskDetail, mockStreamChatResponse, mockAttachChatStream, mockEditChatMessage, mockAddSteeringComment, mockTranslations, mockT } = vi.hoisted(() => {
const { mockEnsureTaskPlannerChatSession, mockFetchTaskPlannerChatSession, mockFetchChatSession, mockFetchChatMessages, mockFetchTaskDetail, mockStreamChatResponse, mockAttachChatStream, mockCancelChatResponse, mockEditChatMessage, mockAddSteeringComment, mockTranslations, mockT } = vi.hoisted(() => {
const translations = new Map<string, string>();
return {
mockEnsureTaskPlannerChatSession: vi.fn(),
@@ -22,6 +22,7 @@ const { mockEnsureTaskPlannerChatSession, mockFetchTaskPlannerChatSession, mockF
mockFetchTaskDetail: vi.fn(),
mockStreamChatResponse: vi.fn(),
mockAttachChatStream: vi.fn(),
mockCancelChatResponse: vi.fn(),
mockEditChatMessage: vi.fn(),
mockAddSteeringComment: vi.fn(),
mockTranslations: translations,
@@ -46,6 +47,7 @@ vi.mock("../../api", async (importOriginal) => {
fetchTaskDetail: mockFetchTaskDetail,
streamChatResponse: mockStreamChatResponse,
attachChatStream: mockAttachChatStream,
cancelChatResponse: mockCancelChatResponse,
editChatMessage: mockEditChatMessage,
addSteeringComment: mockAddSteeringComment,
};
@@ -180,6 +182,7 @@ describe("TaskPlannerChatTab", () => {
mockFetchTaskDetail.mockResolvedValue(makeTask("FN-7310"));
mockStreamChatResponse.mockReturnValue({ close: vi.fn(), isConnected: () => true });
mockAttachChatStream.mockReturnValue({ close: vi.fn(), isConnected: () => true });
mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: false });
mockEditChatMessage.mockResolvedValue({ retained: [] });
mockAddSteeringComment.mockResolvedValue(makeTask("FN-7310"));
});
@@ -220,6 +223,54 @@ describe("TaskPlannerChatTab", () => {
expect(screen.getAllByTestId(/task-planner-chat-starter-/)).toHaveLength(4);
});
it.each([
["desktop", "mouse"],
["mobile", "touch"],
])("FN-016 keeps a planner partial reply after Stop on %s", async (_label, pointerType) => {
let streamHandlers: any;
const interrupted = {
id: "planner-interrupted",
sessionId: "chat-planner",
role: "assistant" as const,
content: "Distinct planner stopped prefix",
thinkingOutput: null,
metadata: { interrupted: true },
createdAt: "2026-08-18T21:55:00.000Z",
};
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
streamHandlers = handlers;
return { close: vi.fn(), isConnected: () => true };
});
const plannerRender = renderPlannerChat();
await screen.findByTestId("task-planner-chat-empty");
mockFetchChatMessages.mockResolvedValue({ messages: [interrupted] });
mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: true, message: interrupted });
await userEvent.click(screen.getByRole("button", { name: /Summarize recent activity/ }));
await waitFor(() => expect(mockStreamChatResponse).toHaveBeenCalledTimes(1));
act(() => streamHandlers?.onText("Distinct planner stopped prefix"));
await screen.findByText("Distinct planner stopped prefix");
const stopButton = screen.getByTestId("chat-stop-btn");
if (pointerType === "touch") {
fireEvent.pointerDown(stopButton, { pointerType: "touch" });
} else {
fireEvent.click(stopButton);
}
await waitFor(() => expect(mockCancelChatResponse).toHaveBeenCalledWith("chat-planner", undefined));
await waitFor(() => expect(screen.getAllByText("Distinct planner stopped prefix")).toHaveLength(1));
expect(screen.getByTestId("chat-send-btn")).toBeInTheDocument();
expect(screen.queryByTestId("chat-stop-btn")).not.toBeInTheDocument();
act(() => streamHandlers?.onText(" stale late callback"));
expect(screen.queryByText("stale late callback")).not.toBeInTheDocument();
plannerRender.unmount();
renderPlannerChat();
expect(await screen.findByText("Distinct planner stopped prefix")).toBeInTheDocument();
});
it("does not create a planner session when no existing history is found on tab activation", async () => {
mockFetchTaskPlannerChatSession.mockResolvedValueOnce({ session: null });

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

@@ -2469,6 +2469,51 @@ describe("useChat", () => {
});
});
it("keeps and reconciles a visible interrupted prefix after Stop", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
const persistedAssistant = makeMessage({
id: "assistant-interrupted",
sessionId: "session-001",
role: "assistant",
content: "Distinct direct prefix",
metadata: { interrupted: true },
createdAt: "2026-08-18T21:55:00.000Z",
});
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
mockFetchChatMessages
.mockResolvedValueOnce({ messages: [] })
.mockResolvedValue({ messages: [
makeMessage({ id: "user-1", sessionId: "session-001", role: "user", content: "Hello" }),
persistedAssistant,
] });
mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: true, message: persistedAssistant });
let streamHandlers: StreamAppendHandlers | undefined;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
streamHandlers = handlers as StreamAppendHandlers;
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.activeSession?.id).toBe("session-001"));
act(() => result.current.sendMessage("Hello"));
await waitFor(() => expect(result.current.isStreaming).toBe(true));
act(() => streamHandlers?.onText("Distinct direct prefix"));
await waitFor(() => expect(result.current.streamingText).toBe("Distinct direct prefix"));
act(() => result.current.stopStreaming());
await waitFor(() => expect(mockCancelChatResponse).toHaveBeenCalledWith("session-001", "proj-123"));
await waitFor(() => {
const assistants = result.current.messages.filter((message) => message.role === "assistant" && message.content === "Distinct direct prefix");
expect(assistants).toHaveLength(1);
expect(result.current.isStreaming).toBe(false);
});
expect(result.current.messages.some((message) => message.failureInfo)).toBe(false);
});
it("stopStreaming with no pendingMessages cancels stream without sending anything", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
@@ -2945,6 +2990,45 @@ describe("useChat", () => {
});
});
it("waits for durable stop reconciliation before starting a queued follow-up", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
const cancellation = createDeferredPromise<{ success: boolean; interrupted: boolean }>();
const reconciliation = createDeferredPromise<{ messages: ChatMessage[] }>();
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
mockFetchChatMessages.mockImplementation(async (_sessionId, options) => (
options?.order === "asc" ? reconciliation.promise : { messages: [] }
));
mockCancelChatResponse.mockReturnValue(cancellation.promise);
mockStreamChatResponse.mockReturnValue({ 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.activeSession?.id).toBe("session-001"));
act(() => result.current.sendMessage("First"));
await waitFor(() => expect(result.current.isStreaming).toBe(true));
act(() => {
result.current.sendMessage("Queued follow-up");
result.current.stopStreaming();
});
expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
await act(async () => {
cancellation.resolve({ success: true, interrupted: false });
await Promise.resolve();
});
expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
await act(async () => {
reconciliation.resolve({ messages: [] });
await Promise.resolve();
});
await waitFor(() => {
expect(mockStreamChatResponse).toHaveBeenCalledTimes(2);
expect(mockStreamChatResponse.mock.calls[1]?.[1]).toBe("Queued follow-up");
});
});
it("stopStreaming sends queued pendingMessages after cancelling the stream", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });

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

@@ -1,4 +1,5 @@
import { useState, useEffect, useCallback, useRef } from "react";
import type { SetStateAction } from "react";
import {
fetchChatSessions,
fetchChatSession,
@@ -485,6 +486,10 @@ export function useChat(
const streamRef = useRef<{ close: () => void } | null>(null);
const lastAttachedGenerationRef = useRef<{ sessionId: string; replayFromEventId: number | null } | null>(null);
const cancelledByUserRef = useRef(false);
const cancellationInProgressRef = useRef<Promise<void> | null>(null);
const streamingTextRef = useRef("");
const streamingThinkingRef = useRef("");
const streamingToolCallsRef = useRef<ToolCallInfo[]>([]);
const pendingMessagesRef = useRef<string[]>([]);
const attachIfGeneratingRef = useRef<(
sessionId: string,
@@ -509,6 +514,31 @@ export function useChat(
activeSessionRef.current = activeSession;
messagesRef.current = messages;
isStreamingRef.current = isStreaming;
streamingTextRef.current = streamingText;
streamingThinkingRef.current = streamingThinking;
streamingToolCallsRef.current = streamingToolCalls;
const updateStreamingText = useCallback((next: SetStateAction<string>) => {
setStreamingText((previous) => {
const resolved = typeof next === "function" ? next(previous) : next;
streamingTextRef.current = resolved;
return resolved;
});
}, []);
const updateStreamingThinking = useCallback((next: SetStateAction<string>) => {
setStreamingThinking((previous) => {
const resolved = typeof next === "function" ? next(previous) : next;
streamingThinkingRef.current = resolved;
return resolved;
});
}, []);
const updateStreamingToolCalls = useCallback((next: SetStateAction<ToolCallInfo[]>) => {
setStreamingToolCalls((previous) => {
const resolved = typeof next === "function" ? next(previous) : next;
streamingToolCallsRef.current = resolved;
return resolved;
});
}, []);
useEffect(() => {
pendingMessagesRef.current = pendingMessages;
@@ -731,6 +761,9 @@ export function useChat(
cancelStreamingFlushesRef.current = null;
pendingMessagesRef.current = [];
setPendingMessages([]);
streamingTextRef.current = "";
streamingThinkingRef.current = "";
streamingToolCallsRef.current = [];
setStreamingText("");
setStreamingThinking("");
setStreamingToolCalls([]);
@@ -868,9 +901,9 @@ export function useChat(
initialText: inFlightGeneration?.streamingText,
initialThinking: inFlightGeneration?.streamingThinking,
initialToolCalls: inFlightGeneration?.toolCalls,
setStreamingText,
setStreamingThinking,
setStreamingToolCalls,
setStreamingText: updateStreamingText,
setStreamingThinking: updateStreamingThinking,
setStreamingToolCalls: updateStreamingToolCalls,
cancelStreamingFlushesRef,
addToast: options?.silent ? undefined : addToast,
onFallbackSession: (data, fallbackSessionId) => {
@@ -925,7 +958,7 @@ export function useChat(
});
streamRef.current = stream;
return true;
}, [addToast, flushPendingMessage, flushPendingMessageAfterAttachedError, hydrateMessagesFromCache, loadMessages, projectId]);
}, [addToast, flushPendingMessage, flushPendingMessageAfterAttachedError, hydrateMessagesFromCache, loadMessages, projectId, updateStreamingText, updateStreamingThinking, updateStreamingToolCalls]);
attachIfGeneratingRef.current = attachIfGenerating;
// Select a session
@@ -1389,7 +1422,8 @@ export function useChat(
}, [activeSession, hasMoreMessages, loadMessages]);
const stopStreaming = useCallback(() => {
if (!activeSession) return;
const session = activeSessionRef.current;
if (!session || cancellationInProgressRef.current) return;
cancelledByUserRef.current = true;
cancelStreamingFlushesRef.current?.();
@@ -1398,17 +1432,86 @@ export function useChat(
streamRef.current = null;
lastAttachedGenerationRef.current = null;
void cancelChatResponse(activeSession.id, projectId).catch(() => {
// Best-effort cancellation; ignore backend errors.
});
const sessionSelectionVersion = activeSessionSelectionRef.current;
const stoppedText = streamingTextRef.current;
const stoppedThinking = streamingThinkingRef.current;
const stoppedToolCalls = streamingToolCallsRef.current;
const interruptedLocalId = `interrupted-${Date.now()}`;
const hasInterruptedOutput = Boolean(stoppedText || stoppedThinking || stoppedToolCalls.length > 0);
if (hasInterruptedOutput) {
// FNXC:ChatCancellation 2026-08-18-21:55:
// Keep the displayed prefix in the transcript while the awaited cancel request
// reconciles its durable PostgreSQL message; a failed request must not erase it.
setMessages((previous) => appendChatMessageChronologically(previous, {
id: interruptedLocalId,
sessionId: session.id,
role: "assistant",
content: stoppedText,
thinkingOutput: stoppedThinking || null,
toolCalls: stoppedToolCalls.length > 0 ? stoppedToolCalls : undefined,
createdAt: new Date().toISOString(),
}));
}
setIsStreaming(false);
isStreamingRef.current = false;
streamingTextRef.current = "";
streamingThinkingRef.current = "";
streamingToolCallsRef.current = [];
setStreamingText("");
setStreamingThinking("");
setStreamingToolCalls([]);
flushPendingMessage();
}, [activeSession, projectId, flushPendingMessage]);
const cancellation = cancelChatResponse(session.id, projectId)
.then(async (result) => {
const cancellationResult = result ?? { success: true, interrupted: false };
if (!cancellationResult.success) {
throw new Error("Chat cancellation did not complete");
}
let refreshedMessages: ChatMessageInfo[] | null = null;
try {
const data = await fetchChatMessages(session.id, { limit: 50, order: "asc" }, projectId);
refreshedMessages = data.messages.map(mapChatMessageToInfo);
} catch {
// The local interrupted row remains recoverable when reconciliation cannot read history.
}
if (activeSessionRef.current?.id !== session.id || activeSessionSelectionRef.current !== sessionSelectionVersion) {
return;
}
const persistedInterruptedMessage = cancellationResult.message
? mapChatMessageToInfo(cancellationResult.message)
: undefined;
if (refreshedMessages || persistedInterruptedMessage) {
const reconciled = [
...(refreshedMessages ?? []),
...(persistedInterruptedMessage && !(refreshedMessages ?? []).some((message) => message.id === persistedInterruptedMessage.id)
? [persistedInterruptedMessage]
: []),
];
setMessages((current) => {
let next = current.filter((message) => message.id !== interruptedLocalId && message.id !== "streaming-assistant");
for (const persisted of reconciled) {
next = reconcileOptimisticSentMessage(next, persisted);
}
return sortChatMessagesChronologically(next);
});
}
flushPendingMessage();
})
.catch(() => {
if (activeSessionRef.current?.id === session.id && activeSessionSelectionRef.current === sessionSelectionVersion) {
addToast?.("Failed to save the interrupted response; it remains visible for recovery.", "error");
}
})
.finally(() => {
if (cancellationInProgressRef.current === cancellation) {
cancellationInProgressRef.current = null;
}
});
cancellationInProgressRef.current = cancellation;
}, [activeSession, addToast, flushPendingMessage, projectId]);
/**
* Send a user message to the active chat session.
@@ -1514,9 +1617,9 @@ export function useChat(
const { handlers } = createChatStreamHandlers({
sessionId: activeSession.id,
tempUserMessageId: tempId,
setStreamingText,
setStreamingThinking,
setStreamingToolCalls,
setStreamingText: updateStreamingText,
setStreamingThinking: updateStreamingThinking,
setStreamingToolCalls: updateStreamingToolCalls,
cancelStreamingFlushesRef,
addToast,
onFallbackSession: (data, sessionId) => {
@@ -1657,7 +1760,7 @@ export function useChat(
onAccepted: () => callbacks?.onAccepted?.(),
}, attachments, projectId);
},
[activeSession, projectId, refreshSessions, addToast, attachIfGenerating, reconnectSessionSilently, flushPendingMessage],
[activeSession, projectId, refreshSessions, addToast, attachIfGenerating, reconnectSessionSilently, flushPendingMessage, updateStreamingText, updateStreamingThinking, updateStreamingToolCalls],
);
sendMessageRef.current = sendMessage;

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

@@ -3349,13 +3349,13 @@ describe("ChatManager.sendMessage", () => {
}
});
it("cancelGeneration returns false when no active generation exists", () => {
it("cancelGeneration returns false when no active generation exists", async () => {
const chatManager = createChatManager();
expect(chatManager.cancelGeneration("chat-001")).toBe(false);
await expect(chatManager.cancelGeneration("chat-001")).resolves.toEqual({ success: false, interrupted: false });
});
it("cancelGeneration returns true and aborts an active generation", () => {
it("cancelGeneration returns true and aborts an active generation", async () => {
const chatManager = createChatManager();
const abortController = new AbortController();
const dispose = vi.fn();
@@ -3363,20 +3363,15 @@ describe("ChatManager.sendMessage", () => {
(chatManager as any).activeGenerations.set("chat-001", {
abortController,
agentResult: { session: { dispose } },
generationId: 1,
cancellationRequested: false,
});
const events: Array<{ type: string; data: unknown }> = [];
const unsubscribe = chatStreamManager.subscribe("chat-001", (event) => {
events.push(event);
});
const result = await chatManager.cancelGeneration("chat-001");
const result = chatManager.cancelGeneration("chat-001");
unsubscribe();
expect(result).toBe(true);
expect(result).toEqual({ success: true, interrupted: false });
expect(abortController.signal.aborted).toBe(true);
expect(dispose).toHaveBeenCalledTimes(1);
expect(events).toContainEqual({ type: "error", data: "Generation cancelled" });
});
it("cancelled generation does not persist assistant message", async () => {
@@ -3405,13 +3400,68 @@ describe("ChatManager.sendMessage", () => {
await new Promise((resolve) => setTimeout(resolve, 0));
expect(chatManager.cancelGeneration("chat-001")).toBe(true);
await expect(chatManager.cancelGeneration("chat-001")).resolves.toEqual({ success: true, interrupted: false });
await sendPromise;
const assistantCalls = mockChatStore.addMessage.mock.calls.filter((call) => call[1].role === "assistant");
expect(assistantCalls).toHaveLength(0);
});
it("persists one interrupted assistant message before clearing the checkpoint", async () => {
let rejectPrompt: ((reason?: unknown) => void) | undefined;
const addMessageCalls: Array<{ role: string; content: string }> = [];
mockChatStore.addMessage.mockImplementation(async (_sessionId: string, input: { role: string; content: string; thinkingOutput?: string; metadata?: Record<string, unknown> }) => {
addMessageCalls.push({ role: input.role, content: input.content });
return {
id: input.role === "user" ? "user-1" : "assistant-interrupted-1",
sessionId: "chat-001",
role: input.role,
content: input.content,
thinkingOutput: input.thinkingOutput ?? null,
metadata: input.metadata ?? null,
createdAt: "2026-08-18T21:55:00.000Z",
};
});
__setCreateFnAgent(async (options: any) => ({
session: {
prompt: vi.fn().mockImplementation(() => {
options.onThinking("thinking prefix");
options.onText("Distinct interrupted prefix");
options.onToolStart("bash", { command: "echo partial" });
return new Promise<void>((_resolve, reject) => {
rejectPrompt = reject;
});
}),
dispose: vi.fn().mockImplementation(() => rejectPrompt?.(new Error("Disposed"))),
state: { messages: [] },
},
}));
const events: Array<{ type: string; data: unknown }> = [];
const unsubscribe = chatStreamManager.subscribe("chat-001", (event) => events.push(event));
const chatManager = createChatManager();
const sendPromise = chatManager.sendMessage("chat-001", "Hello");
await new Promise((resolve) => setTimeout(resolve, 0));
const cancellation = await chatManager.cancelGeneration("chat-001");
await sendPromise;
unsubscribe();
expect(cancellation).toEqual(expect.objectContaining({ success: true, interrupted: true }));
expect(addMessageCalls.filter((call) => call.role === "assistant")).toEqual([
{ role: "assistant", content: "Distinct interrupted prefix" },
]);
expect(mockChatStore.setInFlightGeneration.mock.invocationCallOrder.at(-1)).toBeGreaterThan(
mockChatStore.addMessage.mock.invocationCallOrder.at(-1)!,
);
expect(mockChatStore.setInFlightGeneration).toHaveBeenLastCalledWith("chat-001", null);
expect(events.filter((event) => event.type === "done")).toHaveLength(1);
expect(events.find((event) => event.type === "done")?.data).toEqual(expect.objectContaining({
interrupted: true,
messageId: "assistant-interrupted-1",
}));
});
it("cancelled generation broadcasts error event with cancellation message", async () => {
let rejectPrompt: ((reason?: unknown) => void) | undefined;
@@ -3440,11 +3490,11 @@ describe("ChatManager.sendMessage", () => {
const sendPromise = chatManager.sendMessage("chat-001", "Hello");
await new Promise((resolve) => setTimeout(resolve, 0));
chatManager.cancelGeneration("chat-001");
await chatManager.cancelGeneration("chat-001");
await sendPromise;
unsubscribe();
expect(events.some((event) => event.type === "error" && event.data === "Generation cancelled")).toBe(true);
expect(events.some((event) => event.type === "done" && (event.data as { interrupted?: boolean }).interrupted === true)).toBe(true);
});
it("cleans active generation state even when dispose fails", async () => {

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

@@ -1085,6 +1085,7 @@ export type ChatStreamEvent =
createdAt: string;
};
attachments?: ChatAttachment[];
interrupted?: boolean;
};
}
| { type: "error"; data: string | ChatFailureInfo };
@@ -1414,15 +1415,28 @@ export class RoomReplyGenerationError extends Error {
}
}
interface ChatCancellationResult {
success: boolean;
interrupted: boolean;
message?: ChatMessage;
}
interface ActiveChatGeneration {
abortController: AbortController;
agentResult?: AgentResult;
generationId: number;
cancellationRequested: boolean;
cancellationResult?: ChatCancellationResult;
settled: Promise<void>;
resolveSettled: () => void;
}
export class ChatManager {
private agentStoreReady?: Promise<void>;
private generationCounter = 0;
private inFlightPersistTimers = new Map<string, ReturnType<typeof setTimeout>>();
private activeGenerations = new Map<string, {
abortController: AbortController;
agentResult?: AgentResult;
generationId: number;
}>();
private inFlightPersistChains = new Map<string, Promise<void>>();
private activeGenerations = new Map<string, ActiveChatGeneration>();
constructor(
private chatStore: ChatStore,
@@ -1511,17 +1525,40 @@ export class ChatManager {
its rejection observed so one failed jsonb write cannot become a process-wide
unhandled rejection or interrupt the streaming turn.
*/
private persistInFlightGeneration(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
try {
void this.chatStore.setInFlightGeneration(sessionId, snapshot).catch(() => {
private persistInFlightGeneration(
sessionId: string,
snapshot: ChatInFlightGenerationState | null,
generationId?: number,
): Promise<void> {
const previous = this.inFlightPersistChains.get(sessionId) ?? Promise.resolve();
const write = previous.then(async () => {
// FNXC:ChatCancellation 2026-08-18-21:52:
// Generation-scoped checkpoint writes are serialized so delayed work from an
// interrupted turn cannot clear or overwrite a newer turn's recovery slot.
if (generationId !== undefined && this.activeGenerations.get(sessionId)?.generationId !== generationId) {
return;
}
try {
await this.chatStore.setInFlightGeneration(sessionId, snapshot);
} catch {
diagnostics.warn(`Failed to persist in-flight chat checkpoint for session ${sessionId}`);
});
} catch {
diagnostics.warn(`Failed to persist in-flight chat checkpoint for session ${sessionId}`);
}
}
});
const tracked = write.finally(() => {
if (this.inFlightPersistChains.get(sessionId) === tracked) {
this.inFlightPersistChains.delete(sessionId);
}
});
this.inFlightPersistChains.set(sessionId, tracked);
void tracked.catch(() => undefined);
return tracked;
}
private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
private queueInFlightGenerationPersist(
sessionId: string,
snapshot: ChatInFlightGenerationState | null,
generationId: number,
): void {
const existingTimer = this.inFlightPersistTimers.get(sessionId);
if (existingTimer) {
clearTimeout(existingTimer);
@@ -1529,18 +1566,22 @@ export class ChatManager {
const timer = setTimeout(() => {
this.inFlightPersistTimers.delete(sessionId);
this.persistInFlightGeneration(sessionId, snapshot);
void this.persistInFlightGeneration(sessionId, snapshot, generationId);
}, IN_FLIGHT_PERSIST_DEBOUNCE_MS);
this.inFlightPersistTimers.set(sessionId, timer);
}
private flushInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
private flushInFlightGenerationPersist(
sessionId: string,
snapshot: ChatInFlightGenerationState | null,
generationId?: number,
): Promise<void> {
const existingTimer = this.inFlightPersistTimers.get(sessionId);
if (existingTimer) {
clearTimeout(existingTimer);
this.inFlightPersistTimers.delete(sessionId);
}
this.persistInFlightGeneration(sessionId, snapshot);
return this.persistInFlightGeneration(sessionId, snapshot, generationId);
}
private async getChatModelSettings(): Promise<{
@@ -1671,7 +1712,17 @@ export class ChatManager {
this.generationCounter += 1;
const generationId = this.generationCounter;
const abortController = new AbortController();
this.activeGenerations.set(sessionId, { abortController, generationId });
let resolveSettled!: () => void;
const settled = new Promise<void>((resolve) => {
resolveSettled = resolve;
});
this.activeGenerations.set(sessionId, {
abortController,
generationId,
cancellationRequested: false,
settled,
resolveSettled,
});
return { generationId, abortController };
}
@@ -2306,6 +2357,7 @@ export class ChatManager {
abortController = allocated.abortController;
}
const broadcastOptions = { generationId };
const generationState = this.activeGenerations.get(sessionId);
const session = await this.chatStore.getSession(sessionId);
@@ -2348,6 +2400,7 @@ export class ChatManager {
if (current?.generationId === generationId) {
this.activeGenerations.delete(sessionId);
}
generationState?.resolveSettled?.();
}
return;
}
@@ -2397,7 +2450,7 @@ export class ChatManager {
],
replayFromEventId: lastStreamEventId,
updatedAt: new Date().toISOString(),
});
}, generationId);
};
try {
@@ -2417,7 +2470,7 @@ export class ChatManager {
toolCalls: [],
replayFromEventId: 0,
updatedAt: new Date().toISOString(),
});
}, generationId);
const parsedSkillCommands = parseSkillCommands(content);
@@ -2453,7 +2506,7 @@ export class ChatManager {
void Promise.resolve(emitted).catch(() => undefined);
} catch { /* telemetry must not enter the message-save failure path */ }
} catch (err) {
this.flushInFlightGenerationPersist(sessionId, null);
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
chatStreamManager.broadcast(sessionId, {
type: "error",
data: `Failed to save message: ${err instanceof Error ? err.message : "Unknown error"}`,
@@ -2881,12 +2934,16 @@ export class ChatManager {
...(this.taskStore ? { mcpServers: (await resolveMcpServersForStore(this.taskStore, { agentId: agent?.id })).servers } : {}),
...sessionOptions,
});
this.activeGenerations.set(sessionId, { abortController, agentResult, generationId });
if (abortController.signal.aborted) {
const generationEntry = this.activeGenerations.get(sessionId);
if (!generationEntry || generationEntry.generationId !== generationId) {
agentResult.session.dispose?.();
return;
}
generationEntry.agentResult = agentResult;
if (abortController.signal.aborted) {
throw new Error("Generation cancelled");
}
// Send user message and get response
await enginePromptWithFallback(
@@ -2896,7 +2953,7 @@ export class ChatManager {
);
if (abortController.signal.aborted) {
return;
throw new Error("Generation cancelled");
}
interface AgentMessage {
@@ -2917,7 +2974,7 @@ export class ChatManager {
effectiveModelId,
);
await persistFailureMessage(this.chatStore, sessionId, failureInfo);
this.flushInFlightGenerationPersist(sessionId, null);
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
chatStreamManager.broadcast(sessionId, {
type: "error",
data: failureInfo,
@@ -2985,7 +3042,7 @@ export class ChatManager {
});
}
this.flushInFlightGenerationPersist(sessionId, null);
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
// Broadcast done event with persisted assistant snapshot so clients can
// render completion even when incremental text deltas were absent.
@@ -3007,12 +3064,65 @@ export class ChatManager {
},
}, broadcastOptions);
} catch (err) {
const generationEntry = this.activeGenerations.get(sessionId);
const isExplicitCancellation = abortController.signal.aborted
&& generationEntry?.generationId === generationId
&& generationEntry.cancellationRequested;
if (isExplicitCancellation) {
let interruptedMessage: ChatMessage | undefined;
// FNXC:ChatCancellation 2026-08-18-21:52:
// Stop is a durable conversation transition: save the visible prefix before
// clearing its checkpoint so the next model turn and reload see the same context.
if (accumulatedText || accumulatedThinking || toolCallsAccum.length > 0) {
try {
interruptedMessage = await this.chatStore.addMessage(sessionId, {
role: "assistant",
content: accumulatedText,
thinkingOutput: accumulatedThinking || undefined,
metadata: {
interrupted: true,
...(fallbackInfo ? { fallback: fallbackInfo } : {}),
...(toolCallsAccum.length > 0 ? { toolCalls: toolCallsAccum } : {}),
},
});
} catch (persistErr) {
diagnostics.error(`Failed to persist interrupted response for session ${sessionId}:`, persistErr);
}
}
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
const current = this.activeGenerations.get(sessionId);
if (current?.generationId === generationId) {
current.cancellationResult = {
success: true,
interrupted: Boolean(interruptedMessage),
...(interruptedMessage ? { message: interruptedMessage } : {}),
};
chatStreamManager.broadcast(sessionId, {
type: "done",
data: {
messageId: interruptedMessage?.id ?? "",
...(interruptedMessage ? {
message: {
id: interruptedMessage.id,
sessionId: interruptedMessage.sessionId,
role: "assistant" as const,
content: interruptedMessage.content,
thinkingOutput: interruptedMessage.thinkingOutput,
metadata: interruptedMessage.metadata,
attachments: interruptedMessage.attachments,
createdAt: interruptedMessage.createdAt,
},
} : {}),
interrupted: true,
},
}, broadcastOptions);
}
return;
}
if (abortController.signal.aborted) {
this.flushInFlightGenerationPersist(sessionId, null);
chatStreamManager.broadcast(sessionId, {
type: "error",
data: "Generation cancelled",
}, broadcastOptions);
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
return;
}
@@ -3045,7 +3155,7 @@ export class ChatManager {
diagnostics.error(`Failed to persist failure message for session ${sessionId}:`, persistErr);
}
this.flushInFlightGenerationPersist(sessionId, null);
await this.flushInFlightGenerationPersist(sessionId, null, generationId);
chatStreamManager.broadcast(sessionId, {
type: "error",
@@ -3060,6 +3170,7 @@ export class ChatManager {
if (stillOwnsSlot) {
this.activeGenerations.delete(sessionId);
}
generationState?.resolveSettled?.();
// Dispose the agent session — but ONLY when we still own the slot.
//
@@ -3084,30 +3195,31 @@ export class ChatManager {
}
}
cancelGeneration(sessionId: string): boolean {
async cancelGeneration(sessionId: string): Promise<ChatCancellationResult> {
const entry = this.activeGenerations.get(sessionId);
if (!entry) {
return false;
return { success: false, interrupted: false };
}
entry.abortController.abort();
if (!entry.cancellationRequested) {
entry.cancellationRequested = true;
entry.abortController.abort();
if (entry.agentResult) {
try {
entry.agentResult.session.dispose?.();
} catch (err) {
diagnostics.error(`Error disposing agent session during cancellation:`, err);
if (entry.agentResult) {
try {
entry.agentResult.session.dispose?.();
} catch (err) {
diagnostics.error(`Error disposing agent session during cancellation:`, err);
}
}
}
this.flushInFlightGenerationPersist(sessionId, null);
chatStreamManager.broadcast(sessionId, {
type: "error",
data: "Generation cancelled",
}, { generationId: entry.generationId });
return true;
// The send loop owns persistence and its terminal SSE event. Waiting here
// makes the HTTP cancel response a durable reconciliation barrier for clients.
if (entry.settled) {
await entry.settled;
}
return entry.cancellationResult ?? { success: true, interrupted: false };
}
/**

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

@@ -1124,8 +1124,11 @@ export function registerChatRoutes(ctx: ApiRoutesContext, deps: ChatRouteDeps):
try {
const chatManager = await resolveScopedChatManager(req.query.projectId as string | undefined);
const sessionId = String(req.params.id);
const success = chatManager.cancelGeneration(sessionId);
res.json({ success });
// FNXC:ChatCancellation 2026-08-18-21:52:
// Await cancellation so clients only reconcile or dequeue follow-up sends after
// the interrupted assistant prefix and checkpoint cleanup are durable.
const result = await chatManager.cancelGeneration(sessionId);
res.json(result);
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;