kanban-video-orchestrator

1#!/usr/bin/env python32"""3Monitor a running video-production kanban. Polls `hermes kanban list` and4`events` for a tenant and surfaces issues (stuck tasks, missing heartbeats,5repeated retries, dependency deadlocks).6 7Usage:8    monitor.py --tenant <project-slug> [--interval 30]9 10Outputs a periodic snapshot to stdout. Sends alerts via stderr when issues11are detected. Designed to run alongside the kanban — kill with Ctrl-C when12you're satisfied (or scripted to stop on completion).13 14This is best-effort observability. It does not auto-restart tasks; intervention15decisions should remain human/AI-overseen.16"""17from __future__ import annotations18 19import argparse20import json21import shutil22import subprocess23import sys24import time25from collections import defaultdict26from datetime import datetime, timedelta27 28 29def hermes_available() -> bool:30    return shutil.which("hermes") is not None31 32 33def kanban_list(tenant: str) -> list[dict]:34    """Returns parsed task rows. Falls back to plain stdout parsing if JSON35    output isn't supported by the installed hermes CLI."""36    try:37        out = subprocess.run(38            ["hermes", "kanban", "list", "--tenant", tenant, "--json"],39            capture_output=True, text=True, encoding='utf-8', errors='replace', check=False,40        )41        if out.returncode == 0 and out.stdout.strip().startswith("["):42            return json.loads(out.stdout)43    except (FileNotFoundError, json.JSONDecodeError):44        pass45    # Fallback: textual parse of `hermes kanban list`46    out = subprocess.run(47        ["hermes", "kanban", "list", "--tenant", tenant],48        capture_output=True, text=True, encoding='utf-8', errors='replace', check=False,49    )50    rows = []51    for line in out.stdout.splitlines():52        line = line.strip()53        if not line or line.startswith("#") or "STATUS" in line.upper():54            continue55        parts = line.split()56        if len(parts) >= 4 and parts[0].startswith("t_"):57            rows.append({58                "id": parts[0],59                "status": parts[1] if len(parts) > 1 else "?",60                "assignee": parts[2] if len(parts) > 2 else "?",61                "title": " ".join(parts[3:]) if len(parts) > 3 else "",62                "started_at": None,63                "heartbeat_at": None,64                "max_runtime_s": None,65            })66    return rows67 68 69def kanban_show(task_id: str) -> dict | None:70    out = subprocess.run(71        ["hermes", "kanban", "show", task_id, "--json"],72        capture_output=True, text=True, encoding='utf-8', errors='replace', check=False,73    )74    if out.returncode != 0:75        return None76    try:77        return json.loads(out.stdout)78    except json.JSONDecodeError:79        return None80 81 82def detect_issues(tasks: list[dict]) -> list[str]:83    """Return a list of issue strings, one per concern."""84    now = datetime.now()85    issues: list[str] = []86    by_status = defaultdict(list)87    for t in tasks:88        by_status[t.get("status", "?")].append(t)89 90    # Stuck tasks: RUNNING with no heartbeat in 2 min91    for t in by_status.get("running", []) + by_status.get("RUNNING", []):92        hb = t.get("heartbeat_at")93        if not hb:94            continue95        try:96            hb_dt = datetime.fromisoformat(str(hb).rstrip("Z"))97        except ValueError:98            continue99        if now - hb_dt > timedelta(minutes=2):100            issues.append(101                f"STUCK: {t['id']} ({t.get('assignee', '?')}) — "102                f"no heartbeat in {(now - hb_dt).total_seconds():.0f}s"103            )104 105    # Tasks exceeding max_runtime106    for t in by_status.get("running", []) + by_status.get("RUNNING", []):107        started = t.get("started_at")108        max_rt = t.get("max_runtime_s")109        if not started or not max_rt:110            continue111        try:112            started_dt = datetime.fromisoformat(str(started).rstrip("Z"))113        except ValueError:114            continue115        elapsed = (now - started_dt).total_seconds()116        if elapsed > max_rt:117            issues.append(118                f"OVERTIME: {t['id']} ({t.get('assignee', '?')}) — "119                f"running {elapsed:.0f}s, cap was {max_rt}s"120            )121 122    # Repeated retries123    for t in tasks:124        retries = t.get("retries", 0)125        if retries and retries >= 2:126            issues.append(127                f"FLAPPING: {t['id']} ({t.get('assignee', '?')}) — "128                f"retried {retries}× — fix root cause before next run"129            )130 131    return issues132 133 134def snapshot(tenant: str) -> tuple[list[dict], list[str]]:135    tasks = kanban_list(tenant)136    issues = detect_issues(tasks)137    return tasks, issues138 139 140def print_snapshot(tasks: list[dict], issues: list[str]):141    counts = defaultdict(int)142    for t in tasks:143        counts[str(t.get("status", "?")).lower()] += 1144 145    print(f"\n[{datetime.now().strftime('%H:%M:%S')}] "146          f"Total: {len(tasks)} | "147          + " | ".join(f"{k}: {v}" for k, v in sorted(counts.items())))148 149    for t in tasks:150        bar = "✓" if str(t.get("status", "")).lower() == "done" else \151              "▶" if str(t.get("status", "")).lower() == "running" else \152              "·" if str(t.get("status", "")).lower() == "ready" else \153              "✗" if str(t.get("status", "")).lower() == "failed" else "?"154        print(f"  {bar} {t.get('id', '?'):14} {t.get('assignee', '?'):20}  "155              f"{t.get('title', '')[:60]}")156 157    if issues:158        print("\n  ⚠  ISSUES:", file=sys.stderr)159        for i in issues:160            print(f"     {i}", file=sys.stderr)161 162 163def main():164    ap = argparse.ArgumentParser(description=__doc__,165                                  formatter_class=argparse.RawDescriptionHelpFormatter)166    ap.add_argument("--tenant", required=True,167                    help="Project tenant slug to monitor")168    ap.add_argument("--interval", type=int, default=30,169                    help="Poll interval in seconds (default: 30)")170    ap.add_argument("--once", action="store_true",171                    help="Print one snapshot and exit (no polling loop)")172    args = ap.parse_args()173 174    if not hermes_available():175        print("ERROR: 'hermes' CLI not found in PATH", file=sys.stderr)176        sys.exit(1)177 178    if args.once:179        tasks, issues = snapshot(args.tenant)180        print_snapshot(tasks, issues)181        sys.exit(0 if not issues else 2)182 183    print(f"Monitoring tenant '{args.tenant}' every {args.interval}s. "184          "Ctrl-C to exit.")185    try:186        while True:187            tasks, issues = snapshot(args.tenant)188            print_snapshot(tasks, issues)189            time.sleep(args.interval)190    except KeyboardInterrupt:191        print("\nStopped.")192 193 194if __name__ == "__main__":195    main()196