fix(state): fail loudly when state.db is replaced under a live process

Detect same-inode cp via a generation stamp, halt FTS repair, and divert
unwritten transcripts to sessions/<id>.jsonl plus the gateway pending spool.

Co-authored-by: Cursor <cursoragent@cursor.com>

agent/turn_finalizer.py

@@ -752,7 +752,7 @@ def finalize_turn(
             "health (`hermes doctor`), then send your message again"
         )
         # Machine-readable cause for the gateway/desktop: exactly
-        # 'session_persistence_failed:<locked|compression|turn_lease|corrupt|disk|unknown>'.
+        # 'session_persistence_failed:<locked|compression|turn_lease|corrupt|replaced|disk|unknown>'.
         # Never clobber a failure_reason another path already stamped.
         if "failure_reason" not in result:
             _cause = getattr(agent, "_last_persistence_error_cause", None)

gateway/session.py

@@ -3756,7 +3756,43 @@ class SessionStore:
             try:
                 self._append_transcript_message(session_id, msg)
             except Exception as exc:
-                from hermes_state import CompressionSessionClosedError
+                from hermes_state import CompressionSessionClosedError, StateDbReplacedError
+
+                if isinstance(exc, StateDbReplacedError):
+                    logger.error(
+                        "Session DB was replaced underneath the gateway for %s; "
+                        "stopping SQLite writes and diverting pending "
+                        "transcripts to the on-disk fallback: %s",
+                        session_id, exc,
+                    )
+                    with self._transcript_retry_lock:
+                        remaining = list(self._dirty_transcripts.get(queue_session_id, []))
+                        self._dirty_transcripts.pop(queue_session_id, None)
+                        self._transcript_append_failures.pop(session_id, None)
+                    for dropped in remaining:
+                        try:
+                            from gateway.shutdown_flush import (
+                                spool_dropped_transcript_message,
+                            )
+                            spool_dropped_transcript_message(session_id, dropped)
+                        except Exception:
+                            logger.warning(
+                                "pending fallback failed for replaced "
+                                "state.db transcript on %s",
+                                session_id,
+                                exc_info=True,
+                            )
+                    try:
+                        from hermes_state import divert_session_transcript_jsonl
+                        divert_session_transcript_jsonl(session_id, remaining)
+                    except Exception:
+                        logger.warning(
+                            "JSONL divert failed for replaced state.db "
+                            "transcript on %s",
+                            session_id,
+                            exc_info=True,
+                        )
+                    return
 
                 if isinstance(exc, CompressionSessionClosedError):
                     # Resolve the full continuation chain via the canonical

hermes_state.py

@@ -26,9 +26,11 @@ import queue
 import random
 import re
 import sqlite3
+import struct
 import sys
 import threading
 import time
+import uuid
 import weakref
 from collections import deque
 from contextlib import contextmanager
@@ -2128,6 +2130,7 @@ PERSISTENCE_ERROR_CAUSES = (
     "compression_closed",
     "turn_lease",
     "corrupt",
+    "replaced",
     "disk",
     "unknown",
 )
@@ -2173,6 +2176,9 @@ def classify_persistence_error(exc_or_str) -> str:
       (``database disk image is malformed`` / SQLITE_NOTADB).  Distinct from
       ``"disk"``: freeing space cannot help, the user needs the repair path
       (``hermes doctor`` / automatic schema surgery).
+    * ``"replaced"`` — the ``state.db`` path no longer names the file this
+      process opened (out-of-band ``cp``/``mv``/restore). In-file FTS repair
+      cannot help; writes to the live handle must stop.
     * ``"disk"``    — disk full / read-only / permission-shaped failures
       (delegates the disk-full patterns to :func:`is_disk_full_error` so the
       two classifiers can never drift apart — e.g. ENOSPC).
@@ -2191,6 +2197,8 @@ def classify_persistence_error(exc_or_str) -> str:
         return "compression_closed"
     if isinstance(exc_or_str, CompressionSessionBusyError):
         return "compression"
+    if isinstance(exc_or_str, StateDbReplacedError):
+        return "replaced"
     text = str(exc_or_str).lower()
     if "turn lease" in text:
         return "turn_lease"
@@ -2198,6 +2206,8 @@ def classify_persistence_error(exc_or_str) -> str:
         return "compression_closed"
     if "being compressed" in text or "compression lease" in text:
         return "compression"
+    if "was replaced underneath" in text:
+        return "replaced"
     # Structural corruption BEFORE the lock and disk buckets: "database disk
     # image is malformed" contains "disk" (and some wrapped corruption
     # strings mention "locked" recovery attempts), so later buckets would
@@ -4171,6 +4181,82 @@ class SessionTurnLeaseLostError(RuntimeError):
     """
 
 
+class StateDbReplacedError(RuntimeError):
+    """The state.db path no longer names the file this SessionDB opened.
+
+    Raised when an out-of-band ``cp``/``mv``/restore replaces the database
+    under a live gateway. In-place FTS repair and fail-open trigger
+    dropping cannot fix a generation mismatch; they amplify it.
+    """
+
+
+# SQLite header: 4-byte big-endian application_id at offset 68. Distinct from
+# inode: ``cp`` onto the same path keeps st_ino and truncates+rewrites.
+_STATE_DB_APPLICATION_ID_OFFSET = 68
+_STATE_DB_GENERATION_KEY = "db_file_generation"
+_STATE_DB_REPLACED_MSG = (
+    "FATAL: state.db was replaced underneath the gateway; refusing further "
+    "writes to this file. Divert transcripts to sessions/<id>.jsonl (and the "
+    "gateway pending_messages spool) and restore or reopen after operator "
+    "intervention."
+)
+
+
+def divert_session_transcript_jsonl(session_id: str, messages) -> "Optional[Path]":
+    """Append pending messages as JSON lines under HERMES_HOME/sessions.
+
+    Used when state.db is replaced under a live process so the current
+    turn is not only in RAM. Returns the jsonl path, or None when there
+    is nothing to write.
+    """
+    sid = str(session_id or "").strip()
+    if not sid or not messages:
+        return None
+    sessions_dir = get_hermes_home() / "sessions"
+    sessions_dir.mkdir(parents=True, exist_ok=True)
+    path = sessions_dir / f"{sid}.jsonl"
+    with path.open("a", encoding="utf-8") as handle:
+        for msg in messages:
+            if isinstance(msg, dict):
+                handle.write(json.dumps(msg, ensure_ascii=False, default=str) + "\n")
+            elif msg is not None:
+                handle.write(json.dumps({"content": str(msg)}, ensure_ascii=False) + "\n")
+    return path
+
+
+def _read_sqlite_application_id(db_path: Path) -> "Optional[int]":
+    """Read application_id from the SQLite header without opening a connection."""
+    try:
+        with db_path.open("rb") as handle:
+            header = handle.read(_STATE_DB_APPLICATION_ID_OFFSET + 4)
+    except OSError:
+        return None
+    if len(header) < _STATE_DB_APPLICATION_ID_OFFSET + 4:
+        return None
+    if header[:16] != b"SQLite format 3\x00":
+        return None
+    return int(
+        struct.unpack(
+            ">I",
+            header[_STATE_DB_APPLICATION_ID_OFFSET:_STATE_DB_APPLICATION_ID_OFFSET + 4],
+        )[0]
+    )
+
+
+def _stat_db_file_identity(path: Path) -> "Optional[tuple]":
+    """Return ``(st_dev, st_ino)`` for *path*, or None when identity is unavailable."""
+    try:
+        st = os.stat(path)
+    except OSError:
+        return None
+    # Windows volumes (and some network FS) report st_ino=0; a (0, 0)
+    # identity would false-positive every check. Skip the inode half of
+    # the guard there; generation stamp still applies.
+    if not st.st_dev or not st.st_ino:
+        return None
+    return (st.st_dev, st.st_ino)
+
+
 def _connect_tracked_db(path, tracking_path=None, **kwargs):
     """``sqlite3.connect`` that registers the open fd for lock-safety.
 
@@ -4884,6 +4970,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         self._read_open_failed_at = 0.0
         self._wal_active = False
         self._write_count = 0
+        # File identity of the state.db this instance opened. Compared on
+        # every write (and before FTS fail-open / reopen-after-close) so an
+        # out-of-band replace cannot limp through in-place surgery.
+        # Inode catches mv/new-file; application_id catches cp onto the
+        # same path (same inode, truncate+rewrite).
+        self._db_file_identity: Optional[tuple] = None
+        self._db_file_application_id: int = 0
+        self._db_file_generation_token: str = ""
+        self._db_replaced = False
         # One-shot guard for the usermerge-floor config write on the
         # incremental FTS merge cadence (see _merge_fts_incrementally).
         self._fts_usermerge_floor_applied = False
@@ -4958,6 +5053,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     except Exception:
                         pass
                     raise
+                self._record_db_file_identity()
                 initialization_complete = True
                 return
 
@@ -5115,6 +5211,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             # racing session lifecycle and the surprise disk/latency cost on
             # an unattended open. (An interrupted optimize resumes when the
             # user re-runs the command.)
+            self._ensure_db_file_generation()
+            self._record_db_file_identity()
             initialization_complete = True
         except Exception as exc:
             # Capture the cause so /resume and friends can surface WHY the
@@ -5391,6 +5489,12 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 f"SessionDB for {self.db_path} was closed (read-only handle); "
                 f"cannot serve a {context} after close()"
             )
+        # A reopen resolves the PATH again — if the file at that path is no
+        # longer the one this instance originally opened (out-of-band
+        # restore/cp/mv), reconnecting would write into the new generation
+        # through stale WAL/shm assumptions (#89332). Refuse instead.
+        if self._db_replaced or self._db_file_was_replaced():
+            self._halt_db_replaced()
         logger.warning(
             "state.db connection for %s was closed while a %s was still in "
             "flight — reopening (teardown/worker race, #94736)",
@@ -5728,6 +5832,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             return "no more rows available" in str(exc).lower()
 
         while True:
+            self._raise_if_db_replaced()
             try:
                 with self._lock:
                     if self._conn is None:
@@ -5793,6 +5898,16 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             except sqlite3.DatabaseError as exc:
                 if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):
                     continue
+                # An out-of-band replace of state.db (restore/cp/mv under a
+                # live process) surfaces as this same corruption error class.
+                # In-file repair on a NEW file generation amplifies the
+                # damage (#89332) — halt writes on this handle instead.
+                if (
+                    "not a database" in str(exc).lower()
+                    or is_malformed_db_error(exc)
+                    or self._is_fts_write_corruption_error(exc)
+                ):
+                    self._raise_if_db_replaced()
                 # Corrupt FTS shadow tables make every write raise the
                 # malformed/corrupt error class through the FTS sync triggers
                 # while the canonical messages table is intact. Never run a
@@ -5814,6 +5929,88 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     continue
                 raise
 
+    def _ensure_db_file_generation(self) -> None:
+        """Mint a once-per-file generation stamp (state_meta + application_id).
+
+        First opener wins via INSERT OR IGNORE. application_id is written
+        only when still 0 so racers converge on the same header value.
+        PASSIVE checkpoint only — never TRUNCATE (#45383).
+        """
+        if self.read_only or self._conn is None:
+            return
+        token = uuid.uuid4().hex
+        try:
+            self._conn.execute(
+                "INSERT OR IGNORE INTO state_meta (key, value) VALUES (?, ?)",
+                (_STATE_DB_GENERATION_KEY, token),
+            )
+            row = self._conn.execute(
+                "SELECT value FROM state_meta WHERE key = ?",
+                (_STATE_DB_GENERATION_KEY,),
+            ).fetchone()
+            if row and row[0]:
+                token = str(row[0])
+            self._db_file_generation_token = token
+            current = 0
+            pragma_row = self._conn.execute("PRAGMA application_id").fetchone()
+            if pragma_row:
+                current = int(pragma_row[0] or 0)
+            if current == 0:
+                app_id = int(token[:8], 16) & 0x7FFFFFFF
+                if app_id == 0:
+                    app_id = 1
+                self._conn.execute(f"PRAGMA application_id={app_id}")
+                current = app_id
+            self._db_file_application_id = current
+            try:
+                self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
+            except sqlite3.Error:
+                pass
+        except sqlite3.Error as exc:
+            logger.debug("state.db generation stamp skipped: %s", exc)
+
+    def _record_db_file_identity(self) -> None:
+        """Snapshot inode plus the on-disk generation header when present."""
+        self._db_file_identity = _stat_db_file_identity(self.db_path)
+        disk_id = _read_sqlite_application_id(self.db_path)
+        if disk_id:
+            self._db_file_application_id = disk_id
+        elif self._conn is not None and not self._db_file_application_id:
+            try:
+                pragma_row = self._conn.execute("PRAGMA application_id").fetchone()
+                if pragma_row and pragma_row[0]:
+                    self._db_file_application_id = int(pragma_row[0])
+            except sqlite3.Error:
+                pass
+
+    def _db_file_was_replaced(self) -> bool:
+        """True when the path no longer names the file this instance opened."""
+        recorded = self._db_file_identity
+        if recorded is not None:
+            current = _stat_db_file_identity(self.db_path)
+            if current is None or current != recorded:
+                return True
+        recorded_app = int(self._db_file_application_id or 0)
+        if recorded_app:
+            disk_app = _read_sqlite_application_id(self.db_path)
+            # Header 0 means the WAL has not been checkpointed yet — not a
+            # replace. A copied Hermes DB that minted its own id is nonzero.
+            if disk_app and disk_app != recorded_app:
+                return True
+        return False
+
+    def _halt_db_replaced(self) -> None:
+        """Stop writes and raise; do not run in-file repair on a new generation."""
+        self._db_replaced = True
+        logger.error(_STATE_DB_REPLACED_MSG)
+        raise StateDbReplacedError(_STATE_DB_REPLACED_MSG)
+
+    def _raise_if_db_replaced(self) -> None:
+        if self._db_replaced:
+            raise StateDbReplacedError(_STATE_DB_REPLACED_MSG)
+        if self._db_file_was_replaced():
+            self._halt_db_replaced()
+
     def _sleep_before_write_retry(
         self, deadline: float, patience_s: float
     ) -> bool:
@@ -6035,6 +6232,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         """
         if not self._fts_enabled or not self._is_fts_write_corruption_error(exc):
             return False
+        if self._db_replaced or self._db_file_was_replaced():
+            self._halt_db_replaced()
 
         try:
             with self._lock:

run_agent.py

@@ -2550,10 +2550,24 @@ class AIAgent:
             # ("storage was busy, send it again") from disk-full/read-only.
             from hermes_state import (
                 CompressionSessionClosedError,
+                StateDbReplacedError,
                 classify_persistence_error,
+                divert_session_transcript_jsonl,
             )
 
             self._last_persistence_error_cause = classify_persistence_error(e)
+            if isinstance(e, StateDbReplacedError):
+                try:
+                    divert_session_transcript_jsonl(
+                        getattr(self, "session_id", "") or "",
+                        _batch_rows,
+                    )
+                except Exception:
+                    logger.warning(
+                        "JSONL divert failed after state.db replace for %s",
+                        getattr(self, "session_id", None),
+                        exc_info=True,
+                    )
             if isinstance(e, CompressionSessionClosedError):
                 # Compression race: another path rotated this session while
                 # this turn was still writing against it. The store resolves
@@ -4129,6 +4143,17 @@ class AIAgent:
                     "database). Your message should already be saved — "
                     "please send it again in a moment."
                 )
+            if cause == "replaced":
+                return (
+                    prefix
+                    + "the turn was stopped because the state database file "
+                    "was replaced underneath this process. Do not run "
+                    "`hermes doctor --fix` or in-place FTS repair — stop "
+                    "the process, restore the intended state.db, then "
+                    "restart. Unwritten messages were diverted to "
+                    "sessions/<session_id>.jsonl and, on the gateway, "
+                    "pending_messages/pending-*.json."
+                )
             if cause == "corrupt":
                 return (
                     prefix

tests/gateway/test_session_db_replaced_fallback.py

@@ -0,0 +1,97 @@
+"""Gateway SessionStore must not FTS-repair a replaced state.db (#89332)."""
+
+import json
+import os
+import shutil
+
+import pytest
+
+from gateway.config import GatewayConfig
+from gateway.session import SessionStore
+from hermes_state import SessionDB
+
+
+def _assert_diverted(tmp_path, sid, needle):
+    pending = list((tmp_path / "pending_messages").glob("pending-*.json"))
+    assert pending, "expected pending_messages/pending-*.json spool"
+    spooled = False
+    for path in pending:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        message = (payload.get("data") or {}).get("message") or {}
+        if needle in str(message.get("content", "")):
+            spooled = True
+            break
+    assert spooled, f"{needle!r} missing from pending spool"
+    jsonl = tmp_path / "sessions" / f"{sid}.jsonl"
+    assert jsonl.is_file()
+    assert needle in jsonl.read_text(encoding="utf-8")
+
+
+def test_replaced_state_db_diverts_pending_without_fts_rebuild(tmp_path, monkeypatch):
+    import hermes_state
+
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", live)
+
+    store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig())
+    sid = "gw-replaced"
+    store._db.create_session(session_id=sid, source="cli")
+    store.append_to_transcript(
+        sid, {"role": "user", "content": "before", "timestamp": 1.0}
+    )
+    if store._db._db_file_identity is None:
+        store.close_all_db_handles()
+        pytest.skip("filesystem does not expose st_dev/st_ino")
+
+    alt = SessionDB(db_path=other)
+    alt.create_session("other", "cli")
+    alt.close()
+    os.replace(other, live)
+
+    store.append_to_transcript(
+        sid, {"role": "user", "content": "after-replace", "timestamp": 2.0}
+    )
+
+    assert store._db._db_replaced is True
+    assert store._db._fts_runtime_rebuild_attempted is False
+    assert store._fts_rebuild_attempted is False
+    _assert_diverted(tmp_path, sid, "after-replace")
+    store.close_all_db_handles()
+
+
+def test_copyfile_replaced_state_db_diverts_pending_without_fts_rebuild(
+    tmp_path, monkeypatch
+):
+    import hermes_state
+
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", live)
+
+    store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig())
+    sid = "gw-cp-replaced"
+    store._db.create_session(session_id=sid, source="cli")
+    store.append_to_transcript(
+        sid, {"role": "user", "content": "before-cp", "timestamp": 1.0}
+    )
+    if not store._db._db_file_application_id:
+        store.close_all_db_handles()
+        pytest.skip("generation stamp not recorded")
+
+    alt = SessionDB(db_path=other)
+    alt.create_session("other", "cli")
+    alt.close()
+    shutil.copyfile(other, live)
+
+    store.append_to_transcript(
+        sid, {"role": "user", "content": "after-cp", "timestamp": 2.0}
+    )
+
+    assert store._db._db_replaced is True
+    assert store._db._fts_runtime_rebuild_attempted is False
+    assert store._fts_rebuild_attempted is False
+    _assert_diverted(tmp_path, sid, "after-cp")
+    store.close_all_db_handles()

tests/hermes_state/test_state_db_file_identity.py

@@ -0,0 +1,177 @@
+"""File-identity guard on SessionDB writes (#89332).
+
+When state.db is replaced out-of-band under a live handle, in-place FTS
+rebuild / fail-open cannot help: they operate on a generation mismatch.
+The store must fail loudly instead of limping.
+"""
+
+import json
+import os
+import shutil
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+from hermes_state import (
+    SessionDB,
+    StateDbReplacedError,
+    classify_persistence_error,
+    divert_session_transcript_jsonl,
+)
+
+
+def _make_db(path: Path, session_id: str, content: str) -> SessionDB:
+    db = SessionDB(db_path=path)
+    db.create_session(session_id, "cli")
+    db.append_message(session_id, role="user", content=content)
+    return db
+
+
+def _require_identity(db: SessionDB) -> None:
+    if db._db_file_identity is None:
+        pytest.skip("filesystem does not expose st_dev/st_ino for identity checks")
+
+
+def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path):
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    db = _make_db(live, "live-sess", "original")
+    _require_identity(db)
+    alt = _make_db(other, "other-sess", "replacement")
+    alt.close()
+
+    recorded = db._db_file_identity
+    assert recorded is not None
+    os.replace(other, live)
+    assert _stat_changed(live, recorded)
+
+    with pytest.raises(StateDbReplacedError, match="replaced underneath"):
+        db.append_message("live-sess", role="user", content="after-replace")
+
+    assert db._db_replaced is True
+    assert db._fts_runtime_rebuild_attempted is False
+    assert db._fts_enabled is True
+    db.close()
+
+
+def test_second_write_after_halt_does_not_attempt_repair(tmp_path):
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    db = _make_db(live, "s", "a")
+    _require_identity(db)
+    alt = _make_db(other, "t", "b")
+    alt.close()
+    os.replace(other, live)
+    with pytest.raises(StateDbReplacedError):
+        db.append_message("s", role="user", content="first")
+    with pytest.raises(StateDbReplacedError):
+        db.append_message("s", role="user", content="second")
+    assert db._fts_runtime_rebuild_attempted is False
+    db.close()
+
+
+def test_same_file_fts_corruption_still_rebuilds(tmp_path):
+    """Identity guard must not disable genuine in-file FTS recovery."""
+    db = _make_db(tmp_path / "state.db", "s1", "hello world")
+    _require_identity(db)
+    identity = db._db_file_identity
+    raw = sqlite3.connect(str(tmp_path / "state.db"))
+    raw.execute(
+        "UPDATE messages_fts_data SET block = X'DEADBEEFDEADBEEFDEADBEEFDEADBEEF'"
+    )
+    raw.commit()
+    raw.close()
+    db.append_message("s1", role="user", content="healed append")
+    assert db._db_file_identity == identity
+    assert db._db_replaced is False
+    assert db._fts_runtime_rebuild_attempted is True
+    db.close()
+
+
+def test_classify_replaced_is_not_disk_or_fts_repair():
+    err = StateDbReplacedError(
+        "FATAL: state.db was replaced underneath the gateway; refusing further writes"
+    )
+    assert classify_persistence_error(err) == "replaced"
+    assert classify_persistence_error(str(err)) == "replaced"
+
+
+def test_new_sessiondb_on_replaced_path_records_new_identity(tmp_path):
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    db = _make_db(live, "s", "a")
+    old_id = db._db_file_identity
+    _require_identity(db)
+    db.close()
+    alt = _make_db(other, "t", "b")
+    alt.close()
+    os.replace(other, live)
+    reopened = SessionDB(db_path=live)
+    try:
+        assert reopened._db_file_identity != old_id
+        reopened.append_message("t", role="user", content="adopted after reopen")
+        assert reopened._db_replaced is False
+    finally:
+        reopened.close()
+
+
+def test_malformed_error_on_replaced_file_skips_fts_rebuild(tmp_path):
+    """Even if SQLite surfaces malformed, identity mismatch blocks repair."""
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    db = _make_db(live, "s", "a")
+    _require_identity(db)
+    alt = _make_db(other, "t", "b")
+    alt.close()
+    os.replace(other, live)
+
+    with pytest.raises(StateDbReplacedError):
+        db._try_runtime_fts_rebuild(
+            sqlite3.DatabaseError("database disk image is malformed")
+        )
+    assert db._fts_runtime_rebuild_attempted is False
+    db.close()
+
+
+def test_copyfile_same_inode_fails_loudly_without_fts_repair(tmp_path):
+    """``cp`` keeps st_ino; generation stamp must still halt (#89332)."""
+    live = tmp_path / "state.db"
+    other = tmp_path / "other.db"
+    db = _make_db(live, "live-sess", "original")
+    alt = _make_db(other, "other-sess", "replacement")
+    live_app = db._db_file_application_id
+    other_app = alt._db_file_application_id
+    if not live_app or not other_app:
+        alt.close()
+        db.close()
+        pytest.skip("generation stamp not recorded on this filesystem")
+    assert live_app != other_app
+    recorded = db._db_file_identity
+    alt.close()
+    shutil.copyfile(other, live)
+    if recorded is not None:
+        st = os.stat(live)
+        assert (st.st_dev, st.st_ino) == recorded
+    with pytest.raises(StateDbReplacedError, match="replaced underneath"):
+        db.append_message("live-sess", role="user", content="after-cp")
+    assert db._db_replaced is True
+    assert db._fts_runtime_rebuild_attempted is False
+    db.close()
+
+
+def test_divert_session_transcript_jsonl_appends(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    path = divert_session_transcript_jsonl(
+        "sess-jsonl",
+        [{"role": "user", "content": "hello-jsonl"}],
+    )
+    assert path == tmp_path / "sessions" / "sess-jsonl.jsonl"
+    lines = path.read_text(encoding="utf-8").strip().splitlines()
+    assert json.loads(lines[-1])["content"] == "hello-jsonl"
+    assert divert_session_transcript_jsonl("sess-jsonl", []) is None
+
+
+def _stat_changed(path: Path, recorded) -> bool:
+    st = os.stat(path)
+    return (st.st_dev, st.st_ino) != recorded

tests/run_agent/test_turn_completion_explainer.py

@@ -154,6 +154,17 @@ def test_explanation_persistence_corrupt_cause_never_says_free_space():
     assert "full disk" not in lower
 
 
+def test_explanation_persistence_replaced_cause_forbids_inplace_repair():
+    out = AIAgent._format_turn_completion_explanation(
+        "session_persistence_failed", "replaced"
+    )
+    lower = out.lower()
+    assert "replaced" in lower
+    assert "doctor --fix" in lower or "in-place" in lower
+    assert "free some space" not in lower
+    assert "full disk" not in lower
+
+
 def test_explanation_persistence_unknown_cause_is_neutral():
     """None/'unknown' cause must not claim disk-full — point at diagnostics."""
     for cause in (None, "unknown"):
@@ -305,6 +316,7 @@ def test_persistence_error_causes_tuple_matches_classifier():
         "Session 'abc' is being compressed by another writer",
         "Session turn lease lost; refusing transcript write for 'abc'",
         "database disk image is malformed",
+        "FATAL: state.db was replaced underneath the gateway",
         "database or disk is full",
         "something else entirely",
         None,