fix(ai-chat): hydrate crashed mid-run steps for model replay + cover write/read seams (#492)
F1: the server model-replay loaded history via findAllByChat().map(rowToUiMessage) WITHOUT hydrating parts from ai_chat_run_steps. A HARD crash mid-run (SIGKILL/OOM) fires no terminal callback, so the assistant row stays parts:[] and its partial tool-calls/results/text (durable in the steps table) dropped out of the model's next-turn context. Hydrate needy assistant rows (role==='assistant' && !rowHasInlineParts) via findByMessageIds + hydrateAssistantParts before the replay map — mirroring the controller's withReconstructedParts exactly — guarded on the optional repo. Fix the now-false interrupt-resume comment. F2: add a service int-spec that drives the REAL onStep append-persist WRITE branch through AiChatService.stream with a real AiChatRunStepRepo injected, asserting the per-step rows' stepIndex + parts slice and the step-marker metadata match a single-row flush (catches an stepsPersisted-1 off-by-one). F3: add a controller int-spec that drives withReconstructedParts through getMessages WITH the repo present (a mid-run marker-only row + its step rows), asserting the reconstructed metadata.parts and workspace-scoping. F4: remove the dead countByMessage (zero prod callers; reconstructRunParts derives stepsPersisted inline) + its now-unused sql import and the redundant test assertion. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1121,8 +1121,34 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
|
||||
chatId,
|
||||
workspace.id,
|
||||
);
|
||||
// #492: HYDRATE needy assistant rows from the steps table BEFORE the replay
|
||||
// map. A #492 mid-run assistant row carries only a step marker
|
||||
// (metadata.parts:[]); its real per-step parts live in `ai_chat_run_steps`.
|
||||
// The graceful terminal callbacks (onFinish/onError/onAbort -> flushAssistant)
|
||||
// assemble the full inline parts, so a normally-ended turn already has them.
|
||||
// But a HARD crash mid-run (SIGKILL/OOM) fires NO terminal callback, so the
|
||||
// row stays parts:[]; without this, rowToUiMessage falls back to an empty
|
||||
// text part and the partial tool-calls/results/text — durable in the steps
|
||||
// table — would DROP OUT of the model's replay context (regressing #183
|
||||
// step-granular durability for the model consumer). Mirrors the controller's
|
||||
// withReconstructedParts EXACTLY (same needy predicate + hydration helper).
|
||||
// Guarded on the optional repo: absent (positional test builds) degrades to
|
||||
// the current behavior rather than crashing.
|
||||
let replayHistory = oldHistory;
|
||||
if (this.aiChatRunStepRepo) {
|
||||
const needy = oldHistory.filter(
|
||||
(r) => r.role === 'assistant' && !rowHasInlineParts(r),
|
||||
);
|
||||
if (needy.length > 0) {
|
||||
const stepsByMessage = await this.aiChatRunStepRepo.findByMessageIds(
|
||||
needy.map((r) => r.id),
|
||||
workspace.id,
|
||||
);
|
||||
replayHistory = hydrateAssistantParts(oldHistory, stepsByMessage);
|
||||
}
|
||||
}
|
||||
const uiMessages: Array<Omit<UIMessage, 'id'> & { id: string }> = [
|
||||
...oldHistory.map(rowToUiMessage),
|
||||
...replayHistory.map(rowToUiMessage),
|
||||
{
|
||||
id: 'pending-user',
|
||||
role: 'user',
|
||||
@@ -1161,7 +1187,9 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
|
||||
// hint — confirm it against the persisted history (the preceding assistant
|
||||
// turn must really be aborted/streaming) so a spoofed flag cannot inject the
|
||||
// interrupt note onto an ordinary turn. The partial output the model needs is
|
||||
// already in `messages` (the aborted assistant row replays via findRecent).
|
||||
// already in `messages`: a #492 mid-run row's per-step parts live only in the
|
||||
// `ai_chat_run_steps` table and were hydrated into the replay history above,
|
||||
// so the aborted assistant turn replays WITH its partial parts intact.
|
||||
// Append the new user turn (shape-only) so index -2 is the prior assistant.
|
||||
const interrupted = isInterruptResume(
|
||||
[...oldHistory, { role: 'user', status: null, metadata: null }],
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectKysely } from 'nestjs-kysely';
|
||||
import { sql } from 'kysely';
|
||||
import { KyselyDB, KyselyTransaction } from '../../types/kysely.types';
|
||||
import { dbOrTx } from '../../utils';
|
||||
import { AiChatRunStep } from '@docmost/db/types/entity.types';
|
||||
@@ -93,19 +92,4 @@ export class AiChatRunStepRepo {
|
||||
}
|
||||
return byMessage;
|
||||
}
|
||||
|
||||
/**
|
||||
* How many steps are persisted for a message (its step-marker floor). Exposed
|
||||
* for the reconstruct contract (`reconstructRunParts → { parts, stepsPersisted }`)
|
||||
* so a caller can align a resume attach without materializing every step's parts.
|
||||
*/
|
||||
async countByMessage(messageId: string, workspaceId: string): Promise<number> {
|
||||
const row = await this.db
|
||||
.selectFrom('aiChatRunSteps')
|
||||
.select(sql<number>`count(*)::int`.as('n'))
|
||||
.where('messageId', '=', messageId)
|
||||
.where('workspaceId', '=', workspaceId)
|
||||
.executeTakeFirst();
|
||||
return row?.n ?? 0;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,412 @@
|
||||
import * as http from 'node:http';
|
||||
import { Kysely } from 'kysely';
|
||||
import { tool } from 'ai';
|
||||
import { z } from 'zod';
|
||||
import { MockLanguageModelV3, convertArrayToReadableStream } from 'ai/test';
|
||||
import { AiChatRepo } from '@docmost/db/repos/ai-chat/ai-chat.repo';
|
||||
import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
|
||||
import { AiChatRunStepRepo } from '@docmost/db/repos/ai-chat/ai-chat-run-step.repo';
|
||||
import {
|
||||
AiChatService,
|
||||
assembleStepParts,
|
||||
assistantParts,
|
||||
rowHasInlineParts,
|
||||
stepMarkerMetadata,
|
||||
} from 'src/core/ai-chat/ai-chat.service';
|
||||
import {
|
||||
getTestDb,
|
||||
destroyTestDb,
|
||||
createWorkspace,
|
||||
createUser,
|
||||
createChat,
|
||||
createMessage,
|
||||
} from './db';
|
||||
|
||||
/**
|
||||
* #492 append-persist — the REAL onStep WRITE path (F2) and the model-REPLAY
|
||||
* hydration path (F1), driven through `AiChatService.stream` against a LIVE
|
||||
* Postgres with a REAL `AiChatRunStepRepo` INJECTED. The existing append-persist
|
||||
* int-specs hand-roll the insert+marker cycle via the repos directly and build
|
||||
* the service with `aiChatRunStepRepo: undefined` (only the legacy-fallback branch
|
||||
* is covered), so an off-by-one on `stepsPersisted-1`, a wrong `capturedSteps`
|
||||
* slice, or a broken marker payload would pass all of them. These tests exercise
|
||||
* the actual `updateStreaming` append-persist branch end to end.
|
||||
*
|
||||
* The seam is the injected `model` (a seeded `MockLanguageModelV3` from `ai/test`)
|
||||
* plus a REAL Node `ServerResponse` as the hijacked socket — mirrors
|
||||
* ai-chat-stream.int-spec.ts.
|
||||
*/
|
||||
|
||||
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
async function waitFor(
|
||||
cond: () => Promise<boolean> | boolean,
|
||||
{ timeoutMs = 15_000, stepMs = 25 } = {},
|
||||
): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (Date.now() - start < timeoutMs) {
|
||||
if (await cond()) return;
|
||||
await sleep(stepMs);
|
||||
}
|
||||
throw new Error('waitFor: condition not met within timeout');
|
||||
}
|
||||
|
||||
// A real Node ServerResponse wired to a live socket (identical helper to the
|
||||
// stream int-spec) so the SDK's pipe/heartbeat writes behave as in prod.
|
||||
function makeRealResponse(): Promise<{
|
||||
res: http.ServerResponse;
|
||||
cleanup: () => Promise<void>;
|
||||
}> {
|
||||
return new Promise((resolve) => {
|
||||
const server = http.createServer((_req, res) => {
|
||||
resolve({
|
||||
res,
|
||||
cleanup: () =>
|
||||
new Promise<void>((done) => {
|
||||
try {
|
||||
if (!res.writableEnded) res.end();
|
||||
} catch {
|
||||
/* socket already gone */
|
||||
}
|
||||
server.close(() => done());
|
||||
}),
|
||||
});
|
||||
});
|
||||
server.listen(0, () => {
|
||||
const port = (server.address() as any).port;
|
||||
const creq = http.request({ port, method: 'GET' }, (cres) => {
|
||||
cres.resume();
|
||||
});
|
||||
creq.on('error', () => undefined);
|
||||
creq.end();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Stream parts for a normal, successful single-step turn.
|
||||
function successStream() {
|
||||
return convertArrayToReadableStream([
|
||||
{ type: 'stream-start', warnings: [] },
|
||||
{ type: 'text-start', id: 't1' },
|
||||
{ type: 'text-delta', id: 't1', delta: 'Hello' },
|
||||
{ type: 'text-delta', id: 't1', delta: ' there' },
|
||||
{ type: 'text-end', id: 't1' },
|
||||
{
|
||||
type: 'finish',
|
||||
finishReason: 'stop',
|
||||
usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 },
|
||||
},
|
||||
] as any);
|
||||
}
|
||||
|
||||
// A THREE-step turn: steps 0 and 1 each emit text + an `echo` tool call (the SDK
|
||||
// runs the tool and continues); step 2 answers and stops. Three steps is
|
||||
// deliberate: the LAST finished step's append-persist write races the terminal
|
||||
// finalize (which writes the full inline parts anyway, so a lost last-step row is
|
||||
// by design), but the NON-final steps 0 and 1 always drain to the steps table
|
||||
// before finalize — so those are what the test asserts on deterministically.
|
||||
function threeStepModel(): MockLanguageModelV3 {
|
||||
let step = 0;
|
||||
const toolStep = (i: number) => ({
|
||||
stream: convertArrayToReadableStream([
|
||||
{ type: 'stream-start', warnings: [] },
|
||||
{ type: 'text-start', id: `s${i}` },
|
||||
{ type: 'text-delta', id: `s${i}`, delta: `step ${i} ` },
|
||||
{ type: 'text-end', id: `s${i}` },
|
||||
{
|
||||
type: 'tool-call',
|
||||
toolCallId: `c${i}`,
|
||||
toolName: 'echo',
|
||||
input: JSON.stringify({ msg: `m${i}` }),
|
||||
},
|
||||
{
|
||||
type: 'finish',
|
||||
finishReason: 'tool-calls',
|
||||
usage: { inputTokens: 5, outputTokens: 3, totalTokens: 8 },
|
||||
},
|
||||
] as any),
|
||||
});
|
||||
return new MockLanguageModelV3({
|
||||
doStream: async () => {
|
||||
const n = step++;
|
||||
// Realistic inter-step latency. A real model spends seconds per step, so the
|
||||
// fire-and-forget per-step write chain drains to the steps table BETWEEN
|
||||
// steps; the mock otherwise collapses all steps into microseconds and the
|
||||
// terminal finalize wins the race before any but the first step persists.
|
||||
if (n > 0) await sleep(200);
|
||||
if (n < 2) return toolStep(n);
|
||||
return {
|
||||
stream: convertArrayToReadableStream([
|
||||
{ type: 'stream-start', warnings: [] },
|
||||
{ type: 'text-start', id: 's2' },
|
||||
{ type: 'text-delta', id: 's2', delta: 'final answer' },
|
||||
{ type: 'text-end', id: 's2' },
|
||||
{
|
||||
type: 'finish',
|
||||
finishReason: 'stop',
|
||||
usage: { inputTokens: 6, outputTokens: 4, totalTokens: 10 },
|
||||
},
|
||||
] as any),
|
||||
};
|
||||
},
|
||||
} as any);
|
||||
}
|
||||
|
||||
describe('#492 append-persist service paths [integration]', () => {
|
||||
let db: Kysely<any>;
|
||||
let aiChatRepo: AiChatRepo;
|
||||
let msgRepo: AiChatMessageRepo;
|
||||
let stepRepo: AiChatRunStepRepo;
|
||||
let workspaceId: string;
|
||||
let userId: string;
|
||||
|
||||
let closeCalls: number;
|
||||
const mcpClients = {
|
||||
toolsFor: async () => ({
|
||||
tools: {},
|
||||
clients: [
|
||||
{
|
||||
close: async () => {
|
||||
closeCalls += 1;
|
||||
},
|
||||
},
|
||||
],
|
||||
outcomes: [],
|
||||
instructions: [],
|
||||
}),
|
||||
};
|
||||
|
||||
// Build the service WITH a REAL AiChatRunStepRepo injected (the property under
|
||||
// test) — unlike the legacy-fallback harness that passes it as undefined.
|
||||
const echoTool = tool({
|
||||
description: 'echo the message back',
|
||||
inputSchema: z.object({ msg: z.string() }),
|
||||
execute: async ({ msg }) => ({ echoed: msg }),
|
||||
});
|
||||
|
||||
function buildService(): AiChatService {
|
||||
return new AiChatService(
|
||||
{ getChatModel: async () => null } as any,
|
||||
aiChatRepo,
|
||||
msgRepo,
|
||||
{} as any, // aiChatPageSnapshotRepo
|
||||
{ resolve: async () => null } as any, // aiSettings
|
||||
{ forUser: async () => ({ echo: echoTool }) } as any, // tools
|
||||
mcpClients as any,
|
||||
{} as any, // aiAgentRoleRepo
|
||||
{} as any, // pageRepo
|
||||
{} as any, // pageAccess
|
||||
{
|
||||
isAiChatDeferredToolsEnabled: () => false,
|
||||
isAiChatFinalStepLockdownEnabled: () => false,
|
||||
} as any, // environment (deferred OFF -> all tools active every step)
|
||||
undefined, // streamRegistry
|
||||
undefined, // aiChatRunService
|
||||
stepRepo, // #492 aiChatRunStepRepo — the append-persist backend
|
||||
);
|
||||
}
|
||||
|
||||
function userUiMessage(text: string) {
|
||||
return {
|
||||
id: `u-${Math.random()}`,
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text }],
|
||||
};
|
||||
}
|
||||
|
||||
async function runStream(opts: {
|
||||
model: MockLanguageModelV3;
|
||||
chatId: string;
|
||||
body: any;
|
||||
}): Promise<void> {
|
||||
closeCalls = 0;
|
||||
const service = buildService();
|
||||
const { res, cleanup } = await makeRealResponse();
|
||||
try {
|
||||
await service.stream({
|
||||
user: { id: userId, workspaceId } as any,
|
||||
workspace: { id: workspaceId, name: 'WS' } as any,
|
||||
sessionId: 'sess-1',
|
||||
body: opts.body,
|
||||
res: { raw: res } as any,
|
||||
signal: new AbortController().signal,
|
||||
model: opts.model as any,
|
||||
role: null,
|
||||
} as any);
|
||||
await waitFor(async () => {
|
||||
const rows = await msgRepo.findAllByChat(opts.chatId, workspaceId);
|
||||
return rows.some(
|
||||
(r) =>
|
||||
r.role === 'assistant' &&
|
||||
['completed', 'error', 'aborted'].includes(r.status as string),
|
||||
);
|
||||
});
|
||||
await waitFor(() => closeCalls > 0, { timeoutMs: 5_000 });
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
db = getTestDb();
|
||||
aiChatRepo = new AiChatRepo(db as any);
|
||||
msgRepo = new AiChatMessageRepo(db as any);
|
||||
stepRepo = new AiChatRunStepRepo(db as any);
|
||||
workspaceId = (await createWorkspace(db)).id;
|
||||
userId = (await createUser(db, workspaceId)).id;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await destroyTestDb();
|
||||
});
|
||||
|
||||
// --- F2: the real onStep append-persist WRITE branch -----------------------
|
||||
it('drives steps through the real onStep path: per-step rows + marker match a single-row flush', async () => {
|
||||
const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
|
||||
const model = threeStepModel();
|
||||
|
||||
// Capture the mid-run step-marker UPDATEs the append-persist branch writes on
|
||||
// the assistant row (a { parts: [], toolTraceVersion, stepsPersisted } patch).
|
||||
const updateSpy = jest.spyOn(msgRepo, 'update');
|
||||
try {
|
||||
await runStream({
|
||||
model,
|
||||
chatId,
|
||||
body: { chatId, messages: [userUiMessage('call the tool then answer')] },
|
||||
});
|
||||
|
||||
const rows = await msgRepo.findAllByChat(chatId, workspaceId);
|
||||
const assistant = rows.find((r) => r.role === 'assistant')!;
|
||||
expect(assistant).toBeDefined();
|
||||
expect(assistant.status).toBe('completed');
|
||||
// The turn finalizes with the FULL inline parts assembled by a single-row
|
||||
// flush (assistantParts over every step) — the baseline the per-step slices
|
||||
// must reproduce.
|
||||
expect(rowHasInlineParts(assistant)).toBe(true);
|
||||
const finalParts = (assistant.metadata as { parts: any[] }).parts;
|
||||
|
||||
// The two NON-final finished steps each landed their own row, in stepIndex
|
||||
// order. (The fire-and-forget write chain drains before the next step, so
|
||||
// poll until both are on disk; the LAST step's write may lose the finalize
|
||||
// race, which is by design — its parts are already in `finalParts`.)
|
||||
await waitFor(async () => {
|
||||
const s = await stepRepo.findByMessage(assistant.id, workspaceId);
|
||||
return s.length >= 2;
|
||||
});
|
||||
const steps = await stepRepo.findByMessage(assistant.id, workspaceId);
|
||||
expect(steps[0].stepIndex).toBe(0);
|
||||
expect(steps[1].stepIndex).toBe(1);
|
||||
|
||||
// Each per-step row carries a NON-trivial slice: this step's text part + its
|
||||
// paired tool part (guards a mutation that persists empty/whole-turn parts).
|
||||
const s0 = steps[0].parts as any[];
|
||||
expect(s0).toContainEqual({ type: 'text', text: 'step 0 ' });
|
||||
expect(s0.some((p) => p.type === 'tool-echo')).toBe(true);
|
||||
|
||||
// The per-step slices are EXACTLY the corresponding prefix of the single-row
|
||||
// flush: assembleStepParts([step0, step1]) === finalParts[0 .. len0+len1].
|
||||
// This is what an off-by-one on `stepsPersisted-1` (a wrong `capturedSteps`
|
||||
// slice) or a shifted stepIndex breaks — the prefix no longer aligns.
|
||||
const prefixLen =
|
||||
(steps[0].parts as any[]).length + (steps[1].parts as any[]).length;
|
||||
expect(assembleStepParts([steps[0], steps[1]] as any)).toEqual(
|
||||
finalParts.slice(0, prefixLen),
|
||||
);
|
||||
|
||||
// The mid-run step markers advanced 1 -> 2 -> ... (the resume frontier), each
|
||||
// a shape-stable empty-parts marker equal to a single-row flush's marker.
|
||||
const markerCounts = updateSpy.mock.calls
|
||||
.map((c) => (c[2] as any)?.metadata)
|
||||
.filter(
|
||||
(m) =>
|
||||
m &&
|
||||
Array.isArray(m.parts) &&
|
||||
m.parts.length === 0 &&
|
||||
typeof m.stepsPersisted === 'number',
|
||||
)
|
||||
.map((m) => m.stepsPersisted);
|
||||
// Monotonic from 1, covering at least the two non-final steps.
|
||||
expect(markerCounts.slice(0, 2)).toEqual([1, 2]);
|
||||
expect(
|
||||
updateSpy.mock.calls
|
||||
.map((c) => (c[2] as any)?.metadata)
|
||||
.find((m) => m && m.stepsPersisted === 2),
|
||||
).toEqual(stepMarkerMetadata(2));
|
||||
} finally {
|
||||
updateSpy.mockRestore();
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
// --- F1: model-REPLAY hydrates a hard-crashed mid-run turn from the steps table
|
||||
it('replays a hard-crashed mid-run turn WITH its partial steps hydrated from the steps table', async () => {
|
||||
const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
|
||||
|
||||
// Prior turn: a genuine user question...
|
||||
await createMessage(db, {
|
||||
workspaceId,
|
||||
chatId,
|
||||
userId,
|
||||
role: 'user',
|
||||
content: 'What is in the design doc?',
|
||||
createdAt: new Date(Date.now() - 3000),
|
||||
});
|
||||
// ...and an assistant row that a HARD crash (SIGKILL/OOM) left mid-run: only a
|
||||
// step marker on the row (metadata.parts:[] , content:''), NO terminal
|
||||
// callback ever fired, so its real parts live ONLY in ai_chat_run_steps.
|
||||
const crashed = await createMessage(db, {
|
||||
workspaceId,
|
||||
chatId,
|
||||
role: 'assistant',
|
||||
content: '',
|
||||
status: 'aborted',
|
||||
metadata: stepMarkerMetadata(1),
|
||||
createdAt: new Date(Date.now() - 2000),
|
||||
});
|
||||
// The durable partial step: some reasoning text + a completed getPage tool
|
||||
// call (input + output), exactly what #183 step-granular durability preserves.
|
||||
await stepRepo.insertStep(
|
||||
crashed.id,
|
||||
workspaceId,
|
||||
0,
|
||||
assistantParts(
|
||||
[
|
||||
{
|
||||
text: 'HYDRATED_PARTIAL_STEP the doc says',
|
||||
toolCalls: [
|
||||
{ toolCallId: 'g1', toolName: 'getPage', input: { id: 'p1' } },
|
||||
],
|
||||
toolResults: [
|
||||
{
|
||||
toolCallId: 'g1',
|
||||
toolName: 'getPage',
|
||||
output: { id: 'p1', body: 'PARTIAL_TOOL_OUTPUT budget section' },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
],
|
||||
'',
|
||||
),
|
||||
);
|
||||
|
||||
// The NEXT turn: the model just answers. The service must REPLAY the crashed
|
||||
// assistant turn with its partial parts hydrated from the steps table.
|
||||
const model = new MockLanguageModelV3({
|
||||
doStream: async () => ({ stream: successStream() }),
|
||||
} as any);
|
||||
await runStream({
|
||||
model,
|
||||
chatId,
|
||||
body: { chatId, messages: [userUiMessage('Continue please')] },
|
||||
});
|
||||
|
||||
expect(model.doStreamCalls.length).toBeGreaterThan(0);
|
||||
const prompt = JSON.stringify(model.doStreamCalls[0].prompt);
|
||||
// The partial step's TEXT reached the model context (it would be an empty text
|
||||
// part without hydration — rowToUiMessage falls back to `content:''`).
|
||||
expect(prompt).toContain('HYDRATED_PARTIAL_STEP');
|
||||
// The partial TOOL RESULT survived too (durable in the steps table, replayed).
|
||||
expect(prompt).toContain('PARTIAL_TOOL_OUTPUT');
|
||||
// The genuine prior user turn is present as well (sanity: real history replay).
|
||||
expect(prompt).toContain('What is in the design doc?');
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -0,0 +1,169 @@
|
||||
import { Kysely } from 'kysely';
|
||||
import { AiChatController } from 'src/core/ai-chat/ai-chat.controller';
|
||||
import {
|
||||
assembleStepParts,
|
||||
assistantParts,
|
||||
stepMarkerMetadata,
|
||||
} from 'src/core/ai-chat/ai-chat.service';
|
||||
import { AiChatRepo } from '@docmost/db/repos/ai-chat/ai-chat.repo';
|
||||
import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
|
||||
import { AiChatRunStepRepo } from '@docmost/db/repos/ai-chat/ai-chat-run-step.repo';
|
||||
import type { User, Workspace } from '@docmost/db/types/entity.types';
|
||||
import {
|
||||
getTestDb,
|
||||
destroyTestDb,
|
||||
createWorkspace,
|
||||
createUser,
|
||||
createChat,
|
||||
createMessage,
|
||||
} from './db';
|
||||
|
||||
/**
|
||||
* #492 controller hydration (crash-before-finalize RESUME) on a LIVE Postgres.
|
||||
* `AiChatController.withReconstructedParts` is wired into getMessages/delta/export/
|
||||
* run, but `aiChatRunStepRepo` is OPTIONAL and every controller unit spec passes it
|
||||
* as `undefined`, so the hydration branch early-returns and NEVER executes in those
|
||||
* tests. This drives the real read path — a mid-run streaming row (marker only,
|
||||
* empty inline parts) PLUS its `ai_chat_run_steps` rows — through getMessages WITH
|
||||
* the repo present, exercising the `role==='assistant' && !rowHasInlineParts`
|
||||
* needy predicate, the workspace-scoped batch step fetch, and the endpoint binding.
|
||||
*/
|
||||
describe('#492 controller hydration read path [integration]', () => {
|
||||
let db: Kysely<any>;
|
||||
let aiChatRepo: AiChatRepo;
|
||||
let msgRepo: AiChatMessageRepo;
|
||||
let stepRepo: AiChatRunStepRepo;
|
||||
let workspaceId: string;
|
||||
let otherWorkspaceId: string;
|
||||
let userId: string;
|
||||
|
||||
// Build the controller WITH a real AiChatRunStepRepo injected (position 9), the
|
||||
// seam the unit specs leave undefined. Only the read-path deps are real.
|
||||
function buildController(): AiChatController {
|
||||
return new AiChatController(
|
||||
{} as any, // aiChatService
|
||||
{} as any, // aiChatRunService
|
||||
aiChatRepo,
|
||||
msgRepo,
|
||||
{} as any, // aiTranscription
|
||||
{} as any, // pageRepo
|
||||
undefined, // streamRegistry
|
||||
undefined, // environment
|
||||
stepRepo, // #492 aiChatRunStepRepo
|
||||
);
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
db = getTestDb();
|
||||
aiChatRepo = new AiChatRepo(db as any);
|
||||
msgRepo = new AiChatMessageRepo(db as any);
|
||||
stepRepo = new AiChatRunStepRepo(db as any);
|
||||
workspaceId = (await createWorkspace(db)).id;
|
||||
otherWorkspaceId = (await createWorkspace(db)).id;
|
||||
userId = (await createUser(db, workspaceId)).id;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await destroyTestDb();
|
||||
});
|
||||
|
||||
it('getMessages reconstructs a mid-run row from the steps table (finished rows untouched)', async () => {
|
||||
const chatId = (
|
||||
await createChat(db, { workspaceId, creatorId: userId })
|
||||
).id;
|
||||
const user = { id: userId } as User;
|
||||
const workspace = { id: workspaceId } as Workspace;
|
||||
|
||||
// A prior FINISHED assistant row that already carries inline parts — the needy
|
||||
// predicate must SKIP it (no step fetch), returned untouched.
|
||||
const finishedParts = assistantParts(
|
||||
[{ text: 'done earlier', toolCalls: [], toolResults: [] } as any],
|
||||
'',
|
||||
);
|
||||
await createMessage(db, {
|
||||
workspaceId,
|
||||
chatId,
|
||||
role: 'assistant',
|
||||
content: 'done earlier',
|
||||
status: 'completed',
|
||||
metadata: { parts: finishedParts, toolTraceVersion: 2, stepsPersisted: 1 },
|
||||
createdAt: new Date(Date.now() - 3000),
|
||||
});
|
||||
|
||||
// The mid-run row a crash-before-finalize left behind: a step marker only
|
||||
// (parts:[] , content:''), status 'streaming'. Its real parts live ONLY in the
|
||||
// steps table.
|
||||
const midRun = await createMessage(db, {
|
||||
workspaceId,
|
||||
chatId,
|
||||
role: 'assistant',
|
||||
content: '',
|
||||
status: 'streaming',
|
||||
metadata: stepMarkerMetadata(2),
|
||||
createdAt: new Date(Date.now() - 1000),
|
||||
});
|
||||
|
||||
const step0 = assistantParts(
|
||||
[
|
||||
{
|
||||
text: 'reasoning about the page',
|
||||
toolCalls: [
|
||||
{ toolCallId: 'g1', toolName: 'getPage', input: { id: 'p1' } },
|
||||
],
|
||||
toolResults: [
|
||||
{ toolCallId: 'g1', toolName: 'getPage', output: { id: 'p1', body: 'B' } },
|
||||
],
|
||||
} as any,
|
||||
],
|
||||
'',
|
||||
);
|
||||
const step1 = assistantParts(
|
||||
[{ text: 'partial synthesis so far', toolCalls: [], toolResults: [] } as any],
|
||||
'',
|
||||
);
|
||||
await stepRepo.insertStep(midRun.id, workspaceId, 0, step0);
|
||||
await stepRepo.insertStep(midRun.id, workspaceId, 1, step1);
|
||||
|
||||
// Workspace-scoping guard: a step row for the SAME message id under a DIFFERENT
|
||||
// workspace must NEVER leak into this workspace's reconstruction.
|
||||
await stepRepo.insertStep(midRun.id, otherWorkspaceId, 99, [
|
||||
{ type: 'text', text: 'FOREIGN_WORKSPACE_LEAK' },
|
||||
]);
|
||||
|
||||
const res = await buildController().getMessages(
|
||||
{ chatId } as any,
|
||||
{ limit: 50 } as any,
|
||||
user,
|
||||
workspace,
|
||||
);
|
||||
|
||||
const items = res.items as any[];
|
||||
const finished = items.find((r) => r.status === 'completed');
|
||||
const reconstructed = items.find((r) => r.id === midRun.id);
|
||||
|
||||
// The finished row passed through with its inline parts unchanged.
|
||||
expect(finished.metadata.parts).toEqual(finishedParts);
|
||||
|
||||
// The mid-run row's parts were reconstructed from the two step rows, in order,
|
||||
// exactly as assembleStepParts concatenates them — the client seed sees the
|
||||
// persisted progress with no change to itself.
|
||||
const expected = assembleStepParts([
|
||||
{ stepIndex: 0, parts: step0 },
|
||||
{ stepIndex: 1, parts: step1 },
|
||||
] as any);
|
||||
expect(reconstructed.metadata.parts).toEqual(expected);
|
||||
// The foreign-workspace step row did NOT leak in.
|
||||
expect(JSON.stringify(reconstructed.metadata.parts)).not.toContain(
|
||||
'FOREIGN_WORKSPACE_LEAK',
|
||||
);
|
||||
// Sanity: reconstruction produced real content (text + the paired tool part +
|
||||
// the second step's text), not an empty fallback.
|
||||
expect(reconstructed.metadata.parts).toContainEqual({
|
||||
type: 'text',
|
||||
text: 'reasoning about the page',
|
||||
});
|
||||
expect(
|
||||
(reconstructed.metadata.parts as any[]).some((p) => p.type === 'tool-getPage'),
|
||||
).toBe(true);
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -98,7 +98,6 @@ describe('AiChatRunStepRepo + reconstruct contract [integration]', () => {
|
||||
|
||||
const steps = await stepRepo.findByMessage(row.id, workspaceId);
|
||||
expect(steps.map((s) => s.stepIndex)).toEqual([0, 1]);
|
||||
expect(await stepRepo.countByMessage(row.id, workspaceId)).toBe(2);
|
||||
|
||||
// Batch fetch groups by message id in step order.
|
||||
const map = await stepRepo.findByMessageIds([row.id], workspaceId);
|
||||
|
||||
Reference in New Issue
Block a user