#!/usr/bin/env python3
"""
Periodic candidate scanner. Runs as USER via agent-reaper.timer (60s).

Five candidate classes:
  1. runaway agent -- NO-FORWARD-PROGRESS (transcript mtime AND /proc/pid/io
     both frozen >=STALE_MIN), never CPU% alone. Notify-only by default.
  2. leaked test-run orphan -- a non-agent test subprocess (dev server,
     sleep-infinity fixture) reparented to init/user-systemd after its run
     died, under a test path, idle past TEST_ORPHAN_MIN_AGE. STRUCTURAL, not
     resource-based: these are idle, so class 1's discriminator never sees them.
     Notify-only.
  3. blocked turn -- an agent whose transcript is frozen >=BLOCKED_TURN_MIN
     while a Bash tool child sits in a low-CPU wait (gh run watch, wedged test,
     unbounded poll). Under ENFORCE=1 the child subtree is reaped silently
     (never the agent -- the turn resumes with a failed tool result) and
     journaled to the reaper log. Bounded waits (unexpired `timeout N` in the
     child cmdline) are exempt. Disable with REAPER_UNBLOCK=0.

  4. orphaned sidecar -- a per-session helper daemon (quietmode) reparented
     after its agent session died. Its cmdline carries the agent's own path
     tokens, so class 1 claims it and files it as idle STALE_LOW_CPU forever;
     these accumulated GBs of swap across sessions. Reaped silently under
     ENFORCE=1 -- a sidecar without a session serves nobody. Disable with
     REAPER_SIDECAR=0.
  5. orphaned runtime tree -- a leaked run's whole tree (wrangler/vite/astro/
     esbuild/e2e-remote). Only its ROOT reparents to user-systemd; descendants
     keep live parents inside the dead tree, so class 2's per-process check
     skips them. Ownership is decided by walking the ancestor chain to
     init/user-systemd looking for an agent or a controlling terminal.
     Reaped at the root past ORPHAN_RUNTIME_MIN_AGE under ENFORCE=1.
     Disable with REAPER_ORPHAN_RUNTIME=0.

Only WRITES candidate state; actionable notifications are rendered by the
separate reaper-notifier.py daemon, which offers a one-click reap.

ENFORCE=1 (off by default) is the master switch: with it unset no path signals
any process, and the scan only records candidate state. Under ENFORCE=1 class 1
kills additionally require ppid==1 and tty_nr==0 on top of the stale+cpu-sustained
stack -- absence of a tty is never by itself a reason to kill. Class 2 is never
auto-reaped. The per-class REAPER_* flags narrow enforcement further; they never
widen it.
"""
import os
import sys
import time

sys.dont_write_bytecode = True
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _agent_reaper_lib as lib


def scan_once():
    state = lib.load_state()
    exempt = lib.dispatcher_exempt_pids()
    pids = lib.list_candidate_pids()
    uptime_s = lib.read_uptime_s()
    child_map = lib.build_child_map()

    results = []
    runtime_roots = set()
    for pid in pids:
        snapshot = lib.take_live_snapshot(pid)
        if snapshot is None:
            continue
        sidecar = lib.evaluate_orphaned_sidecar(snapshot, exempt_pids=exempt,
                                                uptime_s=uptime_s)
        if sidecar["verdict"] == "ORPHANED_SIDECAR":
            lib.update_history(state, snapshot)
            results.append(sidecar)
            continue
        if (snapshot["tty_nr"] == 0
                and lib.is_orphan_runtime(snapshot["cmdline"])
                and not lib.is_agent_session(snapshot["comm"], snapshot["cmdline"])):
            root = lib.orphan_tree_root(pid)
            if root is not None:
                runtime_roots.add(root)
        if lib.is_agent_class(snapshot["cmdline"]):
            history = lib.update_history(state, snapshot)
            lib.update_turn_history(history, pid, snapshot["now"], child_map)
            blocked = lib.evaluate_blocked_turn(snapshot, history, exempt_pids=exempt,
                                                uptime_s=uptime_s)
            if blocked["verdict"] == "BLOCKED_TURN":
                results.append(blocked)
            results.append(lib.evaluate(snapshot, history, exempt_pids=exempt))
            continue
        orphan = lib.evaluate_test_orphan(snapshot, exempt_pids=exempt, uptime_s=uptime_s)
        if orphan["verdict"] == "TEST_ORPHAN":
            # track starttime so verify_test_orphan_before_kill can catch pid reuse
            lib.update_history(state, snapshot)
            results.append(orphan)

    for root in runtime_roots:
        root_snapshot = lib.take_live_snapshot(root)
        if root_snapshot is None:
            continue
        runtime = lib.evaluate_orphaned_runtime(root_snapshot, exempt_pids=exempt,
                                                uptime_s=uptime_s, root_of=root)
        if runtime["verdict"] == "ORPHANED_RUNTIME":
            lib.update_history(state, root_snapshot)
            results.append(runtime)

    lib.refresh_flagged(state, results)
    lib.prune_history(state, set(pids))
    lib.save_state(state)
    return results


def run_unblock(results):
    """Auto-reap the blocked tool-command subtree under a stuck turn. The agent
    is never signaled; its turn resumes with a failed tool result. Silent by
    design (user directive: no decision to make -> no notification) -- every
    action is journaled to the reaper log. Disable with REAPER_UNBLOCK=0."""
    if os.environ.get("REAPER_UNBLOCK", "1") != "1":
        return
    state = lib.load_state()
    for r in results:
        if r["verdict"] != "BLOCKED_TURN":
            continue
        verified = lib.verify_blocked_turn_before_kill(r["pid"], state)
        if verified is None:
            lib.log_event({"action": "unblock_refused", "agent_pid": r["pid"],
                           "reason": "criteria no longer hold at kill time"})
            continue
        outcome = lib.reap_tree(verified["child_pid"], verified)
        lib.log_event({"action": "auto_unblock", "agent_pid": r["pid"],
                       "child_pid": verified["child_pid"], "outcome": outcome,
                       "child_age_s": verified.get("child_age_s"),
                       "child_cmdline": (verified.get("child_cmdline") or "")[:200]})


def run_reap_sidecars(results):
    """Reap helper daemons whose agent session is gone. Silent by design (no
    decision to make -> no notification); every action is journaled.
    Disable with REAPER_SIDECAR=0."""
    if os.environ.get("REAPER_SIDECAR", "1") != "1":
        return
    state = lib.load_state()
    for r in results:
        if r["verdict"] != "ORPHANED_SIDECAR":
            continue
        verified = lib.verify_sidecar_before_kill(r["pid"], state)
        if verified is None:
            lib.log_event({"action": "sidecar_reap_refused", "pid": r["pid"],
                           "reason": "criteria no longer hold at kill time"})
            continue
        outcome = lib.reap_tree(r["pid"], verified)
        lib.log_event({"action": "sidecar_reap", "pid": r["pid"], "outcome": outcome,
                       "age_s": verified.get("age_s"),
                       "cmdline": (verified.get("cmdline") or "")[:200]})


def run_reap_runtimes(results):
    """Reap dev-runtime trees whose root lost its owner. Silent by design (no
    decision to make -> no notification); every action is journaled.
    Disable with REAPER_ORPHAN_RUNTIME=0."""
    if os.environ.get("REAPER_ORPHAN_RUNTIME", "1") != "1":
        return
    state = lib.load_state()
    for r in results:
        if r["verdict"] != "ORPHANED_RUNTIME":
            continue
        verified = lib.verify_orphaned_runtime_before_kill(r["pid"], state)
        if verified is None:
            lib.log_event({"action": "runtime_reap_refused", "pid": r["pid"],
                           "reason": "criteria no longer hold at kill time"})
            continue
        outcome = lib.reap_tree(r["pid"], verified)
        lib.log_event({"action": "runtime_reap", "pid": r["pid"], "outcome": outcome,
                       "age_s": verified.get("age_s"),
                       "trigger_pid": verified.get("trigger_pid"),
                       "trigger_cmdline": (verified.get("trigger_cmdline") or "")[:200],
                       "cmdline": (verified.get("cmdline") or "")[:200]})


def run_enforce(results):
    state = lib.load_state()
    for r in results:
        if r["verdict"] != "ENFORCE_ELIGIBLE":
            continue
        verified = lib.verify_live_before_kill(r["pid"], state)
        if verified is None or verified["verdict"] != "ENFORCE_ELIGIBLE":
            lib.log_event({"action": "enforce_refused", "pid": r["pid"], "reason": "criteria no longer hold at kill time"})
            continue
        lib.kill_ladder(r["pid"], verified)


def enforcement_enabled():
    return os.environ.get("ENFORCE") == "1"


def main():
    results = scan_once()
    if not enforcement_enabled():
        return
    run_unblock(results)
    run_reap_sidecars(results)
    run_reap_runtimes(results)
    run_enforce(results)


def self_test():
    now = time.time()

    def mk(pid, comm, cmdline, ppid, tty_nr, mtime):
        return {
            "pid": pid, "comm": comm, "cmdline": cmdline, "ppid": ppid, "tty_nr": tty_nr,
            "starttime": 1000, "utime": 0, "stime": 0, "io_bytes": 0,
            "transcript_mtime": mtime, "now": now,
        }

    # 1. working agent: transcript mtime NOW + io advancing + high cpu -> NEVER flagged
    working_hist = {
        "starttime": 1000,
        "pcpu_history": [90.0, 92.0, 88.0],
        "io_history": [
            {"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 1000},
            {"ts": now, "io_bytes": 999999},
        ],
    }
    working_snap = mk(101, "claude", "/usr/bin/node /usr/bin/claude", 500, 34, now)
    r = lib.evaluate(working_snap, working_hist, exempt_pids=set())
    assert r["verdict"] == "WORKING", f"working agent wrongly flagged: {r}"

    # 2. runaway: stale transcript + frozen io + pegged cpu + no tty + ppid==1 -> flagged in warn, would-kill under enforce
    runaway_hist = {
        "starttime": 1000,
        "pcpu_history": [95.0, 97.0, 96.0],
        "io_history": [
            {"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 5000},
            {"ts": now, "io_bytes": 5000},
        ],
    }
    runaway_snap = mk(102, "claude", "/usr/bin/node /usr/bin/claude", 1, 0, now - lib.STALE_MIN_S - 120)
    r = lib.evaluate(runaway_snap, runaway_hist, exempt_pids=set())
    assert r["verdict"] == "ENFORCE_ELIGIBLE", f"runaway not flagged for enforce: {r}"
    assert r["progress_stale"] is True

    # runaway but WITH a tty must only ever be WARN, never enforce-eligible
    runaway_tty_snap = mk(103, "claude", "/usr/bin/node /usr/bin/claude", 1, 34, now - lib.STALE_MIN_S - 120)
    r_tty = lib.evaluate(runaway_tty_snap, runaway_hist, exempt_pids=set())
    assert r_tty["verdict"] == "WARN", f"tty-present runaway must stay WARN-only: {r_tty}"

    # 3. idle-old-session: stale but ~0% cpu + has tty -> NOT flagged
    idle_hist = {
        "starttime": 1000,
        "pcpu_history": [0.0, 0.1, 0.0],
        "io_history": [
            {"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 5000},
            {"ts": now, "io_bytes": 5000},
        ],
    }
    idle_snap = mk(104, "claude", "/usr/bin/node /usr/bin/claude", 500, 34, now - lib.STALE_MIN_S - 120)
    r = lib.evaluate(idle_snap, idle_hist, exempt_pids=set())
    assert r["verdict"] == "STALE_LOW_CPU", f"idle old session wrongly flagged: {r}"

    # 4. resumed-bg-no-tty: no tty + progress ADVANCING -> NOT flagged (proves no-tty alone never kills)
    resumed_hist = {
        "starttime": 1000,
        "pcpu_history": [80.0, 85.0, 90.0],
        "io_history": [
            {"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 1000},
            {"ts": now, "io_bytes": 500000},
        ],
    }
    resumed_snap = mk(105, "claude", "/usr/bin/node /usr/bin/claude", 1, 0, now)
    r = lib.evaluate(resumed_snap, resumed_hist, exempt_pids=set())
    assert r["verdict"] == "WORKING", f"resumed no-tty background agent wrongly flagged: {r}"

    # denylist + non-agent-class must always short-circuit to NOT_CANDIDATE
    deny_snap = mk(106, "harnessd", "/usr/bin/harnessd", 1, 0, now - lib.STALE_MIN_S - 120)
    r = lib.evaluate(deny_snap, runaway_hist, exempt_pids=set())
    assert r["verdict"] == "NOT_CANDIDATE", f"denylisted proc wrongly flagged: {r}"

    other_snap = mk(107, "pipewire", "/usr/bin/pipewire", 1, 0, now - lib.STALE_MIN_S - 120)
    r = lib.evaluate(other_snap, runaway_hist, exempt_pids=set())
    assert r["verdict"] == "NOT_CANDIDATE", f"non-agent-class proc wrongly evaluated as candidate: {r}"

    # --- leaked test-run orphan class (evaluate_test_orphan) ---
    up = 100000.0  # synthetic uptime
    old_start = int((up - lib.TEST_ORPHAN_MIN_AGE_S - 120) * lib.HZ)
    young_start = int((up - 60) * lib.HZ)

    def orph(pid, cmdline, tty_nr, starttime):
        return {"pid": pid, "comm": "bash", "cmdline": cmdline, "ppid": 8675,
                "tty_nr": tty_nr, "starttime": starttime, "utime": 0, "stime": 0,
                "io_bytes": 0, "transcript_mtime": None, "now": now}

    # reparented + harness token + no tty + old -> TEST_ORPHAN
    o = lib.evaluate_test_orphan(orph(201, "bash ci-integration-http-signal-harness.sh int", 0, old_start),
                                 uptime_s=up, reparented=True, cwd="/home/user/x/.wt-plan-int/apps/web/scripts")
    assert o["verdict"] == "TEST_ORPHAN", f"leaked orphan not flagged: {o}"
    assert o["kind"] == "leaked-test-orphan"

    # generic runtime token but NOT under a test path -> NOT_CANDIDATE
    o = lib.evaluate_test_orphan(orph(202, "node wrangler dev", 0, old_start),
                                 uptime_s=up, reparented=True, cwd="/home/user/Projects/myapp")
    assert o["verdict"] == "NOT_CANDIDATE", f"user dev server wrongly flagged: {o}"

    # generic runtime UNDER a test path -> TEST_ORPHAN
    o = lib.evaluate_test_orphan(orph(203, "node wrangler dev", 0, old_start),
                                 uptime_s=up, reparented=True, cwd="/home/user/x/.wt-plan-int/apps/web")
    assert o["verdict"] == "TEST_ORPHAN", f"orphaned test dev server not flagged: {o}"

    # still has a controlling tty -> interactive, never an orphan
    o = lib.evaluate_test_orphan(orph(204, "bash ci-with-disposable-pg.sh", 34, old_start),
                                 uptime_s=up, reparented=True, cwd="/home/user/x/.wt-plan-int")
    assert o["verdict"] == "NOT_CANDIDATE", f"tty-bearing proc wrongly flagged as orphan: {o}"

    # parent still alive (a live test) -> NOT_CANDIDATE
    o = lib.evaluate_test_orphan(orph(205, "sleep infinity", 0, old_start),
                                 uptime_s=up, reparented=False, cwd="/home/user/x/agent-tmp/int")
    assert o["verdict"] == "NOT_CANDIDATE", f"live (non-reparented) test wrongly flagged: {o}"

    # reparented harness token but too young -> NOT_CANDIDATE (trap may still reap)
    o = lib.evaluate_test_orphan(orph(206, "bash ci-integration-http-signal-harness.sh int", 0, young_start),
                                 uptime_s=up, reparented=True, cwd="/home/user/x/.wt-plan-int")
    assert o["verdict"] == "NOT_CANDIDATE", f"too-young orphan wrongly flagged: {o}"

    # an agent-class process never falls into the orphan class
    o = lib.evaluate_test_orphan({"pid": 207, "comm": "node", "cmdline": "node claude sleep infinity",
                                  "ppid": 1, "tty_nr": 0, "starttime": old_start, "utime": 0, "stime": 0,
                                  "io_bytes": 0, "transcript_mtime": None, "now": now},
                                 uptime_s=up, reparented=True, cwd="/home/user/x/.wt-plan-int")
    assert o["verdict"] == "NOT_CANDIDATE", f"agent-class proc wrongly evaluated as orphan: {o}"

    # --- orphaned-sidecar class (evaluate_orphaned_sidecar) ---
    SIDECAR_CMD = "/usr/bin/bun /home/user/.claude/plugins/sources/quietmode/start.mjs"
    old_sidecar_start = int((up - lib.SIDECAR_MIN_AGE_S - 120) * lib.HZ)

    def sc(pid, cmdline, ppid, starttime, tty_nr=0):
        return {"pid": pid, "comm": "bun", "cmdline": cmdline, "ppid": ppid,
                "tty_nr": tty_nr, "starttime": starttime, "utime": 0, "stime": 0,
                "io_bytes": 0, "transcript_mtime": None, "now": now}

    # reparented (session dead) + old -> ORPHANED_SIDECAR
    s = lib.evaluate_orphaned_sidecar(sc(401, SIDECAR_CMD, 8675, old_sidecar_start),
                                      uptime_s=up, reparented=True)
    assert s["verdict"] == "ORPHANED_SIDECAR", f"orphaned sidecar not flagged: {s}"
    assert s["kind"] == "orphaned-sidecar"

    # parent session still alive -> never reaped, however old
    s = lib.evaluate_orphaned_sidecar(sc(402, SIDECAR_CMD, 3124701, old_sidecar_start),
                                      uptime_s=up, reparented=False)
    assert s["verdict"] == "NOT_CANDIDATE", f"live sidecar wrongly flagged: {s}"

    # reparented but too young -> not yet
    s = lib.evaluate_orphaned_sidecar(sc(403, SIDECAR_CMD, 1, young_start),
                                      uptime_s=up, reparented=True)
    assert s["verdict"] == "NOT_CANDIDATE", f"young sidecar wrongly flagged: {s}"

    # a real agent session is never a sidecar, even reparented and old
    s = lib.evaluate_orphaned_sidecar(sc(404, "/home/user/.local/bin/claude --resume", 1,
                                         old_sidecar_start), uptime_s=up, reparented=True)
    assert s["verdict"] == "NOT_CANDIDATE", f"agent session wrongly flagged as sidecar: {s}"

    # the misroute this class exists to fix: is_agent_class claims the sidecar,
    # so evaluate() files it as idle and it is never reaped
    assert lib.is_agent_class(SIDECAR_CMD), "sidecar no longer matches is_agent_class"
    stale_idle = {
        "starttime": old_sidecar_start,
        "pcpu_history": [0.0, 0.0, 0.0],
        "io_history": [{"ts": now - lib.STALE_MIN_S - 60, "io_bytes": 7000},
                       {"ts": now, "io_bytes": 7000}],
    }
    r = lib.evaluate(sc(405, SIDECAR_CMD, 8675, old_sidecar_start), stale_idle,
                     exempt_pids=set())
    assert r["verdict"] != "ENFORCE_ELIGIBLE", "agent path must not enforce on sidecars"

    # --- orphaned-runtime tree class (evaluate_orphaned_runtime) ---
    old_runtime_start = int((up - lib.ORPHAN_RUNTIME_MIN_AGE_S - 120) * lib.HZ)
    TRIGGER = (599, "node /home/user/x/node_modules/wrangler/bin/wrangler.js dev")

    def rt(pid, cmdline, starttime, tty_nr=0, ppid=8675):
        return {"pid": pid, "comm": "npm exec", "cmdline": cmdline, "ppid": ppid,
                "tty_nr": tty_nr, "starttime": starttime, "utime": 0, "stime": 0,
                "io_bytes": 0, "transcript_mtime": None, "now": now}

    # unowned tree root + old + a runtime in the tree -> ORPHANED_RUNTIME
    t = lib.evaluate_orphaned_runtime(rt(501, "sh -c npm exec wrangler dev", old_runtime_start),
                                      uptime_s=up, root_of=501, trigger=TRIGGER)
    assert t["verdict"] == "ORPHANED_RUNTIME", f"orphaned runtime tree not flagged: {t}"
    assert t["kind"] == "orphaned-runtime" and t["trigger_pid"] == 599

    # an owner was found in the ancestor chain -> live work, never reaped
    t = lib.evaluate_orphaned_runtime(rt(502, "sh -c npm exec wrangler dev", old_runtime_start),
                                      uptime_s=up, root_of=None, trigger=TRIGGER)
    assert t["verdict"] == "NOT_CANDIDATE", f"owned tree wrongly flagged: {t}"

    # root holds a controlling terminal -> the user started it
    t = lib.evaluate_orphaned_runtime(rt(503, "npm exec vite", old_runtime_start, tty_nr=34821),
                                      uptime_s=up, root_of=503, trigger=TRIGGER)
    assert t["verdict"] == "NOT_CANDIDATE", f"tty-bearing tree wrongly flagged: {t}"

    # unowned but younger than the age floor -> not yet
    t = lib.evaluate_orphaned_runtime(rt(504, "npm exec astro dev", young_start),
                                      uptime_s=up, root_of=504, trigger=TRIGGER)
    assert t["verdict"] == "NOT_CANDIDATE", f"young tree wrongly flagged: {t}"

    # unowned and old, but the tree holds no dev runtime -> not our class
    t = lib.evaluate_orphaned_runtime(rt(505, "sh -c ./some-daemon", old_runtime_start),
                                      uptime_s=up, root_of=505)
    assert t["verdict"] == "NOT_CANDIDATE", f"runtime-free tree wrongly flagged: {t}"

    # an orphaned agent session is class 1's call, never reaped here
    t = lib.evaluate_orphaned_runtime(rt(506, "/home/user/.local/bin/claude --resume",
                                         old_runtime_start), uptime_s=up, root_of=506,
                                      trigger=TRIGGER)
    assert t["verdict"] == "NOT_CANDIDATE", f"agent session wrongly flagged as runtime: {t}"

    # a wrapper merely living under ~/.claude is not a session, so it does not
    # shield its own orphaned tree the way a real agent session does
    WRAPPER = "/home/user/.claude/bin/local-gate -- bash -c e2e-remote"
    assert lib.is_agent_class(WRAPPER), "wrapper no longer matches the coarse predicate"
    assert not lib.is_agent_session("node", WRAPPER), "wrapper wrongly read as a session"
    assert lib.is_agent_session("claude", "claude --dangerously-skip-permissions")
    assert lib.is_agent_session("MainThread", "/usr/bin/cursor-agent --use-system-ca")
    t = lib.evaluate_orphaned_runtime(rt(508, WRAPPER, old_runtime_start),
                                      uptime_s=up, root_of=508, trigger=TRIGGER)
    assert t["verdict"] == "ORPHANED_RUNTIME", f"orphaned wrapper tree not flagged: {t}"

    # wrangler dev execs a workerd binary whose path never carries "wrangler"
    WORKERD = ("/home/user/x/node_modules/.pnpm/@cloudflare+workerd-linux-64@1.0.0/"
               "node_modules/@cloudflare/workerd-linux-64/bin/workerd serve --socket-addr")
    assert "wrangler" not in WORKERD, "workerd fixture no longer independent of the wrangler token"
    assert lib.is_orphan_runtime(WORKERD), "workerd not recognised as a dev runtime"

    # the gap this class exists to fix: a descendant inside a dead tree keeps a
    # live parent, so class 2's per-process reparent check never sees the leak
    o = lib.evaluate_test_orphan(orph(507, "node wrangler dev", 0, old_runtime_start),
                                 uptime_s=up, reparented=False,
                                 cwd="/home/user/x/.wt-plan-int/apps/web")
    assert o["verdict"] == "NOT_CANDIDATE", f"class 2 unexpectedly covers deep-tree leaks: {o}"

    # --- blocked-turn class (evaluate_blocked_turn) ---
    up2 = 200000.0
    old_child_start = int((up2 - lib.BLOCKED_TURN_MIN_S - 120) * lib.HZ)
    blocked_since = now - lib.BLOCKED_TURN_MIN_S - 120

    quiet_output_ts = now - lib.BLOCKED_NOEVIDENCE_MIN_S - 120

    def turn_hist(cmdline, pcpu, child_start=old_child_start, status="in-flight",
                  since=blocked_since, wall_start=None, tpath="/tmp/stub.jsonl",
                  out_prog=quiet_output_ts):
        if wall_start is None:
            wall_start = now - up2 + child_start / lib.HZ
        return {"starttime": 1000,
                "turn": {"child_pid": 9001, "child_start": child_start,
                          "child_cmdline": cmdline, "transcript_path": tpath,
                          "turn_status": status, "blocked_since_ts": since,
                          "child_wall_start": wall_start,
                          "last_output_progress": out_prog, "output_size": 0,
                          "output_path": None,
                          "pcpu_history": pcpu, "_last": None}}

    blocked_snap = mk(301, "claude", "/usr/bin/node /usr/bin/claude", 500, 34, None)

    # in-flight tool_use 30+min + old shell child + ~0% subtree cpu -> BLOCKED_TURN
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("gh run watch 123 --exit-status", [0.0, 0.2, 0.1]),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "BLOCKED_TURN", f"blocked turn not flagged: {b}"
    assert b["child_pid"] == 9001

    # output file grew recently (harness heartbeat / streaming build) -> a
    # healthy quiet run (e.g. remote offload wait at ~0%% local cpu), never flagged
    b = lib.evaluate_blocked_turn(blocked_snap,
                                  turn_hist("bash long-running-task.sh", [0.0, 0.0, 0.0], out_prog=now - 300),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"run with recent output wrongly flagged: {b}"

    # child busy (subtree cpu high) -> a working build, never flagged
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("bash tools/gate.sh", [80.0, 75.0, 90.0]),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"busy child wrongly flagged: {b}"

    # child carries an unexpired `timeout` bound -> BOUNDED_WAIT, not flagged
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("timeout 2700 cdx exec -m gpt", [0.0, 0.0, 0.0]),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "BOUNDED_WAIT", f"bounded wait wrongly flagged: {b}"

    # expired bound -> flagged (child outlived its own timeout + slack)
    expired_start = int((up2 - 2700 - lib.BLOCKED_TIMEOUT_SLACK_S - 120) * lib.HZ)
    b = lib.evaluate_blocked_turn(blocked_snap,
                                  turn_hist("timeout 2700 cdx exec -m gpt", [0.0, 0.0, 0.0], expired_start,
                                            since=now - 2700 - lib.BLOCKED_TIMEOUT_SLACK_S - 120),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "BLOCKED_TURN", f"expired bound not flagged: {b}"

    # turn advancing (tool returned / turn ended; child = user background task) -> never flagged
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("tail -f /tmp/x", [0.0, 0.0, 0.0], status="advancing"),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"background task wrongly flagged: {b}"

    # in-flight but child start does not correlate with the pending tool_use
    # timestamp (older user-backgrounded task) -> never flagged
    bg_start = int((up2 - lib.BLOCKED_TURN_MIN_S - 120 - 3600) * lib.HZ)
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("tail -f /tmp/x", [0.0, 0.0, 0.0], bg_start),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"uncorrelated child wrongly flagged: {b}"

    # no transcript evidence: child younger than the 2h no-evidence bar -> not flagged
    b = lib.evaluate_blocked_turn(blocked_snap,
                                  turn_hist("tail -f /tmp/x", [0.0, 0.0, 0.0], status="no-evidence", since=None),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"young no-evidence child wrongly flagged: {b}"

    # no transcript evidence but child older than the 2h bar + ~0% cpu -> flagged
    ancient_start = int((up2 - lib.BLOCKED_NOEVIDENCE_MIN_S - 120) * lib.HZ)
    b = lib.evaluate_blocked_turn(blocked_snap,
                                  turn_hist("tail -f /tmp/x", [0.0, 0.0, 0.0], ancient_start,
                                            status="no-evidence", since=None),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "BLOCKED_TURN", f"ancient no-evidence child not flagged: {b}"

    # session identity never resolved (transcript_path None) -> never killable,
    # however old -- catches is_agent_class false positives (shells, wrappers)
    b = lib.evaluate_blocked_turn(blocked_snap,
                                  turn_hist("tail -f /tmp/x", [0.0, 0.0, 0.0], ancient_start,
                                            status="no-evidence", since=None, tpath=None),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"unidentified session wrongly flagged: {b}"

    # tool_use recent (blocked only a moment) -> not flagged yet
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("sleep 600", [0.0, 0.0, 0.0], since=now - 300),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"young blocked turn wrongly flagged: {b}"

    # no shell child (idle session at prompt) -> never flagged
    b = lib.evaluate_blocked_turn(blocked_snap, {"starttime": 1000, "turn": None},
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"idle session wrongly flagged: {b}"

    # too few cpu samples yet -> not flagged (needs CPU_SAMPLES_REQUIRED)
    b = lib.evaluate_blocked_turn(blocked_snap, turn_hist("gh run watch 1", [0.0]),
                                  exempt_pids=set(), uptime_s=up2)
    assert b["verdict"] == "NOT_CANDIDATE", f"under-sampled turn wrongly flagged: {b}"

    # timeout bound parser
    assert lib.parse_timeout_bound_s("timeout -k 30 1500 gh run watch") == 1500
    assert lib.parse_timeout_bound_s("timeout 900 gh run watch 30263569865") == 900
    assert lib.parse_timeout_bound_s("timeout 5m foo") == 300
    assert lib.parse_timeout_bound_s("gh run watch 99") is None

    # turn-status transcript tail parser (read_turn_status)
    import json as _json
    import tempfile as _tempfile
    with _tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(_json.dumps({"type": "assistant", "timestamp": "2026-07-29T08:00:00.000Z",
                              "message": {"content": [{"type": "tool_use", "name": "Bash"}]}}) + "\n")
        tf.write(_json.dumps({"type": "queue-operation", "timestamp": "2026-07-29T09:00:00.000Z"}) + "\n")
        tpath = tf.name
    try:
        status, since = lib.read_turn_status(tpath)
        assert status == "in-flight", "pending tool_use behind queue noise must read in-flight"
        assert since is not None and abs(since - 1785312000.0) < 86400
        with open(tpath, "a") as f:
            f.write(_json.dumps({"type": "user",
                                 "message": {"content": [{"type": "tool_result"}]}}) + "\n")
        status, _ = lib.read_turn_status(tpath)
        assert status == "advancing", "tool_result after tool_use must read advancing"
        with open(tpath, "a") as f:
            f.write(_json.dumps({"type": "assistant", "timestamp": "2026-07-29T08:10:00.000Z",
                                 "message": {"content": [{"type": "text", "text": "done"}]}}) + "\n")
        status, _ = lib.read_turn_status(tpath)
        assert status == "advancing", "text-final assistant event must read advancing"
        status, since = lib.read_turn_status(tpath + ".does-not-exist")
        assert status == "no-evidence" and since is None, "missing transcript must read no-evidence"
    finally:
        os.unlink(tpath)

    print("agent-reaper self-test OK: runaway + leaked-test-orphan + blocked-turn + orphaned-sidecar + orphaned-runtime classes correctly classified", file=sys.stderr)
    return True


if __name__ == "__main__":
    if "--self-test" in sys.argv:
        sys.exit(0 if self_test() else 1)
    if len(sys.argv) > 1:
        sys.exit(f"usage: {os.path.basename(sys.argv[0])} [--self-test]\n"
                 "a bare run performs a live scan and may kill processes")
    main()