From 0899d49e5c799fff4d51ac06603de63f8cb50dea Mon Sep 17 00:00:00 2001 From: Fusion Agent Date: Tue, 18 Aug 2026 22:28:33 +0000 Subject: [PATCH] FN-016: preserve interrupted chat responses 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 --- .changeset/fn-016-chat-cancel-partial.md | 7 + packages/dashboard/app/api/chat/chat.ts | 13 +- .../app/components/TaskPlannerChatTab.tsx | 91 +++++++- .../ChatView.streaming-thread.test.tsx | 54 +++++ .../__tests__/TaskPlannerChatTab.test.tsx | 53 ++++- .../app/hooks/__tests__/useChat.test.ts | 84 +++++++ packages/dashboard/app/hooks/useChat.ts | 131 +++++++++-- .../src/__tests__/chat-manager.test.ts | 80 +++++-- packages/dashboard/src/chat.ts | 208 ++++++++++++++---- .../src/routes/register-chat-routes.ts | 7 +- 10 files changed, 641 insertions(+), 87 deletions(-) create mode 100644 .changeset/fn-016-chat-cancel-partial.md diff --git a/.changeset/fn-016-chat-cancel-partial.md b/.changeset/fn-016-chat-cancel-partial.md new file mode 100644 index 0000000000..ff1ba39e2a --- /dev/null +++ b/.changeset/fn-016-chat-cancel-partial.md @@ -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. diff --git a/packages/dashboard/app/api/chat/chat.ts b/packages/dashboard/app/api/chat/chat.ts index 3c3a11a999..4a7af165c7 100644 --- a/packages/dashboard/app/api/chat/chat.ts +++ b/packages/dashboard/app/api/chat/chat.ts @@ -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 }) => 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; } diff --git a/packages/dashboard/app/components/TaskPlannerChatTab.tsx b/packages/dashboard/app/components/TaskPlannerChatTab.tsx index 8ee3e6688c..fda6b2119f 100644 --- a/packages/dashboard/app/components/TaskPlannerChatTab.tsx +++ b/packages/dashboard/app/components/TaskPlannerChatTab.tsx @@ -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(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 | null>(null); const transcriptRef = useRef(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 }) => { 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) => { if (showCommandMenu && event.key === "ArrowDown") { diff --git a/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx b/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx index c10b6edce6..1090b91888 100644 --- a/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx +++ b/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx @@ -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(); + 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(); + expect(await screen.findByText("Distinct direct stopped prefix")).toBeInTheDocument(); + }); + it.each([ ["desktop", 1280], ["mobile", 390], diff --git a/packages/dashboard/app/components/__tests__/TaskPlannerChatTab.test.tsx b/packages/dashboard/app/components/__tests__/TaskPlannerChatTab.test.tsx index 700bf6f717..d04a350b71 100644 --- a/packages/dashboard/app/components/__tests__/TaskPlannerChatTab.test.tsx +++ b/packages/dashboard/app/components/__tests__/TaskPlannerChatTab.test.tsx @@ -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(); 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 }); diff --git a/packages/dashboard/app/hooks/__tests__/useChat.test.ts b/packages/dashboard/app/hooks/__tests__/useChat.test.ts index c83727b198..6d4c303d29 100644 --- a/packages/dashboard/app/hooks/__tests__/useChat.test.ts +++ b/packages/dashboard/app/hooks/__tests__/useChat.test.ts @@ -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] }); diff --git a/packages/dashboard/app/hooks/useChat.ts b/packages/dashboard/app/hooks/useChat.ts index 34e4748277..a589d5faa0 100644 --- a/packages/dashboard/app/hooks/useChat.ts +++ b/packages/dashboard/app/hooks/useChat.ts @@ -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 | null>(null); + const streamingTextRef = useRef(""); + const streamingThinkingRef = useRef(""); + const streamingToolCallsRef = useRef([]); const pendingMessagesRef = useRef([]); 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) => { + setStreamingText((previous) => { + const resolved = typeof next === "function" ? next(previous) : next; + streamingTextRef.current = resolved; + return resolved; + }); + }, []); + const updateStreamingThinking = useCallback((next: SetStateAction) => { + setStreamingThinking((previous) => { + const resolved = typeof next === "function" ? next(previous) : next; + streamingThinkingRef.current = resolved; + return resolved; + }); + }, []); + const updateStreamingToolCalls = useCallback((next: SetStateAction) => { + 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; diff --git a/packages/dashboard/src/__tests__/chat-manager.test.ts b/packages/dashboard/src/__tests__/chat-manager.test.ts index ce36ec20bc..68025435d1 100644 --- a/packages/dashboard/src/__tests__/chat-manager.test.ts +++ b/packages/dashboard/src/__tests__/chat-manager.test.ts @@ -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 }) => { + 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((_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 () => { diff --git a/packages/dashboard/src/chat.ts b/packages/dashboard/src/chat.ts index 7cd5efda10..b623f24756 100644 --- a/packages/dashboard/src/chat.ts +++ b/packages/dashboard/src/chat.ts @@ -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; + resolveSettled: () => void; +} + export class ChatManager { private agentStoreReady?: Promise; private generationCounter = 0; private inFlightPersistTimers = new Map>(); - private activeGenerations = new Map(); + private inFlightPersistChains = new Map>(); + private activeGenerations = new Map(); 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 { + 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 { 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((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 { 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 }; } /** diff --git a/packages/dashboard/src/routes/register-chat-routes.ts b/packages/dashboard/src/routes/register-chat-routes.ts index 6f816d32ac..ca4092fcde 100644 --- a/packages/dashboard/src/routes/register-chat-routes.ts +++ b/packages/dashboard/src/routes/register-chat-routes.ts @@ -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;