fix(sessions): verify fully reconstructed recovery

hermes_cli/session_recovery.py

@@ -1094,13 +1094,33 @@ def _verify_recovered_database(
                 else:
                     verification["errors"].append(message)
 
+        cleanup = orphan_cleanup or {}
+        rebuilt_sessions = int(cleanup.get("sessions_reconstructed") or 0)
+        retained_messages = int(cleanup.get("messages_retained") or 0)
+        removed_messages = int(cleanup.get("messages_removed") or 0)
+        # A wholly unreadable sessions b-tree is recoverable when every output
+        # parent was rebuilt from the surviving messages and none were dropped.
+        # This is still data loss, but it is not structural verification failure.
+        sessions_fully_reconstructed = bool(
+            rebuilt_sessions > 0
+            and counts.get("sessions") == rebuilt_sessions
+            and counts.get("messages") == retained_messages
+            and removed_messages == 0
+        )
+
         for table, table_report in copy_report.items():
             status = table_report.get("status")
             if status not in {"failed", "partial"}:
                 continue
             message = f"{table} copy status is {status}"
             if allow_partial and (
-                status == "partial" or table not in {"sessions", "messages"}
+                status == "partial"
+                or table not in {"sessions", "messages"}
+                or (
+                    table == "sessions"
+                    and status == "failed"
+                    and sessions_fully_reconstructed
+                )
             ):
                 verification["warnings"].append(message)
                 verification["loss_detected"] = True

tests/hermes_cli/test_session_recovery.py

@@ -288,6 +288,20 @@ def _corrupt_middle_table_leaf(
     return leaf_page
 
 
+def _corrupt_table_root(path: Path, root_page: int) -> None:
+    data = bytearray(path.read_bytes())
+    page_size = int.from_bytes(data[16:18], "big")
+    if page_size == 1:
+        page_size = 65_536
+    page_start = (root_page - 1) * page_size
+    header_offset = page_start + (100 if root_page == 1 else 0)
+    assert data[header_offset] in {0x02, 0x05, 0x0A, 0x0D}
+    # Damage the root enough that no rowid bounds can be read. This reproduces
+    # a fully failed sessions copy while leaving the messages b-tree intact.
+    data[header_offset + 3 : header_offset + 5] = b"\xff\xff"
+    path.write_bytes(data)
+
+
 def test_snapshot_blocks_connections_opened_during_the_copy(
     tmp_path: Path,
 ) -> None:
@@ -999,6 +1013,59 @@ def test_partial_recovery_reconstructs_unreadable_sessions_without_message_loss(
         conn.close()
 
 
+def test_partial_recovery_verifies_fully_reconstructed_sessions(
+    tmp_path: Path,
+) -> None:
+    """A failed sessions copy is recoverable when every parent is rebuilt.
+
+    Reported July 2026: all 194 original session rows were unreadable, but
+    partial recovery reconstructed 194 placeholders, retained 20,817 readable
+    messages, removed none, and produced clean integrity, FK, and FTS checks.
+    The verifier still treated ``sessions copy status is failed`` as fatal and
+    printed "Do not install it" for a structurally healthy partial output.
+    """
+    source = tmp_path / "sessions-root-destroyed.db"
+    output = tmp_path / "sessions-root-destroyed-recovered.db"
+    session_count = 60
+    sessions_root = _make_many_sessions_source(source, session_count)
+    _corrupt_table_root(source, sessions_root)
+    source_hash = _sha256(source)
+
+    report = recover_session_database(
+        source,
+        output,
+        work_dir=tmp_path,
+        chunk_size=8,
+        allow_partial=True,
+    )
+
+    assert report["copy"]["sessions"]["status"] == "failed"
+    assert report["copy"]["messages"]["status"] == "complete"
+    cleanup = report["orphan_cleanup"]
+    assert cleanup["sessions_reconstructed"] == session_count
+    assert cleanup["messages_retained"] == session_count
+    assert cleanup["messages_removed"] == 0
+
+    verification = report["verification"]
+    assert verification["errors"] == []
+    assert "sessions copy status is failed" in verification["warnings"]
+    assert verification["integrity_check"] == ["ok"]
+    assert verification["foreign_key_check"] == []
+    assert verification["table_counts"]["sessions"] == session_count
+    assert verification["table_counts"]["messages"] == session_count
+    assert report["verified"] is True
+    assert report["complete"] is False
+    assert report["partial"] is True
+    assert report["source_unchanged"] is True
+    assert _sha256(source) == source_hash
+
+    with sqlite3.connect(str(output)) as conn:
+        recovered = conn.execute(
+            "SELECT source, COUNT(*) FROM sessions GROUP BY source"
+        ).fetchall()
+    assert recovered == [("recovered", session_count)]
+
+
 def test_cli_recover_writes_verified_report_without_touching_source(
     tmp_path: Path,
 ) -> None: