Merge remote-tracking branch 'gitea/develop' into fix/541-reconnect-timeout

Reviewed-on: #374

.env.example

@@ -302,6 +302,16 @@ MCP_DOCMOST_PASSWORD=
 # enabled for a workspace, and the same single-instance constraint applies (the
 # registry is process-local).
 # AI_CHAT_RESUMABLE_STREAM=false
+#
+# Per-run replay ring cap (#491), in BYTES, for the resumable-stream registry
+# above. The registry buffers the run's recent SSE tail so a reopened tab can
+# attach and continue from the step it already persisted; the ring is bounded and
+# rotates on every confirmed step-persist. This caps the un-persisted tail between
+# rotations — an overflow evicts the oldest frames and a late attach falls back to
+# 204 -> degraded poll, so correctness never depends on the size. Default 4194304
+# (4MB); a 0/invalid value falls back to the default. The per-subscriber backpressure
+# cap is derived as 2x this value. Only meaningful with AI_CHAT_RESUMABLE_STREAM on.
+# AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES=4194304
 
 # --- Run lifecycle tunables (#487) ---
 # These govern the universal run machinery (every turn is now a first-class run,

CHANGELOG.md

@@ -135,6 +135,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   new resync path re-reads the live anchor so the suggestion applies against the
   current text, and orphaned anchors (whose marked run was deleted) are
   reconciled rather than left blocking. (#496)
+- **Save intentional page versions.** Press `Cmd/Ctrl+S` (or use the page menu)
+  to save a named version of a page. The history panel now distinguishes
+  intentional versions (a "Saved" / "Agent version" badge) from automatic
+  snapshots, dims autosaves, and offers an "Only versions" filter. Automatic
+  snapshots switched from a fixed interval to a trailing idle-flush with a
+  max-wait ceiling, and a boundary snapshot is pinned whenever the editing source
+  changes (e.g. a person's edits followed by the AI agent). (#370)
 
 - **Place several images side by side in a row.** A new "Inline (side by
   side)" alignment mode in the image bubble menu renders consecutive inline

README.md

@@ -206,6 +206,137 @@ start the new migrations apply on top of your existing schema (`CREATE EXTENSION
   existing pages are indexed on their next edit. pgvector is still required for the migration to
   apply at all.
 
+## Local embeddings server
+
+The AI agent's semantic (RAG) search needs an **embeddings model**. Instead of paying a cloud
+provider (e.g. OpenAI `text-embedding-3-*`) to embed every page, you can run a small open-weights
+model yourself with Hugging Face
+[Text Embeddings Inference](https://github.com/huggingface/text-embeddings-inference) (TEI), which
+serves an OpenAI-compatible `/v1/embeddings` endpoint. `intfloat/multilingual-e5-small` is a good
+default: multilingual, 384-dim, and comfortable on CPU (~1–2 GB RAM, 1–2 vCPU). Point Gitmost at it
+under **Workspace settings → AI → Embeddings**.
+
+### Option A — local (same Docker network as Gitmost)
+
+Run TEI as a container on the network Gitmost is already on. The port is never published, so the
+endpoint stays internal and needs no authentication.
+
+```yaml
+services:
+  embeddings:
+    image: ghcr.io/huggingface/text-embeddings-inference:cpu-1.9   # pin version; use a cuda-* tag for GPU
+    container_name: embeddings
+    restart: unless-stopped
+    networks:
+      - gitmost_net          # same network Gitmost is on
+    command:
+      - "--model-id"
+      - "intfloat/multilingual-e5-small"
+      - "--auto-truncate"    # clamp over-long inputs instead of returning 413
+    volumes:
+      - tei-models:/data     # weights are downloaded once and cached here
+
+networks:
+  gitmost_net:
+    external: true           # the network Gitmost already uses
+
+volumes:
+  tei-models:
+```
+
+Gitmost settings (**Workspace settings → AI → Embeddings**):
+
+| Field             | Value                             |
+|-------------------|-----------------------------------|
+| Model             | `intfloat/multilingual-e5-small`  |
+| Base URL          | `http://embeddings:80/v1/`        |
+| Embedding API key | — (leave empty)                   |
+
+> `embeddings` is the container name — Gitmost resolves it over DNS inside the Docker network.
+> The port is not published, so the endpoint is reachable only by containers on that network and
+> no authorization is required.
+
+### Option B — separate host (public via Traefik + Let's Encrypt)
+
+This assumes the host already runs Traefik with an ACME resolver (the example below uses
+`letsEncrypt`, the `websecure` entrypoint and a shared `docker_main_net` network). Replace the
+domain / network / resolver with your own.
+
+**DNS:** add an A record `embeddings.example.com` → the IP of your Traefik host (same
+challenge / port 80 as the rest of your sites).
+
+```yaml
+services:
+  embeddings:
+    image: ghcr.io/huggingface/text-embeddings-inference:cpu-1.9   # pin version; cuda-* tag for GPU
+    container_name: embeddings
+    restart: unless-stopped
+    networks:
+      - docker_main_net      # the network Traefik is attached to
+    command:
+      - "--model-id"
+      - "intfloat/multilingual-e5-small"
+      - "--auto-truncate"
+      - "--api-key"
+      - "sk-emb-REPLACE_WITH_YOUR_KEY"
+    volumes:
+      - tei-models:/data
+    labels:
+      traefik.enable: "true"
+      traefik.http.routers.embeddings.rule: "Host(`embeddings.example.com`)"
+      traefik.http.routers.embeddings.entrypoints: "websecure"
+      traefik.http.routers.embeddings.tls: "true"
+      traefik.http.routers.embeddings.tls.certresolver: "letsEncrypt"
+      traefik.http.routers.embeddings.service: "embeddings"
+      traefik.http.services.embeddings.loadbalancer.server.port: "80"
+      # TEI enforces the Bearer key itself; Traefik only rate-limits to protect the CPU
+      traefik.http.routers.embeddings.middlewares: "embeddings-rl"
+      traefik.http.middlewares.embeddings-rl.ratelimit.average: "20"
+      traefik.http.middlewares.embeddings-rl.ratelimit.burst: "40"
+      traefik.http.middlewares.embeddings-rl.ratelimit.period: "1s"
+
+networks:
+  docker_main_net:
+    external: true
+
+volumes:
+  tei-models:
+```
+
+Gitmost settings (**Workspace settings → AI → Embeddings**):
+
+| Field             | Value                                 |
+|-------------------|---------------------------------------|
+| Model             | `intfloat/multilingual-e5-small`      |
+| Base URL          | `https://embeddings.example.com/v1/`  |
+| Embedding API key | your `sk-emb-…`                       |
+
+Check it from outside:
+
+```bash
+curl -s https://embeddings.example.com/v1/embeddings \
+  -H "Authorization: Bearer sk-emb-REPLACE_WITH_YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"model":"intfloat/multilingual-e5-small","input":"query: hello"}' \
+  | python3 -c 'import sys,json;print("dims:",len(json.load(sys.stdin)["data"][0]["embedding"]))'
+# -> dims: 384
+```
+
+### Embeddings server notes
+
+- **Vector dimension is 384.** If this Gitmost was previously embedded with a different model
+  (e.g. `text-embedding-3-large` = 3072-dim), the old pgvector rows won't match the new dimension —
+  clear the existing embeddings / re-index before switching. Gitmost only compares vectors of the
+  same dimension, so mixed-dimension rows are silently ignored rather than searched.
+- **First start downloads the weights** (hundreds of MB) from `huggingface.co` into the
+  `tei-models` volume; every start after that reads from the volume.
+- **Pin the version.** Pin the image, and optionally the model: add `--revision <commit-sha>` to
+  `command` (the sha is on the model's page on Hugging Face).
+- **Air-gapped / no egress:** seed the `tei-models` volume ahead of time and add
+  `environment: [HF_HUB_OFFLINE=1]`.
+- **GPU:** use the cuda tag of the same release (e.g.
+  `ghcr.io/huggingface/text-embeddings-inference:cuda-1.9`) and start the container with `gpus: all`.
+
 ## Features
 
 - Real-time collaboration

README.ru.md

@@ -193,6 +193,137 @@ dump/restore, существующий каталог данных переис
 > неизменным и бэкапьте вместе с базой данных.
 
 
+## Локальный сервер эмбеддингов
+
+Семантическому (RAG) поиску AI-агента нужна **модель эмбеддингов**. Вместо оплаты облачного
+провайдера (например, OpenAI `text-embedding-3-*`) за эмбеддинг каждой страницы можно запустить
+небольшую open-weights модель у себя через Hugging Face
+[Text Embeddings Inference](https://github.com/huggingface/text-embeddings-inference) (TEI) — он
+отдаёт OpenAI-совместимый эндпоинт `/v1/embeddings`. Хороший дефолт — `intfloat/multilingual-e5-small`:
+многоязычная, 384-мерная, комфортно работает на CPU (~1–2 ГБ RAM, 1–2 vCPU). Пропишите её в
+**Настройки воркспейса → AI → Эмбеддинги**.
+
+### Вариант A — локально (та же Docker-сеть, что и Gitmost)
+
+Запустите TEI контейнером в той же сети, где уже работает Gitmost. Порт наружу не публикуется,
+поэтому эндпоинт остаётся внутренним и не требует авторизации.
+
+```yaml
+services:
+  embeddings:
+    image: ghcr.io/huggingface/text-embeddings-inference:cpu-1.9   # pin version; use a cuda-* tag for GPU
+    container_name: embeddings
+    restart: unless-stopped
+    networks:
+      - gitmost_net          # same network Gitmost is on
+    command:
+      - "--model-id"
+      - "intfloat/multilingual-e5-small"
+      - "--auto-truncate"    # clamp over-long inputs instead of returning 413
+    volumes:
+      - tei-models:/data     # weights are downloaded once and cached here
+
+networks:
+  gitmost_net:
+    external: true           # the network Gitmost already uses
+
+volumes:
+  tei-models:
+```
+
+Настройки Gitmost (**Настройки воркспейса → AI → Эмбеддинги**):
+
+| Поле              | Значение                          |
+|-------------------|-----------------------------------|
+| Model             | `intfloat/multilingual-e5-small`  |
+| Base URL          | `http://embeddings:80/v1/`        |
+| Embedding API key | — (оставить пустым)               |
+
+> `embeddings` — имя контейнера, Gitmost резолвит его по DNS внутри Docker-сети.
+> Наружу порт не публикуется, эндпоинт доступен только контейнерам этой сети, поэтому
+> авторизация не нужна.
+
+### Вариант B — на отдельном хосте (наружу через Traefik + Let's Encrypt)
+
+Предполагается, что на хосте уже есть Traefik с ACME-резолвером (в примере ниже — `letsEncrypt`,
+entrypoint `websecure`, общая сеть `docker_main_net`). Замените домен / сеть / резолвер на свои.
+
+**DNS:** заведите A-запись `embeddings.example.com` → IP хоста с Traefik (тот же challenge / порт 80,
+что и у остальных сайтов).
+
+```yaml
+services:
+  embeddings:
+    image: ghcr.io/huggingface/text-embeddings-inference:cpu-1.9   # pin version; cuda-* tag for GPU
+    container_name: embeddings
+    restart: unless-stopped
+    networks:
+      - docker_main_net      # the network Traefik is attached to
+    command:
+      - "--model-id"
+      - "intfloat/multilingual-e5-small"
+      - "--auto-truncate"
+      - "--api-key"
+      - "sk-emb-REPLACE_WITH_YOUR_KEY"
+    volumes:
+      - tei-models:/data
+    labels:
+      traefik.enable: "true"
+      traefik.http.routers.embeddings.rule: "Host(`embeddings.example.com`)"
+      traefik.http.routers.embeddings.entrypoints: "websecure"
+      traefik.http.routers.embeddings.tls: "true"
+      traefik.http.routers.embeddings.tls.certresolver: "letsEncrypt"
+      traefik.http.routers.embeddings.service: "embeddings"
+      traefik.http.services.embeddings.loadbalancer.server.port: "80"
+      # TEI enforces the Bearer key itself; Traefik only rate-limits to protect the CPU
+      traefik.http.routers.embeddings.middlewares: "embeddings-rl"
+      traefik.http.middlewares.embeddings-rl.ratelimit.average: "20"
+      traefik.http.middlewares.embeddings-rl.ratelimit.burst: "40"
+      traefik.http.middlewares.embeddings-rl.ratelimit.period: "1s"
+
+networks:
+  docker_main_net:
+    external: true
+
+volumes:
+  tei-models:
+```
+
+Настройки Gitmost (**Настройки воркспейса → AI → Эмбеддинги**):
+
+| Поле              | Значение                              |
+|-------------------|---------------------------------------|
+| Model             | `intfloat/multilingual-e5-small`      |
+| Base URL          | `https://embeddings.example.com/v1/`  |
+| Embedding API key | ваш `sk-emb-…`                        |
+
+Проверка снаружи:
+
+```bash
+curl -s https://embeddings.example.com/v1/embeddings \
+  -H "Authorization: Bearer sk-emb-REPLACE_WITH_YOUR_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"model":"intfloat/multilingual-e5-small","input":"query: hello"}' \
+  | python3 -c 'import sys,json;print("dims:",len(json.load(sys.stdin)["data"][0]["embedding"]))'
+# -> dims: 384
+```
+
+### Заметки про сервер эмбеддингов
+
+- **Размерность вектора — 384.** Если раньше этот Gitmost эмбеддился другой моделью
+  (например, `text-embedding-3-large` = 3072-dim), старые строки в pgvector не совпадут по
+  размерности — очистите существующие эмбеддинги / переиндексируйте перед переключением. Gitmost
+  сравнивает только вектора одной размерности, поэтому строки другой размерности не участвуют в
+  поиске, а не ломают его.
+- **Первый старт тянет веса** (сотни МБ) с `huggingface.co` в том `tei-models`; дальше — из тома.
+- **Пин версии.** Пиньте образ, а при желании и модель: добавьте в `command` `--revision <commit-sha>`
+  (sha берётся со страницы модели на Hugging Face).
+- **Без egress (air-gapped):** засейте том `tei-models` заранее и добавьте
+  `environment: [HF_HUB_OFFLINE=1]`.
+- **GPU:** возьмите cuda-тег того же релиза (например,
+  `ghcr.io/huggingface/text-embeddings-inference:cuda-1.9`) и запустите контейнер с `gpus: all`.
+
+
 ## Возможности
 
 - Совместная работа в реальном времени

apps/client/public/locales/en-US/translation.json

@@ -1418,5 +1418,14 @@
   "The commented text changed since this suggestion was made; it was not applied.": "The commented text changed since this suggestion was made; it was not applied.",
   "Dismiss": "Dismiss",
   "Suggestion dismissed": "Suggestion dismissed",
-  "Failed to dismiss suggestion": "Failed to dismiss suggestion"
+  "Failed to dismiss suggestion": "Failed to dismiss suggestion",
+  "Save version": "Save version",
+  "Ctrl+S": "Ctrl+S",
+  "Version saved": "Version saved",
+  "Already saved as the latest version": "Already saved as the latest version",
+  "Agent version": "Agent version",
+  "Boundary": "Boundary",
+  "Autosave": "Autosave",
+  "Only versions": "Only versions",
+  "No saved versions yet.": "No saved versions yet."
 }

apps/client/public/locales/ru-RU/translation.json

@@ -1433,5 +1433,14 @@
   "The commented text changed since this suggestion was made; it was not applied.": "Прокомментированный текст изменился после создания предложения; оно не было применено.",
   "Dismiss": "Не применять",
   "Suggestion dismissed": "Предложение отклонено",
-  "Failed to dismiss suggestion": "Не удалось отклонить предложение"
+  "Failed to dismiss suggestion": "Не удалось отклонить предложение",
+  "Save version": "Сохранить версию",
+  "Ctrl+S": "Ctrl+S",
+  "Version saved": "Версия сохранена",
+  "Already saved as the latest version": "Уже сохранено как последняя версия",
+  "Agent version": "Версия агента",
+  "Boundary": "Граница",
+  "Autosave": "Автосейв",
+  "Only versions": "Только версии",
+  "No saved versions yet.": "Пока нет сохранённых версий."
 }

apps/client/src/features/ai-chat/components/ai-chat-window.tsx

@@ -58,8 +58,11 @@ import ConversationList from "@/features/ai-chat/components/conversation-list.ts
 import ChatThread from "@/features/ai-chat/components/chat-thread.tsx";
 import {
   exportAiChat,
+  getAiChatMessagesDelta,
   stopRun,
 } from "@/features/ai-chat/services/ai-chat-service.ts";
+import { mergeDeltaRowsIntoPages } from "@/features/ai-chat/utils/resume-helpers.ts";
+import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.ts";
 import { useChatSession } from "@/features/ai-chat/hooks/use-chat-session.ts";
 import {
   shouldCollapseOnOutsidePointer,
@@ -269,17 +272,64 @@ export default function AiChatWindow() {
   const { data: messageRows, isLoading: messagesLoading } =
     useAiChatMessagesQuery(
       activeChatId ?? undefined,
-      // DELIBERATELY DUMB: poll every 2.5s WHILE ARMED, otherwise off. NO error
-      // checks (TanStack resets fetchFailureCount each fetch; the poll must survive
-      // a server restart), NO tail checks, NO cap here — the settled/stalled/idle-cap
-      // semantics all live in ChatThread's FSM, which disarms via onResumeFallback.
-      () => (degradedPoll === true ? 2500 : false),
-      // #344: gate on windowOpen too — no message history is fetched (and no
-      // degraded poll runs) while the window is closed; it loads when the window
-      // opens with an active chat.
+      // #491: the full infinite-query no longer POLLS. It seeds the thread ONCE; the
+      // degraded fallback now runs a DELTA poller (below) that augments THIS cache
+      // idempotently, instead of refetching every page (with full parts) every 2.5s.
+      false,
+      // #344: gate on windowOpen too — no message history is fetched while the window
+      // is closed; it loads when the window opens with an active chat.
       windowOpen,
     );
 
+  // #491 degraded DELTA poll. While armed (degradedPoll) and the window is open on a
+  // chat, poll POST /ai-chat/messages/delta every 2.5s: it returns only the rows
+  // CHANGED since the previous cursor (+ the run fact) in ONE round-trip. We merge
+  // those rows into the SAME infinite-query cache the thread reads (idempotently by
+  // id — the delta's overlap window re-delivers rows), so the thread's reconcile
+  // effect follows the detached run to its terminal row from a fraction of the wire
+  // cost. The run-fact settle stays the thread FSM's job (row-status reconcile), so
+  // we do NOT double-poll /run here. Cursor resets when the chat changes / disarms.
+  const deltaCursorRef = useRef<string | undefined>(undefined);
+  useEffect(() => {
+    deltaCursorRef.current = undefined;
+  }, [activeChatId, degradedPoll]);
+  useEffect(() => {
+    if (!degradedPoll || !windowOpen || !activeChatId) return;
+    const chatId = activeChatId;
+    let cancelled = false;
+    const tick = async (): Promise<void> => {
+      try {
+        const res = await getAiChatMessagesDelta(chatId, deltaCursorRef.current);
+        if (cancelled) return;
+        deltaCursorRef.current = res.cursor;
+        if (res.rows.length > 0) {
+          queryClient.setQueryData(
+            AI_CHAT_MESSAGES_RQ_KEY(chatId),
+            (
+              old:
+                | {
+                    pages: { items: IAiChatMessageRow[]; meta: unknown }[];
+                    pageParams: unknown[];
+                  }
+                | undefined,
+            ) =>
+              old
+                ? { ...old, pages: mergeDeltaRowsIntoPages(old.pages, res.rows) }
+                : old,
+          );
+        }
+      } catch {
+        // Transient failure (e.g. a server restart mid-run): swallow and retry on
+        // the next tick — the poll must survive a bounce, like the old dumb refetch.
+      }
+    };
+    const id = setInterval(() => void tick(), 2500);
+    return () => {
+      cancelled = true;
+      clearInterval(id);
+    };
+  }, [degradedPoll, windowOpen, activeChatId, queryClient]);
+
   // #184 reconnect-and-live-follow. Whether detached agent runs are enabled for
   // this workspace. When the feature is off no runs are ever created, so the
   // resume attempt would only ever 204; gating ChatThread's resume on it avoids a

apps/client/src/features/ai-chat/components/chat-thread.test.tsx

@@ -172,9 +172,18 @@ function resetState() {
   h.state.getRun.mockResolvedValue({ run: null, message: null });
 }
 
+// #491: the streaming tail carries a persisted step frontier (metadata.stepsPersisted),
+// which the tail-only attach reads as `n` in `?anchor=<id>&n=<n>`. Seeded WHOLE now.
 const streamingTail = () => [
   row("u1", "user", undefined, "hi"),
-  row("a1", "assistant", "streaming", "partial"),
+  {
+    id: "a1",
+    role: "assistant",
+    content: "partial",
+    status: "streaming",
+    createdAt: "2026-01-01T00:00:00Z",
+    metadata: { stepsPersisted: 2 },
+  } as IAiChatMessageRow,
 ];
 const settledTail = () => [
   row("u1", "user", undefined, "hi"),
@@ -335,20 +344,24 @@ describe("ChatThread — send now", () => {
     expect(screen.getAllByLabelText("Remove queued message")).toHaveLength(1);
   });
 
-  it("Stop then a REAL network-drop finish exits to idle (honor-in-stopping), NOT a false reconnect", () => {
+  it("Stop then a REAL network-drop finish exits to idle (honor-in-stopping), NOT a false reconnect", async () => {
     // Regression for the disconnect-first reorder: on the STOP path, even a drop-
     // form finish { isError:true, isDisconnect:true } arriving in `stopping` must be
     // HONORED (reducer) and exit to idle — it must NOT enter the reconnect ladder.
     startLocalStreamWithRun(); // live local stream, autonomous
     fireEvent.click(screen.getByLabelText("Stop")); // STOP_REQUESTED -> stopping
     h.state.error = { message: "Failed to fetch" };
-    act(() => {
+    // #491: the disconnect re-seeds from persist (async getRun) before dispatching
+    // FINISH_DISCONNECT, which the reducer HONORS in `stopping` -> idle. Flush it.
+    await act(async () => {
       h.state.onFinish?.({
         message: { id: "a1", role: "assistant", parts: [] },
         isAbort: false,
         isDisconnect: true,
         isError: true,
       });
+      await Promise.resolve();
+      await Promise.resolve();
     });
     expect(screen.queryByText(/reconnecting/i)).toBeNull();
   });
@@ -803,19 +816,24 @@ describe("ChatThread — resume (attach) machinery", () => {
     expect(h.state.resumeStream).not.toHaveBeenCalled();
   });
 
-  it("strips the streaming tail from the seed, keeps a user tail whole", () => {
+  it("#491 tail-only: seeds the streaming tail WHOLE (no strip), keeps a user tail whole", () => {
     renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() });
-    expect(h.state.seededMessages).toHaveLength(1);
+    // MUTATION-VERIFY: re-introduce the seed-strip and this goes red — the streaming
+    // tail (steps 0..N-1) MUST be seeded so the SDK continuation appends the tail to
+    // the RIGHT message. Both rows (user + assistant) are seeded.
+    expect(h.state.seededMessages).toHaveLength(2);
     cleanup();
     resetState();
     renderThread({ autonomousRunsEnabled: true, initialRows: userTail() });
     expect(h.state.seededMessages).toHaveLength(1);
   });
 
-  it("builds the attach URL with expect=live&anchor only for a stripped streaming tail", () => {
+  it("#491 tail-only: builds the attach URL with ?anchor=&n= from the persisted step frontier", () => {
     renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() });
+    // n=2 comes from a1's metadata.stepsPersisted (MUTATION-VERIFY: hardcode n=0 and
+    // this fails). No `expect=live` param anymore.
     expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
-      "/api/ai-chat/runs/c1/stream?expect=live&anchor=a1",
+      "/api/ai-chat/runs/c1/stream?anchor=a1&n=2",
     );
     cleanup();
     resetState();
@@ -839,39 +857,41 @@ describe("ChatThread — resume (attach) machinery", () => {
     });
   }
 
-  it("204 on a streaming tail: restore + invalidate + onResumeFallback(true)", async () => {
+  it("204 on a streaming tail: NO restore (row kept) + invalidate + onResumeFallback(true)", async () => {
     const { onResumeFallback, invalidateSpy } = renderThread({
       autonomousRunsEnabled: true,
       initialRows: streamingTail(),
     });
     await attachFetch({ status: 204, ok: false });
-    expect(h.state.setMessages).toHaveBeenCalledTimes(1); // restore
+    // #491 tail-only: the anchor row was never stripped, so there is NOTHING to
+    // restore. MUTATION-VERIFY: re-add a restore setMessages here and it goes red.
+    expect(h.state.setMessages).not.toHaveBeenCalled();
     expect(invalidateSpy).toHaveBeenCalledWith({
       queryKey: ["ai-chat-messages", "c1"],
     });
     expect(onResumeFallback).toHaveBeenCalledWith(true);
   });
 
-  it("F7 restart-survival: a 500 attach failure restores the row AND arms the poll", async () => {
+  it("F7 restart-survival: a 500 attach failure arms the poll WITHOUT a restore", async () => {
     const { onResumeFallback, invalidateSpy } = renderThread({
       autonomousRunsEnabled: true,
       initialRows: streamingTail(),
     });
     await attachFetch({ status: 500, ok: false });
-    expect(h.state.setMessages).toHaveBeenCalledTimes(1);
+    expect(h.state.setMessages).not.toHaveBeenCalled();
     expect(invalidateSpy).toHaveBeenCalledWith({
       queryKey: ["ai-chat-messages", "c1"],
     });
     expect(onResumeFallback).toHaveBeenCalledWith(true);
   });
 
-  it("F7 restart-survival: a network throw restores the row AND arms the poll", async () => {
+  it("F7 restart-survival: a network throw arms the poll WITHOUT a restore", async () => {
     const { onResumeFallback, invalidateSpy } = renderThread({
       autonomousRunsEnabled: true,
       initialRows: streamingTail(),
     });
     await attachFetch(new Error("network down"), true);
-    expect(h.state.setMessages).toHaveBeenCalledTimes(1);
+    expect(h.state.setMessages).not.toHaveBeenCalled();
     expect(invalidateSpy).toHaveBeenCalledWith({
       queryKey: ["ai-chat-messages", "c1"],
     });
@@ -931,7 +951,7 @@ describe("ChatThread — resume (attach) machinery", () => {
     expect(h.state.sendMessage).not.toHaveBeenCalled();
   });
 
-  it("an empty resumed message (starved replay) restores the row AND arms the poll", () => {
+  it("an empty resumed message (starved replay) arms the poll WITHOUT a restore", () => {
     h.state.status = "ready";
     const { onResumeFallback } = renderThread({
       autonomousRunsEnabled: true,
@@ -947,7 +967,9 @@ describe("ChatThread — resume (attach) machinery", () => {
         isError: false,
       });
     });
-    expect(h.state.setMessages).toHaveBeenCalledTimes(1); // restore
+    // #491 tail-only: the seeded steps 0..N-1 are still on screen (the SDK
+    // continuation never wiped them), so there is nothing to restore — just poll.
+    expect(h.state.setMessages).not.toHaveBeenCalled();
     expect(onResumeFallback).toHaveBeenCalledWith(true); // arm
   });
 
@@ -995,24 +1017,41 @@ describe("ChatThread — live reconnect + stalled", () => {
     cleanup();
   });
 
+  // #491: the authoritative PERSISTED assistant row `getRun` projects on a local
+  // disconnect — the re-seed source. Its metadata.stepsPersisted becomes `n`.
+  const persistedAnchor = (steps = 3) => ({
+    run: { id: "run-1", status: "running" },
+    message: {
+      id: "a2",
+      role: "assistant",
+      content: "persisted 0..N-1",
+      status: "streaming",
+      createdAt: "2026-01-01T00:00:00Z",
+      metadata: { stepsPersisted: steps },
+    },
+  });
+
   // A REAL live SSE drop. ai@6.0.207 emits BOTH { isError:true, isDisconnect:true }
-  // for a network TypeError AND sets useChat `error` — NOT the { isError:false,
-  // error:null } form the old tests fed. This is the form browser QA hit; with the
-  // buggy isError-first routing OR without the errorView render-gate these tests go
-  // red (a real drop surfaces the terminal error banner, masking the reconnect
-  // ladder). MUTATION-VERIFY of disconnect-first + the errorView phase-gate.
-  function disconnect(message: unknown = liveMsg) {
+  // for a network TypeError AND sets useChat `error`. #491: an autonomous local drop
+  // now RE-SEEDS from persist (async getRun) BEFORE entering the reconnect ladder, so
+  // this helper is async and flushes the getRun microtask before returning.
+  async function disconnect(message: unknown = liveMsg) {
     h.state.error = { message: "Failed to fetch" }; // the SDK sets error on the drop
-    act(() => {
+    await act(async () => {
       h.state.onFinish?.({
         message,
         isAbort: false,
         isDisconnect: true,
         isError: true,
       });
+      // Flush the getRun().then re-seed + the deferred FINISH_DISCONNECT dispatch.
+      await Promise.resolve();
+      await Promise.resolve();
     });
   }
   function renderLive() {
+    // The persisted-anchor read the local disconnect performs to re-seed from persist.
+    h.state.getRun.mockResolvedValue(persistedAnchor());
     const view = renderThread({
       autonomousRunsEnabled: true,
       initialRows: settledTail(),
@@ -1032,35 +1071,80 @@ describe("ChatThread — live reconnect + stalled", () => {
     });
   }
 
-  it("a live disconnect starts a backoff reconnect (banner + resumeStream after backoff)", () => {
+  it("#491: a live disconnect RE-SEEDS from persist, then backs off to reconnect with ?anchor=&n=", async () => {
     renderLive();
-    disconnect();
+    await disconnect();
+    // The re-seed read the authoritative persisted row and replaced the live partial.
+    // MUTATION-VERIFY: skip the getRun re-seed (send `n` off the live message) and the
+    // n below no longer matches the PERSISTED stepsPersisted.
+    expect(h.state.getRun).toHaveBeenCalledWith("c1");
+    expect(h.state.setMessages).toHaveBeenCalled(); // re-seeded the store from persist
     expect(screen.getByText(/reconnecting/i)).toBeTruthy();
     expect(h.state.resumeStream).not.toHaveBeenCalled();
     advanceToAttempt(1);
     expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
+    // n=3 is the PERSISTED row's stepsPersisted (from getRun), NOT the live store.
     expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
-      "/api/ai-chat/runs/c1/stream?expect=live&anchor=a2",
+      "/api/ai-chat/runs/c1/stream?anchor=a2&n=3",
     );
   });
 
-  it("#488 (browser QA): the reconnect banner is SHOWN, not masked by the residual useChat error", () => {
+  it("#491 regression (#137/#161 dup): getRun REJECT on a live disconnect drops the live partial + nulls the anchor", async () => {
+    // The re-seed source (getRun) FAILS — a flaky-network blip (SSE + getRun both
+    // fail, network recovers in ~1s). The OLD .catch just re-entered the ladder with
+    // NO re-seed and NO filter, so the reconnect could tail-apply the registry's
+    // frames onto the live partial that ALREADY has those steps -> duplicated text.
+    renderLive();
+    h.state.getRun.mockReset();
+    h.state.getRun.mockRejectedValue(new Error("network"));
+    await disconnect(); // live partial = liveMsg (id "a2")
+    expect(h.state.getRun).toHaveBeenCalledWith("c1");
+    // THE GUARANTEE: on the getRun failure the live partial (a2) is FILTERED from the
+    // store, so the reconnect can never tail-apply already-present steps onto it.
+    // MUTATION-VERIFY: revert the .catch fix (enterReconnect only, no filter) and no
+    // setMessages call removes a2 -> this reddens.
+    const removedLivePartial = (
+      h.state.setMessages as unknown as {
+        mock: { calls: [unknown][] };
+      }
+    ).mock.calls.some(([updater]) => {
+      if (typeof updater !== "function") return false;
+      const out = (updater as (p: { id: string }[]) => { id: string }[])([
+        { id: "a2" },
+        { id: "u1" },
+      ]);
+      return !out.some((m) => m.id === "a2");
+    });
+    expect(removedLivePartial).toBe(true);
+    expect(screen.getByText(/reconnecting/i)).toBeTruthy();
+    advanceToAttempt(1);
+    expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
+    // Anchor was nulled -> replay-from-start (no params) / 204 -> poll; never a stale
+    // ?anchor=&n= over the live partial.
+    expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
+      "/api/ai-chat/runs/c1/stream",
+    );
+  });
+
+  it("#488 (browser QA): the reconnect banner is SHOWN, not masked by the residual useChat error", async () => {
     // The drop sets useChat `error` (real SDK), and the terminal errorView describes
     // it ("Lost connection to the server"). The FSM phase-gate must let the
     // `reconnecting` banner WIN over that residual error. MUTATION-VERIFY: revert the
     // errorView phase-gate (show errorView whenever error is set) and the terminal
     // banner masks "reconnecting…" -> red.
     renderLive();
-    disconnect();
+    await disconnect();
     expect(h.state.error).not.toBeNull(); // the SDK error IS set during recovery
     expect(screen.getByText(/reconnecting/i)).toBeTruthy();
     // The terminal "Lost connection… reload" banner must NOT be showing.
     expect(screen.queryByText(/reload and try again/i)).toBeNull();
   });
 
-  it("#488 commit 2: a disconnect BEFORE the first assistant frame reconnects with NO anchor", () => {
+  it("#488 commit 2: a disconnect BEFORE the first assistant frame reconnects with NO anchor", async () => {
     renderLive();
-    disconnect(null); // no assistant message yet (pre-first-frame break)
+    // No persisted assistant row for a pre-first-frame break -> no anchor.
+    h.state.getRun.mockResolvedValue({ run: null, message: null });
+    await disconnect(null); // no assistant message yet (pre-first-frame break)
     expect(screen.getByText(/reconnecting/i)).toBeTruthy();
     expect(
       screen.queryByText("Connection lost — the answer was interrupted."),
@@ -1074,7 +1158,7 @@ describe("ChatThread — live reconnect + stalled", () => {
 
   it("a live re-attach (2xx) clears the reconnect banner", async () => {
     renderLive();
-    disconnect();
+    await disconnect();
     advanceToAttempt(1);
     await reconnect({ status: 200, ok: true });
     expect(screen.queryByText(/reconnecting/i)).toBeNull();
@@ -1082,7 +1166,7 @@ describe("ChatThread — live reconnect + stalled", () => {
 
   it("a 204 arms the degraded poll and backs off to the next attempt", async () => {
     const { onResumeFallback } = renderLive();
-    disconnect();
+    await disconnect();
     advanceToAttempt(1);
     expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
     await reconnect({ status: 204, ok: false });
@@ -1094,7 +1178,7 @@ describe("ChatThread — live reconnect + stalled", () => {
 
   it("exhausts the attempt limit into a manual Retry, which restarts the sequence", async () => {
     renderLive();
-    disconnect();
+    await disconnect();
     for (let n = 1; n <= 5; n++) {
       advanceToAttempt(n);
       expect(h.state.resumeStream).toHaveBeenCalledTimes(n);
@@ -1112,22 +1196,23 @@ describe("ChatThread — live reconnect + stalled", () => {
   it("#488 commit 3: two breaks in a row produce two reconnect cycles", async () => {
     renderLive();
     // First break -> reconnect -> re-attach live.
-    disconnect();
+    await disconnect();
     advanceToAttempt(1);
     expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
     await reconnect({ status: 200, ok: true });
     expect(screen.queryByText(/reconnecting/i)).toBeNull();
-    // The re-attached observer stream drops AGAIN -> a SECOND reconnect cycle
-    // (the old one-shot !wasResumed gate sent this to silent poll).
-    disconnect();
+    // The re-attached observer (live-follow) stream drops AGAIN -> a SECOND reconnect
+    // cycle. #491: this too re-seeds from persist before re-attaching (never tail-
+    // applies over the live-follow partial).
+    await disconnect();
     expect(screen.getByText(/reconnecting/i)).toBeTruthy();
     advanceToAttempt(1);
     expect(h.state.resumeStream).toHaveBeenCalledTimes(2);
   });
 
-  it("does NOT reconnect when autonomous runs are disabled", () => {
+  it("does NOT reconnect when autonomous runs are disabled", async () => {
     renderThread({ autonomousRunsEnabled: false, initialRows: settledTail() });
-    disconnect();
+    await disconnect();
     expect(screen.queryByText(/reconnecting/i)).toBeNull();
     expect(
       screen.getByText("Connection lost — the answer was interrupted."),
@@ -1138,7 +1223,7 @@ describe("ChatThread — live reconnect + stalled", () => {
 
   it("#488 commit 4a: the poll idle cap surfaces a stalled banner + Retry (not silent)", async () => {
     renderLive();
-    disconnect();
+    await disconnect();
     advanceToAttempt(1);
     await reconnect({ status: 204, ok: false }); // arms the poll (reconnecting)
     // No activity for the whole idle cap -> stalled.
@@ -1169,4 +1254,88 @@ describe("ChatThread — live reconnect + stalled", () => {
     });
     expect(onResumeFallback).toHaveBeenCalledWith(false); // POLL_IDLE_CAP -> idle -> disarm
   });
+
+  it("#541: getRun HANGS on a live disconnect — the timeout race still enters reconnect (no silent freeze in `streaming`)", async () => {
+    // MUTATION-VERIFY: revert the race to a bare `getRun(cid).then/.catch` and this
+    // reddens — a HUNG (never-settling, NOT rejected) getRun leaves the FSM stuck in
+    // `streaming` with no reconnect banner and no poll (the axios client sets no
+    // request timeout, and the stalled-idle cap only arms AFTER reconnecting/polling).
+    renderLive();
+    h.state.getRun.mockReset();
+    h.state.getRun.mockReturnValue(new Promise(() => {})); // getRun HANGS forever
+    await disconnect(); // live partial = liveMsg (id "a2")
+    expect(h.state.getRun).toHaveBeenCalledWith("c1");
+    // BEFORE the bound fires the FSM is still in the live turn — the very freeze bug.
+    expect(screen.queryByText(/reconnecting/i)).toBeNull();
+    // The recovery-start bound fires -> the SAME fallback as the reject path.
+    act(() => {
+      vi.advanceTimersByTime(4_000);
+    });
+    expect(screen.getByText(/reconnecting/i)).toBeTruthy();
+    // The live partial (a2) was DROPPED from the store (replay-from-start, no stale
+    // tail-apply base). Mirrors the getRun-REJECT test's inspection.
+    const removedLivePartial = (
+      h.state.setMessages as unknown as {
+        mock: { calls: [unknown][] };
+      }
+    ).mock.calls.some(([updater]) => {
+      if (typeof updater !== "function") return false;
+      const out = (updater as (p: { id: string }[]) => { id: string }[])([
+        { id: "a2" },
+        { id: "u1" },
+      ]);
+      return !out.some((m) => m.id === "a2");
+    });
+    expect(removedLivePartial).toBe(true);
+    // ...and the anchor was NULLED -> replay-from-start (no ?anchor=&n= over the partial).
+    advanceToAttempt(1);
+    expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
+    expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
+      "/api/ai-chat/runs/c1/stream",
+    );
+  });
+
+  it("#541: a getRun resolve AFTER the timeout already fired is IGNORED (no double reconnect, no stale re-seed)", async () => {
+    // The timeout wins first and enters the ladder via replay-from-start. When the
+    // hung getRun FINALLY answers, its late `.then` must be a full no-op: it must not
+    // re-seed the store from the (now stale) persisted row, must not re-set the
+    // anchor, and must not re-enter reconnect. The local `settled` flag makes the
+    // resolve/reject/timeout branches mutually exclusive.
+    // MUTATION-VERIFY: drop the `settled` guard (let the late `.then` run) and the
+    // late re-seed re-sets the anchor (URL regains ?anchor=a2&n=3) + calls setMessages.
+    renderLive();
+    let resolveGetRun!: (v: unknown) => void;
+    h.state.getRun.mockReset();
+    h.state.getRun.mockReturnValue(
+      new Promise((r) => {
+        resolveGetRun = r;
+      }),
+    );
+    await disconnect();
+    act(() => {
+      vi.advanceTimersByTime(4_000); // bound fires -> replay-from-start, reconnecting
+    });
+    expect(screen.getByText(/reconnecting/i)).toBeTruthy();
+    advanceToAttempt(1);
+    expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
+    // Anchor is null -> replay-from-start URL (the pre-condition the late resolve must
+    // not undo).
+    expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
+      "/api/ai-chat/runs/c1/stream",
+    );
+    const setMessagesCallsBefore = h.state.setMessages.mock.calls.length;
+    // NOW the hung getRun finally resolves with a persisted anchor (id a2, steps 3).
+    await act(async () => {
+      resolveGetRun(persistedAnchor());
+      await Promise.resolve();
+    });
+    // The late resolve did NOT re-seed the store...
+    expect(h.state.setMessages.mock.calls.length).toBe(setMessagesCallsBefore);
+    // ...did NOT re-set the anchor (URL stays replay-from-start, no ?anchor=&n=)...
+    expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
+      "/api/ai-chat/runs/c1/stream",
+    );
+    // ...and did NOT trigger a fresh reconnect attach.
+    expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
+  });
 });

apps/client/src/features/ai-chat/components/chat-thread.tsx

@@ -42,7 +42,7 @@ import { assistantMessageHasVisibleContent } from "@/features/ai-chat/utils/mess
 import {
   isStreamingTail,
   isSettledAssistantTail,
-  seedRows,
+  stepsPersistedOf,
   mergeById,
 } from "@/features/ai-chat/utils/resume-helpers.ts";
 import { getRun } from "@/features/ai-chat/services/ai-chat-service.ts";
@@ -81,6 +81,17 @@ const STREAM_THROTTLE_MS = 50;
 // THREAD now (the FSM owns polling->stalled); the window just polls while armed.
 const DEGRADED_POLL_IDLE_MAX_MS = 10 * 60_000;
 
+// #541: the bound on the persist re-seed (getRun) wait when ENTERING the reconnect
+// ladder after a live SSE drop. The axios client (lib/api-client.ts) sets NO request
+// timeout, and the stalled-idle cap only arms AFTER the FSM enters reconnecting/
+// polling — which never happens if getRun HANGS (connection established, no response).
+// Without this bound a live drop + a hung getRun sticks the FSM in `streaming` with
+// no banner and no poll until the browser socket timeout (minutes) — the silent-freeze
+// class #497 eliminates. This is a deliberate RECOVERY-START bound (drop the live
+// partial + enter the ladder so the poll/idle-cap arms), intentionally much shorter
+// than any network socket timeout — not a network read timeout.
+const RECONNECT_RESEED_TIMEOUT_MS = 4_000;
+
 /** The #487 active (non-terminal) run statuses — mirrors the server's
  *  ACTIVE_RUN_STATUSES. A run-fact is "active" only for these. */
 function isActiveRunStatus(status: string | null | undefined): boolean {
@@ -266,25 +277,36 @@ export default function ChatThread({
   // is NOT one of the lifecycle flags the FSM replaced.
   const mountedRef = useRef(true);
 
-  // attachStrategy DATA (behind the resumeStream effect; #491 swaps it to tail-only
-  // WITHOUT touching the FSM). The controller is effect-owned (aborted in cleanup,
-  // I5). `stripRef`/`strippedRowRef` are the current full-replay+strip anchor.
+  // attachStrategy DATA (behind the resumeStream effect; #491 tail-only, WITHOUT
+  // touching the FSM). The controller is effect-owned (aborted in cleanup, I5).
+  // `anchorRef` is the PERSISTED assistant row that pins the run (server invariant
+  // 6) and its persisted step frontier N: it feeds `?anchor=<id>&n=<stepsPersisted>`
+  // so the tail-only attach returns frames for steps >= N (the seed carries 0..N-1).
+  // It is NOT a "stripped" row — the seed keeps every row (tail-only replaces the
+  // old full-replay+strip). Null when there is no streaming/active tail to resume.
   const attachAbortRef = useRef<AbortController | null>(null);
-  const stripRef = useRef(chatId !== null && isStreamingTail(initialRows ?? []));
-  const strippedRowRef = useRef<IAiChatMessageRow | null>(
-    stripRef.current ? (initialRows ?? [])[initialRows!.length - 1] : null,
+  const anchorRef = useRef<{ id: string; stepsPersisted: number } | null>(
+    (() => {
+      if (chatId === null || !isStreamingTail(initialRows ?? [])) return null;
+      const rows = initialRows ?? [];
+      const tail = rows[rows.length - 1];
+      return { id: tail.id, stepsPersisted: stepsPersistedOf(tail) };
+    })(),
   );
   // Effect-owned backoff timers (not lifecycle flags): the reconnect ladder and the
   // stalled inactivity cap. Cleared by the cancelReconnect effect / the cap effect.
   const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
   const idleCapTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
+  // #541: the persist re-seed (getRun) timeout race on the disconnect->reconnect
+  // entry. Held in a ref so unmount (DISPOSE cleanup) clears it — no dangling timer.
+  const reseedTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
 
+  // #491 tail-only: seed EVERY persisted row unchanged (no strip). The streaming
+  // tail holds steps 0..N-1; the run-stream registry's tail (steps >= N) is APPENDED
+  // to it by the SDK continuation (readUIMessageStream({ message })), so it must be
+  // present in the store for the attach to continue the RIGHT message.
   const initialMessages = useMemo<UIMessage[]>(
-    () =>
-      seedRows(
-        initialRows ?? [],
-        stripRef.current && autonomousRunsEnabled === true,
-      ).map(rowToUiMessage),
+    () => (initialRows ?? []).map(rowToUiMessage),
     [initialRows],
   );
 
@@ -335,21 +357,16 @@ export default function ChatThread({
     (eff: RunEffect, epoch: number) => {
       switch (eff.type) {
         case "resumeStream": {
-          // The attach GET. Stamp the outcome's generation (I1). A reconnect
-          // attempt filters the pinned live row from the store first (the mount
-          // seed already stripped it), so the live replay's text-start rebuilds it
-          // without duplicating parts (#430).
+          // The attach GET. Stamp the outcome's generation (I1). #491 tail-only: the
+          // store already holds EXACTLY the persisted steps 0..N-1 (the mount seed IS
+          // persist; a reconnect was re-seeded from persist BEFORE FINISH_DISCONNECT
+          // scheduled it — see the onFinish disconnect handler), so there is nothing
+          // to filter here: the SDK continues that seeded message, appending the tail
+          // (steps >= N) without duplicating the pre-drop partial step.
           pendingAttachEpochRef.current = epoch;
           // The resumed stream's onFinish is stamped with THIS attach generation
           // (F1), so a superseded attempt's late finish is dropped.
           turnEpochRef.current = epoch;
-          if (machineRef.current.phase.name === "reconnecting") {
-            const anchor = strippedRowRef.current;
-            if (anchor)
-              setMessagesRef.current?.((prev) =>
-                prev.filter((m) => m.id !== anchor.id),
-              );
-          }
           void resumeStreamRef.current?.();
           break;
         }
@@ -464,18 +481,23 @@ export default function ChatThread({
       new DefaultChatTransport<UIMessage>({
         api: "/api/ai-chat/stream",
         credentials: "include",
-        prepareReconnectToStreamRequest: () => ({
-          // Build the attach URL from the REAL chat id. ?expect=live&anchor=<row id>
-          // only when a streaming tail was stripped: expect=live opts into a
-          // finished-retained replay (safe only because the row is stripped and the
-          // replay rebuilds it), and the anchor pins the replay to OUR run — a
-          // mismatching (newer) run 204s into the restore+poll path instead.
-          api: `/api/ai-chat/runs/${chatIdRef.current}/stream${
-            stripRef.current
-              ? `?expect=live&anchor=${strippedRowRef.current!.id}`
-              : ""
-          }`,
-        }),
+        prepareReconnectToStreamRequest: () => {
+          // #491 tail-only attach URL. When there is an anchor (a streaming/active
+          // tail to resume) build `?anchor=<assistantRowId>&n=<stepsPersisted>`: the
+          // server returns the TAIL — a synthetic `start` frame + frames for steps
+          // >= n, then live — which the SDK continuation appends to the seeded row.
+          // The server 204s (-> restore-noop + poll) when it cannot cover the
+          // frontier (overflow/rotation gap) or the anchor mismatches (a newer run).
+          // No anchor (a user tail / pre-first-frame break) => no params.
+          const anchor = anchorRef.current;
+          return {
+            api: `/api/ai-chat/runs/${chatIdRef.current}/stream${
+              anchor
+                ? `?anchor=${anchor.id}&n=${anchor.stepsPersisted}`
+                : ""
+            }`,
+          };
+        },
         fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => {
           if ((init.method ?? "GET") !== "GET") {
             // Send path (POST). #488 commit 5: NO client 409 retry ladder anymore
@@ -562,8 +584,9 @@ export default function ChatThread({
 
   // Attach GET outcome -> FSM event. The epoch guard replaces BOTH the one-shot
   // 204 guard (noStreamHandledRef) and the unmount gate: a stale/superseded or
-  // post-DISPOSE outcome is dropped (I1). For a NONE outcome the attachStrategy
-  // recovery (restore the stripped row + invalidate for a fresh poll) runs first.
+  // post-DISPOSE outcome is dropped (I1). #491 tail-only: on a NONE outcome there is
+  // NOTHING to restore — the anchor row was never stripped from the view (the seed
+  // keeps it) — so we only invalidate for a fresh poll + dispatch the FSM event.
   const handleAttachOutcome = useCallback(
     (ep: number, wasReconnecting: boolean, live: boolean) => {
       if (ep !== epochRef.current) return; // stale generation — drop
@@ -575,10 +598,6 @@ export default function ChatThread({
         );
         return;
       }
-      if (strippedRowRef.current)
-        setMessagesRef.current?.((prev) =>
-          mergeById(prev, rowToUiMessage(strippedRowRef.current!)),
-        );
       queryClient.invalidateQueries({
         queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current),
       });
@@ -661,56 +680,31 @@ export default function ChatThread({
       // keeps executing server-side — must win; only a NON-disconnect error (a
       // provider 500, `{ isError:true, isDisconnect:false }`) is terminal.
       if (isDisconnect) {
-        if (wasObserver) {
-          // A resumed/attached OBSERVER stream dropped. Recover via the degraded
-          // poll (restore the stripped row only when there is no visible content;
-          // never clobber a fuller on-screen tail, invariant 9). The FSM decides
-          // reconnect-vs-poll from liveFollow (a live-follow drop reconnects again,
-          // #488 commit 3; a mount-resume drop polls).
-          if (mountedRef.current) {
-            const hasVisible = msgHasVisible;
-            if (!hasVisible && strippedRowRef.current)
-              setMessages((prev) =>
-                mergeById(prev, rowToUiMessage(strippedRowRef.current!)),
-              );
-            queryClient.invalidateQueries({
-              queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current),
-            });
-            dispatch({
-              type: "FINISH_DISCONNECT",
-              hasVisibleContent: hasVisible,
-              epoch: stampEpoch,
-            });
-          }
+        if (!mountedRef.current) {
           setStopNotice(null);
           return;
         }
-        // A LOCAL live turn dropped. #488 commit 2: recover by the RUN-FACT, not by
-        // the presence of an assistant message — a setup-phase break (before the
-        // first frame) still leaves a detached run writing to pages. In autonomous
-        // mode a run is active for the whole turn, so seed the run-fact from the
-        // start-metadata runId when known, else a sentinel (the attach GET goes by
-        // chatId, not runId). Pin the assistant row as the strip/anchor when present.
-        if (autonomousRunsEnabled === true && mountedRef.current) {
-          const hasAnchor =
-            message?.role === "assistant" && typeof message.id === "string";
-          if (hasAnchor) {
-            strippedRowRef.current = {
-              id: message.id,
-              role: "assistant",
-              content: "",
-              status: "streaming",
-              createdAt: new Date().toISOString(),
-              metadata: { parts: message.parts },
-            };
-            stripRef.current = true;
-          } else {
-            strippedRowRef.current = null;
-            stripRef.current = false;
-          }
+        // No detached run to recover (legacy, non-autonomous): a plain disconnect —
+        // terminal notice, no reconnect. (An observer only exists in autonomous mode,
+        // so this is always a local turn.)
+        if (autonomousRunsEnabled !== true) {
           dispatch({
-            type: "RUN_FACT",
-            runFact: { runId: extractRunId(message) ?? "pending" },
+            type: "FINISH_DISCONNECT",
+            hasVisibleContent: false,
+            epoch: stampEpoch,
+          });
+          setStopNotice("disconnect");
+          return;
+        }
+        // A mount-resume OBSERVER (one-shot resume, NOT live-follow) drop falls to
+        // the degraded POLL, which merges by id — it does NOT attach, so there is
+        // nothing to re-seed. #491 tail-only: the anchor row was never removed from
+        // the view (the seed keeps it; the continuation only APPENDED), so nothing to
+        // restore either. The FSM routes this to `polling` (ownership observer,
+        // !liveFollow).
+        if (wasObserver && !machineRef.current.ctx.liveFollow) {
+          queryClient.invalidateQueries({
+            queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current),
           });
           dispatch({
             type: "FINISH_DISCONNECT",
@@ -718,14 +712,123 @@ export default function ChatThread({
             epoch: stampEpoch,
           });
           setStopNotice(null);
-        } else {
+          return;
+        }
+        // We will (re-)ENTER THE RECONNECT LADDER (an attach): a LOCAL live turn's
+        // first drop, OR a live-follow observer's SUBSEQUENT drop (#488 commit 3).
+        // #488 commit 2: recover by the RUN-FACT, not by the presence of an assistant
+        // message — a setup-phase break still leaves a detached run writing to pages.
+        //
+        // #491 tail-only (THE crux): the live store holds a PARTIAL step that is AHEAD
+        // of the persisted boundary; tail-applying the reconnect's step frames over it
+        // would DUPLICATE that partial step. So entering reconnecting is ALWAYS via a
+        // RE-SEED FROM PERSIST — never the live store. Fetch the authoritative
+        // persisted assistant row (`getRun` returns the projected `message`), replace
+        // the live partial by id (mergeById -> the store now holds EXACTLY steps
+        // 0..N-1), and set the anchor to `{ id, n = stepsPersisted }`. Only AFTER the
+        // re-seed is applied do we enter the ladder (FINISH_DISCONNECT schedules the
+        // backoff) — so the attach can never tail-apply over the live partial.
+        const cid = chatIdRef.current;
+        // The live-message runId is the run-fact source (the attach GET keys on
+        // chatId, so a sentinel still recovers a setup-phase break).
+        const runId = extractRunId(message ?? undefined) ?? "pending";
+        const enterReconnect = (fact: string): void => {
+          if (!mountedRef.current) return;
+          // Epoch-stamp the run-fact too (I1): the getRun rtt widens the
+          // onFinish->dispatch window, so a concurrent SEND_LOCAL during it must be
+          // able to drop this stale RUN_FACT (else it clobbers the new turn's
+          // runFact.runId). Consistent with the postRun RUN_FACT stamp.
+          dispatch({ type: "RUN_FACT", runFact: { runId: fact }, epoch: stampEpoch });
           dispatch({
             type: "FINISH_DISCONNECT",
-            hasVisibleContent: false,
+            hasVisibleContent: msgHasVisible,
             epoch: stampEpoch,
           });
-          setStopNotice("disconnect");
+        };
+        // Restore the STRUCTURAL guarantee that the live partial is never the
+        // tail-apply base: drop the live partial from the store by id and null the
+        // anchor, so the reconnect replays from step 0 into a CLEAN store (a full
+        // rebuild) or, past any rotation, 204s -> degraded poll. Used on BOTH the
+        // no-persisted-row and getRun-FAILURE paths — after this there is no path
+        // where the attach tail-applies frames onto a row that already has them
+        // (the #137/#161 duplication class).
+        const dropLivePartialAndReplayFromStart = (): void => {
+          if (message?.role === "assistant" && typeof message.id === "string") {
+            const liveId = message.id;
+            setMessagesRef.current?.((prev) =>
+              prev.filter((m) => m.id !== liveId),
+            );
+          }
+          anchorRef.current = null;
+        };
+        if (cid) {
+          // #541: bound the persist re-seed wait with a timeout race. getRun goes
+          // through the axios client, which has NO request timeout; a HUNG getRun
+          // (connection open, no response) — distinct from a REJECT, which the
+          // `.catch` already handles — would otherwise never let us enter the ladder,
+          // leaving the FSM stuck in `streaming` with no banner and no poll until the
+          // browser socket timeout. `settled` makes the three branches (resolve /
+          // reject / timeout) mutually exclusive: whichever fires FIRST wins and the
+          // others become no-ops. So a LATE getRun resolve AFTER the timeout is fully
+          // ignored — it cannot (i) re-enter reconnect a second time, (ii) overwrite
+          // the timeout's replay-from-start with a stale re-seed, or (iii) re-arm any
+          // timer. On timeout we take the SAME fallback as the reject path (drop the
+          // live partial + enter the ladder, so the poll and the stalled-idle cap arm).
+          let settled = false;
+          const finishReseed = (apply: () => void): void => {
+            if (settled) return;
+            settled = true;
+            if (reseedTimerRef.current) {
+              clearTimeout(reseedTimerRef.current);
+              reseedTimerRef.current = null;
+            }
+            if (!mountedRef.current) return;
+            apply();
+          };
+          reseedTimerRef.current = setTimeout(() => {
+            finishReseed(() => {
+              dropLivePartialAndReplayFromStart();
+              enterReconnect(runId);
+            });
+          }, RECONNECT_RESEED_TIMEOUT_MS);
+          void getRun(cid)
+            .then((res) => {
+              finishReseed(() => {
+                const persisted = res.message;
+                if (persisted && persisted.role === "assistant") {
+                  anchorRef.current = {
+                    id: persisted.id,
+                    stepsPersisted: stepsPersistedOf(persisted),
+                  };
+                  // Replace the live partial with the persisted row IN PLACE by id —
+                  // the re-seed from persist. The attach's tail (steps >= N) then
+                  // appends to a store holding EXACTLY steps 0..N-1: no duplication.
+                  setMessages((prev) => mergeById(prev, rowToUiMessage(persisted)));
+                } else {
+                  // No persisted assistant row (pre-first-frame break): drop the live
+                  // partial + replay from start (no anchor/n) so nothing is duplicated.
+                  dropLivePartialAndReplayFromStart();
+                }
+                enterReconnect(res.run?.id ?? runId);
+              });
+            })
+            .catch(() => {
+              finishReseed(() => {
+                // Persist read FAILED: we cannot re-seed from fresh persist, and a
+                // stale mount-time anchor over the live partial would tail-apply
+                // already-present steps -> duplication (a flaky-network blip:
+                // SSE + getRun both fail, network recovers in ~1s, the registry still
+                // covers from the mount frontier). Restore the removed-filter guarantee
+                // instead: drop the live partial + replay from start / 204 -> poll.
+                dropLivePartialAndReplayFromStart();
+                enterReconnect(runId);
+              });
+            });
+        } else {
+          dropLivePartialAndReplayFromStart();
+          enterReconnect(runId);
         }
+        setStopNotice(null);
         return;
       }
       // A NON-disconnect stream error (a provider 500 etc.) -> terminal error banner.
@@ -746,11 +849,10 @@ export default function ChatThread({
         if (mountedRef.current) {
           const hasVisible = msgHasVisible;
           if (!hasVisible) {
-            // Starved replay: restore the stripped row + poll to the real terminal.
-            if (strippedRowRef.current)
-              setMessages((prev) =>
-                mergeById(prev, rowToUiMessage(strippedRowRef.current!)),
-              );
+            // Starved replay (the tail carried no new steps). #491 tail-only: the
+            // seeded steps 0..N-1 are still on screen (the SDK continuation never
+            // wiped them — `start` does not reset parts), so there is nothing to
+            // restore; just poll to the real terminal.
             queryClient.invalidateQueries({
               queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current),
             });
@@ -846,6 +948,12 @@ export default function ChatThread({
     }
     return () => {
       mountedRef.current = false;
+      // #541: clear the in-flight persist re-seed timeout (not an FSM effect timer,
+      // so DISPOSE does not touch it) — no dangling setTimeout after unmount.
+      if (reseedTimerRef.current) {
+        clearTimeout(reseedTimerRef.current);
+        reseedTimerRef.current = null;
+      }
       dispatch({ type: "DISPOSE" }); // aborts attach + timers, bumps epoch (I5)
     };
     // Mount-only by design; the parent remounts per chat via `key`.
@@ -863,12 +971,12 @@ export default function ChatThread({
     const tail = rows[rows.length - 1];
     if (!tail || tail.role !== "assistant") return;
     setMessages((prev) => mergeById(prev, rowToUiMessage(tail)));
-    // Anchor-mismatch coherence: a restored stripped row A that a DIFFERENT run's
-    // row B has replaced as the tail would linger as an orphan — settle A from
-    // fresh history so no phantom row survives.
-    const stripped = strippedRowRef.current;
-    if (stripped && stripped.id !== tail.id) {
-      const historical = rows.find((r) => r.id === stripped.id);
+    // Anchor-mismatch coherence: if a DIFFERENT run's row B has replaced our anchor
+    // row A as the tail, A would linger as an orphan — reconcile A by id from FRESH
+    // PERSISTED history (not the pinned live row) so no phantom row survives.
+    const anchor = anchorRef.current;
+    if (anchor && anchor.id !== tail.id) {
+      const historical = rows.find((r) => r.id === anchor.id);
       if (historical)
         setMessages((prev) => mergeById(prev, rowToUiMessage(historical)));
     }

apps/client/src/features/ai-chat/services/ai-chat-service.ts

@@ -57,6 +57,31 @@ export async function stopRun(
   return req.data;
 }
 
+/**
+ * Delta poll (#491): the chat's message rows changed since `cursor` (a DB-clock
+ * timestamp echoed from the previous poll) plus the current run fact, in ONE
+ * round-trip — the degraded-poll fallback's payload, replacing the old "refetch
+ * ALL infinite-query pages every 2.5s with full parts" poll. Omit `cursor` on the
+ * first poll (returns just a fresh cursor, no rows, to start the chain). The
+ * overlap window guarantees occasional REPEATS, so the caller MUST merge rows
+ * idempotently by id (mergeById). Owner-gated server-side.
+ */
+export async function getAiChatMessagesDelta(
+  chatId: string,
+  cursor?: string,
+): Promise<{
+  rows: IAiChatMessageRow[];
+  cursor: string;
+  run: { id: string; status: string } | null;
+}> {
+  const req = await api.post<{
+    rows: IAiChatMessageRow[];
+    cursor: string;
+    run: { id: string; status: string } | null;
+  }>("/ai-chat/messages/delta", { chatId, cursor });
+  return req.data;
+}
+
 /**
  * #488: the run-fact — "is a run active on this chat?" — first-class from the
  * server (POST /ai-chat/run). Called on mount to seed the client FSM's run-fact

apps/client/src/features/ai-chat/state/run-fsm.spec.md

@@ -48,6 +48,7 @@ Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`.
 | `RETRY` (manual, stalled banner) | stalled | polling(attach-none) **†** | `[armPoll]` |
 | `POLL_TERMINAL` (settled tail merged) | polling, reconnecting, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4) |
 | `POLL_IDLE_CAP` (inactivity cap) | polling, reconnecting | stalled | `[disarmPoll, cancelReconnect]` (commit 4a — no more silent) |
+| `POLL_IDLE_CAP` (inactivity cap) | stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (Review #4: a Stop-armed poll with no SDK/terminal backstop gets a bounded exit — NOT `stalled`, Stop was already pressed so nothing to retry) |
 | `RUN_FACT{null}` (POST /run → null/terminal, 204) | reconnecting/attaching/polling/stopping | idle | `[cancelReconnect, disarmPoll]`, runFact←null (I3 fresh-negative gate) |
 | `RUN_FACT{runId}` | any | (same) | runFact←runId (pessimism toward an attempt) |
 | `STOP_REQUESTED` (user Stop) | streaming, reconnecting, polling | stopping **†** | `[stopRun, abortAttach, cancelReconnect, armPoll]` (poll drives the terminal — I4 exit by data) |
@@ -121,8 +122,7 @@ holds. **Pending column: empty.**
 | 11 | `stopPendingRef` | **FSM phase `stopping`** | the deferred stop fires from the chat-id adoption effect while `stopping` |
 | 12 | `mountedRef` | **retained (React liveness)** | orthogonal to run-lifecycle; gates imperative onFinish side-effects post-unmount. Epoch (I1) handles stale COMMAND-outcomes; DISPOSE bumps it |
 | 13 | `attemptResumeRef` | **FSM `ATTACH_START` + run-fact** | mount arms attach ONLY on a confirmed active run (commit 4b: streaming-tail status, or POST /run for a user tail) |
-| 14 | `stripRef` | **data** (attachStrategy) | strip+replay detail; the `resumeStream` effect reads it |
-| 15 | `strippedRowRef` | **data** (attachStrategy) | the anchor row |
+| 14–15 | `anchorRef {id, stepsPersisted}` | **data** (attachStrategy) | #491 tail-only: replaced `stripRef`/`strippedRowRef`. The PERSISTED assistant row that pins the run (server invariant 6) + its step frontier N; feeds `?anchor=<id>&n=<stepsPersisted>`. No strip — the seed keeps every row; entering reconnecting re-seeds from persist |
 | 16 | `attachAbortRef` | **effect-owned controller** | aborted by the `abortAttach` effect in cleanup (I5) |
 | 17–25 | `chatIdRef`, `openPageRef`, `getEditorSelectionRef`, `roleIdRef`, `stableIdRef`, `queuedRef`, `sendMessageRef`, `statusRef`, `lastForwardedChatIdRef` | **data** (identity/send mirrors) | unchanged — not lifecycle flags |
 | NEW | `pendingSupersedeRef` | **data** (send-plumbing) | the runId injected into the next `POST /stream {supersede}`; the single replacement for the 3 DELETED one-shots (#8/#9/#10) — net −2 refs |
@@ -151,8 +151,12 @@ message. Sources, in the order they update `ctx.runFact`:
 3. **Attach outcomes:** `ATTACH_LIVE` (2xx) confirms active; a 204 on a non-stripped
    path is an authoritative NEGATIVE fact → the runtime dispatches `RUN_FACT{null}`,
    which cancels recovery (I3 fresh-negative gate).
-4. **Poll (future resume-stack iteration #491):** the delta will carry the run field;
-   until then the poll drives to a terminal ROW, dispatched as `POLL_TERMINAL`.
+4. **Poll (#491, implemented):** the degraded poll now hits the delta endpoint
+   (`POST /ai-chat/messages/delta`), which ALREADY carries the run fact
+   (`run: {id, status} | null`) alongside the changed rows. The client does NOT yet
+   consume that run field — it still drives to a terminal ROW (merged by id),
+   dispatched as `POLL_TERMINAL` — so the run field rides the wire for a future
+   client that settles straight off it.
 
 Pessimism rule: a stale-but-positive fact PERMITS entering recovery (attach); the
 204 then cuts it. A fresh negative fact gates recovery OUT immediately.
@@ -178,6 +182,9 @@ Pessimism rule: a stale-but-positive fact PERMITS entering recovery (attach); th
   /run) are effect-owned and aborted in cleanup (`abortAttach` on `DISPOSE`), not
   render-phase refs. A client abort of an already-sent POST does not cancel the
   server action, so disarming on unmount is safe.
-- **attachStrategy** (strip+replay today) is behind the `resumeStream` effect; the
-  resume-stack iteration (#491) swaps it to tail-only WITHOUT touching the FSM.
+- **attachStrategy** is behind the `resumeStream` effect; #491 swapped it to
+  tail-only (`?anchor=&n=`, `anchorRef` data) WITHOUT touching the FSM. Entering
+  reconnecting always re-seeds from persist; on a getRun failure the live partial
+  is dropped + replay-from-start so it is never the tail-apply base (no #137/#161
+  duplication).
 - **Queue** stays a data structure; flush/interrupt decisions are transitions.

apps/client/src/features/ai-chat/types/ai-chat.types.ts

@@ -181,6 +181,12 @@ export interface IAiChatMessageRow {
   toolCalls?: unknown;
   metadata?: {
     parts?: UIMessage["parts"];
+    // #491 step-alignment anchor: the count of FINISHED steps whose parts are in
+    // THIS row, written atomically with `parts` server-side (flushAssistant). The
+    // resume client reads it as its persisted step frontier N — the tail-only
+    // attach asks the run-stream registry for the frames of step N onward (the
+    // seed already carries steps 0..N-1). Absent on pre-#491 rows -> read as 0.
+    stepsPersisted?: number;
     // AI SDK v6 `totalUsage` persisted on assistant rows. Legacy cumulative
     // figure (sum of every step's usage for the turn); kept for back-compat and
     // as the fallback for older rows that have no `contextTokens`.

apps/client/src/features/ai-chat/utils/resume-helpers.test.ts

@@ -4,7 +4,8 @@ import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.t
 import {
   isStreamingTail,
   isSettledAssistantTail,
-  seedRows,
+  stepsPersistedOf,
+  mergeDeltaRowsIntoPages,
   mergeById,
 } from "./resume-helpers.ts";
 
@@ -12,8 +13,18 @@ function row(
   id: string,
   role: string,
   status?: string,
+  stepsPersisted?: number,
 ): IAiChatMessageRow {
-  return { id, role, content: "", status, createdAt: "2026-01-01T00:00:00Z" };
+  return {
+    id,
+    role,
+    content: "",
+    status,
+    createdAt: "2026-01-01T00:00:00Z",
+    ...(stepsPersisted !== undefined
+      ? { metadata: { stepsPersisted } }
+      : {}),
+  };
 }
 
 function makeMsg(id: string, text: string): UIMessage {
@@ -65,23 +76,92 @@ describe("isSettledAssistantTail", () => {
   });
 });
 
-describe("seedRows", () => {
-  const rows = [row("u1", "user"), row("a1", "assistant", "streaming")];
-
-  it("returns the rows unchanged when not stripping", () => {
-    expect(seedRows(rows, false)).toBe(rows);
+describe("stepsPersistedOf", () => {
+  it("reads metadata.stepsPersisted", () => {
+    expect(stepsPersistedOf(row("a1", "assistant", "streaming", 3))).toBe(3);
+    expect(stepsPersistedOf(row("a1", "assistant", "streaming", 0))).toBe(0);
   });
 
-  it("drops the last row when stripping", () => {
-    const seeded = seedRows(rows, true);
-    expect(seeded).toHaveLength(1);
-    expect(seeded[0].id).toBe("u1");
+  it("defaults to 0 for a pre-#491 row (absent), null/undefined, or a bad value", () => {
+    expect(stepsPersistedOf(row("a1", "assistant", "streaming"))).toBe(0);
+    expect(stepsPersistedOf(null)).toBe(0);
+    expect(stepsPersistedOf(undefined)).toBe(0);
+    expect(
+      stepsPersistedOf({
+        id: "a1",
+        role: "assistant",
+        content: "",
+        createdAt: "x",
+        metadata: { stepsPersisted: -2 },
+      }),
+    ).toBe(0);
   });
 
-  it("returns an empty list when stripping a single-row list", () => {
-    expect(seedRows([row("a1", "assistant", "streaming")], true)).toHaveLength(
-      0,
-    );
+  it("floors a non-integer count", () => {
+    expect(
+      stepsPersistedOf({
+        id: "a1",
+        role: "assistant",
+        content: "",
+        createdAt: "x",
+        metadata: { stepsPersisted: 2.9 },
+      }),
+    ).toBe(2);
+  });
+});
+
+describe("mergeDeltaRowsIntoPages", () => {
+  const pages = () => [
+    { items: [row("u1", "user"), row("a1", "assistant", "streaming", 1)], meta: {} },
+  ];
+
+  it("returns the pages unchanged for an empty delta", () => {
+    const p = pages();
+    expect(mergeDeltaRowsIntoPages(p, [])).toBe(p);
+  });
+
+  it("appends a genuinely new row to the last page in chronological order", () => {
+    const merged = mergeDeltaRowsIntoPages(pages(), [row("a2", "assistant", "streaming", 0)]);
+    expect(merged[0].items.map((i) => i.id)).toEqual(["u1", "a1", "a2"]);
+  });
+
+  it("replaces a grown row in place (per-step growth), never appends a duplicate", () => {
+    const merged = mergeDeltaRowsIntoPages(pages(), [
+      row("a1", "assistant", "streaming", 2),
+    ]);
+    expect(merged[0].items.map((i) => i.id)).toEqual(["u1", "a1"]);
+    // the in-place replacement carries the grown step frontier.
+    expect(stepsPersistedOf(merged[0].items[1])).toBe(2);
+  });
+
+  it("does not mutate the input pages", () => {
+    const input = pages();
+    const before = input[0].items.slice();
+    mergeDeltaRowsIntoPages(input, [row("a2", "assistant", "streaming", 0)]);
+    expect(input[0].items).toEqual(before); // untouched
+  });
+
+  // #491 CONTRACT: the delta overlap window re-delivers the same rows, so merging
+  // MUST be idempotent — applying a delta twice equals applying it once (no growth,
+  // no reorder). A regression re-introduces duplicate assistant bubbles per poll.
+  it("is idempotent: applying the same delta twice equals once", () => {
+    const delta = [
+      row("a1", "assistant", "streaming", 2), // grown existing row
+      row("a2", "assistant", "streaming", 0), // new row
+    ];
+    const once = mergeDeltaRowsIntoPages(pages(), delta);
+    const twice = mergeDeltaRowsIntoPages(once, delta);
+    const thrice = mergeDeltaRowsIntoPages(twice, delta);
+    expect(once[0].items.map((i) => i.id)).toEqual(["u1", "a1", "a2"]);
+    expect(twice[0].items.map((i) => i.id)).toEqual(["u1", "a1", "a2"]);
+    expect(twice).toEqual(once);
+    expect(thrice).toEqual(once);
+  });
+
+  it("seeds a first page when the cache is empty", () => {
+    const merged = mergeDeltaRowsIntoPages([], [row("u1", "user")]);
+    expect(merged).toHaveLength(1);
+    expect(merged[0].items.map((i) => i.id)).toEqual(["u1"]);
   });
 });
 
@@ -109,4 +189,37 @@ describe("mergeById", () => {
     expect(mergeById(prev, null)).toBe(prev);
     expect(mergeById(prev, undefined)).toBe(prev);
   });
+
+  // #491 CONTRACT: the delta poll's overlap window GUARANTEES the same row is
+  // re-delivered across close polls, so merging must be IDEMPOTENT by id — merging
+  // the same row (or an equal-length list of rows) twice must not duplicate or
+  // reorder. This is the property the whole delta-poll design leans on; a
+  // regression here would re-introduce duplicate assistant bubbles on every poll.
+  it("is idempotent by id: re-merging the same row does not duplicate or reorder", () => {
+    const seed = [makeMsg("u1", "hi"), makeMsg("a1", "step 1")];
+    const repeat = makeMsg("a1", "step 1"); // the SAME row the overlap re-delivers
+    const once = mergeById(seed, repeat);
+    const twice = mergeById(once, repeat);
+    const thrice = mergeById(twice, repeat);
+    // Length is stable (no growth), order is stable (user then assistant).
+    expect(once.map((m) => m.id)).toEqual(["u1", "a1"]);
+    expect(twice.map((m) => m.id)).toEqual(["u1", "a1"]);
+    expect(thrice.map((m) => m.id)).toEqual(["u1", "a1"]);
+    // The repeated merge converges: the row is replaced in place, never appended.
+    expect(twice[1]).toBe(repeat);
+  });
+
+  it("is idempotent across a batch of repeated + grown rows (delta re-delivery)", () => {
+    // A delta poll re-delivers a1 (unchanged) and a2 (grown one step). Applying the
+    // batch twice must equal applying it once — the poll can re-send either.
+    const start = [makeMsg("u1", "hi"), makeMsg("a1", "done")];
+    const batch = [makeMsg("a1", "done"), makeMsg("a2", "grown step 2")];
+    const apply = (list: typeof start) =>
+      batch.reduce((acc, row) => mergeById(acc, row), list);
+    const once = apply(start);
+    const twice = apply(once);
+    expect(once.map((m) => m.id)).toEqual(["u1", "a1", "a2"]);
+    expect(twice.map((m) => m.id)).toEqual(["u1", "a1", "a2"]);
+    expect(twice).toEqual(once);
+  });
 });

apps/client/src/features/ai-chat/utils/resume-helpers.ts

@@ -11,9 +11,10 @@ import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.t
 
 /**
  * A STREAMING tail: the last persisted row is an assistant row still marked
- * `status === 'streaming'`. Such a tail is stripped from the seed and rebuilt by
- * the replay (`expect=live`), since the SDK's `text-start` always pushes a new
- * part and replaying over a seeded in-progress row would duplicate its text.
+ * `status === 'streaming'`. #491 (tail-only): such a tail is seeded UNCHANGED —
+ * it carries the persisted steps 0..N-1 — and the run-stream registry's tail
+ * (frames for steps >= N) is APPENDED to it by the SDK's `readUIMessageStream`
+ * continuation. Only the presence of this tail decides WHETHER to attach.
  */
 export function isStreamingTail(rows: IAiChatMessageRow[]): boolean {
   const tail = rows[rows.length - 1];
@@ -32,15 +33,61 @@ export function isSettledAssistantTail(rows: IAiChatMessageRow[]): boolean {
 }
 
 /**
- * Seed rows for `useChat`: return the rows unchanged, or without the last row when
- * `strip` is set (the streaming tail is stripped so the live replay rebuilds it
- * without duplicating parts).
+ * #491 tail-only anchor: the count of FINISHED steps whose parts are persisted in
+ * THIS assistant row (`metadata.stepsPersisted`), written atomically with `parts`
+ * server-side. The resume client reads it as its persisted step frontier N — the
+ * tail-only attach asks the run-stream registry for the frames of step N onward
+ * (the seed already carries steps 0..N-1). Absent on pre-#491 rows => 0.
  */
-export function seedRows(
+export function stepsPersistedOf(
+  row: IAiChatMessageRow | null | undefined,
+): number {
+  const n = row?.metadata?.stepsPersisted;
+  return typeof n === "number" && n >= 0 ? Math.floor(n) : 0;
+}
+
+/** One page of the messages infinite-query cache (`{ items, meta }`). */
+export interface IMessagePage {
+  items: IAiChatMessageRow[];
+  meta: unknown;
+}
+
+/**
+ * #491 delta-poll merge: upsert the delta poll's `rows` into the messages
+ * infinite-query page structure IDEMPOTENTLY by id. The delta endpoint's overlap
+ * window GUARANTEES occasional REPEATS, so this MUST converge: a row already
+ * present is REPLACED IN PLACE (per-step growth of an in-progress row), a new row
+ * is APPENDED to the last page in chronological order (the server returns delta
+ * rows oldest-first). Applying the same delta twice equals applying it once. Never
+ * mutates the input pages (returns fresh page objects with cloned item arrays).
+ */
+export function mergeDeltaRowsIntoPages(
+  pages: IMessagePage[],
   rows: IAiChatMessageRow[],
-  strip: boolean,
-): IAiChatMessageRow[] {
-  return strip ? rows.slice(0, -1) : rows;
+): IMessagePage[] {
+  if (rows.length === 0) return pages;
+  const next: IMessagePage[] = pages.map((p) => ({
+    ...p,
+    items: p.items.slice(),
+  }));
+  const locate = (id: string): [number, number] | null => {
+    for (let pi = 0; pi < next.length; pi++) {
+      const ii = next[pi].items.findIndex((it) => it.id === id);
+      if (ii !== -1) return [pi, ii];
+    }
+    return null;
+  };
+  for (const row of rows) {
+    const at = locate(row.id);
+    if (at) {
+      next[at[0]].items[at[1]] = row; // replace in place — idempotent by id
+    } else if (next.length > 0) {
+      next[next.length - 1].items.push(row); // append chronologically
+    } else {
+      next.push({ items: [row], meta: undefined });
+    }
+  }
+  return next;
 }
 
 /**

apps/client/src/features/ai-chat/utils/sdk-continuation-tripwire.test.ts

@@ -0,0 +1,83 @@
+import { describe, it, expect } from "vitest";
+import { readUIMessageStream, type UIMessage } from "ai";
+import pkg from "../../../../package.json";
+
+/**
+ * PIN-SPEC TRIP-WIRE (#491). The tail-only attach continuation relies on THREE
+ * behaviors of `ai@6.0.207`, verified line-by-line in the issue. Without this
+ * test, an `ai` bump could silently break attach (the client would append the
+ * live tail to the wrong message, or duplicate a step):
+ *
+ *   1. `readUIMessageStream({ message })` CONTINUES the passed message — it does
+ *      not start a fresh one — so the tail streamed after a re-seed is appended to
+ *      the seeded assistant row (the same DB id).
+ *   2. A `start` frame does NOT reset the existing message's parts (so the seeded
+ *      steps 0..N-1 survive; the synthetic `start` the registry prepends only
+ *      carries the run-fact metadata).
+ *   3. Text parts do NOT cross a `finish-step` boundary — a new `text-start` after
+ *      `finish-step` is a NEW part — so the reconstructed steps stay separated and
+ *      the step frontier stays meaningful.
+ *
+ * If an `ai` upgrade changes any of these, this test fails LOUD instead of the
+ * resume path silently corrupting.
+ */
+describe("ai SDK continuation trip-wire (#491, tail-only attach)", () => {
+  it("is pinned to the exact ai version the continuation was verified against", () => {
+    // A caret/range bump is exactly what would silently break attach — require an
+    // exact pin. Bumping ai MUST re-verify the behavior asserted below, then this.
+    expect((pkg as { dependencies: Record<string, string> }).dependencies.ai).toBe(
+      "6.0.207",
+    );
+  });
+
+  it("continues the seeded message: start does not reset parts, the tail appends as new parts", async () => {
+    // A seeded assistant row with ONE finished step already reconstructed.
+    const seeded: UIMessage = {
+      id: "assistant-1",
+      role: "assistant",
+      parts: [
+        { type: "step-start" },
+        { type: "text", text: "STEP0", state: "done" },
+      ],
+    } as UIMessage;
+
+    // The tail the registry delivers on re-attach: a synthetic start (run-fact),
+    // then step 1's frames, then finish. As UI-message chunks (what the SSE frames
+    // decode to).
+    const chunks = [
+      { type: "start", messageMetadata: { runId: "r1", chatId: "c1" } },
+      { type: "start-step" },
+      { type: "text-start", id: "t1" },
+      { type: "text-delta", id: "t1", delta: "STEP1" },
+      { type: "text-end", id: "t1" },
+      { type: "finish-step" },
+      { type: "finish" },
+    ];
+    const stream = new ReadableStream({
+      start(c) {
+        for (const ch of chunks) c.enqueue(ch);
+        c.close();
+      },
+    });
+
+    let last: UIMessage | undefined;
+    for await (const msg of readUIMessageStream({ message: seeded, stream })) {
+      last = msg;
+    }
+
+    expect(last).toBeDefined();
+    // Same message id (continuation, not a fresh message).
+    expect(last!.id).toBe("assistant-1");
+    // The seeded step-0 parts SURVIVED the `start` frame, and step 1 was appended
+    // as SEPARATE parts (text did not cross the finish-step boundary).
+    const shape = last!.parts.map((p) => `${p.type}:${(p as { text?: string }).text ?? ""}`);
+    expect(shape).toEqual([
+      "step-start:",
+      "text:STEP0",
+      "step-start:",
+      "text:STEP1",
+    ]);
+    // The run-fact metadata from the synthetic start frame is applied.
+    expect(last!.metadata).toMatchObject({ runId: "r1", chatId: "c1" });
+  });
+});

apps/client/src/features/editor/atoms/editor-atoms.ts

@@ -3,11 +3,20 @@ import { atom } from "jotai";
 // import would drag the whole @tiptap/core engine into the eager graph of every
 // shell component that reads one of these atoms.
 import type { Editor } from "@tiptap/core";
+import type { HocuspocusProvider } from "@hocuspocus/provider";
 import { PageEditMode } from "@/features/user/types/user.types.ts";
 import type { DictationUnavailableReason } from "@/features/dictation/dictation-status";
 
 export const pageEditorAtom = atom<Editor | null>(null);
 
+// #370 — the active page's collab provider, published by the page editor so the
+// header menu can emit the "save-version" stateless signal (Cmd+S / button).
+// Null when the page is read-only / collab isn't connected. A typed initial
+// value (rather than an explicit generic) keeps jotai's overload resolution on
+// the writable PrimitiveAtom branch.
+const initialCollabProvider: HocuspocusProvider | null = null;
+export const collabProviderAtom = atom(initialCollabProvider);
+
 export const titleEditorAtom = atom<Editor | null>(null);
 
 export const readOnlyEditorAtom = atom<Editor | null>(null);

apps/client/src/features/editor/page-editor.tsx

@@ -31,11 +31,18 @@ import { useAtom, useAtomValue, useSetAtom } from "jotai";
 import useCollaborationUrl from "@/features/editor/hooks/use-collaboration-url";
 import { currentUserAtom } from "@/features/user/atoms/current-user-atom";
 import {
+  collabProviderAtom,
   currentPageEditModeAtom,
   dictationAvailabilityAtom,
   pageEditorAtom,
   yjsConnectionStatusAtom,
 } from "@/features/editor/atoms/editor-atoms";
+import { notifications } from "@mantine/notifications";
+import {
+  VERSION_SAVED_MESSAGE_TYPE,
+  type VersionSavedMessage,
+  saveVersionPending,
+} from "@/features/page-history/version-messages";
 import { asideStateAtom } from "@/components/layouts/global/hooks/atoms/sidebar-atom";
 import {
   activeCommentIdAtom,
@@ -124,6 +131,7 @@ export default function PageEditor({
 
   const [currentUser] = useAtom(currentUserAtom);
   const [, setEditor] = useAtom(pageEditorAtom);
+  const setCollabProvider = useSetAtom(collabProviderAtom);
   const [, setAsideState] = useAtom(asideStateAtom);
   const [, setActiveCommentId] = useAtom(activeCommentIdAtom);
   const [showCommentPopup, setShowCommentPopup] = useAtom(showCommentPopupAtom);
@@ -181,6 +189,24 @@ export default function PageEditor({
       const onStatelessHandler = ({ payload }: onStatelessParameters) => {
         try {
           const message = JSON.parse(payload);
+          // #370 — a version was saved somewhere; live-refresh the history panel
+          // on every client. Only the client that pressed Save (tracked by the
+          // module-level flag) shows the confirmation toast.
+          if (message?.type === VERSION_SAVED_MESSAGE_TYPE) {
+            const versionMsg = message as VersionSavedMessage;
+            queryClient.invalidateQueries({
+              queryKey: ["page-history-list"],
+            });
+            if (saveVersionPending.current) {
+              saveVersionPending.current = false;
+              notifications.show({
+                message: versionMsg.alreadySaved
+                  ? t("Already saved as the latest version")
+                  : t("Version saved"),
+              });
+            }
+            return;
+          }
           if (message?.type !== "page.updated" || !message.updatedAt) return;
           const pageData = queryClient.getQueryData<IPage>(["pages", slugId]);
           if (pageData) {
@@ -238,12 +264,16 @@ export default function PageEditor({
 
       local.on("synced", onLocalSyncedHandler);
       providersRef.current = { socket, local, remote };
+      // #370 — publish the provider so the header menu can emit save-version.
+      setCollabProvider(remote);
       setProvidersReady(true);
     } else {
+      setCollabProvider(providersRef.current.remote);
       setProvidersReady(true);
     }
     // Only destroy on final unmount
     return () => {
+      setCollabProvider(null);
       providersRef.current?.socket.destroy();
       providersRef.current?.remote.destroy();
       providersRef.current?.local.destroy();

apps/client/src/features/page-history/components/history-item.tsx

@@ -1,4 +1,11 @@
-import { Text, Group, UnstyledButton, Avatar, Tooltip } from "@mantine/core";
+import {
+  Text,
+  Group,
+  UnstyledButton,
+  Avatar,
+  Tooltip,
+  Badge,
+} from "@mantine/core";
 import { CustomAvatar } from "@/components/ui/custom-avatar.tsx";
 import { AgentAvatarStack } from "@/components/ui/agent-avatar-stack.tsx";
 import { formattedDate } from "@/lib/time";
@@ -7,36 +14,59 @@ import clsx from "clsx";
 import { IPageHistory } from "@/features/page-history/types/page.types";
 import { memo, useCallback } from "react";
 import { useSetAtom } from "jotai";
+import { useTranslation } from "react-i18next";
 import { historyAtoms } from "@/features/page-history/atoms/history-atoms.ts";
 
 const MAX_VISIBLE_AVATARS = 5;
 
+/**
+ * #370 — map a snapshot's intentionality tier to its badge. `version: true`
+ * marks the intentional points (manual / agent); autosaves (boundary / idle /
+ * legacy null) are non-versions and get dimmed in the list.
+ */
+type HistoryKindMeta = { labelKey: string; color: string; version: boolean };
+export function historyKindMeta(kind?: string | null): HistoryKindMeta {
+  switch (kind) {
+    case "manual":
+      return { labelKey: "Saved", color: "blue", version: true };
+    case "agent":
+      return { labelKey: "Agent version", color: "violet", version: true };
+    case "boundary":
+      return { labelKey: "Boundary", color: "gray", version: false };
+    default: // "idle" | null | undefined (legacy autosave)
+      return { labelKey: "Autosave", color: "gray", version: false };
+  }
+}
+
 interface HistoryItemProps {
   historyItem: IPageHistory;
-  index: number;
-  onSelect: (id: string, index: number) => void;
-  onHover?: (id: string, index: number) => void;
+  // The previous snapshot for diff/restore is resolved by id from the FULL list
+  // in the parent (resolvePrevSnapshotId), so the item only needs to report its
+  // own id — never a list index (which would be the filtered-view index).
+  onSelect: (id: string) => void;
+  onHover?: (id: string) => void;
   onHoverEnd?: () => void;
   isActive: boolean;
 }
 
 const HistoryItem = memo(function HistoryItem({
   historyItem,
-  index,
   onSelect,
   onHover,
   onHoverEnd,
   isActive,
 }: HistoryItemProps) {
   const setHistoryModalOpen = useSetAtom(historyAtoms);
+  const { t } = useTranslation();
+  const kindMeta = historyKindMeta(historyItem.kind);
 
   const handleClick = useCallback(() => {
-    onSelect(historyItem.id, index);
-  }, [onSelect, historyItem.id, index]);
+    onSelect(historyItem.id);
+  }, [onSelect, historyItem.id]);
 
   const handleMouseEnter = useCallback(() => {
-    onHover?.(historyItem.id, index);
-  }, [onHover, historyItem.id, index]);
+    onHover?.(historyItem.id);
+  }, [onHover, historyItem.id]);
 
   const contributors = historyItem.contributors;
   const hasContributors = contributors && contributors.length > 0;
@@ -49,8 +79,20 @@ const HistoryItem = memo(function HistoryItem({
       onMouseEnter={handleMouseEnter}
       onMouseLeave={onHoverEnd}
       className={clsx(classes.history, { [classes.active]: isActive })}
+      // #370 — dim autosnapshots so intentional versions stand out.
+      style={{ opacity: kindMeta.version ? 1 : 0.55 }}
     >
-      <Text size="sm">{formattedDate(new Date(historyItem.createdAt))}</Text>
+      <Group gap={6} wrap="nowrap" justify="space-between">
+        <Text size="sm">{formattedDate(new Date(historyItem.createdAt))}</Text>
+        <Badge
+          size="xs"
+          radius="sm"
+          variant={kindMeta.version ? "filled" : "light"}
+          color={kindMeta.color}
+        >
+          {t(kindMeta.labelKey)}
+        </Badge>
+      </Group>
 
       <Group gap={6} wrap="nowrap" mt={4}>
         {hasContributors ? (

apps/client/src/features/page-history/components/history-list.tsx

@@ -2,14 +2,16 @@ import {
   usePageHistoryListQuery,
   prefetchPageHistory,
 } from "@/features/page-history/queries/page-history-query";
-import HistoryItem from "@/features/page-history/components/history-item";
+import HistoryItem, {
+  historyKindMeta,
+} from "@/features/page-history/components/history-item";
 import {
   activeHistoryIdAtom,
   activeHistoryPrevIdAtom,
   historyAtoms,
 } from "@/features/page-history/atoms/history-atoms";
 import { useAtom, useSetAtom } from "jotai";
-import { useCallback, useEffect, useMemo, useRef } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import {
   Button,
   ScrollArea,
@@ -17,9 +19,12 @@ import {
   Divider,
   Loader,
   Center,
+  Switch,
+  Text,
 } from "@mantine/core";
 import { useTranslation } from "react-i18next";
 import { useHistoryRestore } from "@/features/page-history/hooks";
+import { resolvePrevSnapshotId } from "@/features/page-history/utils/resolve-prev-snapshot";
 
 const PREFETCH_DELAY_MS = 150;
 
@@ -47,6 +52,22 @@ function HistoryList({ pageId }: Props) {
     [pageHistoryData],
   );
 
+  // #370 — "only versions" filter: hide autosnapshots (idle/boundary/legacy
+  // null), keep only intentional points (manual/agent). Filtering is over the
+  // already-loaded pages; the diff/restore still targets the true previous
+  // snapshot, so items carry their index within the FULL list.
+  const [onlyVersions, setOnlyVersions] = useState(false);
+  // Reuse historyKindMeta().version — the SAME predicate the badge (HistoryItem)
+  // uses to mark intentional points — so the "Only versions" filter and the badge
+  // can never drift apart when a future intentional kind is added.
+  const visibleItems = useMemo(
+    () =>
+      onlyVersions
+        ? historyItems.filter((item) => historyKindMeta(item.kind).version)
+        : historyItems,
+    [historyItems, onlyVersions],
+  );
+
   const loadMoreRef = useRef<HTMLDivElement>(null);
   const prefetchTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
 
@@ -60,11 +81,13 @@ function HistoryList({ pageId }: Props) {
   }, []);
 
   const handleHover = useCallback(
-    (historyId: string, index: number) => {
+    (historyId: string) => {
       clearPrefetchTimeout();
       prefetchTimeoutRef.current = setTimeout(() => {
         prefetchPageHistory(historyId);
-        const prevId = historyItems[index + 1]?.id;
+        // The true previous snapshot in the FULL list (not the previous visible
+        // one under the "only versions" filter).
+        const prevId = resolvePrevSnapshotId(historyItems, historyId);
         if (prevId) {
           prefetchPageHistory(prevId);
         }
@@ -78,9 +101,11 @@ function HistoryList({ pageId }: Props) {
   }, [clearPrefetchTimeout]);
 
   const handleSelect = useCallback(
-    (id: string, index: number) => {
+    (id: string) => {
       setActiveHistoryId(id);
-      setActiveHistoryPrevId(historyItems[index + 1]?.id ?? "");
+      // Baseline = true previous snapshot in the FULL list, so the "only
+      // versions" filter never diffs/restores against the wrong item.
+      setActiveHistoryPrevId(resolvePrevSnapshotId(historyItems, id));
     },
     [historyItems, setActiveHistoryId, setActiveHistoryPrevId],
   );
@@ -128,12 +153,27 @@ function HistoryList({ pageId }: Props) {
 
   return (
     <div>
+      <Group px="xs" py={6} justify="flex-end">
+        <Switch
+          size="xs"
+          checked={onlyVersions}
+          onChange={(e) => setOnlyVersions(e.currentTarget.checked)}
+          label={t("Only versions")}
+        />
+      </Group>
+
       <ScrollArea h={620} w="100%" type="scroll" scrollbarSize={5}>
-        {historyItems.map((historyItem, index) => (
+        {onlyVersions && visibleItems.length === 0 && (
+          <Center py="md">
+            <Text size="sm" c="dimmed">
+              {t("No saved versions yet.")}
+            </Text>
+          </Center>
+        )}
+        {visibleItems.map((historyItem) => (
           <HistoryItem
             key={historyItem.id}
             historyItem={historyItem}
-            index={index}
             onSelect={handleSelect}
             onHover={handleHover}
             onHoverEnd={clearPrefetchTimeout}

apps/client/src/features/page-history/types/page.types.ts

@@ -24,6 +24,10 @@ export interface IPageHistory {
   updatedAt: string;
   lastUpdatedBy: IPageHistoryUser;
   contributors?: IPageHistoryUser[];
+  // #370 — intentionality tier: 'manual'/'agent' are versions (intentional
+  // points), 'idle'/'boundary' are autosnapshots; null/undefined = legacy
+  // autosave. Derived server-side, drives the history badge + "versions" filter.
+  kind?: "manual" | "agent" | "idle" | "boundary" | null;
   // Provenance markers copied off the page row when the snapshot was saved.
   // `'agent'` marks a version written by the AI agent; `lastUpdatedAiChatId`
   // (when present) deep-links to the chat that produced the edit.

apps/client/src/features/page-history/utils/resolve-prev-snapshot.test.ts

@@ -0,0 +1,42 @@
+import { describe, it, expect } from "vitest";
+import { resolvePrevSnapshotId } from "./resolve-prev-snapshot";
+
+// #370 F4 — the risky client path: with the "only versions" filter active, diff
+// and restore must still baseline against the TRUE previous snapshot in the FULL
+// list, never the previous VISIBLE version (which would skip the autosnapshots
+// between two versions). These pin that the resolution is by FULL-list order.
+describe("resolvePrevSnapshotId", () => {
+  // Newest-first, as the history list stores it: a version, then two autosaves,
+  // then an older version.
+  const full = [
+    { id: "v2", kind: "manual" },
+    { id: "a2", kind: "idle" },
+    { id: "a1", kind: "boundary" },
+    { id: "v1", kind: "manual" },
+    { id: "a0", kind: null },
+  ];
+
+  it("returns the immediate FULL-list successor, not the previous visible version", () => {
+    // Selecting v2 while filtered to versions-only must baseline against a2 (the
+    // real chronological predecessor), NOT v1 (the previous visible version).
+    expect(resolvePrevSnapshotId(full, "v2")).toBe("a2");
+  });
+
+  it("resolves an autosnapshot's predecessor by full-list order", () => {
+    expect(resolvePrevSnapshotId(full, "a1")).toBe("v1");
+  });
+
+  it("returns '' for the oldest item (no predecessor)", () => {
+    expect(resolvePrevSnapshotId(full, "a0")).toBe("");
+  });
+
+  it("returns '' for an id not in the list", () => {
+    expect(resolvePrevSnapshotId(full, "missing")).toBe("");
+  });
+
+  it("does not depend on a filtered subset — same result whatever is visible", () => {
+    // The helper only ever sees the full list; a filtered view cannot change the
+    // baseline it computes.
+    expect(resolvePrevSnapshotId(full, "v1")).toBe("a0");
+  });
+});

apps/client/src/features/page-history/utils/resolve-prev-snapshot.ts

@@ -0,0 +1,22 @@
+/**
+ * #370 — resolve the TRUE previous snapshot for a history item.
+ *
+ * The history panel can be filtered to "only versions" (manual/agent), but diff
+ * and restore must always compare against the immediately-preceding snapshot in
+ * the FULL, unfiltered list — NOT the previous VISIBLE item. Comparing against
+ * the previous visible version would silently skip the autosnapshots between two
+ * versions and diff/restore the wrong baseline.
+ *
+ * Given the full (newest-first) list and an item id, this returns the id of the
+ * item right after it in the full list (its chronological predecessor), or "" if
+ * it is the oldest / not found. Pure and list-order-preserving so it can be unit
+ * tested without mounting the component.
+ */
+export function resolvePrevSnapshotId(
+  fullItems: ReadonlyArray<{ id: string }>,
+  id: string,
+): string {
+  const index = fullItems.findIndex((item) => item.id === id);
+  if (index === -1) return "";
+  return fullItems[index + 1]?.id ?? "";
+}

apps/client/src/features/page-history/version-messages.ts

@@ -0,0 +1,28 @@
+/**
+ * #370 — page-version stateless wire formats. Kept in one place so the client
+ * emitter (Save hotkey / button) and the client listener (page-editor) agree
+ * with the server (PersistenceExtension) on the message shapes.
+ */
+
+/** Client → server: "save a version now". The server derives the tier
+ * (manual/agent) from the signed connection actor, never from this payload. */
+export const SAVE_VERSION_MESSAGE_TYPE = "save-version";
+
+/** Server → all clients: a version was saved (or promoted / already existed). */
+export const VERSION_SAVED_MESSAGE_TYPE = "version.saved";
+
+export interface VersionSavedMessage {
+  type: typeof VERSION_SAVED_MESSAGE_TYPE;
+  historyId: string;
+  kind: "manual" | "agent";
+  /** True when the latest snapshot was already a manual version (a no-op save). */
+  alreadySaved: boolean;
+}
+
+/**
+ * Cross-component coordination flag so only the client that pressed Save shows
+ * the confirmation toast, while every other client silently refreshes its
+ * history panel on the broadcast. A module-level ref avoids stale-closure
+ * pitfalls in the editor's long-lived stateless handler.
+ */
+export const saveVersionPending = { current: false };

apps/client/src/features/page/components/header/page-header-menu.tsx

@@ -3,6 +3,7 @@ import {
   IconArrowRight,
   IconArrowsHorizontal,
   IconClockHour4,
+  IconDeviceFloppy,
   IconDots,
   IconEye,
   IconEyeOff,
@@ -17,7 +18,7 @@ import {
   IconTrash,
   IconWifiOff,
 } from "@tabler/icons-react";
-import React, { useEffect, useRef, useState } from "react";
+import React, { useCallback, useEffect, useRef, useState } from "react";
 import { useAsideTriggerProps } from "@/hooks/use-toggle-aside.tsx";
 import { useAtom, useAtomValue } from "jotai";
 import { historyAtoms } from "@/features/page-history/atoms/history-atoms.ts";
@@ -39,9 +40,14 @@ import { Trans, useTranslation } from "react-i18next";
 import ExportModal from "@/components/common/export-modal";
 import { convertProseMirrorToMarkdown } from "@docmost/prosemirror-markdown/browser";
 import {
+  collabProviderAtom,
   pageEditorAtom,
   yjsConnectionStatusAtom,
 } from "@/features/editor/atoms/editor-atoms.ts";
+import {
+  SAVE_VERSION_MESSAGE_TYPE,
+  saveVersionPending,
+} from "@/features/page-history/version-messages.ts";
 import { formattedDate } from "@/lib/time.ts";
 import { PageEditModeToggle } from "@/features/user/components/page-state-pref.tsx";
 import MovePageModal from "@/features/page/components/move-page-modal.tsx";
@@ -72,9 +78,34 @@ export default function PageHeaderMenu({ readOnly }: PageHeaderMenuProps) {
   });
   const isDeleted = !!page?.deletedAt;
   const [workspace] = useAtom(workspaceAtom);
+  const collabProvider = useAtomValue(collabProviderAtom);
   // Community public-sharing entry point (replaces the removed EE PageShareModal)
   const workspaceSharingDisabled = workspace?.settings?.sharing?.disabled === true;
 
+  // #370 — explicit "save a version" (Cmd+S / Save button). One path for the
+  // human; the server derives the tier from the signed actor. Readers can't save
+  // (the button is hidden and the collab connection is read-only server-side).
+  const handleSaveVersion = useCallback(() => {
+    if (readOnly || !collabProvider) return;
+    // Flag this client as the initiator so only it shows the confirmation toast;
+    // a safety timeout clears it if no broadcast comes back (e.g. offline).
+    saveVersionPending.current = true;
+    window.setTimeout(() => {
+      saveVersionPending.current = false;
+    }, 5000);
+    collabProvider.sendStateless(
+      JSON.stringify({ type: SAVE_VERSION_MESSAGE_TYPE }),
+    );
+  }, [readOnly, collabProvider]);
+
+  // mod+S must also block the browser's "Save page" dialog. `triggerOnContent-
+  // Editable` + empty ignore-list so it fires while typing in the editor/title.
+  useHotkeys(
+    [["mod+S", handleSaveVersion, { preventDefault: true }]],
+    [],
+    true,
+  );
+
   useHotkeys(
     [
       [
@@ -133,15 +164,16 @@ export default function PageHeaderMenu({ readOnly }: PageHeaderMenuProps) {
         </ActionIcon>
       </Tooltip>
 
-      <PageActionMenu readOnly={readOnly} />
+      <PageActionMenu readOnly={readOnly} onSaveVersion={handleSaveVersion} />
     </>
   );
 }
 
 interface PageActionMenuProps {
   readOnly?: boolean;
+  onSaveVersion?: () => void;
 }
-function PageActionMenu({ readOnly }: PageActionMenuProps) {
+function PageActionMenu({ readOnly, onSaveVersion }: PageActionMenuProps) {
   const { t } = useTranslation();
   const [, setHistoryModalOpen] = useAtom(historyAtoms);
   const clipboard = useClipboard({ timeout: 500 });
@@ -303,6 +335,20 @@ function PageActionMenu({ readOnly }: PageActionMenuProps) {
             </Group>
           </Menu.Item>
 
+          {!readOnly && (
+            <Menu.Item
+              leftSection={<IconDeviceFloppy size={16} />}
+              onClick={onSaveVersion}
+              rightSection={
+                <Text size="xs" c="dimmed">
+                  {t("Ctrl+S")}
+                </Text>
+              }
+            >
+              {t("Save version")}
+            </Menu.Item>
+          )}
+
           <Menu.Item
             leftSection={<IconHistory size={16} />}
             onClick={openHistoryModal}

apps/server/src/collaboration/constants.ts

@@ -1,10 +1,36 @@
-export const HISTORY_INTERVAL = 5 * 60 * 1000;
-export const HISTORY_FAST_INTERVAL = 60 * 1000;
-export const HISTORY_FAST_THRESHOLD = 5 * 60 * 1000;
-
 // #348 — debounce window for the per-page RAG re-embed job. Repeated saves
 // within this window collapse to a single delayed job (coalesced by a stable
 // jobId), so active editing does not pile up expensive re-embeds (external API
 // + page_embeddings rewrite, concurrency 1). The worker reads the CURRENT page
 // state at run time, so the last content within the window wins.
 export const EMBED_DEBOUNCE_MS = 30 * 1000;
+
+/**
+ * #370 — page-history intentionality tiers. Domain of `page_history.kind`.
+ *   - 'manual' / 'agent'  → Tier 1 versions (intentional points)
+ *   - 'idle' / 'boundary' → Tier 0 autosnapshots (safety net)
+ * A legacy `null` kind is treated as an autosave.
+ */
+export type PageHistoryKind = 'manual' | 'agent' | 'idle' | 'boundary';
+
+/**
+ * #370 — trailing idle-flush windows. A page's pending idle snapshot is
+ * re-armed on every store and fires this long after edits go quiet, so a burst
+ * of edits collapses into a single autosnapshot instead of one-per-store. Human
+ * sessions are noisier and less risky, so they flush less often than the agent.
+ */
+export const IDLE_INTERVAL_USER = 60 * 60 * 1000; // 60m
+export const IDLE_INTERVAL_AGENT = 15 * 60 * 1000; // 15m
+
+/**
+ * #370 — max-wait ceiling for the idle flush. Pure trailing debounce starves the
+ * safety net: hocuspocus stores at least every ~45s, so a CONTINUOUS editing
+ * session would re-arm the trailing timer forever and never take an idle
+ * snapshot until edits finally go quiet (up to IDLE_INTERVAL_USER = 60m). This
+ * ceiling bounds the actual wait from the FIRST edit of a burst, so an idle
+ * snapshot fires at least this often during a long unbroken session — restoring
+ * a recovery point cadence closer to the old heuristic without one-per-store
+ * noise. Mirrors hocuspocus's own maxDebounce idea.
+ */
+export const IDLE_MAX_WAIT_USER = 10 * 60 * 1000; // 10m
+export const IDLE_MAX_WAIT_AGENT = 5 * 60 * 1000; // 5m

apps/server/src/collaboration/extensions/compute-history-job.spec.ts

@@ -1,84 +1,93 @@
+import { computeHistoryJob, resolveSource } from './persistence.extension';
 import {
-  computeHistoryJob,
-  resolveSource,
-} from './persistence.extension';
-import {
-  HISTORY_FAST_INTERVAL,
-  HISTORY_FAST_THRESHOLD,
-  HISTORY_INTERVAL,
+  IDLE_INTERVAL_AGENT,
+  IDLE_INTERVAL_USER,
+  IDLE_MAX_WAIT_AGENT,
+  IDLE_MAX_WAIT_USER,
 } from '../constants';
 
-// A fixed clock + fixed createdAt make pageAge deterministic.
-const NOW = 1_700_000_000_000;
 const PAGE_ID = '550e8400-e29b-41d4-a716-446655440000';
 
-// Build a minimal page whose age (NOW - createdAt) is exactly `ageMs`.
-const pageAged = (ageMs: number) => ({
-  id: PAGE_ID,
-  createdAt: new Date(NOW - ageMs),
-});
+const page = { id: PAGE_ID };
 
-describe('computeHistoryJob', () => {
-  it('agent edit → delay MUST be 0 and job id is source-keyed', () => {
-    // INVARIANT (§15 H2 / persistence.extension): the agent delay MUST stay 0.
-    // The worker re-reads the page row at run time, so any non-zero delay risks
-    // snapshotting content a later human edit has already overwritten. This is
-    // the load-bearing assertion of this spec — do not relax it.
-    const { jobId, delay } = computeHistoryJob(pageAged(0), 'agent', NOW);
-    expect(delay).toBe(0);
-    expect(jobId).toBe(`${PAGE_ID}-agent`);
-  });
-
-  it('agent edit on an OLD page is still delay 0 (age never applies to agents)', () => {
-    // Even when the page is far older than the fast threshold, the agent path
-    // must short-circuit to 0 — age-based debounce is a human-only concern.
-    const { jobId, delay } = computeHistoryJob(
-      pageAged(HISTORY_FAST_THRESHOLD + 60_000),
-      'agent',
-      NOW,
-    );
-    expect(delay).toBe(0);
-    expect(jobId).toBe(`${PAGE_ID}-agent`);
-  });
-
-  it('human edit on a YOUNG page (age < threshold) → fast interval, bare job id', () => {
-    const { jobId, delay } = computeHistoryJob(
-      pageAged(HISTORY_FAST_THRESHOLD - 1),
-      'user',
-      NOW,
-    );
-    expect(delay).toBe(HISTORY_FAST_INTERVAL);
+describe('computeHistoryJob (#370 — shared trailing idle pipeline)', () => {
+  it('human edit → user idle window, bare page.id job', () => {
+    // Humans and the agent now share ONE idle job per page (jobId = page.id).
+    // The agent's old delay=0 fast path is GONE — intentional agent points now
+    // arrive via the explicit save-version signal, not a zero-delay snapshot.
+    const { jobId, delay } = computeHistoryJob(page, 'user');
+    expect(delay).toBe(IDLE_INTERVAL_USER);
     expect(jobId).toBe(PAGE_ID);
   });
 
-  it('human edit on an OLD page (age > threshold) → standard interval', () => {
-    const { jobId, delay } = computeHistoryJob(
-      pageAged(HISTORY_FAST_THRESHOLD + 1),
-      'user',
-      NOW,
-    );
-    expect(delay).toBe(HISTORY_INTERVAL);
+  it('agent edit → agent idle window (shorter), still the bare page.id job', () => {
+    const { jobId, delay } = computeHistoryJob(page, 'agent');
+    expect(delay).toBe(IDLE_INTERVAL_AGENT);
+    // No `-agent` suffix anymore: the agent joins the common idle pipeline.
     expect(jobId).toBe(PAGE_ID);
   });
 
-  it('boundary: pageAge EXACTLY === threshold takes the slow branch (the `<` is strict)', () => {
-    // Off-by-one guard: the condition is `pageAge < HISTORY_FAST_THRESHOLD`, so
-    // an age of exactly the threshold is NOT "fast" — it must use HISTORY_INTERVAL.
-    const { delay } = computeHistoryJob(
-      pageAged(HISTORY_FAST_THRESHOLD),
-      'user',
-      NOW,
-    );
-    expect(delay).toBe(HISTORY_INTERVAL);
+  it('agent flushes sooner than a human', () => {
+    expect(IDLE_INTERVAL_AGENT).toBeLessThan(IDLE_INTERVAL_USER);
   });
 
-  it('treats any non-"agent" source string as human', () => {
-    // resolveSource only ever yields 'agent' | 'user', but guard the contract:
-    // the agent branch keys strictly on === 'agent'.
-    const { jobId, delay } = computeHistoryJob(pageAged(0), 'user', NOW);
-    expect(delay).toBe(HISTORY_FAST_INTERVAL);
+  it('treats any non-"agent" source string as human (keys strictly on === agent)', () => {
+    const { jobId, delay } = computeHistoryJob(page, 'user');
+    expect(delay).toBe(IDLE_INTERVAL_USER);
     expect(jobId).toBe(PAGE_ID);
   });
+
+  // #370 review round-1 WARNING: the max-wait ceiling prevents autosnapshot
+  // starvation during a continuous editing session (the trailing timer would
+  // otherwise re-arm forever and never fire).
+  describe('max-wait ceiling', () => {
+    const T0 = 1_000_000; // arbitrary fixed epoch for deterministic tests
+
+    it('once a burst is armed, delay clamps to the remaining max-wait budget', () => {
+      // 1 minute into the burst the USER interval (60m) far exceeds the remaining
+      // max-wait budget (10m - 1m = 9m), so the delay is clamped DOWN to that
+      // remaining budget — the full interval is NOT used once a ceiling applies.
+      const { delay } = computeHistoryJob(page, 'user', T0, T0 + 60_000);
+      expect(delay).toBe(IDLE_MAX_WAIT_USER - 60_000);
+    });
+
+    it('never waits longer than the max-wait budget from the burst start', () => {
+      // A store arriving right at the ceiling → delay 0 (fire promptly).
+      const { delay } = computeHistoryJob(
+        page,
+        'user',
+        T0,
+        T0 + IDLE_MAX_WAIT_USER,
+      );
+      expect(delay).toBe(0);
+    });
+
+    it('past the ceiling never returns a negative delay', () => {
+      const { delay } = computeHistoryJob(
+        page,
+        'user',
+        T0,
+        T0 + IDLE_MAX_WAIT_USER + 5 * 60_000,
+      );
+      expect(delay).toBe(0);
+    });
+
+    it('the agent ceiling is shorter than the user ceiling', () => {
+      expect(IDLE_MAX_WAIT_AGENT).toBeLessThan(IDLE_MAX_WAIT_USER);
+      const { delay } = computeHistoryJob(
+        page,
+        'agent',
+        T0,
+        T0 + IDLE_MAX_WAIT_AGENT,
+      );
+      expect(delay).toBe(0);
+    });
+
+    it('without a burstStart there is no ceiling (backward-compatible)', () => {
+      expect(computeHistoryJob(page, 'user').delay).toBe(IDLE_INTERVAL_USER);
+      expect(computeHistoryJob(page, 'agent').delay).toBe(IDLE_INTERVAL_AGENT);
+    });
+  });
 });
 
 describe('resolveSource (truth table)', () => {

apps/server/src/collaboration/extensions/persistence-store.spec.ts

@@ -40,11 +40,12 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
   let pageHistoryRepo: {
     saveHistory: jest.Mock;
     findPageLastHistory: jest.Mock;
+    updateHistoryKind: jest.Mock;
   };
   let aiQueue: { add: jest.Mock };
-  let historyQueue: { add: jest.Mock };
+  let historyQueue: { add: jest.Mock; remove: jest.Mock };
   let notificationQueue: { add: jest.Mock };
-  let collabHistory: { addContributors: jest.Mock };
+  let collabHistory: { addContributors: jest.Mock; popContributors: jest.Mock };
   let transclusionService: {
     syncPageTransclusions: jest.Mock;
     syncPageReferences: jest.Mock;
@@ -93,13 +94,22 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
     pageHistoryRepo = {
       saveHistory: jest.fn().mockImplementation(async () => {
         callOrder.push('saveHistory');
+        return { id: 'history-1' };
       }),
       findPageLastHistory: jest.fn().mockResolvedValue(null),
+      updateHistoryKind: jest.fn().mockResolvedValue(undefined),
     };
     aiQueue = { add: jest.fn().mockResolvedValue(undefined) };
-    historyQueue = { add: jest.fn().mockResolvedValue(undefined) };
+    historyQueue = {
+      add: jest.fn().mockResolvedValue(undefined),
+      // #370 — enqueuePageHistory now removes any pending idle job before re-adding.
+      remove: jest.fn().mockResolvedValue(undefined),
+    };
     notificationQueue = { add: jest.fn().mockResolvedValue(undefined) };
-    collabHistory = { addContributors: jest.fn().mockResolvedValue(undefined) };
+    collabHistory = {
+      addContributors: jest.fn().mockResolvedValue(undefined),
+      popContributors: jest.fn().mockResolvedValue([]),
+    };
     transclusionService = {
       syncPageTransclusions: jest.fn().mockResolvedValue(undefined),
       syncPageReferences: jest.fn().mockResolvedValue(undefined),
@@ -165,6 +175,50 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
     expect(pageRepo.updatePage.mock.calls[0][0].lastUpdatedSource).toBe('user');
   });
 
+  // #370 review round-1 SUGGESTION: the boundary was GENERALIZED from a
+  // user→agent special-case to ANY lastUpdatedSource transition. These pin the
+  // generalized behaviour it was rebuilt for.
+  describe('generalized boundary — any source transition', () => {
+    // Same persisted page but with an explicit prior source.
+    const pageWithPriorSource = (prior: string | null) => ({
+      ...persistedHumanPage('NEW CONTENT'),
+      lastUpdatedSource: prior,
+    });
+
+    it('agent→user transition fires the boundary (pins the prior agent revision)', async () => {
+      const document = ydocFor(doc('NEW CONTENT'));
+      pageRepo.findById.mockResolvedValue(pageWithPriorSource('agent'));
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+
+      await ext.onStoreDocument(buildData(document, 'user') as any);
+
+      expect(pageHistoryRepo.saveHistory).toHaveBeenCalledTimes(1);
+      expect(callOrder).toEqual(['saveHistory', 'updatePage']);
+      expect(pageRepo.updatePage.mock.calls[0][0].lastUpdatedSource).toBe('user');
+    });
+
+    it('git→user transition fires the boundary (git-sync overwrite is a source change)', async () => {
+      const document = ydocFor(doc('NEW CONTENT'));
+      pageRepo.findById.mockResolvedValue(pageWithPriorSource('git'));
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+
+      await ext.onStoreDocument(buildData(document, 'user') as any);
+
+      expect(pageHistoryRepo.saveHistory).toHaveBeenCalledTimes(1);
+      expect(callOrder).toEqual(['saveHistory', 'updatePage']);
+    });
+
+    it('a null prior source (first-ever edit) does NOT fire the boundary', async () => {
+      const document = ydocFor(doc('NEW CONTENT'));
+      pageRepo.findById.mockResolvedValue(pageWithPriorSource(null));
+
+      await ext.onStoreDocument(buildData(document, 'agent') as any);
+
+      expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
+      expect(pageRepo.updatePage).toHaveBeenCalledTimes(1);
+    });
+  });
+
   it('idempotency: unchanged content → no updatePage, no history, no queues', async () => {
     // The Y.Doc content equals the persisted content deeply → early skip.
     // A Y.Doc round-trip normalizes attrs (e.g. paragraph indent), so derive
@@ -479,4 +533,231 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
     // Contributors keyed by the UUID so they match the PAGE_HISTORY job (page.id).
     expect(collabHistory.addContributors.mock.calls[0][0]).toBe(PAGE_ID);
   });
+
+  // #370 — explicit save-version (Cmd+S / agent save tool) over the stateless
+  // seam. The tier is derived from the SIGNED connection actor, the store path
+  // is reused, and promote-not-dup avoids duplicating heavy content rows.
+  describe('save-version (#370)', () => {
+    const emitSave = (document: any, actor: 'user' | 'agent') =>
+      ext.onStateless({
+        connection: {
+          readOnly: false,
+          context: { user: { id: USER_ID, name: 'Alice' }, actor },
+        } as any,
+        documentName: `page.${PAGE_ID}`,
+        document: document as any,
+        payload: JSON.stringify({ type: 'save-version' }),
+      } as any);
+
+    // findById returns a page whose content already equals the live doc, so the
+    // store path is a no-op and we isolate the versioning decision.
+    const pageMatchingDoc = (document: any) => ({
+      ...persistedHumanPage('IGNORED'),
+      content: TiptapTransformer.fromYdoc(document, 'default'),
+    });
+
+    it('human save with no prior snapshot → writes a manual version + broadcasts', async () => {
+      const document = ydocFor(doc('VERSION ME'));
+      pageRepo.findById.mockResolvedValue(pageMatchingDoc(document));
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+
+      await emitSave(document, 'user');
+
+      expect(pageHistoryRepo.saveHistory).toHaveBeenCalledTimes(1);
+      expect(pageHistoryRepo.saveHistory.mock.calls[0][1]).toEqual(
+        expect.objectContaining({ kind: 'manual' }),
+      );
+      // The pending idle autosnapshot is cancelled by the explicit version.
+      expect(historyQueue.remove).toHaveBeenCalledWith(PAGE_ID);
+      const msg = JSON.parse(
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
+      );
+      expect(msg).toMatchObject({
+        type: 'version.saved',
+        kind: 'manual',
+        alreadySaved: false,
+      });
+    });
+
+    it('agent save derives kind=agent from the signed actor', async () => {
+      const document = ydocFor(doc('AGENT VERSION'));
+      pageRepo.findById.mockResolvedValue(pageMatchingDoc(document));
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+
+      await emitSave(document, 'agent');
+
+      expect(pageHistoryRepo.saveHistory.mock.calls[pageHistoryRepo.saveHistory.mock.calls.length - 1][1]).toEqual(
+        expect.objectContaining({ kind: 'agent' }),
+      );
+    });
+
+    it('promote-not-dup: latest snapshot is an autosave with identical content → upgrades in place', async () => {
+      const document = ydocFor(doc('SAME'));
+      const page = pageMatchingDoc(document);
+      pageRepo.findById.mockResolvedValue(page);
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue({
+        id: 'auto-1',
+        content: page.content,
+        kind: 'idle',
+      });
+
+      await emitSave(document, 'user');
+
+      // No heavy new content row — the existing autosave is promoted to manual.
+      expect(pageHistoryRepo.updateHistoryKind).toHaveBeenCalledWith(
+        'auto-1',
+        'manual',
+        expect.anything(),
+      );
+      expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
+      const msg = JSON.parse(
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
+      );
+      expect(msg).toMatchObject({ historyId: 'auto-1', alreadySaved: false });
+    });
+
+    it('no-op when the latest snapshot is already a manual version of this content', async () => {
+      const document = ydocFor(doc('ALREADY SAVED'));
+      const page = pageMatchingDoc(document);
+      pageRepo.findById.mockResolvedValue(page);
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue({
+        id: 'ver-1',
+        content: page.content,
+        kind: 'manual',
+      });
+
+      await emitSave(document, 'user');
+
+      expect(pageHistoryRepo.updateHistoryKind).not.toHaveBeenCalled();
+      expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
+      const msg = JSON.parse(
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
+      );
+      expect(msg).toMatchObject({ alreadySaved: true, kind: 'manual' });
+    });
+
+    it('a read-only connection cannot save a version', async () => {
+      const document = ydocFor(doc('READER'));
+      pageRepo.findById.mockResolvedValue(pageMatchingDoc(document));
+
+      await ext.onStateless({
+        connection: {
+          readOnly: true,
+          context: { user: { id: USER_ID }, actor: 'user' },
+        } as any,
+        documentName: `page.${PAGE_ID}`,
+        document: document as any,
+        payload: JSON.stringify({ type: 'save-version' }),
+      } as any);
+
+      expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
+      expect(pageHistoryRepo.updateHistoryKind).not.toHaveBeenCalled();
+    });
+
+    // #370 F8-twin — a COMMIT abort (serialization/deadlock/conn-drop) rejects
+    // OUTSIDE the tx callback, AFTER the destructive popContributors (SPOP) and
+    // saveHistory ran but the INSERT rolled back. onStateless has no retry, so
+    // the outer catch MUST re-add (SADD) the popped set or attribution is lost
+    // irrecoverably. MUTATION: drop the outer catch → addContributors is never
+    // called → this reddens.
+    it('restores popped contributors when the commit aborts after the callback', async () => {
+      const document = ydocFor(doc('VERSION ME'));
+      pageRepo.findById.mockResolvedValue(pageMatchingDoc(document));
+      // No matching snapshot → fresh version branch → pops contributors.
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+      collabHistory.popContributors.mockResolvedValue(['u1', 'u2']);
+
+      // A db whose commit REJECTS after the callback body resolved: the SPOP and
+      // saveHistory already ran, then the tx aborts. onStoreDocument's flush uses
+      // the same db but its content matches (no-op branch) and its own retry loop
+      // swallows the throw, so only the versioning tx exercises the restore.
+      const commitFailingDb = {
+        transaction: () => ({
+          execute: async (fn: (trx: any) => Promise<any>) => {
+            await fn(trxStub);
+            throw new Error('commit aborted (serialization_failure)');
+          },
+        }),
+      };
+      const ext2 = new PersistenceExtension(
+        pageRepo as any,
+        pageHistoryRepo as any,
+        commitFailingDb as any,
+        aiQueue as any,
+        historyQueue as any,
+        notificationQueue as any,
+        collabHistory as any,
+        transclusionService as any,
+      );
+      jest.spyOn(ext2['logger'], 'debug').mockImplementation(() => undefined);
+      jest.spyOn(ext2['logger'], 'warn').mockImplementation(() => undefined);
+      jest.spyOn(ext2['logger'], 'error').mockImplementation(() => undefined);
+
+      await expect(
+        ext2.onStateless({
+          connection: {
+            readOnly: false,
+            context: { user: { id: USER_ID, name: 'Alice' }, actor: 'user' },
+          } as any,
+          documentName: `page.${PAGE_ID}`,
+          document: document as any,
+          payload: JSON.stringify({ type: 'save-version' }),
+        } as any),
+      ).rejects.toThrow();
+
+      // Attribution preserved: the popped set is SADD-restored, keyed by the page
+      // UUID it was popped under.
+      expect(collabHistory.addContributors).toHaveBeenCalledWith(PAGE_ID, [
+        'u1',
+        'u2',
+      ]);
+    });
+
+    // #370 #260 — for a `page.<slugId>` document the idle job is armed under the
+    // page UUID (computeHistoryJob's jobId = page.id), so the supersede-remove
+    // must target page.id, not the raw slugId doc-name id, or it silently misses.
+    it('cancels the superseded idle job by the page UUID for a slugId doc', async () => {
+      const SLUG = 'slug-1'; // persistedHumanPage.slugId
+      const document = ydocFor(doc('VERSION ME'));
+      pageRepo.findById.mockResolvedValue(pageMatchingDoc(document));
+      pageHistoryRepo.findPageLastHistory.mockResolvedValue(null);
+
+      await ext.onStateless({
+        connection: {
+          readOnly: false,
+          context: { user: { id: USER_ID, name: 'Alice' }, actor: 'user' },
+        } as any,
+        documentName: `page.${SLUG}`,
+        document: document as any,
+        payload: JSON.stringify({ type: 'save-version' }),
+      } as any);
+
+      // remove() keyed by the UUID (the real jobId), never the slugId.
+      expect(historyQueue.remove).toHaveBeenCalledWith(PAGE_ID);
+      expect(historyQueue.remove).not.toHaveBeenCalledWith(SLUG);
+    });
+  });
+
+  // #370 — the in-memory idle-burst marker must be dropped on doc unload (like
+  // its sibling per-document maps) or it grows unbounded for every page that was
+  // edited but never manually saved. MUTATION: drop the afterUnloadDocument
+  // delete → the entry survives → this reddens.
+  describe('idleBurstStart housekeeping', () => {
+    it('afterUnloadDocument clears the idle-burst marker armed by a store', async () => {
+      const document = ydocFor(doc('EDIT'));
+      pageRepo.findById.mockResolvedValue(persistedHumanPage('EDIT'));
+
+      await ext.onStoreDocument(buildData(document, 'user') as any);
+
+      const map = ext['idleBurstStart'] as Map<string, number>;
+      // Keyed by documentName (buildData uses `page.${PAGE_ID}`).
+      expect(map.has(`page.${PAGE_ID}`)).toBe(true);
+
+      await ext.afterUnloadDocument({
+        documentName: `page.${PAGE_ID}`,
+      } as any);
+
+      expect(map.has(`page.${PAGE_ID}`)).toBe(false);
+    });
+  });
 });

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -37,9 +37,11 @@ import { Page } from '@docmost/db/types/entity.types';
 import { CollabHistoryService } from '../services/collab-history.service';
 import {
   EMBED_DEBOUNCE_MS,
-  HISTORY_FAST_INTERVAL,
-  HISTORY_FAST_THRESHOLD,
-  HISTORY_INTERVAL,
+  IDLE_INTERVAL_AGENT,
+  IDLE_INTERVAL_USER,
+  IDLE_MAX_WAIT_AGENT,
+  IDLE_MAX_WAIT_USER,
+  PageHistoryKind,
 } from '../constants';
 import { TransclusionService } from '../../core/page/transclusion/transclusion.service';
 import {
@@ -56,6 +58,16 @@ import { hasTransclusionFamilyNodes } from '../../core/page/transclusion/utils/t
  */
 export const INTENTIONAL_CLEAR_MESSAGE_TYPE = 'intentional-clear';
 
+/**
+ * #370 — wire format of the client→server "save a version" signal. Sent by the
+ * human (Cmd+S / Save button) and by the agent's explicit save tool over the
+ * SAME stateless channel. The intentionality tier ('manual' vs 'agent') is
+ * derived SERVER-SIDE from the signed connection actor, never from this
+ * payload, so a version's type is unforgeable. The document is taken from the
+ * connection (not the payload), so the signal cannot be aimed at another page.
+ */
+export const SAVE_VERSION_MESSAGE_TYPE = 'save-version';
+
 /**
  * #251 — how long an intentional-clear signal stays "pending" before it is
  * ignored. The signal is set on the clearing keystroke but consumed by the
@@ -92,35 +104,39 @@ export function resolveSource(
 }
 
 /**
- * Compute the BullMQ job id + delay for a page-history snapshot job. Pure so
- * the data-loss-sensitive timing arithmetic is unit-testable; `now` is injected
- * (caller passes `Date.now()`) for determinism.
+ * #370 — compute the BullMQ job id + delay for a page's trailing idle-flush
+ * autosnapshot. Pure so the timing is unit-testable.
  *
- * - Agent edits: delay 0 and a source-keyed job id `${page.id}-agent`. The
- *   delay MUST stay 0 — the worker re-reads the page row at run time, so any
- *   delay risks reading content a later human edit has already overwritten
- *   (mis-tagged snapshot). 0 minimizes that window. The `-agent` suffix keeps
- *   the job from coalescing with the bare-page.id human job.
- * - Human edits: age-based debounce so rapid human edits coalesce into one
- *   snapshot; job id is the bare `page.id`.
- *
- * BullMQ forbids ':' in custom job ids (Redis key separator), so '-' is used;
- * page.id is a UUID, so `${page.id}-agent` cannot collide with a human job.
+ * Both humans and the agent now share ONE idle pipeline (the agent's old
+ * `delay=0` fast path is gone — intentional agent points arrive via the
+ * explicit save-version signal instead). The job id is the bare `page.id`, so a
+ * page has at most one pending idle job; the caller removes-and-re-adds it on
+ * every store to keep it debounced to the trailing edge of an edit burst. The
+ * window differs by source only: the agent flushes sooner than a human.
  */
 export function computeHistoryJob(
-  page: Pick<Page, 'id' | 'createdAt'>,
+  page: Pick<Page, 'id'>,
   source: string,
-  now: number,
+  // Epoch ms of the FIRST edit in the current burst (when the pending idle job
+  // was first armed). Used to enforce the max-wait ceiling so a continuous
+  // editing session cannot re-arm the trailing timer forever. `now` is injectable
+  // for tests; both default to a live clock / no ceiling when omitted.
+  burstStart?: number,
+  now: number = Date.now(),
 ): { jobId: string; delay: number } {
   const isAgent = source === 'agent';
-  const pageAge = now - new Date(page.createdAt).getTime();
-  const delay = isAgent
-    ? 0
-    : pageAge < HISTORY_FAST_THRESHOLD
-      ? HISTORY_FAST_INTERVAL
-      : HISTORY_INTERVAL;
-  const jobId = isAgent ? `${page.id}-agent` : page.id;
-  return { jobId, delay };
+  const interval = isAgent ? IDLE_INTERVAL_AGENT : IDLE_INTERVAL_USER;
+  const maxWait = isAgent ? IDLE_MAX_WAIT_AGENT : IDLE_MAX_WAIT_USER;
+
+  let delay = interval;
+  if (burstStart !== undefined) {
+    // Time already elapsed since the burst's first edit; the snapshot must fire
+    // no later than `maxWait` after that, so shrink the trailing delay to the
+    // remaining budget (never negative, so BullMQ fires it promptly).
+    const remaining = burstStart + maxWait - now;
+    delay = Math.max(0, Math.min(interval, remaining));
+  }
+  return { jobId: page.id, delay };
 }
 
 @Injectable()
@@ -132,6 +148,28 @@ export class PersistenceExtension implements Extension {
   // coalescing window" per document and OR it across all edits in the window,
   // so the snapshot is marked 'agent' regardless of who wrote last.
   private agentTouched: Map<string, boolean> = new Map();
+  // #370 — epoch ms of the FIRST edit in the current idle-flush burst. Keyed by
+  // documentName (like its sibling per-document maps above), NOT by page.id, so
+  // it can be cleaned in afterUnloadDocument alongside `contributors` /
+  // `agentTouched` / `intentionalClear` when the doc unloads — otherwise any page
+  // that was edited but never manually saved (the common case) would keep its
+  // entry forever and the Map would grow unbounded in this long-lived process.
+  // Set when the pending idle job is first armed (empty entry), read to enforce
+  // the max-wait ceiling in computeHistoryJob, and cleared on doc unload or when
+  // a manual save cancels the idle job so the next burst starts a fresh window.
+  //
+  // Single-process assumption (like `contributors` / `agentTouched` above): this
+  // lives only in THIS collab process's memory. A restart, or a page's ownership
+  // moving to another node, loses the burst-start marker. Consequence: a burst
+  // that spans the restart looks like a fresh burst to the surviving process, so
+  // its max-wait ceiling is re-anchored to the first post-restart edit — a single
+  // continuous session straddling a restart can therefore wait up to ~2× the cap
+  // for its idle snapshot (once for the lost pre-restart window, once for the new
+  // one). Bounded and benign (it only DELAYS a safety-net autosnapshot; manual
+  // saves are unaffected and the next quiet period always flushes), but the
+  // assumption and its consequence are recorded here so no one mistakes the
+  // in-memory marker for a durable, cross-process guarantee.
+  private idleBurstStart: Map<string, number> = new Map();
   // #251 — per-document "intentional clear pending" flags. Keyed by
   // documentName, value = expiry timestamp (ms). Set by onStateless when the
   // client reports a deliberate clear; consumed once by the next
@@ -363,20 +401,19 @@ export class PersistenceExtension implements Extension {
             //this.logger.debug('Contributors error:' + err?.['message']);
           }
 
-          // Approach A — boundary snapshot before the agent's first edit.
-          // When this store is the agent's and the page's currently persisted
-          // state was authored by a human, pin that human state as its own
-          // history version BEFORE the agent overwrites it. `page` still holds
-          // the OLD content/provenance here, so saveHistory(page) captures the
-          // pre-agent state tagged 'user'. The agent's new content is
-          // snapshotted later by the debounced PAGE_HISTORY job ('agent'). Skip
-          // if the prior state is already agent-authored (boundary already
-          // pinned on the user->agent transition), if the page is effectively
-          // empty, or if the latest existing snapshot already equals this human
-          // state (avoid duplicates).
+          // #370 — boundary snapshot on ANY source transition. When the store
+          // flips the page's provenance (user↔agent↔git), pin the OUTGOING
+          // state as its own history version BEFORE the incoming source
+          // overwrites it. `page` still holds the OLD content/provenance here,
+          // so saveHistory(page) captures the pre-transition state tagged with
+          // its own source, kind='boundary'. The incoming content is snapshotted
+          // later by the debounced idle job. Skip if the page is effectively
+          // empty or if the latest existing snapshot already equals this state
+          // (the shared isDeepStrictEqual gate — avoids duplicates). Generalizing
+          // beyond the old user→agent special-case also covers git-sync for free.
           if (
-            lastUpdatedSource === 'agent' &&
-            page.lastUpdatedSource !== 'agent'
+            page.lastUpdatedSource &&
+            page.lastUpdatedSource !== lastUpdatedSource
           ) {
             // pageHistory.pageId is uuid-typed; use page.id (never the doc-name
             // slugId) so a `page.<slugId>` doc cannot throw 22P02 here (#260).
@@ -384,15 +421,13 @@ export class PersistenceExtension implements Extension {
               page.id,
               { includeContent: true, trx },
             );
-            const humanBaselineMissing =
+            const baselineMissing =
               !lastHistory ||
               !isDeepStrictEqual(lastHistory.content, page.content);
-            if (
-              !isEmptyParagraphDoc(page.content as any) &&
-              humanBaselineMissing
-            ) {
+            if (!isEmptyParagraphDoc(page.content as any) && baselineMissing) {
               await this.pageHistoryRepo.saveHistory(page, {
                 contributorIds: page.contributorIds ?? undefined,
+                kind: 'boundary',
                 trx,
               });
             }
@@ -522,7 +557,7 @@ export class PersistenceExtension implements Extension {
         { jobId: `embed-${page.id}`, delay: EMBED_DEBOUNCE_MS },
       );
 
-      await this.enqueuePageHistory(page, lastUpdatedSource);
+      await this.enqueuePageHistory(page, documentName, lastUpdatedSource);
     }
 
     // #402 — report the serialized size for the store histogram's size_bucket.
@@ -554,6 +589,14 @@ export class PersistenceExtension implements Extension {
       return; // unrelated / malformed stateless message
     }
 
+    // #370 — explicit "save a version" (human Cmd+S / agent save tool). Edit
+    // rights are already enforced by the readOnly reject above (a reader can't
+    // create a version), exactly as intentional-clear requires.
+    if (message?.type === SAVE_VERSION_MESSAGE_TYPE) {
+      await this.handleSaveVersion(data);
+      return;
+    }
+
     if (message?.type !== INTENTIONAL_CLEAR_MESSAGE_TYPE) return;
 
     this.intentionalClear.set(
@@ -562,6 +605,160 @@ export class PersistenceExtension implements Extension {
     );
   }
 
+  /**
+   * #370 — persist an intentional version from the live in-memory ydoc.
+   *
+   * One stateless path serves BOTH the human and the agent; the tier is derived
+   * SERVER-SIDE from the signed connection actor ('agent' → 'agent', anything
+   * else → 'manual'), so the version type cannot be spoofed by the client. We
+   * take the fresh ydoc from the collab process memory and run it through the
+   * EXISTING store path first (so pages.content/ydoc reflect the exact content
+   * being versioned — a REST endpoint would race the up-to-10s-stale page row),
+   * then snapshot it into page_history with the intentional kind.
+   *
+   * Promote-not-dup: if the latest history row already holds this exact content
+   * and it is an autosave (idle/boundary/legacy-null), upgrade its kind in place
+   * instead of duplicating a heavy content row; if it is already 'manual', it is
+   * a no-op (the client shows an "already saved" toast). Otherwise a fresh
+   * version row is written, popping the aggregated contributors from Redis.
+   */
+  private async handleSaveVersion(data: onStatelessPayload): Promise<void> {
+    const { connection, document, documentName } = data;
+    const context = connection?.context;
+    const pageId = getPageId(documentName);
+    // Unforgeable: 'agent' only for a signed agent connection, else 'manual'.
+    const kind: PageHistoryKind =
+      context?.actor === 'agent' ? 'agent' : 'manual';
+
+    // Flush the live ydoc through the normal store path so the page row + ydoc
+    // hold exactly what we are about to version (also fires the idle enqueue we
+    // supersede below, plus any source-transition boundary). onStoreDocument
+    // only needs document/documentName/context.
+    await this.onStoreDocument({
+      document,
+      documentName,
+      context,
+    } as onStoreDocumentPayload);
+
+    let result:
+      | { historyId: string; kind: PageHistoryKind; alreadySaved: boolean }
+      | undefined;
+
+    // #370 F8-twin — the contributor set popped from Redis (destructive SPOP)
+    // must be restored if the version row does not durably land. The inner
+    // try/catch below only covers a throw INSIDE the callback; but executeTx
+    // COMMITS after the callback, so a commit-abort (serialization/deadlock/
+    // connection drop — the transient class the epic retries in the processor)
+    // rejects OUTSIDE the callback, after saveHistory already ran and the SPOP
+    // already happened, while the INSERT rolls back. onStateless does NOT retry,
+    // so an unrestored pop is a one-shot irrecoverable attribution loss (the
+    // processor got exactly this fix: poppedForRestore + an outer catch). We
+    // track the popped set here (keyed by the page UUID it was popped by — never
+    // the doc-name id, which may be a slugId, #260) and restore it in the outer
+    // catch. addContributors is an idempotent Redis SADD, so a double-restore is
+    // harmless. versionedPageId is also reused below to remove the superseded
+    // idle job by its real jobId (page.id).
+    let poppedForRestore: string[] = [];
+    let versionedPageId: string | undefined;
+
+    try {
+      await executeTx(this.db, async (trx) => {
+        const page = await this.pageRepo.findById(pageId, {
+          withLock: true,
+          includeContent: true,
+          trx,
+        });
+        if (!page) return;
+        versionedPageId = page.id;
+        // Never version an effectively-empty page (mirrors the processor's
+        // first-history guard); there is nothing intentional to pin.
+        if (isEmptyParagraphDoc(page.content as any)) return;
+
+        const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
+          page.id,
+          { includeContent: true, trx },
+        );
+
+        if (
+          lastHistory &&
+          isDeepStrictEqual(lastHistory.content, page.content)
+        ) {
+          // Content is already snapshotted. Promote-not-dup.
+          if (lastHistory.kind === 'manual') {
+            result = {
+              historyId: lastHistory.id,
+              kind: 'manual',
+              alreadySaved: true,
+            };
+            return;
+          }
+          await this.pageHistoryRepo.updateHistoryKind(
+            lastHistory.id,
+            kind,
+            trx,
+          );
+          result = { historyId: lastHistory.id, kind, alreadySaved: false };
+          return;
+        }
+
+        // Fresh version row. Pop the contributors aggregated since the last
+        // snapshot (SPOP); restore them if the write fails so they aren't lost.
+        const contributorIds = await this.collabHistory.popContributors(
+          page.id,
+        );
+        poppedForRestore = contributorIds;
+        try {
+          const saved = await this.pageHistoryRepo.saveHistory(page, {
+            contributorIds,
+            kind,
+            trx,
+          });
+          result = { historyId: saved.id, kind, alreadySaved: false };
+        } catch (err) {
+          await this.collabHistory.addContributors(page.id, contributorIds);
+          poppedForRestore = [];
+          throw err;
+        }
+      });
+    } catch (err) {
+      // A throw here means the tx did NOT commit (callback threw, or the commit
+      // itself failed and rolled back). If we popped contributors and the inner
+      // catch did not already restore them, restore now so attribution is not
+      // lost — onStateless has no retry to recover it. Restore by the page UUID
+      // the pop was keyed under (versionedPageId is always set before the pop).
+      if (poppedForRestore.length && versionedPageId) {
+        await this.collabHistory.addContributors(
+          versionedPageId,
+          poppedForRestore,
+        );
+      }
+      throw err;
+    }
+
+    // Housekeeping: this explicit version supersedes the page's pending idle
+    // autosnapshot, so cancel it and end the current idle burst so the next edit
+    // starts a fresh max-wait window. Remove the idle job by its REAL jobId
+    // (page.id UUID — computeHistoryJob arms it under page.id), not the raw
+    // doc-name id which may be a slugId for a `page.<slugId>` doc (#260), or the
+    // remove silently misses. The burst marker is keyed by documentName (like its
+    // sibling per-document maps), and is also cleaned in afterUnloadDocument.
+    if (versionedPageId) {
+      await this.historyQueue.remove(versionedPageId).catch(() => undefined);
+    }
+    this.idleBurstStart.delete(documentName);
+
+    if (result) {
+      document.broadcastStateless(
+        JSON.stringify({
+          type: 'version.saved',
+          historyId: result.historyId,
+          kind: result.kind,
+          alreadySaved: result.alreadySaved,
+        }),
+      );
+    }
+  }
+
   async onChange(data: onChangePayload) {
     const documentName = data.documentName;
     const userId = data.context?.user?.id;
@@ -586,6 +783,10 @@ export class PersistenceExtension implements Extension {
     this.contributors.delete(documentName);
     this.agentTouched.delete(documentName);
     this.intentionalClear.delete(documentName);
+    // #370 — drop the idle-burst marker with the other per-document maps so it
+    // cannot accumulate across the process lifetime for never-manually-saved
+    // pages. The pending idle job (if any) is a self-expiring BullMQ delayed job.
+    this.idleBurstStart.delete(documentName);
   }
 
   private consumeContributors(documentName: string): string[] {
@@ -617,19 +818,80 @@ export class PersistenceExtension implements Extension {
 
   private async enqueuePageHistory(
     page: Page,
+    documentName: string,
     lastUpdatedSource: string,
   ): Promise<void> {
-    // Job id + delay arithmetic lives in the pure `computeHistoryJob` (see its
-    // doc comment for the agent-delay-0 / age-based-debounce invariants).
+    // #370 — trailing idle debounce with a max-wait ceiling. One pending idle
+    // job per page (jobId = page.id); on every store we remove the pending
+    // delayed job and re-add it, so the snapshot lands `delay` after edits go
+    // quiet rather than once per store (precedent: workspace.service.ts).
+    // remove() on a delayed job simply deletes it (0 if absent, no throw); if the
+    // job is already ACTIVE and the remove is a no-op, the add still de-dups and
+    // the processor's isDeepStrictEqual gate collapses the duplicate content.
+    //
+    // The FIRST arm of a burst records `burstStart`; computeHistoryJob shrinks
+    // the delay to the remaining max-wait budget from that point, so a continuous
+    // session cannot re-arm the trailing timer forever and starve the snapshot.
+    // A burst marker older than THIS TIER's max-wait means the previous idle job
+    // has already fired — start a fresh window instead of firing immediately on
+    // the next edit. Must use the SAME source-specific max-wait computeHistoryJob
+    // uses (agent 5m / user 10m): a hardcoded USER ceiling would leave an agent
+    // burst's marker stale for 5..10m, forcing delay=0 on every store in that
+    // window and writing one idle row per store — exactly the per-store bloat the
+    // debounce exists to prevent, on the continuous-agent path.
+    const maxWait =
+      lastUpdatedSource === 'agent' ? IDLE_MAX_WAIT_AGENT : IDLE_MAX_WAIT_USER;
+    const now = Date.now();
+    // Keyed by documentName (see the map declaration) so afterUnloadDocument can
+    // clean it; the queue jobId stays page.id (computeHistoryJob) as required.
+    let burstStart = this.idleBurstStart.get(documentName);
+    if (burstStart === undefined || now - burstStart >= maxWait) {
+      burstStart = now;
+      this.idleBurstStart.set(documentName, burstStart);
+    }
+
     const { jobId, delay } = computeHistoryJob(
       page,
       lastUpdatedSource,
-      Date.now(),
+      burstStart,
+      now,
     );
 
+    // remove-then-add trailing-debounce idiom, and its ONE race. We delete the
+    // pending delayed job and re-add it under the same jobId so the timer resets
+    // to the trailing edge of the burst. The race is the small window between
+    // these two awaits: if the delayed job's `delay` elapses in that gap it goes
+    // ACTIVE, and then:
+    //   - remove() on an active/locked job is a no-op (BullMQ won't yank a job a
+    //     worker holds), and our `.catch(() => undefined)` swallows that too; and
+    //   - add() with a jobId that already exists (the now-active job's id) is
+    //     DROPPED by BullMQ — a duplicate add is a no-op.
+    // So this store fails to re-arm the trailing job: the just-fired snapshot
+    // captured content up to the moment it went active, and THIS edit is left
+    // without a pending trailing job. It is bounded and self-healing — the NEXT
+    // store re-arms a fresh delayed job (the id is free again once the active job
+    // completes / removeOnComplete frees it), and the processor's
+    // isDeepStrictEqual gate collapses any content-identical duplicate. The only
+    // uncovered case is when the racing store was the LAST in the session: the
+    // tail edits made after the job went active get NO trailing snapshot until
+    // the next edit re-arms one. That is an acceptable safety-net gap (a manual
+    // Save, a source-transition boundary, or simply the next edit all still cover
+    // it), which is why the reviewer accepts documenting it here rather than
+    // adding a post-add "did the add actually arm a job?" re-check.
+    //
+    // NOTE — do NOT "unify" this with the neighbouring embed-debounce idiom
+    // (aiQueue.add of PAGE_CONTENT_UPDATED above): that one uses a STABLE jobId
+    // and NO remove(), relying purely on BullMQ coalescing a repeated add under
+    // the same id, because a re-embed only needs to eventually run once on the
+    // latest content and re-anchoring its delay on every keystroke is undesirable.
+    // THIS idiom deliberately removes-then-adds precisely to PUSH the delay back
+    // to the trailing edge on every store (a true debounce), which coalescing
+    // alone cannot do. Collapsing them would silently change the history cadence.
+    await this.historyQueue.remove(jobId).catch(() => undefined);
+
     await this.historyQueue.add(
       QueueJob.PAGE_HISTORY,
-      { pageId: page.id } as IPageHistoryJob,
+      { pageId: page.id, kind: 'idle' } as IPageHistoryJob,
       { jobId, delay },
     );
   }

apps/server/src/collaboration/processors/history.processor.spec.ts

@@ -66,6 +66,15 @@ describe('HistoryProcessor.process', () => {
     notificationQueue = { add: jest.fn().mockResolvedValue(undefined) };
     generalQueue = { add: jest.fn().mockResolvedValue(undefined) };
 
+    // #370 F3 — the processor now serializes its find+save under a page-row lock
+    // via executeTx. A db whose transaction().execute(fn) runs fn with a trx stub
+    // drives the real executeTx() helper without a database.
+    const db = {
+      transaction: () => ({
+        execute: (fn: (trx: any) => Promise<any>) => fn({ __trx: true }),
+      }),
+    };
+
     // WorkerHost's constructor reads `this.worker`; passing repos positionally
     // matches the constructor and avoids the Nest DI container.
     proc = new HistoryProcessor(
@@ -73,6 +82,7 @@ describe('HistoryProcessor.process', () => {
       pageRepo as any,
       collabHistory as any,
       watcherService as any,
+      db as any,
       notificationQueue as any,
       generalQueue as any,
     );
@@ -126,15 +136,26 @@ describe('HistoryProcessor.process', () => {
     await proc.process(buildJob());
 
     expect(collabHistory.popContributors).toHaveBeenCalledWith(PAGE_ID);
+    // #370 F3/F9 — the snapshot decision runs under a page-row lock. Pin the lock
+    // structurally so a refactor that drops withLock/trx (silently reintroducing
+    // the TOCTOU double-insert) turns this red. The tx stub is { __trx: true }.
+    expect(pageRepo.findById).toHaveBeenCalledWith(
+      PAGE_ID,
+      expect.objectContaining({ withLock: true, trx: { __trx: true } }),
+    );
+    // #370 F7 — addPageWatchers MUST receive the trx, or its FK-check runs on a
+    // separate connection and self-deadlocks against our FOR UPDATE. Asserting
+    // the trx arg here is exactly what would have caught that regression.
     expect(watcherService.addPageWatchers).toHaveBeenCalledWith(
       ['u1', 'u2'],
       PAGE_ID,
       SPACE_ID,
       WORKSPACE_ID,
+      { __trx: true },
     );
     expect(pageHistoryRepo.saveHistory).toHaveBeenCalledWith(
       expect.objectContaining({ id: PAGE_ID }),
-      { contributorIds: ['u1', 'u2'] },
+      { contributorIds: ['u1', 'u2'], kind: 'idle', trx: { __trx: true } },
     );
     expect(generalQueue.add).toHaveBeenCalledWith(
       QueueJob.PAGE_BACKLINKS,
@@ -186,6 +207,48 @@ describe('HistoryProcessor.process', () => {
     ]);
   });
 
+  it('COMMIT failure (throw outside the tx callback) → contributors RESTORED', async () => {
+    // #370 F8 — a commit-time failure throws OUTSIDE the callback, so the inner
+    // try/catch does not run; the outer catch must restore the popped set (else a
+    // BullMQ retry writes an unattributed version). Use a db whose execute() runs
+    // the callback THEN throws, simulating a commit abort.
+    pageHistoryRepo.findPageLastHistory.mockResolvedValue({
+      content: { type: 'doc', content: [] },
+    });
+    const commitFail = {
+      transaction: () => ({
+        execute: async (fn: (trx: any) => Promise<any>) => {
+          await fn({ __trx: true }); // callback succeeds (saveHistory ok)
+          throw new Error('commit aborted'); // ...but the COMMIT fails
+        },
+      }),
+    };
+    const procCommitFail = new HistoryProcessor(
+      pageHistoryRepo as any,
+      pageRepo as any,
+      collabHistory as any,
+      watcherService as any,
+      commitFail as any,
+      notificationQueue as any,
+      generalQueue as any,
+    );
+    jest
+      .spyOn(procCommitFail['logger'], 'error')
+      .mockImplementation(() => undefined);
+
+    await expect(procCommitFail.process(buildJob())).rejects.toThrow(
+      'commit aborted',
+    );
+    // The inner catch did NOT run (save succeeded), so only the outer catch can
+    // restore — assert it did.
+    expect(collabHistory.addContributors).toHaveBeenCalledWith(PAGE_ID, [
+      'u1',
+      'u2',
+    ]);
+    // And the post-snapshot queue work must NOT have run (we rethrew).
+    expect(generalQueue.add).not.toHaveBeenCalled();
+  });
+
   it('backlinks + notification queue failures are swallowed (history still committed)', async () => {
     pageHistoryRepo.findPageLastHistory.mockResolvedValue({
       content: { type: 'doc', content: [] },

apps/server/src/collaboration/processors/history.processor.ts

@@ -19,6 +19,9 @@ import { isDeepStrictEqual } from 'node:util';
 import { CollabHistoryService } from '../services/collab-history.service';
 import { WatcherService } from '../../core/watcher/watcher.service';
 import { isEmptyParagraphDoc } from '../collaboration.util';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { executeTx } from '@docmost/db/utils';
 
 @Processor(QueueName.HISTORY_QUEUE)
 export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
@@ -29,6 +32,7 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
     private readonly pageRepo: PageRepo,
     private readonly collabHistory: CollabHistoryService,
     private readonly watcherService: WatcherService,
+    @InjectKysely() private readonly db: KyselyDB,
     @InjectQueue(QueueName.NOTIFICATION_QUEUE) private notificationQueue: Queue,
     @InjectQueue(QueueName.GENERAL_QUEUE) private generalQueue: Queue,
   ) {
@@ -41,6 +45,9 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
     try {
       const { pageId } = job.data;
 
+      // Read the page WITHOUT a lock first, only to bail early on the two cheap
+      // no-write cases (page gone / empty first snapshot) without opening a
+      // transaction. The authoritative check-then-write happens locked below.
       const page = await this.pageRepo.findById(pageId, {
         includeContent: true,
       });
@@ -51,40 +58,109 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
         return;
       }
 
-      const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
-        pageId,
-        { includeContent: true },
-      );
+      // #370 F3 — the snapshot decision (findPageLastHistory → saveHistory) must
+      // be serialized against manual-save/boundary writers, which run under a
+      // page-row lock in onStoreDocument. Without it, this processor and a
+      // concurrent manual-save each read the same lastHistory (MVCC), both see
+      // content != lastHistory, and both insert — producing two page_history rows
+      // with IDENTICAL content (one 'idle', one 'manual'), defeating
+      // promote-not-dup and the version-vs-autosave split. Taking the same
+      // page-row lock makes the second writer observe the first's committed row so
+      // the isDeepStrictEqual gate collapses the duplicate. Only the read+write
+      // is transacted; the post-snapshot queue work stays outside.
+      let contributorIds: string[] = [];
+      let snapshotWritten = false;
+      let lastHistoryContent: unknown;
+      // #370 F8 — the contributor set popped from Redis (destructive SPOP) must be
+      // restored if the snapshot does not durably land. The inner try/catch only
+      // covers a throw INSIDE the callback; a COMMIT failure (connection drop,
+      // serialization/deadlock abort on commit — the transient class the epic
+      // already retries) throws OUTSIDE it, rolling the snapshot back while the
+      // pop is already gone. We track the popped set here and restore it in the
+      // outer catch so a BullMQ retry re-attributes the version. addContributors
+      // is an idempotent Redis SADD, so a double-restore is harmless.
+      let poppedForRestore: string[] = [];
 
-      if (!lastHistory && isEmptyParagraphDoc(page.content as any)) {
-        this.logger.debug(
-          `Skipping first history for page ${pageId}: empty content`,
-        );
-        await this.collabHistory.clearContributors(pageId);
+      try {
+        await executeTx(this.db, async (trx) => {
+          const lockedPage = await this.pageRepo.findById(pageId, {
+            includeContent: true,
+            withLock: true,
+            trx,
+          });
+          if (!lockedPage) return;
+
+          const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
+            pageId,
+            { includeContent: true, trx },
+          );
+          lastHistoryContent = lastHistory?.content;
+
+          if (!lastHistory && isEmptyParagraphDoc(lockedPage.content as any)) {
+            this.logger.debug(
+              `Skipping first history for page ${pageId}: empty content`,
+            );
+            return;
+          }
+
+          if (
+            lastHistory &&
+            isDeepStrictEqual(lastHistory.content, lockedPage.content)
+          ) {
+            return; // already snapshotted at this content — nothing to write
+          }
+
+          contributorIds = await this.collabHistory.popContributors(pageId);
+          poppedForRestore = contributorIds;
+          try {
+            // Pass `trx` so the watcher insert's FK check (FOR KEY SHARE on
+            // pages[pageId]) runs on the SAME connection that already holds the
+            // FOR UPDATE lock from findById — otherwise it takes the FK lock on a
+            // separate pool connection and self-deadlocks against our own tx.
+            await this.watcherService.addPageWatchers(
+              contributorIds,
+              pageId,
+              lockedPage.spaceId,
+              lockedPage.workspaceId,
+              trx,
+            );
+
+            // #370 — every job on this queue is a trailing idle-flush autosnapshot.
+            await this.pageHistoryRepo.saveHistory(lockedPage, {
+              contributorIds,
+              kind: job.data.kind ?? 'idle',
+              trx,
+            });
+            snapshotWritten = true;
+            this.logger.debug(`History created for page: ${pageId}`);
+          } catch (err) {
+            await this.collabHistory.addContributors(pageId, contributorIds);
+            poppedForRestore = [];
+            throw err;
+          }
+        });
+      } catch (err) {
+        // A throw here means the tx did NOT commit (callback threw, or the commit
+        // itself failed and rolled back). If we popped contributors and the inner
+        // catch did not already restore them, restore now so the retry keeps
+        // attribution. snapshotWritten is irrelevant: it is set before commit, so
+        // it can be true even when the commit rolled the snapshot back.
+        if (poppedForRestore.length) {
+          await this.collabHistory.addContributors(pageId, poppedForRestore);
+        }
+        throw err;
+      }
+
+      // No snapshot written (page vanished / empty-first / unchanged content) →
+      // clear the contributor set for the skip cases and stop.
+      if (!snapshotWritten) {
+        if (!lastHistoryContent && isEmptyParagraphDoc(page.content as any)) {
+          await this.collabHistory.clearContributors(pageId);
+        }
         return;
       }
 
-      if (
-        !lastHistory ||
-        !isDeepStrictEqual(lastHistory.content, page.content)
-      ) {
-        const contributorIds = await this.collabHistory.popContributors(pageId);
-
-        try {
-          await this.watcherService.addPageWatchers(
-            contributorIds,
-            pageId,
-            page.spaceId,
-            page.workspaceId,
-          );
-
-          await this.pageHistoryRepo.saveHistory(page, { contributorIds });
-          this.logger.debug(`History created for page: ${pageId}`);
-        } catch (err) {
-          await this.collabHistory.addContributors(pageId, contributorIds);
-          throw err;
-        }
-
+      {
         const mentions = extractMentions(page.content);
         const pageMentions = extractPageMentions(mentions);
         const internalLinkSlugIds = extractInternalLinkSlugIds(page.content);
@@ -102,7 +178,7 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
             );
           });
 
-        if (contributorIds.length > 0 && lastHistory?.content) {
+        if (contributorIds.length > 0 && lastHistoryContent) {
           await this.notificationQueue
             .add(QueueJob.PAGE_UPDATED, {
               pageId,

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

@@ -1,42 +1,122 @@
 import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
 
 /**
- * In-memory run-stream registry (#184 phase 1.5). A durable agent run tees its
- * SSE frames here (via `pipeUIMessageStreamToResponse({ consumeSseStream })`)
- * so a LATE tab — one that reloaded, or opened after the starter dropped — can
- * attach through `GET /ai-chat/runs/:chatId/stream`, replay the frames buffered
- * so far, and then follow the live tail as a normal streamer.
+ * In-memory run-stream registry (#184 phase 1.5, step-aligned retention #491). A
+ * durable agent run tees its SSE frames here (via
+ * `pipeUIMessageStreamToResponse({ consumeSseStream })`) so a LATE tab — one that
+ * reloaded, or opened after the starter dropped — can attach through
+ * `GET /ai-chat/runs/:chatId/stream`, be handed the TAIL past the step it already
+ * has persisted, and then follow the live tail as a normal streamer.
  *
  * This is deliberately single-process and best-effort: it holds nothing the DB
  * does not (the run + assistant row are the source of truth), so a process
  * restart simply drops in-flight entries and the client falls back to its
  * restore + degraded-poll path. The async `attach` return type is the seam for a
  * future phase-2 cross-process backend (Redis) — the interface does not change.
+ *
+ * ── #491 step-aligned retention (the OOM fix) ────────────────────────────────
+ * The old registry buffered up to 32MB of raw SSE frames PER active run (V8 ~2×
+ * in memory) and, on attach, blasted the WHOLE buffer to the socket synchronously
+ * with no drain — a handful of marathon runs on a 1GB container OOM'd. #491 caps
+ * the ring at a few MB (env-tunable, default 4MB) and keeps it there by ROTATING:
+ *
+ *  - Every buffered frame is STAMPED with a step number at tee (see ingestFrame).
+ *    Convention: the stamp of a frame is the number of `finish-step` parts seen
+ *    BEFORE it (starting at 0). The finish-step frame itself carries the current
+ *    value, THEN the counter increments. So a frame stamped `s` is the content of
+ *    the (s+1)-th step — 0-based step index `s` — and the stamp aligns EXACTLY
+ *    with `metadata.stepsPersisted`: a client whose persisted `stepsPersisted` is
+ *    N has steps 0..N-1 on disk (and in its seed) and needs the tail `stamp >= N`.
+ *
+ *  - The ring rotates ONLY on a CONFIRMED persist of step N
+ *    (`confirmPersistedStep`), dropping frames with `stamp < N` (those steps are
+ *    now on disk and a fresh client seed carries them). A NON-confirmed step is
+ *    never rotated away, so a persist FAILURE just makes the ring cover MORE
+ *    (auto-safe). This is the anti-inversion rule: a naive "rotate in .then()"
+ *    that rotated after an UNwritten step would drop a step nobody has → silent
+ *    hole. Rotation is gated on a real, successful persist.
+ *
+ *  - If the ring still exceeds its byte cap after rotation (a single fat step, or
+ *    a lagging persist), the OLDEST frames are evicted to stay bounded. Evicting a
+ *    not-yet-persisted frame opens a GAP: an attach whose N falls at or below an
+ *    evicted step answers 204 and the client degrades to restore+poll. The gap is
+ *    NOT sticky — the coverage floor is recomputed from the ring, so a later
+ *    persist that rotates past the holey steps clears it.
+ *
+ * ── attach numbering / coverage (the wire convention) ────────────────────────
+ * The step marker N comes ONLY FROM THE CLIENT (a query param). The server never
+ * reads the row to derive N — a server-side N from a stale seed would open a
+ * silent one-step hole. N is the client's persisted `stepsPersisted` (a COUNT):
+ *   - the tail it needs = frames with `stamp >= N`;
+ *   - coverage is OK ⟺ `coverageFloor(entry) <= N`, where coverageFloor is the
+ *     smallest step FULLY present in the ring (its smallest retained stamp, bumped
+ *     by one when that leading step was only partially evicted by overflow). If
+ *     `coverageFloor > N` the ring starts AFTER the client's frontier (a hole, or
+ *     the client's seed simply lagged behind a rotation) → 204 → the client
+ *     refetches (a larger N) and re-attaches.
+ * The N cutoff is applied in ALL branches, INCLUDING the finished-retained replay.
+ *
+ * ── same-tick invariants (unchanged, still load-bearing) ─────────────────────
+ * invariant 1: only the matching run may mutate/observe an entry (runId check).
+ * invariant 2: retention deletes ONLY its own entry (a replacement may own the key).
+ * invariant 3: open() over a live entry mirrors the done-path (subscribers released).
+ * invariant 4: the tail SLICE + subscriber registration happen in ONE synchronous
+ *              tick inside attach() — no await between them — so a concurrently
+ *              ingested frame is EITHER in the snapshot (buffered before the sync
+ *              block, and the just-added subscriber never sees it) OR fanned out to
+ *              the paused subscriber's `pending` (ingested after) — never both and
+ *              never neither: no loss, no duplication. NOTE (#491): the controller
+ *              now AWAITS the drain-respecting tail write BEFORE calling start(), so
+ *              frames ingested during that await accumulate in `pending`; this is
+ *              bounded by the subscriber cap (an overflow degrades start() to an
+ *              end(), a 204-equivalent). It is the SYNCHRONOUS snapshot+registration
+ *              — not a same-tick start() — that makes this correct.
+ * invariant 5: the controller wires close-cleanup BEFORE any write.
+ * invariant 6: no cross-run replay — the `anchor` (the client's assistant row id)
+ *              must match this run's assistant id, or a foreign run's transcript
+ *              would be appended to the client's message.
  */
 
 /** How long a finished entry is retained for late attach (replay + immediate end). */
 export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
 
 /**
- * Per-run replay buffer cap. Past this the buffer is dropped (attach -> 204, and
- * the client falls back to its restore + degraded-poll path, #430).
- *
- * Raised from 4MB to 32MB (#430): marathon autonomous runs (11-25 min observed)
- * stream far more than 4MB of SSE frames, so a live disconnect mid-run would find
- * an already-overflowed buffer and could only degrade-poll instead of re-attaching
- * to the live tail. 32MB comfortably covers those runs while staying bounded.
- *
- * Memory cost: this is the WORST-CASE retained size PER ACTIVE run (the buffer is
- * freed on finish + retention, or dropped immediately on overflow). With the small
- * number of concurrent autonomous runs a single workspace realistically has, 32MB
- * each is an acceptable ceiling; the overflow->204->degraded-poll fallback remains
- * the backstop for anything larger, so correctness never depends on this bound.
+ * DEFAULT per-run replay ring cap (#491, down from 32MB). SSE frames carry
+ * UNcompacted tool outputs + framing overhead (×1.5–2 vs the persisted parts), so
+ * a "2–3 large reads + reasoning" step routinely blows past 2MB; 4MB comfortably
+ * holds a step or two of TAIL, which is all a resuming client needs (steps below
+ * its persisted frontier come from the seed, not the ring). The ring stays bounded
+ * because it rotates on every confirmed persist; this cap is only the ceiling for
+ * the un-persisted tail between rotations. Env-tunable via
+ * AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES (bytes); a 0/invalid value falls back to this.
  */
-export const RUN_STREAM_MAX_BUFFER_BYTES = 32 * 1024 * 1024;
+export const AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024;
 
-// 2x the replay cap: a just-written full-replay burst alone can never trip the
-// per-subscriber cap (see controller); only a genuinely stalled socket can.
-export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES;
+// 2× the ring cap: a just-written full-tail burst alone can never trip the
+// per-subscriber cap (see controller); only a genuinely stalled socket can. This
+// derivative relationship is preserved even when the ring cap is env-overridden.
+export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES;
+
+/**
+ * A finish-step boundary frame is exactly `data: {"type":"finish-step"...}\n\n`
+ * (verified empirically against ai@6.0.207 — each UI-message-stream part is a
+ * single `data: {json}\n\n` event, never split across `data:` lines, and `type`
+ * is always the first key). A prefix match is cheaper than JSON.parse-per-frame
+ * and has no false positives: a literal `"type":"finish-step"` inside a text
+ * delta is JSON-escaped (`\"type\":...`), and the frame would start with
+ * `data: {"type":"text-delta"` anyway.
+ */
+const FINISH_STEP_FRAME_PREFIX = 'data: {"type":"finish-step"';
+
+/** Resolve the ring cap from the environment, falling back to the default. */
+function resolveMaxBufferBytes(): number {
+  const raw = process.env.AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES;
+  if (!raw) return AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES;
+  const parsed = Number(raw);
+  return Number.isFinite(parsed) && parsed > 0
+    ? Math.floor(parsed)
+    : AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES;
+}
 
 export interface RunStreamCallbacks {
   onFrame: (frame: string) => void;
@@ -44,6 +124,9 @@ export interface RunStreamCallbacks {
 }
 
 export interface RunStreamAttachment {
+  // The synthetic `start` frame (carrying { runId, chatId }) followed by the
+  // buffered TAIL filtered to `stamp >= N`. The controller writes these to the
+  // socket in chunks respecting drain, then calls start().
   replay: string[];
   finished: boolean;
   start(): void; // drain pending frames (order preserved) and go live
@@ -53,14 +136,19 @@ export interface RunStreamAttachment {
 interface Subscriber extends RunStreamCallbacks {
   started: boolean;
   pending: string[];
-  // Byte size of `pending`, capped at SUBSCRIBER_MAX_BUFFERED_BYTES. `start()` is
-  // called in the SAME tick as `attach()` today (see attach), so `pending` never
-  // holds more than one microtask of frames — but the async `attach` signature is
-  // a phase-2 seam: an await between attach and start would let a stalled paused
-  // subscriber buffer the WHOLE run here. The cap is the structural backstop.
+  // Byte size of `pending`, capped at the subscriber cap. `start()` is called in
+  // the SAME tick as `attach()` today, so `pending` never holds more than one
+  // microtask of frames — but the controller writes the (potentially large) tail
+  // respecting drain BEFORE start(), so a stalled socket can accumulate here; the
+  // cap is the structural backstop (an overflow degrades start() to an end()).
   pendingBytes: number;
   overflowed: boolean;
   pendingEnd: boolean;
+  // The client's step frontier N: this subscriber only receives frames with
+  // `stamp >= minStamp` (the tail past what it already persisted). Live frames
+  // always satisfy this (their stamp is the current, highest step), so it only
+  // filters the rare out-of-order below-frontier frame.
+  minStamp: number;
 }
 
 interface Entry {
@@ -68,8 +156,20 @@ interface Entry {
   // The persisted assistant row id of this run (set at bind; undefined if the
   // seed failed). Used by the attach anchor check (invariant 6).
   assistantMessageId?: string;
+  // Parallel arrays: frames[i] is the SSE string, stamps[i] its step number.
   frames: string[];
+  stamps: number[];
   bytes: number;
+  // The running step counter used to stamp the NEXT frame (number of finish-step
+  // frames seen so far).
+  currentStamp: number;
+  // The highest confirmed `stepsPersisted`: frames with stamp < persistedFloor are
+  // on disk (safe to drop, never re-buffered). Monotonic (confirmPersistedStep).
+  persistedFloor: number;
+  // The highest stamp EVICTED by an overflow (unsafe) drop, -1 if none. Used to
+  // detect a partially-evicted leading step when computing the coverage floor.
+  overflowThroughStamp: number;
+  // Sticky-for-logging only: at least one unsafe (overflow) eviction happened.
   overflowed: boolean;
   finished: boolean;
   subscribers: Set<Subscriber>;
@@ -80,6 +180,10 @@ interface Entry {
 export class AiChatStreamRegistryService implements OnModuleDestroy {
   private readonly logger = new Logger(AiChatStreamRegistryService.name);
   private readonly entries = new Map<string, Entry>(); // key: chatId
+  // Env-resolved caps (per instance) so a deployment can tune the ceiling without
+  // a code change. The subscriber cap keeps the documented 2× relationship.
+  readonly maxBufferBytes = resolveMaxBufferBytes();
+  readonly subscriberMaxBufferedBytes = 2 * this.maxBufferBytes;
 
   /**
    * Register a fresh entry at the START of a run (before any frame), so a tab
@@ -105,7 +209,11 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
     this.entries.set(chatId, {
       runId,
       frames: [],
+      stamps: [],
       bytes: 0,
+      currentStamp: 0,
+      persistedFloor: 0,
+      overflowThroughStamp: -1,
       overflowed: false,
       finished: false,
       subscribers: new Set<Subscriber>(),
@@ -150,6 +258,34 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
     void pump();
   }
 
+  /**
+   * Confirm that step `stepsPersisted` (a COUNT: steps 0..stepsPersisted-1) is on
+   * disk for this run, and ROTATE the ring: drop the buffered frames of those
+   * now-persisted steps (stamp < stepsPersisted). This is the ONLY thing that
+   * rotates the ring, and it is called ONLY after a genuinely SUCCESSFUL per-step
+   * persist (see ai-chat.service updateStreaming). A failed persist never calls
+   * it, so the ring covers more (auto-safe). Identity-checked (invariant 1) and
+   * monotonic (a stale lower count is ignored).
+   */
+  confirmPersistedStep(
+    chatId: string,
+    runId: string,
+    stepsPersisted: number,
+  ): void {
+    const entry = this.entries.get(chatId);
+    if (!entry || entry.runId !== runId) return;
+    if (!Number.isFinite(stepsPersisted) || stepsPersisted <= entry.persistedFloor)
+      return;
+    entry.persistedFloor = stepsPersisted;
+    // Clean rotation: drop the persisted steps from the head. These frames are on
+    // disk + carried by a fresh client seed, so this NEVER opens a gap.
+    while (entry.frames.length > 0 && entry.stamps[0] < stepsPersisted) {
+      entry.bytes -= Buffer.byteLength(entry.frames[0]);
+      entry.frames.shift();
+      entry.stamps.shift();
+    }
+  }
+
   /**
    * Terminate a run's entry from the OUTER catch of the stream method (a failure
    * before/while wiring the pipe, so `done` will never arrive). Identity-checked
@@ -162,36 +298,77 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
   }
 
   /**
-   * Attach to a run's stream. Async only for the phase-2 Redis seam — the body
-   * runs synchronously so the replay snapshot and the subscriber registration
-   * happen in ONE tick with no await between them (invariant 4): a frame ingested
-   * concurrently cannot slip into the gap and be lost or duplicated.
+   * Attach to a run's stream from the client's step frontier `n` (its persisted
+   * `stepsPersisted`). Async only for the phase-2 Redis seam — the body runs
+   * synchronously so the tail SLICE and the subscriber registration happen in ONE
+   * tick with no await between them (invariant 4).
    *
    * Returns null (-> the caller answers 204) when:
-   *  - there is no entry, or it overflowed (replay is gone);
-   *  - expect=live with an anchor that does not match this run's assistant id
-   *    (invariant 6: a stripped tab must never replay a FOREIGN run's transcript);
-   *  - the run finished and the caller did not expect a live tail.
-   * A finished run with expect=live yields a replay-only attachment (no
-   * subscriber registered). Otherwise a paused subscriber is registered and the
-   * caller replays `replay`, then calls start() to drain and go live.
+   *  - there is no entry;
+   *  - the `anchor` does not match this run's assistant id (invariant 6);
+   *  - the ring does not cover the client's frontier (coverageFloor > n): a hole
+   *    from overflow, or the client's seed simply lagged behind a rotation. The
+   *    client then refetches (a larger n) and re-attaches.
+   *
+   * Otherwise the attachment's `replay` is a synthetic `start` frame (the run-fact
+   * on re-attach) followed by the buffered tail filtered to `stamp >= n`. For a
+   * FINISHED run this is replay-only (no subscriber) and ends after the replay —
+   * with n = N_final that tail is just the run's `finish` frame, so the client
+   * closes the stream. For a LIVE run a paused subscriber is registered; the
+   * caller writes the replay (respecting drain) then calls start() to drain the
+   * pending frames and go live.
    */
   async attach(
     chatId: string,
-    expectLive: boolean,
     anchor: string | undefined,
+    // The client's persisted step frontier. `null` = a NOT-tail-aware client (no
+    // `n` query param) — a legacy/parameterless tab that expects the old
+    // "finished -> 204 -> poll" contract; distinct from `0` (a tail-aware client
+    // with nothing persisted yet).
+    n: number | null,
     cb: RunStreamCallbacks,
   ): Promise<RunStreamAttachment | null> {
     const entry = this.entries.get(chatId);
-    if (!entry || entry.overflowed) return null;
+    if (!entry) return null;
     // Invariant 6: cross-run replay is forbidden. Before bind, assistantMessageId
     // is undefined and mismatches any anchor -> 204 -> client restore+poll path.
-    if (expectLive && anchor && entry.assistantMessageId !== anchor) return null;
-    if (entry.finished && !expectLive) return null;
-    if (entry.finished && expectLive) {
+    if (anchor && entry.assistantMessageId !== anchor) return null;
+    // #491 regression guard (#137/#161 dup): a NOT-tail-aware client (no `n`)
+    // resuming a FINISHED run must 204 and poll — the old `finished && !expectLive`
+    // gate. Without this, a missing `n` collapsing to frontier 0 would serve the
+    // WHOLE tail of a finished, NON-rotated run (coverageFloor 0), and a
+    // parameterless client that never stripped its transcript would APPEND that
+    // full replay onto the steps it already shows -> duplicated text. A tail-aware
+    // client (n present, incl. n=0) still gets the tail past its frontier.
+    if (entry.finished && n === null) return null;
+    // A finished entry with NOTHING in the ring (aborted before the first frame,
+    // or fully overflowed) has no tail to deliver -> 204 -> the client polls.
+    if (entry.finished && entry.frames.length === 0) return null;
+    // A LIVE run with no `n` (legacy parameterless) replays from step 0 (the old
+    // behavior); a tail-aware client resumes from its frontier.
+    const frontier = n ?? 0;
+    const floor = this.coverageFloor(entry);
+    if (floor > frontier) {
+      this.logger.warn(
+        `run-stream attach gap for run=${entry.runId}: coverageFloor=${floor} ` +
+          `> client frontier=${frontier} -> 204 (client refetches + re-attaches)`,
+      );
+      return null;
+    }
+
+    const startFrame = this.buildStartFrame(chatId, entry.runId);
+    const sliceTail = (): string[] => {
+      const out: string[] = [startFrame];
+      for (let i = 0; i < entry.frames.length; i++) {
+        if (entry.stamps[i] >= frontier) out.push(entry.frames[i]);
+      }
+      return out;
+    };
+
+    if (entry.finished) {
       // Replay-only: the run is done, no subscriber is registered.
       return {
-        replay: entry.frames.slice(),
+        replay: sliceTail(),
         finished: true,
         start: () => undefined,
         unsubscribe: () => undefined,
@@ -206,15 +383,12 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
       pendingBytes: 0,
       overflowed: false,
       pendingEnd: false,
+      minStamp: frontier,
     };
+    // Register + snapshot in the SAME synchronous block (invariant 4). No await
+    // separates them, so a concurrently ingested frame cannot be lost/duplicated.
     entry.subscribers.add(sub);
-    // Snapshot in the SAME synchronous block as the registration (invariant 4).
-    const replay = entry.frames.slice();
-    // CONTRACT: the caller MUST call start() in the SAME tick as this attach()
-    // returns — no await between them. While a subscriber is paused, every frame
-    // is buffered in sub.pending; a delayed start() lets a whole run accumulate
-    // there. The pendingBytes cap (see ingestFrame) is the structural backstop if
-    // that contract is ever broken (e.g. the phase-2 Redis await seam).
+    const replay = sliceTail();
     return {
       replay,
       finished: false,
@@ -263,24 +437,83 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
     this.entries.clear();
   }
 
-  /** Buffer + fan-out a single frame. See invariant/overflow semantics inline. */
+  /** The synthetic `start` frame the tail is prefixed with — the source of the
+   *  run-fact (runId/chatId) on re-attach. A `start` frame does NOT reset the
+   *  client's message parts (ai@6.0.207 createStreamingUIMessageState), so it is
+   *  safe to prepend even when the sliced tail begins mid-message. */
+  private buildStartFrame(chatId: string, runId: string): string {
+    return `data: ${JSON.stringify({
+      type: 'start',
+      messageMetadata: { runId, chatId },
+    })}\n\n`;
+  }
+
+  /**
+   * The smallest step FULLY present in the ring: its smallest retained stamp, or
+   * (when the leading step was only partially evicted by an overflow) one past it.
+   * When the ring is empty it is the current step (only the live tail is coming).
+   * An attach at frontier `n` is covered ⟺ coverageFloor <= n.
+   */
+  private coverageFloor(entry: Entry): number {
+    // Empty ring: only the live tail is coming. The floor is the current step,
+    // but never below persistedFloor — a confirmed persist can rotate the ring
+    // empty while currentStamp still lags a beat behind on another connection, so
+    // max() keeps the invariant STRUCTURAL (a client with n = persistedFloor is
+    // always covered) rather than timing-dependent.
+    if (entry.frames.length === 0)
+      return Math.max(entry.currentStamp, entry.persistedFloor);
+    const min = entry.stamps[0];
+    return entry.overflowThroughStamp >= min ? min + 1 : min;
+  }
+
+  /**
+   * Buffer (step-stamped) + fan-out a single frame. The stamp is the number of
+   * finish-step frames seen BEFORE this one; a finish-step frame carries the
+   * current value and THEN increments the counter (so its stamp equals the 0-based
+   * index of the step it closes). Only frames at/above persistedFloor are buffered
+   * (already-persisted steps are on disk); the ring is then trimmed to the byte
+   * cap, an unsafe eviction opening a gap. Fan-out is always live (filtered per
+   * subscriber by its frontier).
+   */
   private ingestFrame(entry: Entry, frame: string): void {
-    entry.bytes += Buffer.byteLength(frame);
-    if (!entry.overflowed) {
+    const size = Buffer.byteLength(frame);
+    const stamp = entry.currentStamp;
+    if (frame.startsWith(FINISH_STEP_FRAME_PREFIX)) {
+      entry.currentStamp = stamp + 1;
+    }
+
+    // Buffer for replay only if this step is not already persisted+rotated away.
+    if (stamp >= entry.persistedFloor) {
       entry.frames.push(frame);
-      if (entry.bytes > RUN_STREAM_MAX_BUFFER_BYTES) {
-        // The crossing frame was already counted AND (below) fanned out; only the
-        // replay buffer is dropped. After overflow no more frames are buffered,
-        // but live fan-out continues.
-        entry.overflowed = true;
-        entry.frames = [];
-        this.logger.warn(
-          `run-stream buffer overflow for run=${entry.runId}; ` +
-            `late attach will 204 until the run ends`,
-        );
+      entry.stamps.push(stamp);
+      entry.bytes += size;
+      // Enforce the ring cap. Evicting a not-yet-persisted frame (stamp >=
+      // persistedFloor) opens a GAP; a leftover persisted frame (< floor) is a
+      // safe drop. Keep evicting until the ring is back under the cap.
+      while (entry.bytes > this.maxBufferBytes && entry.frames.length > 0) {
+        const evStamp = entry.stamps[0];
+        entry.bytes -= Buffer.byteLength(entry.frames[0]);
+        entry.frames.shift();
+        entry.stamps.shift();
+        if (evStamp >= entry.persistedFloor) {
+          if (evStamp > entry.overflowThroughStamp)
+            entry.overflowThroughStamp = evStamp;
+          if (!entry.overflowed) {
+            entry.overflowed = true;
+            this.logger.warn(
+              `run-stream ring overflow for run=${entry.runId}: an un-persisted ` +
+                `step was evicted to stay under ${this.maxBufferBytes}B; a late ` +
+                `attach at an evicted step will 204 until a later persist confirms`,
+            );
+          }
+        }
       }
     }
+
+    // Fan out live, filtered to each subscriber's frontier (a subscriber only
+    // wants the tail past the step it already persisted).
     for (const sub of entry.subscribers) {
+      if (stamp < sub.minStamp) continue;
       if (sub.started) {
         try {
           sub.onFrame(frame);
@@ -289,12 +522,12 @@ export class AiChatStreamRegistryService implements OnModuleDestroy {
         }
       } else {
         sub.pending.push(frame);
-        sub.pendingBytes += Buffer.byteLength(frame);
-        if (sub.pendingBytes > SUBSCRIBER_MAX_BUFFERED_BYTES) {
+        sub.pendingBytes += size;
+        if (sub.pendingBytes > this.subscriberMaxBufferedBytes) {
           // The paused subscriber's buffer overflowed — only possible if start()
-          // was delayed past the same-tick contract (the phase-2 await seam).
-          // Drop it rather than buffer the whole run; on start() it degrades to an
-          // immediate end (a 204-equivalent) instead of replaying a partial.
+          // was delayed (the controller's drain-respecting tail write, or the
+          // phase-2 await seam). Drop it rather than buffer the whole run; on
+          // start() it degrades to an immediate end (a 204-equivalent).
           sub.overflowed = true;
           sub.pending = [];
           entry.subscribers.delete(sub);

apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts

@@ -1,19 +1,27 @@
 import {
   AiChatStreamRegistryService,
-  RUN_STREAM_MAX_BUFFER_BYTES,
+  AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES,
   RUN_STREAM_RETAIN_FINISHED_MS,
-  SUBSCRIBER_MAX_BUFFERED_BYTES,
   RunStreamCallbacks,
 } from './ai-chat-stream-registry.service';
 
 /**
- * Unit tests for the in-memory run-stream registry (#184 phase 1.5). The registry
- * is the whole of the resumable-transport contract: replay ordering, paused ->
- * live hand-off, overflow, retention, the anchor check (invariant 6), and the
- * mirror-the-done-path replace semantics (invariant 3). Every enumerated case in
- * the issue's task 1.5 has a test here.
+ * Unit tests for the in-memory run-stream registry (#184 phase 1.5, step-aligned
+ * retention #491). The registry is the whole of the resumable-transport contract:
+ * step-stamped retention, tail-only attach at the client's frontier N, the
+ * confirmed-persist ring rotation (and the anti-inversion rule), the memory bound,
+ * the overflow gap, paused -> live hand-off, retention, the anchor check
+ * (invariant 6), and the mirror-the-done-path replace semantics (invariant 3).
  */
 
+// Real ai@6 UI-message-stream SSE frames are `data: {json}\n\n`, one part each.
+const sse = (part: Record<string, unknown>): string =>
+  `data: ${JSON.stringify(part)}\n\n`;
+const finishStep = (): string => sse({ type: 'finish-step' });
+const textDelta = (id: string, delta: string): string =>
+  sse({ type: 'text-delta', id, delta });
+const finish = (): string => sse({ type: 'finish' });
+
 // A ReadableStream whose frames the test pushes explicitly, plus close/error.
 function makePushStream(): {
   stream: ReadableStream<string>;
@@ -58,6 +66,9 @@ function collector(): {
   };
 }
 
+// The tail past the synthetic start frame (replay[0] is always the start frame).
+const tail = (replay: string[]): string[] => replay.slice(1);
+
 describe('AiChatStreamRegistryService', () => {
   const CHAT = 'chat-1';
   let registry: AiChatStreamRegistryService;
@@ -71,7 +82,21 @@ describe('AiChatStreamRegistryService', () => {
     registry.onModuleDestroy();
   });
 
-  it('replays frames in arrival order (live attach)', async () => {
+  it('prepends a synthetic start frame carrying { runId, chatId }', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, 'assist-1', 0, c.cb))!;
+    const start = JSON.parse(att.replay[0].replace(/^data: /, '').trim());
+    expect(start.type).toBe('start');
+    expect(start.messageMetadata).toEqual({ runId: 'run-1', chatId: CHAT });
+  });
+
+  it('replays the buffered tail (from frontier 0) in arrival order (live attach)', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
@@ -81,13 +106,13 @@ describe('AiChatStreamRegistryService', () => {
     await flush();
 
     const c = collector();
-    const att = await registry.attach(CHAT, false, undefined, c.cb);
+    const att = await registry.attach(CHAT, 'assist-1', 0, c.cb);
     expect(att).not.toBeNull();
-    expect(att!.replay).toEqual(['a', 'b', 'c']);
+    expect(tail(att!.replay)).toEqual(['a', 'b', 'c']);
     expect(att!.finished).toBe(false);
   });
 
-  it('late attach gets the full prefix as replay plus the live tail', async () => {
+  it('late attach gets the buffered prefix as tail plus the live tail', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
@@ -96,17 +121,16 @@ describe('AiChatStreamRegistryService', () => {
     await flush();
 
     const c = collector();
-    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
-    expect(att.replay).toEqual(['a', 'b']);
+    const att = (await registry.attach(CHAT, 'assist-1', 0, c.cb))!;
+    expect(tail(att.replay)).toEqual(['a', 'b']);
     att.start();
-    // Live tail arrives after start().
     src.push('c');
     src.push('d');
     await flush();
     expect(c.frames).toEqual(['c', 'd']);
   });
 
-  it('a paused subscriber receives frames buffered during pause in order, then live (no loss/reorder)', async () => {
+  it('a paused subscriber receives frames buffered during pause in order, then live', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
@@ -114,81 +138,45 @@ describe('AiChatStreamRegistryService', () => {
     await flush();
 
     const c = collector();
-    // Attach (paused). Frames that arrive BEFORE start() must queue, not drop.
-    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
-    expect(att.replay).toEqual(['a']);
+    const att = (await registry.attach(CHAT, 'assist-1', 0, c.cb))!;
+    expect(tail(att.replay)).toEqual(['a']);
     src.push('b'); // arrives while paused -> pending
     src.push('c');
     await flush();
     expect(c.frames).toEqual([]); // nothing delivered yet (paused)
-    att.start(); // drains pending in order
+    att.start();
     expect(c.frames).toEqual(['b', 'c']);
-    src.push('d'); // now live
+    src.push('d');
     await flush();
     expect(c.frames).toEqual(['b', 'c', 'd']);
   });
 
   it('a run that finishes while a subscriber is paused ends it on start()', async () => {
     registry.open(CHAT, 'run-1');
+    registry.bind(CHAT, 'run-1', 'assist-1', makePushStream().stream);
     const c = collector();
-    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
-    // Terminate the run while the subscriber is still paused.
+    const att = (await registry.attach(CHAT, 'assist-1', 0, c.cb))!;
     registry.abortEntry(CHAT, 'run-1');
     expect(c.ended()).toBe(0); // paused: not ended yet
     att.start();
     expect(c.ended()).toBe(1); // start() drains + ends
   });
 
-  it('finished + expect=live returns a replay WITHOUT registering a subscriber', async () => {
-    registry.open(CHAT, 'run-1');
-    const src = makePushStream();
-    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
-    src.push('a');
-    src.push('b');
-    src.close();
-    await flush();
-
-    const c = collector();
-    const att = (await registry.attach(CHAT, true, undefined, c.cb))!;
-    expect(att.finished).toBe(true);
-    expect(att.replay).toEqual(['a', 'b']);
-    // No subscriber registered: start()/unsubscribe are no-ops and the entry has
-    // zero subscribers.
-    const entry = (registry as any).entries.get(CHAT);
-    expect(entry.subscribers.size).toBe(0);
-    att.start();
-    expect(c.frames).toEqual([]);
-  });
-
-  it('finished WITHOUT expect=live returns null', async () => {
-    registry.open(CHAT, 'run-1');
-    const src = makePushStream();
-    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
-    src.push('a');
-    src.close();
-    await flush();
-
-    const c = collector();
-    expect(await registry.attach(CHAT, false, undefined, c.cb)).toBeNull();
-  });
-
-  it('anchor mismatch with expect=live returns null (and null before bind sets assistantMessageId)', async () => {
+  it('anchor mismatch returns null (and null before bind sets assistantMessageId)', async () => {
     registry.open(CHAT, 'run-1');
     const c = collector();
     // Before bind: assistantMessageId is undefined -> mismatches any anchor.
-    expect(
-      await registry.attach(CHAT, true, 'assist-1', c.cb),
-    ).toBeNull();
+    expect(await registry.attach(CHAT, 'assist-1', 0, c.cb)).toBeNull();
 
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
     src.push('a');
     await flush();
     // Wrong anchor -> null (cross-run replay forbidden, invariant 6).
-    expect(await registry.attach(CHAT, true, 'other-id', c.cb)).toBeNull();
+    expect(await registry.attach(CHAT, 'other-id', 0, c.cb)).toBeNull();
   });
 
-  it('matching anchor with expect=live attaches', async () => {
+  it('matching anchor attaches', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
@@ -196,97 +184,60 @@ describe('AiChatStreamRegistryService', () => {
     await flush();
 
     const c = collector();
-    const att = await registry.attach(CHAT, true, 'assist-1', c.cb);
+    const att = await registry.attach(CHAT, 'assist-1', 0, c.cb);
     expect(att).not.toBeNull();
-    expect(att!.replay).toEqual(['a']);
+    expect(tail(att!.replay)).toEqual(['a']);
   });
 
-  it('overflow: attach returns null, but the LIVE subscriber keeps receiving (incl. the crossing frame)', async () => {
+  it('a throwing onFrame ejects only that subscriber; the ingest loop stays alive', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
 
-    // A live (started) subscriber attached before the flood.
+    const bad = collector();
+    const badAtt = (await registry.attach(CHAT, 'assist-1', 0, {
+      onFrame: () => {
+        throw new Error('boom');
+      },
+      onEnd: bad.cb.onEnd,
+    }))!;
+    badAtt.start();
+
+    const good = collector();
+    const goodAtt = (await registry.attach(CHAT, 'assist-1', 0, good.cb))!;
+    goodAtt.start();
+
+    src.push('a'); // bad throws on this frame -> ejected
+    src.push('b'); // good still receives both
+    await flush();
+
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry.subscribers.size).toBe(1);
+    expect(good.frames).toEqual(['a', 'b']);
+  });
+
+  it('open() over a LIVE entry ends started subscribers once; a late done never touches the new entry (invariant 3)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+
     const c = collector();
-    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    const att = (await registry.attach(CHAT, 'assist-1', 0, c.cb))!;
     att.start();
 
-    // Cap-relative so it survives a buffer-cap change (#430): a quarter-cap frame
-    // means 5 frames comfortably exceed the replay cap; the last one crosses.
-    const chunk = 'x'.repeat(Math.floor(RUN_STREAM_MAX_BUFFER_BYTES / 4));
-    for (let i = 0; i < 5; i++) src.push(chunk + i);
-    await flush();
-
-    const entry = (registry as any).entries.get(CHAT);
-    expect(entry.overflowed).toBe(true);
-    expect(entry.bytes).toBeGreaterThan(RUN_STREAM_MAX_BUFFER_BYTES);
-    // The live subscriber received ALL 5 frames, including the crossing one.
-    expect(c.frames).toHaveLength(5);
-    expect(c.frames[4]).toBe(chunk + 4);
-
-    // A NEW attach after overflow gets null (replay buffer is gone).
-    const c2 = collector();
-    expect(await registry.attach(CHAT, false, undefined, c2.cb)).toBeNull();
-  });
-
-  it('a paused subscriber whose pending buffer overflows is dropped and ends on start(); other subscribers keep receiving', async () => {
-    registry.open(CHAT, 'run-1');
-    const src = makePushStream();
-    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
-
-    // A: paused (start() deliberately delayed to simulate the phase-2 await seam).
-    const a = collector();
-    const attA = (await registry.attach(CHAT, false, undefined, a.cb))!;
-    // B: live (started) — its delivery must be unaffected by A's overflow.
-    const b = collector();
-    const attB = (await registry.attach(CHAT, false, undefined, b.cb))!;
-    attB.start();
-
-    // Cap-relative so it survives a buffer-cap change (#430): a quarter-of-the-
-    // per-subscriber-cap frame means 5 frames exceed A's paused-pending cap while
-    // B streams every frame live.
-    const chunk = 'x'.repeat(Math.floor(SUBSCRIBER_MAX_BUFFERED_BYTES / 4));
-    for (let i = 0; i < 5; i++) src.push(chunk + i);
-    await flush();
-
-    const entry = (registry as any).entries.get(CHAT);
-    // A was dropped from the subscriber set on overflow; B (started) remains.
-    expect(entry.subscribers.size).toBe(1);
-    expect(a.frames).toEqual([]); // paused + overflowed: nothing was delivered
-    // B received every frame live (delivery unaffected by A's overflow).
-    expect(b.frames).toHaveLength(5);
-
-    // A's start() (arriving late) degrades to an immediate end, not a partial replay.
-    attA.start();
-    expect(a.frames).toEqual([]);
-    expect(a.ended()).toBe(1);
-  });
-
-  it('open() over a LIVE entry ends started subscribers exactly once and a late done does not touch the new entry (invariant 3)', async () => {
-    registry.open(CHAT, 'run-1');
-    const src = makePushStream();
-    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
-    src.push('a');
-    await flush();
-
-    const c = collector();
-    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
-    att.start(); // started subscriber on run-1
-
-    // run-2 starts on the same chat while run-1's tee is still reading.
     registry.open(CHAT, 'run-2');
-    expect(c.ended()).toBe(1); // exactly one onEnd from the replace
+    expect(c.ended()).toBe(1);
 
     const newEntry = (registry as any).entries.get(CHAT);
     expect(newEntry.runId).toBe('run-2');
     expect(newEntry.finished).toBe(false);
 
-    // The old tee now completes: its late done must NOT double-end nor delete the
-    // new entry.
     src.push('b');
     src.close();
     await flush();
-    expect(c.ended()).toBe(1); // still exactly one
+    expect(c.ended()).toBe(1);
     const still = (registry as any).entries.get(CHAT);
     expect(still).toBe(newEntry);
     expect(still.runId).toBe('run-2');
@@ -299,7 +250,6 @@ describe('AiChatStreamRegistryService', () => {
     src.push('a');
     await flush();
     const entry = (registry as any).entries.get(CHAT);
-    // Frames were NOT ingested (bind bailed), assistantMessageId untouched.
     expect(entry.frames).toEqual([]);
     expect(entry.assistantMessageId).toBeUndefined();
   });
@@ -310,32 +260,276 @@ describe('AiChatStreamRegistryService', () => {
     const entry = (registry as any).entries.get(CHAT);
     expect(entry.finished).toBe(false);
   });
+});
 
-  it('a throwing onFrame ejects only that subscriber; the ingest loop stays alive', async () => {
+/**
+ * #491 step-stamped retention: the boundary detector, tail-only slicing at the
+ * client's frontier N, the confirmed-persist rotation (+ anti-inversion), the
+ * overflow gap, the memory bound, and the finished-retained tail. All observable
+ * against the REAL registry driven through open/bind/ingest.
+ */
+describe('AiChatStreamRegistryService step-aligned retention (#491)', () => {
+  const CHAT = 'chat-s';
+  let registry: AiChatStreamRegistryService;
+
+  beforeEach(() => {
+    registry = new AiChatStreamRegistryService();
+    jest.spyOn((registry as any).logger, 'warn').mockImplementation(() => {});
+  });
+  afterEach(() => registry.onModuleDestroy());
+
+  const entryOf = () => (registry as any).entries.get(CHAT);
+
+  it('stamps frames by finish-step count, aligned with stepsPersisted', async () => {
     registry.open(CHAT, 'run-1');
     const src = makePushStream();
     registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    // step 0 content, its finish-step, step 1 content, its finish-step, finish.
+    src.push(textDelta('t0', 'a')); // stamp 0
+    src.push(finishStep()); // stamp 0 (the finish-step frame carries the pre value)
+    src.push(textDelta('t1', 'b')); // stamp 1
+    src.push(finishStep()); // stamp 1
+    src.push(finish()); // stamp 2
+    await flush();
+    const e = entryOf();
+    expect(e.stamps).toEqual([0, 0, 1, 1, 2]);
+    expect(e.currentStamp).toBe(2);
+  });
 
-    const bad = collector();
-    const badAtt = (await registry.attach(CHAT, false, undefined, {
-      onFrame: () => {
-        throw new Error('boom');
-      },
-      onEnd: bad.cb.onEnd,
-    }))!;
-    badAtt.start();
+  it('does NOT treat a text delta that merely quotes "finish-step" as a boundary', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    // A model that literally types "type":"finish-step" — JSON-escaped in the frame.
+    src.push(textDelta('t0', '"type":"finish-step"'));
+    await flush();
+    expect(entryOf().currentStamp).toBe(0); // no false boundary
+  });
 
-    const good = collector();
-    const goodAtt = (await registry.attach(CHAT, false, undefined, good.cb))!;
-    goodAtt.start();
-
-    src.push('a'); // bad throws on this frame -> ejected
-    src.push('b'); // good still receives both
+  it('tail-only: attach at N slices frames with stamp >= N', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(textDelta('t1', 'b')); // 1
+    src.push(finishStep()); // 1
+    src.push(textDelta('t2', 'c')); // 2 (in-progress)
     await flush();
 
-    const entry = (registry as any).entries.get(CHAT);
-    expect(entry.subscribers.size).toBe(1); // bad ejected, good remains
-    expect(good.frames).toEqual(['a', 'b']);
+    const c = collector();
+    // Client persisted 2 steps -> wants the tail from step 2.
+    const att = (await registry.attach(CHAT, 'assist-1', 2, c.cb))!;
+    expect(tail(att.replay)).toEqual([textDelta('t2', 'c')]);
+  });
+
+  it('attach in the MIDDLE of a step (N between finish-steps) slices from that step', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(textDelta('t1', 'b1')); // 1
+    src.push(textDelta('t1', 'b2')); // 1 (still step 1, no finish-step yet)
+    await flush();
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, 'assist-1', 1, c.cb))!;
+    // Step 0's frames are dropped from the tail; the whole in-progress step 1 is kept.
+    expect(tail(att.replay)).toEqual([textDelta('t1', 'b1'), textDelta('t1', 'b2')]);
+  });
+
+  it('rotates the ring ONLY on a confirmed persist (drops stamp < N)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(textDelta('t1', 'b')); // 1
+    await flush();
+    expect(entryOf().stamps).toEqual([0, 0, 1]);
+
+    // Confirm step 0 persisted (stepsPersisted = 1) -> drop stamp < 1.
+    registry.confirmPersistedStep(CHAT, 'run-1', 1);
+    expect(entryOf().stamps).toEqual([1]);
+    expect(entryOf().persistedFloor).toBe(1);
+  });
+
+  it('persist FAILED but the ring still fits -> attach SUCCEEDS and the tail includes step N', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(textDelta('t1', 'b')); // 1 (step 1's persist FAILED -> no confirm)
+    await flush();
+    // No confirmPersistedStep for step 1: the ring still holds step 1.
+
+    const c = collector();
+    // Client's last successful persist was step 0 -> stepsPersisted = 1.
+    const att = await registry.attach(CHAT, 'assist-1', 1, c.cb);
+    expect(att).not.toBeNull();
+    expect(tail(att!.replay)).toEqual([textDelta('t1', 'b')]); // includes step 1
+  });
+
+  it('persist failed AND the ring overflowed past N -> 204 (coverage gap)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    // Step 0: a fat step that blows past the cap with NO persist confirmation.
+    const big = 'x'.repeat(Math.floor(AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES / 2));
+    src.push(textDelta('t0', big)); // 0
+    src.push(textDelta('t0', big)); // 0
+    src.push(textDelta('t0', big)); // 0 -> overflow evicts stamp-0 frames
+    await flush();
+    const e = entryOf();
+    expect(e.overflowed).toBe(true);
+    expect(e.bytes).toBeLessThanOrEqual(registry.maxBufferBytes);
+
+    // A client at frontier 0 falls at/below an evicted step -> gap -> null.
+    const c = collector();
+    expect(await registry.attach(CHAT, 'assist-1', 0, c.cb)).toBeNull();
+  });
+
+  it('stale N (client seed lagged behind a rotation) -> 204; after a refetch (larger N) -> success', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(textDelta('t1', 'b')); // 1
+    src.push(finishStep()); // 1
+    src.push(textDelta('t2', 'c')); // 2
+    await flush();
+    // Server confirmed steps 0 and 1 -> rotate away stamp < 2.
+    registry.confirmPersistedStep(CHAT, 'run-1', 2);
+    expect(entryOf().stamps).toEqual([2]);
+
+    // A client whose seed still says stepsPersisted = 1 -> below minStamp -> 204.
+    const stale = collector();
+    expect(await registry.attach(CHAT, 'assist-1', 1, stale.cb)).toBeNull();
+
+    // It refetches (now stepsPersisted = 2) and re-attaches -> success.
+    const fresh = collector();
+    const att = await registry.attach(CHAT, 'assist-1', 2, fresh.cb);
+    expect(att).not.toBeNull();
+    expect(tail(att!.replay)).toEqual([textDelta('t2', 'c')]);
+  });
+
+  it('overflow gap CLEARS once a later persist rotates out the holey steps', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    const big = 'x'.repeat(Math.floor(AI_CHAT_RUN_STREAM_MAX_BUFFER_BYTES / 2));
+    src.push(textDelta('t0', big)); // 0
+    src.push(textDelta('t0', big)); // 0
+    src.push(finishStep()); // 0 (still stamp 0)
+    src.push(textDelta('t1', 'small')); // 1
+    src.push(finishStep()); // 1
+    src.push(textDelta('t2', 'c')); // 2
+    await flush();
+    expect(entryOf().overflowed).toBe(true);
+
+    // Late persist confirms steps 0..1 -> rotates out the holey step-0 frames.
+    registry.confirmPersistedStep(CHAT, 'run-1', 2);
+    // A client at frontier 2 is now cleanly covered (the hole was below it).
+    const c = collector();
+    const att = await registry.attach(CHAT, 'assist-1', 2, c.cb);
+    expect(att).not.toBeNull();
+    expect(tail(att!.replay)).toEqual([textDelta('t2', 'c')]);
+  });
+
+  it('finished-retained + N = N_final -> empty tail plus the finish frame', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(finish()); // 1  (N_final = 1)
+    src.close();
+    await flush();
+    // The last step's per-step persist confirmed stepsPersisted = 1.
+    registry.confirmPersistedStep(CHAT, 'run-1', 1);
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, 'assist-1', 1, c.cb))!;
+    expect(att.finished).toBe(true);
+    // Empty step tail; just the finish frame so the client's SDK closes the stream.
+    expect(tail(att.replay)).toEqual([finish()]);
+    // No subscriber registered for a finished run.
+    expect(entryOf().subscribers.size).toBe(0);
+  });
+
+  it('#491 regression (#137/#161 dup): a PARAMETERLESS attach (n=null) to a finished NON-rotated run -> 204, but n=0 still gets the tail', async () => {
+    // A finished, non-rotated run: frames present, coverageFloor 0. A missing `n`
+    // (null — a legacy/parameterless tab that never stripped its transcript) must
+    // 204 -> poll, NOT receive the whole tail it would append (duplicate). A
+    // tail-aware client (n=0 present) still resumes.
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a')); // 0
+    src.push(finishStep()); // 0
+    src.push(finish()); // 1
+    src.close();
+    await flush();
+    // NOT rotated (no confirmPersistedStep) -> stamps[0]=0, coverageFloor=0.
+    // MUTATION-VERIFY: revert the `finished && n === null -> null` gate (default n
+    // to 0) and the parameterless attach below serves the full tail instead of 204.
+    expect(await registry.attach(CHAT, 'assist-1', null, collector().cb)).toBeNull();
+    // A tail-aware client at frontier 0 IS served (the distinction: null != 0).
+    const tailAware = await registry.attach(CHAT, 'assist-1', 0, collector().cb);
+    expect(tailAware).not.toBeNull();
+    expect(tailAware!.finished).toBe(true);
+  });
+
+  it('confirmPersistedStep is monotonic and identity-checked', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push(textDelta('t0', 'a'));
+    src.push(finishStep());
+    src.push(textDelta('t1', 'b'));
+    await flush();
+    registry.confirmPersistedStep(CHAT, 'run-1', 1);
+    expect(entryOf().persistedFloor).toBe(1);
+    // A stale lower count is ignored.
+    registry.confirmPersistedStep(CHAT, 'run-1', 0);
+    expect(entryOf().persistedFloor).toBe(1);
+    // A foreign runId is ignored.
+    registry.confirmPersistedStep(CHAT, 'WRONG', 5);
+    expect(entryOf().persistedFloor).toBe(1);
+  });
+
+  it('MEMORY BOUND: 5 parallel marathon runs each stream well past 32MB; each ring stays <= the cap', async () => {
+    const cap = registry.maxBufferBytes;
+    const chats = ['m0', 'm1', 'm2', 'm3', 'm4'];
+    const srcs = chats.map((chat) => {
+      registry.open(chat, `run-${chat}`);
+      const s = makePushStream();
+      registry.bind(chat, `run-${chat}`, `assist-${chat}`, s.stream);
+      return s;
+    });
+    // ~256KB frames; 160 per chat = 40MB streamed each, well past the old 32MB.
+    // Interleave a finish-step every 8 frames so steps advance realistically. No
+    // persist confirmation -> the ONLY thing keeping memory bounded is the cap.
+    const frame = 'y'.repeat(256 * 1024);
+    for (let batch = 0; batch < 20; batch++) {
+      for (let i = 0; i < 8; i++) {
+        for (const s of srcs) s.push(textDelta('t', frame));
+      }
+      for (const s of srcs) s.push(finishStep());
+      await flush(); // drain the pump so queues never hold a whole run
+    }
+    let total = 0;
+    for (const chat of chats) {
+      const e = (registry as any).entries.get(chat);
+      expect(e.bytes).toBeLessThanOrEqual(cap);
+      total += e.bytes;
+    }
+    // Total retained across all 5 runs is bounded by 5x the per-run cap — the old
+    // registry would have retained ~5x40MB = 200MB here.
+    expect(total).toBeLessThanOrEqual(cap * chats.length);
   });
 });
 
@@ -361,7 +555,7 @@ describe('AiChatStreamRegistryService retention timers', () => {
 
   it('a finished entry is removed after the retention window', () => {
     registry.open(CHAT, 'run-1');
-    registry.abortEntry(CHAT, 'run-1'); // finalize -> retention armed
+    registry.abortEntry(CHAT, 'run-1');
     expect((registry as any).entries.get(CHAT)).toBeDefined();
     jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
     expect((registry as any).entries.get(CHAT)).toBeUndefined();
@@ -369,20 +563,18 @@ describe('AiChatStreamRegistryService retention timers', () => {
 
   it('retention deletes ONLY its own entry (invariant 2)', () => {
     registry.open(CHAT, 'run-1');
-    registry.abortEntry(CHAT, 'run-1'); // arm retention for entry A
-    // Simulate the race where the key was replaced without clearing A's timer.
+    registry.abortEntry(CHAT, 'run-1');
     const sentinel = { marker: true };
     (registry as any).entries.set(CHAT, sentinel);
     jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
-    // A's timer saw entries.get(CHAT) !== A, so it did NOT delete the successor.
     expect((registry as any).entries.get(CHAT)).toBe(sentinel);
   });
 
   it('open() over a retained entry clears its timer and the successor survives', () => {
     registry.open(CHAT, 'run-1');
-    registry.abortEntry(CHAT, 'run-1'); // retained, timer armed
+    registry.abortEntry(CHAT, 'run-1');
     const clearSpy = jest.spyOn(global, 'clearTimeout');
-    registry.open(CHAT, 'run-2'); // must clear run-1's retain timer
+    registry.open(CHAT, 'run-2');
     expect(clearSpy).toHaveBeenCalled();
     jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
     const entry = (registry as any).entries.get(CHAT);

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

@@ -8,10 +8,12 @@ import { SUBSCRIBER_MAX_BUFFERED_BYTES } from './ai-chat-stream-registry.service
 import type { User, Workspace } from '@docmost/db/types/entity.types';
 
 /**
- * Wiring spec for the #184 phase 1.5 attach endpoint
+ * Wiring spec for the #184 phase 1.5 attach endpoint (tail-only #491)
  * (`GET /ai-chat/runs/:chatId/stream`). Owner-gated via assertOwnedChat; the
- * registry is mocked so this exercises ONLY the controller's replay/live/204/
- * cleanup wiring against a fake raw socket. Constructor order is (aiChatService,
+ * registry is mocked so this exercises ONLY the controller's tail-write/live/204/
+ * cleanup wiring against a fake raw socket. The attach signature is now
+ * `(chatId, anchor, n, cb)` — the client hands its persisted step frontier `n`
+ * and its assistant row id `anchor`. Constructor order is (aiChatService,
  * aiChatRunService, aiChatRepo, aiChatMessageRepo, aiTranscription, pageRepo,
  * streamRegistry, environment).
  */
@@ -86,8 +88,8 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
       attach: jest.fn(
         (
           _chatId: string,
-          _live: boolean,
           _anchor: string | undefined,
+          _n: number,
           cb: RunStreamCallbacks,
         ) => {
           capturedCb = cb;
@@ -156,7 +158,7 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     expect(res.hijack).not.toHaveBeenCalled();
   });
 
-  it('threads expect=live and anchor through to the registry', async () => {
+  it('threads anchor and the numeric frontier n through to the registry', async () => {
     const { controller, streamRegistry } = makeController({
       chat: owned,
       attachment: null,
@@ -165,8 +167,8 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     const { req } = makeReq();
     await controller.attachRunStream(
       'c1',
-      'live',
       'anchor-1',
+      '2',
       req,
       res,
       user,
@@ -174,13 +176,44 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     );
     expect(streamRegistry.attach).toHaveBeenCalledWith(
       'c1',
-      true,
       'anchor-1',
+      2, // parsed to a number
       expect.anything(),
     );
   });
 
-  it('passes expect=false when the query is absent', async () => {
+  it('#491: an ABSENT/invalid n passes null (not 0) so a finished run 204s (not-tail-aware)', async () => {
+    // Distinguishing a MISSING `n` from `n=0` is the #137/#161 dup guard: a
+    // parameterless/legacy tab must be handed null (-> the registry 204s a finished
+    // run) rather than frontier 0 (which would serve a finished non-rotated run's
+    // whole tail). MUTATION-VERIFY: revert to `Number(n) || 0` and this asserts 0.
+    const { controller, streamRegistry } = makeController({
+      chat: owned,
+      attachment: null,
+    });
+    for (const bad of [undefined, '', 'abc']) {
+      streamRegistry.attach.mockClear();
+      const { res } = makeRawRes();
+      const { req } = makeReq();
+      await controller.attachRunStream(
+        'c1',
+        undefined,
+        bad,
+        req,
+        res,
+        user,
+        workspace,
+      );
+      expect(streamRegistry.attach).toHaveBeenCalledWith(
+        'c1',
+        undefined,
+        null,
+        expect.anything(),
+      );
+    }
+  });
+
+  it('#491: a PRESENT n=0 passes 0 (tail-aware, distinct from absent)', async () => {
     const { controller, streamRegistry } = makeController({
       chat: owned,
       attachment: null,
@@ -190,7 +223,7 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     await controller.attachRunStream(
       'c1',
       undefined,
-      undefined,
+      '0',
       req,
       res,
       user,
@@ -198,8 +231,8 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     );
     expect(streamRegistry.attach).toHaveBeenCalledWith(
       'c1',
-      false,
       undefined,
+      0,
       expect.anything(),
     );
   });
@@ -245,8 +278,8 @@ describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
     const { req } = makeReq();
     await controller.attachRunStream(
       'c1',
-      'live',
       'a1',
+      '1',
       req,
       res,
       user,

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

@@ -0,0 +1,108 @@
+import { ForbiddenException } from '@nestjs/common';
+import { AiChatController } from './ai-chat.controller';
+import type { User, Workspace } from '@docmost/db/types/entity.types';
+
+/**
+ * Wiring spec for the #491 delta-poll endpoint (`POST /ai-chat/messages/delta`).
+ * Owner-gated via assertOwnedChat (same gate as the other reads), NOT flag-gated.
+ * The run fact rides IN the delta response (no separate /run poll). Hand-rolled
+ * mocks — no Nest graph, no DB. Constructor order: (aiChatService,
+ * aiChatRunService, aiChatRepo, aiChatMessageRepo, aiTranscription, pageRepo).
+ */
+describe('AiChatController POST /ai-chat/messages/delta (#491)', () => {
+  const user = { id: 'u1' } as User;
+  const workspace = { id: 'ws1' } as Workspace;
+
+  function makeController(opts: {
+    chat?: unknown;
+    delta?: { rows: unknown[]; cursor: string };
+    run?: unknown;
+  }) {
+    const aiChatRunService = {
+      getLatestForChat: jest.fn().mockResolvedValue(opts.run),
+    };
+    const aiChatRepo = {
+      findById: jest.fn().mockResolvedValue(opts.chat),
+    };
+    const aiChatMessageRepo = {
+      findByChatUpdatedAfter: jest
+        .fn()
+        .mockResolvedValue(opts.delta ?? { rows: [], cursor: 'C1' }),
+    };
+    const controller = new AiChatController(
+      {} as never,
+      aiChatRunService as never,
+      aiChatRepo as never,
+      aiChatMessageRepo as never,
+      {} as never,
+      {} as never,
+    );
+    return { controller, aiChatRunService, aiChatRepo, aiChatMessageRepo };
+  }
+
+  it('owner-gates: a chat the user does not own throws, never reaching the repo', async () => {
+    const { controller, aiChatMessageRepo, aiChatRunService } = makeController({
+      chat: { id: 'c1', creatorId: 'someone-else' },
+    });
+    await expect(
+      controller.getMessagesDelta({ chatId: 'c1' }, user, workspace),
+    ).rejects.toBeInstanceOf(ForbiddenException);
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).not.toHaveBeenCalled();
+    expect(aiChatRunService.getLatestForChat).not.toHaveBeenCalled();
+  });
+
+  it('returns { rows, cursor, run:{id,status} } with the run fact inlined', async () => {
+    const rows = [{ id: 'm1' }];
+    const { controller } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+      delta: { rows, cursor: 'C2' },
+      run: { id: 'r1', status: 'running', error: 'ignored', stepCount: 3 },
+    });
+    const res = await controller.getMessagesDelta(
+      { chatId: 'c1', cursor: 'C1' },
+      user,
+      workspace,
+    );
+    expect(res).toEqual({
+      rows,
+      cursor: 'C2',
+      // ONLY id + status — never the whole run row.
+      run: { id: 'r1', status: 'running' },
+    });
+  });
+
+  it('run is null when the chat has never had a run', async () => {
+    const { controller } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+      run: undefined,
+    });
+    const res = await controller.getMessagesDelta(
+      { chatId: 'c1' },
+      user,
+      workspace,
+    );
+    expect(res.run).toBeNull();
+  });
+
+  it('passes cursor through, defaulting a missing cursor to null (first poll)', async () => {
+    const { controller, aiChatMessageRepo } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+    });
+    await controller.getMessagesDelta({ chatId: 'c1' }, user, workspace);
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).toHaveBeenCalledWith(
+      'c1',
+      'ws1',
+      null,
+    );
+    await controller.getMessagesDelta(
+      { chatId: 'c1', cursor: 'CX' },
+      user,
+      workspace,
+    );
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).toHaveBeenLastCalledWith(
+      'c1',
+      'ws1',
+      'CX',
+    );
+  });
+});

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

@@ -51,6 +51,7 @@ import {
   ChatIdDto,
   ExportChatDto,
   GeneratePageTitleDto,
+  GetChatDeltaDto,
   GetChatMessagesDto,
   GetRunDto,
   RenameChatDto,
@@ -63,6 +64,47 @@ import {
   SUBSCRIBER_MAX_BUFFERED_BYTES,
 } from './ai-chat-stream-registry.service';
 import { startSseHeartbeat } from './sse-resilience';
+
+/**
+ * Write the attach TAIL to the hijacked socket in chunks that RESPECT drain
+ * (#491): each `write()` that returns false (the kernel buffer is full) is awaited
+ * on the next 'drain' before continuing. The old code wrote the whole buffer
+ * synchronously, which — with the pre-#491 32MB ring — spiked memory (half the
+ * OOM). Bails immediately if the socket ended/errored mid-write. Frames that the
+ * paused registry subscriber buffers while this awaits are delivered by start().
+ */
+async function writeTailRespectingDrain(
+  raw: {
+    write(chunk: string): boolean;
+    writableEnded?: boolean;
+    destroyed?: boolean;
+    once(event: string, cb: () => void): unknown;
+    removeListener?(event: string, cb: () => void): unknown;
+  },
+  frames: string[],
+): Promise<void> {
+  for (const frame of frames) {
+    if (raw.writableEnded || raw.destroyed) return;
+    const ok = raw.write(frame);
+    if (!ok) {
+      // Kernel buffer full — wait for drain (or an early close/error) before the
+      // next chunk, so a slow reader never forces the whole tail into memory.
+      // Remove ALL three listeners once any fires, so a many-chunk tail with
+      // repeated backpressure never leaks (MaxListenersExceededWarning).
+      await new Promise<void>((resolve) => {
+        const finish = (): void => {
+          raw.removeListener?.('drain', finish);
+          raw.removeListener?.('close', finish);
+          raw.removeListener?.('error', finish);
+          resolve();
+        };
+        raw.once('drain', finish);
+        raw.once('close', finish);
+        raw.once('error', finish);
+      });
+    }
+  }
+}
 import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 /**
@@ -149,6 +191,46 @@ export class AiChatController {
     );
   }
 
+  /**
+   * Delta poll (#491) — the degraded-poll fallback's payload. Returns the chat's
+   * message rows changed since `cursor` (a DB-clock timestamp from the previous
+   * poll), a FRESH cursor, AND the current run fact `{ id, status } | null`. This
+   * replaces the old degraded poll that refetched ALL infinite-query pages (full
+   * parts) every 2.5s: the client seeds once and thereafter merges only the
+   * deltas by id (the overlap window guarantees repeats — the merge is idempotent,
+   * see mergeById). The run fact rides IN the delta (a separate /run poll would
+   * double the poll QPS), so the client FSM gets the run's status on the same tick.
+   * Owner-gated via assertOwnedChat (same gate as the other read endpoints).
+   */
+  @HttpCode(HttpStatus.OK)
+  @Post('messages/delta')
+  async getMessagesDelta(
+    @Body() dto: GetChatDeltaDto,
+    @AuthUser() user: User,
+    @AuthWorkspace() workspace: Workspace,
+  ): Promise<{
+    rows: AiChatMessage[];
+    cursor: string;
+    run: { id: string; status: string } | null;
+  }> {
+    await this.assertOwnedChat(dto.chatId, user, workspace);
+    const { rows, cursor } =
+      await this.aiChatMessageRepo.findByChatUpdatedAfter(
+        dto.chatId,
+        workspace.id,
+        dto.cursor ?? null,
+      );
+    const run = await this.aiChatRunService.getLatestForChat(
+      dto.chatId,
+      workspace.id,
+    );
+    return {
+      rows,
+      cursor,
+      run: run ? { id: run.id, status: run.status } : null,
+    };
+  }
+
   /**
    * Export a chat to Markdown (#183). The DB is the single source of truth: the
    * whole transcript is loaded (oldest -> newest) and rendered server-side. Now
@@ -249,19 +331,25 @@ export class AiChatController {
   }
 
   /**
-   * Attach to a chat's live run stream (#184 phase 1.5). A late/reloaded tab
-   * replays the frames buffered so far and then follows the live tail as a normal
-   * streamer. Owner-gated via assertOwnedChat (same gate as getRun). When there is
-   * nothing to resume — no entry, a finished run without expect=live, an
-   * overflowed buffer, or an anchor that pins a DIFFERENT run — the endpoint
-   * answers 204, the ONLY "nothing to resume" signal the AI SDK's reconnect
-   * accepts (it maps 204 to a silent no-op). With AI_CHAT_RESUMABLE_STREAM off the
-   * registry is never populated, so attach always 204s.
+   * Attach to a chat's live run stream from the client's step frontier (#184 phase
+   * 1.5, tail-only #491). A late/reloaded tab hands the server the step count it
+   * has PERSISTED (`n` = the seeded row's `metadata.stepsPersisted`) and its
+   * assistant row id (`anchor`); the registry answers with the TAIL past step `n`
+   * (a synthetic `start` frame + the buffered frames stamped >= n) and then the
+   * live tail. Owner-gated via assertOwnedChat (same gate as getRun). When there
+   * is nothing to resume — no entry, a ring that does not cover the client's
+   * frontier (overflow gap, or the client's seed lagged a rotation), or an anchor
+   * that pins a DIFFERENT run (invariant 6) — the endpoint answers 204, the ONLY
+   * "nothing to resume" signal the AI SDK's reconnect accepts (it maps 204 to a
+   * silent no-op); the client then refetches (a larger n) and re-attaches. With
+   * AI_CHAT_RESUMABLE_STREAM off the registry is never populated, so attach always
+   * 204s.
    *
-   * `expect=live` opts into replaying a finished-but-retained run (safe only when
-   * the client stripped the streaming tail); `anchor` is the client's assistant
-   * row id, which must match this run's (invariant 6) or a foreign run's
-   * transcript would be replayed into the store.
+   * The step marker `n` comes ONLY from the client — the server never reads the
+   * row to derive it, because a server-side n from a stale seed would open a
+   * silent one-step hole. The tail is written to the socket in CHUNKS respecting
+   * drain (writeTailRespectingDrain): the old code synchronously blasted the whole
+   * buffer, which — with the old 32MB cap — was half the OOM.
    */
   @SkipTransform()
   @UseGuards(JwtAuthGuard, UserThrottlerGuard)
@@ -269,39 +357,49 @@ export class AiChatController {
   @Get('runs/:chatId/stream')
   async attachRunStream(
     @Param('chatId', new ParseUUIDPipe()) chatId: string,
-    @Query('expect') expect: string | undefined,
     @Query('anchor') anchor: string | undefined,
+    @Query('n') n: string | undefined,
     @Req() req: FastifyRequest,
     @Res() res: FastifyReply,
     @AuthUser() user: User,
     @AuthWorkspace() workspace: Workspace,
   ): Promise<void> {
     await this.assertOwnedChat(chatId, user, workspace); // same gate as getRun
+    // The client's persisted step frontier. #491: distinguish a MISSING/invalid `n`
+    // (null — a NOT-tail-aware, legacy/parameterless tab expecting the old
+    // "finished -> 204 -> poll" contract) from `n=0` (a tail-aware client with
+    // nothing persisted yet). Passing 0 for a missing `n` would serve a finished,
+    // non-rotated run's WHOLE tail and a parameterless client would append it onto
+    // the steps it already shows -> #137/#161 duplicate. null makes the registry
+    // 204 such a finished run (see attach); a tail-aware n=0 still resumes.
+    const frontier: number | null =
+      n === undefined || n === '' || !Number.isFinite(Number(n))
+        ? null
+        : Math.max(0, Number(n));
+    // The per-subscriber backpressure cap tracks the (env-tunable) ring cap.
+    const subscriberCap =
+      this.streamRegistry?.subscriberMaxBufferedBytes ??
+      SUBSCRIBER_MAX_BUFFERED_BYTES;
     let stopHeartbeat: () => void = () => undefined;
-    const attachment = await this.streamRegistry?.attach(
-      chatId,
-      expect === 'live',
-      anchor,
-      {
-        onFrame: (frame) => {
-          // Backpressure guard: 2x the replay cap, so the initial replay burst
-          // alone can never trip it; only a genuinely stalled socket can.
-          try {
-            if (res.raw.writableLength > SUBSCRIBER_MAX_BUFFERED_BYTES) {
-              res.raw.destroy(); // 'close' fires -> unsubscribe below
-              return;
-            }
-            if (!res.raw.writableEnded) res.raw.write(frame);
-          } catch {
-            res.raw.destroy();
+    const attachment = await this.streamRegistry?.attach(chatId, anchor, frontier, {
+      onFrame: (frame) => {
+        // Backpressure guard: 2x the ring cap, so the initial tail burst alone
+        // can never trip it; only a genuinely stalled socket can.
+        try {
+          if (res.raw.writableLength > subscriberCap) {
+            res.raw.destroy(); // 'close' fires -> unsubscribe below
+            return;
           }
-        },
-        onEnd: () => {
-          stopHeartbeat();
-          if (!res.raw.writableEnded) res.raw.end();
-        },
+          if (!res.raw.writableEnded) res.raw.write(frame);
+        } catch {
+          res.raw.destroy();
+        }
       },
-    );
+      onEnd: () => {
+        stopHeartbeat();
+        if (!res.raw.writableEnded) res.raw.end();
+      },
+    });
     if (!attachment) {
       res.status(204).send(); // the ONLY "nothing to resume" signal the SDK accepts
       return;
@@ -330,13 +428,16 @@ export class AiChatController {
         // deliberately NO Connection/Keep-Alive (hop-by-hop; Safari/HTTP2)
       });
       res.raw.flushHeaders?.();
-      for (const frame of attachment.replay) res.raw.write(frame);
+      // Write the tail in chunks respecting drain (not a synchronous blast, which
+      // was half the OOM). Frames the paused subscriber buffers meanwhile are
+      // drained by start() below; its cap is the backstop for a stalled socket.
+      await writeTailRespectingDrain(res.raw, attachment.replay);
       if (attachment.finished) {
-        res.raw.end();
+        if (!res.raw.writableEnded) res.raw.end();
         return;
       }
       stopHeartbeat = startSseHeartbeat(res.raw, 15_000);
-      attachment.start(); // drain pending accumulated during replay, go live
+      attachment.start(); // drain pending accumulated during the tail write, go live
     } catch {
       attachment.unsubscribe();
       stopHeartbeat();

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

@@ -1543,30 +1543,39 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
       // Per-step (non-terminal) update: persist the finished steps the moment a
       // step ends. Tolerant — a failed update is logged and swallowed so it never
       // throws into the stream. Keeps status 'streaming'.
-      const updateStreaming = async (): Promise<void> => {
-        if (!assistantId) return;
+      //
+      // #491: it now SIGNALS its outcome — the persisted `stepsPersisted` count on
+      // a CONFIRMED write, or null when it was skipped/failed. The caller rotates
+      // the run-stream registry ring ONLY on a non-null return (a confirmed
+      // persist), so a failed persist never rotates away a step nobody has (the
+      // classic inversion bug); a failure just makes the ring cover more.
+      const updateStreaming = async (): Promise<number | null> => {
+        if (!assistantId) return null;
         // Cheap short-circuit once the turn is finalized (see `finalized` below).
         // The AUTHORITATIVE guard is `onlyIfStreaming` on the UPDATE: a late
         // fire-and-forget step update could still be in flight on another pool
         // connection when finalize runs, so the SQL `WHERE status='streaming'`
         // (not this flag) is what prevents it clobbering the terminal row.
-        if (finalized) return;
+        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;
         try {
-          await this.aiChatMessageRepo.update(
-            assistantId,
-            workspace.id,
-            flushAssistant(capturedSteps, '', 'streaming', {
-              pageChanged,
-              partsCache,
-            }),
-            { onlyIfStreaming: true },
-          );
+          await this.aiChatMessageRepo.update(assistantId, workspace.id, flushed, {
+            onlyIfStreaming: true,
+          });
+          return stepsPersisted;
         } catch (err) {
           this.logger.warn(
             `Failed to update streaming assistant row: ${
               err instanceof Error ? err.message : 'unknown error'
             }`,
           );
+          return null;
         }
       };
 
@@ -1742,7 +1751,24 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
             // this point still recovers the step. Not awaited here (never block the
             // stream), but SERIALIZED via stepUpdateChain so the writes commit in
             // step order; updateStreaming is error-tolerant (logs + swallows).
-            stepUpdateChain = stepUpdateChain.then(() => updateStreaming());
+            // #491: on a CONFIRMED persist, rotate the run-stream registry ring to
+            // drop the now-on-disk steps (stamp < stepsPersisted). Gated on the
+            // resumable flag (same as open/bind) and identity-checked in the
+            // registry; a null return (skipped/failed) rotates NOTHING (auto-safe).
+            stepUpdateChain = stepUpdateChain.then(async () => {
+              const persisted = await updateStreaming();
+              if (
+                persisted != null &&
+                runId &&
+                this.environment?.isAiChatResumableStreamEnabled?.()
+              ) {
+                this.streamRegistry?.confirmPersistedStep(
+                  chatId,
+                  runId,
+                  persisted,
+                );
+              }
+            });
             // #184: persist the run's progress (finished-step count). Fire-and-
             // forget; the hook swallows its own errors.
             if (runId) runHooks?.onStep?.(runId, capturedSteps.length);
@@ -2891,6 +2917,18 @@ export function flushAssistant(
     // `parts`). Old rows have no marker and the legacy { output } shape; a
     // dual-shape query branches on this. Old rows are deliberately NOT migrated.
     toolTraceVersion: 2,
+    // #491 STEP MARKER: the number of FINISHED steps whose parts are in THIS row,
+    // written by the SAME flush that builds `parts` (atomically — they are both
+    // derived from `finished`, so the marker can NEVER disagree with the persisted
+    // parts). This is the step-alignment anchor the resume stack builds on:
+    // - the registry rotates its retention ring only on a CONFIRMED persist of
+    //   step N (commit 3);
+    // - attach slices the tail at "step > N" from the client's persisted seed.
+    // It is NOT `run.stepCount`: recordStep is fire-and-forget and NOT atomic with
+    // the parts write, so stepCount could race ahead of the persisted parts
+    // (seed↔marker drift). The in-progress trailing text (an error/abort partial,
+    // or a mid-stream flush) is NOT a finished step and is excluded from the count.
+    stepsPersisted: finished.length,
   };
   // finishReason: prefer an explicit one; else derive a sensible value from the
   // terminal status (so onError/onAbort records keep their historical reason).

apps/server/src/core/ai-chat/ai-chat.step-marker.spec.ts

@@ -0,0 +1,65 @@
+import { flushAssistant } from './ai-chat.service';
+
+/**
+ * #491 STEP MARKER — `metadata.stepsPersisted` is written by the SAME flush that
+ * builds `metadata.parts`, so the marker can never disagree with the persisted
+ * parts (the step-alignment anchor the resume stack builds on). These are
+ * PROPERTY tests: they assert the marker tracks the number of FINISHED steps for
+ * every flush shape.
+ */
+
+// A finished step carrying one line of text and one tool call/result.
+function step(i: number) {
+  return {
+    text: `step ${i}`,
+    toolCalls: [
+      { toolCallId: `c${i}`, toolName: 'getPage', input: { id: `p${i}` } },
+    ],
+    toolResults: [
+      { toolCallId: `c${i}`, toolName: 'getPage', output: { title: `T${i}` } },
+    ],
+  };
+}
+
+describe('flushAssistant step marker (#491)', () => {
+  it('seed (no steps) → stepsPersisted 0', () => {
+    const f = flushAssistant([], '', 'streaming');
+    expect(f.metadata.stepsPersisted).toBe(0);
+  });
+
+  it('PROPERTY: stepsPersisted equals the number of FINISHED steps, for any N', () => {
+    for (let n = 0; n <= 6; n++) {
+      const steps = Array.from({ length: n }, (_, i) => step(i));
+      const f = flushAssistant(steps, '', 'streaming');
+      expect(f.metadata.stepsPersisted).toBe(n);
+      // ...and the parts actually contain those N steps' text (marker agrees with
+      // the persisted parts — the atomicity the whole design relies on).
+      const parts = f.metadata.parts as Array<Record<string, unknown>>;
+      const textParts = parts.filter((p) => p.type === 'text');
+      expect(textParts).toHaveLength(n);
+    }
+  });
+
+  it('an in-progress trailing partial does NOT increment the marker', () => {
+    // 2 finished steps + a partial (not-yet-finished) trailing text: the marker
+    // counts only the CONFIRMED step boundaries, not the partial.
+    const f = flushAssistant([step(0), step(1)], 'partial third step', 'error', {
+      error: 'boom',
+    });
+    expect(f.metadata.stepsPersisted).toBe(2);
+    // The partial text IS persisted in parts (so the user sees it), but it is not a
+    // counted step.
+    const parts = f.metadata.parts as Array<Record<string, unknown>>;
+    expect(parts[parts.length - 1]).toEqual({
+      type: 'text',
+      text: 'partial third step',
+    });
+  });
+
+  it('terminal completed flush counts all finished steps', () => {
+    const f = flushAssistant([step(0), step(1), step(2)], '', 'completed', {
+      finishReason: 'stop',
+    });
+    expect(f.metadata.stepsPersisted).toBe(3);
+  });
+});

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

@@ -1,4 +1,10 @@
-import { IsOptional, IsString, MaxLength, MinLength } from 'class-validator';
+import {
+  IsISO8601,
+  IsOptional,
+  IsString,
+  MaxLength,
+  MinLength,
+} from 'class-validator';
 
 /** Identify a chat by id (workspace-scoped on the server). */
 export class ChatIdDto {
@@ -37,6 +43,24 @@ export class GetChatMessagesDto {
   cursor?: string;
 }
 
+/**
+ * Delta poll (#491): pull the chat's rows changed since `cursor` (a DB-clock
+ * timestamp from the previous poll) plus the current run fact — the degraded-poll
+ * fallback's payload, replacing the full infinite-query refetch. Omit `cursor` on
+ * the first poll (returns just a fresh cursor to start the chain).
+ */
+export class GetChatDeltaDto {
+  @IsString()
+  chatId: string;
+
+  // ISO-8601 timestamp echoed from the previous poll's response. Validated as
+  // ISO-8601 (not a bare string): a malformed cursor would otherwise reach the
+  // `::timestamptz` cast in findByChatUpdatedAfter and 500 instead of a clean 400.
+  @IsOptional()
+  @IsISO8601()
+  cursor?: string;
+}
+
 /** Resolve the chat bound to a document (the page's most-recent owned chat). */
 export class BoundChatDto {
   @IsString()

apps/server/src/database/migrations/20260707T120000-page-history-kind.ts

@@ -0,0 +1,27 @@
+import { type Kysely } from 'kysely';
+
+/**
+ * #370 — page-versioning intentionality tier on a history snapshot.
+ *
+ * Adds `page_history.kind`, the three-tier "how intentional was this snapshot"
+ * marker that lets versions (intentional points) be told apart from autosaves:
+ *   - 'manual'   — a human explicitly saved a version (Cmd+S / Save button)
+ *   - 'agent'    — the AI agent explicitly saved a version
+ *   - 'idle'     — trailing idle-flush autosnapshot (safety net)
+ *   - 'boundary' — autosnapshot pinned on a source transition (user↔agent↔git)
+ *
+ * Nullable with NO default (mirrors last_updated_source in the agent-provenance
+ * migration): legacy rows predate the marker and read back as `null`, which the
+ * client renders as a plain autosave. Stored as a short varchar to stay
+ * forward-compatible without an enum migration.
+ */
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .alterTable('page_history')
+    .addColumn('kind', 'varchar(20)', (col) => col)
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.alterTable('page_history').dropColumn('kind').execute();
+}

apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.int-spec.ts

@@ -0,0 +1,280 @@
+import { randomUUID } from 'crypto';
+import { CamelCasePlugin, Kysely, sql } from 'kysely';
+import { PostgresJSDialect } from 'kysely-postgres-js';
+// NOT a default import: the project tsconfig is `module: commonjs` with NO
+// esModuleInterop, so `import postgres from 'postgres'` compiles to
+// `postgres_1.default(...)` and the CJS `postgres` export has no `.default` —
+// it threw in beforeAll, was swallowed as "DB unreachable", and SILENTLY voided
+// all six tests. Mirror the working integration harness (test/integration/db.ts).
+import * as postgres from 'postgres';
+import { AiChatMessageRepo } from './ai-chat-message.repo';
+import { AiChatRunRepo } from './ai-chat-run.repo';
+
+/**
+ * #491 delta-poll — OBSERVABLE-PROPERTY tests against a LIVE Postgres (the local
+ * gitmost test DB, docker `gitmost-test-pg` on :5432), not "rows through a mock"
+ * (a mock cannot observe the DB clock nor the overlap-window race — the very
+ * things that matter here). Drives the REAL repo methods (`findByChatUpdatedAfter`,
+ * the now()-stamped `update`) and asserts:
+ *   1. delta-relevant writes stamp `updatedAt` from the DB clock, not the app clock
+ *      (proven by faking the process clock far into the future and observing the
+ *      stamp stays on real DB time);
+ *   2. the poll returns only rows changed after the cursor, ordered, with a fresh
+ *      DB-clock cursor;
+ *   3. the "committed late but stamped earlier than the cursor" RACE is caught by
+ *      the overlap window (a naive `updatedAt > cursor` would MISS it);
+ *   4. the overlap GUARANTEES repeats across close polls — the contract behind the
+ *      client's idempotent merge (mergeById).
+ *
+ * INTEGRATION lane (`*.int-spec.ts`): runs under `test:int`, whose global-setup
+ * DROPS + RE-CREATES + MIGRATES `docmost_test`, so the real `ai_chat_messages` /
+ * `ai_chat_runs` tables EXIST here. (It was previously a `.spec.ts` defaulting to
+ * the UNmigrated dev `docmost`; in the CI unit lane — where `WAL_TEST_DATABASE_URL`
+ * is unset and only `test:int` migrates — that meant 5/6 ERROR
+ * `relation "ai_chat_messages" does not exist`, silently voiding coverage of the
+ * risky cursor/overlap logic. Renaming to `.int-spec.ts` + defaulting the DSN to
+ * `docmost_test` fixes the CI fidelity.)
+ *
+ * FK triggers are bypassed (`session_replication_role = replica`) so synthetic
+ * chat/workspace ids need no parent fixtures; a single pooled connection (max 1)
+ * keeps that session setting for every query. SKIPS cleanly when the DB is
+ * unreachable so a DB-less CI never breaks.
+ */
+const CONN =
+  process.env.WAL_TEST_DATABASE_URL ??
+  process.env.TEST_DATABASE_URL ??
+  'postgresql://docmost:docmost_dev_pw@localhost:5432/docmost_test';
+
+let db: Kysely<any>;
+let sqlClient: ReturnType<typeof postgres>;
+let msgRepo: AiChatMessageRepo;
+let runRepo: AiChatRunRepo;
+let reachable = false;
+
+const workspaceId = randomUUID();
+const chatId = randomUUID();
+
+beforeAll(async () => {
+  try {
+    sqlClient = postgres(CONN, { max: 1, onnotice: () => {} });
+    db = new Kysely<any>({
+      dialect: new PostgresJSDialect({ postgres: sqlClient }),
+      plugins: [new CamelCasePlugin()],
+    });
+    // Single connection keeps this session-scoped bypass for the whole suite.
+    await sql`set session_replication_role = replica`.execute(db);
+    await sql`select 1`.execute(db);
+    reachable = true;
+  } catch (err) {
+    reachable = false;
+    // A genuine connection failure (ECONNREFUSED etc.) is a legitimate skip on a
+    // DB-less CI. A PROGRAMMING error (bad import, typo, driver misuse) must NOT
+    // masquerade as "DB unreachable" and silently void the whole suite (that is
+    // exactly the bug that hid this spec's zero coverage) — rethrow it so the
+    // suite fails LOUDLY.
+    const msg = String((err as Error)?.message ?? err);
+    if (
+      !/ECONNREFUSED|ENOTFOUND|ETIMEDOUT|EHOSTUNREACH|connect|terminating|password|authentication|role .* does not exist|database .* does not exist/i.test(
+        msg,
+      )
+    ) {
+      throw err;
+    }
+  }
+  msgRepo = new AiChatMessageRepo(db as never);
+  runRepo = new AiChatRunRepo(db as never);
+});
+
+afterAll(async () => {
+  if (db) {
+    try {
+      await db
+        .deleteFrom('aiChatMessages')
+        .where('workspaceId', '=', workspaceId)
+        .execute();
+      await db
+        .deleteFrom('aiChatRuns')
+        .where('workspaceId', '=', workspaceId)
+        .execute();
+    } catch {
+      /* best-effort cleanup */
+    }
+    await db.destroy();
+  }
+});
+
+afterEach(() => {
+  jest.useRealTimers();
+});
+
+async function seedMessage(overrides: Record<string, unknown> = {}) {
+  return msgRepo.insert({
+    chatId,
+    workspaceId,
+    userId: null as never,
+    role: 'assistant',
+    content: 'x',
+    status: 'streaming',
+    ...overrides,
+  } as never);
+}
+
+async function dbNow(): Promise<string> {
+  const r = await sql<{ now: Date }>`select now() as now`.execute(db);
+  return r.rows[0].now.toISOString();
+}
+
+// Fake ONLY the Date object (so in-process `new Date()`/`Date.now()` jump), while
+// leaving every TIMER function real. Faking timers wholesale freezes postgres.js's
+// internal connection/query timers, so the awaited DB round-trip would hang the
+// test (and the afterAll cleanup) at the jest 5s cap. With Date-only faking the
+// query resolves normally, and we still prove the stamp is the DB clock (not the
+// skewed process clock).
+function fakeDateOnly(iso: string): void {
+  jest.useFakeTimers({
+    doNotFake: [
+      'hrtime',
+      'nextTick',
+      'performance',
+      'queueMicrotask',
+      'requestAnimationFrame',
+      'cancelAnimationFrame',
+      'requestIdleCallback',
+      'cancelIdleCallback',
+      'setImmediate',
+      'clearImmediate',
+      'setInterval',
+      'clearInterval',
+      'setTimeout',
+      'clearTimeout',
+    ],
+    now: new Date(iso),
+  });
+}
+
+const maybe = (name: string, fn: () => Promise<void>) =>
+  it(name, async () => {
+    if (!reachable) {
+      console.warn(`SKIP (${name}): test DB unreachable at ${CONN}`);
+      return;
+    }
+    await fn();
+  });
+
+describe('AiChatMessageRepo.findByChatUpdatedAfter (#491 delta poll)', () => {
+  maybe('null cursor returns no rows and a fresh DB-clock cursor', async () => {
+    const before = await dbNow();
+    const { rows, cursor } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      null,
+    );
+    expect(rows).toEqual([]);
+    expect(new Date(cursor).getTime()).toBeGreaterThanOrEqual(
+      new Date(before).getTime(),
+    );
+  });
+
+  maybe('returns only rows changed after the cursor', async () => {
+    const { cursor: c0 } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      null,
+    );
+    const m = await seedMessage();
+    const { rows, cursor: c1 } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      c0,
+    );
+    expect(rows.map((r) => r.id)).toContain(m.id);
+    // Cursor is monotonic (advances).
+    expect(new Date(c1).getTime()).toBeGreaterThanOrEqual(
+      new Date(c0).getTime(),
+    );
+  });
+
+  maybe(
+    'RACE: a row stamped BEFORE the cursor but seen after is caught by the overlap',
+    async () => {
+      // Cursor taken now; then a row appears whose updatedAt is 2s in the PAST
+      // (committed late on another connection but stamped earlier). A naive
+      // `updatedAt > cursor` would MISS it; the 5s overlap window catches it.
+      const cursor = await dbNow();
+      const m = await seedMessage();
+      await sql`update ai_chat_messages set updated_at = now() - interval '2 seconds' where id = ${m.id}`.execute(
+        db,
+      );
+      const { rows } = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        cursor,
+      );
+      expect(rows.map((r) => r.id)).toContain(m.id);
+    },
+  );
+
+  maybe(
+    'overlap GUARANTEES repeats across close polls (idempotent-merge contract)',
+    async () => {
+      const { cursor: c0 } = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        null,
+      );
+      const m = await seedMessage();
+      const first = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        c0,
+      );
+      expect(first.rows.map((r) => r.id)).toContain(m.id);
+      // Immediately re-poll with the JUST-returned cursor: the row is still within
+      // the overlap window, so it is returned AGAIN — the client MUST dedupe by id.
+      const second = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        first.cursor,
+      );
+      expect(second.rows.map((r) => r.id)).toContain(m.id);
+    },
+  );
+
+  maybe(
+    'update() stamps updatedAt from the DB clock, not the app clock',
+    async () => {
+      const m = await seedMessage();
+      // Skew the PROCESS clock ~73 years into the future (Date only). If the stamp
+      // came from `new Date()` the row would read year 2099; sql now() keeps it on
+      // DB time.
+      fakeDateOnly('2099-01-01T00:00:00Z');
+      const updated = await msgRepo.update(m.id, workspaceId, {
+        content: 'y',
+      });
+      jest.useRealTimers();
+      expect(updated).toBeDefined();
+      expect(new Date(updated!.updatedAt).getFullYear()).toBeLessThan(2099);
+    },
+  );
+
+  maybe(
+    'run update() also stamps updatedAt from the DB clock',
+    async () => {
+      const run = await runRepo.insert({
+        chatId,
+        workspaceId,
+        createdBy: null as never,
+        trigger: 'user',
+        status: 'running',
+        stepCount: 0,
+      } as never);
+      fakeDateOnly('2099-01-01T00:00:00Z');
+      const updated = await runRepo.update(run.id, workspaceId, {
+        stepCount: 1,
+      });
+      jest.useRealTimers();
+      expect(updated).toBeDefined();
+      expect(new Date(updated!.updatedAt).getFullYear()).toBeLessThan(2099);
+    },
+  );
+});

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

@@ -25,6 +25,20 @@ const SWEEP_STREAMING_STALE_MS = 10 * 60 * 1000; // 10 minutes
 // into memory; far above any realistic transcript length.
 const FIND_ALL_BY_CHAT_LIMIT = 5000;
 
+// Delta-poll overlap (#491): the poll query reaches this far BEHIND the client's
+// echoed cursor, so a row that committed with an `updatedAt` marginally before the
+// previous cursor was taken (on another autocommit connection) is still caught.
+// Sized well above realistic single-row commit skew; the client merge is
+// idempotent by id (mergeById), so the guaranteed repeats the overlap produces are
+// harmless.
+export const DELTA_POLL_OVERLAP_SECONDS = 5;
+
+// Hard cap on rows one delta poll returns — a safety bound (a poll should carry a
+// handful of just-changed rows, never a whole transcript). Ordered by (updatedAt,
+// id) asc, so on the pathological overflow the OLDEST changes win and the newest
+// are picked up by the next poll (its cursor did not advance past them).
+export const DELTA_POLL_MAX_ROWS = 500;
+
 @Injectable()
 export class AiChatMessageRepo {
   private readonly logger = new Logger(AiChatMessageRepo.name);
@@ -139,6 +153,72 @@ export class AiChatMessageRepo {
       .executeTakeFirst();
   }
 
+  /**
+   * Delta read (#491) for the degraded poll: the chat's messages whose row
+   * changed AFTER the client's `cursor`, plus a FRESH cursor taken from the DB
+   * clock. Replaces the old "refetch ALL infinite-query pages every 2.5s with
+   * full parts" poll — the client seeds once (findByChat) and thereafter pulls
+   * only the deltas and merges them by id (mergeById).
+   *
+   * Cursor: a DB-clock timestamp (now()) the client echoes back each poll. All
+   * delta-relevant writes stamp `updatedAt` with now() (see `update` /
+   * `finalizeOwner`), so this is a SINGLE monotonic axis. The query overlaps the
+   * cursor by DELTA_POLL_OVERLAP_SECONDS to catch a row committed with an
+   * `updatedAt` marginally BEFORE the previous cursor was taken on another
+   * connection (single-row autocommit UPDATEs; no long transactions). The overlap
+   * GUARANTEES occasional REPEATS, so the client merge MUST be idempotent by id.
+   *
+   * `cursor === null` (first poll after the full seed) returns NO rows — there is
+   * nothing "new" relative to a just-loaded seed — only the fresh cursor to start
+   * the delta chain. The fresh cursor is read AFTER the rows, so it is >= every
+   * returned row's `updatedAt` (they were read strictly earlier) — a row that
+   * commits between the rows-read and the cursor-read is at most
+   * DELTA_POLL_OVERLAP_SECONDS behind the returned cursor, so the next poll's
+   * overlap window always re-includes it (no miss).
+   */
+  async findByChatUpdatedAfter(
+    chatId: string,
+    workspaceId: string,
+    cursor: string | null,
+  ): Promise<{ rows: AiChatMessage[]; cursor: string }> {
+    if (cursor === null) {
+      const nowRow = await sql<{ now: Date }>`select now() as now`.execute(
+        this.db,
+      );
+      return { rows: [], cursor: nowRow.rows[0].now.toISOString() };
+    }
+    // Overlap the client cursor by DELTA_POLL_OVERLAP_SECONDS, computed in SQL off
+    // the echoed cursor so the whole comparison stays on the DB clock.
+    const rows = await this.db
+      .selectFrom('aiChatMessages')
+      .select(this.baseFields)
+      .where('chatId', '=', chatId)
+      .where('workspaceId', '=', workspaceId)
+      .where('deletedAt', 'is', null)
+      .where(
+        'updatedAt',
+        '>',
+        sql<Date>`${cursor}::timestamptz - make_interval(secs => ${DELTA_POLL_OVERLAP_SECONDS})`,
+      )
+      .orderBy('updatedAt', 'asc')
+      .orderBy('id', 'asc')
+      .limit(DELTA_POLL_MAX_ROWS)
+      .execute();
+    // When the page filled (pathological overflow), DO NOT advance the cursor to
+    // now(): that would skip the changed rows past the cap that this poll did not
+    // return. Resume from the last returned row's updatedAt instead (the next
+    // poll's overlap re-includes ties by id). In the normal case the fresh DB-clock
+    // now() is the cursor.
+    if (rows.length === DELTA_POLL_MAX_ROWS) {
+      return {
+        rows,
+        cursor: rows[rows.length - 1].updatedAt.toISOString(),
+      };
+    }
+    const nowRow = await sql<{ now: Date }>`select now() as now`.execute(this.db);
+    return { rows, cursor: nowRow.rows[0].now.toISOString() };
+  }
+
   async insert(
     insertable: InsertableAiChatMessage,
     trx?: KyselyTransaction,
@@ -172,7 +252,13 @@ export class AiChatMessageRepo {
     const db = dbOrTx(this.db, opts?.trx);
     let query = db
       .updateTable('aiChatMessages')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: stamp `updatedAt` from the DB clock (sql now()), NOT the app clock
+      // (new Date()). The delta-poll cursor (findByChatUpdatedAfter) is a single
+      // DB-clock axis; a per-step 'streaming' UPDATE stamped with the app clock
+      // would be a SECOND, skewed clock source and could leave a row's updatedAt
+      // just under a cursor taken from now() on another connection — an
+      // independent source of delta MISSES. All delta-relevant writes use now().
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId);
     // Concurrency guard (#183 review): a per-step 'streaming' update must NEVER
@@ -214,7 +300,9 @@ export class AiChatMessageRepo {
     const db = dbOrTx(this.db, trx);
     return db
       .updateTable('aiChatMessages')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: DB-clock stamp (see `update`) — this terminal write flips the row's
+      // status, which the delta poll must observe on the shared now() cursor axis.
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId)
       .where((eb) =>
@@ -249,7 +337,9 @@ export class AiChatMessageRepo {
       .set({
         status,
         metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`) so a reconcile status flip lands on
+        // the same now() cursor axis the delta poll reads.
+        updatedAt: sql`now()`,
       })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId)
@@ -307,7 +397,9 @@ export class AiChatMessageRepo {
       .set({
         status: 'aborted',
         metadata: sql`coalesce(m.metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`). The staleness WHERE below stays on
+        // the app clock — a >minutes window makes the ms-scale skew irrelevant.
+        updatedAt: sql`now()`,
       })
       .where('m.status', '=', 'streaming')
       .where('m.updatedAt', '<', staleBefore)
@@ -351,7 +443,8 @@ export class AiChatMessageRepo {
       .set({
         status: 'aborted',
         metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`). Staleness WHERE stays app-clock.
+        updatedAt: sql`now()`,
       })
       .where('status', '=', 'streaming')
       .where('updatedAt', '<', staleBefore)

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

@@ -62,10 +62,17 @@ describe('AiChatRunRepo.sweepRunning', () => {
     // ...but a fresh 'running' run (updatedAt = now) must NOT be skipped: no
     // updatedAt predicate at all on the boot path.
     expect(rec.wheres.some(([col]) => col === 'updatedAt')).toBe(false);
-    // It flips to 'aborted' and stamps finishedAt.
-    expect(rec.set).toEqual(
-      expect.objectContaining({ status: 'aborted', finishedAt: expect.any(Date) }),
-    );
+    // It flips to 'aborted' and stamps finishedAt + updatedAt. #491: the stamps
+    // are now DB-clock `sql now()` expressions (raw builders), NOT app-clock
+    // `new Date()`, so the run row shares the delta poll's single now() cursor axis
+    // — assert they are present and are the sql raw-builder objects (not a Date,
+    // not undefined).
+    expect(rec.set?.status).toBe('aborted');
+    for (const stamp of ['finishedAt', 'updatedAt'] as const) {
+      expect(rec.set?.[stamp]).toBeDefined();
+      expect(rec.set?.[stamp]).not.toBeInstanceOf(Date);
+      expect(typeof rec.set?.[stamp]).toBe('object');
+    }
   });
 
   it('phase-2 path: an explicit staleMs reintroduces the updatedAt window', async () => {

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

@@ -136,7 +136,11 @@ export class AiChatRunRepo {
     const db = dbOrTx(this.db, trx);
     return db
       .updateTable('aiChatRuns')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: DB-clock stamp (sql now()) so the run row shares the delta poll's
+      // single now() cursor axis with the assistant message rows — a run-status
+      // change (the run fact the delta carries) must never sit on a skewed app
+      // clock relative to the message updatedAt cursor.
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId)
       .returning(this.baseFields)
@@ -162,14 +166,15 @@ export class AiChatRunRepo {
     trx?: KyselyTransaction,
   ): Promise<AiChatRun | undefined> {
     const db = dbOrTx(this.db, trx);
-    const now = new Date();
     return db
       .updateTable('aiChatRuns')
       .set({
         status: patch.status,
         error: patch.error,
-        finishedAt: now,
-        updatedAt: now,
+        // #491: DB-clock stamps (finished_at + updated_at) so the terminal run
+        // fact lands on the delta poll's now() cursor axis.
+        finishedAt: sql`now()`,
+        updatedAt: sql`now()`,
       })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId)
@@ -192,7 +197,8 @@ export class AiChatRunRepo {
     const db = dbOrTx(this.db, trx);
     return db
       .updateTable('aiChatRuns')
-      .set({ stopRequestedAt: new Date(), updatedAt: new Date() })
+      // #491: DB-clock stamps (see `update`).
+      .set({ stopRequestedAt: sql`now()`, updatedAt: sql`now()` })
       .where('id', '=', id)
       .where('workspaceId', '=', workspaceId)
       .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[])
@@ -249,13 +255,14 @@ export class AiChatRunRepo {
     trx?: KyselyTransaction,
   ): Promise<number> {
     const db = dbOrTx(this.db, trx);
-    const now = new Date();
     let query = db
       .updateTable('aiChatRuns')
       .set({
         status: 'aborted',
-        finishedAt: now,
-        updatedAt: now,
+        // #491: DB-clock stamps (see `update`). The staleness WHERE below stays on
+        // the app clock — a >minutes window makes the ms-scale skew irrelevant.
+        finishedAt: sql`now()`,
+        updatedAt: sql`now()`,
         error: sql`coalesce(error, ${'Run interrupted by a server restart.'})`,
       })
       .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[]);
@@ -263,7 +270,7 @@ export class AiChatRunRepo {
     // sibling replica's live run is never aborted. Omitted on the phase-1 boot
     // sweep -> unconditional.
     if (typeof opts.staleMs === 'number') {
-      const staleBefore = new Date(now.getTime() - opts.staleMs);
+      const staleBefore = new Date(Date.now() - opts.staleMs);
       query = query.where('updatedAt', '<', staleBefore);
     }
     const rows = await query.returning('id').execute();

apps/server/src/database/repos/page/page-history.repo.ts

@@ -13,6 +13,7 @@ import { jsonArrayFrom, jsonObjectFrom } from 'kysely/helpers/postgres';
 import { ExpressionBuilder, sql } from 'kysely';
 import { DB } from '@docmost/db/types/db';
 import { resolveAgentProvenance } from '../agent-provenance';
+import { PageHistoryKind } from '../../../collaboration/constants';
 
 /**
  * Role-resolution subquery for a page-history row's bound AI chat (#300). Joins
@@ -46,6 +47,9 @@ export class PageHistoryRepo {
     'lastUpdatedById',
     'lastUpdatedSource',
     'lastUpdatedAiChatId',
+    // #370 — intentionality tier ('manual' | 'agent' | 'idle' | 'boundary');
+    // null on legacy rows (= autosave). Selected so callers can read/promote it.
+    'kind',
     'contributorIds',
     'spaceId',
     'workspaceId',
@@ -85,9 +89,15 @@ export class PageHistoryRepo {
 
   async saveHistory(
     page: Page,
-    opts?: { contributorIds?: string[]; trx?: KyselyTransaction },
-  ): Promise<void> {
-    await this.insertPageHistory(
+    opts?: {
+      contributorIds?: string[];
+      // #370 — intentionality tier for this snapshot. Omitted → null (legacy
+      // autosave semantics). Callers derive it server-side, never from a client.
+      kind?: PageHistoryKind;
+      trx?: KyselyTransaction;
+    },
+  ): Promise<PageHistory> {
+    return await this.insertPageHistory(
       {
         pageId: page.id,
         slugId: page.slugId,
@@ -99,6 +109,7 @@ export class PageHistoryRepo {
         // Copy the provenance marker off the page row, as for lastUpdatedById.
         lastUpdatedSource: page.lastUpdatedSource,
         lastUpdatedAiChatId: page.lastUpdatedAiChatId,
+        kind: opts?.kind ?? null,
         contributorIds: opts?.contributorIds,
         spaceId: page.spaceId,
         workspaceId: page.workspaceId,
@@ -107,6 +118,25 @@ export class PageHistoryRepo {
     );
   }
 
+  /**
+   * #370 — promote an existing snapshot's intentionality tier in place. Used by
+   * the manual-save "promote-not-dup" path: when the latest history row already
+   * holds the exact content being versioned, we upgrade its `kind` instead of
+   * duplicating a heavy content row.
+   */
+  async updateHistoryKind(
+    pageHistoryId: string,
+    kind: PageHistoryKind,
+    trx?: KyselyTransaction,
+  ): Promise<void> {
+    const db = dbOrTx(this.db, trx);
+    await db
+      .updateTable('pageHistory')
+      .set({ kind })
+      .where('id', '=', pageHistoryId)
+      .execute();
+  }
+
   async findPageHistoryByPageId(pageId: string, pagination: PaginationOptions) {
     const query = this.db
       .selectFrom('pageHistory')

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

@@ -280,6 +280,7 @@ export interface PageHistory {
   createdAt: Generated<Timestamp>;
   icon: string | null;
   id: Generated<string>;
+  kind: string | null;
   lastUpdatedAiChatId: string | null;
   lastUpdatedById: string | null;
   lastUpdatedSource: string | null;

apps/server/src/integrations/queue/constants/queue.interface.ts

@@ -20,6 +20,10 @@ export interface IStripeSeatsSyncJob {
 
 export interface IPageHistoryJob {
   pageId: string;
+  // #370 — intentionality tier the worker stamps on the snapshot. All jobs on
+  // this queue are trailing idle-flush autosnapshots, so this is 'idle' (absent
+  // → treated as 'idle' by the processor).
+  kind?: 'idle';
 }
 
 /**

apps/server/test/integration/ai-chat-attach.int-spec.ts

@@ -30,11 +30,13 @@ import {
  * tees the SSE frames into it via `consumeSseStream` while stamping the DB row id
  * via `generateMessageId` (both gated on runId + the resumable flag).
  *
- * Proven here: a finished run's replay is the full frame sequence incl `[DONE]`
- * with `start.messageId` == the seeded DB row id; the anchor check (invariant 6);
- * an attach opened BEFORE the first frame follows the live stream from frame 0; an
- * explicit stop surfaces `{"type":"abort"}` + `[DONE]` + end to the subscriber;
- * and the legacy (non-run) path tees nothing.
+ * Proven here (tail-only #491): a finished run attached at its persisted frontier
+ * N_final delivers only the TAIL past N (a synthetic `start` carrying the run-fact
+ * + the terminal `finish`/`[DONE]`) — the step content below N lives in the seeded
+ * DB row, NOT the ring; the anchor check (invariant 6); an attach opened BEFORE the
+ * first frame follows the live stream from frame 0; an explicit stop surfaces
+ * `{"type":"abort"}` + `[DONE]` + end to the subscriber; and the legacy (non-run)
+ * path tees nothing.
  */
 
 const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
@@ -133,14 +135,16 @@ function liveSink(): {
   };
 }
 
-// The SSE `start` frame carries the message id; pull it out of a `data: {...}`.
-function parseStartMessageId(frames: string[]): string | undefined {
+// Parse the first `start` frame's JSON out of a `data: {...}` sequence.
+function parseStartFrame(
+  frames: string[],
+): { messageId?: string; messageMetadata?: any } | undefined {
   for (const f of frames) {
     const m = /^data: (\{.*\})\s*$/m.exec(f.trim());
     if (!m) continue;
     try {
       const json = JSON.parse(m[1]);
-      if (json.type === 'start') return json.messageId;
+      if (json.type === 'start') return json;
     } catch {
       /* not this frame */
     }
@@ -270,7 +274,7 @@ describe('AiChatService run-stream attach [integration]', () => {
     await destroyTestDb();
   });
 
-  it('run-wrapped: replay is the full frame sequence incl [DONE], start.messageId == the seeded DB row id', async () => {
+  it('run-wrapped, tail-only: a finished run at N_final delivers the run-fact start + finish/[DONE]; the step content lives in the seeded row', async () => {
     const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
     const registry = new AiChatStreamRegistryService();
     const runService = new AiChatRunService(runRepo, {
@@ -300,27 +304,37 @@ describe('AiChatService run-stream attach [integration]', () => {
         );
       });
       const rowId = await assistantRowId(chatId);
+      // The client reads its persisted step frontier N from the seeded row.
+      const row: any = await msgRepo.findById(rowId, workspaceId);
+      const nFinal = row.metadata.stepsPersisted as number;
+      expect(nFinal).toBe(1); // a single finished step
+      // The step content is in the SEEDED row (parts/content), not the ring.
+      expect(JSON.stringify(row.metadata.parts)).toContain('Hello');
 
-      // Finished-run replay with expect=live + the correct anchor.
+      // Attach at N_final with the correct anchor: the tail past step 1 is just
+      // the terminal frames; step 0's 'Hello' is BELOW the frontier (seeded).
       const sink = liveSink();
-      const att = await registry.attach(chatId, true, rowId, sink.cb);
+      const att = await registry.attach(chatId, rowId, nFinal, sink.cb);
       expect(att).not.toBeNull();
       expect(att!.finished).toBe(true);
-      // The tee captured frames (consumeSseStream was wired).
-      expect(att!.replay.length).toBeGreaterThan(0);
-      // generateMessageId stamped the DB row id onto the streamed start frame.
-      expect(parseStartMessageId(att!.replay)).toBe(rowId);
-      // The full sequence includes the streamed text and the terminal marker.
-      const joined = att!.replay.join('');
-      expect(joined).toContain('Hello');
+      // The synthetic start frame carries the run-fact (runId/chatId), the source
+      // of the run-fact on re-attach.
+      const start = parseStartFrame(att!.replay);
+      expect(start?.messageMetadata).toMatchObject({
+        runId: box.runId,
+        chatId,
+      });
+      // The terminal marker is delivered so the client's SDK closes the stream.
       expect(att!.replay.some((f) => f.includes('[DONE]'))).toBe(true);
+      // 'Hello' (step 0, below the frontier) is NOT re-streamed — it is seeded.
+      expect(att!.replay.some((f) => f.includes('Hello'))).toBe(false);
     } finally {
       registry.onModuleDestroy();
       await cleanup();
     }
   });
 
-  it('anchor mismatch with expect=live returns null (invariant 6)', async () => {
+  it('anchor mismatch returns null (invariant 6)', async () => {
     const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
     const registry = new AiChatStreamRegistryService();
     const runService = new AiChatRunService(runRepo, {
@@ -347,7 +361,7 @@ describe('AiChatService run-stream attach [integration]', () => {
       const sink = liveSink();
       // A foreign anchor must NOT replay this run's transcript.
       expect(
-        await registry.attach(chatId, true, 'a-different-run-row', sink.cb),
+        await registry.attach(chatId, 'a-different-run-row', 1, sink.cb),
       ).toBeNull();
     } finally {
       registry.onModuleDestroy();
@@ -376,8 +390,11 @@ describe('AiChatService run-stream attach [integration]', () => {
     try {
       // Attach while the entry exists (opened at begin) but before any frame.
       const sink = liveSink();
-      const att = (await registry.attach(chatId, false, undefined, sink.cb))!;
-      expect(att.replay).toEqual([]); // nothing streamed yet -> replay from 0
+      const att = (await registry.attach(chatId, undefined, 0, sink.cb))!;
+      // Nothing streamed yet -> the tail is just the synthetic start frame; the
+      // whole live stream (start..DONE) follows via onFrame after start().
+      expect(att.replay).toHaveLength(1);
+      expect(att.replay[0]).toContain('"type":"start"');
       att.start(); // go live (drains nothing, then follows)
 
       // Now emit the whole turn.
@@ -448,7 +465,7 @@ describe('AiChatService run-stream attach [integration]', () => {
     });
     try {
       const sink = liveSink();
-      const att = (await registry.attach(chatId, false, undefined, sink.cb))!;
+      const att = (await registry.attach(chatId, undefined, 0, sink.cb))!;
       att.start();
 
       // Give streamText a beat to begin consuming the partial output.
@@ -526,7 +543,9 @@ describe('AiChatService run-stream attach [integration]', () => {
       expect(entry).toBeDefined();
       expect(entry.finished).toBe(true);
       const sink = liveSink();
-      expect(await registry.attach(chatId, false, undefined, sink.cb)).toBeNull();
+      // Finished with an EMPTY ring (aborted before any frame) -> null -> the
+      // client degrades to poll instead of hanging on an empty stream.
+      expect(await registry.attach(chatId, undefined, 0, sink.cb)).toBeNull();
     } finally {
       registry.onModuleDestroy();
       await cleanup();
@@ -556,8 +575,8 @@ describe('AiChatService run-stream attach [integration]', () => {
       });
       const sink = liveSink();
       // No entry was ever opened; attach always yields null.
-      expect(await registry.attach(chatId, false, undefined, sink.cb)).toBeNull();
-      expect(await registry.attach(chatId, true, 'anything', sink.cb)).toBeNull();
+      expect(await registry.attach(chatId, undefined, 0, sink.cb)).toBeNull();
+      expect(await registry.attach(chatId, 'anything', 1, sink.cb)).toBeNull();
     } finally {
       registry.onModuleDestroy();
       await cleanup();

apps/server/test/integration/page-history-idle-pipeline.int-spec.ts

@@ -0,0 +1,162 @@
+import { randomUUID } from 'node:crypto';
+import { Queue, Worker } from 'bullmq';
+import { PersistenceExtension } from '../../src/collaboration/extensions/persistence.extension';
+
+/**
+ * #370 — integration property of the idle-snapshot pipeline against REAL BullMQ.
+ *
+ * This is deliberately NOT a unit test of computeHistoryJob (that lives in
+ * compute-history-job.spec.ts). The point here is the OBSERVABLE end-to-end
+ * behaviour of the production `enqueuePageHistory` remove-then-add debounce
+ * driving a real Redis-backed delayed queue + worker (the #431→#439 class: a
+ * locally-correct function whose queue/timer property was never exercised):
+ *
+ *   - a CONTINUOUS burst of stores lasting several caps yields periodic idle
+ *     snapshots — at least one per max-wait cap, NOT one-per-store; and
+ *   - an INTERMITTENT burst (a few stores, then quiet) yields exactly ONE
+ *     trailing snapshot.
+ *
+ * We shrink the idle windows to milliseconds (jest.mock of collaboration
+ * constants) so real BullMQ delayed jobs actually promote within the test —
+ * fake timers cannot advance Redis's own delayed-set clock, so the intervals
+ * must be real but tiny. The production method under test is called verbatim.
+ */
+
+// NOTE: jest.mock is hoisted above the module's const initializers, so its
+// factory cannot close over MAX_WAIT_MS/INTERVAL_MS — the literals are inlined
+// here and MUST stay in sync with the consts below (a single source of truth is
+// impossible across the hoist boundary).
+jest.mock('../../src/collaboration/constants', () => {
+  const actual = jest.requireActual('../../src/collaboration/constants');
+  return {
+    ...actual,
+    IDLE_MAX_WAIT_USER: 300,
+    IDLE_MAX_WAIT_AGENT: 300,
+    IDLE_INTERVAL_USER: 1000,
+    IDLE_INTERVAL_AGENT: 1000,
+  };
+});
+
+// Mirrors the mocked IDLE_MAX_WAIT_* above (IDLE_INTERVAL_* is 1000 > this, so
+// the max-wait ceiling is what actually governs the trailing delay).
+const MAX_WAIT_MS = 300;
+
+const REDIS_CONNECTION = {
+  host: process.env.TEST_REDIS_HOST ?? '127.0.0.1',
+  port: Number(process.env.TEST_REDIS_PORT ?? 6379),
+};
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+describe('#370 idle-snapshot pipeline (real BullMQ)', () => {
+  let queue: Queue;
+  let worker: Worker;
+  let extension: PersistenceExtension;
+  // Every processed snapshot, tagged by pageId so the two scenarios stay isolated.
+  const processed: Array<{ pageId: string; kind: string; at: number }> = [];
+  const queueName = `history-idle-int-${randomUUID()}`;
+
+  beforeAll(async () => {
+    queue = new Queue(queueName, {
+      connection: REDIS_CONNECTION,
+      // Mirror the production default (BullModule.forRoot removeOnComplete): the
+      // enqueue idiom relies on the jobId being freed once a job completes so the
+      // next burst can re-arm the same id.
+      defaultJobOptions: { removeOnComplete: true, removeOnFail: true },
+    });
+    await queue.waitUntilReady();
+
+    worker = new Worker(
+      queueName,
+      async (job) => {
+        processed.push({
+          pageId: job.data?.pageId,
+          kind: job.data?.kind,
+          at: Date.now(),
+        });
+      },
+      { connection: REDIS_CONNECTION },
+    );
+    await worker.waitUntilReady();
+
+    // Construct the real extension; only historyQueue (5th ctor arg) and the
+    // internal idleBurstStart map are exercised by enqueuePageHistory, so the
+    // other collaborators can be null — the constructor only assigns fields.
+    extension = new PersistenceExtension(
+      null as any, // pageRepo
+      null as any, // pageHistoryRepo
+      null as any, // db
+      null as any, // aiQueue
+      queue as any, // historyQueue
+      null as any, // notificationQueue
+      null as any, // collabHistory
+      null as any, // transclusionService
+    );
+  });
+
+  afterAll(async () => {
+    // Force-close and fully drain so no BullMQ background activity (delayed-set
+    // polling, blocking BRPOPLPUSH) bleeds into later suites in this single
+    // shared jest worker (maxWorkers: 1).
+    await worker?.close(true).catch(() => undefined);
+    await queue?.obliterate({ force: true }).catch(() => undefined);
+    await queue?.close();
+    // Let the redis sockets settle before the next suite starts.
+    await sleep(150);
+  });
+
+  const arm = (pageId: string) =>
+    (extension as any).enqueuePageHistory({ id: pageId }, 'user');
+
+  it('continuous burst over several caps → periodic idle snapshots (≥1 per cap, not one-per-store)', async () => {
+    const pageId = randomUUID();
+    const runMs = 6 * MAX_WAIT_MS; // ~6 caps of unbroken editing
+    // Store cadence that does NOT evenly divide the cap: real hocuspocus stores
+    // are not aligned to cap boundaries, so a boundary job promotes in the gap
+    // before the next store's remove(). A cap-aligned cadence would instead land
+    // a store exactly on every boundary and lose the snapshot to the documented
+    // remove-vs-active race — an artefact of the test clock, not the pipeline.
+    const stepMs = 70;
+    const stores = Math.floor(runMs / stepMs);
+
+    const start = Date.now();
+    let count = 0;
+    while (Date.now() - start < runMs) {
+      await arm(pageId);
+      count++;
+      await sleep(stepMs);
+    }
+    // Let the final armed job flush.
+    await sleep(2 * MAX_WAIT_MS);
+
+    const snaps = processed.filter((p) => p.pageId === pageId);
+
+    // Every autosnapshot is an idle-kind row.
+    expect(snaps.every((s) => s.kind === 'idle')).toBe(true);
+    // Periodic: at least one per cap over a multi-cap burst (lower-bounded loosely
+    // to stay robust; the property is "fires at least every cap", not a single
+    // trailing snapshot).
+    expect(snaps.length).toBeGreaterThanOrEqual(3);
+    // But NOT one-per-store: ~`stores` stores were issued; the debounce must
+    // collapse them to a small multiple of the cap count, nowhere near per-store.
+    expect(snaps.length).toBeLessThanOrEqual(Math.ceil(stores / 2));
+  });
+
+  it('intermittent burst (a few stores, then quiet) → exactly ONE trailing snapshot', async () => {
+    const pageId = randomUUID();
+
+    // A short burst well within a single cap window, then silence.
+    await arm(pageId);
+    await sleep(40);
+    await arm(pageId);
+    await sleep(40);
+    await arm(pageId);
+
+    // Wait comfortably past the cap so the single pending trailing job fires.
+    await sleep(4 * MAX_WAIT_MS);
+
+    const snaps = processed.filter((p) => p.pageId === pageId);
+    expect(snaps).toHaveLength(1);
+    expect(snaps[0].kind).toBe('idle');
+  });
+});

docs/features/page-work-time_design.md

@@ -387,7 +387,7 @@ bucketByDay(sessions, tz):
 факты ниже — ground truth, можно дозапросить файлы через gitea MCP по указанному SHA):
 
 - `page_history.kind` — `varchar(20)`, NULLABLE, БЕЗ дефолта (migration
-  `20260705T120000-page-history-kind.ts`). Домен: `manual`/`agent`/`idle`/`boundary`;
+  `20260707T120000-page-history-kind.ts`). Домен: `manual`/`agent`/`idle`/`boundary`;
   legacy `null` = автосейв (`collaboration/constants.ts`, `PageHistoryKind`).
 - `kind` УЖЕ включён в `PageHistoryRepo.baseFields` (`page-history.repo.ts`) — читается всеми
   выборками истории. `saveHistory({kind})` и `updateHistoryKind(id, kind)` существуют.