fix(state): quarantine a state.db whose page 0 is not SQLite, not just a zeroed one

Startup only quarantined a 0-byte / all-NUL state.db. A file whose first page
was clobbered with record bytes (#102198) went straight to sqlite3.connect,
which raised "file is not a database" and deleted the -wal sidecar — the one
piece of evidence that could have been recovered.

`has_invalid_sqlite_header_preopen` generalises the zeroed probe (zeroed is a
subset of "no SQLite header"; same live-connection contract, never raises).
`quarantine_invalid_state_db` moves the file AND its -wal/-shm aside as
`state.db.<zeroed|notadb>-<ts>-<pid>.bak` before anything opens it; a fresh
DB is created as before.

Re-authored on current main (the quarantine helpers moved to
hermes_state_dbfile.py in d15c61b5dc); one regression test proves the
notadb case is quarantined with its sidecars and the new DB passes
integrity_check.

Refs #102198 (the write-after-SIGTERM that clobbers page 0 is not addressed
here; this preserves the evidence instead of destroying it).

hermes_state_dbfile.py

@@ -194,6 +194,23 @@ def is_zeroed_state_db(path: Path, *, probe_bytes: int = 100, force: bool = Fals
     return head is not None and not head.startswith(b"SQLite format 3") and all(b == 0 for b in head)
 
 
+def has_invalid_sqlite_header_preopen(path: Path, *, probe_bytes: int = 100, force: bool = False) -> bool:
+    """Pre-open byte probe: a pre-existing state.db whose first page is not SQLite (0-byte, NUL, or
+    clobbered page 0 as in #102198).  Same live-connection contract as :func:`is_zeroed_state_db`;
+    ``force=True`` only for offline files.  Never raises."""
+    try:
+        if not path.is_file():
+            return False
+        path.stat()
+        from hermes_cli.sqlite_safe_read import has_live_connection, read_header_bytes_preopen
+        if not force and has_live_connection(path):
+            return False
+        head = read_header_bytes_preopen(path, length=max(16, probe_bytes), force=force)
+    except Exception:
+        return False
+    return head is not None and not head.startswith(b"SQLite format 3")
+
+
 @contextlib.contextmanager
 def quarantine_cross_process_lock(path: Path, timeout: float = 5.0):
     """Acquire the cross-process lock for path.quarantine.lock."""
@@ -235,23 +252,24 @@ def quarantine_cross_process_lock(path: Path, timeout: float = 5.0):
             handle.close()
 
 
-def quarantine_zeroed_state_db(path: Path, *, already_locked: bool = False) -> Optional[Path]:
-    """Move a zeroed state.db aside (preserve bytes) and return quarantine path.  A cross-process
-    lock stops two concurrent startups racing: the second re-checks under the lock and finds the
-    file gone (or fresh) instead of clobbering the quarantine."""
+def quarantine_invalid_state_db(path: Path, *, already_locked: bool = False) -> Optional[Path]:
+    """Move a non-SQLite state.db (zeroed or clobbered page 0) aside with its -wal/-shm sidecars,
+    preserving bytes; return the quarantine path.  A cross-process lock stops two concurrent
+    startups racing: the second re-checks under the lock and finds the file gone (or fresh)."""
     def _do_quarantine():
         if not path.exists():
-            logger.info("quarantine_zeroed_state_db: %s already moved by another process", path)
+            logger.info("quarantine_invalid_state_db: %s already moved by another process", path)
             return None
-        if not is_zeroed_state_db(path):
-            logger.info("quarantine_zeroed_state_db: %s is no longer zeroed (another "
+        if not has_invalid_sqlite_header_preopen(path):
+            logger.info("quarantine_invalid_state_db: %s is no longer invalid (another "
                         "process quarantined it and a fresh DB was created)", path)
             return None
+        kind = "zeroed" if is_zeroed_state_db(path) else "notadb"
         try:
             ts = time.strftime("%Y%m%d-%H%M%S")
         except Exception:
             ts = "unknown"
-        stem = f"{path.name}.zeroed-{ts}-{os.getpid()}"
+        stem = f"{path.name}.{kind}-{ts}-{os.getpid()}"
         dest = path.with_name(f"{stem}.bak")
         n = 0
         while dest.exists():
@@ -260,7 +278,7 @@ def quarantine_zeroed_state_db(path: Path, *, already_locked: bool = False) -> O
         try:
             path.rename(dest)
         except OSError as exc:
-            logger.error("Failed to quarantine zeroed %s: %s", path, exc)
+            logger.error("Failed to quarantine invalid %s: %s", path, exc)
             return None
         for suffix in ("-wal", "-shm"):
             side = Path(str(path) + suffix)
@@ -274,7 +292,7 @@ def quarantine_zeroed_state_db(path: Path, *, already_locked: bool = False) -> O
     with quarantine_cross_process_lock(path) as acquired:
         if not acquired:
             logger.error("quarantine lock for %s not acquired within 5s — refusing to "
-                         "quarantine without the cross-process lock. The zeroed file "
+                         "quarantine without the cross-process lock. The invalid file "
                          "is left in place. If sessions fail to load, restore from "
                          "state-snapshots via `hermes snapshot list` / `hermes snapshot restore <id>`.",
                          path)