"""Soul persistence via HF Dataset snapshot/restore. Free + fail-soft. Nothing here ever raises into the caller; if HF_TOKEN/dataset is absent it no-ops (the soul just won't persist, but the app runs fine).""" from __future__ import annotations import os, sqlite3, tempfile, threading, time STATE_FILE = "soul.db" def safe_copy_sqlite(src: str, dst: str) -> None: """Consistent copy of a (possibly open) SQLite db via the backup API.""" s = sqlite3.connect(src) try: d = sqlite3.connect(dst) try: s.backup(d) finally: d.close() finally: s.close() def _token() -> str | None: return os.environ.get("HF_TOKEN") def _download(repo_id: str, token: str) -> str | None: from huggingface_hub import hf_hub_download return hf_hub_download(repo_id=repo_id, repo_type="dataset", filename=STATE_FILE, token=token) def _upload(local_path: str, repo_id: str, token: str) -> None: from huggingface_hub import HfApi HfApi().upload_file(path_or_fileobj=local_path, path_in_repo=STATE_FILE, repo_id=repo_id, repo_type="dataset", token=token) def restore_soul(db_path: str, repo_id: str) -> bool: """Download the latest soul.db into db_path. Returns True if restored.""" token = _token() if not token: return False try: cached = _download(repo_id, token) os.makedirs(os.path.dirname(db_path) or ".", exist_ok=True) import shutil shutil.copyfile(cached, db_path) return True except Exception: return False def snapshot_soul(db_path: str, repo_id: str) -> bool: """Snapshot db_path to the dataset. Fail-soft. Returns True on upload.""" token = _token() if not token or not os.path.exists(db_path): return False try: tmp = os.path.join(tempfile.gettempdir(), "clanker-soul-snap.db") safe_copy_sqlite(db_path, tmp) _upload(tmp, repo_id, token) return True except Exception: return False class Snapshotter: """Throttled, background, fail-soft soul snapshotter.""" def __init__(self, db_path: str, repo_id: str, min_interval: float = 120.0, now=time.monotonic): self._db_path = db_path self._repo = repo_id self._min = min_interval self._now = now self._last = 0.0 self._lock = threading.Lock() self._thread: threading.Thread | None = None def _do(self) -> None: snapshot_soul(self._db_path, self._repo) def maybe_snapshot(self) -> None: with self._lock: t = self._now() if t - self._last < self._min: return self._last = t self._thread = threading.Thread(target=self._do, daemon=True) self._thread.start() def _join(self) -> None: if self._thread: self._thread.join(timeout=5)