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

FN-028: preserve interrupted chat context

Browse Source 
Preserve streamed chat prefixes and durable context when users stop generation.

- Persist interrupted assistant text in the reopened model session and chat history.
- Keep optimistic prefixes visible when cancellation persistence is incomplete.
- Add cancellation coverage, documentation, and a patch changeset.

Files changed:
 .changeset/fn-028-chat-interrupted-context.md      |   7 ++
 docs/dashboard-guide.md                            |   2 +
 .../app/components/TaskPlannerChatTab.tsx          |  11 +-
 .../dashboard/app/hooks/__tests__/useChat.test.ts  |  28 +++++
 packages/dashboard/app/hooks/chatTypes.ts          |   5 +
 packages/dashboard/app/hooks/useChat.ts            |  14 ++-
 .../__tests__/chat-manager-rewind-session.test.ts  |  65 +++++++++++-
 .../dashboard/src/__tests__/chat-manager.test.ts   |   1 +
 .../src/__tests__/routes-chat-cancellation.test.ts | 113 +++++++++++++++++++++
 packages/dashboard/src/chat.ts                     |  72 +++++++++++--
 10 files changed, 305 insertions(+), 13 deletions(-)

Fusion-Task-Id: FN-028

Fusion-Task-Lineage: d26c55a8-1b1f-42cc-89f8-cd39f3ff8ea5

Co-authored-by: Fusion <noreply@runfusion.ai>
This commit is contained in:
Fusion Agent
2026-08-19 05:45:27 +00:00
parent 519180b9e2
commit 9cff3d26ee
10 changed files with 305 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-028-chat-interrupted-context.md Normal file
View File

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

2
docs/dashboard-guide.md
View File

@@ -740,6 +740,8 @@ Mailbox Inbox, Outbox, and agent lists exclude archived correspondence and unrea
<!-- FNXC:Chat-ModelSwitch 2026-07-13-00:00: FN-7934 applies that fitted Brain-popup layout to narrow chat surfaces, including floating Chat windows and compact docks on wide desktop viewports, because the browser viewport alone does not describe the popover's clipping container. -->
- 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)
<!-- FNXC:ChatCancellation 2026-08-19-05:20: Direct/Quick Chat and task Planner Chat must make an explicit Stop durable for reload and the next model turn, while Chat Rooms and CLI-agent-backed sessions retain their separate cancellation semantics. -->
- 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.
<!-- FNXC:ChatEmptyMessage 2026-07-10-00:00: Empty final assistant responses can be legitimate provider output (for example a Grok CLI run ending without text). Document the shared Chat/Planner Chat behavior so operators see "No message" instead of interpreting a blank bubble as a rendering failure. -->
- 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.

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

@@ -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,
));

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

@@ -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] });

5
packages/dashboard/app/hooks/chatTypes.ts
View File

@@ -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<string, unknown> | null;
attachments?: Array<{
id: string;
filename: string;

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

@@ -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);
}

65
packages/dashboard/src/__tests__/chat-manager-rewind-session.test.ts
View File

@@ -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<ChatMessage> {
async addMessage(
sessionId: string,
input: {
role: "user" | "assistant";
content: string;
thinkingOutput?: string;
metadata?: Record<string, unknown>;
},
): Promise<ChatMessage> {
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<void>((_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);

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

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

113
packages/dashboard/src/__tests__/routes-chat-cancellation.test.ts Normal file
View File

@@ -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<typeof vi.fn> }) {
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<typeof vi.fn> };
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();
});
});

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

@@ -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<string, Array<{ toolName: string; args?: Record<string, unknown> }>>();
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 } : {}),
};