#!/usr/bin/env python3 """ Master Deployment: Household Checkpoint System Deploys checkpoint repos and heartbeats for ALL wizards """ import os import sys import json import subprocess import argparse from datetime import datetime from pathlib import Path GITEA_URL = os.environ.get("CLAW_CODE_GITEA_URL", "http://143.198.27.163:3000") GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "") TEMPLATE_SCRIPT = Path("/root/wizards/household-snapshots/scripts/template_checkpoint_heartbeat.py") # Wizard registry WIZARDS = [ {"id": "adagio", "name": "Adagio", "role": "breath-and-design", "home": "/root/wizards/adagio/home"}, {"id": "timmy", "name": "Timmy Time", "role": "father-house", "home": "/root/timmy"}, {"id": "ezra", "name": "Ezra", "role": "archivist", "home": "/root/wizards/ezra/home"}, ] def run_cmd(cmd, cwd=None): result = subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True, text=True) return result.stdout.strip(), result.stderr.strip(), result.returncode def create_gitea_repo(wizard_id): """Create checkpoint repo in Gitea.""" repo_name = f"{wizard_id}-checkpoint" # Check if repo exists check_url = f"{GITEA_URL}/api/v1/repos/allegro/{repo_name}" check_cmd = f'curl -s -o /dev/null -w "%{{http_code}}" -H "Authorization: token {GITEA_TOKEN}" {check_url}' stdout, _, _ = run_cmd(check_cmd) if stdout == "200": print(f" → Repo {repo_name} already exists") return True # Create repo create_url = f"{GITEA_URL}/api/v1/user/repos" data = json.dumps({ "name": repo_name, "description": f"State checkpoint for {wizard_id} - automatic 4-hour backups", "private": False, "auto_init": True, "default_branch": "main" }) cmd = f'curl -s -X POST {create_url} -H "Content-Type: application/json" -H "Authorization: token {GITEA_TOKEN}" -d \'{data}\'' stdout, stderr, code = run_cmd(cmd) if code == 0 and '"id":' in stdout: print(f" ✓ Created repo: {repo_name}") return True else: print(f" ✗ Failed to create repo: {stderr}") return False def setup_checkpoint_repo(wizard): """Setup checkpoint repo for a wizard.""" wizard_id = wizard["id"] wizard_name = wizard["name"] wizard_role = wizard["role"] repo_name = f"{wizard_id}-checkpoint" repo_dir = Path(f"/root/wizards/{repo_name}") home_dir = Path(wizard["home"]) print(f"\n=== Setting up {wizard_name} ===") # Create repo if not create_gitea_repo(wizard_id): return False # Clone or use existing if not repo_dir.exists(): clone_url = f"http://allegro:{GITEA_TOKEN}@143.198.27.163:3000/allegro/{repo_name}.git" stdout, stderr, code = run_cmd(f"git clone {clone_url} {repo_dir}") if code != 0: print(f" ✗ Clone failed: {stderr}") return False print(f" ✓ Cloned {repo_name}") # Setup git config run_cmd("git config user.email 'ezra@hermes.local'", cwd=repo_dir) run_cmd("git config user.name 'Ezra'", cwd=repo_dir) # Create structure (repo_dir / "scripts").mkdir(exist_ok=True) (repo_dir / "memories").mkdir(exist_ok=True) (repo_dir / "skills").mkdir(exist_ok=True) (repo_dir / "work").mkdir(exist_ok=True) # Create MANIFEST.md manifest = repo_dir / "MANIFEST.md" manifest_content = f"""# {wizard_name} State Checkpoint **Wizard:** {wizard_name} **Role:** {wizard_role} **Status:** ACTIVE ## Contents - SOUL.md - Conscience and principles - config.yaml - Harness configuration - memories/ - Durable memories - skills/ - Custom skills - work/ - Active work items ## Checkpoint Schedule Every 4 hours via cron --- *Auto-generated by Household Checkpoint System* """ manifest.write_text(manifest_content) print(f" ✓ Created MANIFEST.md") # Create heartbeat script from template if TEMPLATE_SCRIPT.exists(): script_content = TEMPLATE_SCRIPT.read_text() script_content = script_content.replace('WIZARD_ID_HERE', f'"{wizard_id}"') script_content = script_content.replace('WIZARD_NAME_HERE', f'"{wizard_name}"') script_content = script_content.replace('WIZARD_ROLE_HERE', f'"{wizard_role}"') # Adjust source dir if needed if wizard_id == "timmy": script_content = script_content.replace( f'SOURCE_DIR = Path(f"/root/wizards/{wizard_id}/home")', f'SOURCE_DIR = Path("/root/timmy")' ) script_path = repo_dir / "scripts" / "checkpoint_heartbeat.py" script_path.write_text(script_content) script_path.chmod(0o755) print(f" ✓ Created checkpoint_heartbeat.py") else: print(f" ✗ Template script not found at {TEMPLATE_SCRIPT}") return False # Initial commit run_cmd("git add -A", cwd=repo_dir) stdout, stderr, code = run_cmd('git commit -m "Initial checkpoint structure"', cwd=repo_dir) if code == 0 or "nothing to commit" in stderr.lower(): run_cmd("git push origin main", cwd=repo_dir) print(f" ✓ Pushed to Gitea") else: print(f" ✗ Commit failed: {stderr}") return True def install_cron(wizard_id): """Install cron job for a wizard.""" cron_line = f"0 */4 * * * cd /root/wizards/{wizard_id}-checkpoint && /usr/bin/python3 scripts/checkpoint_heartbeat.py >> /var/log/{wizard_id}-checkpoint.log 2>&1" # Check if already installed stdout, _, _ = run_cmd("crontab -l 2>/dev/null | grep -c checkpoint || echo 0") # Add cron job run_cmd(f'(crontab -l 2>/dev/null | grep -v "{wizard_id}-checkpoint"; echo "{cron_line}") | crontab -') print(f" ✓ Installed cron job") def run_initial_checkpoint(wizard_id): """Run first checkpoint.""" repo_dir = Path(f"/root/wizards/{wizard_id}-checkpoint") script_path = repo_dir / "scripts" / "checkpoint_heartbeat.py" if script_path.exists(): print(f" Running initial checkpoint...") stdout, stderr, code = run_cmd(f"python3 {script_path}", cwd=repo_dir) if code == 0: print(f" ✓ Initial checkpoint complete") else: print(f" ✗ Initial checkpoint failed: {stderr}") else: print(f" ✗ Script not found") def main(): parser = argparse.ArgumentParser(description="Deploy Household Checkpoint System") parser.add_argument("--wizards", help="Comma-separated wizard IDs (default: all)") parser.add_argument("--skip-cron", action="store_true", help="Skip cron installation") parser.add_argument("--dry-run", action="store_true", help="Show what would be done") args = parser.parse_args() # Filter wizards if args.wizards: wizard_ids = args.wizards.split(",") to_deploy = [w for w in WIZARDS if w["id"] in wizard_ids] else: to_deploy = WIZARDS print("=== Household Checkpoint Deployment ===") print(f"Time: {datetime.utcnow().isoformat()}Z") print(f"Wizards to deploy: {[w['id'] for w in to_deploy]}") print() if args.dry_run: print("DRY RUN - No changes will be made") for wizard in to_deploy: print(f"Would deploy: {wizard['name']} ({wizard['id']})") return 0 # Deploy each wizard results = [] for wizard in to_deploy: success = setup_checkpoint_repo(wizard) if success and not args.skip_cron: install_cron(wizard["id"]) run_initial_checkpoint(wizard["id"]) results.append((wizard["id"], success)) # Summary print("\n=== Deployment Summary ===") for wizard_id, success in results: status = "✓" if success else "✗" print(f"{status} {wizard_id}") # Show cron jobs print("\n=== Installed Cron Jobs ===") stdout, _, _ = run_cmd("crontab -l | grep checkpoint || echo 'None found'") print(stdout) return 0 if all(r[1] for r in results) else 1 if __name__ == "__main__": sys.exit(main())