From 62482827c9b4e7d4965304597b2a701ca5dbfe8f Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sun, 12 Jul 2026 00:48:26 +0300 Subject: [PATCH] =?UTF-8?q?fix(ai-chat):=20=D1=80=D0=B5-=D1=80=D0=B5=D0=B2?= =?UTF-8?q?=D1=8C=D1=8E=20#518=20=E2=80=94=20CI-=D1=84=D0=B8=D0=B4=D0=B5?= =?UTF-8?q?=D0=BB=D0=B8=D1=82=D0=B8=20delta,=20attach-=D0=B4=D1=83=D0=B1?= =?UTF-8?q?=D0=BB=D1=8C,=20=D1=87=D0=B8=D1=81=D1=82=D0=BA=D0=B8=20(#491)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ребейз на обновлённый feat/490 (PR #510, Option-A): стек #491 перенесён с 6e872f2d на 6e42752a; run-fsm.ts/chat-thread.tsx приехали из develop-версии (W1/S4: RUN_ALREADY_ACTIVE несёт activeRunId + supersede-CAS адопция) — мои #491-правки легли сверху без конфликтов, W1 НЕ откачен (run-fsm.ts == develop байт-в-байт; RUN_ALREADY_ACTIVE диспатчит activeRunId). 1. [CI-fidelity] Дельта-курсор спек падал в CI unit-лейне: `.spec.ts` дефолтил на НЕмигрированный docmost → 5/6 ERROR relation does not exist (skip-гард ловит только connection-fail). Переименован в `*.int-spec.ts` (исключается из unit- regex `.spec.ts$`, гоняется в test:int, чей global-setup мигрирует docmost_test) + DSN по умолчанию → docmost_test. Теперь 6/6 реально исполняются в CI-верном окружении; overlap-мутация роняет RACE-тесты (не вакуумен). 2. [regression #137/#161] attach: отсутствие `n` схлопывалось в frontier 0 → finished-неротированный ран (coverageFloor 0) отдавал ВЕСЬ tail вместо 204; парамслесс/легаси-вкладка допишет полный replay → дубль. Различаем ОТСУТСТВИЕ `n` (null — не tail-aware) от `n=0` (tail-aware): контроллер шлёт null при missing/invalid; registry.attach(n: number|null) 204-ит finished-ран при n===null (старый `finished && !expectLive` гейт), n=0 по-прежнему отдаёт хвост. Тесты (registry + controller) + mutation-verify: нейтрализация гейта роняет их. 4. [conventions] Ring-кап env-var переименован RUN_STREAM_MAX_BUFFER_BYTES → AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES (префикс как у сиблингов) + запись в .env.example (дефолт 4MB, 0/invalid→дефолт, subscriber-cap=2×). 5. [docs] run-fsm.spec.md: item4 переписан («реализовано в #491 — дельта несёт run:{id,status}|null; клиент run-поле ещё не потребляет») + добавлена строка перехода POLL_IDLE_CAP stopping→idle (Review #4, редьюсер это делает). 6. [simplification] Удалена мёртвая цепочка reconstructRunParts / reconstructPartsFromRow (ноль прод-вызовов) + опц. messageRepo-инъекция в AiChatRunService + спек-блоки; вернётся с первым реальным вызывателем. Маркер metadata.stepsPersisted (реально используемый) сохранён. DROP-пункты ревьюера (осиротевший import, DELTA_POLL_MAX_ROWS) не трогаю. Прогон: server tsc 0, ai-chat unit 202, delta-int 6/6 (int-lane), int attach 6/6, client vitest 403, tsc client 0 ai-chat, mcp 834/0. FSM-инварианты #488 сохранены. Co-Authored-By: Claude Opus 4.8 (1M context) --- .env.example | 10 +++ .../features/ai-chat/state/run-fsm.spec.md | 9 ++- .../core/ai-chat/ai-chat-run.service.spec.ts | 73 ------------------- .../src/core/ai-chat/ai-chat-run.service.ts | 34 --------- .../ai-chat-stream-registry.service.ts | 37 +++++++--- .../ai-chat/ai-chat-stream-registry.spec.ts | 29 +++++++- .../ai-chat/ai-chat.controller.attach.spec.ts | 35 ++++++++- .../src/core/ai-chat/ai-chat.controller.ts | 14 +++- .../src/core/ai-chat/ai-chat.service.ts | 28 ------- .../core/ai-chat/ai-chat.step-marker.spec.ts | 43 +---------- ...=> ai-chat-message.repo.delta.int-spec.ts} | 12 ++- 11 files changed, 126 insertions(+), 198 deletions(-) rename apps/server/src/database/repos/ai-chat/{ai-chat-message.repo.delta.spec.ts => ai-chat-message.repo.delta.int-spec.ts} (92%) diff --git a/.env.example b/.env.example index 6d7dec43..51fc4d4d 100644 --- a/.env.example +++ b/.env.example @@ -302,6 +302,16 @@ MCP_DOCMOST_PASSWORD= # enabled for a workspace, and the same single-instance constraint applies (the # registry is process-local). # AI_CHAT_RESUMABLE_STREAM=false +# +# Per-run replay ring cap (#491), in BYTES, for the resumable-stream registry +# above. The registry buffers the run's recent SSE tail so a reopened tab can +# attach and continue from the step it already persisted; the ring is bounded and +# rotates on every confirmed step-persist. This caps the un-persisted tail between +# rotations — an overflow evicts the oldest frames and a late attach falls back to +# 204 -> degraded poll, so correctness never depends on the size. Default 4194304 +# (4MB); a 0/invalid value falls back to the default. The per-subscriber backpressure +# cap is derived as 2x this value. Only meaningful with AI_CHAT_RESUMABLE_STREAM on. +# AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES=4194304 # --- Run lifecycle tunables (#487) --- # These govern the universal run machinery (every turn is now a first-class run, diff --git a/apps/client/src/features/ai-chat/state/run-fsm.spec.md b/apps/client/src/features/ai-chat/state/run-fsm.spec.md index c40f3fbf..86ae3d73 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.spec.md +++ b/apps/client/src/features/ai-chat/state/run-fsm.spec.md @@ -48,6 +48,7 @@ Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`. | `RETRY` (manual, stalled banner) | stalled | polling(attach-none) **†** | `[armPoll]` | | `POLL_TERMINAL` (settled tail merged) | polling, reconnecting, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4) | | `POLL_IDLE_CAP` (inactivity cap) | polling, reconnecting | stalled | `[disarmPoll, cancelReconnect]` (commit 4a — no more silent) | +| `POLL_IDLE_CAP` (inactivity cap) | stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (Review #4: a Stop-armed poll with no SDK/terminal backstop gets a bounded exit — NOT `stalled`, Stop was already pressed so nothing to retry) | | `RUN_FACT{null}` (POST /run → null/terminal, 204) | reconnecting/attaching/polling/stopping | idle | `[cancelReconnect, disarmPoll]`, runFact←null (I3 fresh-negative gate) | | `RUN_FACT{runId}` | any | (same) | runFact←runId (pessimism toward an attempt) | | `STOP_REQUESTED` (user Stop) | streaming, reconnecting, polling | stopping **†** | `[stopRun, abortAttach, cancelReconnect, armPoll]` (poll drives the terminal — I4 exit by data) | @@ -150,8 +151,12 @@ message. Sources, in the order they update `ctx.runFact`: 3. **Attach outcomes:** `ATTACH_LIVE` (2xx) confirms active; a 204 on a non-stripped path is an authoritative NEGATIVE fact → the runtime dispatches `RUN_FACT{null}`, which cancels recovery (I3 fresh-negative gate). -4. **Poll (future resume-stack iteration #491):** the delta will carry the run field; - until then the poll drives to a terminal ROW, dispatched as `POLL_TERMINAL`. +4. **Poll (#491, implemented):** the degraded poll now hits the delta endpoint + (`POST /ai-chat/messages/delta`), which ALREADY carries the run fact + (`run: {id, status} | null`) alongside the changed rows. The client does NOT yet + consume that run field — it still drives to a terminal ROW (merged by id), + dispatched as `POLL_TERMINAL` — so the run field rides the wire for a future + client that settles straight off it. Pessimism rule: a stale-but-positive fact PERMITS entering recovery (attach); the 204 then cuts it. A fresh negative fact gates recovery OUT immediately. diff --git a/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts b/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts index 41640cb0..ade4563c 100644 --- a/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts @@ -771,76 +771,3 @@ describe('#487 AiChatRunService.supersede (CAS)', () => { expect(svc.hasZombie('run-1')).toBe(false); }); }); - -describe('AiChatRunService.reconstructRunParts (#491)', () => { - const ws = 'ws-1'; - - function makeMsgRepo(row: unknown) { - return { findById: jest.fn(async () => row) }; - } - - it('resolves run -> assistant row -> { parts, stepsPersisted }', async () => { - const repo = makeRepo({ - findById: jest.fn(async () => ({ - id: 'run-1', - assistantMessageId: 'm1', - })), - }); - const msgRepo = makeMsgRepo({ - content: '', - metadata: { - parts: [{ type: 'text', text: 'hi' }], - stepsPersisted: 3, - }, - }); - const svc = new AiChatRunService( - repo as never, - makeEnv() as never, - msgRepo as never, - ); - const res = await svc.reconstructRunParts('run-1', ws); - expect(res).toEqual({ - parts: [{ type: 'text', text: 'hi' }], - stepsPersisted: 3, - }); - expect(msgRepo.findById).toHaveBeenCalledWith('m1', ws); - }); - - it('returns null when the run has no linked assistant row yet (seed window)', async () => { - const repo = makeRepo({ - findById: jest.fn(async () => ({ id: 'run-1', assistantMessageId: null })), - }); - const msgRepo = makeMsgRepo({ content: 'x', metadata: null }); - const svc = new AiChatRunService( - repo as never, - makeEnv() as never, - msgRepo as never, - ); - expect(await svc.reconstructRunParts('run-1', ws)).toBeNull(); - expect(msgRepo.findById).not.toHaveBeenCalled(); - }); - - it('returns null when the run does not exist', async () => { - const repo = makeRepo({ findById: jest.fn(async () => undefined) }); - const msgRepo = makeMsgRepo({}); - const svc = new AiChatRunService( - repo as never, - makeEnv() as never, - msgRepo as never, - ); - expect(await svc.reconstructRunParts('nope', ws)).toBeNull(); - }); - - it('returns null when the linked assistant row was deleted', async () => { - const repo = makeRepo({ - findById: jest.fn(async () => ({ id: 'run-1', assistantMessageId: 'm1' })), - }); - const msgRepo = makeMsgRepo(undefined); - const svc = new AiChatRunService( - repo as never, - makeEnv() as never, - msgRepo as never, - ); - expect(await svc.reconstructRunParts('run-1', ws)).toBeNull(); - }); -}); diff --git a/apps/server/src/core/ai-chat/ai-chat-run.service.ts b/apps/server/src/core/ai-chat/ai-chat-run.service.ts index 37ab3827..1530c3b5 100644 --- a/apps/server/src/core/ai-chat/ai-chat-run.service.ts +++ b/apps/server/src/core/ai-chat/ai-chat-run.service.ts @@ -1,9 +1,6 @@ import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { AiChatRunRepo } from '@docmost/db/repos/ai-chat/ai-chat-run.repo'; -import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo'; import { AiChatRun } from '@docmost/db/types/entity.types'; -import { reconstructPartsFromRow } from './ai-chat.service'; -import type { UIMessage } from 'ai'; import { isUniqueViolation, violatedConstraint } from '@docmost/db/utils'; import { EnvironmentService } from '../../integrations/environment/environment.service'; @@ -215,10 +212,6 @@ export class AiChatRunService implements OnModuleInit { constructor( private readonly runRepo: AiChatRunRepo, private readonly environment: EnvironmentService, - // #491: OPTIONAL so the many 2-arg test constructions of this service compile - // unchanged; Nest always injects the real repo in production. Only touched by - // reconstructRunParts (the live-run read interface). - private readonly messageRepo?: AiChatMessageRepo, ) {} /** @@ -791,33 +784,6 @@ export class AiChatRunService implements OnModuleInit { return this.runRepo.findById(runId, workspaceId); } - /** - * #491 CONTRACT — the SINGLE interface for reading a LIVE run's output: resolve - * the run to its assistant-message projection (#183) and return its persisted - * `parts` plus `stepsPersisted` — the count of FINISHED steps confirmed on disk, - * written atomically with the parts (see {@link flushAssistant} / the step - * marker). Consumers: tail-only attach (commit 3 — slices at "step > N"), the - * degraded-poll delta (rows already carry the marker in metadata), and export. - * - * Returns `null` when the run does not exist, has no linked assistant row yet - * (the seed window), or the row was deleted — a consumer treats null as "nothing - * confirmed" and stays safe (attach 204 / full seed). `stepsPersisted` is 0 for a - * pre-#491 row with no marker, which is likewise the safe floor. - */ - async reconstructRunParts( - runId: string, - workspaceId: string, - ): Promise<{ parts: UIMessage['parts']; stepsPersisted: number } | null> { - const run = await this.runRepo.findById(runId, workspaceId); - if (!run?.assistantMessageId || !this.messageRepo) return null; - const row = await this.messageRepo.findById( - run.assistantMessageId, - workspaceId, - ); - if (!row) return null; - return reconstructPartsFromRow(row); - } - /** The active run on a chat, if any (used to reject a concurrent start with a * clean 409 before committing to the stream). */ getActiveForChat( diff --git a/apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts b/apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts index 743bad30..e338d01a 100644 --- a/apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts +++ b/apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts @@ -88,14 +88,14 @@ export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000; * its persisted frontier come from the seed, not the ring). The ring stays bounded * because it rotates on every confirmed persist; this cap is only the ceiling for * the un-persisted tail between rotations. Env-tunable via - * RUN_STREAM_MAX_BUFFER_BYTES (bytes); a 0/invalid value falls back to this. + * AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES (bytes); a 0/invalid value falls back to this. */ -export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024; +export const AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024; // 2× the ring cap: a just-written full-tail burst alone can never trip the // per-subscriber cap (see controller); only a genuinely stalled socket can. This // derivative relationship is preserved even when the ring cap is env-overridden. -export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES; +export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES; /** * A finish-step boundary frame is exactly `data: {"type":"finish-step"...}\n\n` @@ -110,12 +110,12 @@ const FINISH_STEP_FRAME_PREFIX = 'data: {"type":"finish-step"'; /** Resolve the ring cap from the environment, falling back to the default. */ function resolveMaxBufferBytes(): number { - const raw = process.env.RUN_STREAM_MAX_BUFFER_BYTES; - if (!raw) return RUN_STREAM_MAX_BUFFER_BYTES; + const raw = process.env.AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES; + if (!raw) return AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES; const parsed = Number(raw); return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) - : RUN_STREAM_MAX_BUFFER_BYTES; + : AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES; } export interface RunStreamCallbacks { @@ -321,7 +321,11 @@ export class AiChatStreamRegistryService implements OnModuleDestroy { async attach( chatId: string, anchor: string | undefined, - n: number, + // The client's persisted step frontier. `null` = a NOT-tail-aware client (no + // `n` query param) — a legacy/parameterless tab that expects the old + // "finished -> 204 -> poll" contract; distinct from `0` (a tail-aware client + // with nothing persisted yet). + n: number | null, cb: RunStreamCallbacks, ): Promise { const entry = this.entries.get(chatId); @@ -329,14 +333,25 @@ export class AiChatStreamRegistryService implements OnModuleDestroy { // Invariant 6: cross-run replay is forbidden. Before bind, assistantMessageId // is undefined and mismatches any anchor -> 204 -> client restore+poll path. if (anchor && entry.assistantMessageId !== anchor) return null; + // #491 regression guard (#137/#161 dup): a NOT-tail-aware client (no `n`) + // resuming a FINISHED run must 204 and poll — the old `finished && !expectLive` + // gate. Without this, a missing `n` collapsing to frontier 0 would serve the + // WHOLE tail of a finished, NON-rotated run (coverageFloor 0), and a + // parameterless client that never stripped its transcript would APPEND that + // full replay onto the steps it already shows -> duplicated text. A tail-aware + // client (n present, incl. n=0) still gets the tail past its frontier. + if (entry.finished && n === null) return null; // A finished entry with NOTHING in the ring (aborted before the first frame, // or fully overflowed) has no tail to deliver -> 204 -> the client polls. if (entry.finished && entry.frames.length === 0) return null; + // A LIVE run with no `n` (legacy parameterless) replays from step 0 (the old + // behavior); a tail-aware client resumes from its frontier. + const frontier = n ?? 0; const floor = this.coverageFloor(entry); - if (floor > n) { + if (floor > frontier) { this.logger.warn( `run-stream attach gap for run=${entry.runId}: coverageFloor=${floor} ` + - `> client n=${n} -> 204 (client refetches + re-attaches)`, + `> client frontier=${frontier} -> 204 (client refetches + re-attaches)`, ); return null; } @@ -345,7 +360,7 @@ export class AiChatStreamRegistryService implements OnModuleDestroy { const sliceTail = (): string[] => { const out: string[] = [startFrame]; for (let i = 0; i < entry.frames.length; i++) { - if (entry.stamps[i] >= n) out.push(entry.frames[i]); + if (entry.stamps[i] >= frontier) out.push(entry.frames[i]); } return out; }; @@ -368,7 +383,7 @@ export class AiChatStreamRegistryService implements OnModuleDestroy { pendingBytes: 0, overflowed: false, pendingEnd: false, - minStamp: n, + minStamp: frontier, }; // Register + snapshot in the SAME synchronous block (invariant 4). No await // separates them, so a concurrently ingested frame cannot be lost/duplicated. diff --git a/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts b/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts index 06df6d75..9388b2b6 100644 --- a/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts @@ -1,6 +1,6 @@ import { AiChatStreamRegistryService, - RUN_STREAM_MAX_BUFFER_BYTES, + AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES, RUN_STREAM_RETAIN_FINISHED_MS, RunStreamCallbacks, } from './ai-chat-stream-registry.service'; @@ -377,7 +377,7 @@ describe('AiChatStreamRegistryService step-aligned retention (#491)', () => { const src = makePushStream(); registry.bind(CHAT, 'run-1', 'assist-1', src.stream); // Step 0: a fat step that blows past the cap with NO persist confirmation. - const big = 'x'.repeat(Math.floor(RUN_STREAM_MAX_BUFFER_BYTES / 2)); + const big = 'x'.repeat(Math.floor(AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES / 2)); src.push(textDelta('t0', big)); // 0 src.push(textDelta('t0', big)); // 0 src.push(textDelta('t0', big)); // 0 -> overflow evicts stamp-0 frames @@ -420,7 +420,7 @@ describe('AiChatStreamRegistryService step-aligned retention (#491)', () => { registry.open(CHAT, 'run-1'); const src = makePushStream(); registry.bind(CHAT, 'run-1', 'assist-1', src.stream); - const big = 'x'.repeat(Math.floor(RUN_STREAM_MAX_BUFFER_BYTES / 2)); + const big = 'x'.repeat(Math.floor(AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES / 2)); src.push(textDelta('t0', big)); // 0 src.push(textDelta('t0', big)); // 0 src.push(finishStep()); // 0 (still stamp 0) @@ -460,6 +460,29 @@ describe('AiChatStreamRegistryService step-aligned retention (#491)', () => { expect(entryOf().subscribers.size).toBe(0); }); + it('#491 regression (#137/#161 dup): a PARAMETERLESS attach (n=null) to a finished NON-rotated run -> 204, but n=0 still gets the tail', async () => { + // A finished, non-rotated run: frames present, coverageFloor 0. A missing `n` + // (null — a legacy/parameterless tab that never stripped its transcript) must + // 204 -> poll, NOT receive the whole tail it would append (duplicate). A + // tail-aware client (n=0 present) still resumes. + registry.open(CHAT, 'run-1'); + const src = makePushStream(); + registry.bind(CHAT, 'run-1', 'assist-1', src.stream); + src.push(textDelta('t0', 'a')); // 0 + src.push(finishStep()); // 0 + src.push(finish()); // 1 + src.close(); + await flush(); + // NOT rotated (no confirmPersistedStep) -> stamps[0]=0, coverageFloor=0. + // MUTATION-VERIFY: revert the `finished && n === null -> null` gate (default n + // to 0) and the parameterless attach below serves the full tail instead of 204. + expect(await registry.attach(CHAT, 'assist-1', null, collector().cb)).toBeNull(); + // A tail-aware client at frontier 0 IS served (the distinction: null != 0). + const tailAware = await registry.attach(CHAT, 'assist-1', 0, collector().cb); + expect(tailAware).not.toBeNull(); + expect(tailAware!.finished).toBe(true); + }); + it('confirmPersistedStep is monotonic and identity-checked', async () => { registry.open(CHAT, 'run-1'); const src = makePushStream(); diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.attach.spec.ts b/apps/server/src/core/ai-chat/ai-chat.controller.attach.spec.ts index 05bc19b8..50666847 100644 --- a/apps/server/src/core/ai-chat/ai-chat.controller.attach.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.controller.attach.spec.ts @@ -182,7 +182,38 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => { ); }); - it('floors n to 0 when the query is absent/invalid', async () => { + it('#491: an ABSENT/invalid n passes null (not 0) so a finished run 204s (not-tail-aware)', async () => { + // Distinguishing a MISSING `n` from `n=0` is the #137/#161 dup guard: a + // parameterless/legacy tab must be handed null (-> the registry 204s a finished + // run) rather than frontier 0 (which would serve a finished non-rotated run's + // whole tail). MUTATION-VERIFY: revert to `Number(n) || 0` and this asserts 0. + const { controller, streamRegistry } = makeController({ + chat: owned, + attachment: null, + }); + for (const bad of [undefined, '', 'abc']) { + streamRegistry.attach.mockClear(); + const { res } = makeRawRes(); + const { req } = makeReq(); + await controller.attachRunStream( + 'c1', + undefined, + bad, + req, + res, + user, + workspace, + ); + expect(streamRegistry.attach).toHaveBeenCalledWith( + 'c1', + undefined, + null, + expect.anything(), + ); + } + }); + + it('#491: a PRESENT n=0 passes 0 (tail-aware, distinct from absent)', async () => { const { controller, streamRegistry } = makeController({ chat: owned, attachment: null, @@ -192,7 +223,7 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => { await controller.attachRunStream( 'c1', undefined, - undefined, + '0', req, res, user, diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.ts b/apps/server/src/core/ai-chat/ai-chat.controller.ts index b97ad6ff..2b8aa6ab 100644 --- a/apps/server/src/core/ai-chat/ai-chat.controller.ts +++ b/apps/server/src/core/ai-chat/ai-chat.controller.ts @@ -365,9 +365,17 @@ export class AiChatController { @AuthWorkspace() workspace: Workspace, ): Promise { await this.assertOwnedChat(chatId, user, workspace); // same gate as getRun - // The client's persisted step frontier. A missing/invalid value floors to 0 - // ("give me everything") which, past any rotation, safely 204s. - const frontier = Number.isFinite(Number(n)) ? Math.max(0, Number(n)) : 0; + // The client's persisted step frontier. #491: distinguish a MISSING/invalid `n` + // (null — a NOT-tail-aware, legacy/parameterless tab expecting the old + // "finished -> 204 -> poll" contract) from `n=0` (a tail-aware client with + // nothing persisted yet). Passing 0 for a missing `n` would serve a finished, + // non-rotated run's WHOLE tail and a parameterless client would append it onto + // the steps it already shows -> #137/#161 duplicate. null makes the registry + // 204 such a finished run (see attach); a tail-aware n=0 still resumes. + const frontier: number | null = + n === undefined || n === '' || !Number.isFinite(Number(n)) + ? null + : Math.max(0, Number(n)); // The per-subscriber backpressure cap tracks the (env-tunable) ring cap. const subscriberCap = this.streamRegistry?.subscriberMaxBufferedBytes ?? diff --git a/apps/server/src/core/ai-chat/ai-chat.service.ts b/apps/server/src/core/ai-chat/ai-chat.service.ts index 83b56e8d..babe5f17 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.ts +++ b/apps/server/src/core/ai-chat/ai-chat.service.ts @@ -2735,34 +2735,6 @@ export function rowToUiMessage(row: AiChatMessage): Omit & { return { id: row.id, role, parts: parts as UIMessage['parts'] }; } -/** - * The persisted parts + step marker of a run's assistant row (#491). This is the - * pure core of the `AiChatRunService.reconstructRunParts(runId)` contract — the - * SINGLE interface for reading a LIVE run's output — given the already-resolved - * row. Returns the `metadata.parts` (falling back to a single text part from - * `content` for a pre-#183 row) and `stepsPersisted` — the count of FINISHED steps - * whose parts are present, written atomically with the parts by - * {@link flushAssistant}. A missing marker (an old row) reads as 0, so a consumer - * treats it as "nothing confirmed" and stays safe (attach 204 / full seed). Pure. - */ -export function reconstructPartsFromRow( - row: Pick | null | undefined, -): { parts: UIMessage['parts']; stepsPersisted: number } { - const meta = (row?.metadata ?? {}) as { - parts?: UIMessage['parts']; - stepsPersisted?: number; - }; - const parts = - Array.isArray(meta.parts) && meta.parts.length > 0 - ? meta.parts - : textPart(row?.content ?? ''); - const stepsPersisted = - typeof meta.stepsPersisted === 'number' && meta.stepsPersisted >= 0 - ? meta.stepsPersisted - : 0; - return { parts: parts as UIMessage['parts'], stepsPersisted }; -} - /** * The persisted-row patch shape produced by {@link flushAssistant}. It is the * SAME shape the assistant repo insert/update consume (content + toolCalls + diff --git a/apps/server/src/core/ai-chat/ai-chat.step-marker.spec.ts b/apps/server/src/core/ai-chat/ai-chat.step-marker.spec.ts index 98c53e4b..a4398a3d 100644 --- a/apps/server/src/core/ai-chat/ai-chat.step-marker.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.step-marker.spec.ts @@ -1,15 +1,11 @@ -import { - flushAssistant, - reconstructPartsFromRow, -} from './ai-chat.service'; -import type { AiChatMessage } from '@docmost/db/types/entity.types'; +import { flushAssistant } from './ai-chat.service'; /** * #491 STEP MARKER — `metadata.stepsPersisted` is written by the SAME flush that * builds `metadata.parts`, so the marker can never disagree with the persisted * parts (the step-alignment anchor the resume stack builds on). These are * PROPERTY tests: they assert the marker tracks the number of FINISHED steps for - * every flush shape, and that `reconstructPartsFromRow` reads it back safely. + * every flush shape. */ // A finished step carrying one line of text and one tool call/result. @@ -67,38 +63,3 @@ describe('flushAssistant step marker (#491)', () => { expect(f.metadata.stepsPersisted).toBe(3); }); }); - -describe('reconstructPartsFromRow (#491)', () => { - const row = (metadata: unknown, content = ''): AiChatMessage => - ({ content, metadata }) as unknown as AiChatMessage; - - it('reads parts + stepsPersisted from metadata', () => { - const f = flushAssistant([step(0), step(1)], '', 'streaming'); - const r = reconstructPartsFromRow(row(f.metadata, f.content)); - expect(r.stepsPersisted).toBe(2); - expect(r.parts).toEqual(f.metadata.parts); - }); - - it('defaults stepsPersisted to 0 for a pre-#491 row with no marker (safe floor)', () => { - const r = reconstructPartsFromRow( - row({ parts: [{ type: 'text', text: 'x' }] }), - ); - expect(r.stepsPersisted).toBe(0); - expect(r.parts).toEqual([{ type: 'text', text: 'x' }]); - }); - - it('falls back to a single text part from content when no metadata.parts', () => { - const r = reconstructPartsFromRow(row(null, 'plain content')); - expect(r.stepsPersisted).toBe(0); - expect(r.parts).toEqual([{ type: 'text', text: 'plain content' }]); - }); - - it('null/undefined row → no parts, marker 0 (safe empty floor)', () => { - // textPart('') is empty, matching rowToUiMessage's fallback for an empty row. - expect(reconstructPartsFromRow(null)).toEqual({ - parts: [], - stepsPersisted: 0, - }); - expect(reconstructPartsFromRow(undefined).stepsPersisted).toBe(0); - }); -}); diff --git a/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.int-spec.ts similarity index 92% rename from apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts rename to apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.int-spec.ts index 94ccddeb..063b2d15 100644 --- a/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts +++ b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.int-spec.ts @@ -26,6 +26,15 @@ import { AiChatRunRepo } from './ai-chat-run.repo'; * 4. the overlap GUARANTEES repeats across close polls — the contract behind the * client's idempotent merge (mergeById). * + * INTEGRATION lane (`*.int-spec.ts`): runs under `test:int`, whose global-setup + * DROPS + RE-CREATES + MIGRATES `docmost_test`, so the real `ai_chat_messages` / + * `ai_chat_runs` tables EXIST here. (It was previously a `.spec.ts` defaulting to + * the UNmigrated dev `docmost`; in the CI unit lane — where `WAL_TEST_DATABASE_URL` + * is unset and only `test:int` migrates — that meant 5/6 ERROR + * `relation "ai_chat_messages" does not exist`, silently voiding coverage of the + * risky cursor/overlap logic. Renaming to `.int-spec.ts` + defaulting the DSN to + * `docmost_test` fixes the CI fidelity.) + * * FK triggers are bypassed (`session_replication_role = replica`) so synthetic * chat/workspace ids need no parent fixtures; a single pooled connection (max 1) * keeps that session setting for every query. SKIPS cleanly when the DB is @@ -33,7 +42,8 @@ import { AiChatRunRepo } from './ai-chat-run.repo'; */ const CONN = process.env.WAL_TEST_DATABASE_URL ?? - 'postgresql://docmost:docmost_dev_pw@localhost:5432/docmost'; + process.env.TEST_DATABASE_URL ?? + 'postgresql://docmost:docmost_dev_pw@localhost:5432/docmost_test'; let db: Kysely; let sqlClient: ReturnType;