household-snapshots/scripts/snapshot_heartbeat.py

#!/usr/bin/env python3
"""
Household Snapshot Heartbeat

Captures state snapshots for all wizards and commits to Gitea.
Runs every 15 minutes via cron.
"""

import os
import sys
import json
import subprocess
from datetime import datetime
from pathlib import Path

# Configuration
SNAPSHOTS_DIR = Path("/root/wizards/household-snapshots/snapshots")
REPO_DIR = Path("/root/wizards/household-snapshots")
WIZARDS = ["allegro", "adagio"]
GITEA_URL = "http://143.198.27.163:3000/allegro/household-snapshots"

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 snapshot_wizard(wizard_name):
    """Capture snapshot of a wizard's state."""
    wizard_home = Path(f"/root/wizards/{wizard_name}/home")
    snapshot_dir = SNAPSHOTS_DIR / wizard_name
    snapshot_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    snapshot_file = snapshot_dir / f"{timestamp}.json"
    
    # Gather state information
    state = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "wizard": wizard_name,
        "tick": None,  # Will be filled from Evenia
        "home_exists": wizard_home.exists(),
        "config_exists": (wizard_home / "config.yaml").exists(),
        "soul_exists": (wizard_home / "SOUL.md").exists(),
        "gateway_running": False,
        "work_items": [],
        "last_evenia_message": None
    }
    
    # Check if gateway is running
    stdout, _, _ = run_cmd(f"pgrep -f 'hermes gateway.*{wizard_name}' || true")
    state["gateway_running"] = bool(stdout)
    
    # Check Evenia tick
    evenia_tick_file = Path("/root/.hermes/evenia/tick.json")
    if evenia_tick_file.exists():
        try:
            with open(evenia_tick_file) as f:
                tick_data = json.load(f)
                state["tick"] = tick_data.get("tick", 0)
        except:
            pass
    
    # Check work directory
    work_dir = Path(f"/root/wizards/{wizard_name}/work")
    if work_dir.exists():
        state["work_items"] = [str(p.relative_to(work_dir)) for p in work_dir.rglob("*") if p.is_file()]
    
    # Get last Evenia message
    evenia_log = Path("/root/.hermes/evenia/messages.jsonl")
    if evenia_log.exists():
        try:
            with open(evenia_log) as f:
                lines = f.readlines()
                if lines:
                    last_msg = json.loads(lines[-1])
                    state["last_evenia_message"] = {
                        "tick": last_msg.get("tick"),
                        "from": last_msg.get("from"),
                        "to": last_msg.get("to"),
                        "timestamp": last_msg.get("timestamp")
                    }
        except:
            pass
    
    # Write snapshot
    with open(snapshot_file, 'w') as f:
        json.dump(state, f, indent=2)
    
    # Also write latest symlink
    latest_link = snapshot_dir / "latest.json"
    if latest_link.exists() or latest_link.is_symlink():
        latest_link.unlink()
    latest_link.symlink_to(snapshot_file.name)
    
    return snapshot_file

def snapshot_evenia():
    """Capture Evenia world state."""
    evenia_dir = Path("/root/.hermes/evenia")
    snapshot_dir = SNAPSHOTS_DIR / "evenia"
    snapshot_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    snapshot_file = snapshot_dir / f"{timestamp}.json"
    
    state = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "tick": 0,
        "wizards": [],
        "message_count": 0
    }
    
    # Get tick
    tick_file = evenia_dir / "tick.json"
    if tick_file.exists():
        try:
            with open(tick_file) as f:
                tick_data = json.load(f)
                state["tick"] = tick_data.get("tick", 0)
                state["wizards"] = list(tick_data.get("wizards", {}).keys())
        except:
            pass
    
    # Count messages
    log_file = evenia_dir / "messages.jsonl"
    if log_file.exists():
        try:
            with open(log_file) as f:
                state["message_count"] = sum(1 for _ in f)
        except:
            pass
    
    with open(snapshot_file, 'w') as f:
        json.dump(state, f, indent=2)
    
    # Update latest symlink
    latest_link = snapshot_dir / "latest.json"
    if latest_link.exists() or latest_link.is_symlink():
        latest_link.unlink()
    latest_link.symlink_to(snapshot_file.name)
    
    return snapshot_file

def commit_snapshots():
    """Commit snapshots to Gitea."""
    timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
    
    # Git operations
    run_cmd("git add -A", cwd=REPO_DIR)
    stdout, stderr, code = run_cmd(
        f'git commit -m "Snapshot heartbeat: {timestamp}"',
        cwd=REPO_DIR
    )
    
    if code == 0 or "nothing to commit" in stderr.lower():
        # Push to Gitea
        stdout, stderr, code = run_cmd("git push origin main", cwd=REPO_DIR)
        if code == 0:
            print(f"✓ Snapshots committed and pushed: {timestamp}")
            return True
        else:
            print(f"✗ Push failed: {stderr}")
            return False
    else:
        print(f"✗ Commit failed: {stderr}")
        return False

def main():
    """Main heartbeat function."""
    print(f"=== Household Snapshot Heartbeat ===")
    print(f"Time: {datetime.utcnow().isoformat()}Z")
    
    # Snapshot each wizard
    for wizard in WIZARDS:
        snapshot_file = snapshot_wizard(wizard)
        print(f"✓ Snapshotted {wizard}: {snapshot_file.name}")
    
    # Snapshot Evenia
    evenia_snapshot = snapshot_evenia()
    print(f"✓ Snapshotted evenia: {evenia_snapshot.name}")
    
    # Commit and push
    if commit_snapshots():
        print("✓ State preserved in Gitea")
        return 0
    else:
        print("✗ Failed to preserve state")
        return 1

if __name__ == "__main__":
    sys.exit(main())