test(comments): не-вакуумные тесты audit-swallow + dominant-run; CHANGELOG (#512 ревью)

DO1: тест audit-swallow был вакуумен (log() — fire-and-forget void persist(),
.not.toThrow() зелен независимо от try/catch). Теперь: после failInsert()
шпионим Logger.warn + слушаем unhandledRejection, флашим микро/макротаски,
ассертим warn=1 и 0 floating-rejection. Mutation: убрать try/catch у persist
→ красный.
DO2: dominant-run тест не отличал longest от first (самый длинный ран был и
первым). Перестроено: короткий plain ран впереди, длинный bold — следом →
ассерт bold:true держится только при genuine longest; +тест tie→first.
Mutation: reduce→segments[0] → красный.
DO3: CHANGELOG [Unreleased] — 3 записи по #496.
DO4: убран no-op override insertAttrs.comment (=dominant.markAttrs, а он и есть
attributes['comment'] — сегменты фильтруются по нему; {...attributes,comment:
attributes.comment}={...attributes}); коммент про 'defensive re-assert' исправлен.

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

.env.example

@@ -225,11 +225,26 @@ MCP_DOCMOST_PASSWORD=
 
 # Silence timeout (ms) for EXTERNAL-MCP transport ONLY (not the chat provider).
 # Tighter than AI_STREAM_TIMEOUT_MS so a byte-silent/hung MCP server is broken in
-# ~1 min instead of 15. Note it also cuts a legitimately long but byte-silent
-# single tool call (a slow crawl that emits nothing until done) and an SSE
-# transport idling >1 min BETWEEN tool calls. Default 60000 (1 min).
+# ~1 min instead of 15. It cuts a legitimately long but byte-silent single tool
+# call (a slow crawl that emits nothing until done) on the HTTP (streamable)
+# transport, which opens a fresh request per call. The SSE transport — one
+# long-lived body across many calls — is NO LONGER governed by this timeout
+# (as of #489): its idle-BETWEEN-calls window has its own, raised bodyTimeout,
+# AI_MCP_SSE_BODY_TIMEOUT_MS below. Default 60000 (1 min).
 # AI_MCP_STREAM_TIMEOUT_MS=60000
 
+# bodyTimeout (ms) for the EXTERNAL-MCP SSE transport ONLY (#489). The SSE
+# transport holds ONE response body open across many tool calls, so undici's
+# bodyTimeout (time between body bytes) counts the LEGITIMATE silence BETWEEN the
+# model's tool calls, not just a hung single call. At the tight 1-min silence
+# timeout above, a normal >1-min gap between calls would break the SSE socket and
+# the cache would serve a dead client until TTL — so the SSE transport gets its
+# OWN, RAISED bodyTimeout. A single stuck call is still bounded by the per-call
+# cap (AI_MCP_CALL_TIMEOUT_MS), and a socket that does break is healed by the
+# in-run transport-error retry. The HTTP (streamable) transport keeps the tight
+# timeout. Default 600000 (10 min).
+# AI_MCP_SSE_BODY_TIMEOUT_MS=600000
+
 # Total wall-clock cap (ms) for ONE external MCP tool call (app-level, not
 # transport). Aborts a tool that keeps the socket warm (SSE heartbeats / trickle)
 # but never returns a result — which the silence timeout above never breaks.

CHANGELOG.md

@@ -129,6 +129,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **A drifted comment suggestion can be re-synced instead of failing forever
+  with a 409.** A suggestion whose stored anchor no longer matched the live
+  document used to reject every apply attempt with an unrecoverable conflict; a
+  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)
+
 - **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
   images as a row that wraps onto the next line on narrow screens. The row is
@@ -336,6 +343,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **A chat with one malformed message part no longer 500s on every turn, and a
+  failed send no longer duplicates the user's message.** Incoming client parts
+  are now whitelisted to `text` (a forged tool-result part can no longer reach
+  the persisted history or the model context), and the turn is converted BEFORE
+  the user row is inserted, so a mid-flight failure cannot leave a duplicate
+  user row that a retry then compounds. A single part that still fails to convert
+  degrades to a `[tool context omitted]` marker on that one row instead of
+  bricking the whole chat. (#489)
+- **A transport drop to an external MCP server now heals within the same turn.**
+  On an undici transport error, a read-only MCP tool reconnects its server and
+  retries once within the run; a write is never auto-retried (it may already have
+  applied). One flapping server no longer nulls the shared client cache, so other
+  servers' cached clients are untouched. The SSE transport also gets a raised
+  body-timeout so a legitimate >1-min idle between the model's tool calls no
+  longer breaks a long-lived SSE socket (new `AI_MCP_SSE_BODY_TIMEOUT_MS`, default
+  10 min; see `.env.example`). (#489)
+- **Decisions on comment suggestions now leave a durable audit record.**
+  Applying or dismissing a comment suggestion hard-deletes the (childless)
+  subject comment, so the only surviving trace of who decided what is the audit
+  event — but the audit trail was wired to a Noop service that silently
+  swallowed every event. The trail is now DB-backed, so
+  `comment.suggestion_applied` / `comment.suggestion_dismissed` (and the other
+  comment-decision events) persist to the `audit` table and can be reviewed
+  after the comment is gone. A persistence failure is still swallowed with a
+  warning so it never breaks the originating request. (#496)
+- **Applying a comment suggestion no longer strips the replaced run's inline
+  formatting.** The suggested text was re-inserted carrying only the comment
+  anchor mark, silently dropping bold/italic/code/link on the affected run; the
+  prevailing formatting of the replaced run is now carried onto the applied
+  text. (#496)
 - **The server no longer runs out of heap during long autonomous agent runs.** A
   new pnpm patch on `ai@6.0.134` stops the SDK from building a cumulative
   snapshot of the ENTIRE turn text on every streamed text-delta when no output

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

@@ -86,19 +86,11 @@ const MIN_HEIGHT = 400;
 // Margin kept between the window and the viewport edges while dragging.
 const EDGE_MARGIN = 8;
 
-// #184 phase 1.5 / #430: backstop for the degraded-poll fallback. The poll is
-// armed when a resume attempt could not attach to the live run and disarmed by the
-// thread on settle / local stream; this cap is the ONLY backstop against an endless
-// tick (a stuck 'streaming' row before the boot-sweep, or a user-tail 204 with no
-// run).
-//
-// #430: measured from RUN ACTIVITY, not from arm-time. A real autonomous run takes
-// 11-25 min — longer than a fixed 10-min-from-start cap, which used to cut the poll
-// off mid-run. Instead we cap on INACTIVITY: keep polling as long as the run is
-// still making progress (its persisted rows keep changing), and only give up after
-// this long with NO new activity. A genuinely stuck run produces no row changes, so
-// the idle cap still bounds it; a long-but-progressing run polls to completion.
-const DEGRADED_POLL_IDLE_MAX_MS = 10 * 60_000;
+// #184 phase 1.5 / #430 / #488: the degraded-poll fallback. The window owns only
+// a DUMB 2.5s timer, gated by an armed flag; the THREAD's run-lifecycle FSM owns
+// arm/disarm AND the inactivity cap that turns a stuck run into a `stalled` banner
+// (#488 commit 4a — the cap moved into the thread so polling->stalled is a single
+// FSM transition; the window no longer silently stops polling at the cap).
 
 /** Compact token formatter: 1.2M / 3.4k / 950. */
 function formatTokens(n: number): string {
@@ -259,17 +251,13 @@ export default function AiChatWindow() {
     [roles],
   );
 
-  // #184 phase 1.5: degraded-poll fallback (replaces the F4/F5/F7 latches). When
-  // ChatThread could not attach to a still-running run it arms this via
-  // onResumeFallback(true); the thread disarms it on settle / local stream. The
-  // window only OWNS the timer (armedAtRef stamps when it was armed for the cap).
+  // #184 phase 1.5 / #488: degraded-poll fallback. ChatThread's FSM arms this via
+  // onResumeFallback(true) when it enters a poll-bearing recovery (attach 204 /
+  // starved finish / stop) and disarms it on settle / local stream / stalled. The
+  // window owns ONLY the dumb 2.5s timer; the THREAD owns arm/disarm AND the
+  // inactivity cap (a stuck run -> the thread's `stalled` banner disarms this).
   const [degradedPoll, setDegradedPoll] = useState(false);
-  // #430: timestamp of the LAST run activity while the poll is armed — stamped on
-  // arm and re-stamped whenever the polled rows change (see the effect below). The
-  // idle cap is measured from this, so a long-but-progressing run keeps polling.
-  const lastActivityAtRef = useRef(0);
   const onResumeFallback = useCallback((active: boolean): void => {
-    if (active) lastActivityAtRef.current = Date.now();
     setDegradedPoll(active);
   }, []);
   // Reset the degraded poll whenever the open chat changes: it is scoped to the
@@ -281,33 +269,17 @@ export default function AiChatWindow() {
   const { data: messageRows, isLoading: messagesLoading } =
     useAiChatMessagesQuery(
       activeChatId ?? undefined,
-      // DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed
-      // and while the run is still active (#430: under the INACTIVITY cap, not a
-      // fixed-from-start cap); otherwise off. NO error checks (TanStack v5 resets
-      // fetchFailureCount each fetch, so consecutive errors are not expressible —
-      // and the poll must survive a server restart) and NO tail checks (the
-      // settled/local-stream semantics live in ChatThread, which disarms via
-      // onResumeFallback(false)). The idle cap is the only backstop.
-      () =>
-        degradedPoll === true &&
-        Date.now() - lastActivityAtRef.current < DEGRADED_POLL_IDLE_MAX_MS
-          ? 2500
-          : false,
+      // 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.
       windowOpen,
     );
 
-  // #430: re-stamp the activity clock whenever the polled rows change while the
-  // poll is armed. TanStack keeps the same `messageRows` reference across refetches
-  // that return deep-equal data (structural sharing), so a new reference means the
-  // run genuinely progressed — which extends the inactivity cap above. A stuck run
-  // yields no reference change, so the cap eventually fires and stops the poll.
-  useEffect(() => {
-    if (degradedPoll) lastActivityAtRef.current = Date.now();
-  }, [degradedPoll, messageRows]);
-
   // #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/services/ai-chat-service.ts

@@ -57,6 +57,25 @@ export async function stopRun(
   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
+ * and to VERIFY after a supersede mismatch (an observer following a superseded
+ * run asks for the latest run and follows it). Returns the latest run row (with
+ * its `id` and `status`) and its projected assistant message, or `run: null` when
+ * the chat has never had a run. Owner-gated server-side.
+ */
+export async function getRun(chatId: string): Promise<{
+  run: { id: string; status: string } | null;
+  message: IAiChatMessageRow | null;
+}> {
+  const req = await api.post<{
+    run: { id: string; status: string } | null;
+    message: IAiChatMessageRow | null;
+  }>("/ai-chat/run", { chatId });
+  return req.data;
+}
+
 /**
  * Resolve the chat bound to a document (the current user's most-recent chat
  * created on that page), or null when there is none. Drives auto-open-on-page.

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

@@ -0,0 +1,183 @@
+# AI-chat run-lifecycle FSM — design spec (#488)
+
+This is the written design that `run-fsm.ts` implements. It ships in the PR (issue
+#488 commit 1: "the spec is written FIRST and enters the PR"). It has four parts:
+(1) the event × state transition table, (2) the map of every `chat-thread.tsx` ref
+to {FSM state | FSM context | stays data}, (3) the run-fact protocol, (4) the
+invariants.
+
+The reducer is a **pure function** `reduce(machine, event) → machine`. The returned
+machine carries the **command effects** for that transition; a thin runtime in
+`chat-thread.tsx` dispatches events and executes effects. Because it is pure, the
+whole machine is enumerable and unit-tested directly (event × state → next state is
+the observable property) — see `run-fsm.test.ts`.
+
+---
+
+## 1. Event × state transition table
+
+Phases: `idle | sending | streaming | attaching | reconnecting(attempt,failed) |
+polling(reason) | stalled | stopping | superseding | error(kind)`.
+Context (orthogonal): `epoch`, `ownership: local|observer`, `runFact: {runId}|null`,
+`liveFollow` (are we following a live run we locally streamed — the reconnect
+ladder — vs a one-shot mount-attach resume? both are `observer`, but a live-follow
+drop RE-ENTERS the ladder (#488 commit 3) while a mount-resume drop polls).
+
+Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`.
+
+| Event (source) | From phase(s) | → To phase | Effects / ctx |
+|---|---|---|---|
+| `SEND_LOCAL` (user send) | idle, error, polling, stalled, reconnecting | sending **†** | `[cancelReconnect, disarmPoll]`, ownership=local |
+| `STREAM_START{runId}` (SDK `start` metadata) | sending, attaching, reconnecting, superseding | streaming | `[cancelReconnect, disarmPoll]`, runFact←runId |
+| `FINISH_CLEAN` (onFinish clean) | streaming, … | idle | `[disarmPoll, cancelReconnect]`, runFact←null |
+| `FINISH_ABORT` (onFinish isAbort) | streaming, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4 exits stopping by this DATA) |
+| `FINISH_DISCONNECT` (observer, NOT liveFollow) | streaming(observer) | polling(disconnect-visible) | `[armPoll]` (a mount-resume drop polls) |
+| `FINISH_DISCONNECT{hasVisibleContent}` (local drop OR liveFollow) | streaming | reconnecting(1) **†** *iff runFact\|liveFollow* | `[scheduleReconnect(1)]` (+`armPoll` if visible), ownership=observer, liveFollow=true (commit 3: repeatable) |
+| `FINISH_DISCONNECT` (no runFact, not liveFollow) | streaming | idle | runFact←null (plain terminal "connection lost") |
+| `STREAM_INCOMPLETE{reason}` (observer starved/torn clean finish) | streaming(observer) | polling(reason) | `[armPoll(reason)]` |
+| `FINISH_ERROR{kind}` (onFinish isError) | any | error(kind) | `[disarmPoll, cancelReconnect]`, runFact←null |
+| `STREAM_START{runId}` (first assistant frame of a local turn) | sending | streaming | runFact←runId, `[cancelReconnect, disarmPoll]` |
+| `ATTACH_START{runId}` (mount resume) | **idle only** (F2) | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId; ignored from any non-idle phase |
+| `ATTACH_LIVE` (attach GET 2xx) | attaching | streaming | — |
+| `ATTACH_NONE` (attach GET 204/err/throw) | attaching | polling(attach-none) | `[armPoll(attach-none)]` |
+| `RECONNECT_ATTEMPT{n}` (backoff timer) | reconnecting | reconnecting(n) **†** | `[resumeStream]` |
+| `RECONNECT_ATTACHED` (reconnect GET 2xx) | reconnecting | streaming | `[cancelReconnect, disarmPoll]` — **counter reset** (commit 3) |
+| `RECONNECT_NONE` (reconnect GET 204/err), attempt<MAX | reconnecting | reconnecting(n+1) **†** | `[armPoll(attach-none), scheduleReconnect(n+1)]` |
+| `RECONNECT_NONE`, attempt=MAX | reconnecting | reconnecting(MAX, failed) | `[armPoll(reconnect-exhausted)]` |
+| `RETRY` (manual, failed banner) | reconnecting(failed) | reconnecting(1) **†** | `[resumeStream]` |
+| `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) |
+| `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) |
+| `SUPERSEDE_REQUESTED{targetRunId}` (interrupt+send) | streaming, reconnecting, polling, error | superseding **†** | `[supersede(target), cancelReconnect, disarmPoll]` |
+| `SUPERSEDE_READY{runId}` (CAS ok) | superseding | streaming | ownership=local, runFact←runId |
+| `SUPERSEDE_MISMATCH{currentRunId}` (409 SUPERSEDE_TARGET_MISMATCH) | superseding | error(supersede-mismatch) | `[postRun(verify)]`, runFact←currentRunId |
+| `SUPERSEDE_TIMEOUT` (409 SUPERSEDE_TIMEOUT) | superseding | error(supersede-timeout) | — (composer keeps text; no auto-retry) |
+| `SUPERSEDE_INVALID` (409 SUPERSEDE_INVALID) | superseding | error(supersede-invalid) | — |
+| `RUN_ALREADY_ACTIVE{activeRunId}` (409 A_RUN_ALREADY_ACTIVE, plain POST) | sending | error(run-already-active) | runFact←activeRunId (composer offers supersede; NO auto-retry) |
+| `DISPOSE` (unmount) | any | idle **†** | `[abortAttach, cancelReconnect, disarmPoll]` (I1/I5 — epoch++ kills late callbacks) |
+
+**`stopping` honors any finish (re-review MEDIUM):** BEFORE the epoch filter, a
+stream finish (`FINISH_*`/`STREAM_INCOMPLETE`) arriving in phase `stopping` exits
+`stopping -> idle` regardless of generation. A plain Stop has no successor stream,
+so the aborted stream's finish IS the expected end (I4 exit by data) — and it
+carries the PRE-stop generation (STOP_REQUESTED bumped the epoch), so the filter
+would otherwise strand the machine in `stopping` (no idle-cap covers it). The filter
+stays in force for `superseding` (that is the F1 supersede drop).
+
+**Epoch filter (I1):** the reducer then drops any event carrying an `epoch` that
+does not equal the current `ctx.epoch`. Outcome events (`STREAM_START`, `ATTACH_*`,
+`RECONNECT_*`, `SUPERSEDE_*`, **`FINISH_*`/`STREAM_INCOMPLETE`**, `RUN_FACT`) are
+stamped with the generation the corresponding STREAM started under (the runtime
+holds a per-owned-stream `turnEpoch`); trigger events (user actions, fresh
+disconnects) carry no epoch. **F1:** this is what makes a SUPERSEDED stream's late
+`onFinish` (a dead stream A closing after the CAS started stream B) get dropped, so
+A cannot drive the live new run into a false reconnect or reset its run-fact. The
+supersede path additionally ABORTS A and starts B only from A's onFinish (a
+microtask), because ai@6 `AbstractChat.makeRequest` corrupts overlapping streams
+(A's `finally` reads then nulls the shared `activeResponse`).
+
+**Removed events (scope-cut, internal review):** `RUN_SUPERSEDED` (a ghost feature —
+never dispatched; the observer-superseded case is handled by the degraded poll,
+which follows the latest rows regardless of runId), `RECONNECT_BEGIN` (reconnect is
+entered by `FINISH_DISCONNECT`), and `POLL_ACTIVITY` (the window's activity clock was
+removed when the idle-cap moved into the thread). The reducer and this table now
+share exactly the dispatched event set.
+
+### 409-code → event map (the real #487 contract consumed here)
+
+| Server response | Event dispatched | error kind → banner |
+|---|---|---|
+| 409 `A_RUN_ALREADY_ACTIVE` (+ body.activeRunId) | `RUN_ALREADY_ACTIVE{activeRunId}` | run-already-active → "already answering / interrupt & send" |
+| 409 `SUPERSEDE_TARGET_MISMATCH` (+ body.activeRunId) | `SUPERSEDE_MISMATCH{currentRunId}` | supersede-mismatch → verify via /run |
+| 409 `SUPERSEDE_TIMEOUT` | `SUPERSEDE_TIMEOUT` | supersede-timeout → "couldn't interrupt in time, resend" |
+| 409 `SUPERSEDE_INVALID` | `SUPERSEDE_INVALID` | supersede-invalid → "couldn't interrupt this run" |
+| 503 `A_RUN_BEGIN_FAILED` | `FINISH_ERROR{begin-failed}` | begin-failed → "could not start, temporary" |
+
+---
+
+## 2. Ref-map — every `chat-thread.tsx` ref → its new home  (MIGRATION RESOLVED)
+
+The migration is COMPLETE: the 13 run-lifecycle FLAGS below are GONE from
+`chat-thread.tsx` (collapsed into FSM phase/ctx/effects, or deleted). What remains
+are identity/data mirrors, effect-owned controllers/timers, and ONE React-liveness
+bit — none of which is a run-lifecycle flag, so the post-merge "no new flags" rule
+holds. **Pending column: empty.**
+
+| # | Old ref | Resolved to | Where now |
+|---|---|---|---|
+| 1 | `reconcileTailRef` | **FSM phase** | reconcile-merge gated on `phase ∈ {polling, reconnecting, stopping}` |
+| 2 | `noStreamHandledRef` | **FSM epoch (I1)** | the attach outcome's epoch guard drops the stale/second outcome |
+| 3 | `onNoActiveStreamRef` | **FSM event** | transport → `handleAttachOutcome` dispatches `ATTACH_NONE`/`RECONNECT_NONE` |
+| 4 | `onReconnectAttachedRef` | **FSM event** | transport dispatches `ATTACH_LIVE` / `RECONNECT_ATTACHED` |
+| 5 | `resumedTurnRef` + `resumedTurn` state | **FSM ctx `ownership`** | `ownership==='observer'` ⇒ never flush; hides "Send now" |
+| 6 | `reconnectStateRef` + `reconnectState` state | **FSM phase** | `reconnecting(attempt,failed)` renders the banner |
+| 7 | `reconnectTimerRef` | **effect-owned timer** | owned by `scheduleReconnect`/`cancelReconnect` effects (not a flag) |
+| 8 | `flushOnAbortRef` | **DELETED** | the stop→flush dance is replaced by the CAS supersede (commit 5) |
+| 9 | `interruptNextSendRef` | **DELETED** | the server injects the interrupt note from the supersede itself |
+| 10 | `supersedeRetryRef` | **DELETED** (commit 5) | the client 409 retry ladder is gone; CAS supersede replaces it |
+| 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 |
+| 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 |
+| NEW | `idleCapTimerRef` | **effect-owned timer** | the stalled inactivity cap → `POLL_IDLE_CAP` (commit 4a); not a flag |
+
+Net: the 13 lifecycle flags (#1–#13) are eliminated: **8** → FSM phase/ctx/epoch/event
+(#1–#6, #11, #13), **3** deleted (#8/#9/#10), **`reconnectTimerRef` (#7)** becomes an
+effect-owned controller, and **`mountedRef` (#12)** is retained as React liveness
+(8 + 3 + 1 + 1 = 13). (`attachAbortRef` (#16) is outside the #1–#13 set — it was
+already an effect-owned controller.) Two effect-owned timers + one send-plumbing data
+ref are added — none is a boolean lifecycle latch.
+
+---
+
+## 3. Run-fact protocol (`runFact: {runId} | null`) — I3
+
+"A run is active" is first-class from the SERVER, not inferred from an assistant
+message. Sources, in the order they update `ctx.runFact`:
+
+1. **Init (mount):** `POST /ai-chat/run { chatId }` → `{ run, message }`. A `run`
+   with a non-terminal `status` seeds `runFact = { runId: run.id }`; a null/terminal
+   run seeds `null`. This is what arms the resume attempt (`ATTACH_START`) — the
+   attempt is armed ONLY on a positive fact (commit 4b: a user-tail with no active
+   run no longer arms a pointless poll on every open).
+2. **Live update:** the `start` stream metadata carries `runId` → `STREAM_START{runId}`.
+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`.
+
+Pessimism rule: a stale-but-positive fact PERMITS entering recovery (attach); the
+204 then cuts it. A fresh negative fact gates recovery OUT immediately.
+
+---
+
+## 4. Invariants
+
+- **I1 — Epoch (generation counter).** Every command-emitting transition bumps
+  `ctx.epoch`; every async outcome event carries its issuing epoch; the reducer
+  drops stale-epoch outcomes. Replaces the one-shot-ref zoo (`noStreamHandledRef`,
+  the flush/interrupt/supersede one-shots, the `mountedRef` late-callback gate).
+- **I2 — Ownership is context, not state.** `local | observer` is orthogonal to the
+  transport phase. The queue flushes ONLY under local ownership; an observer
+  following a detached run never flushes (was `resumedTurnRef`).
+- **I3 — Run-fact is first-class from the server.** Reconnect is entered by the
+  run-fact, not by an assistant message (commit 2). A fresh negative fact cancels
+  recovery.
+- **I4 — Exit `stopping` by DATA.** A terminal row / negative run-fact / terminal
+  finish exits `stopping`, never the stopRun HTTP response (which returns after the
+  abort but before finalization — keying off it would unlock the composer on a 409).
+- **I5 — Dispose protocol.** Command controllers (attach GET, POST /stream, POST
+  /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.
+- **Queue** stays a data structure; flush/interrupt decisions are transitions.

apps/client/src/features/ai-chat/state/run-fsm.test.ts

@@ -0,0 +1,482 @@
+import { describe, it, expect } from "vitest";
+import {
+  reduce,
+  initialMachine,
+  reconnectDelayMs,
+  RECONNECT_MAX_ATTEMPTS,
+  type Machine,
+  type Effect,
+  type Event,
+} from "./run-fsm";
+
+// Drive a sequence of events through the reducer, returning the final machine.
+function run(m: Machine, ...events: Event[]): Machine {
+  return events.reduce(reduce, m);
+}
+function withRunFact(runId = "run-1"): Machine {
+  return {
+    ...initialMachine(),
+    ctx: { epoch: 0, ownership: "local", runFact: { runId }, liveFollow: false },
+  };
+}
+function effectTypes(m: Machine): string[] {
+  return m.effects.map((e) => e.type);
+}
+function hasEffect(m: Machine, type: Effect["type"]): boolean {
+  return m.effects.some((e) => e.type === type);
+}
+
+describe("run-fsm — epoch invariant (I1)", () => {
+  it("drops an outcome carrying a stale epoch", () => {
+    // A command bumps the epoch; an outcome stamped with the OLD epoch is dropped.
+    const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); // epoch 0->1, attaching
+    expect(m0.ctx.epoch).toBe(1);
+    expect(m0.phase.name).toBe("attaching");
+    // A late ATTACH_LIVE from a SUPERSEDED attempt (epoch 0) must NOT drive us.
+    const stale = reduce(m0, { type: "ATTACH_LIVE", epoch: 0 });
+    expect(stale.phase.name).toBe("attaching");
+    expect(stale.effects).toEqual([]);
+  });
+
+  it("applies an outcome carrying the current epoch", () => {
+    const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    const live = reduce(m0, { type: "ATTACH_LIVE", epoch: m0.ctx.epoch });
+    expect(live.phase.name).toBe("streaming");
+  });
+
+  it("an outcome with no epoch is never dropped (trigger events)", () => {
+    const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    const disposed = reduce(m0, { type: "DISPOSE" });
+    expect(disposed.phase.name).toBe("idle");
+    expect(hasEffect(disposed, "abortAttach")).toBe(true);
+  });
+
+  it("every command-transition increments the epoch exactly once", () => {
+    let m = initialMachine();
+    const before = m.ctx.epoch;
+    m = reduce(m, { type: "SEND_LOCAL" });
+    expect(m.ctx.epoch).toBe(before + 1);
+    m = reduce(m, { type: "STOP_REQUESTED" });
+    expect(m.ctx.epoch).toBe(before + 2);
+  });
+});
+
+describe("run-fsm — local turn", () => {
+  it("SEND_LOCAL → sending, local ownership, cancels recovery", () => {
+    const m = reduce(withRunFact(), { type: "SEND_LOCAL" });
+    expect(m.phase.name).toBe("sending");
+    expect(m.ctx.ownership).toBe("local");
+    expect(effectTypes(m)).toEqual(
+      expect.arrayContaining(["cancelReconnect", "disarmPoll"]),
+    );
+  });
+
+  it("STREAM_START adopts the runId into the run-fact and goes streaming", () => {
+    const m = run(initialMachine(), { type: "SEND_LOCAL" });
+    const s = reduce(m, { type: "STREAM_START", runId: "run-9", epoch: m.ctx.epoch });
+    expect(s.phase.name).toBe("streaming");
+    expect(s.ctx.runFact).toEqual({ runId: "run-9" });
+  });
+
+  it("FINISH_CLEAN → idle, run-fact cleared, poll/reconnect disarmed", () => {
+    const streaming = run(initialMachine(), { type: "SEND_LOCAL" }, { type: "STREAM_START", runId: "r" });
+    const done = reduce(streaming, { type: "FINISH_CLEAN" });
+    expect(done.phase.name).toBe("idle");
+    expect(done.ctx.runFact).toBeNull();
+  });
+});
+
+// #488 commit 2 — SSE break BEFORE the first assistant frame must still recover.
+describe("run-fsm — commit 2: reconnect by run-fact, not by assistant message", () => {
+  it("FINISH_DISCONNECT with an active run-fact → reconnecting (even with no visible content)", () => {
+    // Setup-phase break: no assistant frame yet, but a run-fact exists.
+    const streaming = withRunFact("run-2");
+    const m = reduce(streaming, {
+      type: "FINISH_DISCONNECT",
+      hasVisibleContent: false,
+      epoch: streaming.ctx.epoch,
+    });
+    expect(m.phase.name).toBe("reconnecting");
+    if (m.phase.name === "reconnecting") expect(m.phase.attempt).toBe(1);
+    expect(m.ctx.ownership).toBe("observer");
+    expect(hasEffect(m, "scheduleReconnect")).toBe(true);
+    // No visible content -> no poll arm yet (the reconnect ladder rebuilds it).
+    expect(hasEffect(m, "armPoll")).toBe(false);
+  });
+
+  it("FINISH_DISCONNECT WITH visible content also arms the poll", () => {
+    const m = reduce(withRunFact("run-2"), {
+      type: "FINISH_DISCONNECT",
+      hasVisibleContent: true,
+      epoch: 0,
+    });
+    expect(m.phase.name).toBe("reconnecting");
+    expect(hasEffect(m, "armPoll")).toBe(true);
+  });
+
+  it("FINISH_DISCONNECT with NO run-fact → idle (plain connection-lost)", () => {
+    const m = reduce(initialMachine(), {
+      type: "FINISH_DISCONNECT",
+      hasVisibleContent: true,
+      epoch: 0,
+    });
+    expect(m.phase.name).toBe("idle");
+  });
+});
+
+// #488 commit 3 — a SECOND break after a successful re-attach starts a NEW ladder.
+describe("run-fsm — commit 3: repeated reconnect cycles", () => {
+  it("two breaks in a row produce two reconnect cycles (counter resets on attach)", () => {
+    let m = withRunFact("run-3");
+    // First break -> reconnecting(1).
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("reconnecting");
+    // Attempt fires, re-attaches live.
+    m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: 1, epoch: m.ctx.epoch });
+    m = reduce(m, { type: "RECONNECT_ATTACHED", epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("streaming");
+    // SECOND break: the counter was reset, so a fresh ladder starts at attempt 1
+    // (the old one-shot !wasResumed gate would have sent this to silent poll).
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("reconnecting");
+    if (m.phase.name === "reconnecting") expect(m.phase.attempt).toBe(1);
+    expect(hasEffect(m, "scheduleReconnect")).toBe(true);
+  });
+
+  it("a MOUNT-attach observer drop falls to POLL, not the reconnect ladder", () => {
+    // Distinguishes commit 3 from a one-shot resume: an observer that never
+    // live-followed (liveFollow false) polls on a drop.
+    let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch });
+    expect(m.ctx.ownership).toBe("observer");
+    expect(m.ctx.liveFollow).toBe(false);
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: true, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("polling");
+    expect(hasEffect(m, "armPoll")).toBe(true);
+  });
+
+  it("STREAM_INCOMPLETE (observer starved/torn finish) → polling", () => {
+    let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch });
+    m = reduce(m, { type: "STREAM_INCOMPLETE", reason: "starved", epoch: m.ctx.epoch });
+    expect(m.phase).toEqual({ name: "polling", reason: "starved" });
+    expect(hasEffect(m, "armPoll")).toBe(true);
+  });
+
+  it("liveFollow is set on the first local drop and kept across a re-attach", () => {
+    let m = withRunFact("run-3");
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch });
+    expect(m.ctx.liveFollow).toBe(true);
+    m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: 1, epoch: m.ctx.epoch });
+    m = reduce(m, { type: "RECONNECT_ATTACHED", epoch: m.ctx.epoch });
+    expect(m.ctx.liveFollow).toBe(true); // kept — so a second drop reconnects
+    // A clean finish clears it.
+    m = reduce(m, { type: "FINISH_CLEAN", epoch: m.ctx.epoch });
+    expect(m.ctx.liveFollow).toBe(false);
+  });
+
+  it("RECONNECT_NONE backs off through the ladder, then fails at the cap", () => {
+    let m = withRunFact("run-3");
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch });
+    for (let n = 1; n < RECONNECT_MAX_ATTEMPTS; n++) {
+      m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: n, epoch: m.ctx.epoch });
+      m = reduce(m, { type: "RECONNECT_NONE", epoch: m.ctx.epoch });
+      expect(m.phase.name).toBe("reconnecting");
+      if (m.phase.name === "reconnecting") {
+        expect(m.phase.attempt).toBe(n + 1);
+        expect(m.phase.failed).toBe(false);
+      }
+      // The belt-and-suspenders poll is armed each failed attempt.
+      expect(hasEffect(m, "armPoll")).toBe(true);
+    }
+    // Final attempt fails -> failed banner (Retry), poll armed.
+    m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: RECONNECT_MAX_ATTEMPTS, epoch: m.ctx.epoch });
+    m = reduce(m, { type: "RECONNECT_NONE", epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("reconnecting");
+    if (m.phase.name === "reconnecting") expect(m.phase.failed).toBe(true);
+    // RETRY restarts at attempt 1.
+    m = reduce(m, { type: "RETRY" });
+    expect(m.phase.name).toBe("reconnecting");
+    if (m.phase.name === "reconnecting") {
+      expect(m.phase.attempt).toBe(1);
+      expect(m.phase.failed).toBe(false);
+    }
+    expect(hasEffect(m, "resumeStream")).toBe(true);
+  });
+
+  it("reconnectDelayMs is the exponential backoff 1s,2s,4s,8s,16s", () => {
+    expect([1, 2, 3, 4, 5].map(reconnectDelayMs)).toEqual([1000, 2000, 4000, 8000, 16000]);
+  });
+});
+
+// #488 commit 4 — polling stalled-state + user-tail gating.
+describe("run-fsm — commit 4: stalled + run-fact gating", () => {
+  it("POLL_IDLE_CAP: polling → stalled with a banner (poll disarmed), not silent", () => {
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("polling");
+    m = reduce(m, { type: "POLL_IDLE_CAP" });
+    expect(m.phase.name).toBe("stalled");
+    expect(hasEffect(m, "disarmPoll")).toBe(true);
+  });
+
+  it("RETRY from stalled re-arms the poll", () => {
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch });
+    m = reduce(m, { type: "POLL_IDLE_CAP" });
+    m = reduce(m, { type: "RETRY" });
+    expect(m.phase.name).toBe("polling");
+    expect(hasEffect(m, "armPoll")).toBe(true);
+  });
+
+  it("a fresh NEGATIVE run-fact while attaching cancels recovery (user-tail, no active run)", () => {
+    // The mount POST /run returns no active run: attaching → idle, no poll armed.
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("idle");
+    expect(m.ctx.runFact).toBeNull();
+    expect(hasEffect(m, "disarmPoll")).toBe(true);
+  });
+
+  it("a negative run-fact while polling stops the poll", () => {
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch });
+    m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("idle");
+  });
+
+  it("POLL_TERMINAL settles polling → idle (I4 data-driven exit)", () => {
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch });
+    m = reduce(m, { type: "POLL_TERMINAL" });
+    expect(m.phase.name).toBe("idle");
+    expect(m.ctx.runFact).toBeNull();
+  });
+});
+
+// #488 commit 5 — error classification + supersede CAS transitions.
+describe("run-fsm — commit 5: supersede CAS + error classification", () => {
+  it("SUPERSEDE_REQUESTED → superseding, fires the CAS effect, bumps epoch", () => {
+    const streaming = withRunFact("run-old");
+    const m = reduce(streaming, { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    expect(m.phase.name).toBe("superseding");
+    expect(m.ctx.epoch).toBe(streaming.ctx.epoch + 1);
+    const sup = m.effects.find((e) => e.type === "supersede");
+    expect(sup).toEqual({ type: "supersede", targetRunId: "run-old" });
+  });
+
+  it("SUPERSEDE_READY → streaming as the new local owner", () => {
+    let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    m = reduce(m, { type: "SUPERSEDE_READY", runId: "run-new", epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("streaming");
+    expect(m.ctx.ownership).toBe("local");
+    expect(m.ctx.runFact).toEqual({ runId: "run-new" });
+  });
+
+  it("SUPERSEDE_MISMATCH → error(supersede-mismatch) + verify via /run (no blind banner)", () => {
+    let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    m = reduce(m, { type: "SUPERSEDE_MISMATCH", currentRunId: "run-x", epoch: m.ctx.epoch });
+    expect(m.phase).toEqual({ name: "error", kind: "supersede-mismatch" });
+    expect(hasEffect(m, "postRun")).toBe(true);
+    expect(m.ctx.runFact).toEqual({ runId: "run-x" });
+  });
+
+  it("SUPERSEDE_TIMEOUT → error(supersede-timeout), no auto-retry effect", () => {
+    let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    m = reduce(m, { type: "SUPERSEDE_TIMEOUT", epoch: m.ctx.epoch });
+    expect(m.phase).toEqual({ name: "error", kind: "supersede-timeout" });
+    expect(m.effects).toEqual([]);
+  });
+
+  it("SUPERSEDE_INVALID → error(supersede-invalid)", () => {
+    let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    m = reduce(m, { type: "SUPERSEDE_INVALID", epoch: m.ctx.epoch });
+    expect(m.phase).toEqual({ name: "error", kind: "supersede-invalid" });
+  });
+
+  it("a stale SUPERSEDE outcome from a superseded epoch is dropped", () => {
+    let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    const supersedingEpoch = m.ctx.epoch;
+    // The user retriggers, bumping the epoch again.
+    m = reduce(m, { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" });
+    // The first CAS's late TIMEOUT (old epoch) must NOT knock us out of superseding.
+    const late = reduce(m, { type: "SUPERSEDE_TIMEOUT", epoch: supersedingEpoch });
+    expect(late.phase.name).toBe("superseding");
+  });
+
+  it("RUN_ALREADY_ACTIVE (plain POST gate) → error(run-already-active), no retry effect", () => {
+    const m = reduce(run(initialMachine(), { type: "SEND_LOCAL" }), { type: "RUN_ALREADY_ACTIVE" });
+    expect(m.phase).toEqual({ name: "error", kind: "run-already-active" });
+    expect(m.effects).toEqual([]);
+  });
+
+  it("#497/S4: RUN_ALREADY_ACTIVE{activeRunId} ADOPTS the server's active run as the run-fact", () => {
+    // The server sends `activeRunId` so a later supersede can TARGET that run
+    // instead of a blind promote+abort. Absorb it into runFact.
+    const m = reduce(run(initialMachine(), { type: "SEND_LOCAL" }), {
+      type: "RUN_ALREADY_ACTIVE",
+      activeRunId: "run-foreign",
+    });
+    expect(m.phase).toEqual({ name: "error", kind: "run-already-active" });
+    expect(m.ctx.runFact).toEqual({ runId: "run-foreign" });
+    expect(m.effects).toEqual([]);
+  });
+
+  it("#497/S4: RUN_ALREADY_ACTIVE without an activeRunId keeps the prior run-fact", () => {
+    const seeded = reduce(run(initialMachine(), { type: "SEND_LOCAL" }), {
+      type: "RUN_FACT",
+      runFact: { runId: "run-prior" },
+    });
+    const m = reduce(seeded, { type: "RUN_ALREADY_ACTIVE" });
+    expect(m.ctx.runFact).toEqual({ runId: "run-prior" });
+  });
+});
+
+// #488 F2 — a late mount `getRun → ATTACH_START` must not hijack a local turn.
+describe("run-fsm — F2: ATTACH_START only from idle", () => {
+  it("ATTACH_START from a local `sending` turn is ignored (no observer hijack)", () => {
+    const sending = reduce(initialMachine(), { type: "SEND_LOCAL" }); // idle -> sending, local
+    const m = reduce(sending, { type: "ATTACH_START", runId: "r" });
+    expect(m.phase.name).toBe("sending");
+    expect(m.ctx.ownership).toBe("local"); // NOT flipped to observer
+    expect(m.effects).toEqual([]); // no resumeStream
+  });
+
+  it("ATTACH_START from idle attaches as normal", () => {
+    const m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    expect(m.phase.name).toBe("attaching");
+    expect(m.ctx.ownership).toBe("observer");
+    expect(hasEffect(m, "resumeStream")).toBe(true);
+  });
+});
+
+describe("run-fsm — stop (I4: exit by data)", () => {
+  it("STOP_REQUESTED → stopping, fires stopRun + abortAttach, no data-independent exit", () => {
+    const m = reduce(withRunFact(), { type: "STOP_REQUESTED" });
+    expect(m.phase.name).toBe("stopping");
+    expect(effectTypes(m)).toEqual(expect.arrayContaining(["stopRun", "abortAttach"]));
+  });
+
+  it("stopping exits on the aborted stream's finish carrying the PRE-STOP epoch", () => {
+    // MEDIUM (#488 re-review): STOP_REQUESTED is a command that BUMPS the epoch, but
+    // the runtime stamps the aborted stream's onFinish with the stream's START (pre-
+    // stop) generation — exactly what the component sends. `stopping` must HONOR
+    // that finish regardless of generation (no idle-cap covers `stopping`).
+    // MUTATION-VERIFY: remove the honor-in-`stopping` branch and this hangs in
+    // `stopping` (the epoch filter drops the pre-stop finish) -> red.
+    const preStopEpoch = withRunFact().ctx.epoch; // E1 (the stream's start epoch)
+    let m = reduce(withRunFact(), { type: "STOP_REQUESTED" }); // E1 -> E2, stopping
+    expect(m.ctx.epoch).toBe(preStopEpoch + 1);
+    m = reduce(m, { type: "FINISH_ABORT", epoch: preStopEpoch }); // NOT the current epoch
+    expect(m.phase.name).toBe("idle");
+    expect(m.ctx.runFact).toBeNull();
+  });
+
+  it("stopping exits on a clean finish carrying the pre-stop epoch too", () => {
+    const preStopEpoch = withRunFact().ctx.epoch;
+    let m = reduce(withRunFact(), { type: "STOP_REQUESTED" });
+    m = reduce(m, { type: "FINISH_CLEAN", epoch: preStopEpoch });
+    expect(m.phase.name).toBe("idle");
+  });
+
+  it("stopping exits on a negative run-fact (data)", () => {
+    let m = reduce(withRunFact(), { type: "STOP_REQUESTED" });
+    m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("idle");
+  });
+
+  // Review #4: `stopping` arms the poll but had no inactivity backstop.
+  it("review-4: POLL_IDLE_CAP in `stopping` exits to idle (bounded), NOT stalled", () => {
+    let m = reduce(withRunFact(), { type: "STOP_REQUESTED" });
+    expect(m.phase.name).toBe("stopping");
+    expect(hasEffect(m, "armPoll")).toBe(true);
+    // MUTATION-VERIFY: drop the `stopping` branch in POLL_IDLE_CAP and this hangs
+    // in `stopping` (poll forever) -> red.
+    m = reduce(m, { type: "POLL_IDLE_CAP" });
+    expect(m.phase.name).toBe("idle");
+    expect(hasEffect(m, "disarmPoll")).toBe(true);
+    expect(m.ctx.ownership).toBe("local");
+  });
+});
+
+// Review #1: positive attach outcomes must be guarded by the SOURCE phase — the
+// epoch filter alone is insufficient because POLL_TERMINAL uses to() (no epoch
+// bump) and does not abort the in-flight GET.
+describe("run-fsm — review-1: attach outcomes guarded by source phase", () => {
+  it("a late RECONNECT_ATTACHED after POLL_TERMINAL stays idle (no phantom streaming)", () => {
+    let m = withRunFact("run-1");
+    m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: true, epoch: m.ctx.epoch });
+    m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: 1, epoch: m.ctx.epoch }); // attach GET
+    const epoch = m.ctx.epoch;
+    // The armed degraded poll reaches the terminal row FIRST (epoch unchanged).
+    m = reduce(m, { type: "POLL_TERMINAL" });
+    expect(m.phase.name).toBe("idle");
+    expect(m.ctx.epoch).toBe(epoch); // POLL_TERMINAL did NOT bump the epoch
+    // The slow GET returns live 2xx under the SAME epoch — must NOT resurrect.
+    m = reduce(m, { type: "RECONNECT_ATTACHED", epoch });
+    expect(m.phase.name).toBe("idle");
+  });
+
+  it("a late ATTACH_LIVE / ATTACH_NONE after leaving `attaching` is ignored", () => {
+    let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    const epoch = m.ctx.epoch;
+    m = reduce(m, { type: "ATTACH_NONE", epoch }); // attaching -> polling
+    m = reduce(m, { type: "POLL_TERMINAL" }); // -> idle (epoch unchanged)
+    expect(m.phase.name).toBe("idle");
+    m = reduce(m, { type: "ATTACH_LIVE", epoch }); // late 2xx, same epoch
+    expect(m.phase.name).toBe("idle");
+    // And a late ATTACH_NONE (not `attaching`) is a no-op too.
+    m = reduce(m, { type: "ATTACH_NONE", epoch });
+    expect(m.phase.name).toBe("idle");
+  });
+});
+
+// Review #2: every terminal transition resets ownership to local.
+describe("run-fsm — review-2: terminal transitions reset ownership to local", () => {
+  const observer = (): Machine => {
+    let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch });
+    expect(m.ctx.ownership).toBe("observer");
+    return m;
+  };
+  it("FINISH_CLEAN resets ownership", () => {
+    const m = reduce(observer(), { type: "FINISH_CLEAN", epoch: observer().ctx.epoch });
+    expect(m.ctx.ownership).toBe("local");
+  });
+  it("FINISH_ERROR / POLL_TERMINAL / RUN_FACT(null) reset ownership", () => {
+    let o = observer();
+    expect(reduce(o, { type: "FINISH_ERROR", kind: "stream", epoch: o.ctx.epoch }).ctx.ownership).toBe("local");
+    // POLL_TERMINAL from an observer polling phase
+    let p = reduce(observer(), { type: "STREAM_INCOMPLETE", reason: "starved", epoch: observer().ctx.epoch });
+    expect(reduce(p, { type: "POLL_TERMINAL" }).ctx.ownership).toBe("local");
+    // RUN_FACT(null) from an observer attaching phase
+    let a = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    expect(reduce(a, { type: "RUN_FACT", runFact: null, epoch: a.ctx.epoch }).ctx.ownership).toBe("local");
+  });
+});
+
+describe("run-fsm — ownership (I2) is context, orthogonal to phase", () => {
+  it("attach/reconnect set observer; send/supersede-ready set local", () => {
+    let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
+    expect(m.ctx.ownership).toBe("observer");
+    m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch });
+    expect(m.phase.name).toBe("streaming");
+    expect(m.ctx.ownership).toBe("observer"); // still observing a detached run
+    // A local send flips ownership back to local.
+    m = reduce(m, { type: "SEND_LOCAL" });
+    expect(m.ctx.ownership).toBe("local");
+  });
+});
+
+describe("run-fsm — dispose (I5)", () => {
+  it("DISPOSE from any phase aborts controllers and bumps epoch", () => {
+    let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" });
+    const before = m.ctx.epoch;
+    m = reduce(m, { type: "DISPOSE" });
+    expect(m.phase.name).toBe("idle");
+    expect(m.ctx.epoch).toBe(before + 1);
+    expect(effectTypes(m)).toEqual(
+      expect.arrayContaining(["abortAttach", "cancelReconnect", "disarmPoll"]),
+    );
+  });
+});

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

@@ -0,0 +1,600 @@
+/**
+ * Run-lifecycle finite state machine for a single AI-chat thread (#488).
+ *
+ * ============================================================================
+ * WHY THIS EXISTS
+ * ----------------------------------------------------------------------------
+ * The resume/reconnect/poll/stop/supersede lifecycle used to be spread across
+ * ~26 `useRef` one-shot flags in `chat-thread.tsx`, each disarmed "on every
+ * path". Ownerless flag combinations produced silent UI freezes, and every fix
+ * added another ref (the #381 -> #432 -> #456 spiral). This module replaces that
+ * ref-zoo with ONE pure reducer whose transitions are enumerable and unit-
+ * testable in isolation (event x state -> next state is the observable property).
+ *
+ * The reducer is PURE: it owns no timers, no fetches, no React state. It maps
+ * `(machine, event) -> machine`, where the returned machine carries the list of
+ * COMMAND EFFECTS to run for that transition. A thin runtime in `chat-thread.tsx`
+ * dispatches events (from SDK callbacks / HTTP outcomes) and executes the
+ * effects (attach GET, POST /stream, POST /run, POST /stop, backoff timers,
+ * poll arm/disarm). The runtime lives in a THREAD, not the window, so a late SDK
+ * callback dies with the owner (kills the "event from a dead view" class, #161).
+ *
+ * ============================================================================
+ * INVARIANTS (see run-fsm.spec.md for the full spec + tables)
+ * ----------------------------------------------------------------------------
+ *  I1  EPOCH (generation counter). Commands (`resumeStream`, `postRun`, `stop`,
+ *      `supersede`, `scheduleReconnect`) are async; their outcomes arrive on the
+ *      SAME SDK/HTTP callbacks. Every command-emitting transition increments
+ *      `ctx.epoch`; every OUTCOME event carries the epoch it was issued under;
+ *      the reducer DROPS an outcome whose epoch != the current epoch. This is
+ *      what the one-shot-ref zoo used to approximate by hand.
+ *  I2  OWNERSHIP is a CONTEXT FIELD (`'local' | 'observer'`), not a state —
+ *      orthogonal to the transport phase. The queue is flushed ONLY by a local
+ *      owner (an observer following a detached run never flushes).
+ *  I3  RUN-FACT ("a run is active") is first-class from the server: `runFact`
+ *      holds the server-confirmed active run id (POST /run on mount, the `start`
+ *      metadata runId, attach outcomes). Reconnect is entered by the RUN-FACT,
+ *      not by the presence of an assistant message (#488 commit 2). A fresh
+ *      negative fact (null) cancels reconnect immediately.
+ *  I4  Exit `stopping` by DATA (a terminal row / negative run-fact), NEVER by the
+ *      stopRun HTTP response (which returns after abort, before finalization).
+ *  I5  Command controllers are effect-owned (abort in cleanup), NOT render-phase
+ *      refs — expressed here as the `abortAttach` effect on disposing transitions.
+ * ============================================================================
+ */
+
+// ---------------------------------------------------------------------------
+// Phases (the transport lifecycle). Ownership / runFact are CONTEXT, not here.
+// ---------------------------------------------------------------------------
+
+/** Why the degraded poll is the active recovery. */
+export type PollReason =
+  | "attach-none" // mount attach returned 204 / error — nothing live to attach
+  | "starved" // a resumed finish carried no visible content
+  | "disconnect-visible" // a live disconnect WITH on-screen content — poll to terminal
+  | "reconnect-exhausted"; // the live re-attach ladder gave up
+
+/** The classified error kind (drives the banner text + composer behavior). */
+export type ErrorKind =
+  | "stream" // a generic provider/network stream error (useChat error)
+  | "run-already-active" // 409 A_RUN_ALREADY_ACTIVE (a plain POST hit the gate)
+  | "supersede-mismatch" // 409 SUPERSEDE_TARGET_MISMATCH (CAS target moved)
+  | "supersede-timeout" // 409 SUPERSEDE_TIMEOUT (old run did not settle in W)
+  | "supersede-invalid" // 409 SUPERSEDE_INVALID (bad supersede target)
+  | "begin-failed"; // 503 A_RUN_BEGIN_FAILED (could not start the run)
+
+export type Phase =
+  | { name: "idle" }
+  | { name: "sending" } // local POST in flight, before the first frame
+  | { name: "streaming" } // receiving frames
+  | { name: "attaching" } // mount-time attach GET in flight
+  | { name: "reconnecting"; attempt: number; failed: boolean }
+  | { name: "polling"; reason: PollReason }
+  | { name: "stalled" } // poll hit the inactivity cap — banner + Retry
+  | { name: "stopping" }
+  | { name: "superseding" }
+  | { name: "error"; kind: ErrorKind };
+
+export type Ownership = "local" | "observer";
+
+/** The server-confirmed active run, or null when no run is active. */
+export type RunFact = { runId: string } | null;
+
+export interface Ctx {
+  /** I1: generation counter — every command-transition increments it. */
+  epoch: number;
+  /** I2: does THIS client own the turn's writes (local streamer) or observe? */
+  ownership: Ownership;
+  /** I3: the server-confirmed active run. */
+  runFact: RunFact;
+  /**
+   * Are we FOLLOWING a live run we were locally streaming (the reconnect ladder),
+   * as opposed to a one-shot mount-attach resume? Both are `ownership: 'observer'`,
+   * but they recover DIFFERENTLY on a drop: a live-follow drop RE-ENTERS the
+   * reconnect ladder (#488 commit 3 — the second break after a successful re-attach
+   * must reconnect again, not fall to silent poll), while a mount-resume drop falls
+   * to the degraded poll. This is the ctx bit that separates the two WITHOUT a new
+   * component ref (it is why commit 3 needs the FSM, not a surgical patch).
+   */
+  liveFollow: boolean;
+}
+
+export interface Machine {
+  phase: Phase;
+  ctx: Ctx;
+  /** Command effects to run for the transition that produced THIS machine.
+   *  The runtime executes them and does not read them again. */
+  effects: Effect[];
+}
+
+// ---------------------------------------------------------------------------
+// Command effects (the reducer's only side-channel — executed by the runtime).
+// ---------------------------------------------------------------------------
+
+export type Effect =
+  /** POST /run to (re)establish or verify the run-fact. `reason` is diagnostic. */
+  | { type: "postRun"; reason: "mount" | "verify" }
+  /** Trigger the SDK `resumeStream()` (attach GET via prepareReconnectToStream). */
+  | { type: "resumeStream" }
+  /** Schedule a reconnect attempt after a backoff, then dispatch RECONNECT_ATTEMPT. */
+  | { type: "scheduleReconnect"; attempt: number; delayMs: number }
+  /** Cancel any pending reconnect backoff timer. */
+  | { type: "cancelReconnect" }
+  /** Arm the degraded poll (the window's dumb timer follows the run in the DB). */
+  | { type: "armPoll"; reason: PollReason }
+  /** Disarm the degraded poll. */
+  | { type: "disarmPoll" }
+  /** POST /stop the chat's active run (authoritative detached-run stop). */
+  | { type: "stopRun" }
+  /** POST /stream { supersede: { runId } } — the CAS "interrupt and send now". */
+  | { type: "supersede"; targetRunId: string }
+  /** Abort the in-flight attach/reconnect GET controller (dispose / observer stop). */
+  | { type: "abortAttach" };
+
+// ---------------------------------------------------------------------------
+// Events. An OUTCOME event MAY carry `epoch`; if it does and it does not equal
+// the current epoch, the reducer drops it (I1). Trigger events (user actions,
+// fresh disconnects) carry no epoch and are never dropped.
+// ---------------------------------------------------------------------------
+
+export type Event =
+  // -- local turn --
+  | { type: "SEND_LOCAL" }
+  | { type: "STREAM_START"; runId?: string; epoch?: number }
+  /** An OBSERVER's attached stream ended WITHOUT reaching terminal (a starved
+   *  clean replay, or a torn resume) — fall to the degraded poll to drive the row
+   *  to its real terminal state. (A live-follow drop uses FINISH_DISCONNECT.) */
+  | { type: "STREAM_INCOMPLETE"; reason: PollReason; epoch?: number }
+  | { type: "FINISH_CLEAN"; epoch?: number }
+  | { type: "FINISH_ABORT"; epoch?: number }
+  | { type: "FINISH_DISCONNECT"; hasVisibleContent: boolean; epoch?: number }
+  | { type: "FINISH_ERROR"; kind: ErrorKind; epoch?: number }
+  // -- mount attach (resume) --
+  | { type: "ATTACH_START"; runId?: string }
+  | { type: "ATTACH_LIVE"; epoch?: number }
+  | { type: "ATTACH_NONE"; epoch?: number }
+  // -- reconnect after a live disconnect (entered by FINISH_DISCONNECT, #488 c2) --
+  | { type: "RECONNECT_ATTEMPT"; attempt: number; epoch?: number }
+  | { type: "RECONNECT_ATTACHED"; epoch?: number }
+  | { type: "RECONNECT_NONE"; epoch?: number }
+  | { type: "RETRY" }
+  // -- degraded poll --
+  | { type: "POLL_TERMINAL" }
+  | { type: "POLL_IDLE_CAP" }
+  // -- run-fact (server-confirmed active run) --
+  | { type: "RUN_FACT"; runFact: RunFact; epoch?: number }
+  // -- stop --
+  | { type: "STOP_REQUESTED" }
+  // -- supersede (CAS) --
+  | { type: "SUPERSEDE_REQUESTED"; targetRunId: string }
+  | { type: "SUPERSEDE_READY"; runId?: string; epoch?: number }
+  | { type: "SUPERSEDE_MISMATCH"; currentRunId?: string; epoch?: number }
+  | { type: "SUPERSEDE_TIMEOUT"; epoch?: number }
+  | { type: "SUPERSEDE_INVALID"; epoch?: number }
+  | { type: "RUN_ALREADY_ACTIVE"; activeRunId?: string }
+  // -- lifecycle --
+  | { type: "DISPOSE" };
+
+export const RECONNECT_MAX_ATTEMPTS = 5;
+export const RECONNECT_BASE_DELAY_MS = 1000;
+/** Backoff before attempt N (1-based): 1s, 2s, 4s, 8s, 16s. */
+export function reconnectDelayMs(attempt: number): number {
+  return RECONNECT_BASE_DELAY_MS * 2 ** (attempt - 1);
+}
+
+// ---------------------------------------------------------------------------
+// Constructors / helpers.
+// ---------------------------------------------------------------------------
+
+export function initialMachine(overrides?: Partial<Ctx>): Machine {
+  return {
+    phase: { name: "idle" },
+    ctx: { epoch: 0, ownership: "local", runFact: null, liveFollow: false, ...overrides },
+    effects: [],
+  };
+}
+
+/** Build a machine result: a phase, optional ctx patch, and effects. Empty
+ *  effects by default. Never mutates the input. */
+function to(
+  m: Machine,
+  phase: Phase,
+  opts?: { ctx?: Partial<Ctx>; effects?: Effect[] },
+): Machine {
+  return {
+    phase,
+    ctx: { ...m.ctx, ...(opts?.ctx ?? {}) },
+    effects: opts?.effects ?? [],
+  };
+}
+
+/** No transition: keep the phase, clear effects (so a re-run does not re-fire). */
+function stay(m: Machine): Machine {
+  return { phase: m.phase, ctx: m.ctx, effects: [] };
+}
+
+/** A command-transition: same as `to` but bumps the epoch (I1). Any outcome
+ *  event issued under the old epoch is dropped once this lands. */
+function command(
+  m: Machine,
+  phase: Phase,
+  effects: Effect[],
+  ctx?: Partial<Ctx>,
+): Machine {
+  return {
+    phase,
+    ctx: { ...m.ctx, ...(ctx ?? {}), epoch: m.ctx.epoch + 1 },
+    effects,
+  };
+}
+
+// ---------------------------------------------------------------------------
+// The pure reducer.
+// ---------------------------------------------------------------------------
+
+/** The terminal stream-finish events (one turn's stream ended). */
+function isFinishEvent(event: Event): boolean {
+  return (
+    event.type === "FINISH_ABORT" ||
+    event.type === "FINISH_CLEAN" ||
+    event.type === "FINISH_DISCONNECT" ||
+    event.type === "FINISH_ERROR" ||
+    event.type === "STREAM_INCOMPLETE"
+  );
+}
+
+export function reduce(m: Machine, event: Event): Machine {
+  // MEDIUM (#488 re-review): honor ANY stream finish in `stopping` regardless of
+  // generation. A plain user Stop has NO successor stream — the aborted stream's
+  // finish IS the expected end of the stop, so exit `stopping -> idle` by that DATA
+  // (I4). The epoch filter below must NOT drop it: STOP_REQUESTED bumped the epoch,
+  // but the finish carries the PRE-stop generation (the runtime stamps it with the
+  // stream's start epoch), so I1 would otherwise strand the machine in `stopping`
+  // forever (no idle-cap covers `stopping`). The epoch filter stays in force for
+  // `superseding` (a successor B owns) — that is the F1 supersede drop.
+  if (m.phase.name === "stopping" && isFinishEvent(event)) {
+    return to(m, { name: "idle" }, {
+      // Reset ownership to local on this terminal transition (review #2): otherwise
+      // an observer-stop leaves ownership 'observer' and hides "Send now" forever.
+      ctx: { runFact: null, liveFollow: false, ownership: "local" },
+      effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+    });
+  }
+
+  // I1: drop a stale outcome (an event issued under a superseded epoch).
+  if ("epoch" in event && event.epoch !== undefined && event.epoch !== m.ctx.epoch) {
+    return stay(m);
+  }
+
+  switch (event.type) {
+    // ---- local turn ----------------------------------------------------
+    case "SEND_LOCAL":
+      // A local send owns the view: leave any recovery, become the local
+      // streamer, disarm poll/reconnect. epoch++ so a late recovery outcome
+      // from the previous phase is dropped.
+      return command(
+        m,
+        { name: "sending" },
+        [{ type: "cancelReconnect" }, { type: "disarmPoll" }],
+        { ownership: "local", liveFollow: false },
+      );
+
+    case "STREAM_INCOMPLETE":
+      // An OBSERVER's attached stream ended incomplete (starved / torn) — follow
+      // the run to terminal via the degraded poll.
+      return to(m, { name: "polling", reason: event.reason }, {
+        effects: [{ type: "armPoll", reason: event.reason }],
+      });
+
+    case "STREAM_START": {
+      // First frame arrived. Adopt the run-fact runId if present. sending ->
+      // streaming; a reconnect/attach that just went live also lands here.
+      const runFact = event.runId ? { runId: event.runId } : m.ctx.runFact;
+      return to(m, { name: "streaming" }, {
+        ctx: { runFact },
+        effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }],
+      });
+    }
+
+    case "FINISH_CLEAN":
+      // A clean terminal outcome. The run is done — clear the run-fact and go
+      // idle. (The queue flush is a component concern gated by ownership; the
+      // FSM only models the phase.) Review #2: reset ownership to local so a
+      // just-finished observer-attach turn re-exposes "Send now" for the queue.
+      return to(m, { name: "idle" }, {
+        ctx: { runFact: null, liveFollow: false, ownership: "local" },
+        effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+      });
+
+    case "FINISH_ABORT":
+      // A user Stop / intentional abort finished. If we were stopping, the
+      // terminal data has now arrived (I4) — go idle. The run-fact is cleared.
+      return to(m, { name: "idle" }, {
+        ctx: { runFact: null, liveFollow: false, ownership: "local" },
+        effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+      });
+
+    case "FINISH_DISCONNECT":
+      // A LIVE SSE drop. Recovery depends on WHO we are (I2 + liveFollow):
+      //  - a mount-attach OBSERVER (a one-shot resume, NOT live-follow) that drops
+      //    -> the degraded poll drives the row to terminal from the DB.
+      if (m.ctx.ownership === "observer" && !m.ctx.liveFollow) {
+        return to(m, { name: "polling", reason: "disconnect-visible" }, {
+          effects: [{ type: "armPoll", reason: "disconnect-visible" }],
+        });
+      }
+      //  - a LOCAL live turn (first drop) OR a live-follow re-attach (a SUBSEQUENT
+      //    drop) -> (re-)enter the reconnect ladder. #488 commit 3: allowed
+      //    REPEATEDLY — `liveFollow` is kept across a successful re-attach, so the
+      //    second break reconnects again instead of falling to silent poll.
+      // #488 commit 2: gated on the RUN-FACT (or an existing live-follow), NOT on
+      // the presence of an assistant message — a setup-phase break still recovers.
+      //  - visible content already on screen -> keep it, ALSO poll to terminal
+      //    (a full replay could clobber the fuller live tail);
+      //  - no visible content -> the reconnect ladder rebuilds it.
+      if (m.ctx.runFact || m.ctx.liveFollow) {
+        const effects: Effect[] = [
+          { type: "scheduleReconnect", attempt: 1, delayMs: reconnectDelayMs(1) },
+        ];
+        if (event.hasVisibleContent) effects.push({ type: "armPoll", reason: "disconnect-visible" });
+        return command(m, { name: "reconnecting", attempt: 1, failed: false }, effects, {
+          ownership: "observer",
+          liveFollow: true,
+        });
+      }
+      // No run to recover: a plain disconnect. Surface the terminal notice.
+      return to(m, { name: "idle" }, {
+        ctx: { runFact: null, liveFollow: false, ownership: "local" },
+      });
+
+    case "FINISH_ERROR":
+      return to(m, { name: "error", kind: event.kind }, {
+        ctx: { runFact: null, liveFollow: false, ownership: "local" },
+        effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+      });
+
+    // ---- mount attach (resume) ----------------------------------------
+    case "ATTACH_START":
+      // A reopened tab attaches to a still-running run: observer ownership.
+      // #488 F2: ONLY from idle. The mount `getRun` round-trip resolves async, and
+      // a local send may have started meanwhile (phase `sending`, ownership local);
+      // a late ATTACH_START must NOT hijack that local turn into an observer-attach
+      // (queue would stop flushing, "Send now" would hide). Guarding in the reducer
+      // covers every dispatch source.
+      if (m.phase.name !== "idle") return stay(m);
+      return command(m, { name: "attaching" }, [{ type: "resumeStream" }], {
+        ownership: "observer",
+        runFact: event.runId ? { runId: event.runId } : m.ctx.runFact,
+      });
+
+    case "ATTACH_LIVE":
+      // The attach GET returned a live 2xx stream — follow it as an observer.
+      // Review #1: guard by SOURCE phase. The epoch filter alone is not enough — a
+      // POLL_TERMINAL uses to() (no epoch bump) and does not abort the in-flight
+      // GET, so a slow 2xx landing after the machine already left `attaching` (e.g.
+      // the armed poll saw the terminal row -> idle) would resurrect a settled run
+      // into a phantom `streaming`. Only enter streaming FROM `attaching`.
+      if (m.phase.name !== "attaching") return stay(m);
+      return to(m, { name: "streaming" });
+
+    case "ATTACH_NONE":
+      // 204 / non-2xx / throw: nothing live to attach. Arm the degraded poll to
+      // follow the run to terminal from the DB. This is a soft-negative run-fact
+      // (204 on a non-stripped path is authoritative-negative; the runtime may
+      // pass a RUN_FACT null separately). Keep the run-fact as-is here.
+      // Review #1: guard by source phase for consistency (a late outcome after the
+      // machine already left `attaching` must not re-arm a poll).
+      if (m.phase.name !== "attaching") return stay(m);
+      return to(m, { name: "polling", reason: "attach-none" }, {
+        effects: [{ type: "armPoll", reason: "attach-none" }],
+      });
+
+    // ---- reconnect after a live disconnect ----------------------------
+    case "RECONNECT_ATTEMPT":
+      // A scheduled backoff fired — fire the attach GET. epoch++ so the previous
+      // attempt's late outcome cannot drive this one.
+      if (m.phase.name !== "reconnecting") return stay(m);
+      return command(
+        m,
+        { name: "reconnecting", attempt: event.attempt, failed: false },
+        [{ type: "resumeStream" }],
+      );
+
+    case "RECONNECT_ATTACHED":
+      // #488 commit 3: a live re-attach succeeded. Reset to streaming — the
+      // attempt counter is dropped, so a LATER disconnect can start a fresh
+      // ladder from attempt 1 (the old one-shot `!wasResumed` gate forbade a
+      // second cycle, sending the second break to silent poll).
+      // Review #1: guard by SOURCE phase. The armed degraded poll can reach the
+      // terminal row (POLL_TERMINAL -> idle, via to(), NO epoch bump, GET not
+      // aborted) BEFORE a slow reconnect GET returns 2xx; without this guard that
+      // late RECONNECT_ATTACHED (same epoch) would resurrect a settled run into a
+      // phantom `streaming`. Only re-enter streaming FROM `reconnecting`.
+      if (m.phase.name !== "reconnecting") return stay(m);
+      return to(m, { name: "streaming" }, {
+        effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }],
+      });
+
+    case "RECONNECT_NONE": {
+      // 204 / error during a reconnect attempt. Arm the degraded poll as the
+      // belt-and-suspenders fallback, then either back off to the next attempt
+      // or, at the cap, surface the manual Retry ("failed").
+      if (m.phase.name !== "reconnecting") return stay(m);
+      const attempt = m.phase.attempt;
+      if (attempt < RECONNECT_MAX_ATTEMPTS) {
+        return command(
+          m,
+          { name: "reconnecting", attempt: attempt + 1, failed: false },
+          [
+            { type: "armPoll", reason: "attach-none" },
+            { type: "scheduleReconnect", attempt: attempt + 1, delayMs: reconnectDelayMs(attempt + 1) },
+          ],
+        );
+      }
+      return to(m, { name: "reconnecting", attempt, failed: true }, {
+        effects: [{ type: "armPoll", reason: "reconnect-exhausted" }],
+      });
+    }
+
+    case "RETRY":
+      // Manual Retry from the "failed" reconnect banner OR the stalled banner.
+      if (m.phase.name === "reconnecting" && m.phase.failed) {
+        return command(
+          m,
+          { name: "reconnecting", attempt: 1, failed: false },
+          [{ type: "resumeStream" }],
+        );
+      }
+      if (m.phase.name === "stalled") {
+        // Re-arm the poll to try to catch the run up again.
+        return command(m, { name: "polling", reason: "attach-none" }, [
+          { type: "armPoll", reason: "attach-none" },
+        ]);
+      }
+      return stay(m);
+
+    // ---- degraded poll -------------------------------------------------
+    case "POLL_TERMINAL":
+      // The run reached a terminal row via the poll (or the reconcile merge). Go
+      // idle and disarm everything (I4: this is a DATA-driven exit, incl. exit
+      // from `stopping`). Review #2: reset ownership to local.
+      return to(m, { name: "idle" }, {
+        ctx: { runFact: null, liveFollow: false, ownership: "local" },
+        effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+      });
+
+    case "POLL_IDLE_CAP":
+      // Review #4: `stopping` also arms the poll (STOP_REQUESTED) but has NO other
+      // backstop — an observer-stop with no SDK stream to fire onFinish, whose
+      // server stop never drives the run terminal, would poll the DB forever. Give
+      // it a bounded exit: cap -> idle + disarm (NOT `stalled`; Stop was already
+      // pressed, so there is nothing for the user to retry).
+      if (m.phase.name === "stopping") {
+        return to(m, { name: "idle" }, {
+          ctx: { runFact: null, liveFollow: false, ownership: "local" },
+          effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+        });
+      }
+      // #488 commit 4a: the poll hit the inactivity cap. Instead of going SILENT
+      // (the old "forever half-done answer"), surface a stalled banner + Retry.
+      if (m.phase.name !== "polling" && m.phase.name !== "reconnecting") return stay(m);
+      return to(m, { name: "stalled" }, {
+        effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }],
+      });
+
+    // ---- run-fact ------------------------------------------------------
+    case "RUN_FACT": {
+      const runFact = event.runFact;
+      // A fresh NEGATIVE fact (no active run) cancels recovery immediately (I3):
+      // there is nothing to reconnect to / poll for.
+      if (!runFact) {
+        if (
+          m.phase.name === "reconnecting" ||
+          m.phase.name === "attaching" ||
+          m.phase.name === "polling" ||
+          m.phase.name === "stopping"
+        ) {
+          return to(m, { name: "idle" }, {
+            // Review #2: reset ownership to local on this terminal transition.
+            ctx: { runFact: null, liveFollow: false, ownership: "local" },
+            effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }],
+          });
+        }
+        return to(m, m.phase, { ctx: { runFact: null } });
+      }
+      // A positive fact just updates the context (pessimism toward an attempt: a
+      // stale-but-positive fact permits entering recovery; a 204 will cut it).
+      return to(m, m.phase, { ctx: { runFact } });
+    }
+
+    // ---- stop ----------------------------------------------------------
+    case "STOP_REQUESTED":
+      // Authoritative stop of a detached run. Enter `stopping` and fire stopRun +
+      // abort the local/attach reader. ALSO arm the poll so the terminal row is
+      // observed — the exit is by DATA (I4: a terminal row / negative run-fact),
+      // never by the stopRun HTTP response (which returns after abort, before
+      // finalization). For a local turn the aborted stream's onFinish (ANY finish)
+      // is HONORED in `stopping` at the top of reduce() — regardless of generation
+      // — and exits to idle; the armed poll is the fallback for an observer stop
+      // with no local onFinish.
+      return command(
+        m,
+        { name: "stopping" },
+        [
+          { type: "stopRun" },
+          { type: "abortAttach" },
+          { type: "cancelReconnect" },
+          { type: "armPoll", reason: "attach-none" },
+        ],
+      );
+
+    // ---- supersede (CAS) ----------------------------------------------
+    case "SUPERSEDE_REQUESTED":
+      // "Interrupt and send now": CAS POST /stream { supersede }. epoch++ so a
+      // late outcome of the interrupted run is dropped.
+      return command(
+        m,
+        { name: "superseding" },
+        [{ type: "supersede", targetRunId: event.targetRunId }, { type: "cancelReconnect" }, { type: "disarmPoll" }],
+      );
+
+    case "SUPERSEDE_READY": {
+      // CAS succeeded (old run stopped/settled, slot taken, new run begun). We
+      // are now the local streamer of the NEW run. Adopt its runId if provided.
+      const runFact = event.runId ? { runId: event.runId } : m.ctx.runFact;
+      return to(m, { name: "streaming" }, {
+        ctx: { ownership: "local", runFact, liveFollow: false },
+      });
+    }
+
+    case "SUPERSEDE_MISMATCH":
+      // The active run moved between the click and the CAS. Per the spec: verify
+      // via /run rather than blindly banner — the mismatch may be our own already-
+      // superseded run. Surface a classified error AND fire a run-fact verify.
+      return to(m, { name: "error", kind: "supersede-mismatch" }, {
+        ctx: { runFact: event.currentRunId ? { runId: event.currentRunId } : m.ctx.runFact },
+        effects: [{ type: "postRun", reason: "verify" }],
+      });
+
+    case "SUPERSEDE_TIMEOUT":
+      // The old run did not settle within W. Nothing persisted; the composer keeps
+      // its text. Classified error, NO auto-retry (the old client retry ladder is
+      // removed in #488 commit 5).
+      return to(m, { name: "error", kind: "supersede-timeout" });
+
+    case "SUPERSEDE_INVALID":
+      return to(m, { name: "error", kind: "supersede-invalid" });
+
+    case "RUN_ALREADY_ACTIVE":
+      // A plain POST hit the one-active-run gate. NO auto-retry — the composer
+      // offers "interrupt and send" (supersede) instead. #497/S4: adopt the
+      // server's activeRunId as the run-fact so that supersede can TARGET the
+      // (possibly foreign-tab) active run via the CAS, rather than a blind
+      // promote+abort that just 409s again. A stale/absent id keeps the prior fact.
+      return to(m, { name: "error", kind: "run-already-active" }, {
+        ctx: { runFact: event.activeRunId ? { runId: event.activeRunId } : m.ctx.runFact },
+      });
+
+    // ---- lifecycle -----------------------------------------------------
+    case "DISPOSE":
+      // Unmount: abort in-flight controllers, drop timers, and bump the epoch so
+      // NO late callback can drive this (now dead) machine (I5).
+      return command(
+        m,
+        { name: "idle" },
+        [
+          { type: "abortAttach" },
+          { type: "cancelReconnect" },
+          { type: "disarmPoll" },
+        ],
+        { liveFollow: false },
+      );
+
+    default: {
+      // Exhaustiveness guard.
+      const _never: never = event;
+      void _never;
+      return stay(m);
+    }
+  }
+}

apps/client/src/features/ai-chat/utils/adopt-chat-id.test.ts

@@ -3,6 +3,7 @@ import {
   resolveAdoptedChatId,
   newlyAddedChatIds,
   extractServerChatId,
+  extractRunId,
 } from "./adopt-chat-id";
 
 describe("resolveAdoptedChatId", () => {
@@ -70,3 +71,17 @@ describe("extractServerChatId", () => {
     expect(extractServerChatId(undefined)).toBeUndefined();
   });
 });
+
+describe("extractRunId", () => {
+  it("reads a string runId from the start metadata", () => {
+    expect(extractRunId({ metadata: { runId: "run-1" } })).toBe("run-1");
+  });
+  it("returns undefined when runId is absent", () => {
+    expect(extractRunId({ metadata: { chatId: "c" } })).toBeUndefined();
+    expect(extractRunId({})).toBeUndefined();
+    expect(extractRunId(undefined)).toBeUndefined();
+  });
+  it("returns undefined for a non-string runId", () => {
+    expect(extractRunId({ metadata: { runId: 7 } })).toBeUndefined();
+  });
+});

apps/client/src/features/ai-chat/utils/adopt-chat-id.ts

@@ -56,6 +56,20 @@ export function extractServerChatId(
   return typeof m?.chatId === "string" ? m.chatId : undefined;
 }
 
+/**
+ * #488: read the authoritative RUN id off a streaming assistant message. The
+ * server attaches it as `message.metadata.runId` on the `start` part when a run
+ * wraps the turn (see server `chatStreamMetadata`, #184/#487). This is the live
+ * run-fact update the client FSM adopts (mirrors `extractServerChatId`). Returns
+ * it only when it is a string; undefined otherwise.
+ */
+export function extractRunId(
+  message: { metadata?: unknown } | undefined,
+): string | undefined {
+  const m = message?.metadata as { runId?: string } | undefined;
+  return typeof m?.runId === "string" ? m.runId : undefined;
+}
+
 /**
  * The deduped set of ids present in `afterIds` but not in `beforeIds`. A
  * paginated/flatMapped list can repeat the same id, so dedupe: one genuinely-new

apps/client/src/features/ai-chat/utils/error-message.test.ts

@@ -42,6 +42,53 @@ describe("describeChatError", () => {
     );
   });
 
+  // #488 commit 5: the #487 concurrency-gate / supersede 409s. FULL real bodies:
+  // a ConflictException(object) whose response is serialized verbatim, carrying a
+  // `code` and statusCode 409. Each must classify to a human text, not raw JSON.
+  it("classifies A_RUN_ALREADY_ACTIVE (409) as already-answering, not raw JSON", () => {
+    const body =
+      '{"message":"A run is already active for this chat","code":"A_RUN_ALREADY_ACTIVE","statusCode":409}';
+    expect(describeChatError(body, t).title).toBe(
+      "The agent is already answering",
+    );
+    // Never leaks the raw code as the detail.
+    expect(describeChatError(body, t).detail).not.toContain("A_RUN_ALREADY_ACTIVE");
+  });
+
+  it("classifies SUPERSEDE_TARGET_MISMATCH (409) as run-changed", () => {
+    // Real server body shape: the current run id is `activeRunId` (NOT `runId`) —
+    // see ai-chat.controller.ts. describeChatError classifies off `code` only.
+    const body =
+      '{"message":"active run does not match the supersede target","code":"SUPERSEDE_TARGET_MISMATCH","activeRunId":"run-x","statusCode":409}';
+    expect(describeChatError(body, t).title).toBe(
+      "Couldn't interrupt — the run changed",
+    );
+  });
+
+  it("classifies SUPERSEDE_TIMEOUT (409) as couldn't-interrupt-in-time", () => {
+    const body =
+      '{"message":"the run did not settle within the supersede window","code":"SUPERSEDE_TIMEOUT","statusCode":409}';
+    expect(describeChatError(body, t).title).toBe("Couldn't interrupt in time");
+  });
+
+  it("classifies SUPERSEDE_INVALID (409) as couldn't-interrupt-that-run", () => {
+    const body =
+      '{"message":"supervise requires chatId","code":"SUPERSEDE_INVALID","statusCode":409}';
+    expect(describeChatError(body, t).title).toBe(
+      "Couldn't interrupt that run",
+    );
+  });
+
+  it("ORDER GUARD: A_RUN_ALREADY_ACTIVE wins over any generic status branch", () => {
+    // Even though the body could superficially look 4xx-ish, the code branch runs
+    // first, so it is never mislabeled by a generic status heading.
+    const body =
+      '{"message":"conflict","code":"A_RUN_ALREADY_ACTIVE","statusCode":409}';
+    const view = describeChatError(body, t);
+    expect(view.title).not.toBe("Something went wrong");
+    expect(view.title).not.toBe("AI provider not configured");
+  });
+
   it("classifies a dropped connection (ECONNRESET) as a lost-connection error", () => {
     expect(
       describeChatError("Cannot connect to API: read ECONNRESET", t).title,

apps/client/src/features/ai-chat/utils/error-message.ts

@@ -39,6 +39,44 @@ export function describeChatError(
     };
   }
 
+  // #488 commit 5: the #487 concurrency-gate / supersede 409s. These arrive as a
+  // ConflictException(object) body carrying a `code` (and statusCode 409). They
+  // MUST be classified by `code` STRICTLY BEFORE any generic status branch, or the
+  // user sees the raw JSON `{"code":"A_RUN_ALREADY_ACTIVE",…}`. The code strings
+  // are the real #487 server contract (ai-chat.controller.ts) — do not invent.
+  if (/"code"\s*:\s*"A_RUN_ALREADY_ACTIVE"/.test(msg)) {
+    return {
+      title: t("The agent is already answering"),
+      detail: t(
+        "This chat already has a run in progress. Wait for it to finish, or interrupt it and send now.",
+      ),
+    };
+  }
+  if (/"code"\s*:\s*"SUPERSEDE_TARGET_MISMATCH"/.test(msg)) {
+    return {
+      title: t("Couldn't interrupt — the run changed"),
+      detail: t(
+        "The run you tried to interrupt is no longer the active one. Check the latest answer and try again.",
+      ),
+    };
+  }
+  if (/"code"\s*:\s*"SUPERSEDE_TIMEOUT"/.test(msg)) {
+    return {
+      title: t("Couldn't interrupt in time"),
+      detail: t(
+        "The previous run didn't stop in time. Nothing was sent — try sending again.",
+      ),
+    };
+  }
+  if (/"code"\s*:\s*"SUPERSEDE_INVALID"/.test(msg)) {
+    return {
+      title: t("Couldn't interrupt that run"),
+      detail: t(
+        "The run to interrupt doesn't belong to this chat. Reload and try again.",
+      ),
+    };
+  }
+
   if (/"statusCode"\s*:\s*403\b/.test(msg)) {
     return {
       title: t("AI chat is disabled"),

apps/server/src/app.module.ts

@@ -25,7 +25,7 @@ import { CacheModule } from '@nestjs/cache-manager';
 import KeyvRedis from '@keyv/redis';
 import { LoggerModule } from './common/logger/logger.module';
 import { ClsModule } from 'nestjs-cls';
-import { NoopAuditModule } from './integrations/audit/audit.module';
+import { AuditModule } from './integrations/audit/audit.module';
 import { ThrottleModule } from './integrations/throttle/throttle.module';
 import { McpModule } from './integrations/mcp/mcp.module';
 import { SandboxModule } from './integrations/sandbox/sandbox.module';
@@ -55,7 +55,7 @@ try {
       middleware: { mount: true },
     }),
     LoggerModule,
-    NoopAuditModule,
+    AuditModule,
     CoreModule,
     DatabaseModule,
     EnvironmentModule,

apps/server/src/collaboration/yjs.util.spec.ts

@@ -529,4 +529,107 @@ describe('replaceYjsMarkedText', () => {
     expect(result).toEqual({ applied: false, currentText: 'abcdef' });
     expect(text.toDelta()).toEqual(before);
   });
+
+  // #496: apply must NOT silently strip the replaced run's inline formatting.
+  // Build a paragraph and format the marked range with extra marks, then assert
+  // the replacement carries them.
+  function buildFormatted(
+    runs: Array<{ text: string; attrs?: Record<string, any> }>,
+  ): { fragment: Y.XmlFragment; text: Y.XmlText } {
+    const ydoc = new Y.Doc();
+    const fragment = ydoc.getXmlFragment('default');
+    const para = new Y.XmlElement('paragraph');
+    fragment.insert(0, [para]);
+    const text = new Y.XmlText();
+    para.insert(0, [text]);
+    text.insert(0, runs.map((r) => r.text).join(''));
+    let offset = 0;
+    for (const run of runs) {
+      if (run.attrs) text.format(offset, run.text.length, run.attrs);
+      offset += run.text.length;
+    }
+    return { fragment, text };
+  }
+
+  it('preserves the original run formatting (bold + link) on the replacement', () => {
+    const { fragment, text } = buildFormatted([
+      { text: 'see ' },
+      {
+        text: 'old',
+        attrs: {
+          comment: { commentId: 'c1', resolved: false },
+          bold: true,
+          link: { href: 'https://x.test' },
+        },
+      },
+      { text: ' end' },
+    ]);
+
+    const result = replaceYjsMarkedText(fragment, 'c1', 'old', 'new');
+
+    expect(result).toEqual({ applied: true, currentText: 'new' });
+    // The comment anchor AND the bold/link marks survive the delete+insert.
+    expect(text.toDelta()).toEqual([
+      { insert: 'see ' },
+      {
+        insert: 'new',
+        attributes: {
+          comment: { commentId: 'c1', resolved: false },
+          bold: true,
+          link: { href: 'https://x.test' },
+        },
+      },
+      { insert: ' end' },
+    ]);
+  });
+
+  it('mixed formatting under the mark: replacement takes the DOMINANT (longest) run, NOT the leading one', () => {
+    // Leading run is SHORT + plain ("x", 1 char); the following run is LONGER +
+    // bold ("bolded", 6 chars), same commentId. The longest run is deliberately
+    // NOT first: a "first-wins" pick would carry plain (no bold), so asserting
+    // bold on the result only holds if the code genuinely selects the LONGEST run.
+    const { fragment, text } = buildFormatted([
+      { text: 'x', attrs: { comment: { commentId: 'c1', resolved: false } } },
+      {
+        text: 'bolded',
+        attrs: { comment: { commentId: 'c1', resolved: false }, bold: true },
+      },
+    ]);
+
+    const result = replaceYjsMarkedText(fragment, 'c1', 'xbolded', 'Z');
+
+    expect(result).toEqual({ applied: true, currentText: 'Z' });
+    expect(text.toDelta()).toEqual([
+      {
+        insert: 'Z',
+        attributes: { comment: { commentId: 'c1', resolved: false }, bold: true },
+      },
+    ]);
+  });
+
+  it('mixed formatting under the mark: on a length tie the FIRST run wins', () => {
+    // Two equal-length runs (2 chars each) with different formatting, same
+    // commentId. The reduce keeps the accumulator on a tie, so the FIRST run
+    // (italic) prevails over the later bold one.
+    const { fragment, text } = buildFormatted([
+      {
+        text: 'AA',
+        attrs: { comment: { commentId: 'c1', resolved: false }, italic: true },
+      },
+      {
+        text: 'BB',
+        attrs: { comment: { commentId: 'c1', resolved: false }, bold: true },
+      },
+    ]);
+
+    const result = replaceYjsMarkedText(fragment, 'c1', 'AABB', 'Z');
+
+    expect(result).toEqual({ applied: true, currentText: 'Z' });
+    expect(text.toDelta()).toEqual([
+      {
+        insert: 'Z',
+        attributes: { comment: { commentId: 'c1', resolved: false }, italic: true },
+      },
+    ]);
+  });
 });

apps/server/src/collaboration/yjs.util.ts

@@ -145,6 +145,10 @@ type MarkedSegment = {
   length: number;
   text: string;
   markAttrs: Record<string, any>;
+  // The FULL attribute set of this delta run — the `comment` mark plus any
+  // inline formatting (bold/italic/code/link/…). Captured so apply can carry the
+  // original run's formatting onto the replacement instead of dropping it.
+  attributes: Record<string, any>;
 };
 
 /**
@@ -202,6 +206,7 @@ export function replaceYjsMarkedText(
             length,
             text: insert,
             markAttrs: markAttr,
+            attributes,
           });
         }
         offset += length;
@@ -251,15 +256,25 @@ export function replaceYjsMarkedText(
     return { applied: false, currentText: joinedText };
   }
 
-  // 3. All guards passed: delete the marked run and re-insert newText with the
-  // same comment attributes at the same offset. Atomic within the caller's
-  // transaction.
+  // 3. All guards passed: delete the marked run and re-insert newText at the
+  // same offset. Atomic within the caller's transaction.
   const start = segments[0].offset;
   const len = segments.reduce((sum, s) => sum + s.length, 0);
-  const markAttrs = segments[0].markAttrs;
+
+  // Carry the ORIGINAL run's formatting onto the replacement (#496): inserting
+  // with only the `comment` mark silently dropped bold/italic/code/link of the
+  // replaced text. Yjs applies one flat attribute set to the whole insert, so
+  // when the marked run mixes formatting we pick the DOMINANT segment (the one
+  // covering the most characters) and apply its attributes — a v1 that preserves
+  // the common single-format case exactly and, for a mixed run, keeps the
+  // prevailing style rather than losing all of it. `attributes` already carries
+  // the `comment` mark (every collected segment is filtered on it above), so the
+  // anchor is preserved by copying the run's attribute set verbatim.
+  const dominant = segments.reduce((a, b) => (b.length > a.length ? b : a));
+  const insertAttrs = { ...dominant.attributes };
 
   node.delete(start, len);
-  node.insert(start, newText, { comment: markAttrs });
+  node.insert(start, newText, insertAttrs);
 
   return { applied: true, currentText: newText };
 }

apps/server/src/core/ai-chat/ai-chat.parts-resilience.spec.ts

@@ -0,0 +1,142 @@
+// #489 — client-parts validation + resilient history conversion.
+//
+// These unit tests exercise the two exported helpers against the REAL
+// `convertToModelMessages` from `ai` (NOT a mock): a genuinely malformed part
+// (a `null` element inside a parts array) makes the real converter throw
+// ("Cannot read properties of null"), which is the actual production
+// "bricked chat" mechanism this fix defends against. Asserting against the real
+// converter (rather than a mock-shaped error) is the whole point — a mock would
+// hide a version change in the converter's throw behaviour.
+import { convertToModelMessages, type UIMessage } from 'ai';
+import {
+  sanitizeUserParts,
+  convertHistoryResilient,
+  TOOL_CONTEXT_OMITTED_MARKER,
+} from './ai-chat.service';
+
+type Row = Omit<UIMessage, 'id'> & { id: string };
+
+describe('sanitizeUserParts (#489, branch: validation on receipt)', () => {
+  it('keeps whitelisted text parts unchanged', () => {
+    const drops: string[] = [];
+    const out = sanitizeUserParts(
+      [
+        { type: 'text', text: 'a' },
+        { type: 'text', text: 'b' },
+      ] as UIMessage['parts'],
+      (t) => drops.push(t),
+    );
+    expect(out).toEqual([
+      { type: 'text', text: 'a' },
+      { type: 'text', text: 'b' },
+    ]);
+    expect(drops).toEqual([]);
+  });
+
+  it('drops a non-text part (a tool-part in input-available) and reports its type', () => {
+    const drops: string[] = [];
+    const out = sanitizeUserParts(
+      [
+        { type: 'text', text: 'hi' },
+        {
+          type: 'tool-getPage',
+          toolCallId: 't1',
+          state: 'input-available',
+          input: { pageId: 'p' },
+        },
+      ] as unknown as UIMessage['parts'],
+      (t) => drops.push(t),
+    );
+    expect(out).toEqual([{ type: 'text', text: 'hi' }]);
+    expect(drops).toEqual(['tool-getPage']);
+  });
+
+  it('drops a null part (the shape that would poison convertToModelMessages)', () => {
+    const drops: string[] = [];
+    const out = sanitizeUserParts(
+      [{ type: 'text', text: 'hi' }, null] as unknown as UIMessage['parts'],
+      (t) => drops.push(t),
+    );
+    expect(out).toEqual([{ type: 'text', text: 'hi' }]);
+    expect(drops).toEqual(['(unknown)']);
+  });
+
+  it('returns undefined when nothing survives (so a null metadata is persisted)', () => {
+    const out = sanitizeUserParts(
+      [
+        { type: 'tool-x', toolCallId: 't', state: 'input-available' },
+      ] as unknown as UIMessage['parts'],
+      () => undefined,
+    );
+    expect(out).toBeUndefined();
+  });
+
+  it('returns undefined for a non-array input', () => {
+    expect(
+      sanitizeUserParts(undefined as unknown as UIMessage['parts'], () => undefined),
+    ).toBeUndefined();
+  });
+});
+
+describe('convertHistoryResilient (#489, branches: happy + per-row degradation)', () => {
+  it('happy path: healthy history converts identically to convertToModelMessages, no degrade', async () => {
+    const history: Row[] = [
+      { id: 'u1', role: 'user', parts: [{ type: 'text', text: 'hi' }] },
+      { id: 'a1', role: 'assistant', parts: [{ type: 'text', text: 'hello' }] },
+    ];
+    const degrades: number[] = [];
+    const out = await convertHistoryResilient(history, (i) => degrades.push(i));
+    const expected = await convertToModelMessages(history as UIMessage[]);
+    expect(out).toEqual(expected);
+    expect(degrades).toEqual([]);
+  });
+
+  it('REAL poison: a null part throws in the batch converter but is isolated and degraded to a marker', async () => {
+    // Sanity: the real converter genuinely throws on this shape.
+    const poisoned: Row = {
+      id: 'a1',
+      role: 'assistant',
+      parts: [
+        { type: 'text', text: 'earlier answer' },
+        null,
+      ] as unknown as UIMessage['parts'],
+    };
+    await expect(
+      convertToModelMessages([poisoned as UIMessage]),
+    ).rejects.toThrow();
+
+    const history: Row[] = [
+      { id: 'u1', role: 'user', parts: [{ type: 'text', text: 'first' }] },
+      poisoned,
+      { id: 'u2', role: 'user', parts: [{ type: 'text', text: 'second' }] },
+    ];
+    const degrades: number[] = [];
+    const out = await convertHistoryResilient(history, (i) => degrades.push(i));
+
+    // Only the poisoned row (index 1) is degraded.
+    expect(degrades).toEqual([1]);
+    // Healthy rows survive verbatim.
+    const flat = JSON.stringify(out);
+    expect(flat).toContain('first');
+    expect(flat).toContain('second');
+    // The degraded row carries its readable text AND the truncation marker so the
+    // model sees that tool context was omitted (never a silent loss).
+    expect(flat).toContain('earlier answer');
+    expect(flat).toContain(TOOL_CONTEXT_OMITTED_MARKER);
+    // The whole batch converted (3 model messages, none dropped).
+    expect(out).toHaveLength(3);
+  });
+
+  it('a fully-poisoned row (no readable text) still degrades to just the marker', async () => {
+    const history: Row[] = [
+      {
+        id: 'a1',
+        role: 'assistant',
+        parts: [null] as unknown as UIMessage['parts'],
+      },
+    ];
+    const out = await convertHistoryResilient(history, () => undefined);
+    expect(out).toHaveLength(1);
+    expect(JSON.stringify(out)).toContain(TOOL_CONTEXT_OMITTED_MARKER);
+  });
+});

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

@@ -97,8 +97,14 @@ describe('AiChatService.stream run-lifecycle safety net (#184)', () => {
     };
     const runService = new AiChatRunService(runRepo as never, { isCloud: () => false } as never);
 
-    // The user-message insert (the first bare await after beginRun) throws.
+    // The user-message insert throws. #489 runs the history load + convert BEFORE
+    // the insert (convert-before-insert, so a retry cannot duplicate the user row),
+    // so `findAllByChat` (a real repo method) is now called first — stub it to an
+    // empty history so the flow reaches the insert. Both awaits are AFTER beginRun,
+    // so the "exception after beginRun -> settled to error" invariant is unchanged;
+    // the throw point simply moved from insert to a later insert after a no-op load.
     const aiChatMessageRepo = {
+      findAllByChat: jest.fn().mockResolvedValue([]),
       insert: jest.fn().mockRejectedValue(new Error('insert boom')),
     };
     const aiChatRepo = {

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

@@ -1440,7 +1440,7 @@ describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', ()
   }
 
   // Wire only the deps reached on the way to the pipe call, plus a spy registry.
-  function makeService(opts: { resumable: boolean }) {
+  function makeService(opts: { resumable: boolean; history?: unknown[] }) {
     const aiChatRepo = {
       findById: jest.fn(async () => ({ id: 'chat-1', workspaceId: 'ws-1' })),
       insert: jest.fn(),
@@ -1448,7 +1448,7 @@ describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', ()
     const aiChatMessageRepo = {
       // Both the user insert and the assistant seed return the same row id.
       insert: jest.fn(async () => ({ id: 'msg-1' })),
-      findAllByChat: jest.fn(async () => []),
+      findAllByChat: jest.fn(async () => opts.history ?? []),
       update: jest.fn(async () => ({ id: 'msg-1' })),
       // #487: the terminal owner-write + the opportunistic reconcile query.
       finalizeOwner: jest.fn(async () => ({ id: 'msg-1' })),
@@ -1487,7 +1487,7 @@ describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', ()
       } as never,
       streamRegistry as never,
     );
-    return { svc, streamRegistry };
+    return { svc, streamRegistry, aiChatMessageRepo };
   }
 
   const body = {
@@ -1570,6 +1570,86 @@ describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', ()
     await expect(drive(svc, makeRunHooks())).rejects.toThrow('boom');
     expect(streamRegistry.abortEntry).toHaveBeenCalledWith('chat-1', 'run-1');
   });
+
+  // #489 REGRESSION (against the REAL convertToModelMessages — not mocked here):
+  // a persisted history row whose parts contain a `null` element makes the real
+  // convertToModelMessages THROW ("Cannot read properties of null"). Pre-fix that
+  // 500-ed every turn forever and each retry appended a duplicate user row. The
+  // fix converts BEFORE the insert and isolates the poisoned row per-row, degrading
+  // it to text with a "[tool context omitted]" marker. Assert the turn still runs,
+  // the marker reaches the model, and exactly ONE user row is inserted.
+  it('#489: a poisoned OLD-history row keeps the chat working; the marker reaches the model; one user insert', async () => {
+    const { svc, aiChatMessageRepo } = makeService({
+      resumable: false,
+      history: [
+        {
+          id: 'old-1',
+          role: 'assistant',
+          content: 'earlier answer',
+          // A null part is the poison: rowToUiMessage keeps it (the array is
+          // non-empty) and the real convertToModelMessages throws on it.
+          metadata: { parts: [{ type: 'text', text: 'earlier answer' }, null] },
+          status: 'completed',
+        },
+      ],
+    });
+    // Must NOT throw — the poisoned row is degraded, not fatal.
+    await drive(svc, makeRunHooks());
+    expect(streamTextMock).toHaveBeenCalledTimes(1);
+    const passedMessages = streamTextMock.mock.calls[0][0].messages;
+    const serialized = JSON.stringify(passedMessages);
+    // The model sees the truncation marker (silent tool-context loss is not ok)
+    // AND the row's readable text is preserved alongside it.
+    expect(serialized).toContain('[tool context omitted]');
+    expect(serialized).toContain('earlier answer');
+    // Exactly ONE user row inserted (no duplicate), inserted AFTER conversion.
+    const userInserts = aiChatMessageRepo.insert.mock.calls
+      .map((c: unknown[]) => c[0] as { role?: string })
+      .filter((r) => r.role === 'user');
+    expect(userInserts).toHaveLength(1);
+  });
+
+  // #489: client-supplied non-text parts (a tool-part in `input-available`, the
+  // exact "bricking" payload) are dropped ON RECEIPT — never persisted — so they
+  // can never poison future turns. Only the text survives into metadata.parts.
+  it('#489: a non-text client part is stripped before persist (only text survives)', async () => {
+    const { svc, aiChatMessageRepo } = makeService({ resumable: false });
+    await svc.stream({
+      user: { id: 'u1' } as never,
+      workspace: { id: 'ws-1' } as never,
+      sessionId: 's1',
+      body: {
+        chatId: 'chat-1',
+        messages: [
+          {
+            id: 'm1',
+            role: 'user',
+            parts: [
+              { type: 'text', text: 'hello' },
+              // untrusted tool-part — must be dropped, never persisted
+              {
+                type: 'tool-getPage',
+                toolCallId: 't1',
+                state: 'input-available',
+                input: { pageId: 'p' },
+              },
+            ],
+          },
+        ],
+      } as never,
+      res: makeRes() as never,
+      signal: new AbortController().signal,
+      model: {} as never,
+      role: null,
+      runHooks: makeRunHooks() as never,
+    });
+    const userInsert = aiChatMessageRepo.insert.mock.calls
+      .map((c: unknown[]) => c[0] as { role?: string; metadata?: unknown })
+      .find((r) => r.role === 'user');
+    const parts = (userInsert?.metadata as { parts?: Array<{ type: string }> })
+      ?.parts;
+    expect(parts).toEqual([{ type: 'text', text: 'hello' }]);
+  });
 });
 
 /**

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

@@ -14,6 +14,7 @@ import {
   convertToModelMessages,
   stepCountIs,
   type UIMessage,
+  type ModelMessage,
   type LanguageModel,
 } from 'ai';
 import { AiService } from '../../integrations/ai/ai.service';
@@ -1042,7 +1043,58 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
       const incoming = lastUserMessage(body.messages);
       const incomingText = uiMessageText(incoming);
 
-      // Persist the user message before contacting the model.
+      // #489: sanitize client-supplied parts ON RECEIPT. The client only ever
+      // sends `sendMessage({ text })` (a single text part); there is no
+      // file/attachment path. Any other part — most dangerously a tool-part in
+      // `input-available` state — is untrusted data that, once persisted to
+      // `metadata.parts` verbatim, is REPLAYED through convertToModelMessages on
+      // every later turn. A malformed tool-part makes that conversion throw,
+      // 500-ing every future turn of the chat forever ("bricked"). Drop any
+      // non-whitelisted part with a warn.
+      const sanitizedParts = sanitizeUserParts(incoming?.parts, (type) =>
+        this.logger.warn(
+          `Dropping unsupported user message part '${type}' on chat ${chatId}`,
+        ),
+      );
+
+      // #489: rebuild the conversation from persisted history (not the client
+      // payload) and CONVERT it to model messages BEFORE persisting the user row.
+      // Load the OLD history (WITHOUT the new row) and append the incoming turn in
+      // memory for the conversion. This makes the insert happen only after a
+      // successful conversion, so a conversion failure cannot leave a DUPLICATE
+      // user row behind on the client's retry (the "bricked chat" that accreted a
+      // dup on every 500). `findAllByChat` returns chronological order (oldest ->
+      // newest) and keeps a 5000-row memory-safety backstop (on overflow it keeps
+      // the NEWEST rows and logs a warning); that is a safety net far above any
+      // realistic chat, not a conversational limit.
+      const oldHistory = await this.aiChatMessageRepo.findAllByChat(
+        chatId,
+        workspace.id,
+      );
+      const uiMessages: Array<Omit<UIMessage, 'id'> & { id: string }> = [
+        ...oldHistory.map(rowToUiMessage),
+        {
+          id: 'pending-user',
+          role: 'user',
+          parts: (sanitizedParts && sanitizedParts.length > 0
+            ? sanitizedParts
+            : textPart(incomingText)) as UIMessage['parts'],
+        },
+      ];
+      // convertToModelMessages is async in ai@6.0.134 (returns Promise<ModelMessage[]>).
+      // Resilient (#489): a single poisoned row in the OLD history is isolated via
+      // per-row conversion and degraded to plain text with a "[tool context
+      // omitted]" marker rather than 500-ing the whole turn (silent loss of tool
+      // context is not acceptable — the model must see the truncation).
+      const messages = await convertHistoryResilient(uiMessages, (index, err) =>
+        this.logger.warn(
+          `Degraded unconvertible history row ${index} on chat ${chatId} to text: ${
+            err instanceof Error ? err.message : 'unknown error'
+          }`,
+        ),
+      );
+
+      // Persist the user message only AFTER a successful conversion (#489).
       await this.aiChatMessageRepo.insert({
         chatId,
         workspaceId: workspace.id,
@@ -1050,31 +1102,21 @@ export class AiChatService implements OnModuleInit, OnModuleDestroy {
         role: 'user',
         content: incomingText,
         // jsonb column: UIMessage parts are JSON-serializable at runtime but not
-        // structurally `JsonValue`, so cast through unknown.
-        metadata: (incoming?.parts ? { parts: incoming.parts } : null) as never,
+        // structurally `JsonValue`, so cast through unknown. Persist the SANITIZED
+        // parts (never the raw client parts) so the row is always convertible.
+        metadata: (sanitizedParts ? { parts: sanitizedParts } : null) as never,
       });
 
-      // Rebuild the conversation from persisted history (not the client payload),
-      // so the model always sees the authoritative server-side transcript. Load
-      // the FULL history in chronological order (oldest -> newest, incl. the user
-      // message just inserted above) so NO turns are dropped — there is no
-      // recent-tail window anymore. `findAllByChat` keeps a 5000-row memory-safety
-      // backstop (on overflow it keeps the NEWEST rows and logs a warning); that
-      // is a safety net far above any realistic chat, not a conversational limit.
-      const history = await this.aiChatMessageRepo.findAllByChat(
-        chatId,
-        workspace.id,
-      );
-      const uiMessages = history.map(rowToUiMessage);
-      // convertToModelMessages is async in ai@6.0.134 (returns Promise<ModelMessage[]>).
-      const messages = await convertToModelMessages(uiMessages);
-
       // Interrupt-resume detection (#198): the client "send now" flag is only a
       // hint — confirm it against the persisted history (the preceding assistant
       // turn must really be aborted/streaming) so a spoofed flag cannot inject the
       // interrupt note onto an ordinary turn. The partial output the model needs is
       // already in `messages` (the aborted assistant row replays via findRecent).
-      const interrupted = isInterruptResume(history, body.interrupted);
+      // Append the new user turn (shape-only) so index -2 is the prior assistant.
+      const interrupted = isInterruptResume(
+        [...oldHistory, { role: 'user', status: null, metadata: null }],
+        body.interrupted,
+      );
 
       // Per-turn page-change detection (#274): if the open page was hand-edited by
       // the user since the agent's last turn ended, compute the unified diff so the
@@ -2074,6 +2116,82 @@ function textPart(text: string): Array<{ type: 'text'; text: string }> {
   return text ? [{ type: 'text', text }] : [];
 }
 
+/**
+ * Part types accepted on an INCOMING user turn (#489). The client only ever
+ * sends `sendMessage({ text })` (a single text part); there is no file/attachment
+ * path. Everything else on a client-supplied user message — most dangerously a
+ * tool-part in `input-available` state — is untrusted data that would be
+ * persisted to `metadata.parts` verbatim and replayed through
+ * `convertToModelMessages` on every later turn, potentially bricking the chat.
+ */
+const ALLOWED_USER_PART_TYPES: ReadonlySet<string> = new Set(['text']);
+
+/**
+ * Keep only whitelisted parts on a client-supplied user message; report each
+ * dropped part's type via `onDrop` (the caller warns). Returns `undefined` when
+ * nothing survives (no parts / none whitelisted), so the caller persists a null
+ * metadata rather than an empty-parts object. Never throws.
+ */
+export function sanitizeUserParts(
+  parts: UIMessage['parts'] | undefined,
+  onDrop: (type: string) => void,
+): UIMessage['parts'] | undefined {
+  if (!Array.isArray(parts)) return undefined;
+  const kept = parts.filter((p) => {
+    const type =
+      typeof (p as { type?: unknown })?.type === 'string'
+        ? (p as { type: string }).type
+        : '';
+    if (ALLOWED_USER_PART_TYPES.has(type)) return true;
+    onDrop(type || '(unknown)');
+    return false;
+  });
+  return kept.length > 0 ? (kept as UIMessage['parts']) : undefined;
+}
+
+/** Marker for a history row whose tool parts could not be replayed (#489). */
+export const TOOL_CONTEXT_OMITTED_MARKER = '[tool context omitted]';
+
+/**
+ * Convert persisted UI history to model messages, tolerating a single poisoned
+ * row (#489). `convertToModelMessages` over the WHOLE array throws if ANY row is
+ * malformed (e.g. a tool-part left unbalanced / in `input-available` state),
+ * which would otherwise 500 every turn of the chat forever. On a batch failure we
+ * fall back to per-row conversion so the bad row is isolated: it is degraded to
+ * plain text carrying its readable text plus a `[tool context omitted]` marker
+ * (the model MUST see that its tool context was truncated — silent loss is not
+ * acceptable), while every healthy row converts normally. Because AI SDK v6
+ * carries a tool call and its result inside the SAME assistant UIMessage's parts,
+ * per-row conversion preserves call/result pairing.
+ */
+export async function convertHistoryResilient(
+  uiMessages: Array<Omit<UIMessage, 'id'> & { id: string }>,
+  onDegrade: (index: number, err: unknown) => void,
+): Promise<ModelMessage[]> {
+  try {
+    return await convertToModelMessages(uiMessages as UIMessage[]);
+  } catch {
+    const out: ModelMessage[] = [];
+    for (let i = 0; i < uiMessages.length; i++) {
+      const m = uiMessages[i];
+      try {
+        out.push(...(await convertToModelMessages([m as UIMessage])));
+      } catch (err) {
+        onDegrade(i, err);
+        const text = uiMessageText(m as UIMessage);
+        const degraded = text
+          ? `${text}\n\n${TOOL_CONTEXT_OMITTED_MARKER}`
+          : TOOL_CONTEXT_OMITTED_MARKER;
+        out.push({
+          role: m.role === 'assistant' ? 'assistant' : 'user',
+          content: degraded,
+        } as ModelMessage);
+      }
+    }
+    return out;
+  }
+}
+
 /**
  * Minimal shapes of the AI SDK v6 step objects we read to rebuild UIMessage
  * parts (see ai@6.0.134 `StepResult`: `text`, `toolCalls` -> TypedToolCall,

apps/server/src/core/ai-chat/external-mcp/mcp-clients.recovery.spec.ts

@@ -0,0 +1,261 @@
+import { errors } from 'undici';
+import {
+  McpClientsService,
+  isRetryableConnectError,
+} from './mcp-clients.service';
+
+/**
+ * #489 — external-MCP in-run transport recovery.
+ *
+ * The transport-error classification + retry gate are exercised against the REAL
+ * undici error CLASSES prod throws (`errors.SocketError` / `errors.BodyTimeoutError`,
+ * carrying the true `UND_ERR_*` codes and class names), wrapped EXACTLY as undici's
+ * `fetch` wraps them — a `TypeError('fetch failed'|'terminated')` whose `.cause` is
+ * the undici error. These are the real classes, not hand-rolled `{code:'...'}`
+ * mocks: constructing the genuine class is what makes this a faithful test of the
+ * prod predicate (epic root-cause #4 — a mock-shaped predicate would leave the
+ * evict/retry path silently dead in production while CI stays green). We construct
+ * rather than drive a live fetch because Jest's environment degrades the live-fetch
+ * error to a generic `Error` cause (no undici code), which would NOT be the prod
+ * shape.
+ */
+
+/** A REAL undici socket reset, wrapped as fetch wraps it. */
+function realSocketResetError(): unknown {
+  const err = new TypeError('fetch failed');
+  (err as { cause?: unknown }).cause = new errors.SocketError('other side closed');
+  return err;
+}
+
+/** A REAL undici body timeout, wrapped as fetch wraps it. */
+function realBodyTimeoutError(): unknown {
+  const err = new TypeError('terminated');
+  (err as { cause?: unknown }).cause = new errors.BodyTimeoutError();
+  return err;
+}
+
+type FakeServer = {
+  id: string;
+  name: string;
+  transport: 'http' | 'sse';
+  url: string;
+  headersEnc: string | null;
+  toolAllowlist: string[] | null;
+  instructions: string | null;
+};
+
+const server = (over: Partial<FakeServer> = {}): FakeServer => ({
+  id: 's1',
+  name: 'srv',
+  transport: 'http',
+  url: 'http://example.test/mcp',
+  headersEnc: null,
+  toolAllowlist: null,
+  instructions: null,
+  ...over,
+});
+
+function buildService(servers: FakeServer[], trusted = false) {
+  const repo = { listEnabled: jest.fn().mockResolvedValue(servers) };
+  const service = new McpClientsService(repo as never, {} as never);
+  // Seed a DETERMINISTIC write-class map so the retry gate is controlled here
+  // (the production map loads from @docmost/mcp via a dynamic ESM import). getPage
+  // is a read, patchNode is a write — the real classifications.
+  (
+    service as unknown as { writeClassMapPromise: Promise<unknown> }
+  ).writeClassMapPromise = Promise.resolve({
+    getPage: 'readOnly',
+    patchNode: 'write',
+  });
+  // The service only APPLIES that map to a TRUSTED internal Docmost server
+  // (isInternalDocmostServer, really false for every third-party row). A retry
+  // test needs a trusted server to exercise the readOnly-retry path at all, so it
+  // passes trusted=true to model a Docmost-origin server; the third-party
+  // double-apply test leaves it at the real value (false).
+  if (trusted) {
+    jest
+      .spyOn(
+        service as unknown as {
+          isInternalDocmostServer: (s: FakeServer) => boolean;
+        },
+        'isInternalDocmostServer',
+      )
+      .mockReturnValue(true);
+  }
+  return { service, repo };
+}
+
+/** Spy the private `connect` so each call yields a controlled fake client whose
+ *  single tool's execute is the supplied function. Returns the connect spy. */
+function stubConnect(
+  service: McpClientsService,
+  toolName: string,
+  execs: Array<(...a: unknown[]) => Promise<unknown>>,
+) {
+  let n = 0;
+  return jest
+    .spyOn(
+      service as unknown as { connect: (s: FakeServer) => Promise<unknown> },
+      'connect',
+    )
+    .mockImplementation(async () => {
+      const exec = execs[Math.min(n, execs.length - 1)];
+      n += 1;
+      return {
+        tools: async () => ({ [toolName]: { description: 'x', execute: exec } }),
+        close: jest.fn().mockResolvedValue(undefined),
+      };
+    });
+}
+
+const opts = (abortSignal?: AbortSignal) =>
+  ({ toolCallId: 't', messages: [], abortSignal }) as never;
+
+describe('isRetryableConnectError (#489, REAL error shapes)', () => {
+  it('classifies a real undici socket reset and body timeout as retryable', async () => {
+    const socketErr = await realSocketResetError();
+    const bodyErr = await realBodyTimeoutError();
+    expect(isRetryableConnectError(socketErr)).toBe(true);
+    expect(isRetryableConnectError(bodyErr)).toBe(true);
+    // Unwraps a wrapped cause chain (e.g. an MCPClientError around the socket err).
+    const wrapped = new Error('mcp call failed');
+    (wrapped as { cause?: unknown }).cause = socketErr;
+    expect(isRetryableConnectError(wrapped)).toBe(true);
+  });
+
+  it('does NOT classify an application-level error as a transport break', () => {
+    expect(isRetryableConnectError(new Error('validation failed'))).toBe(false);
+    expect(isRetryableConnectError({ name: 'HttpError', status: 400 })).toBe(false);
+    expect(isRetryableConnectError(undefined)).toBe(false);
+    expect(isRetryableConnectError('boom')).toBe(false);
+  });
+});
+
+describe('McpClientsService in-run transport recovery (#489)', () => {
+  afterEach(() => jest.restoreAllMocks());
+
+  it('a readOnly tool whose transport breaks reconnects and retries WITHIN the same run', async () => {
+    const realErr = await realSocketResetError();
+    const { service } = buildService([server()], true);
+    const first = jest.fn().mockRejectedValue(realErr);
+    const second = jest.fn().mockResolvedValue({ ok: true });
+    const connectSpy = stubConnect(service, 'getPage', [first, second]);
+
+    const toolset = await service.toolsFor('ws-1');
+    const tool = toolset.tools['srv_getPage'];
+    const result = await (tool.execute as (a: unknown, o: unknown) => Promise<unknown>)(
+      { pageId: 'p' },
+      opts(),
+    );
+
+    // The repeat call within the run got a LIVE client and succeeded.
+    expect(result).toEqual({ ok: true });
+    expect(first).toHaveBeenCalledTimes(1);
+    expect(second).toHaveBeenCalledTimes(1);
+    // Exactly one reconnect was minted (initial build connect + one recovery).
+    expect(connectSpy).toHaveBeenCalledTimes(2);
+    // The run accumulated BOTH leases (old + reconnected) — released together at end.
+    expect(toolset.clients).toHaveLength(2);
+    await Promise.all(toolset.clients.map((c) => c.close()));
+  });
+
+  it('a WRITE tool does NOT auto-retry on a transport error (indeterminate)', async () => {
+    const realErr = await realSocketResetError();
+    const { service } = buildService([server()], true);
+    const exec = jest.fn().mockRejectedValue(realErr);
+    const connectSpy = stubConnect(service, 'patchNode', [exec]);
+
+    const toolset = await service.toolsFor('ws-2');
+    const tool = toolset.tools['srv_patchNode'];
+    await expect(
+      (tool.execute as (a: unknown, o: unknown) => Promise<unknown>)(
+        { pageId: 'p' },
+        opts(),
+      ),
+    ).rejects.toThrow(/MAY have already applied/);
+
+    // Called exactly once — NO blind retry (avoids double-apply, the #435 class).
+    expect(exec).toHaveBeenCalledTimes(1);
+    // No fresh connection was minted for a write.
+    expect(connectSpy).toHaveBeenCalledTimes(1);
+    await Promise.all(toolset.clients.map((c) => c.close()));
+  });
+
+  it('does NOT retry (or reconnect) after the run is aborted (Stop)', async () => {
+    const realErr = await realSocketResetError();
+    const { service } = buildService([server()], true);
+    const controller = new AbortController();
+    // The transport error arrives, but the run was Stopped in the same tick.
+    const first = jest.fn().mockImplementation(async () => {
+      controller.abort();
+      throw realErr;
+    });
+    const second = jest.fn().mockResolvedValue({ ok: true });
+    const connectSpy = stubConnect(service, 'getPage', [first, second]);
+
+    const toolset = await service.toolsFor('ws-3');
+    const tool = toolset.tools['srv_getPage'];
+    await expect(
+      (tool.execute as (a: unknown, o: unknown) => Promise<unknown>)(
+        { pageId: 'p' },
+        opts(controller.signal),
+      ),
+    ).rejects.toBeDefined();
+
+    // getPage IS readOnly, but the Stop blocks the retry — no second call, no mint.
+    expect(second).not.toHaveBeenCalled();
+    expect(connectSpy).toHaveBeenCalledTimes(1);
+    await Promise.all(toolset.clients.map((c) => c.close()));
+  });
+
+  it('an app-level (non-transport) tool error is surfaced verbatim, never retried', async () => {
+    const { service } = buildService([server()], true);
+    const appErr = new Error('tool says: bad input');
+    const exec = jest.fn().mockRejectedValue(appErr);
+    const connectSpy = stubConnect(service, 'getPage', [exec]);
+
+    const toolset = await service.toolsFor('ws-4');
+    const tool = toolset.tools['srv_getPage'];
+    await expect(
+      (tool.execute as (a: unknown, o: unknown) => Promise<unknown>)(
+        { pageId: 'p' },
+        opts(),
+      ),
+    ).rejects.toThrow('tool says: bad input');
+    expect(exec).toHaveBeenCalledTimes(1);
+    expect(connectSpy).toHaveBeenCalledTimes(1); // no reconnect for an app error
+    await Promise.all(toolset.clients.map((c) => c.close()));
+  });
+
+  // #489 (review, MEDIUM) — the Docmost write-class map keys by DOCMOST tool
+  // names; a THIRD-PARTY server may name a WRITE tool `getPage` (a Docmost read
+  // name). It must NOT inherit readOnly and must NOT auto-retry on a transport
+  // error — a blind retry of that write is a double-apply (the #435 class). Here
+  // the server is UNTRUSTED (buildService default, isInternalDocmostServer=false),
+  // so the map is not applied and `getPage` classifies as a write.
+  //
+  // MUTATION-VERIFY: forcing the server "trusted" (buildService(..., true)) makes
+  // `getPage` inherit readOnly -> it WOULD reconnect+retry (connect twice) and the
+  // assertions below fail — i.e. removing the trust scope re-opens the bug.
+  it('a THIRD-PARTY WRITE tool named like a Docmost read does NOT auto-retry (no double-apply)', async () => {
+    const realErr = await realSocketResetError();
+    // Untrusted: default trusted=false — a real third-party server.
+    const { service } = buildService([server()]);
+    const exec = jest.fn().mockRejectedValue(realErr);
+    const connectSpy = stubConnect(service, 'getPage', [exec, exec]);
+
+    const toolset = await service.toolsFor('ws-5');
+    const tool = toolset.tools['srv_getPage'];
+    await expect(
+      (tool.execute as (a: unknown, o: unknown) => Promise<unknown>)(
+        { pageId: 'p' },
+        opts(),
+      ),
+    ).rejects.toThrow(/MAY have already applied/);
+
+    // Exactly one call, NO reconnect — the name collision granted no readOnly-retry.
+    expect(exec).toHaveBeenCalledTimes(1);
+    expect(connectSpy).toHaveBeenCalledTimes(1);
+    await Promise.all(toolset.clients.map((c) => c.close()));
+  });
+});

apps/server/src/core/ai-chat/external-mcp/mcp-clients.service.spec.ts

@@ -106,8 +106,11 @@ describe('McpClientsService.decryptHeaders', () => {
 
 describe('McpClientsService.guardedFetch (SSRF per-request guard)', () => {
   // The bound guardedFetch closure lives on the instance as a private field.
+  // #489 split it into per-transport HTTP/SSE bindings (they differ only in the
+  // dispatcher's bodyTimeout); the SSRF guard is identical, so testing the HTTP
+  // one is sufficient.
   const guardedFetchOf = (service: McpClientsService) =>
-    (service as unknown as { guardedFetch: typeof fetch }).guardedFetch;
+    (service as unknown as { guardedFetchHttp: typeof fetch }).guardedFetchHttp;
 
   let fetchSpy: jest.SpiedFunction<typeof fetch>;
 

apps/server/src/core/ai-chat/external-mcp/mcp-clients.service.ts

@@ -1,5 +1,6 @@
 import { isIP } from 'node:net';
 import { lookup as dnsLookup, type LookupAddress } from 'node:dns';
+import { pathToFileURL } from 'node:url';
 import { Injectable, Logger } from '@nestjs/common';
 import { type Tool, type ToolCallOptions } from 'ai';
 import { createMCPClient } from '@ai-sdk/mcp';
@@ -10,9 +11,29 @@ import {
   streamingDispatcherOptions,
   mcpStreamTimeoutMs,
   mcpCallTimeoutMs,
+  mcpSseBodyTimeoutMs,
 } from '../../../integrations/ai/ai-streaming-fetch';
 import { SecretBoxService } from '../../../integrations/crypto/secret-box';
 import { isUrlAllowed, isIpAllowed } from './ssrf-guard';
+// TYPE-ONLY (erased at compile): @docmost/mcp is ESM-only and cannot be a runtime
+// `require()` from this commonjs module (same constraint as docmost-client.loader).
+// The write-class MAP is loaded lazily via the dynamic-import trick below.
+import type { ToolWriteClass } from '@docmost/mcp';
+
+// TS(commonjs) downlevels a literal `import()` to `require()`, which cannot load
+// the ESM-only @docmost/mcp. Indirect through Function so the real dynamic
+// `import()` survives compilation (same trick as docmost-client.loader.ts).
+const esmImport = new Function(
+  'specifier',
+  'return import(specifier)',
+) as (specifier: string) => Promise<unknown>;
+
+/** Local read-only predicate — avoids a value import of the ESM-only package.
+ *  Only a pure read is retry-safe after a transport break (a write is
+ *  indeterminate). Kept in lockstep with @docmost/mcp's isRetryableWriteClass. */
+function isReadOnlyWriteClass(writeClass: ToolWriteClass | undefined): boolean {
+  return writeClass === 'readOnly';
+}
 
 /** A closable external MCP client handle. */
 export interface Closable {
@@ -81,12 +102,52 @@ const MAX_TOOL_NAME_LENGTH = 64;
  * close until the turn releases it, so a TTL expiry mid-turn never closes a
  * client a stream is still executing against.
  */
+/**
+ * Where a merged (namespaced) tool came from, so the per-run recovery wrapper
+ * (#489) can, on a transport error, reconnect THAT server and re-resolve the SAME
+ * underlying tool by its raw name. `writeClass` gates the single auto-retry (a
+ * read is retry-safe; a write is indeterminate). `serverIndex` indexes the
+ * entry's `servers` array (which server config to reconnect).
+ */
+interface ToolProvenance {
+  serverIndex: number;
+  rawName: string;
+  writeClass: ToolWriteClass | undefined;
+}
+
+/** A live reconnected server (its fresh client + raw call-timeout-wrapped tools). */
+interface RecoveredServerState {
+  client: McpClient;
+  tools: Record<string, Tool>;
+}
+
+/**
+ * Per-run, per-server recovery binding (#489). `current` is the server's LIVE
+ * target for this run: `null` means "use the ORIGINAL cached client/template";
+ * a non-null value is a reconnected throwaway client all this server's tools now
+ * call. `reconnecting` dedupes concurrent reconnects so only ONE fresh client is
+ * minted per death (a losing concurrent call awaits it and retries on the SAME
+ * new client — the CAS-by-identity rule).
+ */
+interface ServerBinding {
+  current: RecoveredServerState | null;
+  reconnecting?: Promise<RecoveredServerState>;
+}
+
 interface CacheEntry {
   tools: Record<string, Tool>;
   clients: McpClient[];
   outcomes: ServerOutcome[];
   /** Prompt guidance for qualifying servers (see McpServerInstruction). */
   instructions: McpServerInstruction[];
+  /**
+   * The enabled server configs used to build this entry (#489), so the per-run
+   * recovery wrapper can reconnect a specific server by index. Parallel to the
+   * indices referenced by {@link toolMeta}.
+   */
+  servers: AiMcpServer[];
+  /** merged-tool-key -> provenance (#489), for the per-run recovery wrapper. */
+  toolMeta: Record<string, ToolProvenance>;
   expiresAt: number;
   /** Active leases (turns currently using these clients). */
   refCount: number;
@@ -120,20 +181,82 @@ export class McpClientsService {
    */
   private readonly cache = new Map<string, Promise<CacheEntry>>();
   /**
-   * A single shared SSRF-pinned dispatcher for ALL outbound external-MCP fetches.
-   * Its custom connect.lookup runs per connection, so one instance safely guards
-   * every server's connections (we never connect to an unvalidated IP).
+   * SSRF-pinned dispatchers for outbound external-MCP fetches. Both use the SAME
+   * custom connect.lookup (so every connection is IP-validated), but carry a
+   * DIFFERENT `bodyTimeout` (#489): the HTTP (streamable) transport opens a fresh
+   * request per call, so it keeps the tight silence timeout; the SSE transport
+   * holds ONE long-lived body open across many calls, so a >1-min idle BETWEEN
+   * calls is LEGITIMATE and must not break the socket — it gets a much larger
+   * bodyTimeout. (headersTimeout stays tight on both.)
    */
-  private readonly dispatcher: Dispatcher = buildPinnedDispatcher();
-  /** guardedFetch bound to the pinned dispatcher; reused by every transport. */
-  private readonly guardedFetch: typeof fetch = (input, init) =>
-    guardedFetch(this.dispatcher, input, init);
+  private readonly dispatcherHttp: Dispatcher = buildPinnedDispatcher(
+    mcpStreamTimeoutMs(),
+  );
+  private readonly dispatcherSse: Dispatcher = buildPinnedDispatcher(
+    mcpSseBodyTimeoutMs(),
+  );
+  /** guardedFetch bound to each dispatcher; picked by transport type in connect(). */
+  private readonly guardedFetchHttp: typeof fetch = (input, init) =>
+    guardedFetch(this.dispatcherHttp, input, init);
+  private readonly guardedFetchSse: typeof fetch = (input, init) =>
+    guardedFetch(this.dispatcherSse, input, init);
+
+  /**
+   * Memoized write-class map (#489), loaded lazily from @docmost/mcp via the
+   * dynamic-import trick. Keyed by tool name (=== mcpName). A tool NOT in the map
+   * (any third-party external MCP tool) classifies as `undefined` -> treated as a
+   * write by the retry gate (the safe default: never blind-retry an unknown tool).
+   * On any load failure the map is `{}` (every tool -> no auto-retry), so a
+   * missing/older @docmost/mcp build only DISABLES retries, never mis-retries.
+   */
+  private writeClassMapPromise: Promise<Record<string, ToolWriteClass>> | null =
+    null;
 
   constructor(
     private readonly repo: AiMcpServerRepo,
     private readonly secretBox: SecretBoxService,
   ) {}
 
+  /**
+   * Whether an external MCP server is the TRUSTED internal Docmost MCP server —
+   * the only server whose tools may be classified by the Docmost write-class map
+   * (#489 review). Today this is ALWAYS false: every `ai_mcp_servers` row is an
+   * admin-configured THIRD-PARTY endpoint (there is no builtin/self flag, sentinel
+   * URL, or synthetic server in this path — Docmost's OWN tools are exposed via the
+   * separate in-app tools path, never through this external-MCP client). So no
+   * third-party tool can inherit `readOnly` by a name collision with a Docmost read
+   * tool, and none is ever auto-retried on a transport error (which would risk a
+   * double-apply — the #435 class). Flip this (an explicit `kind`/`isBuiltin`
+   * column, or a configured self-MCP URL) if a trusted internal server is ever
+   * introduced. A method (not a free function) so it is a single, mockable seam.
+   */
+  private isInternalDocmostServer(_server: AiMcpServer): boolean {
+    return false;
+  }
+
+  /** Lazily load + memoize the shared write-class map (see the field doc). */
+  private getWriteClassMap(): Promise<Record<string, ToolWriteClass>> {
+    if (!this.writeClassMapPromise) {
+      this.writeClassMapPromise = (async () => {
+        try {
+          const entry = require.resolve('@docmost/mcp');
+          const mod = (await esmImport(pathToFileURL(entry).href)) as {
+            SHARED_TOOL_WRITE_CLASS?: Record<string, ToolWriteClass>;
+          };
+          return mod.SHARED_TOOL_WRITE_CLASS ?? {};
+        } catch (err) {
+          this.logger.warn(
+            `Could not load MCP write-class map (auto-retry disabled): ${shortError(
+              err,
+            )}`,
+          );
+          return {};
+        }
+      })();
+    }
+    return this.writeClassMapPromise;
+  }
+
   /**
    * Build (or reuse a cached) external toolset for a workspace. Returns the
    * merged tools, the open client handles to release, and per-server outcomes.
@@ -162,11 +285,37 @@ export class McpClientsService {
         }
       },
     };
-    // One release handle drives the whole leased entry; closing it releases all
-    // underlying clients together (they share the same lease lifecycle).
+
+    // #489: the run accumulates a SET of leases — the primary cache lease PLUS any
+    // throwaway client minted by an in-run transport-recovery reconnect. They are
+    // NEVER released mid-run (releasing a swapped-out client while a concurrent
+    // in-flight call still holds it would INDUCE a second failure); the caller
+    // releases the WHOLE set together at turn-end. A recovery reconnect pushes its
+    // lease onto this live array, which the consumer closes over.
+    const leaseSet: Closable[] = [release];
+
+    // #489: per-RUN transport-recovery binding, one per server, SHARED by all of
+    // that server's tools so a swap by one call is seen by the next (CAS by
+    // identity). Kept per-run (here, not in the cached entry) because the binding
+    // + lease-set state is per-run.
+    const bindings = new Map<number, ServerBinding>();
+    const capMs = mcpCallTimeoutMs();
+
+    // Wrap each cached tool with the recovery layer. On a transport error a
+    // declared readOnly tool reconnects its server and retries ONCE; a write is
+    // never blind-retried (indeterminate — may have applied before the reset). A
+    // tool without provenance (a minimal stub entry in a test) passes through raw.
+    const tools: Record<string, Tool> = {};
+    for (const [key, tool] of Object.entries(entry.tools)) {
+      const meta = entry.toolMeta?.[key];
+      tools[key] = meta
+        ? this.wrapWithTransportRecovery(entry, meta, tool, leaseSet, bindings, capMs)
+        : tool;
+    }
+
     return {
-      tools: entry.tools,
-      clients: [release],
+      tools,
+      clients: leaseSet,
       outcomes: entry.outcomes,
       instructions: entry.instructions,
     };
@@ -254,6 +403,16 @@ export class McpClientsService {
     // Per-call total wall-clock cap, read once for this build (env-overridable).
     const callTimeoutMs = mcpCallTimeoutMs();
     const instructions: McpServerInstruction[] = [];
+    // merged-key -> provenance for the per-run recovery wrapper (#489).
+    const toolMeta: Record<string, ToolProvenance> = {};
+    // Shared Docmost write-class map (#489) — classifies a tool by its raw name.
+    // Loaded ONLY when at least one server is a TRUSTED internal Docmost server
+    // (see isInternalDocmostServer): for third-party servers the map is never
+    // applied (a name collision must not grant readOnly-retry), so we skip the
+    // dynamic ESM load entirely in that (currently universal) case.
+    const writeClassMap = servers.some((s) => this.isInternalDocmostServer(s))
+      ? await this.getWriteClassMap()
+      : null;
 
     // Per-server connect+tools result, still tagged with its server so the merge
     // below can be applied in the SAME order as `servers` (see the parallel note).
@@ -327,11 +486,23 @@ export class McpClientsService {
       // against names already merged from earlier servers, so no external
       // tool is silently overwritten on collision. The returned count drives
       // whether this server's prompt guidance is included (≥1 tool merged).
+      // #489 (review): the Docmost write-class map keys by DOCMOST tool names and
+      // may ONLY be trusted for a server KNOWN to be the internal Docmost MCP
+      // server. Every row here is an admin-configured THIRD-PARTY endpoint, so a
+      // third-party WRITE tool that happens to be named like a Docmost read
+      // (getPage, listPages, ...) must NOT inherit readOnly — that would auto-retry
+      // a mutation on a transport error (double-apply, the #435 class). Gate the
+      // map on the trust check; untrusted servers get writeClass=undefined -> the
+      // recovery wrapper treats them as writes and never auto-retries.
+      const trustWriteClass = this.isInternalDocmostServer(server);
       const merged = this.mergeNamespaced(
         tools,
         result.guarded,
         server.name,
         server.id,
+        toolMeta,
+        i,
+        trustWriteClass ? writeClassMap : null,
       );
       outcomes.push({ name: server.name, ok: true });
       // Include this server's guidance ONLY when it actually contributed at
@@ -353,6 +524,8 @@ export class McpClientsService {
       clients,
       outcomes,
       instructions,
+      servers,
+      toolMeta,
       expiresAt: Date.now() + CACHE_TTL_MS,
       refCount: 0,
       evicted: false,
@@ -379,18 +552,33 @@ export class McpClientsService {
     picked: Record<string, Tool>,
     serverName: string,
     serverId: string,
+    toolMeta: Record<string, ToolProvenance>,
+    serverIndex: number,
+    // The Docmost write-class map, or `null` for an UNTRUSTED (third-party)
+    // server whose tools must all default to write (never auto-retried).
+    writeClassMap: Record<string, ToolWriteClass> | null,
   ): { count: number; prefix: string } {
     let count = 0;
-    for (const [name, tool] of Object.entries(namespace(picked, serverName))) {
-      let key = name;
+    for (const { full, raw, tool } of namespace(picked, serverName)) {
+      let key = full;
       if (key in target) {
         const original = key;
-        key = disambiguate(name, serverId, (candidate) => candidate in target);
+        key = disambiguate(full, serverId, (candidate) => candidate in target);
         this.logger.debug(
           `External MCP tool name "${original}" collided; renamed to "${key}"`,
         );
       }
       target[key] = tool;
+      // Record provenance so the per-run recovery wrapper (#489) can reconnect
+      // this tool's server and re-resolve it by its raw name. writeClass is set
+      // ONLY from a TRUSTED (internal-Docmost) map; for a third-party server the
+      // map is null -> writeClass stays undefined -> the wrapper treats the tool
+      // as a write and never auto-retries it (no double-apply on name collision).
+      toolMeta[key] = {
+        serverIndex,
+        rawName: raw,
+        writeClass: writeClassMap ? writeClassMap[raw] : undefined,
+      };
       count += 1;
     }
     return { count, prefix: namespacePrefix(serverName) };
@@ -424,7 +612,10 @@ export class McpClientsService {
         // Defense in depth: re-validate the actual request host on EVERY fetch
         // AND pin the socket to a validated IP via the dispatcher's connect
         // lookup, closing the DNS-rebinding TOCTOU between check and connect.
-        fetch: this.guardedFetch,
+        // #489: the SSE transport uses the raised-bodyTimeout dispatcher (idle
+        // between calls is legit); HTTP uses the tight one.
+        fetch:
+          transportType === 'sse' ? this.guardedFetchSse : this.guardedFetchHttp,
       },
     })) as unknown as McpClient;
     return client;
@@ -505,6 +696,176 @@ export class McpClientsService {
     }
   }
 
+  /**
+   * Wrap one merged external tool with the per-run transport-recovery layer (#489).
+   *
+   * attempt 1 runs on the server's CURRENT binding (the cached client, or a client
+   * a sibling tool already reconnected this run). On a REAL transport error
+   * (undici/@ai-sdk socket/body-timeout shapes — {@link isRetryableConnectError},
+   * NOT a mock) and ONLY for a declared readOnly tool, it reconnects the server
+   * and retries EXACTLY ONCE on the fresh client; a write is surfaced as an
+   * indeterminate error (it may have applied before the reset — never
+   * blind-retried). A single per-call cap bounds BOTH attempts + the reconnect,
+   * and the run's abort signal is checked before the retry AND before minting a
+   * fresh connection (no connection is opened for a stopped run).
+   */
+  private wrapWithTransportRecovery(
+    entry: CacheEntry,
+    meta: ToolProvenance,
+    template: Tool,
+    leaseSet: Closable[],
+    bindings: Map<number, ServerBinding>,
+    capMs: number,
+  ): Tool {
+    const original = template.execute;
+    if (typeof original !== 'function') return template;
+    const service = this;
+    const { serverIndex, rawName, writeClass } = meta;
+
+    let binding = bindings.get(serverIndex);
+    if (!binding) {
+      binding = { current: null };
+      bindings.set(serverIndex, binding);
+    }
+    const boundBinding = binding;
+
+    const execute = async (args: unknown, options: ToolCallOptions) => {
+      // The per-call cap governs the WHOLE sequence (attempt1 + reconnect +
+      // attempt2). Compose it with the run's abort signal so a Stop or the cap
+      // ends any awaited call — @ai-sdk/mcp does not settle on abort, so we RACE.
+      const capController = new AbortController();
+      const capTimer = setTimeout(() => {
+        capController.abort(new Error(`MCP tool call timed out after ${capMs}ms`));
+      }, capMs);
+      capTimer.unref?.();
+      const runSignal = options?.abortSignal;
+      const composed = runSignal
+        ? AbortSignal.any([runSignal, capController.signal])
+        : capController.signal;
+      const stopped = () => runSignal?.aborted === true || capController.signal.aborted;
+
+      const callOn = async (
+        exec: NonNullable<Tool['execute']>,
+      ): Promise<unknown> => {
+        const aborted = new Promise<never>((_, reject) => {
+          const fail = () => reject(abortReason(composed));
+          if (composed.aborted) fail();
+          else composed.addEventListener('abort', fail, { once: true });
+        });
+        return Promise.race([exec(args, { ...options, abortSignal: composed }), aborted]);
+      };
+
+      const execFor = (
+        state: RecoveredServerState | null,
+      ): NonNullable<Tool['execute']> | undefined =>
+        state ? (state.tools[rawName]?.execute as NonNullable<Tool['execute']>) : original;
+
+      try {
+        // Snapshot the target BEFORE the call so a swap by a concurrent call is
+        // detected by identity in the catch.
+        const attemptState = boundBinding.current;
+        const attemptExec = execFor(attemptState);
+        if (typeof attemptExec !== 'function') {
+          throw new Error(`external MCP tool "${rawName}" is not callable`);
+        }
+        try {
+          return await callOn(attemptExec);
+        } catch (err) {
+          // Never retry on a Stop or an exhausted cap.
+          if (stopped()) throw err;
+          // Only a genuine transport break is a recovery candidate.
+          if (!isRetryableConnectError(err)) throw err;
+          // A write tool is INDETERMINATE on a transport error (may have applied
+          // before the reset) — surface that; do NOT auto-retry (double-apply is
+          // the #435 incident class).
+          if (!isReadOnlyWriteClass(writeClass)) {
+            throw new Error(
+              `external MCP tool "${rawName}" hit a transport error and MAY have already ` +
+                `applied on the server — not retried automatically; verify state before ` +
+                `retrying. (${shortError(err)})`,
+            );
+          }
+          // Abort check BEFORE minting a fresh connection (no socket for a
+          // stopped run). LIMITATION (#489, LOW): the reconnect's own connect is
+          // bounded by CONNECT_TIMEOUT_MS but does NOT itself observe `composed`,
+          // so a Stop that lands DURING the handshake is only honored at the next
+          // `stopped()` gate (before the retry) — a bounded ≤5s late-abort window;
+          // the throwaway client is closed at turn-end regardless. Threading
+          // `composed` into the SHARED (CAS-deduped) reconnect is deliberately
+          // avoided: it would let the first caller's abort tear down a reconnect a
+          // concurrent still-live caller depends on.
+          if (stopped()) throw err;
+          // CAS-swap by IDENTITY: mint+swap only if nobody swapped since this
+          // call's snapshot; a losing concurrent call awaits the same reconnect
+          // and retries on the SAME fresh client.
+          let target: RecoveredServerState;
+          if (boundBinding.current === attemptState) {
+            if (!boundBinding.reconnecting) {
+              boundBinding.reconnecting = (async () => {
+                const server = entry.servers[serverIndex];
+                const fresh = await service.reconnectServer(server, capMs);
+                leaseSet.push(fresh.lease); // accumulate; released at turn-end
+                boundBinding.current = fresh.state;
+                return fresh.state;
+              })();
+              // Clear the in-flight marker once it settles (success or failure) so
+              // a LATER death of the new client can reconnect again.
+              void boundBinding.reconnecting.then(
+                () => (boundBinding.reconnecting = undefined),
+                () => (boundBinding.reconnecting = undefined),
+              );
+            }
+            target = await boundBinding.reconnecting;
+          } else {
+            target = boundBinding.current as RecoveredServerState;
+          }
+          // Abort check BEFORE the retry.
+          if (stopped()) throw err;
+          const retryExec = execFor(target);
+          if (typeof retryExec !== 'function') throw err;
+          return await callOn(retryExec);
+        }
+      } finally {
+        clearTimeout(capTimer);
+      }
+    };
+    return { ...template, execute } as unknown as Tool;
+  }
+
+  /**
+   * Reconnect ONE server for an in-run recovery (#489): open a fresh client and
+   * list+wrap its tools. The throwaway client is NOT cached — it is owned by the
+   * RUN via the returned lease (closed at turn-end), independent of the shared
+   * cache entry (whose TTL rebuild heals future turns). On a failure the fresh
+   * client is closed so its socket never leaks.
+   */
+  private async reconnectServer(
+    server: AiMcpServer,
+    capMs: number,
+  ): Promise<{ state: RecoveredServerState; lease: Closable }> {
+    const client = await this.connectWithTimeout(server, CONNECT_TIMEOUT_MS);
+    let tools: Record<string, Tool>;
+    try {
+      const raw = await withTimeout(client.tools(), CONNECT_TIMEOUT_MS);
+      const allow = server.toolAllowlist;
+      const picked =
+        Array.isArray(allow) && allow.length > 0 ? pick(raw, allow) : raw;
+      tools = wrapToolsWithCallTimeout(picked, capMs);
+    } catch (err) {
+      void client.close().catch(() => undefined);
+      throw err;
+    }
+    let released = false;
+    const lease: Closable = {
+      close: async () => {
+        if (released) return;
+        released = true;
+        await client.close().catch(() => undefined);
+      },
+    };
+    return { state: { client, tools }, lease };
+  }
+
   /** Mark an entry evicted; close its clients now if nothing is leasing them. */
   private evict(entry: CacheEntry): void {
     clearTimeout(entry.timer);
@@ -554,22 +915,21 @@ export function validateResolvedAddresses(addrs: readonly LookupAddress[]): {
  * certificate validation still uses the real hostname (we never rewrite the URL
  * to an IP literal).
  */
-function buildPinnedDispatcher(): Agent {
-  // External-MCP traffic uses a DEDICATED, shorter silence timeout
+function buildPinnedDispatcher(bodyTimeoutMs: number): Agent {
+  // External-MCP traffic uses a DEDICATED, shorter HEADERS silence timeout
   // (`AI_MCP_STREAM_TIMEOUT_MS`, default 1 min) — deliberately tighter than the
   // chat provider's 15-min `streamTimeoutMs()` — so a byte-silent/hung MCP
   // upstream is broken in ~1 min instead of 15. We keep the keep-alive options
-  // from `streamingDispatcherOptions()` but OVERRIDE headers/body timeouts.
-  // Accepted trade-off: a legitimately long but byte-silent single tool call,
-  // and an SSE transport idling >1 min BETWEEN tool calls, are also cut here; the
-  // per-call total cap (wrapToolsWithCallTimeout, `AI_MCP_CALL_TIMEOUT_MS`) is the
-  // complementary guard for chatty-but-stuck calls that keep the socket warm yet
-  // never return.
-  const mcpSilenceMs = mcpStreamTimeoutMs();
+  // from `streamingDispatcherOptions()` but OVERRIDE the timeouts. `bodyTimeout`
+  // is passed in per-transport (#489): tight for HTTP (fresh request per call),
+  // raised for SSE (one long-lived body across calls — idle BETWEEN calls is
+  // legit). The per-call total cap (`AI_MCP_CALL_TIMEOUT_MS`) is the complementary
+  // guard for chatty-but-stuck calls that keep the socket warm yet never return.
+  const headersMs = mcpStreamTimeoutMs();
   return new Agent({
     ...streamingDispatcherOptions(),
-    headersTimeout: mcpSilenceMs,
-    bodyTimeout: mcpSilenceMs,
+    headersTimeout: headersMs,
+    bodyTimeout: bodyTimeoutMs,
     connect: {
       lookup: (hostname, _options, callback) => {
         // Always resolve ALL addresses ourselves; do not trust the caller's
@@ -669,18 +1029,22 @@ function pick(
 function namespace(
   tools: Record<string, Tool>,
   serverName: string,
-): Record<string, Tool> {
+): Array<{ full: string; raw: string; tool: Tool }> {
   const prefix = namespacePrefix(serverName);
-  const out: Record<string, Tool> = {};
+  const out: Array<{ full: string; raw: string; tool: Tool }> = [];
+  const taken: Record<string, true> = {};
   for (const [name, t] of Object.entries(tools)) {
     const safe = sanitizeName(name);
     let full = capName(`${prefix}_${safe}`);
     // Duplicate names within ONE server can still collide after sanitize/
     // truncate — suffix-disambiguate so the second tool is not overwritten.
-    if (full in out) {
-      full = disambiguate(full, '', (candidate) => candidate in out);
+    if (full in taken) {
+      full = disambiguate(full, '', (candidate) => candidate in taken);
     }
-    out[full] = t;
+    taken[full] = true;
+    // Keep the RAW (un-namespaced) name alongside the merged key so the per-run
+    // recovery wrapper (#489) can re-resolve the same tool on a fresh client.
+    out.push({ full, raw: name, tool: t });
   }
   return out;
 }
@@ -804,6 +1168,69 @@ export function wrapToolWithCallTimeout(tool: Tool, ms: number): Tool {
   return { ...tool, execute } as unknown as Tool;
 }
 
+/**
+ * undici / Node network error CODES that mean the connection broke (not an
+ * application-level error) — a transient transport failure a readOnly call may
+ * safely retry after reconnecting. Matched against the REAL error shapes (#489):
+ * a socket reset surfaces as `TypeError: fetch failed` whose `.cause` is an
+ * undici `SocketError { code:'UND_ERR_SOCKET' }`; a body-timeout as
+ * `TypeError: terminated` whose `.cause` is `BodyTimeoutError`. Classifying by
+ * these real codes/names (not by mock errors) is essential — a mock-shaped
+ * predicate would leave eviction silently dead in production while CI is green.
+ */
+const RETRYABLE_TRANSPORT_ERROR_CODES: ReadonlySet<string> = new Set([
+  'ECONNRESET',
+  'ECONNREFUSED',
+  'ECONNABORTED',
+  'EPIPE',
+  'ETIMEDOUT',
+  'EAI_AGAIN',
+  'ENETUNREACH',
+  'EHOSTUNREACH',
+  'UND_ERR_SOCKET',
+  'UND_ERR_CONNECT_TIMEOUT',
+  'UND_ERR_HEADERS_TIMEOUT',
+  'UND_ERR_BODY_TIMEOUT',
+  'UND_ERR_CLOSED',
+  'UND_ERR_DESTROYED',
+]);
+
+/** undici error CLASS names for the same transport-break conditions. */
+const RETRYABLE_TRANSPORT_ERROR_NAMES: ReadonlySet<string> = new Set([
+  'SocketError',
+  'BodyTimeoutError',
+  'HeadersTimeoutError',
+  'ConnectTimeoutError',
+  'ClientClosedError',
+  'ClientDestroyedError',
+]);
+
+/**
+ * Whether `err` is a retryable TRANSPORT break (a broken socket / body timeout),
+ * classified by the REAL undici/@ai-sdk error shapes (#489). undici surfaces a
+ * reset as `TypeError('fetch failed'|'terminated')` with the real error in
+ * `.cause`, and @ai-sdk/mcp may wrap it again in an `MCPClientError` (cause
+ * chain), so we walk `.cause` (bounded depth) checking `.code` and `.name`. An
+ * app-level tool error (a 4xx, a validation failure) is NOT retryable and returns
+ * false — only a connection-level failure heals with a reconnect.
+ */
+export function isRetryableConnectError(err: unknown, depth = 0): boolean {
+  if (!err || typeof err !== 'object' || depth > 6) return false;
+  const e = err as {
+    code?: unknown;
+    name?: unknown;
+    cause?: unknown;
+  };
+  if (typeof e.code === 'string' && RETRYABLE_TRANSPORT_ERROR_CODES.has(e.code)) {
+    return true;
+  }
+  if (typeof e.name === 'string' && RETRYABLE_TRANSPORT_ERROR_NAMES.has(e.name)) {
+    return true;
+  }
+  if (e.cause != null) return isRetryableConnectError(e.cause, depth + 1);
+  return false;
+}
+
 /** The signal's reason as an Error (informative thrown value on abort/timeout). */
 function abortReason(signal: AbortSignal): Error {
   const r = signal.reason;

apps/server/src/core/comment/comment.controller.ts

@@ -16,6 +16,7 @@ import { UpdateCommentDto } from './dto/update-comment.dto';
 import { ResolveCommentDto } from './dto/resolve-comment.dto';
 import { ApplySuggestionDto } from './dto/apply-suggestion.dto';
 import { DismissSuggestionDto } from './dto/dismiss-suggestion.dto';
+import { ResyncSuggestionAnchorDto } from './dto/resync-suggestion-anchor.dto';
 import { PageIdDto, CommentIdDto } from './dto/comments.input';
 import { AuthUser } from '../../common/decorators/auth-user.decorator';
 import { AuthWorkspace } from '../../common/decorators/auth-workspace.decorator';
@@ -235,6 +236,39 @@ export class CommentController {
     return this.commentService.applySuggestion(comment, user, provenance);
   }
 
+  @HttpCode(HttpStatus.OK)
+  @Post('resync-suggestion-anchor')
+  async resyncSuggestionAnchor(
+    @Body() dto: ResyncSuggestionAnchorDto,
+    @AuthUser() user: User,
+    @AuthWorkspace() workspace: Workspace,
+  ) {
+    const comment = await this.commentRepo.findById(dto.commentId, {
+      includeCreator: true,
+      includeResolvedBy: true,
+    });
+    if (!comment) {
+      throw new NotFoundException('Comment not found');
+    }
+
+    const page = await this.pageRepo.findById(comment.pageId);
+    if (!page || page.deletedAt) {
+      throw new NotFoundException('Page not found');
+    }
+
+    // Authorize BEFORE revealing structural detail (mirrors apply/dismiss).
+    // Re-anchoring does NOT change the page text — it only corrects the stored
+    // selection metadata — so the page-level gate is comment access. The service
+    // further restricts it to the suggestion's own author.
+    await this.pageAccessService.validateCanComment(page, user, workspace.id);
+
+    return this.commentService.resyncSuggestionAnchor(
+      comment,
+      dto.selection,
+      user,
+    );
+  }
+
   @HttpCode(HttpStatus.OK)
   @Post('dismiss-suggestion')
   async dismissSuggestion(

apps/server/src/core/comment/comment.service.apply-suggestion.spec.ts

@@ -146,11 +146,19 @@ describe('CommentService — applySuggestion', () => {
       'page-1',
       expect.objectContaining({ operation: 'commentDeleted', commentId: 'c-1' }),
     );
+    // #496: hard-deleted row → the audit payload is the only surviving record.
     expect(auditService.log).toHaveBeenCalledWith(
       expect.objectContaining({
         event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
         resourceType: AuditResource.COMMENT,
         resourceId: 'c-1',
+        metadata: expect.objectContaining({
+          pageId: 'page-1',
+          suggestedText: 'new text',
+          selection: 'old text',
+          commentAuthor: 'user-1',
+          decidedBy: 'user-1',
+        }),
       }),
     );
     expect(result.outcome).toBe('deleted');
@@ -189,17 +197,25 @@ describe('CommentService — applySuggestion', () => {
     expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
     expect(resolvePatch.resolvedById).toBe('user-1');
 
-    // NOT deleted; broadcast an update, not a deletion.
+    // NOT deleted.
     expect(commentRepo.deleteComment).not.toHaveBeenCalled();
     expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith(
       'deleteCommentMark',
       expect.anything(),
       expect.anything(),
     );
+    // #496 dedup: resolveComment broadcasts `commentResolved` with the enriched
+    // row; finalize must NOT ALSO emit a redundant `commentUpdated`. So the
+    // thread receives exactly ONE resolve broadcast and no update broadcast.
     expect(wsService.emitCommentEvent).toHaveBeenCalledWith(
       'space-1',
       'page-1',
-      expect.objectContaining({ operation: 'commentUpdated', comment: UPDATED }),
+      expect.objectContaining({ operation: 'commentResolved', comment: UPDATED }),
+    );
+    expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith(
+      'space-1',
+      'page-1',
+      expect.objectContaining({ operation: 'commentUpdated' }),
     );
 
     expect(auditService.log).toHaveBeenCalledWith(
@@ -211,6 +227,36 @@ describe('CommentService — applySuggestion', () => {
     expect(result.outcome).toBe('resolved');
   });
 
+  it('re-entry: already applied+resolved WITH replies → emits commentUpdated (dedup does not over-suppress)', async () => {
+    // suggestionAppliedAt set → idempotent finalize; resolvedAt set → resolveComment
+    // is skipped, so there is NO commentResolved broadcast. The applied-stamp state
+    // must still reach clients via a single commentUpdated.
+    const { service, wsService } = makeService(
+      { applied: false, currentText: 'new text' },
+      true,
+    );
+
+    await service.applySuggestion(
+      suggestionComment({
+        suggestionAppliedAt: new Date(),
+        resolvedAt: new Date(),
+      }),
+      user(),
+    );
+
+    expect(wsService.emitCommentEvent).toHaveBeenCalledWith(
+      'space-1',
+      'page-1',
+      expect.objectContaining({ operation: 'commentUpdated', comment: UPDATED }),
+    );
+    // Nothing resolved this time (already resolved) → no resolve broadcast.
+    expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith(
+      'space-1',
+      'page-1',
+      expect.objectContaining({ operation: 'commentResolved' }),
+    );
+  });
+
   // --- error / rejection branches -----------------------------------------
 
   it('applied=false and currentText differs → ConflictException with currentText in payload', async () => {

apps/server/src/core/comment/comment.service.dismiss-suggestion.spec.ts

@@ -107,11 +107,21 @@ describe('CommentService — dismissSuggestion', () => {
       'page-1',
       expect.objectContaining({ operation: 'commentDeleted', commentId: 'c-1' }),
     );
+    // #496: the row is hard-deleted, so the audit payload must carry the
+    // decision's substance (what was suggested, the anchored text, who authored
+    // it, who decided) — it is the only surviving record.
     expect(auditService.log).toHaveBeenCalledWith(
       expect.objectContaining({
         event: AuditEvent.COMMENT_SUGGESTION_DISMISSED,
         resourceType: AuditResource.COMMENT,
         resourceId: 'c-1',
+        metadata: expect.objectContaining({
+          pageId: 'page-1',
+          suggestedText: 'new text',
+          selection: 'old text',
+          commentAuthor: 'user-1',
+          decidedBy: 'user-1',
+        }),
       }),
     );
     expect(result.outcome).toBe('deleted');

apps/server/src/core/comment/comment.service.resync-anchor.spec.ts

@@ -0,0 +1,126 @@
+import { BadRequestException, ForbiddenException } from '@nestjs/common';
+import { CommentService } from './comment.service';
+
+/**
+ * Coverage for CommentService.resyncSuggestionAnchor (#496): re-anchoring a
+ * suggestion's stored selection (== apply-time expectedText) to the live-doc
+ * substring. The service is built directly with jest-mocked deps (the
+ * @InjectQueue tokens can't be resolved by Test.createTestingModule — see the
+ * sibling specs).
+ */
+describe('CommentService — resyncSuggestionAnchor', () => {
+  const UPDATED = { id: 'c-1', selection: 'new anchor', __updated: true } as any;
+
+  function makeService() {
+    const commentRepo: any = {
+      updateComment: jest.fn(async () => undefined),
+      findById: jest.fn(async () => UPDATED),
+    };
+    const service = new CommentService(
+      commentRepo,
+      {} as any,
+      { emitCommentEvent: jest.fn() } as any,
+      {} as any,
+      { add: jest.fn() } as any,
+      { add: jest.fn() } as any,
+      { log: jest.fn() } as any,
+    );
+    return { service, commentRepo };
+  }
+
+  const suggestion = (over?: Partial<any>): any => ({
+    id: 'c-1',
+    creatorId: 'user-1',
+    parentCommentId: null,
+    selection: 'old anchor',
+    suggestedText: 'new text',
+    suggestionAppliedAt: null,
+    resolvedAt: null,
+    ...over,
+  });
+  const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });
+
+  it('persists the new selection and returns the enriched comment', async () => {
+    const { service, commentRepo } = makeService();
+
+    const out = await service.resyncSuggestionAnchor(
+      suggestion(),
+      'new anchor',
+      user(),
+    );
+
+    expect(commentRepo.updateComment).toHaveBeenCalledWith(
+      { selection: 'new anchor' },
+      'c-1',
+    );
+    expect(out).toBe(UPDATED);
+  });
+
+  it('is idempotent: no write when the anchor already matches', async () => {
+    const { service, commentRepo } = makeService();
+
+    const out = await service.resyncSuggestionAnchor(
+      suggestion({ selection: 'same' }),
+      'same',
+      user(),
+    );
+
+    expect(commentRepo.updateComment).not.toHaveBeenCalled();
+    expect(out).toEqual(suggestion({ selection: 'same' }));
+  });
+
+  it('rejects a non-author (only the suggestion owner may re-anchor)', async () => {
+    const { service, commentRepo } = makeService();
+    await expect(
+      service.resyncSuggestionAnchor(suggestion(), 'new anchor', user({ id: 'other' })),
+    ).rejects.toBeInstanceOf(ForbiddenException);
+    expect(commentRepo.updateComment).not.toHaveBeenCalled();
+  });
+
+  it('rejects a reply / a comment with no suggestion', async () => {
+    const { service } = makeService();
+    await expect(
+      service.resyncSuggestionAnchor(
+        suggestion({ parentCommentId: 'p-1' }),
+        'x',
+        user(),
+      ),
+    ).rejects.toBeInstanceOf(BadRequestException);
+    await expect(
+      service.resyncSuggestionAnchor(
+        suggestion({ suggestedText: null }),
+        'x',
+        user(),
+      ),
+    ).rejects.toBeInstanceOf(BadRequestException);
+  });
+
+  it('rejects re-anchoring an already applied or resolved suggestion', async () => {
+    const { service } = makeService();
+    await expect(
+      service.resyncSuggestionAnchor(
+        suggestion({ suggestionAppliedAt: new Date() }),
+        'x',
+        user(),
+      ),
+    ).rejects.toBeInstanceOf(BadRequestException);
+    await expect(
+      service.resyncSuggestionAnchor(
+        suggestion({ resolvedAt: new Date() }),
+        'x',
+        user(),
+      ),
+    ).rejects.toBeInstanceOf(BadRequestException);
+  });
+
+  it('rejects a no-op selection equal to the suggested text', async () => {
+    const { service } = makeService();
+    await expect(
+      service.resyncSuggestionAnchor(
+        suggestion({ suggestedText: 'new text' }),
+        'new text',
+        user(),
+      ),
+    ).rejects.toBeInstanceOf(BadRequestException);
+  });
+});

apps/server/src/core/comment/comment.service.ts

@@ -370,6 +370,76 @@ export class CommentService {
     return updatedComment;
   }
 
+  /**
+   * Re-sync a suggestion's stored `selection` (== apply-time expectedText) to the
+   * RAW substring the inline mark actually covers in the LIVE document (#496).
+   *
+   * The MCP client creates the comment from a DEBOUNCED REST snapshot, then
+   * anchors the mark in the live collab doc. When the two disagree (the doc moved
+   * on in the debounce window) the stored selection no longer equals the marked
+   * text, so EVERY apply 409s ("the commented text changed"). After anchoring the
+   * client re-reads the exact marked substring and calls this to store it, making
+   * apply's strict equality hold.
+   *
+   * Only meaningful for an un-settled top-level suggestion authored by the
+   * caller: applying/resolving freezes the anchor, and a reply-carrying thread is
+   * preserved rather than mutated. The new text must still differ from the
+   * suggestion (else "apply" would be a no-op), preserving create()'s invariant.
+   */
+  async resyncSuggestionAnchor(
+    comment: Comment,
+    selection: string,
+    user: User,
+  ): Promise<Comment> {
+    if (comment.creatorId !== user.id) {
+      throw new ForbiddenException(
+        'You can only re-anchor your own suggestion',
+      );
+    }
+    if (comment.parentCommentId) {
+      throw new BadRequestException(
+        'Only a top-level comment can carry a suggested edit',
+      );
+    }
+    if (!comment.suggestedText) {
+      throw new BadRequestException('This comment has no suggested edit');
+    }
+    // A settled suggestion's anchor is frozen: re-anchoring an applied/resolved
+    // thread is meaningless and could resurrect a stale expectedText.
+    if (comment.suggestionAppliedAt || comment.resolvedAt) {
+      throw new BadRequestException(
+        'Cannot re-anchor a suggestion that was already applied or resolved',
+      );
+    }
+    const trimmed = selection.trim();
+    if (trimmed.length === 0) {
+      throw new BadRequestException('The re-anchored selection cannot be empty');
+    }
+    // Same no-op guard as create(): the suggestion must differ from the text it
+    // replaces, or apply becomes indistinguishable from already-applied.
+    if (trimmed === comment.suggestedText.trim()) {
+      throw new BadRequestException(
+        'A suggested edit must differ from the selected text',
+      );
+    }
+
+    // Idempotent: nothing to persist when the anchor already matches.
+    if (comment.selection === selection) {
+      return comment;
+    }
+
+    await this.commentRepo.updateComment({ selection }, comment.id);
+
+    const updatedComment = await this.commentRepo.findById(comment.id, {
+      includeCreator: true,
+      includeResolvedBy: true,
+    });
+
+    // Re-anchoring only corrects stored metadata; it does not change the page
+    // text or the comment body, so no ws broadcast / notification is warranted.
+    return updatedComment;
+  }
+
   /**
    * Apply the suggested edit carried by a top-level inline comment: atomically
    * replace the text under the comment mark in the collaborative document with
@@ -524,7 +594,7 @@ export class CommentService {
         resourceType: AuditResource.COMMENT,
         resourceId: comment.id,
         spaceId: comment.spaceId,
-        metadata: { pageId: comment.pageId },
+        metadata: this.suggestionAuditMetadata(comment, user),
       });
       return { ...updatedComment, outcome: 'resolved' };
     }
@@ -538,7 +608,7 @@ export class CommentService {
       resourceType: AuditResource.COMMENT,
       resourceId: comment.id,
       spaceId: comment.spaceId,
-      metadata: { pageId: comment.pageId },
+      metadata: this.suggestionAuditMetadata(comment, user),
     });
     return settled;
   }
@@ -577,8 +647,10 @@ export class CommentService {
 
       // Auto-resolve the thread. resolveComment handles the resolve mark, its ws
       // broadcast and the resolve notification. Stay defensive on re-entry.
+      let didResolveBroadcast = false;
       if (!comment.resolvedAt) {
         await this.resolveComment(comment, true, user, provenance);
+        didResolveBroadcast = true;
       }
 
       const updatedComment = await this.commentRepo.findById(comment.id, {
@@ -586,18 +658,27 @@ export class CommentService {
         includeResolvedBy: true,
       });
 
-      this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
-        operation: 'commentUpdated',
-        pageId: comment.pageId,
-        comment: updatedComment,
-      });
+      // #496 dedup: resolveComment already broadcast `commentResolved` carrying
+      // the fully-enriched row (the applied stamps were persisted above, before
+      // that call, so its re-read reflects them). Emitting `commentUpdated` here
+      // too made the client receive TWO events for one apply. Broadcast the
+      // update ONLY when we did NOT resolve — i.e. the rare re-entry on an
+      // already-resolved thread, where the applied-stamp change still needs a
+      // broadcast and resolveComment did not run.
+      if (!didResolveBroadcast) {
+        this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
+          operation: 'commentUpdated',
+          pageId: comment.pageId,
+          comment: updatedComment,
+        });
+      }
 
       this.auditService.log({
         event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
         resourceType: AuditResource.COMMENT,
         resourceId: comment.id,
         spaceId: comment.spaceId,
-        metadata: { pageId: comment.pageId },
+        metadata: this.suggestionAuditMetadata(comment, user),
       });
 
       return { ...updatedComment, outcome: 'resolved' };
@@ -616,7 +697,7 @@ export class CommentService {
       resourceType: AuditResource.COMMENT,
       resourceId: comment.id,
       spaceId: comment.spaceId,
-      metadata: { pageId: comment.pageId },
+      metadata: this.suggestionAuditMetadata(comment, user),
     });
 
     return settled;
@@ -627,14 +708,17 @@ export class CommentService {
    * inline `comment` anchor mark, then ATOMICALLY hard-delete the row only if it
    * is still childless. Shared by the apply/dismiss no-replies branches (#329).
    *
-   * ORDER MATTERS: the anchor mark is removed FIRST and FATALLY (mirrors
-   * applySuggestion, which mutates the doc before writing the DB). The row
-   * delete is irreversible, so if the mark removal fails — including the
-   * COLLAB_DISABLE_REDIS "no live instance" hard-error — we must NOT delete the
-   * row and report success, or the document is left with a permanent orphan
-   * anchor pointing at a comment that no longer exists (the exact data-integrity
-   * bug #329 targets). Let the exception propagate (→ 5xx); the operation is
-   * then repeatable with row + mark still consistent.
+   * ORDER MATTERS (updated #399 → #496): what runs FIRST and FATALLY here is the
+   * mark-removal ENQUEUE (a fast, durable Redis add), NOT the mark op itself.
+   * deleteCommentMark awaits only the enqueue, so a failed add throws BEFORE the
+   * irreversible row delete — the row + mark stay consistent and the operation is
+   * repeatable. The actual anchor strip then runs off the HTTP path in the worker
+   * (idempotent, 3 retries). Only an EXHAUSTED-retries job could leave the doc
+   * with an orphan anchor pointing at a hard-deleted comment (the data-integrity
+   * bug #329 targets); that residual divergence is now self-healed by the
+   * resolve/unresolve mark worker, which strips an orphan mark whenever its
+   * comment row is gone (#496), and it is meanwhile VISIBLE via BullMQ failed-job
+   * metrics rather than a silently-swallowed warn.
    *
    * RACE (#338 F4): the caller read `hasChildren` BEFORE the (slow) mark
    * removal, so a reply can land in that window. `comments.parent_comment_id` is
@@ -732,6 +816,27 @@ export class CommentService {
     return this.generalQueue.add(QueueJob.COMMENT_MARK_UPDATE, jobData);
   }
 
+  /**
+   * Build the audit metadata for a suggestion apply/dismiss decision (#496).
+   * The subject comment is HARD-DELETED on the childless path, so the audit row
+   * is the only surviving record — capture the decision's substance (what was
+   * suggested, the anchored text it replaced, who authored it, who decided)
+   * before the row can vanish. `decidedBy` is the acting user; `commentAuthor`
+   * is the suggestion's creator.
+   */
+  private suggestionAuditMetadata(
+    comment: Comment,
+    user: User,
+  ): Record<string, any> {
+    return {
+      pageId: comment.pageId,
+      suggestedText: comment.suggestedText ?? null,
+      selection: comment.selection ?? null,
+      commentAuthor: comment.creatorId ?? null,
+      decidedBy: user.id,
+    };
+  }
+
   private async queueCommentNotification(
     content: any,
     oldMentionIds: string[],

apps/server/src/core/comment/dto/resync-suggestion-anchor.dto.ts

@@ -0,0 +1,20 @@
+import { IsString, IsUUID, MaxLength, MinLength } from 'class-validator';
+
+/**
+ * #496: after the MCP client anchors a suggestion in the LIVE collab doc, it
+ * re-reads the exact substring under the new mark and syncs it here as the
+ * comment's stored `selection` (== apply-time expectedText). Fixes the perpetual
+ * 409 where expectedText came from a debounced REST snapshot while the mark sat
+ * in the live doc.
+ */
+export class ResyncSuggestionAnchorDto {
+  @IsUUID()
+  commentId: string;
+
+  // The raw substring the mark now covers in the live document. Bounded like the
+  // create-time selection (2000) so a legitimate anchored span is never cut.
+  @IsString()
+  @MinLength(1)
+  @MaxLength(2000)
+  selection: string;
+}

apps/server/src/integrations/ai/ai-streaming-fetch.ts

@@ -129,6 +129,12 @@ const DEFAULT_MCP_STREAM_TIMEOUT_MS = 60_000;
 /** Default total wall-clock cap for ONE external MCP tool call (2 min). */
 const DEFAULT_MCP_CALL_TIMEOUT_MS = 120_000;
 
+/**
+ * Default `bodyTimeout` for the EXTERNAL-MCP SSE transport (10 min) — #489.
+ * Deliberately much LARGER than {@link DEFAULT_MCP_STREAM_TIMEOUT_MS}.
+ */
+const DEFAULT_MCP_SSE_BODY_TIMEOUT_MS = 600_000;
+
 /**
  * SILENCE timeout (ms) for EXTERNAL-MCP transport ONLY. Override with
  * `AI_MCP_STREAM_TIMEOUT_MS`; a missing/invalid/non-positive value falls back to
@@ -164,6 +170,26 @@ export function mcpCallTimeoutMs(): number {
   return positiveEnv('AI_MCP_CALL_TIMEOUT_MS', DEFAULT_MCP_CALL_TIMEOUT_MS);
 }
 
+/**
+ * `bodyTimeout` (ms) for the EXTERNAL-MCP **SSE** transport ONLY — #489. Override
+ * with `AI_MCP_SSE_BODY_TIMEOUT_MS`; a missing/invalid/non-positive value falls
+ * back to {@link DEFAULT_MCP_SSE_BODY_TIMEOUT_MS} (10 min).
+ *
+ * The SSE transport holds ONE long-lived response body open across many tool
+ * calls, so undici's `bodyTimeout` (time between body bytes) counts the LEGITIMATE
+ * silence BETWEEN calls, not just a hung single call. At the tight HTTP silence
+ * timeout ({@link mcpStreamTimeoutMs}, 1 min) a normal >1-min gap between the
+ * model's tool calls would break the SSE socket, and the cache would then serve a
+ * dead client until TTL. So the SSE transport gets its OWN, RAISED bodyTimeout;
+ * the per-call total cap ({@link mcpCallTimeoutMs}) still bounds a single stuck
+ * call, and the app-level transport-error retry heals a socket that does break.
+ * The HTTP (streamable) transport keeps the tight timeout — it opens a fresh
+ * request per call, so idle-between-calls does not apply there.
+ */
+export function mcpSseBodyTimeoutMs(): number {
+  return positiveEnv('AI_MCP_SSE_BODY_TIMEOUT_MS', DEFAULT_MCP_SSE_BODY_TIMEOUT_MS);
+}
+
 /**
  * undici `Agent` options for streaming AI traffic — the (generous, finite)
  * silence timeouts plus the keep-alive recycle window. Shared by the chat

apps/server/src/integrations/audit/audit.module.ts

@@ -1,14 +1,19 @@
 import { Global, Module } from '@nestjs/common';
-import { AUDIT_SERVICE, NoopAuditService } from './audit.service';
+import { AUDIT_SERVICE } from './audit.service';
+import { DatabaseAuditService } from './database-audit.service';
 
+// #496: bind the audit token to a real DB-backed trail (was NoopAuditService,
+// which silently dropped every event). Kysely (@Global DatabaseModule) and
+// ClsService (@Global ClsModule) are both globally available, so this module
+// needs no extra imports.
 @Global()
 @Module({
   providers: [
     {
       provide: AUDIT_SERVICE,
-      useClass: NoopAuditService,
+      useClass: DatabaseAuditService,
     },
   ],
   exports: [AUDIT_SERVICE],
 })
-export class NoopAuditModule {}
+export class AuditModule {}

apps/server/src/integrations/audit/database-audit.service.spec.ts

@@ -0,0 +1,198 @@
+import { Logger } from '@nestjs/common';
+import { DatabaseAuditService } from './database-audit.service';
+import { AUDIT_CONTEXT_KEY } from '../../common/middlewares/audit-context.middleware';
+import { AuditEvent, AuditResource } from '../../common/events/audit-events';
+
+/**
+ * Observable-property coverage for the DB-backed audit trail (#496): every
+ * assertion pins what actually reaches the `audit` table (or that nothing does),
+ * driven through a chainable Kysely mock that captures the inserted rows.
+ */
+describe('DatabaseAuditService', () => {
+  function makeService(clsContext: any) {
+    const inserted: any[] = [];
+    const updated: any[] = [];
+    let failNextInsert = false;
+
+    const db: any = {
+      insertInto: jest.fn(() => ({
+        values: jest.fn((rows: any) => ({
+          execute: jest.fn(async () => {
+            if (failNextInsert) {
+              failNextInsert = false;
+              throw new Error('boom');
+            }
+            inserted.push(rows);
+          }),
+        })),
+      })),
+      updateTable: jest.fn(() => ({
+        set: jest.fn((patch: any) => ({
+          where: jest.fn(() => ({
+            execute: jest.fn(async () => {
+              updated.push(patch);
+            }),
+          })),
+        })),
+      })),
+    };
+
+    const store: Record<string, any> = { [AUDIT_CONTEXT_KEY]: clsContext };
+    const cls: any = {
+      get: jest.fn((key: string) => store[key]),
+      set: jest.fn((key: string, val: any) => {
+        store[key] = val;
+      }),
+    };
+
+    const service = new DatabaseAuditService(db, cls);
+    return {
+      service,
+      inserted,
+      updated,
+      cls,
+      store,
+      failInsert: () => {
+        failNextInsert = true;
+      },
+    };
+  }
+
+  const applyPayload = () => ({
+    event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
+    resourceType: AuditResource.COMMENT,
+    resourceId: 'c-1',
+    spaceId: 'space-1',
+    metadata: { pageId: 'p-1', suggestedText: 'new', decidedBy: 'u-2' },
+  });
+
+  const ctx = (over?: any) => ({
+    workspaceId: 'ws-1',
+    actorId: 'u-2',
+    actorType: 'user',
+    ipAddress: '10.0.0.1',
+    userAgent: 'jest',
+    ...over,
+  });
+
+  it('log() persists a row with the CLS context merged onto the payload', async () => {
+    const { service, inserted } = makeService(ctx());
+    service.log(applyPayload());
+    // log() is fire-and-forget; flush the microtask queue.
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(inserted).toHaveLength(1);
+    expect(inserted[0]).toMatchObject({
+      workspaceId: 'ws-1',
+      actorId: 'u-2',
+      actorType: 'user',
+      event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
+      resourceType: AuditResource.COMMENT,
+      resourceId: 'c-1',
+      spaceId: 'space-1',
+      ipAddress: '10.0.0.1',
+      metadata: { pageId: 'p-1', suggestedText: 'new', decidedBy: 'u-2' },
+    });
+  });
+
+  it('log() is a no-op when there is no workspace in scope', async () => {
+    const { service, inserted } = makeService(ctx({ workspaceId: null }));
+    service.log(applyPayload());
+    await Promise.resolve();
+    expect(inserted).toHaveLength(0);
+  });
+
+  it('log() drops EXCLUDED_AUDIT_EVENTS (e.g. comment.created)', async () => {
+    const { service, inserted } = makeService(ctx());
+    service.log({
+      event: AuditEvent.COMMENT_CREATED,
+      resourceType: AuditResource.COMMENT,
+      resourceId: 'c-1',
+      spaceId: 'space-1',
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(inserted).toHaveLength(0);
+  });
+
+  it('a failed insert is swallowed with a warn and floats no rejection (audit is a side-record)', async () => {
+    // This pins the load-bearing swallow inside persist(). Because log() is
+    // fire-and-forget (`void this.persist(...)`), it always returns synchronously
+    // without throwing — so `not.toThrow()` alone would stay green even if the
+    // try/catch were removed. We instead observe the two effects the catch is
+    // responsible for: a warn IS emitted, and NO unhandled rejection floats.
+    // Removing persist()'s try/catch reddens both assertions (warn count 0 + a
+    // captured rejection).
+    const warnSpy = jest
+      .spyOn(Logger.prototype, 'warn')
+      .mockImplementation(() => undefined as any);
+    const rejections: unknown[] = [];
+    const onRejection = (err: unknown) => rejections.push(err);
+    process.on('unhandledRejection', onRejection);
+    try {
+      const { service, failInsert } = makeService(ctx());
+      failInsert();
+      expect(() => service.log(applyPayload())).not.toThrow();
+      // Flush microtasks so the rejected insert settles, then give any floated
+      // rejection a macrotask tick to be reported by the runtime.
+      await Promise.resolve();
+      await Promise.resolve();
+      await new Promise((r) => setImmediate(r));
+
+      expect(warnSpy).toHaveBeenCalledTimes(1);
+      expect(String(warnSpy.mock.calls[0][0])).toContain(
+        'Failed to persist audit event',
+      );
+      expect(rejections).toHaveLength(0);
+    } finally {
+      process.off('unhandledRejection', onRejection);
+      warnSpy.mockRestore();
+    }
+  });
+
+  it('logWithContext() persists with an explicit (non-request) context', async () => {
+    const { service, inserted } = makeService(undefined);
+    service.logWithContext(applyPayload(), ctx({ actorType: 'system' }) as any);
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(inserted).toHaveLength(1);
+    expect(inserted[0].actorType).toBe('system');
+    expect(inserted[0].workspaceId).toBe('ws-1');
+  });
+
+  it('logBatchWithContext() inserts only non-excluded events', async () => {
+    const { service, inserted } = makeService(undefined);
+    service.logBatchWithContext(
+      [
+        applyPayload(),
+        {
+          event: AuditEvent.COMMENT_CREATED,
+          resourceType: AuditResource.COMMENT,
+          resourceId: 'c-2',
+        },
+      ],
+      ctx() as any,
+    );
+    await Promise.resolve();
+    await Promise.resolve();
+    // Batch is a single insert call carrying only the applied event.
+    expect(inserted).toHaveLength(1);
+    expect(inserted[0]).toHaveLength(1);
+    expect(inserted[0][0].event).toBe(AuditEvent.COMMENT_SUGGESTION_APPLIED);
+  });
+
+  it('setActorId / setActorType mutate the ambient CLS context', () => {
+    const { service, store } = makeService(ctx({ actorId: null }));
+    service.setActorId('u-9');
+    service.setActorType('api_key');
+    expect(store[AUDIT_CONTEXT_KEY].actorId).toBe('u-9');
+    expect(store[AUDIT_CONTEXT_KEY].actorType).toBe('api_key');
+  });
+
+  it('updateRetention() writes the workspace retention window', async () => {
+    const { service, updated } = makeService(ctx());
+    await service.updateRetention('ws-1', 30);
+    expect(updated).toEqual([{ auditRetentionDays: 30 }]);
+  });
+});

apps/server/src/integrations/audit/database-audit.service.ts

@@ -0,0 +1,160 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectKysely } from 'nestjs-kysely';
+import { ClsService } from 'nestjs-cls';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import {
+  AuditContext,
+  AUDIT_CONTEXT_KEY,
+} from '../../common/middlewares/audit-context.middleware';
+import {
+  AuditLogPayload,
+  ActorType,
+  EXCLUDED_AUDIT_EVENTS,
+} from '../../common/events/audit-events';
+import { AuditLogContext, IAuditService } from './audit.service';
+
+/**
+ * Minimal DB-backed audit trail (#496). Replaces NoopAuditService so that
+ * decision-bearing events — notably comment.suggestion_applied /
+ * comment.suggestion_dismissed, whose subject comment is HARD-DELETED on the
+ * childless path — leave a durable record of who decided what. Without this the
+ * events were emitted (comment.service / *.controller) but swallowed, so an
+ * applied/dismissed suggestion was unrecoverable once the row was gone.
+ *
+ * Rows land in the pre-existing `audit` table (migration 20260228T223532). The
+ * per-request actor/workspace/ip come from the CLS AuditContext populated by
+ * AuditContextMiddleware + AuditActorInterceptor; callers that run OUTSIDE a
+ * request (queue workers, imports) pass an explicit context via
+ * logWithContext / logBatchWithContext.
+ *
+ * Audit is a side-record: a write failure MUST NOT break the originating
+ * request, so every persistence path swallows its error with a warn. Events in
+ * EXCLUDED_AUDIT_EVENTS (high-volume/low-signal) are dropped.
+ */
+@Injectable()
+export class DatabaseAuditService implements IAuditService {
+  private readonly logger = new Logger(DatabaseAuditService.name);
+
+  constructor(
+    @InjectKysely() private readonly db: KyselyDB,
+    private readonly cls: ClsService,
+  ) {}
+
+  /**
+   * Persist a single event using the ambient request-scoped AuditContext. A
+   * no-op when there is no workspace in scope (the table's workspace_id is NOT
+   * NULL) or the event is excluded. Fire-and-forget: the returned promise is not
+   * awaited by hot callers, and its rejection is swallowed here.
+   */
+  log(payload: AuditLogPayload): void {
+    const context = this.cls?.get<AuditContext>(AUDIT_CONTEXT_KEY);
+    if (!context?.workspaceId) {
+      // No workspace in scope — nothing we can attribute the row to. This is
+      // expected for events emitted outside an HTTP request; those callers must
+      // use logWithContext instead.
+      return;
+    }
+    void this.persist(payload, {
+      workspaceId: context.workspaceId,
+      actorId: context.actorId ?? undefined,
+      actorType: context.actorType,
+      ipAddress: context.ipAddress ?? undefined,
+      userAgent: context.userAgent ?? undefined,
+    });
+  }
+
+  /** Persist a single event with an explicit (non-request) context. */
+  logWithContext(payload: AuditLogPayload, context: AuditLogContext): void {
+    if (!context?.workspaceId) return;
+    void this.persist(payload, context);
+  }
+
+  /** Persist a batch of events sharing one explicit context (imports). */
+  logBatchWithContext(
+    payloads: AuditLogPayload[],
+    context: AuditLogContext,
+  ): void {
+    if (!context?.workspaceId || payloads.length === 0) return;
+    const rows = payloads
+      .filter((p) => !EXCLUDED_AUDIT_EVENTS.has(p.event))
+      .map((p) => this.toRow(p, context));
+    if (rows.length === 0) return;
+    this.db
+      .insertInto('audit')
+      .values(rows)
+      .execute()
+      .catch((err: any) =>
+        this.logger.warn(`Failed to persist ${rows.length} audit events: ${err?.message}`),
+      );
+  }
+
+  /** Update the ambient request actor (e.g. after login resolves the user). */
+  setActorId(actorId: string): void {
+    const context = this.cls?.get<AuditContext>(AUDIT_CONTEXT_KEY);
+    if (context) {
+      context.actorId = actorId;
+      this.cls.set(AUDIT_CONTEXT_KEY, context);
+    }
+  }
+
+  /** Update the ambient request actor type (user | system | api_key). */
+  setActorType(actorType: ActorType): void {
+    const context = this.cls?.get<AuditContext>(AUDIT_CONTEXT_KEY);
+    if (context) {
+      context.actorType = actorType;
+      this.cls.set(AUDIT_CONTEXT_KEY, context);
+    }
+  }
+
+  /** Persist a workspace's audit-log retention window (days). */
+  async updateRetention(
+    workspaceId: string,
+    retentionDays: number,
+  ): Promise<void> {
+    try {
+      await this.db
+        .updateTable('workspaces')
+        .set({ auditRetentionDays: retentionDays })
+        .where('id', '=', workspaceId)
+        .execute();
+    } catch (err: any) {
+      this.logger.warn(
+        `Failed to update audit retention for workspace ${workspaceId}: ${err?.message}`,
+      );
+    }
+  }
+
+  private async persist(
+    payload: AuditLogPayload,
+    context: AuditLogContext,
+  ): Promise<void> {
+    if (EXCLUDED_AUDIT_EVENTS.has(payload.event)) return;
+    try {
+      await this.db
+        .insertInto('audit')
+        .values(this.toRow(payload, context))
+        .execute();
+    } catch (err: any) {
+      // Audit is a side-record; never let a failed write surface to the caller.
+      this.logger.warn(
+        `Failed to persist audit event ${payload.event}: ${err?.message}`,
+      );
+    }
+  }
+
+  private toRow(payload: AuditLogPayload, context: AuditLogContext) {
+    return {
+      workspaceId: context.workspaceId,
+      actorId: context.actorId ?? null,
+      actorType: context.actorType ?? 'user',
+      event: payload.event,
+      resourceType: payload.resourceType,
+      resourceId: payload.resourceId ?? null,
+      spaceId: payload.spaceId ?? null,
+      // jsonb columns: node-postgres serializes plain objects to JSON.
+      changes: payload.changes ? (payload.changes as any) : null,
+      metadata: payload.metadata ? (payload.metadata as any) : null,
+      ipAddress: context.ipAddress ?? null,
+    };
+  }
+}

apps/server/src/integrations/queue/processors/general-queue.processor.comment-mark.spec.ts

@@ -139,13 +139,19 @@ describe('GeneralQueueProcessor — COMMENT_MARK_UPDATE (#399)', () => {
     );
   });
 
-  it('skips (no throw) when the comment row has vanished', async () => {
+  it('reconcile (#496): comment row vanished → strips the orphan anchor mark', async () => {
     const { proc, collaborationGateway, commentRepo } = makeProc();
     commentRepo.findById.mockResolvedValue(undefined);
 
     await expect(
       proc.process(job({ ...base, action: 'resolve', ts: 1000 })),
     ).resolves.toBeUndefined();
-    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+    // A resolve/unresolve mark job whose comment row is gone leaves a silent
+    // orphan; the worker self-heals by stripping the anchor instead of returning.
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+      'deleteCommentMark',
+      'page.page-1',
+      { commentId: 'c-1', user: { id: 'user-1' } },
+    );
   });
 });

apps/server/src/integrations/queue/processors/general-queue.processor.ts

@@ -95,7 +95,8 @@ export class GeneralQueueProcessor
    * #399: apply a comment's inline-mark mirror in the collab Y.Doc, off the HTTP
    * critical path. Runs the SAME gateway path the synchronous comment.service
    * code used (byte-identical mark op):
-   *   - resolve / unresolve → resolveCommentMark (flip the `resolved` attribute);
+   *   - resolve / unresolve → resolveCommentMark (flip the `resolved` attribute),
+   *     OR strip an orphan anchor when the comment row has vanished (#496);
    *   - delete → deleteCommentMark (strip the ephemeral-suggestion anchor #329).
    * The op is idempotent, so a BullMQ retry is safe. Throwing propagates to
    * WorkerHost → the job is retried and, on exhaustion, surfaces in failed-job
@@ -133,7 +134,18 @@ export class GeneralQueueProcessor
     // of this resolve), skip it rather than flip the mark to a stale state.
     const comment = await this.commentRepo.findById(commentId);
     if (!comment) {
-      // The comment vanished (e.g. hard-deleted) → nothing left to mirror.
+      // #496 reconcile: the comment row is GONE (e.g. an ephemeral apply/dismiss
+      // hard-deleted it while this resolve/unresolve mark job sat in the queue),
+      // but its inline anchor may still live in the doc — a silent orphan mark
+      // pointing at a comment that no longer exists. Self-heal by stripping it
+      // instead of just returning: this closes the divergence the fire-and-forget
+      // resolve/unresolve enqueue (comment.service resolveComment) could leave.
+      // Idempotent — deleteCommentMark on an already-absent mark is a no-op.
+      await this.getCollaborationGateway().handleYjsEvent(
+        'deleteCommentMark',
+        documentName,
+        { commentId, user },
+      );
       return;
     }
     const wantResolved = action === 'resolve';

packages/mcp/src/client/comments.ts

@@ -450,6 +450,12 @@ export function CommentsMixin<TBase extends GConstructor<DocmostClientContext>>(
     // can surface the closest-block / spans-multiple-blocks hint built from the
     // LIVE document (the pre-check page is not in scope there).
     let liveNotFoundError: Error | null = null;
+    // #496: the RAW substring the mark actually covers in the LIVE doc. The
+    // stored selection (payload.selection) came from a DEBOUNCED REST snapshot,
+    // which can differ from the live doc — and apply compares the marked live
+    // text to the stored selection strictly, so a stale snapshot 409s on EVERY
+    // apply. Captured here (same doc version the mark is set in) and synced below.
+    let liveAnchoredSelection: string | null = null;
     try {
       const collabToken = await this.getCollabTokenWithReauth();
       // Open the collab doc by the canonical UUID, never the slugId (#260). The
@@ -489,6 +495,12 @@ export function CommentsMixin<TBase extends GConstructor<DocmostClientContext>>(
           }
           if (applyAnchorInDoc(doc, selection as string, newCommentId)) {
             anchored = true;
+            // For a suggestion, re-read the exact substring now under the mark
+            // (the mark is an attribute, so it does not change the raw text) to
+            // sync as the stored expectedText after the mutation resolves.
+            if (hasSuggestion) {
+              liveAnchoredSelection = getAnchoredText(doc, selection as string);
+            }
             return doc;
           }
           // Selection text not found in the LIVE document: abort the write. The
@@ -527,6 +539,36 @@ export function CommentsMixin<TBase extends GConstructor<DocmostClientContext>>(
       );
     }
 
+    // #496: sync the stored selection (== apply-time expectedText) to the RAW
+    // substring the mark actually covers in the LIVE doc when it diverged from
+    // the debounced REST snapshot we stored at create time. Without this, apply
+    // strictly compares the marked live text to a stale stored selection and
+    // 409s every time. Best-effort: the comment is already correctly anchored, so
+    // a resync failure must NOT roll it back — it only risks a later apply 409,
+    // which we surface as a soft warning.
+    if (
+      hasSuggestion &&
+      liveAnchoredSelection != null &&
+      liveAnchoredSelection !== payload.selection
+    ) {
+      try {
+        await this.client.post("/comments/resync-suggestion-anchor", {
+          commentId: newCommentId,
+          selection: liveAnchoredSelection,
+        });
+        // Reflect the corrected anchor in the returned comment.
+        if (result.data) result.data.selection = liveAnchoredSelection;
+      } catch (e) {
+        if (process.env.DEBUG) {
+          console.error("Failed to resync suggestion anchor:", e);
+        }
+        result.warning =
+          "The suggestion was anchored, but its stored selection could not be " +
+          "synced to the live document; applying it may report a conflict if the " +
+          "text changed. Re-create the suggestion if Apply fails.";
+      }
+    }
+
     // Soft warning (like editPageText): the selection only matched after
     // stripping markdown, so the caller likely quoted a styled fragment.
     if (anchorNormalized) {

packages/mcp/src/index.ts

@@ -30,6 +30,13 @@ export { destroyAllSessions } from "./lib/collab-session.js";
 // internals directly; it goes through loadDocmostMcp()).
 export { SHARED_TOOL_SPECS } from "./tool-specs.js";
 export type { SharedToolSpec } from "./tool-specs.js";
+// #489 — write-class registry consumed by the in-app external-MCP retry gate.
+export {
+  SHARED_TOOL_WRITE_CLASS,
+  isRetryableWriteClass,
+  assertEverySpecDeclaresWriteClass,
+} from "./tool-specs.js";
+export type { ToolWriteClass } from "./tool-specs.js";
 
 // Re-export the build-time REGISTRY_STAMP (issue #447): a deterministic hash of
 // the tool-specs registry content, generated into src/registry-stamp.generated.ts

packages/mcp/src/tool-specs.ts

@@ -127,6 +127,18 @@ export interface SharedToolSpec {
   mcpName: string;
   /** camelCase key in the ai-SDK tools object (the in-app layer). */
   inAppKey: string;
+  /**
+   * Write-class of the tool (#489), declared on EVERY spec (a registration-time
+   * assert enforces completeness; `satisfies Record<string, SharedToolSpec>`
+   * makes it a compile error to omit). 'readOnly' = a pure read that mutates
+   * NOTHING durable, so it is safe to auto-retry once after a transport break.
+   * 'write' = anything that mutates a page/comment/share/diagram/etc — a
+   * transport error is INDETERMINATE (the server may have applied it before the
+   * connection reset), so it is NEVER blind-retried (a retry would double-apply,
+   * the #435 incident class). Consumed by the external-MCP retry path
+   * (mcp-clients.service.ts) to gate its single auto-retry.
+   */
+  writeClass: 'readOnly' | 'write';
   /** Single canonical model-facing description used by both layers. */
   description: string;
   /**
@@ -240,6 +252,7 @@ export const SHARED_TOOL_SPECS = {
   getWorkspace: {
     mcpName: 'getWorkspace',
     inAppKey: 'getWorkspace',
+    writeClass: 'readOnly',
     description: 'Fetch metadata about the current workspace (name, settings).',
     tier: 'core',
     catalogLine: 'getWorkspace — fetch current workspace metadata (name, settings).',
@@ -249,6 +262,7 @@ export const SHARED_TOOL_SPECS = {
   listSpaces: {
     mcpName: 'listSpaces',
     inAppKey: 'listSpaces',
+    writeClass: 'readOnly',
     description:
       'List the spaces the current user can access. Returns the array of ' +
       'spaces (id, name, slug, ...).',
@@ -260,6 +274,7 @@ export const SHARED_TOOL_SPECS = {
   listShares: {
     mcpName: 'listShares',
     inAppKey: 'listShares',
+    writeClass: 'readOnly',
     description:
       'List all public shares in the workspace with page titles and public URLs.',
     tier: 'deferred',
@@ -272,6 +287,7 @@ export const SHARED_TOOL_SPECS = {
   getPageJson: {
     mcpName: 'getPageJson',
     inAppKey: 'getPageJson',
+    writeClass: 'readOnly',
     description:
       'Get page details with the raw ProseMirror JSON content (lossless: ' +
       'includes block ids, callouts, tables, link/image attributes) plus the ' +
@@ -289,6 +305,7 @@ export const SHARED_TOOL_SPECS = {
   getOutline: {
     mcpName: 'getOutline',
     inAppKey: 'getOutline',
+    writeClass: 'readOnly',
     description:
       "Return a COMPACT outline of a page's top-level blocks ({index, type, " +
       'id, level, firstText}; tables add rows/cols/header; lists add item ' +
@@ -309,6 +326,7 @@ export const SHARED_TOOL_SPECS = {
   getNode: {
     mcpName: 'getNode',
     inAppKey: 'getNode',
+    writeClass: 'readOnly',
     description:
       "Fetch a single block for editing. `nodeId` is a block id from the page " +
       'outline or page-JSON view (works for headings/paragraphs/callouts/images), OR ' +
@@ -350,6 +368,7 @@ export const SHARED_TOOL_SPECS = {
   searchInPage: {
     mcpName: 'searchInPage',
     inAppKey: 'searchInPage',
+    writeClass: 'readOnly',
     description:
       'Find every occurrence of a string (or regex) INSIDE one page and get ' +
       'WHERE each is — instead of pulling blocks one-by-one with getNode. ' +
@@ -413,6 +432,7 @@ export const SHARED_TOOL_SPECS = {
   deleteNode: {
     mcpName: 'deleteNode',
     inAppKey: 'deleteNode',
+    writeClass: 'write',
     description:
       'Remove a single block by its attrs.id (from the page outline or ' +
       'page-JSON view) WITHOUT resending the whole document.',
@@ -438,6 +458,7 @@ export const SHARED_TOOL_SPECS = {
   patchNode: {
     mcpName: 'patchNode',
     inAppKey: 'patchNode',
+    writeClass: 'write',
     description:
       'Replace a single content block identified by its attrs.id, WITHOUT ' +
       'resending the whole document; the replacement keeps the same block id. ' +
@@ -505,6 +526,7 @@ export const SHARED_TOOL_SPECS = {
   insertNode: {
     mcpName: 'insertNode',
     inAppKey: 'insertNode',
+    writeClass: 'write',
     description:
       'Insert content before/after another block (by attrs.id or anchor text) ' +
       'or append it at the end (top level). For before/after you MUST provide ' +
@@ -597,6 +619,7 @@ export const SHARED_TOOL_SPECS = {
   sharePage: {
     mcpName: 'sharePage',
     inAppKey: 'sharePage',
+    writeClass: 'write',
     // CANONICAL: merges the MCP copy's URL-format + idempotency detail with the
     // in-app copy's reversibility note; keeps the security framing both had.
     description:
@@ -626,6 +649,7 @@ export const SHARED_TOOL_SPECS = {
   unsharePage: {
     mcpName: 'unsharePage',
     inAppKey: 'unsharePage',
+    writeClass: 'write',
     description: 'Remove the public share of a page (revokes the public URL).',
     tier: 'deferred',
     catalogLine: "unsharePage — revoke a page's public share (removes the public URL).",
@@ -640,6 +664,7 @@ export const SHARED_TOOL_SPECS = {
   diffPageVersions: {
     mcpName: 'diffPageVersions',
     inAppKey: 'diffPageVersions',
+    writeClass: 'readOnly',
     description:
       'Diff two versions of a page and return a Docmost-equivalent change set ' +
       '(inserted/deleted text, integrity counts for images/links/tables/' +
@@ -672,6 +697,7 @@ export const SHARED_TOOL_SPECS = {
   listPageHistory: {
     mcpName: 'listPageHistory',
     inAppKey: 'listPageHistory',
+    writeClass: 'readOnly',
     description:
       "List a page's saved versions (Docmost auto-snapshots on every save), " +
       'newest first, cursor-paginated. Returns { items, nextCursor }; each ' +
@@ -693,6 +719,7 @@ export const SHARED_TOOL_SPECS = {
   restorePageVersion: {
     mcpName: 'restorePageVersion',
     inAppKey: 'restorePageVersion',
+    writeClass: 'write',
     description:
       'Restore a page to a saved version: writes that version\'s content back ' +
       'as the page\'s current content (Docmost has no restore endpoint, so ' +
@@ -713,6 +740,7 @@ export const SHARED_TOOL_SPECS = {
   importPageMarkdown: {
     mcpName: 'importPageMarkdown',
     inAppKey: 'importPageMarkdown',
+    writeClass: 'write',
     // IN-APP ONLY (issue #411): the external /mcp surface no longer exposes
     // importPageMarkdown — the registry loop in index.ts skips inAppOnly specs,
     // so this stays available to the in-app agent (round-tripping an EXPORTED
@@ -742,6 +770,7 @@ export const SHARED_TOOL_SPECS = {
   copyPageContent: {
     mcpName: 'copyPageContent',
     inAppKey: 'copyPageContent',
+    writeClass: 'write',
     description:
       "Replace targetPageId's content with a copy of sourcePageId's content, " +
       'entirely server-side — the document is NOT sent through the model. The ' +
@@ -770,6 +799,7 @@ export const SHARED_TOOL_SPECS = {
   editPageText: {
     mcpName: 'editPageText',
     inAppKey: 'editPageText',
+    writeClass: 'write',
     description:
       "Surgical find/replace inside a page's text, preserving all block " +
       'ids and marks. A find MAY cross bold/italic/link boundaries; the ' +
@@ -819,6 +849,7 @@ export const SHARED_TOOL_SPECS = {
   stashPage: {
     mcpName: 'stashPage',
     inAppKey: 'stashPage',
+    writeClass: 'readOnly',
     description:
       'Serialize a whole page (the full ProseMirror JSON, as getPageJson ' +
       'returns) into an ephemeral in-memory blob and return ONLY a short ' +
@@ -880,6 +911,7 @@ export const SHARED_TOOL_SPECS = {
   getPage: {
     mcpName: 'getPage',
     inAppKey: 'getPage',
+    writeClass: 'readOnly',
     description:
       'Fetch a single page as Markdown by its id. Returns the page title and ' +
       'its Markdown content. The converter is canonical (round-trips text and ' +
@@ -919,6 +951,7 @@ export const SHARED_TOOL_SPECS = {
   listPages: {
     mcpName: 'listPages',
     inAppKey: 'listPages',
+    writeClass: 'readOnly',
     description:
       'List the most recent pages (ordered by updatedAt, descending), ' +
       'optionally scoped to a single space. Returns a bounded list (default ' +
@@ -965,6 +998,7 @@ export const SHARED_TOOL_SPECS = {
   getTree: {
     mcpName: 'getTree',
     inAppKey: 'getTree',
+    writeClass: 'readOnly',
     description:
       "Get a space's page hierarchy (or one subtree) as a nested tree in a " +
       'SINGLE request — completely and without loss. Each node is ' +
@@ -1009,6 +1043,7 @@ export const SHARED_TOOL_SPECS = {
   getPageContext: {
     mcpName: 'getPageContext',
     inAppKey: 'getPageContext',
+    writeClass: 'readOnly',
     description:
       'Given a pageId, get its LOCATION and immediate surroundings (metadata ' +
       'only, no page content) in one call — answers "where am I / what is ' +
@@ -1038,6 +1073,7 @@ export const SHARED_TOOL_SPECS = {
   createPage: {
     mcpName: 'createPage',
     inAppKey: 'createPage',
+    writeClass: 'write',
     description:
       'Create a new page with a Markdown body in a space, optionally under a ' +
       'parent page (omit parentPageId to create at the space root). Returns ' +
@@ -1088,6 +1124,7 @@ export const SHARED_TOOL_SPECS = {
   movePage: {
     mcpName: 'movePage',
     inAppKey: 'movePage',
+    writeClass: 'write',
     description:
       'Move a page under a new parent page, or to the space root when no ' +
       'parent is given. Reversible: move it back at any time.',
@@ -1181,6 +1218,7 @@ export const SHARED_TOOL_SPECS = {
   renamePage: {
     mcpName: 'renamePage',
     inAppKey: 'renamePage',
+    writeClass: 'write',
     description:
       'Rename a page (change its title only; the body is untouched, never ' +
       'resent). Reversible: rename back at any time.',
@@ -1202,6 +1240,7 @@ export const SHARED_TOOL_SPECS = {
   deletePage: {
     mcpName: 'deletePage',
     inAppKey: 'deletePage',
+    writeClass: 'write',
     description:
       'Move a page to the trash — SOFT delete only: the page can be restored ' +
       'from trash and nothing is ever permanently deleted.',
@@ -1235,6 +1274,7 @@ export const SHARED_TOOL_SPECS = {
   updatePageJson: {
     mcpName: 'updatePageJson',
     inAppKey: 'updatePageJson',
+    writeClass: 'write',
     description:
       "Replace a page's content with a raw ProseMirror JSON document (lossless " +
       'write: preserves the block ids, callouts, tables and attributes you pass ' +
@@ -1291,6 +1331,7 @@ export const SHARED_TOOL_SPECS = {
   updatePageMarkdown: {
     mcpName: 'updatePageMarkdown',
     inAppKey: 'updatePageMarkdown',
+    writeClass: 'write',
     description:
       "Replace a page's body with new Markdown content (and optionally its " +
       'title). The whole body is re-imported from the markdown (block ids ' +
@@ -1325,6 +1366,7 @@ export const SHARED_TOOL_SPECS = {
   exportPageMarkdown: {
     mcpName: 'exportPageMarkdown',
     inAppKey: 'exportPageMarkdown',
+    writeClass: 'readOnly',
     // CANONICAL: the MCP copy (a strict superset of the terse in-app wording).
     description:
       'Export a page to a single self-contained Docmost-flavoured Markdown ' +
@@ -1373,6 +1415,7 @@ export const SHARED_TOOL_SPECS = {
   createComment: {
     mcpName: 'createComment',
     inAppKey: 'createComment',
+    writeClass: 'write',
     // CANONICAL: the in-app copy (the more-maintained one). It keeps the same
     // rules as the MCP copy — inline-only, top-level requires a `selection`, no
     // page-level comments, replies inherit the anchor, suggestedText must be
@@ -1505,6 +1548,7 @@ export const SHARED_TOOL_SPECS = {
   listComments: {
     mcpName: 'listComments',
     inAppKey: 'listComments',
+    writeClass: 'readOnly',
     // CANONICAL: the two copies are near-identical; the MCP copy is the
     // superset (it keeps the "(pagination is handled internally)" note the
     // in-app copy dropped), so it is used verbatim.
@@ -1532,6 +1576,7 @@ export const SHARED_TOOL_SPECS = {
   resolveComment: {
     mcpName: 'resolveComment',
     inAppKey: 'resolveComment',
+    writeClass: 'write',
     // CANONICAL: the MCP copy's richer wording, minus its reference
     // to `deleteComment` (a sibling tool that does NOT exist in the in-app
     // layer) — rephrased transport-neutrally per the registry convention.
@@ -1572,6 +1617,7 @@ export const SHARED_TOOL_SPECS = {
   checkNewComments: {
     mcpName: 'checkNewComments',
     inAppKey: 'checkNewComments',
+    writeClass: 'readOnly',
     // CANONICAL: the MCP copy (the more detailed of the two). The MCP layer's
     // execute-side guard that rejects an unparseable `since` timestamp stays in
     // its execute body (per-layer logic), not in the shared schema.
@@ -1651,6 +1697,7 @@ export const SHARED_TOOL_SPECS = {
   tableInsertRow: {
     mcpName: 'tableInsertRow',
     inAppKey: 'tableInsertRow',
+    writeClass: 'write',
     description:
       'Insert a row of plain-text cells into a table. `table` is `#<index>` ' +
       'from the page outline, or a block id inside it. `cells` is the text per ' +
@@ -1684,6 +1731,7 @@ export const SHARED_TOOL_SPECS = {
   tableDeleteRow: {
     mcpName: 'tableDeleteRow',
     inAppKey: 'tableDeleteRow',
+    writeClass: 'write',
     description:
       'Delete the row at 0-based `index` from a table (`table` is `#<index>` ' +
       'from the page outline, or a block id inside it). Refuses to delete the ' +
@@ -1707,6 +1755,7 @@ export const SHARED_TOOL_SPECS = {
   tableUpdateCell: {
     mcpName: 'tableUpdateCell',
     inAppKey: 'tableUpdateCell',
+    writeClass: 'write',
     description:
       'Set the plain-text content of cell [row, col] (0-based) in a table ' +
       '(`table` is `#<index>` from the page outline, or a block id inside it). ' +
@@ -1747,6 +1796,7 @@ export const SHARED_TOOL_SPECS = {
   insertFootnote: {
     mcpName: 'insertFootnote',
     inAppKey: 'insertFootnote',
+    writeClass: 'write',
     description:
       'Insert an AUTHOR-INLINE footnote: you specify only WHERE (anchorText) ' +
       'and WHAT (text). The footnote marker is placed right after anchorText in ' +
@@ -1784,6 +1834,7 @@ export const SHARED_TOOL_SPECS = {
   insertImage: {
     mcpName: 'insertImage',
     inAppKey: 'insertImage',
+    writeClass: 'write',
     description:
       'Download an image from a web (http/https) URL and insert it into ' +
       'a page in one step. By default ' +
@@ -1828,6 +1879,7 @@ export const SHARED_TOOL_SPECS = {
   replaceImage: {
     mcpName: 'replaceImage',
     inAppKey: 'replaceImage',
+    writeClass: 'write',
     description:
       'Replace an existing image on a page with a new image fetched from a web ' +
       '(http/https) URL: uploads the new file as a NEW ' +
@@ -1866,6 +1918,7 @@ export const SHARED_TOOL_SPECS = {
   drawioGet: {
     mcpName: 'drawioGet',
     inAppKey: 'drawioGet',
+    writeClass: 'readOnly',
     description:
       'Read a draw.io diagram on a page as mxGraph XML (default) or as its raw ' +
       '`.drawio.svg`. `node` is the drawio node\'s attrs.id (from getOutline / ' +
@@ -1899,6 +1952,7 @@ export const SHARED_TOOL_SPECS = {
   drawioCreate: {
     mcpName: 'drawioCreate',
     inAppKey: 'drawioCreate',
+    writeClass: 'write',
     description:
       'Create a draw.io diagram from mxGraph XML and insert it as a diagram ' +
       'block. `xml` is a bare `<mxGraphModel>` OR a list of `<mxCell>` elements ' +
@@ -1969,6 +2023,7 @@ export const SHARED_TOOL_SPECS = {
   drawioUpdate: {
     mcpName: 'drawioUpdate',
     inAppKey: 'drawioUpdate',
+    writeClass: 'write',
     description:
       'Replace a draw.io diagram\'s content with new mxGraph XML (same lint ' +
       'pipeline as drawioCreate). `baseHash` is MANDATORY: pass the hash from ' +
@@ -2020,6 +2075,7 @@ export const SHARED_TOOL_SPECS = {
   drawioEditCells: {
     mcpName: 'drawioEditCells',
     inAppKey: 'drawioEditCells',
+    writeClass: 'write',
     description:
       'Make TARGETED, id-based edits to an existing draw.io diagram instead of ' +
       'resending the whole XML (a full-XML diff is fragile — draw.io reorders ' +
@@ -2078,6 +2134,7 @@ export const SHARED_TOOL_SPECS = {
   drawioFromGraph: {
     mcpName: 'drawioFromGraph',
     inAppKey: 'drawioFromGraph',
+    writeClass: 'write',
     description:
       'Build a draw.io diagram from a SEMANTIC graph — you describe nodes, groups ' +
       'and edges by MEANING and the server picks every coordinate, color and icon ' +
@@ -2202,6 +2259,7 @@ export const SHARED_TOOL_SPECS = {
   drawioFromMermaid: {
     mcpName: 'drawioFromMermaid',
     inAppKey: 'drawioFromMermaid',
+    writeClass: 'write',
     description:
       'Convert Mermaid `flowchart` text into an EDITABLE draw.io diagram (LLMs ' +
       'write Mermaid reliably). Best for STANDARD flowcharts/decision trees: ' +
@@ -2248,6 +2306,7 @@ export const SHARED_TOOL_SPECS = {
   drawioShapes: {
     mcpName: 'drawioShapes',
     inAppKey: 'drawioShapes',
+    writeClass: 'readOnly',
     description:
       'Look up VERIFIED draw.io stencil style-strings so you never guess a ' +
       '`shape=mxgraph.*` name (a wrong name renders as an EMPTY BOX). Searches a ' +
@@ -2294,6 +2353,7 @@ export const SHARED_TOOL_SPECS = {
   drawioGuide: {
     mcpName: 'drawioGuide',
     inAppKey: 'drawioGuide',
+    writeClass: 'readOnly',
     description:
       'Progressive-disclosure draw.io authoring reference. Call with a `section` ' +
       'to pull one focused, <=4KB chapter instead of bloating context: ' +
@@ -2323,3 +2383,53 @@ export const SHARED_TOOL_SPECS = {
     inlineBothHosts: true,
   },
 } satisfies Record<string, SharedToolSpec>;
+
+// --- write-class registry (#489) ------------------------------------------
+
+/** A tool's retry-safety class. 'readOnly' may be auto-retried once after a
+ *  transport break; 'write' is indeterminate and must never be blind-retried. */
+export type ToolWriteClass = 'readOnly' | 'write';
+
+/**
+ * Name → write-class map for the shared registry, keyed by mcpName (=== inAppKey).
+ * The external-MCP retry path (mcp-clients.service.ts) looks a tool up here by its
+ * RAW (un-namespaced) name to decide whether a transport failure may be retried.
+ * A tool NOT in this map (a third-party external MCP tool) is treated as 'write'
+ * by the consumer — the safe default (never blind-retry an unknown tool).
+ */
+export const SHARED_TOOL_WRITE_CLASS: Record<string, ToolWriteClass> =
+  Object.fromEntries(
+    Object.values(SHARED_TOOL_SPECS).map((spec) => [spec.mcpName, spec.writeClass]),
+  );
+
+/** Whether a write-class permits a single automatic retry after a transport
+ *  break. Only a pure read is retry-safe; everything mutating is indeterminate. */
+export function isRetryableWriteClass(
+  writeClass: ToolWriteClass | undefined,
+): boolean {
+  return writeClass === 'readOnly';
+}
+
+/**
+ * Registration-time assert (#489): EVERY spec must declare a valid write-class.
+ * `satisfies Record<string, SharedToolSpec>` already makes an omission a compile
+ * error, but this guards a raw/cast construction path and documents the invariant
+ * at the point of use. Runs once on import — both hosts import this module, so
+ * both get the check. Throws (fails startup) rather than silently mis-gating a
+ * retry in production.
+ */
+export function assertEverySpecDeclaresWriteClass(): void {
+  for (const [key, spec] of Object.entries(SHARED_TOOL_SPECS)) {
+    const wc = (spec as SharedToolSpec).writeClass;
+    if (wc !== 'readOnly' && wc !== 'write') {
+      throw new Error(
+        `tool-specs: spec "${key}" must declare writeClass ('readOnly' | 'write'), got ${JSON.stringify(
+          wc,
+        )}`,
+      );
+    }
+  }
+}
+
+// Enforce at module load (registration time) on both hosts.
+assertEverySpecDeclaresWriteClass();

packages/mcp/test/mock/create-comment.test.mjs

@@ -553,6 +553,189 @@ test("suggestedText: the stored selection is the doc's RAW typographic substring
   assert.equal(createPayload.suggestedText, "goodbye");
 });
 
+// -----------------------------------------------------------------------------
+// 8b) #496: the DEBOUNCED REST snapshot (pages/info) DIFFERS from the LIVE collab
+//     doc (mutatePage) — the doc moved on in the debounce window. The stored
+//     selection is captured from the snapshot at create time, but the mark is set
+//     in the live doc, so apply would 409 forever. After anchoring, the client
+//     re-reads the RAW substring under the mark from the LIVE doc and POSTs it to
+//     /comments/resync-suggestion-anchor so the stored expectedText matches.
+// -----------------------------------------------------------------------------
+test("suggestion: re-syncs the stored selection to the LIVE doc substring when the REST snapshot lagged", async () => {
+  let createPayload = null;
+  let resyncPayload = null;
+
+  const { baseURL } = await spawn(async (req, res) => {
+    const raw = await readBody(req);
+    if (req.url === "/api/auth/login") {
+      sendJson(res, 200, { success: true }, {
+        "Set-Cookie": "authToken=t; Path=/; HttpOnly",
+      });
+      return;
+    }
+    if (req.url === "/api/pages/info") {
+      // DEBOUNCED snapshot: ASCII quotes (stale — the live doc has since been
+      // typographically corrected).
+      sendJson(res, 200, {
+        data: {
+          id: "33333333-3333-3333-3333-333333333333",
+          content: {
+            type: "doc",
+            content: [
+              {
+                type: "paragraph",
+                content: [{ type: "text", text: 'he said "hello" loudly' }],
+              },
+            ],
+          },
+        },
+      });
+      return;
+    }
+    if (req.url === "/api/comments/create") {
+      createPayload = JSON.parse(raw);
+      sendJson(res, 200, {
+        data: {
+          id: "cmt-resync-1",
+          content: createPayload.content,
+          selection: createPayload.selection,
+          suggestedText: createPayload.suggestedText,
+          type: createPayload.type,
+        },
+      });
+      return;
+    }
+    if (req.url === "/api/comments/resync-suggestion-anchor") {
+      resyncPayload = JSON.parse(raw);
+      sendJson(res, 200, { data: { id: "cmt-resync-1" } });
+      return;
+    }
+    sendJson(res, 404, { message: "not found" });
+  });
+
+  class TestClient extends DocmostClient {
+    async getCollabTokenWithReauth() {
+      return "collab-token";
+    }
+    async resolvePageId() {
+      return "33333333-3333-3333-3333-333333333333";
+    }
+    async mutatePage(pageId, collabToken, apiUrl, transform) {
+      // LIVE doc: SMART quotes (what the mark is actually set over).
+      const doc = {
+        type: "doc",
+        content: [
+          {
+            type: "paragraph",
+            content: [{ type: "text", text: "he said “hello” loudly" }],
+          },
+        ],
+      };
+      const out = transform(doc);
+      return { doc: out, verify: { ok: true } };
+    }
+  }
+
+  const client = new TestClient(baseURL, "user@example.com", "pw");
+
+  const result = await client.createComment(
+    "33333333-3333-3333-3333-333333333333",
+    "please change",
+    "inline",
+    '"hello"', // ASCII quotes
+    undefined,
+    "goodbye",
+  );
+
+  assert.equal(result.success, true);
+  assert.equal(result.anchored, true);
+  // Create stored the STALE snapshot substring (ASCII quotes).
+  assert.equal(createPayload.selection, '"hello"');
+  // …then the client re-synced to the LIVE marked substring (smart quotes).
+  assert.ok(resyncPayload, "/comments/resync-suggestion-anchor must be called");
+  assert.equal(resyncPayload.commentId, "cmt-resync-1");
+  assert.equal(resyncPayload.selection, "“hello”");
+  // The returned comment reflects the corrected anchor.
+  assert.equal(result.data.selection, "“hello”");
+});
+
+// -----------------------------------------------------------------------------
+// 8c) #496: when the REST snapshot and the LIVE doc AGREE, no resync round-trip
+//     is made (the common case must stay a single write).
+// -----------------------------------------------------------------------------
+test("suggestion: no resync call when the snapshot already matches the live doc", async () => {
+  let resyncCalls = 0;
+
+  const { baseURL } = await spawn(async (req, res) => {
+    const raw = await readBody(req);
+    if (req.url === "/api/auth/login") {
+      sendJson(res, 200, { success: true }, {
+        "Set-Cookie": "authToken=t; Path=/; HttpOnly",
+      });
+      return;
+    }
+    if (req.url === "/api/pages/info") {
+      sendJson(res, 200, {
+        data: {
+          id: "44444444-4444-4444-4444-444444444444",
+          content: {
+            type: "doc",
+            content: [
+              { type: "paragraph", content: [{ type: "text", text: "Hello brave world" }] },
+            ],
+          },
+        },
+      });
+      return;
+    }
+    if (req.url === "/api/comments/create") {
+      const p = JSON.parse(raw);
+      sendJson(res, 200, {
+        data: { id: "cmt-nosync-1", content: p.content, selection: p.selection, suggestedText: p.suggestedText, type: p.type },
+      });
+      return;
+    }
+    if (req.url === "/api/comments/resync-suggestion-anchor") {
+      resyncCalls++;
+      sendJson(res, 200, { data: {} });
+      return;
+    }
+    sendJson(res, 404, { message: "not found" });
+  });
+
+  class TestClient extends DocmostClient {
+    async getCollabTokenWithReauth() {
+      return "collab-token";
+    }
+    async resolvePageId() {
+      return "44444444-4444-4444-4444-444444444444";
+    }
+    async mutatePage(pageId, collabToken, apiUrl, transform) {
+      const doc = {
+        type: "doc",
+        content: [
+          { type: "paragraph", content: [{ type: "text", text: "Hello brave world" }] },
+        ],
+      };
+      const out = transform(doc);
+      return { doc: out, verify: { ok: true } };
+    }
+  }
+
+  const client = new TestClient(baseURL, "user@example.com", "pw");
+  const result = await client.createComment(
+    "44444444-4444-4444-4444-444444444444",
+    "rename",
+    "inline",
+    "brave",
+    undefined,
+    "bold",
+  );
+
+  assert.equal(result.anchored, true);
+  assert.equal(resyncCalls, 0, "matching snapshot must NOT trigger a resync round-trip");
+});
+
 // -----------------------------------------------------------------------------
 // 8) #408: a not-found selection error QUOTES the closest block text so the
 //    model can self-correct instead of blind-retrying.

packages/mcp/test/unit/tool-specs.test.mjs

@@ -2,7 +2,12 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 import { z } from "zod";
 
-import { SHARED_TOOL_SPECS } from "../../build/tool-specs.js";
+import {
+  SHARED_TOOL_SPECS,
+  SHARED_TOOL_WRITE_CLASS,
+  isRetryableWriteClass,
+  assertEverySpecDeclaresWriteClass,
+} from "../../build/tool-specs.js";
 
 // The shared registry is consumed by BOTH the zod-v3 MCP server and the zod-v4
 // in-app AI-SDK service, so every spec must carry the cross-layer wiring
@@ -43,6 +48,41 @@ test("mcpName and inAppKey are each unique across the registry", () => {
   }
 });
 
+// #489 — every spec must declare its write-class so the external-MCP retry path
+// can gate a single auto-retry ONLY on a pure read (a blind retry of a write =
+// double-apply). The declaration is enforced at registration time.
+test("#489: every spec declares a valid writeClass ('readOnly' | 'write')", () => {
+  for (const [key, spec] of Object.entries(SHARED_TOOL_SPECS)) {
+    assert.ok(
+      spec.writeClass === "readOnly" || spec.writeClass === "write",
+      `${key}: missing/invalid writeClass: ${JSON.stringify(spec.writeClass)}`,
+    );
+  }
+  // The registration-time assert must not throw for the shipped registry.
+  assert.doesNotThrow(() => assertEverySpecDeclaresWriteClass());
+});
+
+test("#489: SHARED_TOOL_WRITE_CLASS maps every mcpName to its class; helper gates on readOnly", () => {
+  const specs = Object.values(SHARED_TOOL_SPECS);
+  assert.equal(Object.keys(SHARED_TOOL_WRITE_CLASS).length, specs.length);
+  for (const spec of specs) {
+    assert.equal(SHARED_TOOL_WRITE_CLASS[spec.mcpName], spec.writeClass);
+  }
+  // Only a readOnly tool is retry-eligible; a write tool and an unknown tool are not.
+  assert.equal(isRetryableWriteClass("readOnly"), true);
+  assert.equal(isRetryableWriteClass("write"), false);
+  assert.equal(isRetryableWriteClass(undefined), false);
+});
+
+test("#489: representative reads are readOnly and representative writes are write", () => {
+  for (const name of ["getPage", "getTree", "searchInPage", "listComments"]) {
+    assert.equal(SHARED_TOOL_SPECS[name].writeClass, "readOnly", `${name} should be readOnly`);
+  }
+  for (const name of ["patchNode", "createPage", "deletePage", "createComment", "drawioCreate"]) {
+    assert.equal(SHARED_TOOL_SPECS[name].writeClass, "write", `${name} should be write`);
+  }
+});
+
 test("buildShape (when present) returns a usable ZodRawShape with a real zod", () => {
   for (const [key, spec] of Object.entries(SHARED_TOOL_SPECS)) {
     if (!spec.buildShape) continue;