fix(backup): bound locked database snapshot waits

hermes_cli/backup.py

@@ -155,6 +155,10 @@ class _SQLiteSnapshotError(RuntimeError):
     pass
 
 
+class _SQLiteBackupTimeout(RuntimeError):
+    """Raised when a SQLite snapshot remains busy past its deadline."""
+
+
 @contextmanager
 def _backup_operation_lock(hermes_home: Path, timeout_seconds: float = 0.25):
     """Acquire one cross-process backup slot for full and quick snapshots."""
@@ -347,7 +351,12 @@ def _should_skip_backup_file(abs_path: Path, rel_path: Path, out_path: Path) ->
 # SQLite safe copy
 # ---------------------------------------------------------------------------
 
-def _safe_copy_db(src: Path, dst: Path) -> bool:
+def _safe_copy_db(
+    src: Path,
+    dst: Path,
+    *,
+    timeout_seconds: float = 10.0,
+) -> bool:
     """Copy a SQLite database safely using the backup() API.
 
     Handles WAL mode — produces a consistent snapshot even while
@@ -357,12 +366,42 @@ def _safe_copy_db(src: Path, dst: Path) -> bool:
     conn = None
     backup_conn = None
     try:
-        conn = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
+        # Disable sqlite3's implicit busy wait so backup() progress callbacks
+        # control the full locked-source deadline instead of adding the
+        # connection's default timeout before each callback.
+        conn = sqlite3.connect(f"file:{src}?mode=ro", uri=True, timeout=0.0)
         backup_conn = sqlite3.connect(str(dst))
-        conn.backup(backup_conn)
+        busy_deadline = time.monotonic() + max(0.0, timeout_seconds)
+
+        def _check_backup_progress(status: int, _remaining: int, _total: int) -> None:
+            nonlocal busy_deadline
+            now = time.monotonic()
+            if status in (sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED):
+                if now >= busy_deadline:
+                    raise _SQLiteBackupTimeout(
+                        f"database remained locked for {timeout_seconds:g} seconds"
+                    )
+            else:
+                busy_deadline = now + max(0.0, timeout_seconds)
+
+        conn.backup(
+            backup_conn,
+            pages=256,
+            progress=_check_backup_progress,
+            sleep=0.1,
+        )
         return True
     except Exception as exc:
         logger.warning("SQLite safe copy failed for %s: %s", src, exc)
+        # Windows will not remove the partial destination while SQLite still
+        # has it open. Close it before fail-closed cleanup; the finally block
+        # still owns the source and any close failure.
+        if backup_conn is not None:
+            try:
+                backup_conn.close()
+            except Exception:
+                pass
+            backup_conn = None
         try:
             dst.unlink(missing_ok=True)
         except OSError:

tests/hermes_cli/test_backup.py

@@ -1043,6 +1043,55 @@ class TestSafeCopyDb:
         conn.close()
         assert rows == [(42,)]
 
+    def test_aborts_when_source_remains_busy_past_deadline(
+        self, tmp_path, monkeypatch
+    ):
+        from hermes_cli import backup as backup_mod
+
+        src = tmp_path / "locked.db"
+        dst = tmp_path / "copy.db"
+        src.touch()
+        dst.write_bytes(b"partial")
+
+        clock = iter((100.0, 100.5, 101.1))
+
+        class FakeSourceConnection:
+            def backup(self, _destination, *, pages, progress, sleep):
+                assert pages > 0
+                assert sleep > 0
+                progress(sqlite3.SQLITE_BUSY, 0, 1)
+                progress(sqlite3.SQLITE_BUSY, 0, 1)
+
+            def close(self):
+                pass
+
+        destination_closed = []
+
+        class FakeDestinationConnection:
+            def close(self):
+                destination_closed.append(True)
+
+        connections = iter((FakeSourceConnection(), FakeDestinationConnection()))
+        real_unlink = Path.unlink
+
+        def assert_closed_before_unlink(path, *args, **kwargs):
+            assert destination_closed
+            return real_unlink(path, *args, **kwargs)
+
+        connect_calls = []
+
+        def fake_connect(*args, **kwargs):
+            connect_calls.append((args, kwargs))
+            return next(connections)
+
+        monkeypatch.setattr(backup_mod.sqlite3, "connect", fake_connect)
+        monkeypatch.setattr(backup_mod.time, "monotonic", lambda: next(clock))
+        monkeypatch.setattr(Path, "unlink", assert_closed_before_unlink)
+
+        assert backup_mod._safe_copy_db(src, dst, timeout_seconds=1.0) is False
+        assert connect_calls[0][1]["timeout"] == 0.0
+        assert not dst.exists()
+
 
     def test_is_zeroed_sqlite_file_detects_nul_header(self, tmp_path):
         from hermes_cli.backup import is_zeroed_sqlite_file