diff --git a/.changeset/fn-028-chat-interrupted-context.md b/.changeset/fn-028-chat-interrupted-context.md new file mode 100644 index 0000000000..49905e18e7 --- /dev/null +++ b/.changeset/fn-028-chat-interrupted-context.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Preserve streamed Direct, Quick, and Planner Chat prefixes after Stop. +category: fix +dev: Explicit cancellation now records one interrupted assistant turn and keeps its text in the reopened model session context. diff --git a/docs/dashboard-guide.md b/docs/dashboard-guide.md index 7971a44046..74f8c20442 100644 --- a/docs/dashboard-guide.md +++ b/docs/dashboard-guide.md @@ -740,6 +740,8 @@ Mailbox Inbox, Outbox, and agent lists exclude archived correspondence and unrea - A small **Brain**-icon button next to the composer's attach button lets you change an already-created direct chat session's target and thinking level mid-conversation, without starting a new chat. Its **Model / Agent** section can switch the session to another model via the shared model picker or to a real agent from the agent list; its **Thinking level** section still lists the six thinking levels plus **Default** (clear/inherit, labeled with the current resolved default such as **Default (medium)**). Each selection persists immediately and applies starting with the session's next send, including on mobile/tablet touch viewports and narrow floating Chat windows or compact docks where the popup stays fitted to the chat surface. This control appears only for non-CLI Direct sessions — it is not shown for CLI-agent-backed sessions or in Chat Rooms, neither of which support this per-session retargeting control. - Full Chat and Quick Chat both consume the same streamed `/api/chat/sessions/:id/messages` response contract, and both now prefer the authoritative assistant `message` snapshot on `done` while still accumulating `text` chunks when present (so providers without incremental text streaming still render output immediately) + +- Stopping a Direct/Quick Chat or task Planner model-loop response retains any non-empty text already streamed as one interrupted assistant conversation message, including after refresh/remount and in the next turn's file-backed model context. Chat Rooms and CLI-agent-backed chat sessions are excluded from this model-loop continuity contract. - Final assistant messages with no text, tool calls, thinking output, attachments, or failure details render a muted **No message** placeholder instead of a blank bubble. In-progress responses still use the existing **Working…** / **Thinking…** streaming state until the run finishes. - In-progress assistant responses now survive refresh/navigation while generation is still active: Chat restores the last durable in-flight text/thinking/tool state immediately, keeps the prior persisted conversation visible, then resumes streaming from the stored replay point; any new text, thinking, or tool-call updates append to that restored bubble instead of replacing it or starting from an empty "Working…" placeholder. diff --git a/packages/dashboard/app/components/TaskPlannerChatTab.tsx b/packages/dashboard/app/components/TaskPlannerChatTab.tsx index a3ac4c0e48..242ba2f018 100644 --- a/packages/dashboard/app/components/TaskPlannerChatTab.tsx +++ b/packages/dashboard/app/components/TaskPlannerChatTab.tsx @@ -1071,8 +1071,17 @@ export function TaskPlannerChatTab({ task, columnFlags, projectId, active, expan ...refreshed, ...persisted.filter((message) => !refreshed.some((candidate) => candidate.id === message.id)), ]; + const hasDurableInterruptedMessage = Boolean(cancellationResult.message) + || reconciled.some((message) => + message.role === "assistant" + && message.content === snapshot.text + && message.metadata?.interrupted === true, + ); setMessages((current) => mergePlannerTranscriptWithOptimistic( - current.filter((message) => message.id !== interruptedLocalId), + current.filter((message) => + message.id !== "streaming-assistant" + && (!hasDurableInterruptedMessage || message.id !== interruptedLocalId), + ), reconciled, )); diff --git a/packages/dashboard/app/hooks/__tests__/useChat.test.ts b/packages/dashboard/app/hooks/__tests__/useChat.test.ts index ac16b3036b..a000732426 100644 --- a/packages/dashboard/app/hooks/__tests__/useChat.test.ts +++ b/packages/dashboard/app/hooks/__tests__/useChat.test.ts @@ -2527,6 +2527,34 @@ describe("useChat", () => { expect(result.current.messages.some((message) => message.failureInfo)).toBe(false); }); + it("retains the optimistic interrupted prefix when history has no matching durable row", async () => { + const session = makeSession({ id: "session-001", agentId: "agent-001" }); + mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] }); + mockFetchChatMessages.mockResolvedValue({ messages: [] }); + mockCancelChatResponse.mockResolvedValue({ success: true, interrupted: true }); + + 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 retained prefix")); + await waitFor(() => expect(result.current.streamingText).toBe("Distinct retained prefix")); + + act(() => result.current.stopStreaming()); + await waitFor(() => expect(mockCancelChatResponse).toHaveBeenCalledWith("session-001", "proj-123")); + await waitFor(() => { + expect(result.current.messages.filter((message) => message.content === "Distinct retained prefix")).toHaveLength(1); + }); + }); + it("stopStreaming with no pendingMessages cancels stream without sending anything", async () => { const session = makeSession({ id: "session-001", agentId: "agent-001" }); mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] }); diff --git a/packages/dashboard/app/hooks/chatTypes.ts b/packages/dashboard/app/hooks/chatTypes.ts index db6522c61d..8cff8a959f 100644 --- a/packages/dashboard/app/hooks/chatTypes.ts +++ b/packages/dashboard/app/hooks/chatTypes.ts @@ -43,6 +43,11 @@ export interface ChatMessageInfo { toolCalls?: ToolCallInfo[]; fallbackInfo?: FallbackInfo; failureInfo?: FailureInfo; + /** + * FNXC:ChatCancellation 2026-08-19-05:20: + * Retain server metadata so interrupted-stop reconciliation can distinguish a durable row from an older identical reply. + */ + metadata?: Record | null; attachments?: Array<{ id: string; filename: string; diff --git a/packages/dashboard/app/hooks/useChat.ts b/packages/dashboard/app/hooks/useChat.ts index 32ad3b594d..4a0545ecbd 100644 --- a/packages/dashboard/app/hooks/useChat.ts +++ b/packages/dashboard/app/hooks/useChat.ts @@ -348,6 +348,7 @@ function mapChatMessageToInfo(message: ChatMessage): ChatMessageInfo { toolCalls: extractCompletedToolCalls(message.metadata), fallbackInfo: extractFallbackInfo(message.metadata), failureInfo: extractFailureInfo(message.metadata), + ...(message.metadata ? { metadata: message.metadata } : {}), attachments: message.attachments, createdAt: message.createdAt, }; @@ -1488,8 +1489,19 @@ export function useChat( ? [persistedInterruptedMessage] : []), ]; + const hasDurableInterruptedMessage = Boolean(persistedInterruptedMessage) + || reconciled.some((message) => + message.role === "assistant" + && message.content === stoppedText + && message.metadata?.interrupted === true, + ); setMessages((current) => { - let next = current.filter((message) => message.id !== interruptedLocalId && message.id !== "streaming-assistant"); + // FNXC:ChatCancellation 2026-08-19-05:20: + // A successful but incomplete history read must not erase the optimistic prefix. Remove it only after the cancel response or history proves the interrupted row exists; otherwise merge durable history around the still-visible local recovery bubble. + let next = current.filter((message) => + message.id !== "streaming-assistant" + && (!hasDurableInterruptedMessage || message.id !== interruptedLocalId), + ); for (const persisted of reconciled) { next = reconcileOptimisticSentMessage(next, persisted); } diff --git a/packages/dashboard/src/__tests__/chat-manager-rewind-session.test.ts b/packages/dashboard/src/__tests__/chat-manager-rewind-session.test.ts index e68f66b498..48ae5d2725 100644 --- a/packages/dashboard/src/__tests__/chat-manager-rewind-session.test.ts +++ b/packages/dashboard/src/__tests__/chat-manager-rewind-session.test.ts @@ -19,12 +19,16 @@ truncation semantics themselves are covered against real PostgreSQL in packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts; the seam under test here is the pi session branch/repoint behavior, which is store-agnostic. */ -import { describe, it, expect, beforeAll, afterAll } from "vitest"; +import { describe, it, expect, beforeAll, afterAll, vi } from "vitest"; import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { rm } from "node:fs/promises"; -import { ChatManager } from "../chat.js"; +import { + ChatManager, + __resetChatState, + __setCreateResolvedAgentSession, +} from "../chat.js"; import type { ChatMessage, ChatSession } from "@fusion/core"; import { SessionManager } from "@earendil-works/pi-coding-agent"; @@ -98,14 +102,22 @@ class FakeChatStore { return this.sessions.get(id); } - async addMessage(sessionId: string, input: { role: "user" | "assistant"; content: string }): Promise { + async addMessage( + sessionId: string, + input: { + role: "user" | "assistant"; + content: string; + thinkingOutput?: string; + metadata?: Record; + }, + ): Promise { const message: ChatMessage = { id: `msg-${++this.counter}`, sessionId, role: input.role, content: input.content, - thinkingOutput: null, - metadata: null, + thinkingOutput: input.thinkingOutput ?? null, + metadata: input.metadata ?? null, createdAt: new Date().toISOString(), }; this.messages.push(message); @@ -231,6 +243,49 @@ describe("ChatManager.rewindSessionForEdit — pi session context seam (real Ses expect(oldFileStillHasDiscardedTurn).toContain("second turn"); }); + it("persists an interrupted prefix into the reopened pi context exactly once", async () => { + __resetChatState(); + const session = chatStore.createSession({ agentId: "agent-001" }); + session.title = "Existing title"; + const seedManager = SessionManager.create(tmpDir); + await chatStore.setCliSessionFile(session.id, seedManager.getSessionFile()!); + + const prefix = "Distinct interrupted prefix"; + let rejectPrompt: ((reason?: unknown) => void) | undefined; + __setCreateResolvedAgentSession(async (options: any) => ({ + session: { + prompt: vi.fn().mockImplementation(async () => { + options.onThinking?.("thinking prefix"); + options.onText?.(prefix); + return new Promise((_resolve, reject) => { + rejectPrompt = reject; + }); + }), + dispose: vi.fn().mockImplementation(() => rejectPrompt?.(new Error("disposed"))), + state: { messages: [] }, + }, + }) as any); + + const sendPromise = chatManager.sendMessage(session.id, "hello"); + await new Promise((resolve) => setImmediate(resolve)); + const cancellation = await chatManager.cancelGeneration(session.id); + await sendPromise; + + expect(cancellation).toEqual(expect.objectContaining({ success: true, interrupted: true })); + const assistantRows = (await chatStore.getMessages(session.id)).filter((message) => message.role === "assistant"); + expect(assistantRows).toHaveLength(1); + expect(assistantRows[0]).toEqual(expect.objectContaining({ + content: prefix, + metadata: expect.objectContaining({ interrupted: true }), + })); + + const reopened = SessionManager.open((await chatStore.getSession(session.id))!.cliSessionFile!); + const contextTexts = extractText(reopened.buildSessionContext()); + expect(contextTexts.filter((text) => text === prefix)).toHaveLength(1); + + __resetChatState(); + }); + it("falls back to a rebuilt retained session when branch materialization fails", async () => { const session = chatStore.createSession({ agentId: "agent-001" }); const seedManager = SessionManager.create(tmpDir); diff --git a/packages/dashboard/src/__tests__/chat-manager.test.ts b/packages/dashboard/src/__tests__/chat-manager.test.ts index e9ac1496bb..49424ca27e 100644 --- a/packages/dashboard/src/__tests__/chat-manager.test.ts +++ b/packages/dashboard/src/__tests__/chat-manager.test.ts @@ -64,6 +64,7 @@ const { mockSessionManagerCreate, mockSessionManagerOpen } = vi.hoisted(() => { branch: () => {}, resetLeaf: () => {}, appendMessage: () => "entry-fake", + buildSessionContext: () => ({ messages: [] }), createBranchedSession: () => "/tmp/test/.pi-fake/session-branched.jsonl", }; return { diff --git a/packages/dashboard/src/__tests__/routes-chat-cancellation.test.ts b/packages/dashboard/src/__tests__/routes-chat-cancellation.test.ts new file mode 100644 index 0000000000..cd904a8107 --- /dev/null +++ b/packages/dashboard/src/__tests__/routes-chat-cancellation.test.ts @@ -0,0 +1,113 @@ +// @vitest-environment node + +import express from "express"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { request } from "../test-request.js"; +import { registerChatRoutes } from "../routes/register-chat-routes.js"; + +const { mockResolveProjectChatContext, mockGetOrCreateScopedChatManager } = vi.hoisted(() => ({ + mockResolveProjectChatContext: vi.fn(), + mockGetOrCreateScopedChatManager: vi.fn(), +})); + +vi.mock("../chat-project-services.js", () => ({ + resolveProjectChatContext: mockResolveProjectChatContext, + getOrCreateScopedChatManager: mockGetOrCreateScopedChatManager, +})); + +function makeApp(manager: { cancelGeneration: ReturnType }) { + const router = express.Router(); + const app = express(); + app.use(express.json()); + app.use("/api", router); + + registerChatRoutes({ + router, + store: {} as any, + options: { chatManager: manager } as any, + runtimeLogger: {} as any, + planningLogger: {} as any, + chatLogger: { error: vi.fn(), warn: vi.fn(), log: vi.fn() } as any, + getProjectIdFromRequest: () => undefined, + getScopedStore: vi.fn(), + getProjectContext: vi.fn(), + getProjectPluginLoader: vi.fn(), + prioritizeProjectsForCurrentDirectory: (projects: unknown[]) => projects, + emitRemoteRouteDiagnostic: vi.fn(), + emitAuthSyncAuditLog: vi.fn(), + parseScopeParam: vi.fn(), + resolveAutomationStore: vi.fn(), + resolveRoutineStore: vi.fn(), + resolveRoutineRunner: vi.fn(), + registerDispose: vi.fn(), + dispose: vi.fn(), + rethrowAsApiError: (error: unknown): never => { throw error; }, + } as any, { + parseLastEventId: () => undefined, + replayBufferedSSE: () => false, + validateOptionalModelField: () => undefined, + upload: { + single: () => (_req: unknown, _res: unknown, next: () => void) => next(), + array: () => (_req: unknown, _res: unknown, next: () => void) => next(), + } as any, + }); + + return app; +} + +describe("POST /api/chat/sessions/:id/cancel", () => { + beforeEach(() => { + mockResolveProjectChatContext.mockResolvedValue({ store: {}, chatStore: {} }); + mockGetOrCreateScopedChatManager.mockImplementation(() => currentManager); + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + let currentManager: { cancelGeneration: ReturnType }; + + it("awaits the scoped manager's durable result before responding", async () => { + let resolveCancellation!: (value: unknown) => void; + const cancellation = new Promise((resolve) => { resolveCancellation = resolve; }); + currentManager = { cancelGeneration: vi.fn(() => cancellation) }; + const app = makeApp(currentManager); + + let responseSettled = false; + const responsePromise = request(app, "POST", "/api/chat/sessions/chat-1/cancel?projectId=project-a") + .then((response) => { + responseSettled = true; + return response; + }); + + await new Promise((resolve) => setImmediate(resolve)); + expect(responseSettled).toBe(false); + expect(mockResolveProjectChatContext).toHaveBeenCalledWith(expect.objectContaining({ projectId: "project-a" })); + + resolveCancellation({ + success: true, + interrupted: true, + message: { id: "assistant-1", content: "partial", metadata: { interrupted: true } }, + }); + const response = await responsePromise; + + expect(response.status).toBe(200); + expect(response.body).toEqual(expect.objectContaining({ + success: true, + interrupted: true, + message: expect.objectContaining({ id: "assistant-1" }), + })); + expect(currentManager.cancelGeneration).toHaveBeenCalledWith("chat-1"); + }); + + it("returns no invented message when the scoped session is idle", async () => { + currentManager = { + cancelGeneration: vi.fn().mockResolvedValue({ success: false, interrupted: false }), + }; + const response = await request(makeApp(currentManager), "POST", "/api/chat/sessions/chat-idle/cancel?projectId=project-a"); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ success: false, interrupted: false }); + expect(response.body.message).toBeUndefined(); + }); +}); diff --git a/packages/dashboard/src/chat.ts b/packages/dashboard/src/chat.ts index 9dde3aeac7..c488634b83 100644 --- a/packages/dashboard/src/chat.ts +++ b/packages/dashboard/src/chat.ts @@ -1601,6 +1601,54 @@ export class ChatManager { return this.persistInFlightGeneration(sessionId, snapshot, generationId); } + /* + * FNXC:ChatCancellation 2026-08-19-05:20: + * An explicit Stop must write the streamed textual prefix to the same file-backed pi session that the next turn reopens. PostgreSQL history alone cannot restore model context, while appending a second assistant entry would make the model see the prefix twice when pi already recorded it during cancellation. + */ + private persistInterruptedSessionContext( + sessionManager: SessionManager | undefined, + session: ChatSession | null | undefined, + text: string, + ): void { + if (!text) { + return; + } + if (!sessionManager || !session) { + throw new Error("Interrupted chat context has no file-backed session"); + } + + const context = sessionManager.buildSessionContext(); + const lastAssistant = [...context.messages].reverse().find((message) => message.role === "assistant"); + const lastAssistantText = lastAssistant && Array.isArray(lastAssistant.content) + ? lastAssistant.content + .filter((part): part is { type: "text"; text: string } => part?.type === "text" && typeof part.text === "string") + .map((part) => part.text) + .join("") + : typeof lastAssistant?.content === "string" ? lastAssistant.content : ""; + + if (lastAssistantText === text) { + return; + } + + sessionManager.appendMessage({ + role: "assistant", + content: [{ type: "text", text }], + api: "chat", + provider: session.modelProvider ?? "unknown", + model: session.modelId ?? "unknown", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }); + } + private async getChatModelSettings(): Promise<{ fallbackProvider?: string; fallbackModelId?: string; @@ -2447,6 +2495,7 @@ export class ChatManager { }; const toolCallsAccum: ToolCallRecord[] = []; const pendingToolStarts = new Map }>>(); + let sessionManager: SessionManager | undefined; let fallbackInfo: | { primaryModel: string; fallbackModel: string; triggerPoint: "session-creation" | "prompt-time" } | undefined; @@ -2708,7 +2757,7 @@ export class ChatManager { // the Claude CLI --resume session it owns) is keyed off the chat. On the // first user message we create a fresh, file-backed session and persist // its path; subsequent messages reopen the same file. - const sessionManager = await this.resolveCliSessionManager(session); + sessionManager = await this.resolveCliSessionManager(session); /* * FNXC:ChatMessageEdit 2026-07-07-09:00: @@ -3100,10 +3149,18 @@ export class ChatManager { && 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. + let interruptionDurable = true; + // FNXC:ChatCancellation 2026-08-19-05:20: + // Stop is a durable conversation transition: save the visible prefix to both the PostgreSQL transcript and the reopened pi session before clearing its checkpoint. A failed durable write keeps the checkpoint available for recovery and reports failure so clients retain their local prefix. if (accumulatedText || accumulatedThinking || toolCallsAccum.length > 0) { + if (accumulatedText) { + try { + this.persistInterruptedSessionContext(sessionManager, session, accumulatedText); + } catch (persistErr) { + interruptionDurable = false; + diagnostics.error(`Failed to persist interrupted pi context for session ${sessionId}:`, persistErr); + } + } try { interruptedMessage = await this.chatStore.addMessage(sessionId, { role: "assistant", @@ -3116,15 +3173,18 @@ export class ChatManager { }, }); } catch (persistErr) { + interruptionDurable = false; diagnostics.error(`Failed to persist interrupted response for session ${sessionId}:`, persistErr); } } - await this.flushInFlightGenerationPersist(sessionId, null, generationId); + if (interruptionDurable) { + await this.flushInFlightGenerationPersist(sessionId, null, generationId); + } const current = this.activeGenerations.get(sessionId); if (current?.generationId === generationId) { current.cancellationResult = { - success: true, + success: interruptionDurable, interrupted: Boolean(interruptedMessage), ...(interruptedMessage ? { message: interruptedMessage } : {}), };