fix(state): loud failed state.db snapshot + zeroed-file quarantine

Hardening for the Windows zeroed-state.db class (#68474):
- Surface critical stdout when pre-update/quick snapshot cannot copy a
  present *.db (was log-only; update still looked successful).
- Detect all-NUL SQLite header on SessionDB open, quarantine the bytes,
  and open a fresh DB with recovery guidance to state-snapshots.

Does not claim storage-stack root cause.

hermes_cli/backup.py

@@ -283,6 +283,32 @@ def _safe_copy_db(src: Path, dst: Path) -> bool:
                     pass
 
 
+def is_zeroed_sqlite_file(path: Path, *, probe_bytes: int = 100) -> bool:
+    """True when *path* looks like the #68474 zeroed-state.db signature.
+
+    Signature: size > 0, first *probe_bytes* are all NUL (no ``SQLite format 3``
+    header). Used at SessionDB open and for snapshot diagnostics so a silent
+    all-zero file becomes a guided recovery instead of a generic failure.
+    """
+    try:
+        size = path.stat().st_size
+    except OSError:
+        return False
+    if size <= 0:
+        return False
+    try:
+        with open(path, "rb") as fh:
+            head = fh.read(max(16, probe_bytes))
+    except OSError:
+        return False
+    if not head:
+        return False
+    if head.startswith(b"SQLite format 3"):
+        return False
+    return all(byte == 0 for byte in head)
+
+
+
 # ---------------------------------------------------------------------------
 # SQLite integrity verification
 # ---------------------------------------------------------------------------
@@ -1031,6 +1057,7 @@ def create_quick_snapshot(
     snap_dir.mkdir(parents=True, exist_ok=True)
 
     manifest: Dict[str, int] = {}  # rel_path -> file size
+    failed_dbs: list[str] = []  # present *.db that could not be snapshotted
 
     for rel in _QUICK_STATE_FILES:
         src = home / rel
@@ -1060,6 +1087,16 @@ def create_quick_snapshot(
                     # snapshot time) is captured consistently.
                     if sub.suffix == ".db":
                         if not _safe_copy_db(sub, dst):
+                            failed_dbs.append(sub_rel)
+                            print(
+                                f"  ⚠ Snapshot: SQLite safe copy FAILED for {sub_rel} "
+                                f"— file may be locked or corrupted"
+                            )
+                            if is_zeroed_sqlite_file(sub):
+                                print(
+                                    f"  ⚠ Snapshot: {sub_rel} looks ZEROED "
+                                    f"(no SQLite header; {sub.stat().st_size} bytes of NULs?)"
+                                )
                             continue
                     else:
                         shutil.copy2(sub, dst)
@@ -1080,6 +1117,16 @@ def create_quick_snapshot(
         try:
             if src.suffix == ".db":
                 if not _safe_copy_db(src, dst):
+                    failed_dbs.append(rel)
+                    print(
+                        f"  ⚠ Snapshot: SQLite safe copy FAILED for {rel} "
+                        f"— file may be locked or corrupted"
+                    )
+                    if is_zeroed_sqlite_file(src):
+                        print(
+                            f"  ⚠ Snapshot: {rel} looks ZEROED "
+                            f"(no SQLite header; {src.stat().st_size} bytes)"
+                        )
                     continue
             else:
                 shutil.copy2(src, dst)
@@ -1087,8 +1134,31 @@ def create_quick_snapshot(
         except (OSError, PermissionError) as exc:
             logger.warning("Could not snapshot %s: %s", rel, exc)
 
+    if failed_dbs:
+        # Critical: update path used to log-and-continue with exit 0, so a
+        # missing state.db backup looked like a successful pre-update snapshot
+        # (#68474). Surface this on stdout where operators actually look.
+        print(
+            "  ⚠ CRITICAL: could not snapshot DB file(s): "
+            + ", ".join(failed_dbs)
+        )
+        print(
+            "  ⚠ If sessions disappear after update, check "
+            f"{root} and run: hermes snapshot list"
+        )
+        logger.error(
+            "Quick snapshot failed to capture DB file(s): %s",
+            ", ".join(failed_dbs),
+        )
+
     if not manifest:
         shutil.rmtree(snap_dir, ignore_errors=True)
+        if failed_dbs:
+            # Distinguish "nothing to snapshot" from "state.db present but unreadable"
+            print(
+                "  ⚠ Snapshot aborted: no files captured "
+                f"(failed DBs: {', '.join(failed_dbs)})"
+            )
         return None
 
     # Write manifest
@@ -1099,6 +1169,7 @@ def create_quick_snapshot(
         "file_count": len(manifest),
         "total_size": sum(manifest.values()),
         "files": manifest,
+        "failed_dbs": failed_dbs,
     }
     with open(snap_dir / "manifest.json", "w", encoding="utf-8") as f:
         json.dump(meta, f, indent=2)

hermes_state.py

@@ -1575,6 +1575,63 @@ class CompressionSessionBusyError(RuntimeError):
     """A non-owner tried to write while compression owns the session."""
 
 
+def is_zeroed_state_db(path: Path, *, probe_bytes: int = 100) -> bool:
+    """Detect the #68474 zeroed state.db signature (size>0, NUL header).
+
+    Prefer ``hermes_cli.backup.is_zeroed_sqlite_file`` when available; this
+    local copy keeps SessionDB openable without importing the CLI package
+    in constrained embed paths.
+    """
+    try:
+        from hermes_cli.backup import is_zeroed_sqlite_file
+
+        return is_zeroed_sqlite_file(path, probe_bytes=probe_bytes)
+    except Exception:
+        pass
+    try:
+        size = path.stat().st_size
+    except OSError:
+        return False
+    if size <= 0:
+        return False
+    try:
+        with open(path, "rb") as fh:
+            head = fh.read(max(16, probe_bytes))
+    except OSError:
+        return False
+    if not head or head.startswith(b"SQLite format 3"):
+        return False
+    return all(byte == 0 for byte in head)
+
+
+def quarantine_zeroed_state_db(path: Path) -> Optional[Path]:
+    """Move a zeroed state.db aside (preserve bytes) and return quarantine path."""
+    try:
+        ts = time.strftime("%Y%m%d-%H%M%S")
+    except Exception:
+        ts = "unknown"
+    dest = path.with_name(f"{path.name}.zeroed-{ts}.bak")
+    # Avoid clobbering a prior quarantine
+    n = 0
+    while dest.exists():
+        n += 1
+        dest = path.with_name(f"{path.name}.zeroed-{ts}-{n}.bak")
+    try:
+        path.rename(dest)
+    except OSError as exc:
+        logger.error("Failed to quarantine zeroed %s: %s", path, exc)
+        return None
+    # Also move empty WAL/SHM if present so a fresh open is clean
+    for suffix in ("-wal", "-shm"):
+        side = Path(str(path) + suffix)
+        if side.exists():
+            try:
+                side.rename(Path(str(dest) + suffix))
+            except OSError:
+                pass
+    return dest
+
+
 class SessionDB:
     """
     SQLite-backed session storage with FTS5 search.
@@ -1661,6 +1718,35 @@ class SessionDB:
 
             self.db_path.parent.mkdir(parents=True, exist_ok=True)
 
+            # #68474: zeroed state.db (size>0, all-NUL header) used to fail as a
+            # generic "file is not a database" with no recovery path. Quarantine
+            # the bytes (do not delete) and continue so a fresh DB can open;
+            # point the operator at pre-update snapshots.
+            if (
+                not read_only
+                and self.db_path.exists()
+                and is_zeroed_state_db(self.db_path)
+            ):
+                try:
+                    zsize = self.db_path.stat().st_size
+                except OSError:
+                    zsize = -1
+                qpath = quarantine_zeroed_state_db(self.db_path)
+                snaps = self.db_path.parent / "state-snapshots"
+                msg = (
+                    f"state.db looks ZEROED ({zsize} bytes, no SQLite header). "
+                    f"Preserved at {qpath or '(quarantine failed — file left in place)'}. "
+                    f"Restore from {snaps} via `hermes snapshot list` / "
+                    f"`hermes snapshot restore <id>` if available. "
+                    "Opening a fresh empty database so the agent can start."
+                )
+                logger.error(msg)
+                _set_last_init_error(msg)
+                # If quarantine failed, do not open the zeroed file (would fail
+                # opaquely or risk further damage). Raise with the clear message.
+                if qpath is None and self.db_path.exists() and is_zeroed_state_db(self.db_path):
+                    raise sqlite3.DatabaseError(msg)
+
             def _connect_and_init():
                 self._conn = sqlite3.connect(
                     str(self.db_path),

tests/hermes_cli/test_backup.py

@@ -1431,6 +1431,29 @@ class TestSafeCopyDb:
         assert rows == [("wal-test",)]
 
 
+
+    def test_is_zeroed_sqlite_file_detects_nul_header(self, tmp_path):
+        from hermes_cli.backup import is_zeroed_sqlite_file
+        p = tmp_path / "state.db"
+        p.write_bytes(bytes(4096))  # all NULs
+        assert is_zeroed_sqlite_file(p) is True
+
+    def test_is_zeroed_sqlite_file_rejects_valid_db(self, tmp_path):
+        from hermes_cli.backup import is_zeroed_sqlite_file
+        p = tmp_path / "ok.db"
+        conn = sqlite3.connect(str(p))
+        conn.execute("CREATE TABLE t (x INT)")
+        conn.commit()
+        conn.close()
+        assert is_zeroed_sqlite_file(p) is False
+
+    def test_is_zeroed_sqlite_file_empty_file(self, tmp_path):
+        from hermes_cli.backup import is_zeroed_sqlite_file
+        p = tmp_path / "empty.db"
+        p.write_bytes(b"")
+        assert is_zeroed_sqlite_file(p) is False
+
+
 # ---------------------------------------------------------------------------
 # Quick state snapshot tests
 # ---------------------------------------------------------------------------
@@ -1477,13 +1500,32 @@ class TestQuickSnapshot:
         snap_id = create_quick_snapshot(hermes_home=hermes_home)
         db_copy = hermes_home / "state-snapshots" / snap_id / "state.db"
         assert db_copy.exists()
-
         conn = sqlite3.connect(str(db_copy))
         rows = conn.execute("SELECT * FROM sessions").fetchall()
         conn.close()
         assert len(rows) == 1
         assert rows[0] == ("s1", "hello world")
 
+    def test_failed_state_db_copy_is_loud(self, hermes_home, monkeypatch, capsys):
+        """#68474: unreadable state.db must not look like a silent success."""
+        from hermes_cli import backup as backup_mod
+
+        def boom(src, dst):
+            return False
+
+        monkeypatch.setattr(backup_mod, "_safe_copy_db", boom)
+        snap_id = backup_mod.create_quick_snapshot(hermes_home=hermes_home)
+        err = capsys.readouterr().out
+        assert "SQLite safe copy FAILED" in err or "CRITICAL" in err
+        assert "state.db" in err
+        # Other small files may still snapshot
+        if snap_id:
+            manifest = (hermes_home / "state-snapshots" / snap_id / "manifest.json")
+            assert manifest.exists()
+            data = json.loads(manifest.read_text(encoding="utf-8"))
+            assert "state.db" not in data.get("files", {})
+            assert "state.db" in data.get("failed_dbs", [])
+
     def test_copies_nested_files(self, hermes_home):
         from hermes_cli.backup import create_quick_snapshot
         snap_id = create_quick_snapshot(hermes_home=hermes_home)

tests/test_zeroed_state_db.py

@@ -0,0 +1,41 @@
+"""#68474 hardening: zeroed state.db detection + quarantine."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+def test_is_zeroed_state_db_and_quarantine(tmp_path):
+    import hermes_state as hs
+
+    db = tmp_path / "state.db"
+    db.write_bytes(bytes(1024))
+    assert hs.is_zeroed_state_db(db) is True
+
+    q = hs.quarantine_zeroed_state_db(db)
+    assert q is not None
+    assert q.exists()
+    assert not db.exists()
+    assert q.read_bytes() == bytes(1024)
+
+
+def test_sessiondb_opens_fresh_after_zeroed_quarantine(tmp_path, monkeypatch):
+    import hermes_state as hs
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    db = tmp_path / "state.db"
+    db.write_bytes(bytes(4096))
+
+    sdb = hs.SessionDB(db_path=db)
+    try:
+        # Fresh DB should open and accept schema
+        assert db.exists()
+        assert not hs.is_zeroed_state_db(db)
+        # Quarantine retained
+        backups = list(tmp_path.glob("state.db.zeroed-*.bak"))
+        assert len(backups) == 1
+        assert backups[0].stat().st_size == 4096
+    finally:
+        sdb.close()