fix(sessions): close the snapshot check/use race and guard damaged state_meta

hermes_cli/session_recovery.py

@@ -239,34 +239,33 @@ def _disk_space_preflight(
 def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[str]]:
     """Copy the source DB bundle aside so SQLite never opens the original.
 
-    Refuses when a connection to *source* is live in this process. Copying a
-    database file is an ``open()``/``close()`` on it, and ``close()`` cancels
-    every POSIX advisory lock the process holds on that file -- including a
-    running VACUUM's EXCLUSIVE lock (see ``hermes_cli.sqlite_safe_read``).
+    The whole copy runs inside ``offline_file_access``, which holds the
+    connection-lifecycle lock for its duration. Checking for a live connection
+    and *then* copying would be a check/use race: a connection could open in
+    that window, and the copy's ``close()`` would cancel its POSIX advisory
+    locks -- the failure class ``hermes_cli.sqlite_safe_read`` exists to
+    prevent (see #71724). Holding the lock means no connection can appear
+    mid-copy, across the main file and every sidecar.
+
     Recovery normally runs as its own short-lived CLI process against an
-    offline/quarantined file, so this should never fire; the check keeps this
-    path consistent with ``hermes_state._backup_db_file``, which refuses the
-    same situation, rather than leaving two policies for one hazard.
+    offline/quarantined file, so the refusal should never fire; the guard
+    keeps this path consistent with ``hermes_state._backup_db_file``.
     """
-    from hermes_cli.sqlite_safe_read import has_live_connection
-
-    if has_live_connection(source):
-        raise SessionRecoverySafetyError(
-            f"Refusing to snapshot {source}: a connection to it is still open "
-            "in this process, and copying the file would cancel that "
-            "connection's POSIX locks. Close all database handles (stop the "
-            "gateway/dashboard) and re-run."
-        )
+    from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access
 
     snapshot_source = snapshot_dir / source.name
     copied: list[str] = []
-    for suffix in _SIDECAR_SUFFIXES:
-        source_part = _sidecar_path(source, suffix)
-        if not source_part.exists():
-            continue
-        destination_part = _sidecar_path(snapshot_source, suffix)
-        shutil.copy2(source_part, destination_part)
-        copied.append(destination_part.name)
+    try:
+        with offline_file_access(source, what="snapshot"):
+            for suffix in _SIDECAR_SUFFIXES:
+                source_part = _sidecar_path(source, suffix)
+                if not source_part.exists():
+                    continue
+                destination_part = _sidecar_path(snapshot_source, suffix)
+                shutil.copy2(source_part, destination_part)
+                copied.append(destination_part.name)
+    except LiveConnectionError as exc:
+        raise SessionRecoverySafetyError(str(exc)) from exc
     return snapshot_source, copied
 
 
@@ -751,7 +750,59 @@ def _copy_state_meta_salvage(
     progress_cb: Optional[ProgressCallback],
     source_rows: Optional[int],
 ) -> dict[str, Any]:
-    """Salvage readable user metadata while regenerating derived FTS state."""
+    """Salvage readable user metadata while regenerating derived FTS state.
+
+    Requires both ``key`` and ``value``, matching the non-partial
+    :func:`_copy_state_meta`. A damaged ``state_meta`` can retain one column
+    and lose the other; without this check a missing ``key`` raised
+    ``ValueError`` from ``columns.index("key")`` and aborted the entire
+    partial recovery, and a missing ``value`` would have copied key-only rows
+    while reporting the table complete.
+
+    Status matters here. An unusable-but-PRESENT table reports ``failed``, not
+    ``missing``: verification only escalates ``failed``/``partial`` into a
+    warning + ``loss_detected``, so reporting ``missing`` would silently drop
+    real metadata and still claim ``complete=True``. ``missing`` is reserved
+    for a table that genuinely is not there. Either way ``state_meta`` is
+    optional, so ``--allow-partial`` records the loss and carries on
+    recovering sessions and messages.
+    """
+    source_columns = _table_columns(source, "state_meta")
+    destination_columns = _table_columns(destination, "state_meta")
+    if not source_columns:
+        # Genuinely absent from the source — nothing was lost.
+        return {
+            "mode": "rowid_range_salvage",
+            "source_meta_rows": source_rows,
+            "copied_rows": 0,
+            "columns": ["key", "value"],
+            "excluded_keys": sorted(_GENERATED_META_KEYS),
+            "status": "missing",
+        }
+    if not {"key", "value"}.issubset(source_columns):
+        # Present but unusable: this IS data loss and must be reported.
+        return {
+            "mode": "rowid_range_salvage",
+            "source_meta_rows": source_rows,
+            "copied_rows": 0,
+            "columns": ["key", "value"],
+            "excluded_keys": sorted(_GENERATED_META_KEYS),
+            "status": "failed",
+            "error": (
+                "source state_meta exists but is missing the key/value "
+                f"columns (found: {', '.join(source_columns) or 'none'})"
+            ),
+        }
+    if not {"key", "value"}.issubset(destination_columns):
+        return {
+            "mode": "rowid_range_salvage",
+            "source_meta_rows": source_rows,
+            "copied_rows": 0,
+            "columns": ["key", "value"],
+            "excluded_keys": sorted(_GENERATED_META_KEYS),
+            "status": "failed",
+            "error": "destination state_meta schema is incomplete",
+        }
 
     def keep_user_meta(
         row: tuple[Any, ...],

hermes_cli/sqlite_safe_read.py

@@ -61,6 +61,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import os
 import sqlite3
@@ -372,3 +373,37 @@ def read_header_bytes_preopen(
                 return handle.read(length)
         except OSError:
             return None
+
+
+class LiveConnectionError(RuntimeError):
+    """A raw file operation was attempted on a database with live connections."""
+
+
+@contextlib.contextmanager
+def offline_file_access(path: Path | str, *, what: str = "read"):
+    """Hold the connection-lifecycle lock across a raw read of a database file.
+
+    Checking :func:`has_live_connection` and *then* doing the raw I/O is a
+    check/use race: a connection can be opened in the window between the two,
+    and the raw ``close()`` will cancel its POSIX advisory locks — the exact
+    failure class the registry exists to prevent. Any multi-step raw access
+    (copying a database plus its ``-wal``/``-shm``/``-journal`` sidecars,
+    hashing a file, moving a bundle aside) must therefore run *inside* this
+    context manager rather than after a bare check.
+
+    While held, :func:`connect_tracked` blocks, so no new connection can
+    appear mid-copy. Raises :class:`LiveConnectionError` if a connection is
+    already live when the guard is entered.
+
+    The lock is only held for the duration of the raw I/O; it never spans
+    caller work on an open connection, so it does not serialise database use.
+    """
+    with _live_lock:
+        if _key(path) in _live_connections:
+            raise LiveConnectionError(
+                f"Refusing to {what} {path}: a connection to it is still open "
+                "in this process, and raw file access would cancel that "
+                "connection's POSIX advisory locks. Close all database "
+                "handles (stop the gateway/dashboard) and retry."
+            )
+        yield

tests/hermes_cli/test_session_recovery.py

@@ -288,6 +288,222 @@ def _corrupt_middle_table_leaf(
     return leaf_page
 
 
+def test_snapshot_blocks_connections_opened_during_the_copy(
+    tmp_path: Path,
+) -> None:
+    """A connection must not be able to open while raw copy descriptors exist.
+
+    Checking has_live_connection() and then copying leaves a window: a
+    connection can open between the two, and the copy's close() cancels its
+    POSIX advisory locks. The guard must hold the lifecycle lock across the
+    whole bundle copy.
+
+    Runs the copy in a worker thread and pauses it inside the patched copy, so
+    the assertion is about lock ordering rather than which thread the
+    scheduler happens to resume first: while the copy is parked, a
+    connect_tracked() attempt must NOT complete; once released, it must.
+    """
+    import threading
+
+    from hermes_cli import session_recovery as recovery_module
+    from hermes_cli.sqlite_safe_read import connect_tracked
+
+    source = tmp_path / "racy-state.db"
+    snapshot_dir = tmp_path / "snapshot"
+    snapshot_dir.mkdir()
+    _make_source(source)
+
+    inside_copy = threading.Event()
+    release_copy = threading.Event()
+    connect_attempted = threading.Event()
+    connection_opened = threading.Event()
+    errors: list[str] = []
+    real_copy2 = recovery_module.shutil.copy2
+
+    def slow_copy2(src, dst, *args, **kwargs):
+        result = real_copy2(src, dst, *args, **kwargs)
+        if str(src).endswith("racy-state.db"):
+            inside_copy.set()
+            release_copy.wait(30)
+        return result
+
+    def do_copy():
+        try:
+            recovery_module._copy_source_bundle(source, snapshot_dir)
+        except Exception as exc:  # pragma: no cover - surfaced via errors
+            errors.append(f"copy failed: {exc}")
+
+    def do_connect():
+        # Signal immediately before the blocking call so a timed "still
+        # blocked" assertion cannot pass merely because this thread had not
+        # been scheduled yet.
+        connect_attempted.set()
+        try:
+            conn = connect_tracked(source, isolation_level=None, timeout=30.0)
+            connection_opened.set()
+            conn.close()
+        except Exception as exc:  # pragma: no cover - surfaced via errors
+            errors.append(f"connect failed: {exc}")
+
+    recovery_module.shutil.copy2 = slow_copy2
+    copier = threading.Thread(target=do_copy, daemon=True)
+    connector = threading.Thread(target=do_connect, daemon=True)
+    try:
+        copier.start()
+        assert inside_copy.wait(30), "copy never reached the patched operation"
+
+        connector.start()
+        assert connect_attempted.wait(30), "connector thread never started"
+        # The connector is at the lock. While the copy holds it, the
+        # connection must not open.
+        assert not connection_opened.wait(1.0), (
+            "connect_tracked() completed while raw copy descriptors were open "
+            "— the guard is not holding the lifecycle lock across the copy"
+        )
+
+        release_copy.set()
+        # Once the copy finishes and releases the lock, it must open promptly.
+        assert connection_opened.wait(30), (
+            "connect_tracked() never completed after the copy released the lock"
+        )
+    finally:
+        release_copy.set()
+        recovery_module.shutil.copy2 = real_copy2
+        copier.join(30)
+        connector.join(30)
+
+    assert not errors, errors[0]
+
+
+def test_partial_recovery_reports_damaged_state_meta_as_loss(
+    tmp_path: Path,
+) -> None:
+    """A damaged optional state_meta must degrade AND be reported as loss.
+
+    state_meta can lose ``key`` while ``value`` stays readable. Indexing
+    ``key`` unconditionally raised ValueError and killed the whole partial
+    recovery. Reporting it as ``missing`` instead is just as wrong in the
+    other direction: verification only escalates ``failed``/``partial``, so
+    the run would claim ``complete=True`` after silently dropping real
+    metadata. A present-but-unusable table must be ``failed``, surface a
+    warning, and mark the output partial — while sessions and messages still
+    recover.
+    """
+    source = tmp_path / "meta-damaged.db"
+    output = tmp_path / "meta-recovered.db"
+    expected = _make_source(source)
+
+    conn = sqlite3.connect(str(source), isolation_level=None)
+    try:
+        conn.execute("DROP TABLE IF EXISTS state_meta")
+        conn.execute("CREATE TABLE state_meta(value TEXT)")
+        conn.execute("INSERT INTO state_meta(value) VALUES ('orphaned')")
+    finally:
+        conn.close()
+
+    report = recover_session_database(
+        source,
+        output,
+        work_dir=tmp_path,
+        chunk_size=4,
+        allow_partial=True,
+    )
+
+    # The table existed and could not be salvaged: that is loss, not absence.
+    assert report["copy"]["state_meta"]["status"] == "failed"
+    assert any(
+        "state_meta" in warning for warning in report["verification"]["warnings"]
+    ), report["verification"]["warnings"]
+    assert report["verification"]["loss_detected"] is True
+    assert report["partial"] is True
+    assert report["complete"] is False
+
+    # Structurally sound, so still installable-with-review.
+    assert report["verified"] is True
+    assert report["verification"]["healthy"] is True
+    assert report["installed"] is False
+
+    # The canonical data still recovers.
+    assert report["verification"]["table_counts"]["sessions"] == expected["sessions"]
+    assert report["verification"]["table_counts"]["messages"] == expected["messages"]
+    assert report["verification"]["integrity_check"] == ["ok"]
+
+
+def test_partial_recovery_treats_absent_state_meta_as_no_loss(
+    tmp_path: Path,
+) -> None:
+    """A genuinely absent state_meta is not data loss and must not warn."""
+    source = tmp_path / "meta-absent.db"
+    output = tmp_path / "meta-absent-recovered.db"
+    _make_source(source)
+
+    conn = sqlite3.connect(str(source), isolation_level=None)
+    try:
+        conn.execute("DROP TABLE IF EXISTS state_meta")
+    finally:
+        conn.close()
+
+    report = recover_session_database(
+        source,
+        output,
+        work_dir=tmp_path,
+        chunk_size=4,
+        allow_partial=True,
+    )
+
+    assert report["copy"]["state_meta"]["status"] == "missing"
+    assert not any(
+        "state_meta" in warning for warning in report["verification"]["warnings"]
+    ), report["verification"]["warnings"]
+    assert report["verification"]["integrity_check"] == ["ok"]
+
+
+def test_state_meta_salvage_distinguishes_absent_from_unusable(
+    tmp_path: Path,
+) -> None:
+    """Unit-level: the salvage helper must not conflate absent with damaged.
+
+    ``recover_session_database`` short-circuits on the inspection result when
+    state_meta is entirely absent, so the helper's own absent-branch is never
+    reached through the public path — a regression there would be invisible
+    end-to-end. Exercise it directly so both statuses are pinned:
+    absent -> ``missing`` (no loss), present-but-unusable -> ``failed``
+    (loss, which verification escalates into a warning).
+    """
+    destination_path = tmp_path / "dest.db"
+    destination = sqlite3.connect(str(destination_path), isolation_level=None)
+    destination.execute("CREATE TABLE state_meta(key TEXT PRIMARY KEY, value TEXT)")
+
+    absent_path = tmp_path / "absent.db"
+    absent = sqlite3.connect(str(absent_path), isolation_level=None)
+    absent.execute("CREATE TABLE unrelated(x)")
+
+    damaged_path = tmp_path / "damaged.db"
+    damaged = sqlite3.connect(str(damaged_path), isolation_level=None)
+    damaged.execute("CREATE TABLE state_meta(value TEXT)")
+    damaged.execute("INSERT INTO state_meta(value) VALUES ('orphaned')")
+
+    try:
+        absent_result = session_recovery._copy_state_meta_salvage(
+            absent, destination, chunk_size=4, progress_cb=None, source_rows=0
+        )
+        assert absent_result["status"] == "missing", (
+            "a table that was never there is not data loss"
+        )
+
+        damaged_result = session_recovery._copy_state_meta_salvage(
+            damaged, destination, chunk_size=4, progress_cb=None, source_rows=1
+        )
+        assert damaged_result["status"] == "failed", (
+            "a present-but-unusable table is data loss; reporting 'missing' "
+            "would let verification claim complete=True after dropping it"
+        )
+    finally:
+        destination.close()
+        absent.close()
+        damaged.close()
+
+
 def test_recovery_refuses_to_snapshot_a_live_database(tmp_path: Path) -> None:
     """Snapshotting must refuse while a connection to the source is live.