The script that wakes up a frozen AI agent.

#!/usr/bin/env python3
"""stall_guard.py: checks every AI agent on your Mac every 5 minutes so none of them sits frozen all night.

Why: one of my agents froze for 4.5 hours overnight with no error. It had started a tool in the
background, macOS paused that tool (it tried to read the terminal), and the agent just waited on it.

Three checks, every run:
  It only touches processes an AI agent started (anything under claude or codex, see AGENT_PARENTS),
  so a program you paused on purpose is left alone.

  1. FROZEN   an agent, or a process an agent started, stopped (state T) for 2 runs in a row -> resume it (SIGCONT).
              If the same tool freezes again after a resume, stop it (SIGTERM) so the agent gets an
              error and moves on. Agents themselves are only ever resumed, never killed.
  2. HUNG     a tool (ffmpeg, yt-dlp, whisper...) an agent started, at least 40 min old, whose CPU time
              has not moved for 3 runs in a row (15 min) -> stop it. Network tools get 6 runs (30 min).
              So the earliest a hung tool gets stopped is about 55 min after it started.
  3. WAITING  a tmux pane showing a yes/no or permission prompt that has not changed for 20+ min
              -> alert. It never answers for you.

Every action goes to events.jsonl. Alerts go out as ONE message: right away between 8 AM and 11 PM,
night alerts wait for the morning. Pause it any time: touch ~/.stall-guard/PAUSE

Setup (edit the lists below to match your tools):
    python3 stall_guard.py --dry-run      # prints what it would do; sends no signals and no alerts
                                          # (it still writes its own notes in ~/.stall-guard)
    then load the launchd file (com.you.stallguard.plist) to run it every 5 minutes.

Alerts: set STALL_ALERT_CMD to any command that takes the message as its last argument. Under launchd,
set it in the plist (EnvironmentVariables), not in your terminal. If it is not set, you get a macOS
notification instead. Texting your phone needs your own script (iMessage, Pushover, ntfy...).
"""
import argparse, datetime as dt, fcntl, hashlib, json, os, re, shlex, shutil, signal, subprocess, time

# ---- edit these ----
TOOL_NAMES = ["ffmpeg", "ffprobe", "yt-dlp", "whisper", "rclone"]   # helper tools that may hang
IO_TOOLS = ["yt-dlp", "rclone", "whisper"]                             # slow network/disk tools get 30 min
AGENT_NAMES = ["claude", "codex", "node", "tmux", "launchd"]           # never killed, only resumed
AGENT_PARENTS = ["claude", "codex"]    # the guard only touches these and processes started under them
TOOL_MIN_AGE = 40 * 60       # a tool must be this old before it can count as hung
WAIT_ALERT_SECS = 20 * 60    # prompt unchanged this long -> alert
DAY_START, DAY_END = 8, 23   # texts only in these hours (local time)
# --------------------

ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true", help="report only, send no signals and no alerts")
a = ap.parse_args()

D = os.path.expanduser("~/.stall-guard")
os.makedirs(D, exist_ok=True)
STATE, EV = f"{D}/state.json", f"{D}/events.jsonl"
if os.path.exists(f"{D}/PAUSE"):
    raise SystemExit(0)
lock = open(f"{D}/.lock", "w")
try:
    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)   # one run at a time
except OSError:
    raise SystemExit(0)

TOOLS_RE = re.compile(r"^(" + "|".join(map(re.escape, TOOL_NAMES)) + r")$")
AGENT_RE = re.compile(r"^(" + "|".join(map(re.escape, AGENT_NAMES)) + r")$")
PROMPT = re.compile(r"(Do you want to (proceed|make this edit|create)|Allow (once|always)|\(y/n\)|"
                    r"Press Enter to continue|Waiting for your (answer|input))", re.I)
TMUX = shutil.which("tmux")


def is_tool(cmd):
    """Match the program itself (or the script python/bash runs), never a shell that only mentions it."""
    parts = cmd.split()
    names = [os.path.basename(parts[0])] if parts else []
    if names and re.match(r"^(python3?(\.\d+)?|Python|bash|sh|env)$", names[0]):
        rest = [x for x in parts[1:6] if not x.startswith("-") and "=" not in x]
        if rest:
            names.append(os.path.basename(rest[0]))
    return any(TOOLS_RE.match(n) for n in names)


def tool_name(cmd):
    parts = cmd.split()
    names = [os.path.basename(parts[0])] if parts else []
    if names and re.match(r"^(python3?(\.\d+)?|Python|bash|sh|env)$", names[0]):
        rest = [x for x in parts[1:6] if not x.startswith("-") and "=" not in x]
        if rest:
            names.append(os.path.basename(rest[0]))
    return next((n for n in names if TOOLS_RE.match(n)), "")


def is_agent(cmd):
    exe = os.path.basename(cmd.split()[0]) if cmd.split() else ""
    return not is_tool(cmd) and (bool(AGENT_RE.match(exe)) or ".app/" in cmd.split()[0])


def run(cmd):
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=20, stdin=subprocess.DEVNULL)
        return r.stdout if r.returncode == 0 else ""
    except Exception:
        return ""


def ev(kind, **kw):
    kw.update(kind=kind, at=dt.datetime.now().isoformat(timespec="seconds"), dry_run=a.dry_run)
    with open(EV, "a") as f:
        f.write(json.dumps(kw) + "\n")
    if a.dry_run:
        print("WOULD:", kind, kw.get("cmd") or kw.get("session") or "")
    return kw


def secs(t):
    m = re.match(r"(?:(\d+)-)?(?:(\d+):)?(\d+):(\d+)(?:\.\d+)?$", t.strip())
    if not m:
        return 0
    d, h, mi, s = (int(g) if g else 0 for g in m.groups())
    return d * 86400 + h * 3600 + mi * 60 + s


def sig(pid, s, expect_cmd):
    """Signal only if the PID still runs the same command (guards against PID reuse)."""
    if a.dry_run:
        return True
    try:
        if run(["/bin/ps", "-o", "command=", "-p", str(pid)]).strip() != expect_cmd:
            return False
        os.kill(int(pid), s)
        return True
    except Exception as e:
        ev("signal_failed", pid=pid, err=str(e)[:120])
        return False


def short(cmd):
    return re.sub(r"\s+", " ", cmd)[:160]


now = time.time()
try:
    st = json.load(open(STATE))
    assert isinstance(st, dict)
except Exception:
    st = {}
frozen, cpu, resumed, panes, logged = (st.get(k) if isinstance(st.get(k), dict) else {}
                                       for k in ("frozen", "cpu", "resumed", "panes", "logged"))
pending = st.get("pending") if isinstance(st.get("pending"), list) else []

procs = {}
for line in run(["/bin/ps", "-U", str(os.getuid()), "-o", "pid=,ppid=,state=,etime=,time=,command="]).splitlines():
    p = line.split(None, 5)
    if len(p) == 6:
        procs[p[0]] = dict(ppid=p[1], state=p[2], age=secs(p[3]), cpu=secs(p[4]), cmd=p[5])
kids = {}
for pid, p in procs.items():
    kids.setdefault(p["ppid"], []).append(pid)


PARENT_RE = re.compile(r"^(" + "|".join(map(re.escape, AGENT_PARENTS)) + r")$")


def from_agent(pid):
    """True if this process is, or was started under, one of AGENT_PARENTS."""
    seen = set()
    while pid in procs and pid not in seen:
        seen.add(pid)
        parts = procs[pid]["cmd"].split()
        if parts and PARENT_RE.match(os.path.basename(parts[0])):
            return True
        pid = procs[pid]["ppid"]
    return False


def tree_cpu(pid):
    tot, stack = 0, [pid]
    while stack:
        x = stack.pop()
        tot += procs.get(x, {}).get("cpu", 0)
        stack += kids.get(x, [])
    return tot


def key_of(pid, cmd):
    return pid + ":" + hashlib.md5(cmd.encode()).hexdigest()[:8]


alerts = []

# 1. frozen processes
seen = {}
for pid, p in procs.items():
    if not p["state"].startswith("T") or p["age"] < 120 or not from_agent(pid):
        continue
    k = key_of(pid, p["cmd"])
    n = frozen.get(k, 0) + 1
    seen[k] = n
    if n < 2:
        continue
    if is_agent(p["cmd"]):
        if sig(pid, signal.SIGCONT, p["cmd"]):
            resumed[k] = now
        if k not in logged:
            logged[k] = now
            alerts.append(ev("frozen_agent_resumed", pid=pid, cmd=short(p["cmd"])))
    elif k in resumed and is_tool(p["cmd"]):
        if sig(pid, signal.SIGTERM, p["cmd"]):
            alerts.append(ev("frozen_again_tool_stopped", pid=pid, cmd=short(p["cmd"])))
    else:
        if sig(pid, signal.SIGCONT, p["cmd"]):
            resumed[k] = now
        ev("frozen_resumed", pid=pid, cmd=short(p["cmd"]))
frozen = seen

# 2. hung tools (no CPU movement)
new_cpu = {}
for pid, p in procs.items():
    if not is_tool(p["cmd"]) or is_agent(p["cmd"]) or p["age"] < TOOL_MIN_AGE or not from_agent(pid):
        continue
    k = key_of(pid, p["cmd"])
    c = tree_cpu(pid)
    prev = cpu.get(k, {"c": -1, "still": 0})
    still = prev["still"] + 1 if abs(c - prev["c"]) < 1 else 0
    new_cpu[k] = {"c": c, "still": still}
    io = tool_name(p["cmd"]) in IO_TOOLS
    if still >= (6 if io else 3) and sig(pid, signal.SIGTERM, p["cmd"]):
        alerts.append(ev("hung_tool_stopped", pid=pid, age_min=p["age"] // 60, cmd=short(p["cmd"])))
cpu = new_cpu

# 3. sessions waiting on a prompt (tmux only; skipped if tmux is not installed)
new_panes = {}
if not TMUX and a.dry_run:
    print("note: tmux not found, so the 'waiting on you' check is off")
if TMUX:
    for s in run([TMUX, "list-panes", "-a", "-F", "#{session_name}:#{window_index}.#{pane_index}"]).split():
        txt = run([TMUX, "capture-pane", "-p", "-t", s])
        h = hashlib.md5(txt.encode()).hexdigest()
        prev = panes.get(s, {})
        since = prev["since"] if prev.get("h") == h else now
        new_panes[s] = {"h": h, "since": since}
        tail = "\n".join(txt.strip().splitlines()[-15:])
        if PROMPT.search(tail) and now - since > WAIT_ALERT_SECS:
            k = f"{s}:{h}"
            if k not in logged:
                logged[k] = now
                alerts.append(ev("session_waiting_on_prompt", session=s, idle_min=int((now - since) // 60)))
panes = new_panes

# alerts: one message in the daytime, night ones wait for morning
pending += alerts
hour = dt.datetime.now().hour
if pending and DAY_START <= hour < DAY_END and not a.dry_run:
    words = {"session_waiting_on_prompt": "waiting on you", "hung_tool_stopped": "hung tool stopped",
             "frozen_again_tool_stopped": "freezing tool stopped", "frozen_agent_resumed": "frozen agent resumed"}
    lines = [f"{words.get(x['kind'], x['kind'])}: {(x.get('session') or x.get('cmd', ''))[:50]}" for x in pending[-5:]]
    more = f"\n+{len(pending) - 5} more" if len(pending) > 5 else ""
    msg = f"Stall guard, {len(pending)} item(s)\n" + "\n".join(lines) + more
    custom = os.environ.get("STALL_ALERT_CMD")
    cmd = shlex.split(custom) + [msg] if custom else \
        ["osascript", "-e", f"display notification {json.dumps(msg)} with title \"Stall guard\""]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=60, stdin=subprocess.DEVNULL)
    if r.returncode == 0:
        pending = []          # cleared only after a real send
    else:
        ev("alert_failed", err=(r.stderr or "")[-200:])

if not a.dry_run:
    pending = pending[-20:]
    logged = {k: v for k, v in logged.items() if now - v < 86400}
    resumed = {k: v for k, v in resumed.items() if now - v < 86400 and k.split(":")[0] in procs}
    json.dump(dict(frozen=frozen, cpu=cpu, resumed=resumed, panes=panes, logged=logged, pending=pending),
              open(STATE + ".tmp", "w"))
    os.replace(STATE + ".tmp", STATE)
else:
    print(f"dry run done: {len(procs)} processes checked, {len(alerts)} alert(s) would be sent")

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<!--
  Runs stall_guard.py every 5 minutes.
  1. Replace YOURNAME with your Mac username (run: whoami) and fix the script path if you put it somewhere else.
  2. Check your python path with: which python3
  3. cp com.you.stallguard.plist ~/Library/LaunchAgents/
     launchctl load ~/Library/LaunchAgents/com.you.stallguard.plist
  Stop it:  launchctl unload ~/Library/LaunchAgents/com.you.stallguard.plist
-->
<plist version="1.0"><dict>
  <key>Label</key><string>com.you.stallguard</string>
  <key>ProgramArguments</key><array>
    <string>/usr/bin/python3</string>
    <string>/Users/YOURNAME/stall-guard/stall_guard.py</string>
  </array>
  <key>EnvironmentVariables</key><dict>
    <!-- launchd does not read your shell profile: give it a PATH that finds tmux, ffmpeg and friends -->
    <key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
    <!-- optional: your own alert script, called with the message as its last argument -->
    <!-- <key>STALL_ALERT_CMD</key><string>/Users/YOURNAME/bin/text-me</string> -->
  </dict>
  <key>StartInterval</key><integer>300</integer>
  <key>RunAtLoad</key><true/>
  <key>StandardInputPath</key><string>/dev/null</string>
  <key>StandardOutPath</key><string>/tmp/stallguard.out.log</string>
  <key>StandardErrorPath</key><string>/tmp/stallguard.err.log</string>
</dict></plist>