Checkpoint: 2026-04-02 03:12:10 UTC

scripts/checkpoint_heartbeat.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""
+Allegro Checkpoint Heartbeat
+
+Commits state changes to allegro-checkpoint repo every 4 hours.
+Captures: memories, config, skills, and SOUL state.
+"""
+
+import os
+import sys
+import json
+import subprocess
+from datetime import datetime
+from pathlib import Path
+import shutil
+
+REPO_DIR = Path("/root/wizards/allegro-checkpoint")
+SOURCE_DIR = Path("/root/wizards/allegro/home")
+CHECKPOINT_DIRS = ["memories", "skills"]
+CHECKPOINT_FILES = ["SOUL.md", "config.yaml", ".env"]
+
+def run_cmd(cmd, cwd=None):
+    """Run shell command and return output."""
+    result = subprocess.run(
+        cmd, shell=True, cwd=cwd, capture_output=True, text=True
+    )
+    return result.stdout.strip(), result.stderr.strip(), result.returncode
+
+def sync_directory(src, dst):
+    """Sync source directory to destination."""
+    if not src.exists():
+        print(f"  ✗ Source not found: {src}")
+        return False
+    
+    dst.mkdir(parents=True, exist_ok=True)
+    
+    # Remove old contents
+    for item in dst.iterdir():
+        if item.is_dir():
+            shutil.rmtree(item)
+        else:
+            item.unlink()
+    
+    # Copy new contents
+    for item in src.iterdir():
+        if item.is_dir():
+            shutil.copytree(item, dst / item.name)
+        else:
+            shutil.copy2(item, dst / item.name)
+    
+    return True
+
+def sync_file(src, dst):
+    """Sync a single file."""
+    if not src.exists():
+        print(f"  ✗ Source not found: {src}")
+        return False
+    
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(src, dst)
+    return True
+
+def capture_state():
+    """Capture current state from Allegro home."""
+    print("=== Capturing Allegro State ===")
+    
+    # Sync directories
+    for dirname in CHECKPOINT_DIRS:
+        src = SOURCE_DIR / dirname
+        dst = REPO_DIR / dirname
+        if sync_directory(src, dst):
+            print(f"  ✓ Synced {dirname}/")
+    
+    # Sync checkpoint files
+    for filename in CHECKPOINT_FILES:
+        src = SOURCE_DIR / filename
+        dst = REPO_DIR / filename
+        if sync_file(src, dst):
+            print(f"  ✓ Synced {filename}")
+    
+    # Update MANIFEST with timestamp
+    manifest = REPO_DIR / "MANIFEST.md"
+    if manifest.exists():
+        content = manifest.read_text()
+        now = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
+        # Only add timestamp line if not already present for today
+        timestamp_line = f"**Last Checkpoint:** {now}"
+        if timestamp_line not in content:
+            content = content.replace(
+                "**Status:** COMPLETE",
+                f"**Status:** COMPLETE  \n{timestamp_line}"
+            )
+            manifest.write_text(content)
+            print(f"  ✓ Updated MANIFEST.md")
+
+def has_changes():
+    """Check if there are changes to commit."""
+    stdout, _, _ = run_cmd("git status --porcelain", cwd=REPO_DIR)
+    return bool(stdout.strip())
+
+def commit_checkpoint():
+    """Commit state to Gitea."""
+    timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
+    
+    # Add all changes
+    run_cmd("git add -A", cwd=REPO_DIR)
+    
+    # Check if there are changes
+    if not has_changes():
+        print(f"  → No changes to commit")
+        return True
+    
+    # Commit
+    stdout, stderr, code = run_cmd(
+        f'git commit -m "Checkpoint: {timestamp}"',
+        cwd=REPO_DIR
+    )
+    
+    if code != 0:
+        print(f"  ✗ Commit failed: {stderr}")
+        return False
+    
+    # Push
+    stdout, stderr, code = run_cmd("git push origin main", cwd=REPO_DIR)
+    if code != 0:
+        print(f"  ✗ Push failed: {stderr}")
+        return False
+    
+    print(f"  ✓ Committed to Gitea: {timestamp}")
+    return True
+
+def main():
+    """Main heartbeat function."""
+    print(f"=== Allegro Checkpoint Heartbeat ===")
+    print(f"Time: {datetime.utcnow().isoformat()}Z")
+    print()
+    
+    # Capture state
+    capture_state()
+    print()
+    
+    # Commit
+    if commit_checkpoint():
+        print("\n✓ Checkpoint complete")
+        return 0
+    else:
+        print("\n✗ Checkpoint failed")
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(main())