Merge remote-tracking branch 'gitea/develop' into feat/492-incremental-render

Раньше каждый onStepFinish переписывал ВСЮ строку ассистента (растущий
metadata.parts jsonb со всеми выводами инструментов) → O(n²) объёма записи
на прогон: под MVCC/TOAST апдейт jsonb переписывает всю версию строки, так
что шаг k пишет ~k×вывод. Прогон из 50 шагов по ~100 КБ = сотни МБ WAL и
мёртвых кортежей за ход, что молотит autovacuum. (#490 убрал только ВТОРУЮ
копию в tool_calls; сам metadata.parts всё ещё рос и переписывался.)

Теперь каждый завершённый шаг ДОПИСЫВАЕТСЯ отдельной строкой в лёгкую
таблицу ai_chat_run_steps (только парты этого шага), а строка сообщения
получает дешёвый маркер (stepsPersisted + toolTraceVersion, без растущего
блоба parts). Полный metadata.parts собирается ОДИН раз на финализации.
НЕ jsonb-append (||): апдейт всё равно переписывает всю TOAST-версию —
экономится только сетевой payload, а WAL/мёртвые кортежи остаются; поэтому
именно ОТДЕЛЬНАЯ таблица + INSERT.

Три обязательные интеграции:
- reconstructRunParts(row, stepRows) → { parts, stepsPersisted }: единый
  шов переключения бэкенда. Читает парты из СТРОКИ, если она уже несёт
  inline-parts (старые записи + ЛЮБАЯ финализированная), иначе из ТАБЛИЦЫ
  ШАГОВ (mid-run запись #492). Дискриминатор — наличие непустого
  metadata.parts (флаг схемы не нужен). Потребители (attach-seed,
  delta-poll, export, reconnect) прогоняют строки через hydrateAssistantParts
  на границе чтения — их контракт/вывод не меняется, старые и новые записи
  восстанавливаются идентично.
- сигнал ротации кольца реестра #491 (confirmPersistedStep) теперь стреляет
  на подтверждённый INSERT шага, под тем же контрактом (updateStreaming
  возвращает stepsPersisted / null).
- era-marker toolTraceVersion (#490) больше не ставится полной переписью —
  ставится в маркере шага и на финализации (flushAssistant), остаётся
  консистентным.

Полная обратная совместимость: прогон, записанный по-старому (полная строка,
без строк шагов), восстанавливается/attach/export идентично. При отсутствии
репозитория шагов (позиционные тест-конструкции) — фолбэк на прежний
полнострочный flush (без регрессии, только без выигрыша WAL).

Тесты (реальный pg, int-lane):
- WAL-дельта (pg_current_wal_lsn) на прогоне 40×100КБ: new=4.3МБ vs
  old=90.3МБ (20.8x) — O(Σ шагов) против O(n²); старый путь в тесте И есть
  ревертнутое поведение (мутация-проверка).
- reconstruct-контракт: new-style (таблица шагов) и old-style (inline) прогоны
  восстанавливаются в идентичные parts; hydrate заполняет строку.
- миграция up/down roundtrip.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/client/src/features/ai-chat/components/message-item-memo.test.tsx

@@ -27,6 +27,7 @@ vi.mock("@/features/ai-chat/utils/markdown.ts", async () => {
 
 import MessageItem from "./message-item";
 import { messageSignature } from "@/features/ai-chat/utils/message-signature.ts";
+import { splitPlainChunks } from "./streaming-plain-text";
 
 // matchMedia (read by MantineProvider) is stubbed globally in vitest.setup.ts.
 
@@ -114,3 +115,89 @@ describe("MessageItem markdown memoization", () => {
     expect(queryByText("streamed answer")).not.toBeNull();
   });
 });
+
+// PERF SMOKE (#492): the whole point of the incremental streaming render is that
+// the ANSWER path costs O(number of markdown blocks), NOT O(number of throttled
+// ~20Hz ticks). Pre-#492 the finalized MarkdownPart re-parsed the WHOLE growing
+// answer on every delta — a synthetic ~100 KB stream measured 394 renderChatMarkdown
+// calls (one per tick). With the incremental render each STABILIZED block is parsed
+// exactly once (memoized in MarkdownChunk) and the live tail is cheap plain text, so
+// the call count collapses to ~= the block count regardless of tick granularity.
+describe("MessageItem streaming answer render is O(blocks), not O(ticks)", () => {
+  // ~100 KB answer. Each section is a heading + a paragraph — TWO blank-line
+  // delimited markdown blocks — so the safe-cut block count is ~2× the section
+  // count. The perf claim is about the BLOCK count (the memoization granularity),
+  // measured directly with splitPlainChunks below, not the section count.
+  const buildAnswer = () => {
+    const SECTIONS = 100;
+    const paragraphs: string[] = [];
+    for (let i = 0; i < SECTIONS; i++) {
+      paragraphs.push(`## Section ${i}\n\n` + "lorem ipsum dolor ".repeat(55));
+    }
+    const full = paragraphs.join("\n\n");
+    // The number of memoized markdown blocks the incremental render splits into
+    // (all but the live tail are parsed once each).
+    return { full, blocks: splitPlainChunks(full).length };
+  };
+
+  const streamMsg = (text: string, state: "streaming" | "done"): UIMessage =>
+    ({
+      id: "m1",
+      role: "assistant",
+      parts: [{ type: "text", text, state }],
+    }) as UIMessage;
+
+  it("parses each block ~once over a 100KB stream (≈blocks, ≪ ticks)", () => {
+    renderChatMarkdownSpy.mockClear();
+    const { full, blocks } = buildAnswer();
+    const CHUNK = 128; // a realistic ~20Hz throttled delta size
+    const ticks = Math.ceil(full.length / CHUNK);
+
+    let msg = streamMsg(full.slice(0, CHUNK), "streaming");
+    const { rerender } = render(
+      <MantineProvider>
+        <MessageItem
+          message={msg}
+          signature={messageSignature(msg)}
+          turnStreaming
+        />
+      </MantineProvider>,
+    );
+    for (let end = 2 * CHUNK; end < full.length; end += CHUNK) {
+      msg = streamMsg(full.slice(0, end), "streaming");
+      rerender(
+        <MantineProvider>
+          <MessageItem
+            message={msg}
+            signature={messageSignature(msg)}
+            turnStreaming
+          />
+        </MantineProvider>,
+      );
+    }
+    // Finalize: the streaming→done flip renders the whole answer through ONE
+    // canonical pass (visual parity), so the finished DOM matches the pre-#492
+    // output. This is the single extra parse on top of the per-block ones.
+    const done = streamMsg(full, "done");
+    rerender(
+      <MantineProvider>
+        <MessageItem message={done} signature={messageSignature(done)} />
+      </MantineProvider>,
+    );
+
+    const calls = renderChatMarkdownSpy.mock.calls.length;
+    // Sanity: the stream really had far more ticks than blocks (else the test is
+    // vacuous — the point is that calls scale with blocks, not ticks).
+    expect(ticks).toBeGreaterThan(blocks * 3);
+    // O(blocks): each stabilized block parsed once + the single final whole-text
+    // parse. A small constant absorbs the finalize render and the live-tail block;
+    // the load-bearing claim is the bound below.
+    expect(calls).toBeLessThanOrEqual(blocks + 2);
+    // ≪ ticks — and, non-vacuously, the blocks WERE parsed (not skipped entirely).
+    expect(calls).toBeLessThan(ticks / 3);
+    expect(calls).toBeGreaterThan(blocks / 2);
+    // MUTATION-VERIFY (documented, not run here): dropping the `memo()` wrapper on
+    // MarkdownChunk (so every stable block re-parses each tick) drives `calls`
+    // toward `ticks` (~394), reddening both upper-bound assertions above.
+  });
+});

apps/client/src/features/ai-chat/components/message-item.render.test.tsx

@@ -0,0 +1,112 @@
+import { describe, expect, it, vi } from "vitest";
+import { render } from "@testing-library/react";
+import { MantineProvider } from "@mantine/core";
+import type { UIMessage } from "@ai-sdk/react";
+
+// Stub react-i18next (the component reads `useTranslation`). Mirrors the other
+// message-item specs.
+vi.mock("react-i18next", () => ({
+  useTranslation: () => ({ t: (key: string) => key }),
+}));
+
+import MessageItem from "./message-item";
+import { messageSignature } from "@/features/ai-chat/utils/message-signature.ts";
+// The REAL canonical renderer (NOT the spy the memo test installs): this file
+// exercises the actual markdown output so the visual-regression assertions below
+// compare against genuine HTML (incl. the schema's `<li><p>` wrappers).
+import { renderChatMarkdown } from "@/features/ai-chat/utils/markdown.ts";
+import classes from "./ai-chat.module.css";
+
+const msg = (
+  parts: UIMessage["parts"],
+  extra?: Partial<UIMessage>,
+): UIMessage =>
+  ({ id: "m1", role: "assistant", parts, ...extra }) as UIMessage;
+
+const renderRow = (message: UIMessage, turnStreaming = false) =>
+  render(
+    <MantineProvider>
+      <MessageItem
+        message={message}
+        signature={messageSignature(message)}
+        turnStreaming={turnStreaming}
+      />
+    </MantineProvider>,
+  );
+
+// A rich multi-block answer that exercises headings, a list (the `<li><p>` case
+// the scoped CSS tightens), inline emphasis, and multiple paragraphs.
+const ANSWER = [
+  "# Заголовок",
+  "",
+  "Первый абзац с **жирным** и `кодом`.",
+  "",
+  "- пункт один",
+  "- пункт два",
+  "",
+  "Второй абзац.",
+].join("\n");
+
+describe("MessageItem final render — visual parity with the canonical pipeline", () => {
+  it("a finalized text part renders exactly renderChatMarkdown(text)", () => {
+    const { container } = renderRow(
+      msg([{ type: "text", text: ANSWER, state: "done" }]),
+    );
+    const block = container.querySelector(`.${classes.markdown}`);
+    expect(block).not.toBeNull();
+    // Byte-for-byte the canonical output (the SAME whole-text pass the pre-#492
+    // MarkdownPart produced), including `<li><p>…</p></li>` wrappers.
+    expect(block!.innerHTML).toBe(renderChatMarkdown(ANSWER, {}));
+    // The list wrapper is really present (guards against a vacuous empty render).
+    expect(container.querySelectorAll("li p").length).toBe(2);
+  });
+
+  it("the streaming incremental view CONVERGES to the canonical render on finish", () => {
+    // Mount mid-stream (live tail) — the DOM here is the incremental view.
+    const { container, rerender } = render(
+      <MantineProvider>
+        <MessageItem
+          message={msg([{ type: "text", text: ANSWER, state: "streaming" }])}
+          signature={messageSignature(
+            msg([{ type: "text", text: ANSWER, state: "streaming" }]),
+          )}
+          turnStreaming
+        />
+      </MantineProvider>,
+    );
+    // Finish the turn: state flips to done AND the turn is no longer streaming.
+    const done = msg([{ type: "text", text: ANSWER, state: "done" }]);
+    rerender(
+      <MantineProvider>
+        <MessageItem message={done} signature={messageSignature(done)} />
+      </MantineProvider>,
+    );
+    // After finish there is exactly ONE canonical markdown container whose HTML is
+    // the whole-text render — identical to the non-streaming path above.
+    const blocks = container.querySelectorAll(`.${classes.markdown}`);
+    expect(blocks.length).toBe(1);
+    expect(blocks[0].innerHTML).toBe(renderChatMarkdown(ANSWER, {}));
+  });
+
+  it("neutralizeInternalLinks is honored on the finalized render", () => {
+    const linkAnswer = "См. [страницу](/p/abc).";
+    const { container } = render(
+      <MantineProvider>
+        <MessageItem
+          message={msg([{ type: "text", text: linkAnswer, state: "done" }])}
+          signature={messageSignature(
+            msg([{ type: "text", text: linkAnswer, state: "done" }]),
+          )}
+          neutralizeInternalLinks
+        />
+      </MantineProvider>,
+    );
+    const block = container.querySelector(`.${classes.markdown}`);
+    expect(block!.innerHTML).toBe(
+      renderChatMarkdown(linkAnswer, { neutralizeInternalLinks: true }),
+    );
+    // The internal link was made inert (no href) by the neutralization flag.
+    const a = container.querySelector("a");
+    expect(a?.hasAttribute("href")).toBe(false);
+  });
+});

apps/client/src/features/ai-chat/components/message-item.tsx

@@ -4,6 +4,7 @@ import { useTranslation } from "react-i18next";
 import type { UIMessage } from "@ai-sdk/react";
 import ToolCallCard from "@/features/ai-chat/components/tool-call-card.tsx";
 import ReasoningBlock from "@/features/ai-chat/components/reasoning-block.tsx";
+import { StreamingMarkdownText } from "@/features/ai-chat/components/streaming-markdown-text.tsx";
 import ChatErrorAlert from "@/features/ai-chat/components/chat-error-alert.tsx";
 import ChatStoppedNotice from "@/features/ai-chat/components/chat-stopped-notice.tsx";
 import { ToolUiPart, isToolPart } from "@/features/ai-chat/utils/tool-parts.tsx";
@@ -86,17 +87,39 @@ interface MessageItemProps {
  * One assistant text part rendered as sanitized markdown. Memoized on its inputs
  * so a finalized text part is NOT re-parsed on every streamed delta: during a
  * turn only the actively-growing tail part changes its `text`, so every earlier
- * part hits the memo and skips the expensive marked + DOMPurify pass. Props are
- * primitives, so React.memo's default shallow compare is exactly right (the
- * `text` string is compared by value).
+ * part hits the memo and skips the expensive canonical parse + DOMPurify pass.
+ * Props are primitives, so React.memo's default shallow compare is exactly right
+ * (the `text` string is compared by value).
+ *
+ * Streaming gate (#492) — mirrors ReasoningBlock:
+ *  - `streaming` (this is the live, actively-growing tail part of an in-flight
+ *    turn): render incrementally via StreamingMarkdownText — the stabilized blocks
+ *    go through the canonical pipeline (each parsed ONCE, memoized) and only the
+ *    live tail is cheap plain text. This makes the per-tick cost O(new blocks),
+ *    not the pre-#492 O(ticks) whole-answer re-parse on every ~20Hz delta.
+ *  - finalized (the common case, and the turn-end flip): render the WHOLE text
+ *    through ONE canonical pass — byte-identical to the pre-#492 output (visual
+ *    parity). The row re-renders on the streaming→done flip because
+ *    `messageSignature` tracks each part's `state` (and `turnStreaming` flips at
+ *    turn end), so the incremental view always converges to this single render.
  */
 const MarkdownPart = memo(function MarkdownPart({
   text,
   neutralizeInternalLinks,
+  streaming,
 }: {
   text: string;
   neutralizeInternalLinks: boolean;
+  streaming: boolean;
 }) {
+  if (streaming) {
+    return (
+      <StreamingMarkdownText
+        text={text}
+        neutralizeInternalLinks={neutralizeInternalLinks}
+      />
+    );
+  }
   const html = renderChatMarkdown(text, { neutralizeInternalLinks });
   if (html) {
     return (
@@ -179,47 +202,10 @@ function MessageItem({
         {resolveAssistantName(assistantName) ?? t("AI agent")}
       </Text>
       {message.parts.map((part, index) => {
-        if (part.type === "reasoning") {
-          // Reasoning ("thinking") -> a collapsible block with its own token
-          // count. Empty/whitespace reasoning with no authoritative count carries
-          // nothing to show, so skip it (avoids an empty 0-token block).
-          const text = (part as { text?: string }).text ?? "";
-          if (!text.trim() && !(reasoningTokens && reasoningTokens > 0))
-            return null;
-          // Absent state (persisted rows) and "done" both mean finalized.
-          // `messageSignature` already includes each part's `state`, so the
-          // streaming→done flip changes the row signature and re-renders this
-          // row — which is what lets ReasoningBlock switch from chunked plain
-          // text to its one-time markdown parse (see reasoning-block.tsx).
-          // ALSO require the turn to be live: a part stranded at
-          // `state:"streaming"` after the turn ended (no `reasoning-end` — see
-          // the `turnStreaming` prop doc) must still finalize and parse.
-          const streaming =
-            turnStreaming && (part as { state?: string }).state === "streaming";
-          return (
-            <ReasoningBlock
-              key={index}
-              text={text}
-              tokens={reasoningTokens}
-              streaming={streaming}
-            />
-          );
-        }
-
-        if (part.type === "text") {
-          // Skip empty/whitespace-only text parts (a streaming message often
-          // starts with an empty text part before the first token arrives); the
-          // typing indicator covers that gap until real content streams in.
-          if (!part.text.trim()) return null;
-          return (
-            <MarkdownPart
-              key={index}
-              text={part.text}
-              neutralizeInternalLinks={neutralizeInternalLinks}
-            />
-          );
-        }
-
+        // Tool parts (`tool-*` / `dynamic-tool`) are template-literal kinds, so
+        // they cannot be a `switch` case; the runtime guard handles them, and the
+        // switch below covers every CLOSED (literal-typed) part kind with a
+        // compile-time exhaustiveness check in its default.
         if (isToolPart(part.type)) {
           return (
             <ToolCallCard
@@ -232,7 +218,76 @@ function MessageItem({
           );
         }
 
-        return null;
+        switch (part.type) {
+          case "reasoning": {
+            // Reasoning ("thinking") -> a collapsible block with its own token
+            // count. Empty/whitespace reasoning with no authoritative count
+            // carries nothing to show, so skip it (avoids an empty 0-token block).
+            const text = part.text ?? "";
+            if (!text.trim() && !(reasoningTokens && reasoningTokens > 0))
+              return null;
+            // Absent state (persisted rows) and "done" both mean finalized.
+            // `messageSignature` already includes each part's `state`, so the
+            // streaming→done flip changes the row signature and re-renders this
+            // row — which is what lets ReasoningBlock switch from chunked plain
+            // text to its one-time markdown parse (see reasoning-block.tsx).
+            // ALSO require the turn to be live: a part stranded at
+            // `state:"streaming"` after the turn ended (no `reasoning-end` — see
+            // the `turnStreaming` prop doc) must still finalize and parse.
+            const streaming = turnStreaming && part.state === "streaming";
+            return (
+              <ReasoningBlock
+                key={index}
+                text={text}
+                tokens={reasoningTokens}
+                streaming={streaming}
+              />
+            );
+          }
+
+          case "text": {
+            // Skip empty/whitespace-only text parts (a streaming message often
+            // starts with an empty text part before the first token arrives); the
+            // typing indicator covers that gap until real content streams in.
+            if (!part.text.trim()) return null;
+            // The live, actively-growing tail part of the in-flight turn renders
+            // incrementally (see MarkdownPart); a finalized part (persisted, or
+            // the turn-end flip) renders the whole text through one canonical
+            // pass. Same liveness rule as the reasoning branch above.
+            const streaming = turnStreaming && part.state === "streaming";
+            return (
+              <MarkdownPart
+                key={index}
+                text={part.text}
+                neutralizeInternalLinks={neutralizeInternalLinks}
+                streaming={streaming}
+              />
+            );
+          }
+
+          case "source-url":
+          case "source-document":
+          case "file":
+          case "step-start":
+            // Not surfaced in the chat bubble (v1) — same as the pre-#492 default.
+            return null;
+
+          default: {
+            // Compile-time exhaustiveness over the CLOSED union members: every
+            // literal-typed part kind is handled above, so the only kinds that
+            // can reach here are the OPEN template-literal ones (`tool-*` — caught
+            // by the guard at runtime — and `data-*`) plus `dynamic-tool`. Adding
+            // a NEW closed part kind to UIMessagePart makes this assignment fail
+            // to compile, forcing it to be handled instead of silently ignored
+            // (this replaces the pre-#492 fall-through `return null` + WARNING).
+            const _exhaustive:
+              | `tool-${string}`
+              | "dynamic-tool"
+              | `data-${string}` = part.type;
+            void _exhaustive;
+            return null;
+          }
+        }
       })}
       {/* A persisted turn error (server stored it in metadata.error). Rendered
           here so it survives a thread remount and shows in reopened history. */}

apps/client/src/features/ai-chat/components/streaming-markdown-text.tsx

@@ -0,0 +1,96 @@
+import { memo, useMemo } from "react";
+import { splitPlainChunks } from "@/features/ai-chat/components/streaming-plain-text.tsx";
+import { renderChatMarkdown } from "@/features/ai-chat/utils/markdown.ts";
+import classes from "@/features/ai-chat/components/ai-chat.module.css";
+
+/**
+ * One STABILIZED markdown block, rendered through the canonical pipeline and
+ * memoized on its string prop. During streaming only the TAIL chunk grows (the
+ * `splitPlainChunks` append-only invariant guarantees every earlier chunk is
+ * byte-identical across deltas), so React skips every stable block and each one
+ * is parsed by `renderChatMarkdown` EXACTLY ONCE — turning the pre-#492
+ * "re-parse the whole accumulated answer on every ~20Hz tick" (O(ticks)) into
+ * O(number of blocks). The markup is DOMPurify-sanitized inside renderChatMarkdown
+ * before it reaches `dangerouslySetInnerHTML`.
+ *
+ * NOTE (transient streaming-only artifact): a safe cut is a blank-line boundary,
+ * so a construct that legitimately contains a blank line (e.g. a fenced code block
+ * with an empty line) can be split across chunks and render oddly WHILE it is still
+ * streaming. This is cosmetic and self-heals: the moment the part finalizes,
+ * MarkdownPart renders the WHOLE text through one canonical pass (visual parity
+ * with the pre-#492 output). The reasoning path makes the same trade (plain text
+ * while streaming, one markdown parse at the end).
+ */
+const MarkdownChunk = memo(function MarkdownChunk({
+  text,
+  neutralizeInternalLinks,
+}: {
+  text: string;
+  neutralizeInternalLinks: boolean;
+}) {
+  const html = renderChatMarkdown(text, { neutralizeInternalLinks });
+  if (html) {
+    return (
+      <div
+        className={classes.markdown}
+        // Sanitized by renderChatMarkdown (DOMPurify) before insertion.
+        dangerouslySetInnerHTML={{ __html: html }}
+      />
+    );
+  }
+  // Malformed/unsupported markdown could not render synchronously: raw text.
+  return (
+    <div className={classes.markdown} style={{ whiteSpace: "pre-wrap" }}>
+      {text}
+    </div>
+  );
+});
+
+/**
+ * The cheap streaming-time stand-in for the finalized answer's one-time markdown
+ * parse (see MarkdownPart in message-item.tsx). Mirrors StreamingPlainText's
+ * chunked-memo pattern but renders the STABILIZED prefix as real markdown (each
+ * block parsed once, memoized) and only the LIVE tail as flat plain text — so the
+ * user sees formatted output for everything up to the last safe cut, and the not-
+ * yet-stable tail (which markdown-parsing every tick would make O(ticks)) stays a
+ * single cheap escaped text node until it stabilizes into a new block.
+ *
+ * `splitPlainChunks` yields chunks where, under append-only growth, every chunk
+ * except the LAST is immutable; the last chunk is the live tail. Index keys are
+ * therefore stable (a given index never changes to a different chunk's content).
+ */
+export function StreamingMarkdownText({
+  text,
+  neutralizeInternalLinks,
+}: {
+  text: string;
+  neutralizeInternalLinks: boolean;
+}) {
+  const chunks = useMemo(() => splitPlainChunks(text), [text]);
+  return (
+    <>
+      {chunks.map((chunk, index) =>
+        index < chunks.length - 1 ? (
+          <MarkdownChunk
+            key={index}
+            text={chunk}
+            neutralizeInternalLinks={neutralizeInternalLinks}
+          />
+        ) : (
+          // The live tail: flat, React-escaped plain text (no markdown parse, no
+          // sanitizer, no innerHTML). `pre-wrap` preserves its newlines; trailing
+          // separator newlines are dropped at display time so the block gap comes
+          // from the markdown margins, not a doubled empty line (mirrors
+          // PlainChunk in streaming-plain-text.tsx).
+          <div
+            key={index}
+            className={classes.markdown}
+            style={{ whiteSpace: "pre-wrap" }}
+          >
+            {chunk.replace(/\n+$/, "")}
+          </div>
+        ),
+      )}
+    </>
+  );
+}

apps/server/src/core/ai-chat/ai-chat.controller.ts

@@ -35,6 +35,7 @@ import {
 import { PaginationOptions } from '@docmost/db/pagination/pagination-options';
 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 { PageRepo } from '@docmost/db/repos/page/page.repo';
 import { UserThrottlerGuard } from '../../integrations/throttle/user-throttler.guard';
 import { AI_CHAT_THROTTLER } from '../../integrations/throttle/throttler-names';
@@ -43,6 +44,8 @@ import {
   AiChatRunHooks,
   AiChatService,
   AiChatStreamBody,
+  rowHasInlineParts,
+  hydrateAssistantParts,
 } from './ai-chat.service';
 import { AiChatRunService } from './ai-chat-run.service';
 import { AiTranscriptionService } from './ai-transcription.service';
@@ -129,8 +132,39 @@ export class AiChatController {
     // production. Only touched on the resumable-stream (flag-on) path.
     private readonly streamRegistry?: AiChatStreamRegistryService,
     private readonly environment?: EnvironmentService,
+    // #492: reconstruct a #492 mid-run record's parts from the steps table before
+    // returning rows to the client / export. OPTIONAL so positional controller
+    // specs compile unchanged; when absent, hydration is skipped (old-era rows
+    // already carry inline parts, so nothing to reconstruct).
+    private readonly aiChatRunStepRepo?: AiChatRunStepRepo,
   ) {}
 
+  /**
+   * Reconstruct parts for any assistant rows that don't carry them INLINE — a
+   * #492 mid-run record whose per-step parts live in `ai_chat_run_steps` (the
+   * append-persist backend). Every FINISHED row (old-era + #492) and every old-era
+   * streaming snapshot already has inline `metadata.parts`, so the common path
+   * fetches NOTHING and returns the rows untouched; only an actively-streaming
+   * new-style row triggers the batch step fetch. Consumers (seed/poll/export) read
+   * `metadata.parts` off the returned rows exactly as before — the era switch is
+   * invisible to them (reconstructRunParts contract).
+   */
+  private async withReconstructedParts(
+    rows: AiChatMessage[],
+    workspaceId: string,
+  ): Promise<AiChatMessage[]> {
+    if (!this.aiChatRunStepRepo) return rows;
+    const needy = rows.filter(
+      (r) => r.role === 'assistant' && !rowHasInlineParts(r),
+    );
+    if (needy.length === 0) return rows;
+    const stepsByMessage = await this.aiChatRunStepRepo.findByMessageIds(
+      needy.map((r) => r.id),
+      workspaceId,
+    );
+    return hydrateAssistantParts(rows, stepsByMessage);
+  }
+
   /** List the requesting user's chats in this workspace (paginated). */
   @HttpCode(HttpStatus.OK)
   @Post('chats')
@@ -184,11 +218,17 @@ export class AiChatController {
     @AuthWorkspace() workspace: Workspace,
   ) {
     await this.assertOwnedChat(dto.chatId, user, workspace);
-    return this.aiChatMessageRepo.findByChat(
+    const page = await this.aiChatMessageRepo.findByChat(
       dto.chatId,
       workspace.id,
       pagination,
     );
+    // #492: reconstruct parts for any active new-style row so the client seed sees
+    // `metadata.parts` unchanged (a no-op for the finished rows that fill a page).
+    return {
+      ...page,
+      items: await this.withReconstructedParts(page.items, workspace.id),
+    };
   }
 
   /**
@@ -225,7 +265,10 @@ export class AiChatController {
       workspace.id,
     );
     return {
-      rows,
+      // #492: the delta of an actively-streaming new-style row carries its parts
+      // reconstructed from the steps table, so the degraded poll shows persisted
+      // progress exactly as the pre-#492 full-row snapshot did.
+      rows: await this.withReconstructedParts(rows, workspace.id),
       cursor,
       run: run ? { id: run.id, status: run.status } : null,
     };
@@ -247,8 +290,10 @@ export class AiChatController {
     @AuthWorkspace() workspace: Workspace,
   ): Promise<{ markdown: string }> {
     const chat = await this.assertOwnedChat(dto.chatId, user, workspace);
-    const rows = await this.aiChatMessageRepo.findAllByChat(
-      dto.chatId,
+    const rows = await this.withReconstructedParts(
+      await this.aiChatMessageRepo.findAllByChat(dto.chatId, workspace.id),
+      // #492: an interrupted-but-still-active turn exports its persisted steps
+      // (reconstructed from the steps table) just like the pre-#492 full row did.
       workspace.id,
     );
     const markdown = buildChatMarkdown({
@@ -288,7 +333,13 @@ export class AiChatController {
           workspace.id,
         )
       : undefined;
-    return { run, message: message ?? null };
+    // #492: reconnect to an IN-FLIGHT run reconstructs the projection row's parts
+    // from the steps table (the row itself carries only the step marker mid-run);
+    // a finished run's row already has inline parts, so this is a no-op.
+    const [hydrated] = message
+      ? await this.withReconstructedParts([message], workspace.id)
+      : [undefined];
+    return { run, message: hydrated ?? null };
   }
 
   /**

apps/server/src/core/ai-chat/ai-chat.service.ts

@@ -22,6 +22,7 @@ import { AiSettingsService } from '../../integrations/ai/ai-settings.service';
 import { describeProviderError } from '../../integrations/ai/ai-error.util';
 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 { AiChatPageSnapshotRepo } from '@docmost/db/repos/ai-chat/ai-chat-page-snapshot.repo';
 import { AiAgentRoleRepo } from '@docmost/db/repos/ai-agent-roles/ai-agent-roles.repo';
 import { PageRepo } from '@docmost/db/repos/page/page.repo';
@@ -518,6 +519,12 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
     // constructions compile unchanged; Nest always injects the real singleton, so
     // reconcile sees the SAME in-memory active/zombie maps the runner mutates.
     private readonly aiChatRunService?: AiChatRunService,
+    // #492 append-persist: per-step INSERT into the lightweight steps table (the
+    // O(Σ steps) replacement for the O(n²) full-row `metadata.parts` rewrite).
+    // OPTIONAL so existing positional constructions (int-specs) compile unchanged;
+    // Nest injects the real singleton. When ABSENT the per-step path falls back to
+    // the pre-#492 full-row flush (no regression, only no WAL win).
+    private readonly aiChatRunStepRepo?: AiChatRunStepRepo,
   ) {}
 
   // #487: periodic reconcile timer (single-process phase 1). Started in
@@ -1559,17 +1566,57 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
         // connection when finalize runs, so the SQL `WHERE status='streaming'`
         // (not this flag) is what prevents it clobbering the terminal row.
         if (finalized) return null;
-        // Build the flush ONCE so the returned count is EXACTLY the persisted
-        // `stepsPersisted` (both derive from capturedSteps.length at this instant).
-        const flushed = flushAssistant(capturedSteps, '', 'streaming', {
-          pageChanged,
-          partsCache,
-        });
-        const stepsPersisted = flushed.metadata.stepsPersisted as number;
+        // The count derives from capturedSteps.length at THIS instant, so the
+        // returned value is EXACTLY the persisted `stepsPersisted` the ring rotates
+        // on (whether we take the append-persist path or the legacy fallback).
+        const stepsPersisted = capturedSteps.length;
         try {
-          await this.aiChatMessageRepo.update(assistantId, workspace.id, flushed, {
-            onlyIfStreaming: true,
-          });
+          if (this.aiChatRunStepRepo) {
+            // #492 APPEND-PERSIST: write only THIS finished step's parts to the
+            // steps table (O(step) WAL), then bump the row's CHEAP step marker —
+            // NO growing `metadata.parts` blob (that O(n²) full-row rewrite is
+            // exactly what this removes). The full `metadata.parts` is assembled
+            // once at finalize; a mid-run resume seed is reconstructed from the
+            // step rows (reconstructRunParts). The INSERT is idempotent
+            // (ON CONFLICT DO NOTHING), so a re-fired step never doubles the parts.
+            const index = stepsPersisted - 1;
+            if (index >= 0) {
+              const stepParts = assistantParts(
+                [capturedSteps[index]],
+                '',
+                partsCache,
+              );
+              await this.aiChatRunStepRepo.insertStep(
+                assistantId,
+                workspace.id,
+                index,
+                stepParts,
+              );
+            }
+            // Marker UPDATE: advance stepsPersisted + keep the toolTrace era marker
+            // (bumps updatedAt so the delta poll observes the step, and carries the
+            // frontier a resuming client attaches from). Scoped onlyIfStreaming so a
+            // late marker never clobbers the terminal finalize.
+            await this.aiChatMessageRepo.update(
+              assistantId,
+              workspace.id,
+              { metadata: stepMarkerMetadata(stepsPersisted) },
+              { onlyIfStreaming: true },
+            );
+          } else {
+            // Legacy fallback (no steps table wired — positional test builds): the
+            // pre-#492 full-row flush, so parts still land inline on the row.
+            const flushed = flushAssistant(capturedSteps, '', 'streaming', {
+              pageChanged,
+              partsCache,
+            });
+            await this.aiChatMessageRepo.update(
+              assistantId,
+              workspace.id,
+              flushed,
+              { onlyIfStreaming: true },
+            );
+          }
           return stepsPersisted;
         } catch (err) {
           this.logger.warn(
@@ -2749,6 +2796,122 @@ export function rowToUiMessage(row: AiChatMessage): Omit<UIMessage, 'id'> & {
   return { id: row.id, role, parts: parts as UIMessage['parts'] };
 }
 
+/**
+ * Cheap step-marker metadata for the #492 per-step UPDATE. Advances
+ * `stepsPersisted` (the resume attach frontier) and keeps the `toolTraceVersion`
+ * era marker, WITHOUT the growing `parts` blob (those live in the steps table
+ * now; the full `metadata.parts` is assembled once at finalize by flushAssistant).
+ * `parts: []` is kept for shape stability — it reads as an empty inline-parts row,
+ * which is exactly the discriminator that routes reconstruction to the steps table.
+ */
+export function stepMarkerMetadata(
+  stepsPersisted: number,
+): Record<string, unknown> {
+  return { parts: [], toolTraceVersion: 2, stepsPersisted };
+}
+
+/**
+ * Whether an assistant row already carries its full UI parts INLINE on the row
+ * (`metadata.parts`). TRUE for every FINISHED row — old-era rows AND #492 rows,
+ * whose full parts are assembled once at finalize — and for old-era streaming
+ * snapshots (the pre-#492 per-step full-row flush). FALSE for a #492 MID-RUN
+ * record, whose per-step parts live in the `ai_chat_run_steps` table. This is the
+ * era discriminator the reconstruct seam branches on — no schema flag needed.
+ */
+export function rowHasInlineParts(row: { metadata?: unknown }): boolean {
+  const meta = (row.metadata ?? {}) as { parts?: unknown };
+  return Array.isArray(meta.parts) && meta.parts.length > 0;
+}
+
+/**
+ * Concatenate persisted per-step parts (in `stepIndex` order) into the turn's UI
+ * parts (#492). Reproduces EXACTLY what flushAssistant → assistantParts would have
+ * written to `metadata.parts` for those finished steps, since each step row stored
+ * `assistantParts([step])` at persist time.
+ */
+export function assembleStepParts(
+  stepRows: ReadonlyArray<{ stepIndex: number; parts: unknown }>,
+): UIMessage['parts'] {
+  const parts: Array<Record<string, unknown>> = [];
+  for (const step of [...stepRows].sort((a, b) => a.stepIndex - b.stepIndex)) {
+    if (Array.isArray(step.parts)) {
+      parts.push(...(step.parts as Array<Record<string, unknown>>));
+    }
+  }
+  return parts as UIMessage['parts'];
+}
+
+/**
+ * reconstructRunParts (#492) — the single backend-switch seam. Given an assistant
+ * ROW and its persisted step rows, return the turn's UI `parts` + the persisted
+ * step count, reading from the ROW when it already carries inline parts (old-era
+ * records AND every finished record) and from the STEPS TABLE otherwise (a #492
+ * mid-run record). The higher-level consumers (attach seed, delta poll, export)
+ * route their row→parts through this / {@link hydrateAssistantParts}, so old and
+ * new records reconstruct identically WITHOUT the consumers branching on the era.
+ */
+export function reconstructRunParts(
+  row: { metadata?: unknown; content?: string | null },
+  stepRows: ReadonlyArray<{ stepIndex: number; parts: unknown }>,
+): { parts: UIMessage['parts']; stepsPersisted: number } {
+  if (rowHasInlineParts(row)) {
+    const meta = row.metadata as {
+      parts: UIMessage['parts'];
+      stepsPersisted?: number;
+    };
+    return {
+      parts: meta.parts,
+      stepsPersisted:
+        typeof meta.stepsPersisted === 'number'
+          ? meta.stepsPersisted
+          : stepRows.length,
+    };
+  }
+  if (stepRows.length > 0) {
+    return {
+      parts: assembleStepParts(stepRows),
+      stepsPersisted: stepRows.length,
+    };
+  }
+  // No inline parts and no step rows: an old-era seed / empty streaming row. Fall
+  // back to a single text part from `content` (mirrors rowToUiMessage).
+  return {
+    parts: textPart(row.content ?? '') as UIMessage['parts'],
+    stepsPersisted: 0,
+  };
+}
+
+/**
+ * Fill each assistant row's `metadata.parts` from its step rows when the row does
+ * not already carry them inline (a #492 mid-run record), so a consumer that reads
+ * `metadata.parts` off the RAW row (the client seed/poll, the Markdown export)
+ * sees the reconstructed parts with NO change to itself. Rows that already have
+ * inline parts (old-era + finished) and non-assistant rows pass through untouched.
+ * Pure: returns new row objects, never mutates the inputs.
+ */
+export function hydrateAssistantParts<
+  T extends { id: string; role?: string; metadata?: unknown },
+>(
+  rows: ReadonlyArray<T>,
+  stepsByMessage: Map<
+    string,
+    ReadonlyArray<{ stepIndex: number; parts: unknown }>
+  >,
+): T[] {
+  return rows.map((row) => {
+    if (row.role !== 'assistant' || rowHasInlineParts(row)) return row;
+    const steps = stepsByMessage.get(row.id);
+    if (!steps || steps.length === 0) return row;
+    return {
+      ...row,
+      metadata: {
+        ...((row.metadata ?? {}) as Record<string, unknown>),
+        parts: assembleStepParts(steps),
+      },
+    };
+  });
+}
+
 /**
  * The persisted-row patch shape produced by {@link flushAssistant}. It is the
  * SAME shape the assistant repo insert/update consume (content + toolCalls +

apps/server/src/database/database.module.ts

@@ -32,6 +32,7 @@ import { TemplateRepo } from '@docmost/db/repos/template/template.repo';
 import { AiChatRepo } from '@docmost/db/repos/ai-chat/ai-chat.repo';
 import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
 import { AiChatRunRepo } from '@docmost/db/repos/ai-chat/ai-chat-run.repo';
+import { AiChatRunStepRepo } from '@docmost/db/repos/ai-chat/ai-chat-run-step.repo';
 import { AiChatPageSnapshotRepo } from '@docmost/db/repos/ai-chat/ai-chat-page-snapshot.repo';
 import { AiProviderCredentialsRepo } from '@docmost/db/repos/ai-chat/ai-provider-credentials.repo';
 import { AiMcpServerRepo } from '@docmost/db/repos/ai-chat/ai-mcp-server.repo';
@@ -125,6 +126,7 @@ import { firstSqlToken } from '../integrations/metrics/metrics.constants';
     AiChatRepo,
     AiChatMessageRepo,
     AiChatRunRepo,
+    AiChatRunStepRepo,
     AiChatPageSnapshotRepo,
     AiProviderCredentialsRepo,
     AiMcpServerRepo,
@@ -161,6 +163,7 @@ import { firstSqlToken } from '../integrations/metrics/metrics.constants';
     AiChatRepo,
     AiChatMessageRepo,
     AiChatRunRepo,
+    AiChatRunStepRepo,
     AiChatPageSnapshotRepo,
     AiProviderCredentialsRepo,
     AiMcpServerRepo,

apps/server/src/database/migrations/20260708T120000-ai-chat-run-steps.ts

@@ -0,0 +1,70 @@
+import { type Kysely, sql } from 'kysely';
+
+/**
+ * `ai_chat_run_steps` — append-only per-step persistence for an assistant turn
+ * (#492 wave C). Each finished agent step's UI `parts` (its text part + a part
+ * per tool call, WITH the tool output) is INSERTed as its own lightweight row the
+ * moment the step ends, instead of REWRITING the whole assistant row's growing
+ * `metadata.parts` jsonb on every `onStepFinish`.
+ *
+ * WHY a separate table + INSERT (not a jsonb `||` append on the message row): a
+ * Postgres jsonb UPDATE rewrites the ENTIRE TOASTed row version under MVCC, so
+ * re-persisting a growing `metadata.parts` on every step is O(n²) write volume
+ * (a 50-step run with ~100 KB tool outputs wrote hundreds of MB of WAL / dead
+ * tuples per turn, hammering autovacuum). `||` would only shave the network
+ * payload — the WAL/TOAST rewrite harm remains. An INSERT into a per-step table
+ * writes ONLY that step's bytes, so the per-turn write volume is O(Σ steps).
+ *
+ * The full `metadata.parts` on the message row is assembled ONCE at finalize (the
+ * terminal completed/error/aborted write). Mid-run, a resuming client's seed is
+ * reconstructed by concatenating these step rows in `step_index` order — which
+ * reproduces exactly what the old per-step full-row rewrite persisted. Records
+ * written the OLD way (full `metadata.parts` on the row, no step rows) still
+ * reconstruct from the row unchanged; the two eras are distinguished by whether
+ * the row already carries non-empty `metadata.parts` (see reconstructRunParts /
+ * assembleStepParts in ai-chat.service.ts).
+ *
+ * ON DELETE CASCADE on `message_id`: the step rows are a derived projection of the
+ * assistant message; they must vanish with it (or with its workspace).
+ */
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .createTable('ai_chat_run_steps')
+    .ifNotExists()
+    .addColumn('id', 'uuid', (col) =>
+      col.primaryKey().defaultTo(sql`gen_uuid_v7()`),
+    )
+    // The assistant message row this step belongs to (the #183 projection). The
+    // step rows are a derived, per-step slice of that message, so they cascade.
+    .addColumn('message_id', 'uuid', (col) =>
+      col.references('ai_chat_messages.id').onDelete('cascade').notNull(),
+    )
+    .addColumn('workspace_id', 'uuid', (col) =>
+      col.references('workspaces.id').onDelete('cascade').notNull(),
+    )
+    // 0-based index of the finished step within the turn. Ordering key for
+    // reconstruction; unique per message (idempotent step re-persist).
+    .addColumn('step_index', 'integer', (col) => col.notNull())
+    // The step's UI parts (text part + a `tool-*` part per call, WITH output).
+    // Concatenated in step order to rebuild the turn's `metadata.parts`.
+    .addColumn('parts', 'jsonb', (col) => col.notNull())
+    .addColumn('created_at', 'timestamptz', (col) =>
+      col.notNull().defaultTo(sql`now()`),
+    )
+    .execute();
+
+  // Idempotent per-step persist: a retried INSERT of the same (message, step)
+  // is a no-op (the service uses ON CONFLICT DO NOTHING). This also serves the
+  // reconstruction read (WHERE message_id ORDER BY step_index).
+  await db.schema
+    .createIndex('ai_chat_run_steps_message_step_uidx')
+    .ifNotExists()
+    .on('ai_chat_run_steps')
+    .columns(['message_id', 'step_index'])
+    .unique()
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.dropTable('ai_chat_run_steps').ifExists().execute();
+}

apps/server/src/database/repos/ai-chat/ai-chat-run-step.repo.ts

@@ -0,0 +1,111 @@
+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';
+
+/**
+ * Append-only per-step persistence for an assistant turn (#492). Each finished
+ * agent step's UI `parts` (its text part + a `tool-*` part per call, WITH the
+ * tool output) is INSERTed as its own lightweight row the moment the step ends —
+ * instead of REWRITING the assistant row's growing `metadata.parts` jsonb on every
+ * `onStepFinish` (a Postgres jsonb UPDATE rewrites the whole TOASTed row version
+ * under MVCC, so that was O(n²) WAL/dead-tuple churn per turn).
+ *
+ * The full `metadata.parts` on the message row is assembled ONCE at finalize;
+ * mid-run, a resuming client's seed is rebuilt from these rows in `stepIndex`
+ * order (see `assembleStepParts` / the reconstruct seam in ai-chat.service.ts).
+ * Every method is workspace-scoped as defense-in-depth.
+ */
+@Injectable()
+export class AiChatRunStepRepo {
+  constructor(@InjectKysely() private readonly db: KyselyDB) {}
+
+  /**
+   * Append one finished step's parts. Idempotent: a retried persist of the SAME
+   * (message, stepIndex) is a no-op via ON CONFLICT DO NOTHING — the per-step
+   * writes are fired fire-and-forget + serialized, and a duplicate must never
+   * throw into the stream or double the parts. Returns whether a NEW row landed
+   * (false = the step was already persisted).
+   */
+  async insertStep(
+    messageId: string,
+    workspaceId: string,
+    stepIndex: number,
+    parts: unknown,
+    trx?: KyselyTransaction,
+  ): Promise<boolean> {
+    const db = dbOrTx(this.db, trx);
+    const inserted = await db
+      .insertInto('aiChatRunSteps')
+      .values({
+        messageId,
+        workspaceId,
+        stepIndex,
+        // jsonb column: cast through never (same pattern as the message repo).
+        parts: parts as never,
+      })
+      .onConflict((oc) => oc.columns(['messageId', 'stepIndex']).doNothing())
+      .returning('id')
+      .executeTakeFirst();
+    return inserted !== undefined;
+  }
+
+  /** All persisted steps for ONE assistant message, in step order. */
+  async findByMessage(
+    messageId: string,
+    workspaceId: string,
+  ): Promise<AiChatRunStep[]> {
+    return this.db
+      .selectFrom('aiChatRunSteps')
+      .selectAll('aiChatRunSteps')
+      .where('messageId', '=', messageId)
+      .where('workspaceId', '=', workspaceId)
+      .orderBy('stepIndex', 'asc')
+      .execute();
+  }
+
+  /**
+   * All persisted steps for a SET of assistant messages, grouped by messageId
+   * (each group in step order). One query for the batch — the hydration seam
+   * (getMessages / delta / export) calls this only for the rows that actually
+   * need reconstruction (an active new-style row whose `metadata.parts` is still
+   * empty), which is usually none, so this is skipped on the common path.
+   */
+  async findByMessageIds(
+    messageIds: string[],
+    workspaceId: string,
+  ): Promise<Map<string, AiChatRunStep[]>> {
+    const byMessage = new Map<string, AiChatRunStep[]>();
+    if (messageIds.length === 0) return byMessage;
+    const rows = await this.db
+      .selectFrom('aiChatRunSteps')
+      .selectAll('aiChatRunSteps')
+      .where('messageId', 'in', messageIds)
+      .where('workspaceId', '=', workspaceId)
+      .orderBy('stepIndex', 'asc')
+      .execute();
+    for (const row of rows) {
+      const list = byMessage.get(row.messageId);
+      if (list) list.push(row);
+      else byMessage.set(row.messageId, [row]);
+    }
+    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;
+  }
+}

apps/server/src/database/types/db.d.ts

@@ -692,6 +692,22 @@ export interface AiChatRuns {
   updatedAt: Generated<Timestamp>;
 }
 
+// Append-only per-step persistence for an assistant turn (#492). Mirrors
+// migration 20260708T120000-ai-chat-run-steps.ts. Each finished agent step's UI
+// `parts` are INSERTed as their own row (instead of rewriting the message row's
+// growing `metadata.parts` jsonb every step — an O(n²) WAL/TOAST churn). The full
+// `metadata.parts` is assembled once at finalize; mid-run a resuming client's seed
+// is rebuilt by concatenating these rows in `stepIndex` order. Cascades with the
+// assistant message row it projects.
+export interface AiChatRunSteps {
+  id: Generated<string>;
+  messageId: string;
+  workspaceId: string;
+  stepIndex: number;
+  parts: Json;
+  createdAt: Generated<Timestamp>;
+}
+
 // Per-(chat,page) snapshot of the open page's Markdown at the END of the agent's
 // previous turn (#274). Mirrors migration 20260702T120000-ai-chat-page-snapshot.ts.
 // The next turn diffs the CURRENT Markdown against `contentMd` to surface edits a
@@ -729,6 +745,7 @@ export interface DB {
   aiChats: AiChats;
   aiChatMessages: AiChatMessages;
   aiChatRuns: AiChatRuns;
+  aiChatRunSteps: AiChatRunSteps;
   aiChatPageSnapshots: AiChatPageSnapshots;
   apiKeys: ApiKeys;
   attachments: Attachments;

apps/server/src/database/types/entity.types.ts

@@ -4,6 +4,7 @@ import {
   AiChats,
   AiChatMessages,
   AiChatRuns,
+  AiChatRunSteps,
   AiChatPageSnapshots,
   Attachments,
   Comments,
@@ -64,6 +65,12 @@ export type InsertableAiChatMessage = Omit<Insertable<AiChatMessages>, 'tsv'>;
 export type AiChatRun = Selectable<AiChatRuns>;
 export type InsertableAiChatRun = Insertable<AiChatRuns>;
 
+// AI Chat Run Step (#492): append-only per-step parts persistence. Each finished
+// agent step's UI parts are stored as their own row; the full turn's parts are
+// assembled from these (in stepIndex order) for a mid-run resume seed.
+export type AiChatRunStep = Selectable<AiChatRunSteps>;
+export type InsertableAiChatRunStep = Insertable<AiChatRunSteps>;
+
 // AI Chat Page Snapshot (#274): per-(chat,page) Markdown snapshot taken at the
 // end of the agent's previous turn, diffed against the current page next turn to
 // detect human edits made between turns.

apps/server/test/integration/ai-chat-append-persist-wal.int-spec.ts

@@ -0,0 +1,173 @@
+import { randomBytes } from 'crypto';
+import { Kysely, sql } from 'kysely';
+import { AiChatRunStepRepo } from '@docmost/db/repos/ai-chat/ai-chat-run-step.repo';
+import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
+import {
+  assistantParts,
+  flushAssistant,
+  stepMarkerMetadata,
+} from '../../src/core/ai-chat/ai-chat.service';
+import {
+  getTestDb,
+  destroyTestDb,
+  createWorkspace,
+  createUser,
+  createChat,
+} from './db';
+
+/**
+ * #492 append-persist — WRITE-VOLUME regression on a LIVE Postgres, measured via
+ * the `pg_current_wal_lsn()` delta around a realistic multi-step run driven through
+ * the REAL repos (not a mock — a mock cannot observe MVCC/TOAST rewrite volume, the
+ * whole point). Proves the core claim:
+ *
+ *   NEW (per-step INSERT into ai_chat_run_steps + a CHEAP step-marker UPDATE on the
+ *   message row) writes O(Σ steps) of WAL — each step writes only its own bytes.
+ *
+ *   OLD (the pre-#492 full-row rewrite: re-persist the GROWING metadata.parts on
+ *   every onStepFinish) writes O(n²) — step k rewrites the whole TOASTed jsonb of
+ *   all k prior outputs.
+ *
+ * The OLD path here IS the reverted behavior, so this doubles as the mutation
+ * check: swapping the new path back to `flushAssistant` full-row UPDATEs reddens
+ * the assertion (OLD is many times larger).
+ */
+type Step = {
+  text: string;
+  toolCalls: Array<{ toolCallId: string; toolName: string; input: unknown }>;
+  toolResults: Array<{ toolCallId: string; toolName: string; output: unknown }>;
+};
+
+// ~100 KB INCOMPRESSIBLE output per step (a page read). Random base64 so TOAST
+// cannot compress it away and hide the real write volume.
+function makeStep(i: number, outputBytes = 100_000): Step {
+  const body = randomBytes(Math.ceil(outputBytes * 0.75)).toString('base64');
+  return {
+    text: `step ${i} reasoning`,
+    toolCalls: [
+      { toolCallId: `c${i}`, toolName: 'getPage', input: { id: `p${i}` } },
+    ],
+    toolResults: [
+      {
+        toolCallId: `c${i}`,
+        toolName: 'getPage',
+        output: { id: `p${i}`, title: `Page ${i}`, body },
+      },
+    ],
+  };
+}
+
+async function walDelta(
+  db: Kysely<any>,
+  fn: () => Promise<void>,
+): Promise<number> {
+  const before = (
+    await sql<{ l: string }>`select pg_current_wal_lsn() as l`.execute(db)
+  ).rows[0].l;
+  await fn();
+  // NOTE: no pg_switch_wal() — a segment switch pads the LSN to the next 16 MB
+  // boundary and would swamp the delta. The raw LSN advances by the WAL bytes.
+  const after = (
+    await sql<{ l: string }>`select pg_current_wal_lsn() as l`.execute(db)
+  ).rows[0].l;
+  return Number(
+    (
+      await sql<{
+        d: string;
+      }>`select pg_wal_lsn_diff(${after}::pg_lsn, ${before}::pg_lsn) as d`.execute(
+        db,
+      )
+    ).rows[0].d,
+  );
+}
+
+describe('#492 append-persist write volume (pg_current_wal_lsn delta) [integration]', () => {
+  let db: Kysely<any>;
+  let stepRepo: AiChatRunStepRepo;
+  let msgRepo: AiChatMessageRepo;
+  let workspaceId: string;
+  let userId: string;
+  let chatId: string;
+
+  beforeAll(async () => {
+    db = getTestDb();
+    stepRepo = new AiChatRunStepRepo(db as any);
+    msgRepo = new AiChatMessageRepo(db as any);
+    workspaceId = (await createWorkspace(db)).id;
+    userId = (await createUser(db, workspaceId)).id;
+    chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+  });
+
+  afterAll(async () => {
+    await destroyTestDb();
+  });
+
+  const seedRow = () =>
+    msgRepo.insert({
+      chatId,
+      workspaceId,
+      userId,
+      role: 'assistant',
+      content: '',
+      status: 'streaming',
+      metadata: stepMarkerMetadata(0) as never,
+    });
+
+  const STEPS = 40;
+
+  it('NEW per-step INSERT is O(Σ steps); OLD full-row rewrite is O(n²)', async () => {
+    const steps: Step[] = [];
+    for (let i = 0; i < STEPS; i++) steps.push(makeStep(i));
+
+    // NEW: per-step INSERT of THIS step's parts + a cheap marker UPDATE.
+    const newRow = await seedRow();
+    const newWal = await walDelta(db, async () => {
+      for (let i = 0; i < STEPS; i++) {
+        await stepRepo.insertStep(
+          newRow.id,
+          workspaceId,
+          i,
+          assistantParts([steps[i]], ''),
+        );
+        await msgRepo.update(
+          newRow.id,
+          workspaceId,
+          { metadata: stepMarkerMetadata(i + 1) },
+          { onlyIfStreaming: true },
+        );
+      }
+    });
+
+    // OLD (the pre-#492 revert): re-persist the GROWING metadata.parts on the
+    // message row on every step.
+    const oldRow = await seedRow();
+    const oldWal = await walDelta(db, async () => {
+      const acc: Step[] = [];
+      for (let i = 0; i < STEPS; i++) {
+        acc.push(steps[i]);
+        await msgRepo.update(
+          oldRow.id,
+          workspaceId,
+          flushAssistant(acc as never, '', 'streaming'),
+          { onlyIfStreaming: true },
+        );
+      }
+    });
+
+    // eslint-disable-next-line no-console
+    console.log(
+      `[#492 WAL] ${STEPS} steps ×100KB: new=${(newWal / 1e6).toFixed(1)}MB ` +
+        `old=${(oldWal / 1e6).toFixed(1)}MB (${(oldWal / newWal).toFixed(
+          1,
+        )}x smaller)`,
+    );
+
+    // O(Σ steps): ~STEPS × (100KB output + marker) of WAL. 40 × ~100KB parts plus
+    // 40 tiny markers is a few tens of MB at most — bounded, linear in step count.
+    expect(newWal).toBeLessThan(30_000_000);
+    // O(n²): step k rewrites ~k × 100KB. Σ over 40 steps ≈ 80+ MB — far larger.
+    expect(oldWal).toBeGreaterThan(30_000_000);
+    // The load-bearing claim: the new path writes a small FRACTION of the old.
+    expect(newWal).toBeLessThan(oldWal * 0.35);
+  }, 120_000);
+});

apps/server/test/integration/ai-chat-run-step-repo.int-spec.ts

@@ -0,0 +1,164 @@
+import { randomBytes } from 'crypto';
+import { Kysely } from 'kysely';
+import { AiChatRunStepRepo } from '@docmost/db/repos/ai-chat/ai-chat-run-step.repo';
+import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
+import {
+  assistantParts,
+  reconstructRunParts,
+  hydrateAssistantParts,
+  stepMarkerMetadata,
+  rowHasInlineParts,
+} from '../../src/core/ai-chat/ai-chat.service';
+import {
+  getTestDb,
+  destroyTestDb,
+  createWorkspace,
+  createUser,
+  createChat,
+} from './db';
+
+/**
+ * #492 append-persist — the reconstruct CONTRACT on a live Postgres. Proves that a
+ * turn persisted the NEW way (per-step rows in `ai_chat_run_steps`, only a step
+ * marker on the message row) reconstructs to the SAME UI parts as a turn persisted
+ * the OLD way (full `metadata.parts` inline on the row, no step rows) — so the
+ * era-switch is invisible to attach / delta-poll / export. Real repos + real jsonb
+ * roundtrip, not a mock (a mock cannot prove the parts survive the jsonb column
+ * byte-identical).
+ */
+type Step = {
+  text: string;
+  toolCalls: Array<{ toolCallId: string; toolName: string; input: unknown }>;
+  toolResults: Array<{ toolCallId: string; toolName: string; output: unknown }>;
+};
+
+// A realistic step: some text + a getPage tool call whose ~100 KB body is
+// INCOMPRESSIBLE random base64 (a 'x'.repeat filler would TOAST away and hide the
+// real bytes). Under MAX_TOOL_OUTPUT_BYTES (200 KB) it is stored uncompacted.
+function makeStep(i: number, outputBytes = 4_000): Step {
+  const body = randomBytes(Math.ceil(outputBytes * 0.75)).toString('base64');
+  return {
+    text: `step ${i} text`,
+    toolCalls: [
+      { toolCallId: `c${i}`, toolName: 'getPage', input: { id: `p${i}` } },
+    ],
+    toolResults: [
+      {
+        toolCallId: `c${i}`,
+        toolName: 'getPage',
+        output: { id: `p${i}`, title: `Page ${i}`, body },
+      },
+    ],
+  };
+}
+
+describe('AiChatRunStepRepo + reconstruct contract [integration]', () => {
+  let db: Kysely<any>;
+  let stepRepo: AiChatRunStepRepo;
+  let msgRepo: AiChatMessageRepo;
+  let workspaceId: string;
+  let userId: string;
+  let chatId: string;
+
+  beforeAll(async () => {
+    db = getTestDb();
+    stepRepo = new AiChatRunStepRepo(db as any);
+    msgRepo = new AiChatMessageRepo(db as any);
+    workspaceId = (await createWorkspace(db)).id;
+    userId = (await createUser(db, workspaceId)).id;
+    chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+  });
+
+  afterAll(async () => {
+    await destroyTestDb();
+  });
+
+  const seedRow = (metadata: unknown, status: string) =>
+    msgRepo.insert({
+      chatId,
+      workspaceId,
+      userId,
+      role: 'assistant',
+      content: '',
+      status,
+      metadata: metadata as never,
+    });
+
+  it('insertStep is idempotent per (message, stepIndex) and reads back in order', async () => {
+    const row = await seedRow(stepMarkerMetadata(0), 'streaming');
+    const parts0 = assistantParts([makeStep(0)], '');
+    const parts1 = assistantParts([makeStep(1)], '');
+
+    expect(await stepRepo.insertStep(row.id, workspaceId, 0, parts0)).toBe(true);
+    expect(await stepRepo.insertStep(row.id, workspaceId, 1, parts1)).toBe(true);
+    // A retried persist of the SAME step is a no-op (ON CONFLICT DO NOTHING).
+    expect(await stepRepo.insertStep(row.id, workspaceId, 0, parts0)).toBe(
+      false,
+    );
+
+    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);
+    expect(map.get(row.id)!.map((s) => s.stepIndex)).toEqual([0, 1]);
+  });
+
+  it('a NEW-style (step-table) run reconstructs identically to an OLD-style (inline) run', async () => {
+    const steps = [makeStep(10), makeStep(11)];
+    // The inline parts the OLD full-row flush would have written.
+    const fullParts = assistantParts(steps, '');
+
+    // OLD-style record: full parts inline on the row, NO step rows.
+    const oldRow = await seedRow(
+      { parts: fullParts, toolTraceVersion: 2, stepsPersisted: 2 },
+      'completed',
+    );
+
+    // NEW-style record: only a step marker on the row + per-step rows.
+    const newRow = await seedRow(stepMarkerMetadata(2), 'streaming');
+    for (let i = 0; i < steps.length; i++) {
+      await stepRepo.insertStep(
+        newRow.id,
+        workspaceId,
+        i,
+        assistantParts([steps[i]], ''),
+      );
+    }
+
+    // Re-read both from the DB (proves the jsonb roundtrip).
+    const oldFetched = await msgRepo.findById(oldRow.id, workspaceId);
+    const newFetched = await msgRepo.findById(newRow.id, workspaceId);
+    const oldSteps = await stepRepo.findByMessage(oldRow.id, workspaceId);
+    const newSteps = await stepRepo.findByMessage(newRow.id, workspaceId);
+
+    // The discriminator: the old row carries inline parts, the new one does not.
+    expect(rowHasInlineParts(oldFetched!)).toBe(true);
+    expect(rowHasInlineParts(newFetched!)).toBe(false);
+    expect(oldSteps).toHaveLength(0);
+    expect(newSteps).toHaveLength(2);
+
+    const oldRecon = reconstructRunParts(oldFetched!, oldSteps);
+    const newRecon = reconstructRunParts(newFetched!, newSteps);
+
+    // Both reconstruct to the SAME parts + step count — the era is invisible.
+    expect(newRecon.parts).toEqual(fullParts);
+    expect(oldRecon.parts).toEqual(fullParts);
+    expect(newRecon.parts).toEqual(oldRecon.parts);
+    expect(newRecon.stepsPersisted).toBe(2);
+    expect(oldRecon.stepsPersisted).toBe(2);
+
+    // hydrateAssistantParts fills the new row's metadata.parts to match the old
+    // row's inline parts — so a consumer reading `metadata.parts` off the raw row
+    // (the client seed/poll, export) is unchanged across the era.
+    const map = await stepRepo.findByMessageIds([newRow.id], workspaceId);
+    const [hydrated] = hydrateAssistantParts([newFetched!], map);
+    expect((hydrated.metadata as { parts: unknown }).parts).toEqual(fullParts);
+    // A row that already has inline parts passes through untouched (same ref-shape).
+    const [oldPassThrough] = hydrateAssistantParts([oldFetched!], map);
+    expect((oldPassThrough.metadata as { parts: unknown }).parts).toEqual(
+      fullParts,
+    );
+  });
+});

apps/server/test/integration/ai-chat-run-steps-migration.int-spec.ts

@@ -0,0 +1,63 @@
+import { Kysely, sql } from 'kysely';
+import {
+  up,
+  down,
+} from '../../src/database/migrations/20260708T120000-ai-chat-run-steps';
+import { getTestDb, destroyTestDb } from './db';
+
+/**
+ * #492 migration up/down roundtrip on a LIVE Postgres. global-setup already
+ * migrated docmost_test to latest (so the table exists at start); this drives the
+ * migration's own down()/up() and asserts the table presence toggles, then leaves
+ * it PRESENT (up) so the shared test DB is intact for any spec that runs after.
+ */
+async function tableExists(db: Kysely<any>): Promise<boolean> {
+  const row = (
+    await sql<{ t: string | null }>`select to_regclass('ai_chat_run_steps') as t`.execute(
+      db,
+    )
+  ).rows[0];
+  return row.t !== null;
+}
+
+async function uniqueIndexExists(db: Kysely<any>): Promise<boolean> {
+  const row = (
+    await sql<{
+      t: string | null;
+    }>`select to_regclass('ai_chat_run_steps_message_step_uidx') as t`.execute(db)
+  ).rows[0];
+  return row.t !== null;
+}
+
+describe('20260708 ai_chat_run_steps migration roundtrip [integration]', () => {
+  let db: Kysely<any>;
+
+  beforeAll(() => {
+    db = getTestDb();
+  });
+
+  afterAll(async () => {
+    // Belt-and-suspenders: guarantee the table is present for later specs even if
+    // an assertion threw mid-roundtrip.
+    if (!(await tableExists(db))) await up(db);
+    await destroyTestDb();
+  });
+
+  it('down() drops the table+index and up() recreates them (idempotent)', async () => {
+    // Starts applied (global-setup migrated to latest).
+    expect(await tableExists(db)).toBe(true);
+    expect(await uniqueIndexExists(db)).toBe(true);
+
+    await down(db);
+    expect(await tableExists(db)).toBe(false);
+    expect(await uniqueIndexExists(db)).toBe(false);
+
+    await up(db);
+    expect(await tableExists(db)).toBe(true);
+    expect(await uniqueIndexExists(db)).toBe(true);
+
+    // up() is idempotent (ifNotExists) — a second run is a harmless no-op.
+    await expect(up(db)).resolves.not.toThrow();
+    expect(await tableExists(db)).toBe(true);
+  });
+});

packages/prosemirror-markdown/src/lib/node-ops.ts

@@ -633,130 +633,6 @@ function findAnchorChain(
   return null;
 }
 
-// ===========================================================================
-// List seam coalescing (#535)
-//
-// When a markdown/JSON insert places a list block directly next to an existing
-// sibling list of the SAME node type, the two must become ONE list (items
-// appended/prepended) rather than two adjacent lists — otherwise the serializer
-// correctly emits a `<!-- -->` separator between them (that separator is right
-// for two genuinely-separate lists; the bug is that the extra sibling was ever
-// created). Coalescing is STRICTLY LOCAL to the two seams of the active
-// insertion — never a global "collapse all adjacent lists" normalization, which
-// would destroy intentionally-separate lists elsewhere.
-// ===========================================================================
-
-/**
- * The three list container types we structurally coalesce at an insertion seam.
- * Deliberately an explicit set, NOT a `type.endsWith("List")` test — that also
- * matches `footnotesList`, which must NEVER be structurally merged.
- */
-function isCoalescibleList(n: any): boolean {
-  return (
-    isObject(n) &&
-    (n.type === "bulletList" ||
-      n.type === "orderedList" ||
-      n.type === "taskList")
-  );
-}
-
-/**
- * True when two adjacent list nodes may have their ITEMS merged into one list.
- * Requires identical node type and, for orderedList, a compatible numbering
- * style: a missing/null `attrs.type` counts as the default, and the merge is
- * blocked ONLY when both lists carry an explicit, differing `attrs.type`.
- */
-function listsMergeable(a: any, b: any): boolean {
-  if (!isCoalescibleList(a) || !isCoalescibleList(b)) return false;
-  if (a.type !== b.type) return false;
-  if (!Array.isArray(a.content) || !Array.isArray(b.content)) return false;
-  if (a.type === "orderedList") {
-    const ta = a.attrs?.type ?? null;
-    const tb = b.attrs?.type ?? null;
-    if (ta != null && tb != null && ta !== tb) return false;
-  }
-  return true;
-}
-
-/**
- * Coalesce list seams around a freshly inserted run occupying indices `[i, j)`
- * in `parent`. At MOST the left seam (`parent[i-1]` ↔ `parent[i]`) and the right
- * seam (`parent[j-1]` ↔ `parent[j]`) are merged, each AT MOST ONCE — never a
- * `while (neighbours same type) merge` loop, which would swallow a further-out
- * intentionally-separate list. Mutates `parent` in place.
- *
- * Survivor choice is POSITIONAL, never by id: the neighbour OUTSIDE the `[i, j)`
- * range is pre-existing (the survivor, keeping its block id and list-level
- * attrs); the boundary block INSIDE the range is the freshly inserted list,
- * whose items move into the survivor and whose wrapper is then deleted
- * (appended when the survivor is on the left, prepended when on the right).
- *
- * Empty inserted list: if the inserted boundary list has zero items, its seam is
- * NOT coalesced — the block is left exactly as inserted.
- */
-function coalesceSeams(parent: any[], i: number, j: number): void {
-  if (!Array.isArray(parent)) return;
-  const n = parent.length;
-
-  const left = parent[i - 1];
-  const boundaryLeft = parent[i];
-  const boundaryRight = parent[j - 1];
-  const right = parent[j];
-
-  // A seam fires only when the pre-existing neighbour and the inserted boundary
-  // list are mergeable AND the inserted boundary list is non-empty.
-  const singleBlock = i === j - 1;
-  const leftMergeable =
-    i - 1 >= 0 &&
-    listsMergeable(left, boundaryLeft) &&
-    boundaryLeft.content.length > 0;
-  let rightMergeable =
-    j < n &&
-    listsMergeable(boundaryRight, right) &&
-    boundaryRight.content.length > 0;
-
-  // Three-way collision: a SINGLE inserted list (singleBlock) landed exactly
-  // between two pre-existing lists. The LEFT pre-existing list wins: the
-  // inserted items then the right list's items fold into it, and both the
-  // inserted wrapper and the right pre-existing list are deleted (the right
-  // block id is NOT preserved — rare, documented).
-  //
-  // The `listsMergeable(left, right)` guard is REQUIRED: leftMergeable and
-  // rightMergeable only check each PRE-EXISTING list against the inserted one.
-  // A default-typed inserted orderedList is compatible with BOTH neighbours
-  // even when the neighbours carry explicit DIFFERENT numbering styles, so
-  // without this guard the two would collapse transitively through the middle
-  // and the right list's style would be silently lost. When it fails we fall
-  // through to the single-seam path below (never a transitive merge).
-  if (singleBlock && leftMergeable && rightMergeable && listsMergeable(left, right)) {
-    left.content.push(...boundaryLeft.content, ...right.content);
-    parent.splice(i, 2);
-    return;
-  }
-
-  // A single inserted block can be absorbed by at most ONE neighbour. When both
-  // seams are individually valid but the neighbours are mutually incompatible
-  // (the three-way guard above failed), prefer the LEFT seam — consistent with
-  // the three-way survivor choice — and drop the right so the incompatible
-  // right list stays separate with its own style.
-  if (singleBlock && leftMergeable && rightMergeable) {
-    rightMergeable = false;
-  }
-
-  // Otherwise the two seams are independent. Process the RIGHT seam FIRST so its
-  // deletion at the higher indices cannot shift the left seam's [i-1, i].
-  if (rightMergeable) {
-    // Survivor is the pre-existing right neighbour; PREPEND the inserted items.
-    right.content.unshift(...boundaryRight.content);
-    parent.splice(j - 1, 1);
-  }
-  if (leftMergeable) {
-    // Survivor is the pre-existing left neighbour; APPEND the inserted items.
-    left.content.push(...boundaryLeft.content);
-    parent.splice(i, 1);
-  }
-}
-
 /** Options controlling where `insertNodeRelative` places the new node. */
 export interface InsertOptions {
   position: "before" | "after" | "append";
@@ -809,10 +685,7 @@ export function insertNodeRelative(
     }
     if (isObject(out)) {
       if (!Array.isArray(out.content)) out.content = [];
-      const at = out.content.length;
       out.content.push(fresh);
-      // Coalesce the left seam with the prior tail block (#535).
-      coalesceSeams(out.content, at, out.content.length);
       return { doc: out, inserted: true };
     }
     return { doc: out, inserted: false };
@@ -864,19 +737,40 @@ export function insertNodeRelative(
     return { doc: out, inserted: true };
   }
 
-  // before/after (non-structural): resolve the anchor's ancestor chain and
-  // splice into the anchor's IMMEDIATE parent, so seam coalescing runs against
-  // the ACTUAL parent array (also correctly handling a list nested in a callout
-  // / table cell). The first-match / top-level-only semantics of anchorNodeId /
-  // anchorText are preserved by findAnchorChain (identical to the old walk).
-  const chain = findAnchorChain(out, opts);
-  if (chain == null || chain.length < 2) return { doc: out, inserted: false };
-  const parent = chain[chain.length - 2].node.content;
-  if (!Array.isArray(parent)) return { doc: out, inserted: false };
-  const at = chain[chain.length - 1].index + offset;
-  parent.splice(at, 0, fresh);
-  coalesceSeams(parent, at, at + 1);
-  return { doc: out, inserted: true };
+  // Resolve by id anywhere in the tree: splice into the parent content array.
+  if (opts.anchorNodeId != null) {
+    let inserted = false;
+    const walkContent = (content: any[]): void => {
+      for (let i = 0; i < content.length; i++) {
+        const child = content[i];
+        if (matchesId(child, opts.anchorNodeId as string)) {
+          content.splice(i + offset, 0, fresh);
+          inserted = true;
+          return;
+        }
+        if (isObject(child) && Array.isArray(child.content)) {
+          walkContent(child.content);
+          if (inserted) return;
+        }
+      }
+    };
+    if (isObject(out) && Array.isArray(out.content)) {
+      walkContent(out.content);
+    }
+    return { doc: out, inserted };
+  }
+
+  // Resolve by text: only top-level doc.content blocks are scanned. Exact
+  // match wins; a markdown-stripped fallback is tried only on a miss.
+  if (opts.anchorText != null && isObject(out) && Array.isArray(out.content)) {
+    const i = findAnchorTextIndex(out.content, opts.anchorText);
+    if (i !== -1) {
+      out.content.splice(i + offset, 0, fresh);
+      return { doc: out, inserted: true };
+    }
+  }
+
+  return { doc: out, inserted: false };
 }
 
 /**
@@ -917,11 +811,7 @@ export function insertNodesRelative(
   if (opts.position === "append") {
     if (isObject(out)) {
       if (!Array.isArray(out.content)) out.content = [];
-      const at = out.content.length;
       out.content.push(...fresh);
-      // Coalesce the left seam with the prior tail block; the run's right side
-      // has no neighbour at the top level (#535).
-      coalesceSeams(out.content, at, out.content.length);
       return { doc: out, inserted: true };
     }
     return { doc: out, inserted: false };
@@ -929,19 +819,40 @@ export function insertNodesRelative(
 
   const offset = opts.position === "after" ? 1 : 0;
 
-  // before/after: resolve the anchor's ancestor chain and splice the whole run
-  // into the anchor's IMMEDIATE parent, then coalesce ONLY the run's two
-  // boundary seams (inner blocks are untouched). Converting to findAnchorChain
-  // makes coalescing run against the ACTUAL parent array (incl. lists nested in
-  // callouts / table cells); first-match / top-level-only semantics preserved.
-  const chain = findAnchorChain(out, opts);
-  if (chain == null || chain.length < 2) return { doc: out, inserted: false };
-  const parent = chain[chain.length - 2].node.content;
-  if (!Array.isArray(parent)) return { doc: out, inserted: false };
-  const at = chain[chain.length - 1].index + offset;
-  parent.splice(at, 0, ...fresh);
-  coalesceSeams(parent, at, at + fresh.length);
-  return { doc: out, inserted: true };
+  // Resolve by id anywhere in the tree: splice the whole array into the parent.
+  if (opts.anchorNodeId != null) {
+    let inserted = false;
+    const walkContent = (content: any[]): void => {
+      for (let i = 0; i < content.length; i++) {
+        const child = content[i];
+        if (matchesId(child, opts.anchorNodeId as string)) {
+          content.splice(i + offset, 0, ...fresh);
+          inserted = true;
+          return;
+        }
+        if (isObject(child) && Array.isArray(child.content)) {
+          walkContent(child.content);
+          if (inserted) return;
+        }
+      }
+    };
+    if (isObject(out) && Array.isArray(out.content)) {
+      walkContent(out.content);
+    }
+    return { doc: out, inserted };
+  }
+
+  // Resolve by text: only top-level doc.content blocks are scanned. Exact match
+  // wins; a markdown-stripped fallback is tried only on a miss.
+  if (opts.anchorText != null && isObject(out) && Array.isArray(out.content)) {
+    const i = findAnchorTextIndex(out.content, opts.anchorText);
+    if (i !== -1) {
+      out.content.splice(i + offset, 0, ...fresh);
+      return { doc: out, inserted: true };
+    }
+  }
+
+  return { doc: out, inserted: false };
 }
 
 // ===========================================================================

packages/prosemirror-markdown/test/node-ops-list-coalesce.test.ts

@@ -1,390 +0,0 @@
-import { describe, expect, it } from "vitest";
-import {
-  insertNodesRelative,
-  insertNodeRelative,
-} from "../src/lib/node-ops.js";
-import { markdownToProseMirrorSync } from "../src/lib/markdown-to-prosemirror.js";
-import { convertProseMirrorToMarkdown } from "../src/lib/markdown-converter.js";
-
-// #535: inserting a markdown/JSON list next to an existing same-type sibling
-// list must APPEND/PREPEND its items into that list (one list), not leave two
-// adjacent lists that the serializer separates with `<!-- -->`. Coalescing is
-// strictly local to the two seams of the active insertion.
-
-// ---- Minimal builders (structure is what these tests assert) ---------------
-
-function item(label: string | number): any {
-  return {
-    type: "listItem",
-    content: [
-      { type: "paragraph", content: [{ type: "text", text: String(label) }] },
-    ],
-  };
-}
-function taskItem(label: string | number, checked = false): any {
-  return {
-    type: "taskItem",
-    attrs: { checked },
-    content: [
-      { type: "paragraph", content: [{ type: "text", text: String(label) }] },
-    ],
-  };
-}
-function bulletList(id: string, ...labels: (string | number)[]): any {
-  return { type: "bulletList", attrs: { id }, content: labels.map(item) };
-}
-function orderedList(
-  id: string,
-  start: number,
-  attrsType: string | null,
-  ...labels: (string | number)[]
-): any {
-  return {
-    type: "orderedList",
-    attrs: { id, start, type: attrsType },
-    content: labels.map(item),
-  };
-}
-function taskList(id: string, ...labels: (string | number)[]): any {
-  return { type: "taskList", attrs: { id }, content: labels.map((l) => taskItem(l)) };
-}
-
-/** Read a list block's item labels (the text of each item's first paragraph). */
-function labels(list: any): string[] {
-  return (list.content ?? []).map(
-    (it: any) => it.content?.[0]?.content?.[0]?.text,
-  );
-}
-
-describe("#535 list seam coalescing — markdown path (insertNodesRelative)", () => {
-  it("crit1: append a bulletList onto an adjacent bulletList -> ONE list", () => {
-    const doc = { type: "doc", content: [bulletList("A", 1, 2, 3)] };
-    const { doc: out, inserted } = insertNodesRelative(
-      doc,
-      [bulletList("B", 4)],
-      { position: "append" },
-    );
-    expect(inserted).toBe(true);
-    // Exactly one block, and it is the pre-existing survivor (id "A").
-    expect(out.content.length).toBe(1);
-    expect(out.content[0].type).toBe("bulletList");
-    expect(out.content[0].attrs.id).toBe("A");
-    expect(labels(out.content[0])).toEqual(["1", "2", "3", "4"]);
-  });
-
-  it("crit2: append an orderedList -> ONE list, original start kept, 1..4 continuous, no start:1 sibling", () => {
-    const doc = { type: "doc", content: [orderedList("A", 1, null, 1, 2, 3)] };
-    const { doc: out } = insertNodesRelative(
-      doc,
-      [orderedList("B", 1, null, 4)],
-      { position: "append" },
-    );
-    expect(out.content.length).toBe(1);
-    expect(out.content[0].type).toBe("orderedList");
-    expect(out.content[0].attrs.start).toBe(1);
-    expect(labels(out.content[0])).toEqual(["1", "2", "3", "4"]);
-  });
-
-  it("crit3: after/before by anchorNodeId on a same-type list -> one merged list", () => {
-    const docA = { type: "doc", content: [bulletList("A", 1, 2)] };
-    const afterRes = insertNodesRelative(docA, [bulletList("B", 3)], {
-      position: "after",
-      anchorNodeId: "A",
-    });
-    expect(afterRes.doc.content.length).toBe(1);
-    expect(labels(afterRes.doc.content[0])).toEqual(["1", "2", "3"]);
-
-    const docB = { type: "doc", content: [bulletList("A", 2, 3)] };
-    const beforeRes = insertNodesRelative(docB, [bulletList("B", 1)], {
-      position: "before",
-      anchorNodeId: "A",
-    });
-    expect(beforeRes.doc.content.length).toBe(1);
-    // Survivor is the pre-existing "A"; inserted items PREPEND.
-    expect(beforeRes.doc.content[0].attrs.id).toBe("A");
-    expect(labels(beforeRes.doc.content[0])).toEqual(["1", "2", "3"]);
-  });
-
-  it("crit3: anchorText resolving to a list block also merges", () => {
-    const doc = { type: "doc", content: [bulletList("A", "alpha", "beta")] };
-    const { doc: out } = insertNodesRelative(doc, [bulletList("B", "gamma")], {
-      position: "after",
-      anchorText: "alpha",
-    });
-    expect(out.content.length).toBe(1);
-    expect(labels(out.content[0])).toEqual(["alpha", "beta", "gamma"]);
-  });
-
-  it("crit4: both seams (three-way) — inserting a list between two same-type lists -> ONE list in order", () => {
-    const doc = {
-      type: "doc",
-      content: [bulletList("A", 1, 2), bulletList("C", 4, 5)],
-    };
-    const { doc: out } = insertNodesRelative(doc, [bulletList("B", 3)], {
-      position: "after",
-      anchorNodeId: "A",
-    });
-    expect(out.content.length).toBe(1);
-    // LEFT pre-existing "A" survives; order A-items, inserted, C-items.
-    expect(out.content[0].attrs.id).toBe("A");
-    expect(labels(out.content[0])).toEqual(["1", "2", "3", "4", "5"]);
-  });
-
-  it("crit4: multi-block run coalesces BOTH boundary seams, inner block untouched", () => {
-    // parent = [preA, Bi, para, Bj, preB]; run [Bi, para, Bj] inserted after preA.
-    const doc = {
-      type: "doc",
-      content: [bulletList("A", 1), bulletList("E", 9)],
-    };
-    const run = [
-      bulletList("Bi", 2),
-      { type: "paragraph", content: [{ type: "text", text: "mid" }] },
-      bulletList("Bj", 8),
-    ];
-    const { doc: out } = insertNodesRelative(doc, run, {
-      position: "after",
-      anchorNodeId: "A",
-    });
-    // Expect: [A(1,2), para(mid), E(8,9)] — preA absorbed Bi, preE absorbed Bj.
-    expect(out.content.map((n: any) => n.type)).toEqual([
-      "bulletList",
-      "paragraph",
-      "bulletList",
-    ]);
-    expect(labels(out.content[0])).toEqual(["1", "2"]);
-    expect(out.content[1].content[0].text).toBe("mid");
-    expect(labels(out.content[2])).toEqual(["8", "9"]);
-    expect(out.content[0].attrs.id).toBe("A");
-    expect(out.content[2].attrs.id).toBe("E");
-  });
-
-  it("crit4b: three-way with DIFFERING explicit ordered types does NOT transitively merge through a default-typed insertion", () => {
-    // [ol(type:"a")[1], ol(type:"i")[9]] + insert ol(type:null)[5] after A.
-    // The null-typed middle is pairwise-compatible with BOTH neighbours, but the
-    // neighbours are mutually incompatible ("a" vs "i"), so they must NOT
-    // collapse. The inserted list is absorbed by the LEFT survivor; the RIGHT
-    // list stays separate and keeps its roman numbering style.
-    const doc = {
-      type: "doc",
-      content: [orderedList("A", 1, "a", 1), orderedList("C", 9, "i", 9)],
-    };
-    const { doc: out } = insertNodeRelative(doc, orderedList("B", 5, null, 5), {
-      position: "after",
-      anchorNodeId: "A",
-    });
-    expect(out.content.length).toBe(2);
-    // Left survivor absorbs the inserted item.
-    expect(out.content[0].attrs.id).toBe("A");
-    expect(out.content[0].attrs.type).toBe("a");
-    expect(labels(out.content[0])).toEqual(["1", "5"]);
-    // Right list is untouched: its explicit style survives.
-    expect(out.content[1].attrs.id).toBe("C");
-    expect(out.content[1].attrs.type).toBe("i");
-    expect(labels(out.content[1])).toEqual(["9"]);
-  });
-
-  it("crit6: before for orderedList{start:5} keeps start:5 (positional survivor = pre-existing)", () => {
-    const doc = { type: "doc", content: [orderedList("A", 5, null, "a", "b")] };
-    const { doc: out } = insertNodesRelative(
-      doc,
-      [orderedList("B", 1, null, "c")],
-      { position: "before", anchorNodeId: "A" },
-    );
-    expect(out.content.length).toBe(1);
-    expect(out.content[0].attrs.id).toBe("A");
-    expect(out.content[0].attrs.start).toBe(5);
-    expect(labels(out.content[0])).toEqual(["c", "a", "b"]);
-  });
-
-  it("crit7: different list types do NOT coalesce", () => {
-    // bulletList next to orderedList
-    const d1 = { type: "doc", content: [bulletList("A", 1)] };
-    const r1 = insertNodesRelative(d1, [orderedList("B", 1, null, 2)], {
-      position: "append",
-    });
-    expect(r1.doc.content.length).toBe(2);
-    expect(r1.doc.content.map((n: any) => n.type)).toEqual([
-      "bulletList",
-      "orderedList",
-    ]);
-
-    // bulletList next to taskList
-    const d2 = { type: "doc", content: [bulletList("A", 1)] };
-    const r2 = insertNodesRelative(d2, [taskList("B", 2)], {
-      position: "append",
-    });
-    expect(r2.doc.content.length).toBe(2);
-    expect(r2.doc.content.map((n: any) => n.type)).toEqual([
-      "bulletList",
-      "taskList",
-    ]);
-  });
-
-  it("crit7b: orderedLists with explicit DIFFERING attrs.type do NOT coalesce", () => {
-    const doc = { type: "doc", content: [orderedList("A", 1, "a", 1)] };
-    const { doc: out } = insertNodesRelative(
-      doc,
-      [orderedList("B", 1, "i", 2)],
-      { position: "append" },
-    );
-    expect(out.content.length).toBe(2);
-  });
-
-  it("footnotesList is NEVER structurally coalesced (allow-list excludes it, guarding against endsWith(\"List\"))", () => {
-    const fn = (id: string): any => ({
-      type: "footnotesList",
-      attrs: { id },
-      content: [
-        { type: "footnoteDefinition", attrs: { id: id + "d" }, content: [] },
-      ],
-    });
-    const doc = { type: "doc", content: [fn("A")] };
-    const { doc: out } = insertNodeRelative(doc, fn("B"), {
-      position: "append",
-    });
-    // Two footnotesLists must stay two separate blocks — merging them would
-    // corrupt footnotes.
-    expect(out.content.length).toBe(2);
-    expect(out.content.map((n: any) => n.type)).toEqual([
-      "footnotesList",
-      "footnotesList",
-    ]);
-  });
-
-  it("taskList next to taskList coalesces, item checked attrs move with items", () => {
-    const doc = {
-      type: "doc",
-      content: [
-        {
-          type: "taskList",
-          attrs: { id: "A" },
-          content: [taskItem("x", true)],
-        },
-      ],
-    };
-    const { doc: out } = insertNodesRelative(
-      doc,
-      [{ type: "taskList", attrs: { id: "B" }, content: [taskItem("y", false)] }],
-      { position: "append" },
-    );
-    expect(out.content.length).toBe(1);
-    expect(out.content[0].content.map((it: any) => it.attrs.checked)).toEqual([
-      true,
-      false,
-    ]);
-  });
-});
-
-describe("#535 list seam coalescing — node path (insertNodeRelative)", () => {
-  it("append a single bulletList node onto an adjacent bulletList -> ONE list", () => {
-    const doc = { type: "doc", content: [bulletList("A", 1, 2)] };
-    const { doc: out } = insertNodeRelative(doc, bulletList("B", 3), {
-      position: "append",
-    });
-    expect(out.content.length).toBe(1);
-    expect(labels(out.content[0])).toEqual(["1", "2", "3"]);
-  });
-
-  it("crit8: EMPTY inserted list is NOT coalesced (stays as inserted)", () => {
-    const doc = { type: "doc", content: [bulletList("A", 1, 2)] };
-    const empty = { type: "bulletList", attrs: { id: "B" }, content: [] };
-    const { doc: out, inserted } = insertNodeRelative(doc, empty, {
-      position: "append",
-    });
-    expect(inserted).toBe(true);
-    expect(out.content.length).toBe(2);
-    expect(out.content[1].attrs.id).toBe("B");
-    expect(out.content[1].content.length).toBe(0);
-  });
-
-  it("crit5b: nested parent — merging a list next to a list INSIDE a callout, top level untouched", () => {
-    const doc = {
-      type: "doc",
-      content: [
-        {
-          type: "callout",
-          attrs: { id: "co" },
-          content: [bulletList("A", 1, 2)],
-        },
-      ],
-    };
-    const { doc: out } = insertNodeRelative(doc, bulletList("B", 3), {
-      position: "after",
-      anchorNodeId: "A",
-    });
-    // Top level unchanged: still one callout.
-    expect(out.content.length).toBe(1);
-    expect(out.content[0].type).toBe("callout");
-    // The callout's content array holds ONE merged bulletList.
-    expect(out.content[0].content.length).toBe(1);
-    expect(out.content[0].content[0].type).toBe("bulletList");
-    expect(labels(out.content[0].content[0])).toEqual(["1", "2", "3"]);
-  });
-});
-
-describe("#535 locality (regression guard) + round-trip", () => {
-  it("crit5a: appending next to ONE of two intentionally-separate lists merges only that seam; the further-out separate list survives and still emits <!-- -->", () => {
-    // Reimport a doc with two deliberately-separate bullet lists: [(x,y),(z)].
-    const doc = markdownToProseMirrorSync("- x\n- y\n\n<!-- -->\n\n- z");
-    expect(doc.content.filter((n: any) => n.type === "bulletList").length).toBe(
-      2,
-    );
-
-    // Append a list. It lands at the end, adjacent to the SECOND list (z) and
-    // must merge ONLY into it — the FIRST list (x,y) is the intentionally-
-    // separate one two hops away and must NOT be swallowed (a greedy `while`
-    // merge would reach and collapse it, which this guards against).
-    const { doc: out, inserted } = insertNodesRelative(
-      doc,
-      [bulletList("NEW", "q")],
-      { position: "append" },
-    );
-    expect(inserted).toBe(true);
-
-    // Still TWO lists: (x,y) stayed separate; (z) absorbed the appended item.
-    expect(out.content.filter((n: any) => n.type === "bulletList").length).toBe(
-      2,
-    );
-    expect(labels(out.content[0])).toEqual(["x", "y"]);
-    expect(labels(out.content[1])).toEqual(["z", "q"]);
-
-    // Serialize: the separator between the two separate lists remains.
-    const md = convertProseMirrorToMarkdown(out);
-    expect(md).toContain("<!-- -->");
-  });
-
-  it("idempotency: re-running the transform on a fresh doc yields the same result", () => {
-    const doc = { type: "doc", content: [bulletList("A", 1, 2, 3)] };
-    const r1 = insertNodesRelative(doc, [bulletList("B", 4)], {
-      position: "append",
-    });
-    const r2 = insertNodesRelative(doc, [bulletList("B", 4)], {
-      position: "append",
-    });
-    expect(r2.doc).toEqual(r1.doc);
-    // Input never mutated.
-    expect(doc.content.length).toBe(1);
-    expect(labels(doc.content[0])).toEqual(["1", "2", "3"]);
-  });
-
-  it("#351 P1/P2: a merged list serializes to one list and is byte-fixpoint on the 2nd pass", () => {
-    // Build a merged bulletList via a canonical import so the round-trip is real.
-    const base = markdownToProseMirrorSync("- one\n- two");
-    const add = markdownToProseMirrorSync("- three");
-    const { doc: merged } = insertNodesRelative(base, add.content, {
-      position: "append",
-    });
-    expect(merged.content.length).toBe(1);
-
-    const md1 = convertProseMirrorToMarkdown(merged);
-    // A single list — no separator inside.
-    expect(md1).not.toContain("<!-- -->");
-    expect(md1).toBe("- one\n- two\n- three");
-
-    // Reimport -> still ONE bulletList; re-serialize -> byte-identical (fixpoint).
-    const reimported = markdownToProseMirrorSync(md1);
-    expect(reimported.content.filter((n: any) => n.type === "bulletList").length).toBe(1);
-    const md2 = convertProseMirrorToMarkdown(reimported);
-    expect(md2).toBe(md1);
-  });
-});