Allegro
885d08b5e5
- Snapshot heartbeat script for all wizards - Allegro's heartbeat for Adagio monitoring - Directory structure for snapshots and heartbeats - README with household documentation
149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Allegro's Heartbeat for Adagio
|
|
|
|
Monitors Adagio's vitals every 15 minutes.
|
|
Records heartbeat status to heartbeats/allegro/ directory.
|
|
Communicates via Evenia world tick.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
HEARTBEAT_DIR = Path("/root/wizards/household-snapshots/heartbeats/allegro")
|
|
EVENIA_SCRIPT = Path("/root/.hermes/evenia/world_tick.py")
|
|
ADAGIO_HOME = Path("/root/wizards/adagio/home")
|
|
|
|
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 check_adagio_vitals():
|
|
"""Check Adagio's vital signs."""
|
|
vitals = {
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"wizard": "adagio",
|
|
"checked_by": "allegro",
|
|
"checks": {}
|
|
}
|
|
|
|
# Check 1: Home directory exists
|
|
vitals["checks"]["home_exists"] = ADAGIO_HOME.exists()
|
|
|
|
# Check 2: SOUL.md present
|
|
vitals["checks"]["soul_present"] = (ADAGIO_HOME / "SOUL.md").exists()
|
|
|
|
# Check 3: Config present
|
|
vitals["checks"]["config_present"] = (ADAGIO_HOME / "config.yaml").exists()
|
|
|
|
# Check 4: Gateway running
|
|
stdout, _, _ = run_cmd("pgrep -f 'hermes gateway.*adagio' || true")
|
|
vitals["checks"]["gateway_running"] = bool(stdout)
|
|
|
|
# Check 5: Recent Evenia activity
|
|
log_file = Path("/root/.hermes/evenia/messages.jsonl")
|
|
if log_file.exists():
|
|
try:
|
|
with open(log_file) as f:
|
|
lines = f.readlines()
|
|
if lines:
|
|
last_msg = json.loads(lines[-1])
|
|
msg_time = datetime.fromisoformat(last_msg.get("timestamp", "").replace('Z', '+00:00'))
|
|
time_diff = datetime.utcnow() - msg_time.replace(tzinfo=None)
|
|
vitals["checks"]["evenia_recent"] = time_diff.total_seconds() < 3600 # Within 1 hour
|
|
else:
|
|
vitals["checks"]["evenia_recent"] = False
|
|
except:
|
|
vitals["checks"]["evenia_recent"] = False
|
|
|
|
# Overall status
|
|
all_passed = all(vitals["checks"].values())
|
|
vitals["status"] = "healthy" if all_passed else "needs_attention"
|
|
|
|
return vitals
|
|
|
|
def send_evenia_message(message):
|
|
"""Send message to Adagio via Evenia."""
|
|
if EVENIA_SCRIPT.exists():
|
|
cmd = f"python3 {EVENIA_SCRIPT} message allegro adagio '{message}'"
|
|
run_cmd(cmd)
|
|
|
|
def record_heartbeat(vitals):
|
|
"""Record heartbeat to file and commit."""
|
|
HEARTBEAT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
|
heartbeat_file = HEARTBEAT_DIR / f"{timestamp}.json"
|
|
|
|
with open(heartbeat_file, 'w') as f:
|
|
json.dump(vitals, f, indent=2)
|
|
|
|
# Update latest
|
|
latest_link = HEARTBEAT_DIR / "latest.json"
|
|
if latest_link.exists() or latest_link.is_symlink():
|
|
latest_link.unlink()
|
|
latest_link.symlink_to(heartbeat_file.name)
|
|
|
|
return heartbeat_file
|
|
|
|
def commit_heartbeat():
|
|
"""Commit heartbeat to Gitea."""
|
|
repo_dir = Path("/root/wizards/household-snapshots")
|
|
timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
|
|
run_cmd("git add -A", cwd=repo_dir)
|
|
stdout, stderr, code = run_cmd(
|
|
f'git commit -m "Allegro heartbeat for Adagio: {timestamp}"',
|
|
cwd=repo_dir
|
|
)
|
|
|
|
if code == 0:
|
|
run_cmd("git push origin main", cwd=repo_dir)
|
|
return True
|
|
return "nothing to commit" in stderr.lower()
|
|
|
|
def main():
|
|
"""Main heartbeat function."""
|
|
print(f"=== Allegro Heartbeat for Adagio ===")
|
|
print(f"Time: {datetime.utcnow().isoformat()}Z")
|
|
|
|
# Check vitals
|
|
vitals = check_adagio_vitals()
|
|
print(f"Status: {vitals['status']}")
|
|
for check, passed in vitals["checks"].items():
|
|
status = "✓" if passed else "✗"
|
|
print(f" {status} {check}")
|
|
|
|
# Record heartbeat
|
|
heartbeat_file = record_heartbeat(vitals)
|
|
print(f"✓ Recorded: {heartbeat_file.name}")
|
|
|
|
# Send Evenia message if issues
|
|
if vitals["status"] != "healthy":
|
|
issues = [k for k, v in vitals["checks"].items() if not v]
|
|
msg = f"Adagio, I detect issues: {', '.join(issues)}. Please respond. - Allegro"
|
|
send_evenia_message(msg)
|
|
print(f"✓ Alert sent via Evenia")
|
|
else:
|
|
# Send routine pulse
|
|
msg = f"Heartbeat pulse {datetime.utcnow().strftime('%H:%M')}. All vitals good. - Allegro"
|
|
send_evenia_message(msg)
|
|
|
|
# Commit
|
|
if commit_heartbeat():
|
|
print("✓ Committed to Gitea")
|
|
return 0
|
|
else:
|
|
print("✗ Commit failed")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|