scripts.swarm_supervisor

Reference › scripts › swarm_supervisor

scripts/swarm_supervisor.py provides heartbeat supervision for delegated swarm tasks: it flags a worktree whose git activity has stalled and can raise an alarm over IRC. A human or the orchestrating session runs it; it has no internal imports. See the pytest live observer and slot_lease for related tooling.

Watches are kept in a state file under a cross-process lock (state_path, _locked, _read_state, _write_state). Subcommands: watch --worktree PATH --task-name NAME with exactly one of --stall-minutes or --stall-seconds (the latter a precision and test seam), plus --irc, --irc-host, --irc-port, --irc-chan and repeatable --watch-path to bound the scan to part of the repository; check evaluates every registered watch once; status prints a table of watches; and unwatch --task-name NAME removes one. compute_heartbeat runs git in the worktree (_git) to find the time of the latest activity, and a watch is stalled when that is older than its limit. With --irc, send_irc_line sends a best-effort single-line message to the channel, and raises on a real connection error.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/swarm_supervisor.py

#!/usr/bin/env python3
"""swarm_supervisor.py — heartbeat supervision for delegated swarm tasks.

A retro of swarm-orchestration cost centers found stalled delegated agents
(Claude subagents + kimi + minimax over IRC #agents) going unnoticed for
hours: nothing watched whether a worktree a task was delegated to was still
making git-visible progress. This is that watchdog: register a worktree +
stall threshold, then evaluate it (by hand, cron, or a loop) via ``check``.

Heartbeat = the most recent of three real-git signals under the worktree:
the git-dir's ``HEAD``/``index`` mtimes, the newest TRACKED file's mtime,
and the last commit's committer time. A stalled agent that edits files but
never commits still counts (mtimes); a commit with an otherwise-clean tree
since still counts (the commit-time floor).

State file: ``~/.local/state/orchestrata/swarm-supervisor.json``,
overridable via ``SWARM_SUPERVISOR_STATE`` (tests point this at a
``tmp_path`` file).

Alarms are edge-triggered: a watch alarms (appends to the alerts log, and
optionally sends one IRC line) the FIRST time it crosses the stall
threshold, then stays quiet on every subsequent ``check`` while still
stalled. A fresh heartbeat (a new commit, a touched tracked file) re-arms it
so the NEXT stall episode alarms again.

CLI:
    python scripts/swarm_supervisor.py watch --worktree /path/to/wt \\
        --task-name build-EFF --stall-minutes 30 --irc
    python scripts/swarm_supervisor.py check
    python scripts/swarm_supervisor.py status
    python scripts/swarm_supervisor.py unwatch --task-name build-EFF

``--stall-seconds`` is a precision/test seam alongside ``--stall-minutes``
(mutually exclusive, one required) — sub-minute thresholds let a test set
up an already-stalled watch without sleeping.
"""

from __future__ import annotations

import argparse
import contextlib
import fcntl
import json
import os
import socket
import subprocess
import sys
import time
from collections.abc import Sequence
from pathlib import Path
from typing import Any

STATE_ENV = "SWARM_SUPERVISOR_STATE"
_DEFAULT_STATE = (
    Path.home() / ".local" / "state" / "orchestrata" / "swarm-supervisor.json"
)

DEFAULT_IRC_HOST = "127.0.0.1"
DEFAULT_IRC_PORT = 6667
DEFAULT_IRC_CHAN = "#agents"
DEFAULT_IRC_NICK = "swarm-supervisor"


def state_path() -> Path:
    override = os.environ.get(STATE_ENV, "").strip()
    return Path(override) if override else _DEFAULT_STATE


# ---------------------------------------------------------------------------
# State I/O — flock-protected read-modify-write (same pattern as slot_lease.py;
# duplicated rather than shared so each tool stays a self-contained, dependency-
# free CLI per the swarm-efficiency-tooling brief).
# ---------------------------------------------------------------------------


@contextlib.contextmanager
def _locked(path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    lock_path = path.with_name(path.name + ".lock")
    with open(lock_path, "a+") as lock_fh:
        fcntl.flock(lock_fh.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock_fh.fileno(), fcntl.LOCK_UN)


def _read_state(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {"watches": {}, "alerts": []}
    try:
        with path.open(encoding="utf-8") as fh:
            data = json.load(fh)
    except json.JSONDecodeError, OSError:
        return {"watches": {}, "alerts": []}
    if not isinstance(data, dict):
        return {"watches": {}, "alerts": []}
    data.setdefault("watches", {})
    data.setdefault("alerts", [])
    return data


def _write_state(path: Path, state: dict[str, Any]) -> None:
    tmp = path.with_name(path.name + ".tmp")
    with tmp.open("w", encoding="utf-8") as fh:
        json.dump(state, fh, indent=2, sort_keys=True)
    tmp.replace(path)


# ---------------------------------------------------------------------------
# Heartbeat — real git subprocess calls only
# ---------------------------------------------------------------------------


def _git(worktree: Path, *args: str, timeout: float = 10.0) -> str:
    result = subprocess.run(
        ["git", "-C", str(worktree), *args],
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"git {' '.join(args)} failed in {worktree}: {result.stderr.strip()}"
        )
    return result.stdout.strip()


def compute_heartbeat(
    worktree: Path,
    *,
    paths: Sequence[str] | None = None,
    timeout: float = 10.0,
) -> float:
    """Most recent of: git-dir HEAD/index mtime, newest tracked-file mtime,
    last commit's committer time. Raises ``RuntimeError`` when ``worktree``
    isn't a directory or carries no git signal at all (not a repo, or an
    unreadable git-dir) — a watch on a bad path must surface loudly, not
    silently read as "never stalls".

    ``paths`` (repo-relative) bounds the tracked-file scan to those paths via
    ``git ls-files -- <paths>`` instead of the whole tree — a worktree here
    is typically a full project checkout (~10k tracked files), and stat()ing
    every one of them on every ``check`` is wasted work when a watch only
    cares about, say, ``src/`` for one delegated task. ``None``/empty
    preserves the original whole-tree behavior.

    ``timeout`` is a precision/test seam (mirrors ``--stall-seconds``
    alongside ``--stall-minutes``) — passed straight through to every
    ``_git`` call, letting a test force a REAL ``subprocess.TimeoutExpired``
    without waiting out the 10s production default.
    """
    worktree = Path(worktree)
    if not worktree.is_dir():
        raise RuntimeError(f"worktree does not exist: {worktree}")

    git_dir_raw = _git(worktree, "rev-parse", "--git-dir", timeout=timeout)
    git_dir = Path(git_dir_raw)
    if not git_dir.is_absolute():
        git_dir = (worktree / git_dir).resolve()

    candidates: list[float] = []
    for name in ("HEAD", "index"):
        p = git_dir / name
        if p.exists():
            candidates.append(p.stat().st_mtime)

    ls_files_args = ("ls-files", "--", *paths) if paths else ("ls-files",)
    for rel in _git(worktree, *ls_files_args, timeout=timeout).splitlines():
        fp = worktree / rel
        try:
            candidates.append(fp.stat().st_mtime)
        except OSError:
            continue

    try:
        commit_ts_raw = _git(worktree, "log", "-1", "--format=%ct", timeout=timeout)
    except RuntimeError:
        commit_ts_raw = ""  # no commits yet — not an error, just no signal
    if commit_ts_raw:
        candidates.append(float(commit_ts_raw))

    if not candidates:
        raise RuntimeError(f"no git heartbeat signal found under {worktree}")
    return max(candidates)


# ---------------------------------------------------------------------------
# IRC — minimal, dependency-free (mirrors _irc_send.py's connect/NICK/USER/
# JOIN/PRIVMSG/QUIT sequence; deliberately does NOT import the project's
# _irc_conversation capture helper).
# ---------------------------------------------------------------------------


def _drain(sock: socket.socket) -> None:
    with contextlib.suppress(OSError):
        sock.recv(4096)


def send_irc_line(
    text: str,
    *,
    host: str = DEFAULT_IRC_HOST,
    port: int = DEFAULT_IRC_PORT,
    chan: str = DEFAULT_IRC_CHAN,
    nick: str = DEFAULT_IRC_NICK,
    timeout: float = 2.0,
) -> None:
    """Best-effort single-line IRC PRIVMSG. Raises on a real connection
    failure — callers decide whether that should abort or just be logged
    (``run_check`` logs it, never lets it prevent recording the alert)."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((host, port))
        sock.sendall(f"NICK {nick}\r\n".encode())
        sock.sendall(f"USER {nick} 0 * :swarm-supervisor\r\n".encode())
        time.sleep(0.2)
        _drain(sock)
        sock.sendall(f"JOIN {chan}\r\n".encode())
        time.sleep(0.2)
        _drain(sock)
        sock.sendall(f"PRIVMSG {chan} :{text}\r\n".encode())
        time.sleep(0.2)
        _drain(sock)
        sock.sendall(b"QUIT :done\r\n")
    finally:
        sock.close()


# ---------------------------------------------------------------------------
# Watch registry
# ---------------------------------------------------------------------------


def register_watch(
    task_name: str,
    worktree: str | Path,
    stall_seconds: float,
    *,
    irc: bool = False,
    irc_host: str = DEFAULT_IRC_HOST,
    irc_port: int = DEFAULT_IRC_PORT,
    irc_chan: str = DEFAULT_IRC_CHAN,
    paths: Sequence[str] | None = None,
    git_timeout: float | None = None,
    path: Path | None = None,
    now: float | None = None,
) -> None:
    p = path or state_path()
    ts = now if now is not None else time.time()
    with _locked(p):
        state = _read_state(p)
        state["watches"][task_name] = {
            "worktree": str(Path(worktree).resolve()),
            "stall_seconds": stall_seconds,
            "irc": irc,
            "irc_host": irc_host,
            "irc_port": irc_port,
            "irc_chan": irc_chan,
            # Repo-relative paths bounding compute_heartbeat's tracked-file
            # scan (F11 perf fix) — [] preserves the original whole-tree scan.
            "paths": list(paths) if paths else [],
            # Per-watch git-subprocess timeout override — None means
            # compute_heartbeat's own 10s default. Precision/test seam (F11):
            # lets a test force a REAL subprocess.TimeoutExpired.
            "git_timeout": git_timeout,
            "registered_at": ts,
            "alarmed": False,
            "last_heartbeat": None,
            "last_checked": None,
        }
        _write_state(p, state)


def unwatch(task_name: str, *, path: Path | None = None) -> bool:
    p = path or state_path()
    with _locked(p):
        state = _read_state(p)
        existed = task_name in state["watches"]
        state["watches"].pop(task_name, None)
        _write_state(p, state)
    return existed


def run_check(
    *, path: Path | None = None, now: float | None = None
) -> list[dict[str, Any]]:
    """Evaluate every registered watch once. Returns the NEW alerts raised
    THIS call (empty when nothing newly crossed into stall)."""
    p = path or state_path()
    ts = now if now is not None else time.time()
    new_alerts: list[dict[str, Any]] = []
    with _locked(p):
        state = _read_state(p)
        for task_name, watch in state["watches"].items():
            try:
                heartbeat = compute_heartbeat(
                    Path(watch["worktree"]),
                    paths=watch.get("paths") or None,
                    timeout=watch.get("git_timeout") or 10.0,
                )
            except (RuntimeError, subprocess.SubprocessError, OSError) as exc:
                # subprocess.SubprocessError (incl. TimeoutExpired) / OSError:
                # a hung or missing `git` binary used to escape uncaught here,
                # skipping _write_state entirely (it's called once AFTER this
                # loop) — one bad watch silently dropped every OTHER watch's
                # freshly-computed heartbeat on the floor too (F11).
                watch["last_error"] = str(exc)
                continue
            watch["last_error"] = None
            watch["last_heartbeat"] = heartbeat
            watch["last_checked"] = ts
            stalled = (ts - heartbeat) > watch["stall_seconds"]
            if stalled and not watch.get("alarmed"):
                alert = {
                    "task_name": task_name,
                    "at": ts,
                    "heartbeat": heartbeat,
                    "stalled_for_seconds": round(ts - heartbeat, 1),
                }
                state["alerts"].append(alert)
                new_alerts.append(alert)
                watch["alarmed"] = True
                if watch.get("irc"):
                    text = (
                        f"STALL ALERT: task {task_name!r} in {watch['worktree']} has had "
                        f"no git activity for {alert['stalled_for_seconds']:.0f}s "
                        f"(threshold {watch['stall_seconds']:.0f}s)"
                    )
                    try:
                        send_irc_line(
                            text,
                            host=watch.get("irc_host", DEFAULT_IRC_HOST),
                            port=watch.get("irc_port", DEFAULT_IRC_PORT),
                            chan=watch.get("irc_chan", DEFAULT_IRC_CHAN),
                        )
                    except OSError as exc:
                        watch["last_irc_error"] = str(exc)
            elif not stalled:
                watch["alarmed"] = False
        _write_state(p, state)
    return new_alerts


def format_status(state: dict[str, Any], now: float) -> str:
    watches = state.get("watches", {})
    if not watches:
        return "(no watches registered)"
    lines = [
        f"{'TASK':<24}{'STALL(s)':<10}{'STATUS':<10}{'LAST HEARTBEAT':<18}WORKTREE"
    ]
    for name in sorted(watches):
        w = watches[name]
        hb = w.get("last_heartbeat")
        age = f"{now - hb:.0f}s ago" if hb is not None else "unknown"
        status = "STALLED" if w.get("alarmed") else "ok"
        lines.append(
            f"{name:<24}{w['stall_seconds']:<10.0f}{status:<10}{age:<18}{w['worktree']}"
        )
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Heartbeat supervision for delegated swarm tasks — flags git-"
            "activity stalls in a worktree, optionally alarms over IRC."
        )
    )
    sub = parser.add_subparsers(dest="command", required=True)

    w = sub.add_parser("watch", help="Register a heartbeat watch for a worktree")
    w.add_argument("--worktree", required=True)
    w.add_argument("--task-name", required=True)
    stall_group = w.add_mutually_exclusive_group(required=True)
    stall_group.add_argument("--stall-minutes", type=float)
    stall_group.add_argument(
        "--stall-seconds",
        type=float,
        help="Precision/test seam alongside --stall-minutes — see module docstring.",
    )
    w.add_argument("--irc", action="store_true", help="Send an IRC line on stall")
    w.add_argument("--irc-host", default=DEFAULT_IRC_HOST)
    w.add_argument("--irc-port", type=int, default=DEFAULT_IRC_PORT)
    w.add_argument("--irc-chan", default=DEFAULT_IRC_CHAN)
    w.add_argument(
        "--watch-path",
        action="append",
        dest="watch_paths",
        default=None,
        help="Repo-relative path to bound the heartbeat file scan to "
        "(repeatable). Default: scan the whole worktree.",
    )

    sub.add_parser("check", help="Evaluate every registered watch once")
    sub.add_parser("status", help="Print a table of registered watches")

    u = sub.add_parser("unwatch", help="Remove a watch")
    u.add_argument("--task-name", required=True)

    args = parser.parse_args(argv)

    if args.command == "watch":
        stall_seconds = (
            args.stall_seconds
            if args.stall_seconds is not None
            else args.stall_minutes * 60.0
        )
        register_watch(
            args.task_name,
            args.worktree,
            stall_seconds,
            irc=args.irc,
            irc_host=args.irc_host,
            irc_port=args.irc_port,
            irc_chan=args.irc_chan,
            paths=args.watch_paths,
        )
        print(
            f"[swarm_supervisor] watching {args.task_name!r} @ {args.worktree} "
            f"(stall threshold {stall_seconds:.0f}s"
            + (", IRC on stall)" if args.irc else ")")
        )
        return 0

    if args.command == "check":
        new_alerts = run_check()
        new_task_names = {alert["task_name"] for alert in new_alerts}
        state = _read_state(state_path())
        stalled_now = {
            name: w for name, w in state["watches"].items() if w.get("alarmed")
        }
        # Edge-triggering only governs the alert LOG / IRC ping (avoid
        # re-paging every cron tick for the same stall episode) -- the
        # check's own reported status must reflect CURRENT state every
        # time, or a still-stalled task silently reads as "all watches OK"
        # on the second and every later poll.
        if stalled_now:
            for name in sorted(stalled_now):
                w = stalled_now[name]
                tag = "NEW STALL" if name in new_task_names else "STILL STALLED"
                heartbeat = w.get("last_heartbeat")
                age = (
                    f"{time.time() - heartbeat:.0f}s"
                    if heartbeat is not None
                    else "unknown"
                )
                print(f"[swarm_supervisor] {tag} {name!r}: {age} since last heartbeat")
            return 1
        print("[swarm_supervisor] all watches OK")
        return 0

    if args.command == "status":
        print(format_status(_read_state(state_path()), time.time()))
        return 0

    if args.command == "unwatch":
        existed = unwatch(args.task_name)
        print(
            f"[swarm_supervisor] {'removed' if existed else 'no such watch:'} {args.task_name!r}"
        )
        return 0 if existed else 1

    return 1


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