fix(state): v25 prompt dedupe degrades gracefully on a contended DB

Only the initial SELECT of _dedupe_legacy_system_prompts was guarded;
a 'database is locked' on any per-row write propagated out, aborted
schema init, left the schema version below 25, and made every later
SessionDB.__init__ re-enter the same migration against the same
contended DB - the second half of the enterprise crash-loop report.

The per-row loop now catches OperationalError, logs once, and returns.
Partial migration is safe by design: the legacy system_prompt column
is the documented read fallback for unmigrated rows, and the next
schema init resumes where the contention stopped. Tests prove rows
migrated before the failure stay migrated, the remainder stays
readable, and a later run completes it.

hermes_state_schema.py

@@ -86,7 +86,18 @@ class SessionSchemaMixin:
     """See module docstring — mixin for SessionDB (Schema cluster)."""
 
     def _dedupe_legacy_system_prompts(self, cursor: sqlite3.Cursor) -> None:
-        """Move inline prompt snapshots into the shared content-addressed table."""
+        """Move inline prompt snapshots into the shared content-addressed table.
+
+        Contention-safe by design: a ``database is locked`` (or any other
+        ``OperationalError``) mid-loop returns instead of raising. Partial
+        migration is safe — the legacy ``system_prompt`` column is kept as a
+        read fallback for unmigrated rows, and the next schema init picks up
+        the remainder. Letting the error propagate aborted schema init
+        entirely, left the version below 25, and made every subsequent
+        ``SessionDB.__init__`` re-enter this migration against the same
+        contended DB (Coatue field report, 2026-08-14: gateway watchdog
+        crash loop).
+        """
         try:
             rows = cursor.execute(
                 "SELECT id, system_prompt FROM sessions "
@@ -98,13 +109,22 @@ class SessionSchemaMixin:
         for row in rows:
             session_id = row["id"] if isinstance(row, sqlite3.Row) else row[0]
             prompt = row["system_prompt"] if isinstance(row, sqlite3.Row) else row[1]
-            prompt_hash = self._store_system_prompt(cursor, prompt)
-            cursor.execute(
-                "UPDATE sessions "
-                "SET system_prompt_hash = ?, system_prompt = NULL "
-                "WHERE id = ?",
-                (prompt_hash, session_id),
-            )
+            try:
+                prompt_hash = self._store_system_prompt(cursor, prompt)
+                cursor.execute(
+                    "UPDATE sessions "
+                    "SET system_prompt_hash = ?, system_prompt = NULL "
+                    "WHERE id = ?",
+                    (prompt_hash, session_id),
+                )
+            except sqlite3.OperationalError as exc:
+                logger.warning(
+                    "v25 prompt dedupe paused after contention (%s); "
+                    "unmigrated rows keep the legacy inline prompt and the "
+                    "next schema init resumes the migration.",
+                    exc,
+                )
+                return
 
     def _sqlite_supports_fts5(self, cursor: sqlite3.Cursor) -> bool:
         try:

tests/state/test_dedupe_migration_contention.py

@@ -0,0 +1,120 @@
+"""The v25 dedupe migration must degrade gracefully on a contended DB.
+
+Coatue field report (2026-08-14): with state.db locked by another process,
+``_dedupe_legacy_system_prompts`` raised ``sqlite3.OperationalError``
+mid-loop (only the initial SELECT was guarded), which aborted schema init,
+left the schema version below 25, and made EVERY subsequent
+``SessionDB.__init__`` re-enter the same blocking migration — the fuel of
+the gateway's watchdog crash loop.
+
+Contract:
+- A write failure mid-migration returns gracefully (no raise). Partial
+  migration is safe by design: the legacy ``system_prompt`` column is kept
+  as a read fallback for unmigrated rows.
+- Rows migrated before the failure stay migrated; unmigrated rows remain
+  readable and are picked up by a later successful run.
+"""
+
+import sqlite3
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+def _make_legacy_db(tmp_path, n_rows=5):
+    """Open a real SessionDB, then regress it to a pre-v25 shape."""
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+    db.close()
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    cur = conn.cursor()
+    for i in range(n_rows):
+        cur.execute(
+            "INSERT OR IGNORE INTO sessions (id, source, started_at) "
+            "VALUES (?, 'test', 1.0)",
+            (f"sess-{i}",),
+        )
+        cur.execute(
+            "UPDATE sessions SET system_prompt = ?, system_prompt_hash = NULL "
+            "WHERE id = ?",
+            (f"legacy prompt {i}", f"sess-{i}"),
+        )
+    conn.commit()
+    inserted = cur.execute(
+        "SELECT COUNT(*) FROM sessions WHERE id LIKE 'sess-%'"
+    ).fetchone()[0]
+    conn.close()
+    assert inserted == n_rows, f"fixture only created {inserted}/{n_rows} rows"
+    return db_path
+
+
+class _FailAfterN:
+    """Cursor proxy: UPDATE statements start failing after N successes."""
+
+    def __init__(self, cursor, fail_after):
+        self._cursor = cursor
+        self._updates = 0
+        self._fail_after = fail_after
+
+    def execute(self, sql, *args, **kwargs):
+        if sql.lstrip().upper().startswith("UPDATE"):
+            if self._updates >= self._fail_after:
+                raise sqlite3.OperationalError("database is locked")
+            self._updates += 1
+        return self._cursor.execute(sql, *args, **kwargs)
+
+    def __getattr__(self, name):
+        return getattr(self._cursor, name)
+
+
+def test_mid_loop_lock_error_returns_instead_of_raising(tmp_path):
+    db_path = _make_legacy_db(tmp_path)
+    db = SessionDB(db_path=db_path)
+    try:
+        conn = sqlite3.connect(db_path)
+        conn.row_factory = sqlite3.Row
+        raw = conn.cursor()
+        proxy = _FailAfterN(raw, fail_after=2)
+        # Must NOT raise even though the third UPDATE hits "database is locked".
+        db._dedupe_legacy_system_prompts(proxy)
+        conn.commit()
+
+        rows = raw.execute(
+            "SELECT id, system_prompt, system_prompt_hash FROM sessions "
+            "WHERE id LIKE 'sess-%' ORDER BY id"
+        ).fetchall()
+        migrated = [r for r in rows if r["system_prompt"] is None]
+        legacy = [r for r in rows if r["system_prompt"] is not None]
+        assert migrated, "no rows migrated before the simulated lock"
+        assert legacy, "expected unmigrated remainder after the failure"
+        # Migrated rows carry a hash; legacy rows keep their readable prompt.
+        assert all(r["system_prompt_hash"] for r in migrated)
+        assert all(r["system_prompt"].startswith("legacy prompt") for r in legacy)
+        conn.close()
+    finally:
+        db.close()
+
+
+def test_later_run_completes_the_remainder(tmp_path):
+    db_path = _make_legacy_db(tmp_path)
+    db = SessionDB(db_path=db_path)
+    try:
+        conn = sqlite3.connect(db_path)
+        conn.row_factory = sqlite3.Row
+        raw = conn.cursor()
+        db._dedupe_legacy_system_prompts(_FailAfterN(raw, fail_after=2))
+        conn.commit()
+        # Second run with no failures finishes the job.
+        db._dedupe_legacy_system_prompts(raw)
+        conn.commit()
+        remaining = raw.execute(
+            "SELECT COUNT(*) FROM sessions "
+            "WHERE id LIKE 'sess-%' AND system_prompt IS NOT NULL"
+        ).fetchone()[0]
+        assert remaining == 0
+        conn.close()
+    finally:
+        db.close()