#!/usr/bin/env python3 """CLI for agent-reaper: list flagged candidates, status, manual kill. `kill` always re-verifies the candidate criteria against the LIVE process before signaling -- a pid can be reused, or the agent may have resumed progress since the last periodic scan. """ import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import _agent_reaper_lib as lib def cmd_list(): state = lib.load_state() flagged = state.get("flagged", {}) if not flagged: print("no flagged candidates") return 0 for pid_key, entry in sorted(flagged.items(), key=lambda kv: int(kv[0])): if entry.get("kind") == "leaked-test-orphan": age_m = (entry.get("age_s") or 0) / 60 print(f"PID {pid_key} verdict={entry['verdict']} comm={entry['comm']} " f"orphaned={age_m:.0f}m reason={entry.get('reason', '?')} " f"ppid={entry['ppid']}") else: age_m = (entry.get("transcript_age_s") or 0) / 60 print(f"PID {pid_key} verdict={entry['verdict']} comm={entry['comm']} " f"stale={age_m:.0f}m cpu={entry.get('pcpu_now', 0):.0f}% " f"tty_nr={entry['tty_nr']} ppid={entry['ppid']}") return 0 def cmd_status(): state = lib.load_state() pids_tracked = len(state.get("pids", {})) flagged = state.get("flagged", {}) enforce = os.environ.get("ENFORCE") == "1" print(f"tracked pids: {pids_tracked}") print(f"flagged candidates: {len(flagged)}") print(f"ENFORCE: {'ON' if enforce else 'off (warn-only)'}") print(f"state file: {lib.STATE_PATH}") print(f"log file: {lib.LOG_PATH}") return 0 def cmd_kill(pid_arg): try: pid = int(pid_arg) except ValueError: print(f"invalid pid: {pid_arg}", file=sys.stderr) return 2 state = lib.load_state() verified = lib.verify_live_before_kill(pid, state) if verified is None: print(f"refused: pid {pid} not found in tracked history, or was reused (identity mismatch)", file=sys.stderr) return 1 if verified["verdict"] not in ("WARN", "ENFORCE_ELIGIBLE"): print(f"refused: pid {pid} no longer meets candidate criteria live " f"(verdict={verified['verdict']}) -- criteria no longer hold", file=sys.stderr) print(json.dumps(verified, default=str), file=sys.stderr) return 1 outcome = lib.kill_ladder(pid, verified) print(f"pid {pid}: {outcome}") return 0 if outcome in ("terminated", "killed") else 1 def cmd_reap(pid_arg): try: pid = int(pid_arg) except ValueError: print(f"invalid pid: {pid_arg}", file=sys.stderr) return 2 state = lib.load_state() verified = lib.verify_test_orphan_before_kill(pid, state) if verified is None: print(f"refused: pid {pid} not found in tracked history, or was reused (identity mismatch)", file=sys.stderr) return 1 if verified["verdict"] != "TEST_ORPHAN": print(f"refused: pid {pid} no longer a leaked test orphan live " f"(verdict={verified['verdict']})", file=sys.stderr) return 1 outcome = lib.reap_tree(pid, verified) print(f"pid {pid} (tree): {outcome}") return 0 if outcome == "reaped" else 1 def cmd_unblock(pid_arg): """Kill the blocked tool-command subtree under a stuck agent turn -- the agent itself is never signaled; its turn resumes with a failed tool result.""" try: pid = int(pid_arg) except ValueError: print(f"invalid pid: {pid_arg}", file=sys.stderr) return 2 state = lib.load_state() verified = lib.verify_blocked_turn_before_kill(pid, state) if verified is None: print(f"refused: pid {pid} not a live blocked turn (resumed, exited, or identity mismatch)", file=sys.stderr) return 1 outcome = lib.reap_tree(verified["child_pid"], verified) print(f"agent {pid}: blocked child {verified['child_pid']} (tree): {outcome}") return 0 if outcome == "reaped" else 1 def main(): argv = sys.argv[1:] if not argv: print(__doc__) return 2 cmd = argv[0] if cmd == "list": return cmd_list() if cmd == "status": return cmd_status() if cmd == "kill": if len(argv) < 2: print("usage: reaper-ctl kill ", file=sys.stderr) return 2 return cmd_kill(argv[1]) if cmd == "reap": if len(argv) < 2: print("usage: reaper-ctl reap ", file=sys.stderr) return 2 return cmd_reap(argv[1]) if cmd == "unblock": if len(argv) < 2: print("usage: reaper-ctl unblock ", file=sys.stderr) return 2 return cmd_unblock(argv[1]) if cmd == "--self-test": return 0 if self_test() else 1 print(f"unknown command: {cmd}", file=sys.stderr) return 2 def self_test(): import shutil import tempfile import time as _time tmpdir = tempfile.mkdtemp(prefix="reaper-ctl-selftest-") orig_state_path = lib.STATE_PATH orig_log_path = lib.LOG_PATH try: lib.STATE_PATH = os.path.join(tmpdir, "state.json") lib.LOG_PATH = os.path.join(tmpdir, "log.jsonl") now = _time.time() state = { "pids": { "999999": { "starttime": 4242, "pcpu_history": [10.0, 10.0, 10.0], "io_history": [{"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 10}], "last_seen": now, } }, "flagged": { "999999": {"verdict": "WARN", "comm": "claude", "cmdline": "claude", "ppid": 500, "tty_nr": 34, "transcript_age_s": lib.STALE_MIN_S + 60, "pcpu_now": 55.0} }, } lib.save_state(state) rc = cmd_list() assert rc == 0 rc = cmd_status() assert rc == 0 rc = cmd_kill("999999") assert rc == 1, "kill of a nonexistent live pid must refuse, not crash" rc = cmd_reap("999998") assert rc == 1, "reap of an untracked/dead pid must refuse, not crash" rc = cmd_unblock("999997") assert rc == 1, "unblock of an untracked/dead agent must refuse, not crash" print("reaper-ctl self-test OK", file=sys.stderr) return True finally: lib.STATE_PATH = orig_state_path lib.LOG_PATH = orig_log_path shutil.rmtree(tmpdir, ignore_errors=True) if __name__ == "__main__": sys.exit(main())