fix(ai-chat): ре-ревью #518 — CI-фиделити delta, attach-дубль, чистки (#491)

Ребейз на обновлённый 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) <noreply@anthropic.com>
2026-07-12 00:48:26 +03:00
parent a0ecf21cb5
commit 3550bfa411
11 changed files with 126 additions and 198 deletions
@@ -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,
@@ -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.
@@ -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();
-  });
-});
@@ -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(
@@ -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<RunStreamAttachment | null> {
    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.
@@ -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();
@@ -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,
@@ -365,9 +365,17 @@ export class AiChatController {
    @AuthWorkspace() workspace: Workspace,
  ): Promise<void> {
    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 ??
@@ -2747,34 +2747,6 @@ export function rowToUiMessage(row: AiChatMessage): Omit<UIMessage, 'id'> & {
  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<AiChatMessage, 'content' | 'metadata'> | 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 +
@@ -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);
-  });
-});
@@ -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<any>;
 let sqlClient: ReturnType<typeof postgres>;