`snippets/agent_status_check.py`

81 lines
#!/usr/bin/env python3
"""
agent_status_check.py — Report the status of running AI agents.

A lightweight observability tool that checks for running agent processes
and reports their status in a human-readable format. Designed to make
invisible agent activity visible at a glance.

Usage:
    python3 snippets/agent_status_check.py
    python3 snippets/agent_status_check.py --json
"""

from __future__ import annotations

import json
import subprocess
import sys
from datetime import datetime, timezone


def get_agent_processes() -> list[dict]:
    """Discover running agent-like processes via pm2 or process list."""
    agents = []
    try:
        result = subprocess.run(
            ["pm2", "jlist"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            for proc in json.loads(result.stdout):
                agents.append({
                    "name": proc.get("name", "unknown"),
                    "status": proc.get("pm2_env", {}).get("status", "unknown"),
                    "uptime_ms": proc.get("pm2_env", {}).get("pm_uptime", 0),
                    "restarts": proc.get("pm2_env", {}).get("restart_time", 0),
                    "pid": proc.get("pid", None),
                    "memory_mb": round(
                        proc.get("monit", {}).get("memory", 0) / 1_048_576, 1
                    ),
                })
    except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError):
        pass
    return agents


def print_status(agents: list[dict], as_json: bool = False) -> None:
    """Print agent status in human-readable or JSON format."""
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")

    if as_json:
        print(json.dumps({"timestamp": now, "agents": agents}, indent=2))
        return

    print(f"Agent Status Report — {now}")
    print("=" * 60)
    if not agents:
        print("No agent processes detected.")
        return

    for a in agents:
        icon = "ONLINE" if a["status"] == "online" else "OFFLINE"
        print(
            f"  [{icon}] {a['name']:30s}  "
            f"PID {a['pid'] or '—':>6}  "
            f"MEM {a['memory_mb']:>6.1f} MB  "
            f"restarts={a['restarts']}"
        )
    print(f"\nTotal agents: {len(agents)}")


def main() -> int:
    as_json = "--json" in sys.argv
    agents = get_agent_processes()
    print_status(agents, as_json=as_json)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())