FN-8791: prevent NUL-containing chat checkpoints from crashing
Prevent malformed tool output from breaking chat checkpoint persistence. - Sanitize NUL characters at chat JSONB persistence boundaries without mutating clean values. - Observe rejected best-effort checkpoint writes so streaming sessions do not emit unhandled rejections. - Add PostgreSQL, dashboard, and sanitizer regression coverage with operator-facing release notes. Files changed: .changeset/fn-8791-chat-nul-checkpoint.md | 7 ++ .../reliability/chat-jsonb-nul-sanitization.md | 44 +++++++++++ packages/core/src/__tests__/nul-sanitize.test.ts | 21 ++++- .../chat-store-content-search-edit.pg.test.ts | 91 +++++++++++++++++++++- packages/core/src/async-stores/async-chat-store.ts | 61 ++++++++++----- packages/core/src/postgres/nul-sanitize.ts | 17 ++-- .../dashboard/src/__tests__/chat-manager.test.ts | 90 +++++++++++++++++++++ packages/dashboard/src/chat.ts | 21 ++++- 8 files changed, 322 insertions(+), 30 deletions(-) Fusion-Task-Id: FN-8791 Fusion-Task-Lineage: 51e13634-04cc-4827-b72f-f33e40760940 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
7
.changeset/fn-8791-chat-nul-checkpoint.md
Normal file
7
.changeset/fn-8791-chat-nul-checkpoint.md
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
summary: Prevent chat checkpoints from failing on NUL-containing tool output.
|
||||||
|
category: fix
|
||||||
|
dev: Sanitizes chat JSONB persistence boundaries and observes best-effort checkpoint failures.
|
||||||
44
docs/solutions/reliability/chat-jsonb-nul-sanitization.md
Normal file
44
docs/solutions/reliability/chat-jsonb-nul-sanitization.md
Normal file
@@ -0,0 +1,44 @@
|
|||||||
|
---
|
||||||
|
title: "Chat JSONB persistence must strip U+0000 at every PostgreSQL boundary"
|
||||||
|
date: 2026-08-05
|
||||||
|
problem_type: reliability
|
||||||
|
module: "@fusion/core"
|
||||||
|
component: chat-persistence
|
||||||
|
tags:
|
||||||
|
- postgres
|
||||||
|
- jsonb
|
||||||
|
- chat
|
||||||
|
- nul
|
||||||
|
- checkpoints
|
||||||
|
symptoms:
|
||||||
|
- "unsupported Unicode escape sequence during chat session checkpointing"
|
||||||
|
- "unhandled rejection after a tool reads NUL-bearing output"
|
||||||
|
root_cause: "Raw model and tool text reached PostgreSQL jsonb bindings without U+0000 sanitization, while fire-and-forget checkpoint promises had no rejection observer."
|
||||||
|
resolution_type: code_fix
|
||||||
|
---
|
||||||
|
|
||||||
|
## Problem
|
||||||
|
|
||||||
|
PostgreSQL rejects U+0000 in `text` and `jsonb`. Chat tool results are arbitrary environment-controlled output, so a tool reading redirected Fusion logs can include the logger marker `\u0000fnlvl=info\u0000`. Persisting that result in an in-flight generation snapshot caused the checkpoint write to reject; because checkpoints are intentionally fire-and-forget, the dashboard then emitted an unhandled rejection.
|
||||||
|
|
||||||
|
## Protected surfaces
|
||||||
|
|
||||||
|
The invariant belongs at PostgreSQL persistence boundaries, not provider callbacks. `sanitizeJsonbValue` now recursively strips U+0000 from JSONB values and keys before chat-session snapshot creation/checkpoint writes, message and room-message inserts, attachment appends, and message metadata merges. This covers direct chat, QuickChat, all provider callbacks, and future tool providers without mutating the caller's value. Clean values retain identity; `null` and `undefined` retain their existing semantics.
|
||||||
|
|
||||||
|
## Degradation behavior
|
||||||
|
|
||||||
|
In-flight checkpoints are crash-recovery state, not turn control flow. `ChatManager` routes debounced timer writes and immediate flushes through one observing helper. A failed checkpoint emits a concise session-scoped warning without payload or database-error text, drops that attempt, and allows streaming, completion, cancellation, and later clears to continue. It does not retry, wait, or create another scheduler.
|
||||||
|
|
||||||
|
## Why strip U+0000
|
||||||
|
|
||||||
|
Stripping is deterministic, preserves the readable surrounding tool output, and is already the established PostgreSQL sanitizer behavior. Escaping into a literal sequence would change tool text semantics and still require every reader to understand a special encoding.
|
||||||
|
|
||||||
|
## Regression commands
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pnpm --filter @fusion/core exec vitest run src/__tests__/nul-sanitize.test.ts --silent=passed-only --reporter=default
|
||||||
|
pnpm --filter @fusion/core exec vitest run src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts --silent=passed-only --reporter=default
|
||||||
|
pnpm --filter @fusion/dashboard exec vitest run src/__tests__/chat-manager.test.ts --project dashboard-api-quality --reporter=default --silent=passed-only
|
||||||
|
```
|
||||||
|
|
||||||
|
The PostgreSQL test persists the reported full in-flight tool-result shape and reads it back; dashboard tests exercise both debounced and flush rejection paths with no unhandled rejection.
|
||||||
@@ -59,8 +59,25 @@ describe("sanitizeTextValue", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("sanitizeJsonbValue", () => {
|
describe("sanitizeJsonbValue", () => {
|
||||||
it("deep-strips NUL from an object destined for a jsonb column", () => {
|
it("deep-strips NUL from nested values, keys, and repeated logger markers", () => {
|
||||||
expect(sanitizeJsonbValue({ note: "tail\u0000end" })).toEqual({ note: "tailend" });
|
const snapshot = {
|
||||||
|
["tool\u0000result"]: {
|
||||||
|
content: ["\u0000fnlvl=info\u0000first", { text: "\u0000fnlvl=warn\u0000second" }],
|
||||||
|
},
|
||||||
|
};
|
||||||
|
expect(sanitizeJsonbValue(snapshot)).toEqual({
|
||||||
|
toolresult: {
|
||||||
|
content: ["fnlvl=infofirst", { text: "fnlvl=warnsecond" }],
|
||||||
|
},
|
||||||
|
});
|
||||||
|
expect(snapshot["tool\u0000result"].content[0]).toBe("\u0000fnlvl=info\u0000first");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("preserves clean JSON identity while leaving null and undefined intact", () => {
|
||||||
|
const clean = { nested: ["clean", { value: true }] };
|
||||||
|
expect(sanitizeJsonbValue(clean)).toBe(clean);
|
||||||
|
expect(sanitizeJsonbValue(null)).toBe(null);
|
||||||
|
expect(sanitizeJsonbValue(undefined)).toBe(undefined);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("passes through null/undefined unchanged", () => {
|
it("passes through null/undefined unchanged", () => {
|
||||||
|
|||||||
@@ -19,13 +19,16 @@ import { execSync } from "node:child_process";
|
|||||||
import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js";
|
import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js";
|
||||||
import {
|
import {
|
||||||
addChatMessage,
|
addChatMessage,
|
||||||
|
addChatMessageAttachment,
|
||||||
createChatSession,
|
createChatSession,
|
||||||
deleteChatMessagesFrom,
|
deleteChatMessagesFrom,
|
||||||
getChatMessages,
|
getChatMessages,
|
||||||
|
getChatSession,
|
||||||
|
setInFlightGeneration,
|
||||||
searchChatSessionsByMessageContent,
|
searchChatSessionsByMessageContent,
|
||||||
updateChatMessageMetadata,
|
updateChatMessageMetadata,
|
||||||
} from "../../async-stores/async-chat-store.js";
|
} from "../../async-stores/async-chat-store.js";
|
||||||
import type { ChatMessage, ChatSession } from "../../chat/chat-types.js";
|
import type { ChatInFlightGenerationState, ChatMessage, ChatSession } from "../../chat/chat-types.js";
|
||||||
|
|
||||||
const PG_TEST_URL_BASE =
|
const PG_TEST_URL_BASE =
|
||||||
process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432";
|
process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432";
|
||||||
@@ -216,6 +219,40 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", ()
|
|||||||
expect((await getChatMessages(ctx.layer.db, sessionA.id)).length).toBe(1);
|
expect((await getChatMessages(ctx.layer.db, sessionA.id)).length).toBe(1);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
/*
|
||||||
|
FNXC:ChatPersistence 2026-08-05-01:54:
|
||||||
|
Attachment appends and metadata merges are separate jsonb update paths from
|
||||||
|
initial inserts. They must independently enforce the arbitrary-text NUL
|
||||||
|
invariant so a later tool-derived mutation cannot poison a chat row.
|
||||||
|
*/
|
||||||
|
it("sanitizes attachment appends and metadata merge updates at their jsonb boundaries", async () => {
|
||||||
|
ctx = await setupCtx();
|
||||||
|
const session = await makeSession(ctx);
|
||||||
|
const message = await addMessage(ctx, session.id, "user", "hello", { clean: true });
|
||||||
|
|
||||||
|
const attached = await addChatMessageAttachment(ctx.layer.db, session.id, message.id, {
|
||||||
|
id: "att\u0000-1",
|
||||||
|
filename: "fu\u0000sion.log",
|
||||||
|
originalName: "fu\u0000sion.log",
|
||||||
|
mimeType: "text/plain\u0000",
|
||||||
|
size: 1,
|
||||||
|
createdAt: "2026-08-05T01:54:00.000Z",
|
||||||
|
});
|
||||||
|
expect(attached.attachments).toEqual([{
|
||||||
|
id: "att-1",
|
||||||
|
filename: "fusion.log",
|
||||||
|
originalName: "fusion.log",
|
||||||
|
mimeType: "text/plain",
|
||||||
|
size: 1,
|
||||||
|
createdAt: "2026-08-05T01:54:00.000Z",
|
||||||
|
}]);
|
||||||
|
|
||||||
|
const merged = await updateChatMessageMetadata(ctx.layer.db, message.id, {
|
||||||
|
["tool\u0000result"]: { text: "\u0000fnlvl=info\u0000" },
|
||||||
|
});
|
||||||
|
expect(merged.metadata).toEqual({ clean: true, toolresult: { text: "fnlvl=info" } });
|
||||||
|
});
|
||||||
|
|
||||||
it("updateChatMessageMetadata merges by default, replaces on merge:false, and throws for missing messages", async () => {
|
it("updateChatMessageMetadata merges by default, replaces on merge:false, and throws for missing messages", async () => {
|
||||||
ctx = await setupCtx();
|
ctx = await setupCtx();
|
||||||
|
|
||||||
@@ -235,6 +272,58 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", ()
|
|||||||
).rejects.toThrow(/not found/);
|
).rejects.toThrow(/not found/);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
/*
|
||||||
|
FNXC:ChatPersistence 2026-08-05-01:54:
|
||||||
|
A completed tool result can contain Fusion's raw logger framing
|
||||||
|
(U+0000fnlvl=infoU+0000). Checkpoint persistence must strip it recursively
|
||||||
|
before jsonb binding, preserve all replay fields, and leave the input owned
|
||||||
|
by the live chat callback untouched.
|
||||||
|
*/
|
||||||
|
it("persists the NUL-marked in-flight tool snapshot and reads back its sanitized shape", async () => {
|
||||||
|
ctx = await setupCtx();
|
||||||
|
const session = await makeSession(ctx);
|
||||||
|
const snapshot: ChatInFlightGenerationState = {
|
||||||
|
status: "generating",
|
||||||
|
streamingText: "prefix\u0000fnlvl=info\u0000suffix",
|
||||||
|
streamingThinking: "thought\u0000stream",
|
||||||
|
toolCalls: [{
|
||||||
|
toolName: "ba\u0000sh",
|
||||||
|
args: { command: "tail\u0000 /tmp/fusion.log" },
|
||||||
|
isError: false,
|
||||||
|
result: {
|
||||||
|
content: [{ type: "text", text: "log \u0000fnlvl=info\u0000 line" }],
|
||||||
|
["nested\u0000key"]: ["a\u0000b"],
|
||||||
|
},
|
||||||
|
status: "completed",
|
||||||
|
}],
|
||||||
|
replayFromEventId: 4,
|
||||||
|
updatedAt: "2026-08-05T01:54:00.000Z",
|
||||||
|
};
|
||||||
|
|
||||||
|
const updated = await setInFlightGeneration(ctx.layer.db, session.id, snapshot);
|
||||||
|
expect(updated?.inFlightGeneration).toMatchObject({
|
||||||
|
status: "generating",
|
||||||
|
streamingText: "prefixfnlvl=infosuffix",
|
||||||
|
streamingThinking: "thoughtstream",
|
||||||
|
replayFromEventId: 4,
|
||||||
|
toolCalls: [{
|
||||||
|
toolName: "bash",
|
||||||
|
args: { command: "tail /tmp/fusion.log" },
|
||||||
|
result: {
|
||||||
|
content: [{ type: "text", text: "log fnlvl=info line" }],
|
||||||
|
nestedkey: ["ab"],
|
||||||
|
},
|
||||||
|
}],
|
||||||
|
});
|
||||||
|
expect(snapshot.streamingText).toBe("prefix\u0000fnlvl=info\u0000suffix");
|
||||||
|
expect((snapshot.toolCalls[0].result as { content: Array<{ text: string }> }).content[0]?.text).toBe("log \u0000fnlvl=info\u0000 line");
|
||||||
|
|
||||||
|
const persisted = await getChatSession(ctx.layer.db, session.id);
|
||||||
|
expect(persisted?.inFlightGeneration).toEqual(updated?.inFlightGeneration);
|
||||||
|
await expect(setInFlightGeneration(ctx.layer.db, session.id, null)).resolves.toBeDefined();
|
||||||
|
expect((await getChatSession(ctx.layer.db, session.id))?.inFlightGeneration).toBeNull();
|
||||||
|
});
|
||||||
|
|
||||||
/*
|
/*
|
||||||
FNXC:PostgresMigrationNulSanitize 2026-07-20:
|
FNXC:PostgresMigrationNulSanitize 2026-07-20:
|
||||||
Production incident: the CEO agent's chat turn crashed mid-conversation
|
Production incident: the CEO agent's chat turn crashed mid-conversation
|
||||||
|
|||||||
@@ -128,24 +128,34 @@ function rowToRoomMessage(row: Record<string, unknown>): ChatRoomMessage {
|
|||||||
* Create a chat session.
|
* Create a chat session.
|
||||||
*/
|
*/
|
||||||
export async function createChatSession(handle: QueryHandle, session: ChatSession): Promise<ChatSession> {
|
export async function createChatSession(handle: QueryHandle, session: ChatSession): Promise<ChatSession> {
|
||||||
|
/*
|
||||||
|
FNXC:ChatPersistence 2026-08-05-01:54:
|
||||||
|
Chat snapshot fields may include arbitrary model and tool text. PostgreSQL
|
||||||
|
rejects U+0000 anywhere in jsonb, so strip it only at this persistence
|
||||||
|
boundary and return the same sanitized shape that is stored.
|
||||||
|
*/
|
||||||
|
const sanitized: ChatSession = {
|
||||||
|
...session,
|
||||||
|
inFlightGeneration: sanitizeJsonbValue(session.inFlightGeneration),
|
||||||
|
};
|
||||||
await handle.insert(schema.project.chatSessions).values({
|
await handle.insert(schema.project.chatSessions).values({
|
||||||
id: session.id,
|
id: sanitized.id,
|
||||||
agentId: session.agentId,
|
agentId: sanitized.agentId,
|
||||||
title: session.title,
|
title: sanitized.title,
|
||||||
status: session.status,
|
status: sanitized.status,
|
||||||
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition.
|
// FNXC:MultiProjectIsolation 2026-07-15-23:40: write the caller's domain project to owner_project_id and never project_id — the trigger/GUC owns the partition.
|
||||||
ownerProjectId: session.projectId,
|
ownerProjectId: sanitized.projectId,
|
||||||
modelProvider: session.modelProvider,
|
modelProvider: sanitized.modelProvider,
|
||||||
modelId: session.modelId,
|
modelId: sanitized.modelId,
|
||||||
thinkingLevel: session.thinkingLevel ?? null,
|
thinkingLevel: sanitized.thinkingLevel ?? null,
|
||||||
createdAt: session.createdAt,
|
createdAt: sanitized.createdAt,
|
||||||
updatedAt: session.updatedAt,
|
updatedAt: sanitized.updatedAt,
|
||||||
pinnedAt: session.pinnedAt,
|
pinnedAt: sanitized.pinnedAt,
|
||||||
cliSessionFile: session.cliSessionFile,
|
cliSessionFile: sanitized.cliSessionFile,
|
||||||
inFlightGeneration: session.inFlightGeneration,
|
inFlightGeneration: sanitized.inFlightGeneration,
|
||||||
cliExecutorAdapterId: session.cliExecutorAdapterId,
|
cliExecutorAdapterId: sanitized.cliExecutorAdapterId,
|
||||||
});
|
});
|
||||||
return session;
|
return sanitized;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -767,9 +777,16 @@ export async function setInFlightGeneration(
|
|||||||
): Promise<ChatSession | undefined> {
|
): Promise<ChatSession | undefined> {
|
||||||
const existing = await getChatSession(handle, id);
|
const existing = await getChatSession(handle, id);
|
||||||
if (!existing) return undefined;
|
if (!existing) return undefined;
|
||||||
|
/*
|
||||||
|
FNXC:ChatPersistence 2026-08-05-01:54:
|
||||||
|
Every checkpoint reaches this JSONB boundary from live streaming callbacks,
|
||||||
|
including tool results. Sanitizing here protects regular chat, QuickChat,
|
||||||
|
and future providers without mutating the caller's snapshot.
|
||||||
|
*/
|
||||||
|
const sanitizedInFlightGeneration = sanitizeJsonbValue(inFlightGeneration);
|
||||||
await handle
|
await handle
|
||||||
.update(schema.project.chatSessions)
|
.update(schema.project.chatSessions)
|
||||||
.set({ inFlightGeneration })
|
.set({ inFlightGeneration: sanitizedInFlightGeneration })
|
||||||
.where(eq(schema.project.chatSessions.id, id));
|
.where(eq(schema.project.chatSessions.id, id));
|
||||||
return getChatSession(handle, id);
|
return getChatSession(handle, id);
|
||||||
}
|
}
|
||||||
@@ -790,7 +807,10 @@ export async function addChatMessageAttachment(
|
|||||||
if (!message || message.sessionId !== sessionId) {
|
if (!message || message.sessionId !== sessionId) {
|
||||||
throw new Error(`Message ${messageId} not found in session ${sessionId}`);
|
throw new Error(`Message ${messageId} not found in session ${sessionId}`);
|
||||||
}
|
}
|
||||||
const updatedAttachments = [...(message.attachments ?? []), attachment];
|
const updatedAttachments = sanitizeJsonbValue([
|
||||||
|
...(message.attachments ?? []),
|
||||||
|
attachment,
|
||||||
|
]);
|
||||||
await handle
|
await handle
|
||||||
.update(schema.project.chatMessages)
|
.update(schema.project.chatMessages)
|
||||||
.set({ attachments: updatedAttachments })
|
.set({ attachments: updatedAttachments })
|
||||||
@@ -930,7 +950,7 @@ export async function updateChatMessageMetadata(
|
|||||||
|
|
||||||
await handle
|
await handle
|
||||||
.update(schema.project.chatMessages)
|
.update(schema.project.chatMessages)
|
||||||
.set({ metadata: nextMetadata ?? null })
|
.set({ metadata: sanitizeJsonbValue(nextMetadata) ?? null })
|
||||||
.where(eq(schema.project.chatMessages.id, messageId));
|
.where(eq(schema.project.chatMessages.id, messageId));
|
||||||
|
|
||||||
const updated = await getChatMessage(handle, messageId);
|
const updated = await getChatMessage(handle, messageId);
|
||||||
@@ -1109,7 +1129,10 @@ export async function addChatRoomMessageAttachment(
|
|||||||
if (!message || message.roomId !== roomId) {
|
if (!message || message.roomId !== roomId) {
|
||||||
throw new Error(`Message ${messageId} not found in room ${roomId}`);
|
throw new Error(`Message ${messageId} not found in room ${roomId}`);
|
||||||
}
|
}
|
||||||
const updatedAttachments = [...(message.attachments ?? []), attachment];
|
const updatedAttachments = sanitizeJsonbValue([
|
||||||
|
...(message.attachments ?? []),
|
||||||
|
attachment,
|
||||||
|
]);
|
||||||
await handle
|
await handle
|
||||||
.update(schema.project.chatRoomMessages)
|
.update(schema.project.chatRoomMessages)
|
||||||
.set({ attachments: updatedAttachments })
|
.set({ attachments: updatedAttachments })
|
||||||
|
|||||||
@@ -32,14 +32,19 @@ export function deepStripNulChars(value: unknown): unknown {
|
|||||||
return stripNulChars(value);
|
return stripNulChars(value);
|
||||||
}
|
}
|
||||||
if (Array.isArray(value)) {
|
if (Array.isArray(value)) {
|
||||||
return value.map(deepStripNulChars);
|
const sanitized = value.map(deepStripNulChars);
|
||||||
|
return sanitized.some((entry, index) => entry !== value[index]) ? sanitized : value;
|
||||||
}
|
}
|
||||||
if (value !== null && typeof value === "object") {
|
if (value !== null && typeof value === "object") {
|
||||||
return Object.fromEntries(
|
const entries = Object.entries(value as Record<string, unknown>);
|
||||||
Object.entries(value as Record<string, unknown>).map(
|
const sanitizedEntries = entries.map(([key, entry]) => [
|
||||||
([key, entry]) => [stripNulChars(key), deepStripNulChars(entry)],
|
stripNulChars(key),
|
||||||
),
|
deepStripNulChars(entry),
|
||||||
);
|
] as const);
|
||||||
|
const changed = sanitizedEntries.some(([key, entry], index) => (
|
||||||
|
key !== entries[index]?.[0] || entry !== entries[index]?.[1]
|
||||||
|
));
|
||||||
|
return changed ? Object.fromEntries(sanitizedEntries) : value;
|
||||||
}
|
}
|
||||||
return value;
|
return value;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -169,6 +169,7 @@ describe("ChatManager.sendMessage", () => {
|
|||||||
});
|
});
|
||||||
mockChatStore.getMessages.mockReturnValue([]);
|
mockChatStore.getMessages.mockReturnValue([]);
|
||||||
mockChatStore.getRoomMessages.mockReturnValue([]);
|
mockChatStore.getRoomMessages.mockReturnValue([]);
|
||||||
|
mockChatStore.setInFlightGeneration.mockResolvedValue(undefined);
|
||||||
|
|
||||||
mockAgentStore.init.mockResolvedValue(undefined);
|
mockAgentStore.init.mockResolvedValue(undefined);
|
||||||
mockAgentStore.getAgent.mockResolvedValue({
|
mockAgentStore.getAgent.mockResolvedValue({
|
||||||
@@ -203,6 +204,8 @@ describe("ChatManager.sendMessage", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
afterEach(() => {
|
afterEach(() => {
|
||||||
|
__setChatDiagnostics(null);
|
||||||
|
vi.useRealTimers();
|
||||||
vi.restoreAllMocks();
|
vi.restoreAllMocks();
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -1019,6 +1022,93 @@ describe("ChatManager.sendMessage", () => {
|
|||||||
expect(mockChatStore.setInFlightGeneration).toHaveBeenLastCalledWith("chat-001", null);
|
expect(mockChatStore.setInFlightGeneration).toHaveBeenLastCalledWith("chat-001", null);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("observes a debounced checkpoint rejection without an unhandled rejection", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
const warn = vi.fn();
|
||||||
|
const unhandled = vi.fn();
|
||||||
|
process.on("unhandledRejection", unhandled);
|
||||||
|
__setChatDiagnostics({ log: vi.fn(), warn, error: vi.fn() });
|
||||||
|
|
||||||
|
let resolvePrompt: (() => void) | undefined;
|
||||||
|
__setCreateFnAgent(async (options: any) => ({
|
||||||
|
session: {
|
||||||
|
prompt: vi.fn().mockImplementation(() => new Promise<void>((resolve) => {
|
||||||
|
options.onToolEnd("bash", false, {
|
||||||
|
content: [{ type: "text", text: "log \u0000fnlvl=info\u0000 line" }],
|
||||||
|
});
|
||||||
|
resolvePrompt = resolve;
|
||||||
|
})),
|
||||||
|
dispose: vi.fn(),
|
||||||
|
state: { messages: [{ role: "assistant", content: "done" }] },
|
||||||
|
},
|
||||||
|
}));
|
||||||
|
mockChatStore.setInFlightGeneration
|
||||||
|
.mockResolvedValueOnce(undefined)
|
||||||
|
.mockRejectedValueOnce(new Error("jsonb rejected"))
|
||||||
|
.mockResolvedValue(undefined);
|
||||||
|
|
||||||
|
const sending = createChatManager().sendMessage("chat-001", "Read the log");
|
||||||
|
await vi.advanceTimersByTimeAsync(200);
|
||||||
|
await Promise.resolve();
|
||||||
|
expect(warn.mock.calls.filter(([message]) => message === "Failed to persist in-flight chat checkpoint for session chat-001")).toEqual([
|
||||||
|
["Failed to persist in-flight chat checkpoint for session chat-001"],
|
||||||
|
]);
|
||||||
|
expect(unhandled).not.toHaveBeenCalled();
|
||||||
|
expect(mockChatStore.setInFlightGeneration).toHaveBeenCalledWith("chat-001", expect.objectContaining({
|
||||||
|
toolCalls: [expect.objectContaining({
|
||||||
|
result: { content: [{ type: "text", text: "log \u0000fnlvl=info\u0000 line" }] },
|
||||||
|
})],
|
||||||
|
}));
|
||||||
|
|
||||||
|
resolvePrompt?.();
|
||||||
|
await sending;
|
||||||
|
process.off("unhandledRejection", unhandled);
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("observes an immediate flush rejection and preserves the final clear", async () => {
|
||||||
|
const warn = vi.fn();
|
||||||
|
__setChatDiagnostics({ log: vi.fn(), warn, error: vi.fn() });
|
||||||
|
mockChatStore.setInFlightGeneration
|
||||||
|
.mockRejectedValueOnce(new Error("checkpoint unavailable"))
|
||||||
|
.mockResolvedValue(undefined);
|
||||||
|
__setCreateFnAgent(async () => ({
|
||||||
|
session: {
|
||||||
|
prompt: vi.fn().mockResolvedValue(undefined),
|
||||||
|
dispose: vi.fn(),
|
||||||
|
state: { messages: [{ role: "assistant", content: "done" }] },
|
||||||
|
},
|
||||||
|
}));
|
||||||
|
|
||||||
|
await expect(createChatManager().sendMessage("chat-001", "Hello")).resolves.toBeUndefined();
|
||||||
|
await Promise.resolve();
|
||||||
|
expect(warn.mock.calls.filter(([message]) => message === "Failed to persist in-flight chat checkpoint for session chat-001")).toEqual([
|
||||||
|
["Failed to persist in-flight chat checkpoint for session chat-001"],
|
||||||
|
]);
|
||||||
|
expect(mockChatStore.setInFlightGeneration).toHaveBeenLastCalledWith("chat-001", null);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("cancels a stale debounced snapshot before flushing the latest clear", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
let onText: ((delta: string) => void) | undefined;
|
||||||
|
__setCreateFnAgent(async (options: any) => {
|
||||||
|
onText = options.onText;
|
||||||
|
return {
|
||||||
|
session: {
|
||||||
|
prompt: vi.fn().mockImplementation(async () => onText?.("queued")),
|
||||||
|
dispose: vi.fn(),
|
||||||
|
state: { messages: [{ role: "assistant", content: "done" }] },
|
||||||
|
},
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
|
await createChatManager().sendMessage("chat-001", "Hello");
|
||||||
|
await vi.advanceTimersByTimeAsync(200);
|
||||||
|
expect(mockChatStore.setInFlightGeneration).toHaveBeenCalledTimes(2);
|
||||||
|
expect(mockChatStore.setInFlightGeneration).toHaveBeenLastCalledWith("chat-001", null);
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
it("broadcasts done with persisted assistant message snapshot", async () => {
|
it("broadcasts done with persisted assistant message snapshot", async () => {
|
||||||
const events: Array<{ type: string; data: unknown }> = [];
|
const events: Array<{ type: string; data: unknown }> = [];
|
||||||
const unsubscribe = chatStreamManager.subscribe("chat-001", (event) => {
|
const unsubscribe = chatStreamManager.subscribe("chat-001", (event) => {
|
||||||
|
|||||||
@@ -1518,6 +1518,23 @@ export class ChatManager {
|
|||||||
this.cliChatProjectId = projectId;
|
this.cliChatProjectId = projectId;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
FNXC:ChatPersistence 2026-08-05-01:54:
|
||||||
|
Checkpoint snapshots contain environment-controlled tool output. Persistence
|
||||||
|
is best-effort for crash recovery, but every fire-and-forget write must have
|
||||||
|
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(() => {
|
||||||
|
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}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
|
private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
|
||||||
const existingTimer = this.inFlightPersistTimers.get(sessionId);
|
const existingTimer = this.inFlightPersistTimers.get(sessionId);
|
||||||
if (existingTimer) {
|
if (existingTimer) {
|
||||||
@@ -1526,7 +1543,7 @@ export class ChatManager {
|
|||||||
|
|
||||||
const timer = setTimeout(() => {
|
const timer = setTimeout(() => {
|
||||||
this.inFlightPersistTimers.delete(sessionId);
|
this.inFlightPersistTimers.delete(sessionId);
|
||||||
this.chatStore.setInFlightGeneration(sessionId, snapshot);
|
this.persistInFlightGeneration(sessionId, snapshot);
|
||||||
}, IN_FLIGHT_PERSIST_DEBOUNCE_MS);
|
}, IN_FLIGHT_PERSIST_DEBOUNCE_MS);
|
||||||
this.inFlightPersistTimers.set(sessionId, timer);
|
this.inFlightPersistTimers.set(sessionId, timer);
|
||||||
}
|
}
|
||||||
@@ -1537,7 +1554,7 @@ export class ChatManager {
|
|||||||
clearTimeout(existingTimer);
|
clearTimeout(existingTimer);
|
||||||
this.inFlightPersistTimers.delete(sessionId);
|
this.inFlightPersistTimers.delete(sessionId);
|
||||||
}
|
}
|
||||||
this.chatStore.setInFlightGeneration(sessionId, snapshot);
|
this.persistInFlightGeneration(sessionId, snapshot);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async getChatModelSettings(): Promise<{
|
private async getChatModelSettings(): Promise<{
|
||||||
|
|||||||
Reference in New Issue
Block a user