fix(state): refuse checkpoint and repair on unprovable live-writer state (#103339)

hermes_cli/doctor_state.py

@@ -243,10 +243,35 @@ def _state_db_wal(f: Finding, should_fix: bool, state_db_path: Path) -> None:
             check_warn(f"WAL file is large ({size // (1024*1024)} MB)", "(may indicate missed checkpoints)")
             if not should_fix:
                 return f.issues.append("Large WAL file — run 'hermes doctor --fix' to checkpoint")
-            import sqlite3
-            conn = sqlite3.connect(str(state_db_path))
-            conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
-            conn.close()
+            # Checkpoint-lock premise (#40177): a bare connect runs WAL recovery and the checkpoint joins the
+            # live WAL — under a running gateway that second-writer handling corrupts state.db. Skip instead.
+            from hermes_state_holders import live_writer_holds_db
+            from hermes_state_repair import _connect_repair_durable, _exclusive_repair_db_guard
+
+            def _skip_unquiet() -> None:
+                # Honest disjunction: a refusal here means "held OR
+                # unprovable" — the DatabaseError lane fires when SQLite
+                # cannot open the file at all, with nobody holding it. Never
+                # assert a live writer as fact.
+                check_warn("WAL checkpoint skipped: cannot prove state.db is quiet",
+                           "(a live writer holds it, or it is unreadable — stop the profile's gateway "
+                           "and re-run 'hermes doctor --fix')")
+                f.issues.append("Large WAL file — cannot prove state.db is quiet (stop the profile's "
+                                "gateway first, then re-run 'hermes doctor --fix' to checkpoint)")
+
+            if live_writer_holds_db(state_db_path, connect_repair_durable=_connect_repair_durable):
+                _skip_unquiet()
+                return
+            # Lifetime: the preflight above is check-then-act by itself — a
+            # writer entering between it and the checkpoint would still race.
+            # Run the checkpoint ON the exclusive-guard connection so
+            # authority is held across the operation, not just probed before
+            # it (same contract as repair surgery).
+            with _exclusive_repair_db_guard(state_db_path) as (guard, _guard_error):
+                if guard is None:
+                    _skip_unquiet()
+                    return
+                guard.execute("PRAGMA wal_checkpoint(PASSIVE)")
             check_ok(f"WAL checkpoint performed ({size // 1024}K → {wal_size() // 1024}K)")
             f.fixed += 1
         elif size > 10 * 1024 * 1024:  # 10 MB

tests/hermes_cli/test_doctor_wal_holder_guard.py

@@ -0,0 +1,141 @@
+"""Regression: ``hermes doctor --fix`` must not checkpoint the live WAL under a running gateway.
+
+Checkpoint-lock premise (#40177): a bare ``sqlite3.connect`` runs WAL recovery and
+``PRAGMA wal_checkpoint(PASSIVE)`` joins the live WAL — that second-writer handling
+on a gateway-held state.db is the corruption class #103339 tracks. The check must
+skip with an actionable finding while a live writer holds the database.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import subprocess
+import sys
+from pathlib import Path
+
+from hermes_cli.doctor_report import Finding
+from hermes_cli.doctor_state import _state_db_wal
+
+
+# NOTE: no ``requires_wal`` marker here on purpose. That gate exists for tests
+# that depend on Hermes *choosing* WAL mode (declined on vulnerable SQLite
+# builds). This test forces WAL explicitly through raw SQL and asserts only on
+# the holder-guard skip, so the probe mechanics work on any build.
+def test_wal_checkpoint_skipped_while_live_writer_holds_db(tmp_path):
+    """A held database skips the checkpoint; nothing is checkpointed or fixed."""
+    db = tmp_path / "state.db"
+    setup = sqlite3.connect(str(db))
+    try:
+        setup.execute("CREATE TABLE t(x)")
+        setup.execute("PRAGMA journal_mode=WAL")
+        setup.execute("INSERT INTO t VALUES (1)")
+        setup.commit()
+    finally:
+        setup.close()
+    holder = sqlite3.connect(str(db))
+    holder.execute("SELECT count(*) FROM t").fetchone()
+    try:
+        wal = Path(f"{db}-wal")
+        assert wal.exists()
+        # Push past the 50 MB fix threshold without 50 MB of real frames: the
+        # guard runs before any WAL byte is parsed, so padding is never read.
+        with open(wal, "ab") as handle:
+            handle.truncate(51 * 1024 * 1024)
+        finding = Finding()
+        _state_db_wal(finding, True, db)
+    finally:
+        try:
+            holder.close()
+        except Exception:
+            pass
+
+    assert finding.fixed == 0
+    assert any("gateway" in issue for issue in finding.issues)
+
+
+def _make_padded_wal_db(tmp_path):
+    """A WAL db past the 50 MB fix threshold with zero open connections.
+
+    The writer subprocess exits via ``os._exit`` (no clean close), so its
+    committed frames stay in the -wal instead of being auto-checkpointed
+    away on last-close — the crash-recovery shape ``doctor --fix`` exists
+    for. Pure-sqlite3 child: no Hermes imports, no HERMES_HOME involved.
+    """
+    db = tmp_path / "state.db"
+    writer_code = (
+        "import os, sqlite3, sys;"
+        "c = sqlite3.connect(sys.argv[1]);"
+        "c.execute('CREATE TABLE t(x)');"
+        "c.execute('PRAGMA journal_mode=WAL');"
+        "c.execute('INSERT INTO t VALUES (1)');"
+        "c.commit();"
+        "os._exit(0)"
+    )
+    proc = subprocess.run([sys.executable, "-c", writer_code, str(db)], timeout=60)
+    assert proc.returncode == 0
+    wal = Path(f"{db}-wal")
+    assert wal.exists()
+    # Push past the 50 MB fix threshold without 50 MB of real frames: the
+    # guard runs before any WAL byte is parsed, so padding is never read.
+    with open(wal, "ab") as handle:
+        handle.truncate(51 * 1024 * 1024)
+    return db
+
+
+def test_wal_checkpoint_refuses_unprovable_operational_error(tmp_path, monkeypatch):
+    """A non-lock OperationalError is unprovable, not quiet: skip, fix nothing."""
+    import hermes_state_repair
+
+    db = _make_padded_wal_db(tmp_path)
+
+    def _raise_io_error(*_args, **_kwargs):
+        raise sqlite3.OperationalError("disk I/O error")
+
+    # Patched at the source: _state_db_wal late-imports this name at call time.
+    monkeypatch.setattr(hermes_state_repair, "_connect_repair_durable", _raise_io_error)
+
+    finding = Finding()
+    _state_db_wal(finding, True, db)
+
+    assert finding.fixed == 0
+    assert any("gateway" in issue for issue in finding.issues)
+
+
+def test_wal_checkpoint_refuses_writer_admitted_after_quiet_probe(tmp_path, monkeypatch):
+    """Quiet preflight + writer entering before the checkpoint still refuses.
+
+    Deterministic interleaving for the check-then-act race: the preflight
+    proves quiet, then guard acquisition fails (a writer took exclusion
+    first). The checkpoint must not run and nothing is reported fixed.
+    """
+    import contextlib
+
+    import hermes_state_holders
+    import hermes_state_repair
+
+    db = _make_padded_wal_db(tmp_path)
+    # Patched at the source modules: _state_db_wal late-imports both names
+    # at call time, so module-attribute patches take effect.
+    monkeypatch.setattr(hermes_state_holders, "live_writer_holds_db", lambda *_a, **_k: False)
+
+    @contextlib.contextmanager
+    def _contended_guard(_path):
+        yield None, sqlite3.OperationalError("database is locked")
+
+    monkeypatch.setattr(hermes_state_repair, "_exclusive_repair_db_guard", _contended_guard)
+
+    finding = Finding()
+    _state_db_wal(finding, True, db)
+
+    assert finding.fixed == 0
+    assert any("gateway" in issue for issue in finding.issues)
+
+
+def test_wal_checkpoint_runs_once_quiet_and_guarded(tmp_path):
+    """A quiet database checkpoints through the held guard and reports fixed."""
+    db = _make_padded_wal_db(tmp_path)
+
+    finding = Finding()
+    _state_db_wal(finding, True, db)
+
+    assert finding.fixed == 1

tests/test_state_db_repair_live_writer_guard.py

@@ -399,3 +399,40 @@ def test_repair_proceeds_once_the_database_is_quiescent(tmp_path):
     report = repair_state_db_schema(db, backup=False)
 
     assert "live writer" not in (report["error"] or "").lower()
+
+
+@pytest.mark.parametrize(
+    "probe_error",
+    [
+        sqlite3.DatabaseError("file is not a database"),
+        sqlite3.OperationalError("unable to open database file"),
+        sqlite3.OperationalError("disk I/O error"),
+    ],
+)
+def test_repair_refuses_when_guard_cannot_open(tmp_path, monkeypatch, probe_error):
+    """Surgery never runs when exclusion cannot be proven (#103339).
+
+    The preflight probe answers quiet on this database, but the
+    exclusive guard fails on the same fault and repair degrades to an
+    honest skip — never surgery under an unseen holder.
+    """
+    db = _make_wal_db(tmp_path)
+    # Real damage (not a stubbed probe): drop a canonical table so the
+    # healthy fast-path (already_healthy) cannot fire and the flow must
+    # reach the exclusive guard.
+    damaged = sqlite3.connect(str(db))
+    try:
+        damaged.execute("DROP TABLE sessions")
+        damaged.commit()
+    finally:
+        damaged.close()
+
+    def _raise_guard_error(*_args, **_kwargs):
+        raise probe_error
+
+    monkeypatch.setattr(hermes_state_repair, "_open_exclusive", _raise_guard_error)
+
+    report = repair_state_db_schema(db, backup=False)
+
+    assert report["repaired"] is False
+    assert "exclusive" in (report["error"] or "").lower()