ckriti/HuggingClaw
0
1#!/usr/bin/env python32"""3OpenClaw Full Directory Persistence for Hugging Face Spaces4========================================================5 6This script provides atomic, complete persistence of the entire ~/.openclaw directory.7It implements the comprehensive persistence plan:8 9- Config & Credentials (openclaw.json, credentials/)10- Workspace (workspace/ with AGENTS.md, SOUL.md, TOOLS.md, MEMORY.md, skills/, memory/)11- Sessions (agents/*/sessions/*.jsonl)12- Memory Index (memory/*.sqlite)13- QMD Backend (agents/*/qmd/)14- Extensions (extensions/)15- All other state in ~/.openclaw16 17Usage:18 # Backup (save)19 python3 openclaw_persist.py save20 21 # Restore (load)22 python3 openclaw_persist.py load23 24Environment Variables:25 HF_TOKEN - Hugging Face access token with write permissions26 OPENCLAW_DATASET_REPO - Dataset repo ID (e.g., "username/openclaw-state")27 OPENCLAW_HOME - OpenClaw home directory (default: ~/.openclaw)28"""29 30import os31import sys32import json33import tarfile34import tempfile35import shutil36import hashlib37import time38import signal39from datetime import datetime40from pathlib import Path41from typing import Optional, List, Set, Dict, Any42 43from huggingface_hub import HfApi, hf_hub_download44from huggingface_hub.utils import RepositoryNotFoundError45 46 47# ============================================================================48# Configuration49# ============================================================================50 51class Config:52 """Configuration for persistence system"""53 54 # Paths55 OPENCLAW_HOME = Path(os.environ.get("OPENCLAW_HOME", "~/.openclaw")).expanduser()56 BACKUP_FILENAME = "openclaw-full.tar.gz"57 BACKUP_STATE_FILE = ".persistence-state.json"58 LOCK_FILE = ".persistence.lock"59 60 # Backup rotation settings61 MAX_BACKUPS = 562 BACKUP_PREFIX = "backup-"63 64 # Patterns to exclude from backup65 EXCLUDE_PATTERNS = [66 "*.lock",67 "*.tmp",68 "*.pyc",69 "*__pycache__*",70 "*.socket",71 "*.pid",72 "node_modules",73 ".DS_Store",74 ".git",75 ]76 77 # Directories to skip entirely (relative to OPENCLAW_HOME)78 SKIP_DIRS = {79 ".cache",80 "logs",81 "temp",82 "tmp",83 }84 85 86# ============================================================================87# Utility Functions88# ============================================================================89 90def log(level: str, message: str, **kwargs):91 """Structured logging"""92 timestamp = datetime.now().isoformat()93 log_entry = {94 "timestamp": timestamp,95 "level": level,96 "message": message,97 **kwargs98 }99 print(json.dumps(log_entry), flush=True)100 101 102def calculate_file_hash(filepath: Path) -> str:103 """Calculate SHA256 hash of a file"""104 sha256 = hashlib.sha256()105 try:106 with open(filepath, "rb") as f:107 for chunk in iter(lambda: f.read(65536), b""):108 sha256.update(chunk)109 return sha256.hexdigest()110 except Exception:111 return ""112 113 114def get_directory_size(directory: Path) -> int:115 """Calculate total size of directory in bytes"""116 total_size = 0117 try:118 for dirpath, dirnames, filenames in os.walk(directory):119 for filename in filenames:120 filepath = Path(dirpath) / filename121 try:122 total_size += filepath.stat().st_size123 except Exception:124 pass125 except Exception:126 pass127 return total_size128 129 130def should_exclude(path: str, exclude_patterns: List[str]) -> bool:131 """Check if a path should be excluded based on patterns"""132 path_normalized = path.replace("\\", "/")133 134 for pattern in exclude_patterns:135 pattern = pattern.lstrip("/")136 if pattern.startswith("*"):137 suffix = pattern[1:]138 if path_normalized.endswith(suffix):139 return True140 elif pattern in path_normalized:141 return True142 143 return False144 145 146# ============================================================================147# Persistence Manager148# ============================================================================149 150class OpenClawPersistence:151 """152 Manages persistence of OpenClaw state to Hugging Face Dataset153 154 Features:155 - Atomic full-directory backup/restore156 - Proper exclusion of lock files and temporary data157 - Safe handling of SQLite databases158 - Backup rotation159 - Integrity verification160 """161 162 def __init__(self):163 self.api = None164 self.repo_id = os.environ.get("OPENCLAW_DATASET_REPO")165 self.token = os.environ.get("HF_TOKEN")166 self.home_dir = Config.OPENCLAW_HOME167 self.lock_file = self.home_dir / Config.LOCK_FILE168 self.state_file = self.home_dir / Config.BACKUP_STATE_FILE169 170 # Validate configuration171 if not self.repo_id:172 log("ERROR", "OPENCLAW_DATASET_REPO not set")173 raise ValueError("OPENCLAW_DATASET_REPO environment variable required")174 175 if not self.token:176 log("ERROR", "HF_TOKEN not set")177 raise ValueError("HF_TOKEN environment variable required")178 179 # Initialize API180 self.api = HfApi(token=self.token)181 182 log("INFO", "Initialized persistence manager",183 repo_id=self.repo_id,184 home_dir=str(self.home_dir))185 186 # -----------------------------------------------------------------------187 # Backup Operations188 # -----------------------------------------------------------------------189 190 def save(self) -> Dict[str, Any]:191 """192 Save current state to Hugging Face Dataset193 194 Creates a complete backup of ~/.openclaw directory as a tar.gz file.195 """196 operation_id = f"save-{int(time.time())}"197 start_time = time.time()198 199 log("INFO", "Starting save operation", operation_id=operation_id)200 201 # Check if home directory exists202 if not self.home_dir.exists():203 log("WARNING", "OpenClaw home directory does not exist, creating")204 self.home_dir.mkdir(parents=True, exist_ok=True)205 206 # Check for existing lock207 if self.lock_file.exists():208 log("WARNING", "Lock file exists, another operation may be in progress")209 # Continue anyway, but log warning210 211 # Create lock file212 try:213 self.lock_file.write_text(str(os.getpid()))214 except Exception as e:215 log("WARNING", "Could not create lock file", error=str(e))216 217 try:218 # Get directory info219 dir_size = get_directory_size(self.home_dir)220 log("INFO", "Directory size calculated",221 size_bytes=dir_size,222 size_mb=f"{dir_size / (1024*1024):.2f}")223 224 # Create tar archive225 with tempfile.TemporaryDirectory() as tmpdir:226 tar_path = Path(tmpdir) / Config.BACKUP_FILENAME227 manifest = self._create_tar_archive(tar_path)228 229 # Read archive info230 tar_size = tar_path.stat().st_size231 log("INFO", "Archive created",232 size_bytes=tar_size,233 size_mb=f"{tar_size / (1024*1024):.2f}",234 files_count=manifest["file_count"])235 236 # Upload to dataset237 remote_path = f"{Config.BACKUP_PREFIX}{datetime.now().strftime('%Y%m%d_%H%M%S')}.tar.gz"238 upload_result = self._upload_archive(tar_path, remote_path)239 240 # Update state file241 self._update_state({242 "last_save_time": datetime.now().isoformat(),243 "last_save_operation": operation_id,244 "last_save_remote_path": remote_path,245 "last_save_commit": upload_result.get("commit_id"),246 "last_save_manifest": manifest,247 })248 249 # Rotate old backups250 self._rotate_backups()251 252 duration = time.time() - start_time253 log("INFO", "Save completed successfully",254 operation_id=operation_id,255 duration_seconds=f"{duration:.2f}")256 257 return {258 "success": True,259 "operation_id": operation_id,260 "remote_path": remote_path,261 "commit_id": upload_result.get("commit_id"),262 "duration": duration,263 "manifest": manifest264 }265 266 except Exception as e:267 log("ERROR", "Save operation failed",268 operation_id=operation_id,269 error=str(e),270 exc_info=True)271 return {272 "success": False,273 "operation_id": operation_id,274 "error": str(e)275 }276 finally:277 # Remove lock file278 if self.lock_file.exists():279 try:280 self.lock_file.unlink()281 except Exception:282 pass283 284 def _create_tar_archive(self, tar_path: Path) -> Dict[str, Any]:285 """Create tar.gz archive of OpenClaw home directory"""286 manifest = {287 "created_at": datetime.now().isoformat(),288 "version": "1.0",289 "file_count": 0,290 "excluded_patterns": [],291 "included_dirs": [],292 "skipped_dirs": [],293 }294 295 excluded_count = 0296 297 def tar_filter(tarinfo: tarfile.TarInfo) -> Optional[tarfile.TarInfo]:298 nonlocal excluded_count, manifest299 300 # Skip lock file itself301 if tarinfo.name.endswith(Config.LOCK_FILE):302 excluded_count += 1303 return None304 305 # Skip state file (will be written after backup)306 if tarinfo.name.endswith(Config.BACKUP_STATE_FILE):307 return None308 309 # Get relative path310 rel_path = tarinfo.name311 if rel_path.startswith("./"):312 rel_path = rel_path[2:]313 314 # Check exclusion patterns315 if should_exclude(rel_path, Config.EXCLUDE_PATTERNS):316 excluded_count += 1317 manifest["excluded_patterns"].append(rel_path)318 return None319 320 # Check if parent directory should be skipped321 path_parts = Path(rel_path).parts322 if path_parts and path_parts[0] in Config.SKIP_DIRS:323 excluded_count += 1324 return None325 326 # Track included327 manifest["file_count"] += 1328 if path_parts and path_parts[0] not in manifest["included_dirs"]:329 manifest["included_dirs"].append(path_parts[0])330 331 return tarinfo332 333 # Create archive334 with tarfile.open(tar_path, "w:gz") as tar:335 tar.add(self.home_dir, arcname=".", filter=tar_filter)336 337 manifest["excluded_count"] = excluded_count338 manifest["skipped_dirs"] = list(Config.SKIP_DIRS)339 340 return manifest341 342 def _upload_archive(self, local_path: Path, remote_path: str) -> Dict[str, Any]:343 """Upload archive to Hugging Face Dataset"""344 try:345 # Ensure repo exists346 try:347 self.api.repo_info(repo_id=self.repo_id, repo_type="dataset")348 except RepositoryNotFoundError:349 log("INFO", "Creating new dataset repository")350 self.api.create_repo(351 repo_id=self.repo_id,352 repo_type="dataset",353 private=True354 )355 356 # Upload file357 commit_info = self.api.upload_file(358 path_or_fileobj=str(local_path),359 path_in_repo=remote_path,360 repo_id=self.repo_id,361 repo_type="dataset",362 commit_message=f"OpenClaw state backup - {datetime.now().isoformat()}"363 )364 365 log("INFO", "File uploaded successfully",366 remote_path=remote_path,367 commit_url=commit_info.commit_url)368 369 return {370 "success": True,371 "commit_id": commit_info.oid,372 "commit_url": commit_info.commit_url373 }374 375 except Exception as e:376 log("ERROR", "Upload failed", error=str(e))377 raise378 379 def _update_state(self, state_update: Dict[str, Any]):380 """Update persistence state file"""381 try:382 current_state = {}383 if self.state_file.exists():384 with open(self.state_file, 'r') as f:385 current_state = json.load(f)386 387 current_state.update(state_update)388 389 self.state_file.parent.mkdir(parents=True, exist_ok=True)390 with open(self.state_file, 'w') as f:391 json.dump(current_state, f, indent=2)392 393 except Exception as e:394 log("WARNING", "Could not update state file", error=str(e))395 396 def _rotate_backups(self):397 """Rotate old backups, keeping only MAX_BACKUPS most recent"""398 try:399 files = self.api.list_repo_files(400 repo_id=self.repo_id,401 repo_type="dataset"402 )403 404 # Get backup files405 backups = [406 f for f in files407 if f.startswith(Config.BACKUP_PREFIX) and f.endswith(".tar.gz")408 ]409 410 # Sort by name (which includes timestamp)411 backups = sorted(backups)412 413 # Delete old backups414 if len(backups) > Config.MAX_BACKUPS:415 to_delete = backups[:-Config.MAX_BACKUPS]416 log("INFO", "Rotating backups",417 total=len(backups),418 keeping=Config.MAX_BACKUPS,419 deleting=len(to_delete))420 421 for old_backup in to_delete:422 try:423 self.api.delete_file(424 path_in_repo=old_backup,425 repo_id=self.repo_id,426 repo_type="dataset"427 )428 log("INFO", "Deleted old backup", file=old_backup)429 except Exception as e:430 log("WARNING", "Could not delete backup",431 file=old_backup,432 error=str(e))433 434 except Exception as e:435 log("WARNING", "Backup rotation failed", error=str(e))436 437 # -----------------------------------------------------------------------438 # Restore Operations439 # -----------------------------------------------------------------------440 441 def load(self, force: bool = False) -> Dict[str, Any]:442 """443 Load state from Hugging Face Dataset444 445 Restores the most recent backup. If force is False and local state446 exists, it will create a backup before restoring.447 """448 operation_id = f"load-{int(time.time())}"449 start_time = time.time()450 451 log("INFO", "Starting load operation",452 operation_id=operation_id,453 force=force)454 455 try:456 # Get latest backup457 backup_info = self._find_latest_backup()458 459 if not backup_info:460 log("WARNING", "No backups found, starting fresh")461 # Ensure home directory exists462 self.home_dir.mkdir(parents=True, exist_ok=True)463 return {464 "success": True,465 "operation_id": operation_id,466 "restored": False,467 "message": "No backups found, starting fresh"468 }469 470 log("INFO", "Found backup to restore",471 backup_file=backup_info["filename"],472 timestamp=backup_info.get("timestamp"))473 474 # Create local backup if state exists475 if self.home_dir.exists() and not force:476 backup_dir = self._create_local_backup()477 log("INFO", "Created local backup", backup_dir=str(backup_dir))478 479 # Download and extract480 with tempfile.TemporaryDirectory() as tmpdir:481 tar_path = Path(tmpdir) / "backup.tar.gz"482 483 # Download backup484 log("INFO", "Downloading backup...")485 downloaded_path = hf_hub_download(486 repo_id=self.repo_id,487 filename=backup_info["filename"],488 repo_type="dataset",489 token=self.token,490 local_dir=tmpdir,491 local_dir_use_symlinks=False492 )493 494 # Extract archive495 log("INFO", "Extracting archive...")496 self._extract_archive(downloaded_path)497 498 duration = time.time() - start_time499 log("INFO", "Load completed successfully",500 operation_id=operation_id,501 duration_seconds=f"{duration:.2f}")502 503 return {504 "success": True,505 "operation_id": operation_id,506 "restored": True,507 "backup_file": backup_info["filename"],508 "duration": duration509 }510 511 except Exception as e:512 log("ERROR", "Load operation failed",513 operation_id=operation_id,514 error=str(e),515 exc_info=True)516 return {517 "success": False,518 "operation_id": operation_id,519 "error": str(e)520 }521 522 def _find_latest_backup(self) -> Optional[Dict[str, Any]]:523 """Find the latest backup file in the dataset"""524 try:525 files = self.api.list_repo_files(526 repo_id=self.repo_id,527 repo_type="dataset"528 )529 530 # Get backup files sorted by name (timestamp)531 backups = sorted(532 [f for f in files if f.startswith(Config.BACKUP_PREFIX) and f.endswith(".tar.gz")],533 reverse=True534 )535 536 if not backups:537 return None538 539 latest = backups[0]540 541 # Extract timestamp from filename542 timestamp_str = latest.replace(Config.BACKUP_PREFIX, "").replace(".tar.gz", "")543 try:544 timestamp = datetime.strptime(timestamp_str, "%Y%m%d_%H%M%S").isoformat()545 except ValueError:546 timestamp = None547 548 return {549 "filename": latest,550 "timestamp": timestamp551 }552 553 except Exception as e:554 log("ERROR", "Could not find latest backup", error=str(e))555 return None556 557 def _create_local_backup(self) -> Optional[Path]:558 """Create a backup of local state before restore"""559 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")560 backup_dir = self.home_dir.parent / f"{self.home_dir.name}_backup_{timestamp}"561 562 try:563 if self.home_dir.exists():564 shutil.copytree(self.home_dir, backup_dir)565 return backup_dir566 except Exception as e:567 log("WARNING", "Could not create local backup", error=str(e))568 569 return None570 571 def _extract_archive(self, tar_path: Path):572 """Extract tar.gz archive to home directory"""573 # Ensure home directory exists574 self.home_dir.mkdir(parents=True, exist_ok=True)575 576 # Extract archive577 with tarfile.open(tar_path, "r:gz") as tar:578 tar.extractall(self.home_dir)579 580 log("INFO", "Archive extracted successfully",581 destination=str(self.home_dir))582 583 584# ============================================================================585# CLI Interface586# ============================================================================587 588def main():589 if len(sys.argv) < 2:590 print("Usage: python openclaw_persist.py [save|load|status]", file=sys.stderr)591 print("", file=sys.stderr)592 print("Commands:", file=sys.stderr)593 print(" save - Save current state to dataset", file=sys.stderr)594 print(" load - Load state from dataset", file=sys.stderr)595 print(" status - Show persistence status", file=sys.stderr)596 sys.exit(1)597 598 command = sys.argv[1].lower()599 600 try:601 manager = OpenClawPersistence()602 603 if command == "save":604 result = manager.save()605 print(json.dumps(result, indent=2))606 sys.exit(0 if result.get("success") else 1)607 608 elif command == "load":609 force = "--force" in sys.argv or "-f" in sys.argv610 result = manager.load(force=force)611 print(json.dumps(result, indent=2))612 sys.exit(0 if result.get("success") else 1)613 614 elif command == "status":615 # Show status information616 status = {617 "configured": True,618 "repo_id": manager.repo_id,619 "home_dir": str(manager.home_dir),620 "home_exists": manager.home_dir.exists(),621 }622 623 # Load state file624 if manager.state_file.exists():625 with open(manager.state_file, 'r') as f:626 state = json.load(f)627 status["state"] = state628 629 # List backups630 backups = manager._find_latest_backup()631 status["latest_backup"] = backups632 633 print(json.dumps(status, indent=2))634 sys.exit(0)635 636 else:637 print(f"Unknown command: {command}", file=sys.stderr)638 sys.exit(1)639 640 except Exception as e:641 print(json.dumps({642 "success": False,643 "error": str(e)644 }, indent=2))645 sys.exit(1)646 647 648if __name__ == "__main__":649 main()650