refactor(state): one read-only URI builder; probe a held store via snapshot only

read_only_db_uri() replaces four inline mode=ro URI sites (two of which
still used the raw f-string that truncates on ?/# in the home path:
state_db_has_structural_damage and collect_state_db_stats). The doctor
write probe now applies the live-holder gate in both modes: a quiet store
is probed in place as on main, a held store is probed through a read-only
snapshot, and a held store over 1 GB is skipped with an info line unless
--fix is given (the unconditional copy cost one full DB write per plain
doctor run). Connect/backup failures propagate to the existing
classification instead of being reported as FTS write-health failures.
Observational sessions commands print a migration hint instead of a raw
traceback when a read-only opener meets an older schema.

Co-authored-by: Ahmett101 <Ahmett101@users.noreply.github.com>

hermes_cli/doctor_state.py

@@ -10,7 +10,7 @@ from hermes_cli.doctor_report import (
     warn_on_error,
 )
 from hermes_cli.sizefmt import format_bytes as _human_bytes
-from hermes_state_common import FTS_STORAGE_VERSION
+from hermes_state_common import FTS_STORAGE_VERSION, read_only_db_uri
 
 
 def _honcho_is_configured_for_doctor() -> bool:
@@ -151,39 +151,40 @@ def _check_directory_structure(should_fix: bool, f: Finding) -> None:
 def _session_count(state_db_path: Path):
     import sqlite3
     # mode=ro: doctor is a reader; a writable open of a gateway-held WAL DB is the second-writer class (#103339).
-    # as_uri() percent-encodes '?' / '#' in the home path; a raw f-string URI truncates there.
-    conn = sqlite3.connect(Path(state_db_path).resolve().as_uri() + "?mode=ro", uri=True)
+    conn = sqlite3.connect(read_only_db_uri(state_db_path), uri=True)
     try:
         return conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
     finally:
         conn.close()
 
 
-def _write_health_reason(state_db_path: Path, *, isolate: bool):
-    """FTS/write-health probe. Isolated copies never join the live store WAL lifecycle."""
+# Above this the snapshot copy a held store needs costs more than the probe is worth; --fix still probes.
+_WRITE_PROBE_SNAPSHOT_MAX_BYTES = 1 << 30
+
+
+def _write_health_reason(state_db_path: Path, *, should_fix: bool):
+    """FTS/write-health probe (a rolled-back BEGIN IMMEDIATE). Against a store a live writer holds,
+    that probe is the second-writer class (#103339), so probe a read-only snapshot instead; a quiet
+    store is probed in place. Returns the failure reason, or None when healthy or skipped."""
     from hermes_state_repair import _connect_repair_durable, _db_opens_cleanly
     from hermes_state_holders import live_writer_holds_db
-    # Even under --fix the probe's BEGIN IMMEDIATE is a second writer against a gateway-held DB (#103339):
-    # only probe the live file when the holder scan proves it quiet, else fall back to a snapshot.
-    if not isolate and not live_writer_holds_db(state_db_path, connect_repair_durable=_connect_repair_durable):
+    if not live_writer_holds_db(state_db_path, connect_repair_durable=_connect_repair_durable):
         return _db_opens_cleanly(state_db_path)
+    if not should_fix and state_db_path.stat().st_size > _WRITE_PROBE_SNAPSHOT_MAX_BYTES:
+        check_info("state.db write-health probe skipped: store is held by a live writer and larger than 1 GB "
+                   "(run 'hermes doctor --fix' to probe it)")
+        return None
     import sqlite3
     import tempfile
     with tempfile.TemporaryDirectory() as tmp:
         snapshot = Path(tmp) / "state.db"
-        try:
-            # as_uri() percent-encodes '?' / '#' in the home path; a raw f-string URI truncates there.
-            src = sqlite3.connect(Path(state_db_path).resolve().as_uri() + "?mode=ro", uri=True, timeout=1.0)
-        except sqlite3.Error as exc:
-            return str(exc)
+        src = sqlite3.connect(read_only_db_uri(state_db_path), uri=True, timeout=1.0)
         try:
             dest = sqlite3.connect(str(snapshot))
             try:
                 src.backup(dest)
             finally:
                 dest.close()
-        except sqlite3.Error as exc:
-            return str(exc)
         finally:
             src.close()
         return _db_opens_cleanly(snapshot)
@@ -256,11 +257,10 @@ def _state_db_health(f: Finding, should_fix: bool, state_db_path: Path, _DHH: st
     from hermes_state_repair import state_db_has_structural_damage
     try:
         check_ok(f"{_DHH}/state.db exists ({_session_count(state_db_path)} sessions)")
+        # COUNT(*) succeeds even when the FTS index is corrupt and every write fails through the triggers.
+        _write_reason = _write_health_reason(state_db_path, should_fix=should_fix)
     except Exception as e:
         return _classify_unreadable_state_db(f, should_fix, state_db_path, _DHH, e)
-    # COUNT(*) succeeds even when the FTS index is corrupt and every write fails through the triggers.
-    # Non-fixing doctor snapshots first so the write probe cannot join the live WAL lifecycle (#50502).
-    _write_reason = _write_health_reason(state_db_path, isolate=not should_fix)
     if _write_reason is not None:
         if state_db_has_structural_damage(state_db_path):
             check_warn(f"{_DHH}/state.db has structural corruption (canonical tables/indexes damaged, "

hermes_cli/sessions_cmd.py

@@ -9,6 +9,7 @@ import — must run without opening ``SessionDB()``, which a malformed schema pr
 import json
 import os
 import shutil
+import sqlite3
 import sys
 from functools import partial
 from pathlib import Path
@@ -994,6 +995,13 @@ def cmd_sessions(args, sessions_parser=None):
         if handler is None:
             sessions_parser.print_help()
             return
-        return handler(db, args)
+        try:
+            return handler(db, args)
+        except sqlite3.OperationalError as e:
+            if not observational:
+                raise
+            # A read-only opener skips schema migration, so a store from an older release can lack a column.
+            print(f"Error: session database needs migration — run any writing hermes command first ({e})")
+            return 1
     finally:
         db.close()

hermes_state.py

@@ -27,7 +27,9 @@ from pathlib import Path
 from hermes_constants import get_hermes_home, mkdir_under_hermes_home
 from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TypeVar, cast
 
-from hermes_state_common import escape_like as _escape_like, stat_db_file_identity as _stat_db_file_identity
+from hermes_state_common import (
+    escape_like as _escape_like, read_only_db_uri, stat_db_file_identity as _stat_db_file_identity,
+)
 from hermes_state_errors import (
     _DELETED_WAL_GENERATION_MSG, _DISK_IO_ERROR_MARKER, _STATE_DB_CORRUPT_MSG, _STATE_DB_GENERATION_KEY,
     _STATE_DB_REPLACED_MSG, DeletedWalGenerationError, SessionCompressionInProgressError, StateDbCorruptError,
@@ -642,9 +644,8 @@ class SessionDB(
     def _connect_read_only(self, timeout: float) -> sqlite3.Connection:
         """``mode=ro`` tracked connection with Row factory. check_same_thread=False: pooled connections
         are borrowed by whichever thread reads next; exclusive ownership is enforced by pool checkout."""
-        # as_uri() percent-encodes '?' / '#' in the home path; a raw f-string URI truncates there.
         conn = _connect_tracked_db(
-            Path(self.db_path).resolve().as_uri() + "?mode=ro", tracking_path=self.db_path, uri=True,
+            read_only_db_uri(self.db_path), tracking_path=self.db_path, uri=True,
             check_same_thread=False, timeout=timeout, isolation_level=None,
         )
         conn.row_factory = sqlite3.Row

hermes_state_common.py

@@ -8,12 +8,19 @@ import logging
 import os
 import sys
 import time
+from pathlib import Path
 from typing import Any
 
 from agent.skill_commands import SKILL_EXCERPT_JOINT, SKILL_SCAFFOLD_SQL_LIKE, describe_skill_invocation
 from agent.context_compressor import (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX, _MERGED_PRIOR_CONTEXT_HEADER,
     _MERGED_SUMMARY_DELIMITER, _SUMMARY_END_MARKER)
 
+def read_only_db_uri(db_path) -> str:
+    """``file:`` URI for a ``mode=ro`` open. ``as_uri()`` percent-encodes ``?``/``#`` in the home
+    path; a raw ``f"file:{path}?mode=ro"`` truncates there and opens the wrong (empty) database."""
+    return Path(db_path).resolve().as_uri() + "?mode=ro"
+
+
 
 # Session preview = head of the first user message (shown when a session has no title).  A /skill invocation
 # embeds the whole skill body, so scaffolded rows take a wider excerpt (whole message under budget, else head +

hermes_state_dbfile.py

@@ -26,6 +26,7 @@ from typing import Any, Callable, Dict, List, Optional, Tuple
 
 from hermes_state_holders import canonical_sqlite_path
 from hermes_state_common import (
+    read_only_db_uri,
     FTS_REBUILD_DEFERRAL_KEY, stat_db_file_identity as _stat_db_file_identity
 )
 
@@ -716,7 +717,7 @@ def collect_state_db_stats(db_path: Path) -> Dict[str, Any]:
     try:
         # A short timeout keeps doctor snappy when a writer holds the lock.  The tracked connect
         # lets byte-probe helpers see this connection and refuse raw opens that would cancel locks.
-        conn = _connect_tracked_db(f"file:{Path(db_path)}?mode=ro", tracking_path=Path(db_path),
+        conn = _connect_tracked_db(read_only_db_uri(db_path), tracking_path=Path(db_path),
                                    uri=True, timeout=2.0)
     except Exception as exc:
         logger.debug("collect_state_db_stats: cannot open %s read-only: %s", db_path, exc)

hermes_state_repair.py

@@ -23,6 +23,7 @@ from typing import Any, Dict, List, Optional, Tuple
 from hermes_constants import get_hermes_home
 from hermes_startup_watchdog import report_startup_progress
 from hermes_state_common import (
+    read_only_db_uri,
     _acquire_db_flock, _clear_lock_holder_record, _describe_lock_holder, _read_lock_holder_record,
     is_advisory_lock_contention,
 )
@@ -727,7 +728,7 @@ def state_db_has_structural_damage(db_path: Path) -> bool:
     while ``messages``/``sessions`` read cleanly, and the FTS rebuild ladder cannot help.
     Cannot-open / locked stays False so the caller keeps the FTS path."""
     try:
-        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=1.0)
+        conn = sqlite3.connect(read_only_db_uri(db_path), uri=True, timeout=1.0)
     except sqlite3.Error:
         return False
     try:

tests/hermes_cli/test_observational_sessiondb_modes.py

@@ -39,3 +39,25 @@ def test_sessions_observational_commands_on_missing_store_stay_empty(monkeypatch
     assert sessions_cmd.cmd_sessions(pinned_args) is None
     assert "No pinned sessions" in capsys.readouterr().out
     assert not db_path.exists()
+
+
+def test_doctor_write_probe_never_touches_a_store_a_live_writer_holds(monkeypatch, tmp_path):
+    """The write probe runs against a read-only snapshot when a gateway holds state.db, in place otherwise."""
+    import sqlite3
+
+    import hermes_state_holders
+    import hermes_state_repair
+    from hermes_cli import doctor_state
+
+    state_db = tmp_path / "state.db"
+    sqlite3.connect(state_db).execute("CREATE TABLE sessions (id TEXT)").connection.close()
+    probed: list = []
+    monkeypatch.setattr(hermes_state_repair, "_db_opens_cleanly", lambda path: probed.append(path))
+
+    monkeypatch.setattr(hermes_state_holders, "live_writer_holds_db", lambda *_a, **_k: True)
+    assert doctor_state._write_health_reason(state_db, should_fix=False) is None
+    monkeypatch.setattr(hermes_state_holders, "live_writer_holds_db", lambda *_a, **_k: False)
+    assert doctor_state._write_health_reason(state_db, should_fix=False) is None
+
+    assert probed[0] != state_db and probed[0].name == "state.db"
+    assert probed[1] == state_db