"""Shared candidate-classification + kill-ladder logic for agent-reaper.py
and reaper-ctl. Single source of truth for the never-kill-interactive
safety stack -- do not duplicate this logic in either caller.
"""
import glob
import json
import os
import re
import signal
import time

STATE_PATH = os.path.expanduser("~/.claude/run/agent-reaper-state.json")
LOG_PATH = os.path.expanduser("~/.claude/run/agent-reaper.log")

STALE_MIN_S = int(os.environ.get("REAPER_STALE_MIN_S", 45 * 60))
CPU_THRESHOLD = float(os.environ.get("REAPER_CPU_THRESHOLD", 50.0))
CPU_SAMPLES_REQUIRED = int(os.environ.get("REAPER_CPU_SAMPLES", 3))

# Blocked turn: an interactive agent whose in-flight Bash tool call sits in a
# low-CPU wait (gh run watch, wedged test, unbounded poll) -- the inverse of the
# runaway class, invisible to it. Discriminator is a live direct shell child
# older than the transcript freeze: an idle session at the prompt has none, and
# MCP/LSP children are node/bun, never a shell.
BLOCKED_TURN_MIN_S = int(os.environ.get("REAPER_BLOCKED_TURN_MIN_S", 30 * 60))
BLOCKED_NOEVIDENCE_MIN_S = int(os.environ.get("REAPER_BLOCKED_NOEVIDENCE_MIN_S", 2 * 3600))
BLOCKED_CPU_MAX = float(os.environ.get("REAPER_BLOCKED_CPU_MAX", 10.0))
BLOCKED_TIMEOUT_SLACK_S = int(os.environ.get("REAPER_BLOCKED_TIMEOUT_SLACK_S", 600))
BLOCKED_CORRELATE_S = int(os.environ.get("REAPER_BLOCKED_CORRELATE_S", 600))
TOOL_SHELL_COMMS = ("bash", "sh", "zsh")
TRANSCRIPT_GLOBS = (
    "~/.claude/projects/*/{u}.jsonl",
    "~/.systray-ai/claude-accounts/*/CLAUDE_HOME/projects/*/{u}.jsonl",
)
HISTORY_MAX = 90  # scan cadence is 60s; 90 samples covers STALE_MIN_S with margin
KILL_WAIT_S = 15

DENYLIST_TOKENS = {
    "ccr", "harnessd", "virt-manager", "qemu", "qemu-system", "pipewire",
    "cinnamon", "agent_guard", "hw-blackbox", "agent-reaper", "reaper-ctl",
    "reaper-notifier",
}
AGENT_CLASS_TOKENS = ("claude", "cursor-agent")

# Leaked test-run orphans: a distinct candidate class from runaway agents.
# Signature is STRUCTURAL, not resource-based -- a killed run/agent leaves test
# subprocesses (dev servers, sleep-infinity fixtures) reparented to the user
# systemd manager or init; nothing ever reaps them. They are idle, so the
# agent-class CPU/no-progress discriminator can never see them.
TEST_ORPHAN_MIN_AGE_S = int(os.environ.get("REAPER_TEST_ORPHAN_MIN_AGE_S", 10 * 60))
# Specific harness fixtures -- unambiguous, match on their own.
TEST_ORPHAN_HARNESS_TOKENS = (
    "ci-integration-http-signal-harness",
    "ci-integration-http",
    "ci-with-disposable",
    "ci-neon-branch",
    "ci-oracle-database",
    "runner-integration",
)
# Generic runtimes -- a leak ONLY when also reparented AND under a test path,
# so a user's own backgrounded dev server never trips.
TEST_ORPHAN_RUNTIME_TOKENS = ("sleep infinity", "wrangler", "vitest", "playwright", "esbuild")
TEST_ORPHAN_PATH_MARKERS = (".wt-", "/.claude/worktrees/", "agent-tmp", "runner-integration",
                            "/apps/web/scripts/ci-")

# Orphaned agent sidecars: per-session helper daemons that serve only their
# spawning agent. Their cmdline carries the agent's own path tokens, so
# is_agent_class() claims them and evaluate() then files them as idle
# STALE_LOW_CPU -- they are never reaped and accumulate GBs of swap across
# sessions. Reparenting alone is conclusive here: a sidecar whose session died
# serves nobody, so no path or resource qualification applies.
SIDECAR_MIN_AGE_S = int(os.environ.get("REAPER_SIDECAR_MIN_AGE_S", 10 * 60))
SIDECAR_TOKENS = ("quietmode/start.mjs",)

# Orphaned dev-runtime trees: a leaked run's whole process tree, not one process.
# Only the tree ROOT reparents to user-systemd; its descendants keep live parents
# inside the dead tree, so the per-process is_reparented() check that class 2 uses
# sees a live parent and skips them. Ownership is therefore decided by walking the
# ancestor chain: a chain reaching init/user-systemd without passing an agent or a
# controlling terminal has no owner, and the whole tree is garbage.
ORPHAN_RUNTIME_MIN_AGE_S = int(os.environ.get("REAPER_ORPHAN_RUNTIME_MIN_AGE_S", 4 * 3600))
ORPHAN_RUNTIME_TOKENS = TEST_ORPHAN_RUNTIME_TOKENS + ("vite", "astro", "e2e-remote",
                                                      "workerd")

HZ = os.sysconf("SC_CLK_TCK") or 100


def _read_ancestor_pids(start_pid):
    pids = set()
    pid = start_pid
    for _ in range(64):
        pids.add(pid)
        try:
            with open(f"/proc/{pid}/stat", "rb") as f:
                raw = f.read().decode(errors="replace")
        except OSError:
            break
        rp = raw.rfind(")")
        if rp == -1:
            break
        rest = raw[rp + 2:].split()
        try:
            ppid = int(rest[1])
        except (IndexError, ValueError):
            break
        if ppid == pid or ppid <= 1:
            pids.add(ppid)
            break
        pid = ppid
    return pids


def dispatcher_exempt_pids():
    """Pids in this process's own ancestor chain -- protects the invoking
    claude session (e.g. when this tool is run manually from within one)
    from ever becoming its own candidate."""
    return _read_ancestor_pids(os.getpid())


def is_agent_class(cmdline):
    low = cmdline.lower()
    return any(tok in low for tok in AGENT_CLASS_TOKENS)


def is_agent_session(comm, cmdline):
    """True only when the process IS an agent CLI. is_agent_class() matches any
    cmdline carrying an agent path token, which every wrapper and helper under
    ~/.claude also satisfies -- too coarse to decide tree ownership."""
    if comm in AGENT_CLASS_TOKENS:
        return True
    parts = cmdline.split()
    return bool(parts) and os.path.basename(parts[0]) in AGENT_CLASS_TOKENS


def is_denylisted(comm, cmdline):
    low_comm = comm.lower()
    low_cmd = cmdline.lower()
    return any(tok in low_comm or tok in low_cmd for tok in DENYLIST_TOKENS)


def read_uptime_s():
    try:
        with open("/proc/uptime") as f:
            return float(f.read().split()[0])
    except (OSError, ValueError, IndexError):
        return None


def read_proc_cwd(pid):
    try:
        return os.readlink(f"/proc/{pid}/cwd")
    except OSError:
        return None


def is_reparented(ppid):
    """True when a process's parent is init or the user systemd manager --
    i.e. its real spawner died and it was reparented. A live test's runtime
    still has its runner as parent, so this cleanly separates leak from live."""
    if ppid <= 1:
        return True
    pst = read_proc_stat(ppid)
    if pst is None:
        return True  # parent vanished between reads -> reparenting now
    if pst["comm"] == "systemd":
        return "--user" in read_proc_cmdline(ppid)
    return False


def is_sidecar(cmdline):
    return any(tok in cmdline for tok in SIDECAR_TOKENS)


def is_orphan_runtime(cmdline):
    low = cmdline.lower()
    return any(tok in low for tok in ORPHAN_RUNTIME_TOKENS)


def orphan_tree_root(pid, max_depth=64):
    """Topmost ancestor whose chain reaches init/user-systemd without passing a
    live owner. None when an owner is found: an ancestor that is agent-class or
    holds a controlling terminal."""
    cur = pid
    seen = set()
    for _ in range(max_depth):
        if cur in seen:
            return None
        seen.add(cur)
        st = read_proc_stat(cur)
        if st is None:
            return None
        ppid = st["ppid"]
        if ppid <= 1:
            return cur
        pst = read_proc_stat(ppid)
        if pst is None:
            return cur
        pcmd = read_proc_cmdline(ppid)
        if pst["comm"] == "systemd" and "--user" in pcmd:
            return cur
        if pst["tty_nr"] != 0 or is_agent_session(pst["comm"], pcmd):
            return None
        cur = ppid
    return None


def test_orphan_reason(cmdline, cwd):
    low = cmdline.lower()
    for tok in TEST_ORPHAN_HARNESS_TOKENS:
        if tok in low:
            return tok
    if any(tok in low for tok in TEST_ORPHAN_RUNTIME_TOKENS):
        hay = low + " " + (cwd or "").lower()
        if any(marker in hay for marker in TEST_ORPHAN_PATH_MARKERS):
            return "runtime-under-test-path"
    return None


def read_proc_stat(pid):
    try:
        with open(f"/proc/{pid}/stat", "rb") as f:
            raw = f.read().decode(errors="replace")
    except OSError:
        return None
    rp = raw.rfind(")")
    lp = raw.find("(")
    if rp == -1 or lp == -1:
        return None
    comm = raw[lp + 1:rp]
    rest = raw[rp + 2:].split()
    try:
        return {
            "comm": comm,
            "ppid": int(rest[1]),
            "tty_nr": int(rest[4]),
            "utime": int(rest[11]),
            "stime": int(rest[12]),
            "starttime": int(rest[19]),
        }
    except (IndexError, ValueError):
        return None


def read_proc_cmdline(pid):
    try:
        with open(f"/proc/{pid}/cmdline", "rb") as f:
            raw = f.read()
    except OSError:
        return ""
    return raw.replace(b"\x00", b" ").decode(errors="replace").strip()


def read_proc_io_bytes(pid):
    try:
        with open(f"/proc/{pid}/io") as f:
            fields = {}
            for line in f:
                k, _, v = line.partition(":")
                fields[k.strip()] = v.strip()
    except OSError:
        return None
    try:
        return (int(fields.get("rchar", 0)) + int(fields.get("wchar", 0)) +
                int(fields.get("read_bytes", 0)) + int(fields.get("write_bytes", 0)))
    except ValueError:
        return None


def find_transcript_mtime(pid):
    fd_dir = f"/proc/{pid}/fd"
    projects_dir = os.path.expanduser("~/.claude/projects") + os.sep
    try:
        entries = os.listdir(fd_dir)
    except OSError:
        return None
    newest = None
    for entry in entries:
        try:
            target = os.readlink(os.path.join(fd_dir, entry))
        except OSError:
            continue
        if not target.endswith(".jsonl") or not target.startswith(projects_dir):
            continue
        try:
            mtime = os.stat(target).st_mtime
        except OSError:
            continue
        if newest is None or mtime > newest:
            newest = mtime
    return newest


def take_live_snapshot(pid):
    st = read_proc_stat(pid)
    if st is None:
        return None
    cmdline = read_proc_cmdline(pid)
    io_bytes = read_proc_io_bytes(pid)
    transcript_mtime = find_transcript_mtime(pid)
    return {
        "pid": pid,
        "comm": st["comm"],
        "cmdline": cmdline,
        "ppid": st["ppid"],
        "tty_nr": st["tty_nr"],
        "starttime": st["starttime"],
        "utime": st["utime"],
        "stime": st["stime"],
        "io_bytes": io_bytes,
        "transcript_mtime": transcript_mtime,
        "now": time.time(),
    }


def list_candidate_pids():
    pids = []
    for d in glob.glob("/proc/[0-9]*"):
        pid = os.path.basename(d)
        try:
            pids.append(int(pid))
        except ValueError:
            continue
    return pids


def load_state():
    try:
        with open(STATE_PATH) as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError):
        return {}


def save_state(state):
    os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
    tmp = STATE_PATH + ".tmp"
    with open(tmp, "w") as f:
        json.dump(state, f)
    os.replace(tmp, STATE_PATH)


def update_history(state, snapshot):
    """Mutates state['pids'] in place with a new sample for snapshot['pid'].
    Resets history if starttime shows the pid was reused."""
    pids = state.setdefault("pids", {})
    pid_key = str(snapshot["pid"])
    entry = pids.get(pid_key)
    if entry is None or entry.get("starttime") != snapshot["starttime"]:
        entry = {"starttime": snapshot["starttime"], "pcpu_history": [], "io_history": [],
                  "last_notified": None}

    prev_cpu = entry.get("_last_cpu")
    if prev_cpu is not None and snapshot["utime"] is not None:
        dt = snapshot["now"] - prev_cpu["ts"]
        dticks = (snapshot["utime"] + snapshot["stime"]) - prev_cpu["ticks"]
        if dt > 0 and dticks >= 0:
            pcpu = (dticks / HZ) / dt * 100.0
            entry["pcpu_history"] = (entry.get("pcpu_history", []) + [pcpu])[-HISTORY_MAX:]
    entry["_last_cpu"] = {"ticks": snapshot["utime"] + snapshot["stime"], "ts": snapshot["now"]}

    if snapshot["io_bytes"] is not None:
        entry["io_history"] = (entry.get("io_history", []) + [
            {"ts": snapshot["now"], "io_bytes": snapshot["io_bytes"]}
        ])[-HISTORY_MAX:]

    entry["last_seen"] = snapshot["now"]
    pids[pid_key] = entry
    return entry


def prune_history(state, alive_pids, max_age_s=3 * 3600):
    pids = state.setdefault("pids", {})
    now = time.time()
    for pid_key in list(pids.keys()):
        entry = pids[pid_key]
        if int(pid_key) in alive_pids and now - entry.get("last_seen", 0) < max_age_s:
            continue
        del pids[pid_key]


def _io_advancing(io_history, now, stale_min_s):
    if not io_history:
        return None
    window_start = now - stale_min_s
    baseline = None
    for point in io_history:
        if point["ts"] <= window_start:
            baseline = point
        else:
            break
    if baseline is None:
        if io_history[0]["ts"] > window_start:
            return None
        baseline = io_history[0]
    current = io_history[-1]["io_bytes"]
    return current > baseline["io_bytes"]


def evaluate(snapshot, history, exempt_pids=None, stale_min_s=None, cpu_threshold=None):
    exempt_pids = exempt_pids or set()
    stale_min_s = STALE_MIN_S if stale_min_s is None else stale_min_s
    cpu_threshold = CPU_THRESHOLD if cpu_threshold is None else cpu_threshold

    result = {
        "pid": snapshot["pid"], "comm": snapshot["comm"], "cmdline": snapshot["cmdline"],
        "ppid": snapshot["ppid"], "tty_nr": snapshot["tty_nr"], "kind": "runaway-agent",
        "agent_class": is_agent_class(snapshot["cmdline"]),
        "denylisted": is_denylisted(snapshot["comm"], snapshot["cmdline"]),
        "exempt": snapshot["pid"] in exempt_pids,
        "transcript_mtime": snapshot["transcript_mtime"],
        "transcript_age_s": None,
        "io_advancing": None,
        "pcpu_now": None,
        "cpu_sustained_high": False,
        "progress_stale": None,
        "verdict": "NOT_CANDIDATE",
    }

    if result["denylisted"] or result["exempt"] or not result["agent_class"]:
        return result

    now = snapshot["now"]
    if snapshot["transcript_mtime"] is not None:
        result["transcript_age_s"] = now - snapshot["transcript_mtime"]
    transcript_frozen = (result["transcript_age_s"] is not None and
                          result["transcript_age_s"] >= stale_min_s)

    io_history = history.get("io_history", []) if history else []
    io_advancing = _io_advancing(io_history, now, stale_min_s)
    result["io_advancing"] = io_advancing
    io_frozen = (io_advancing is False)

    if snapshot["transcript_mtime"] is None:
        # headless/detached agent (no transcript fd): io is the only progress signal
        result["progress_stale"] = bool(io_frozen)
    else:
        result["progress_stale"] = bool(transcript_frozen and io_frozen)

    pcpu_history = history.get("pcpu_history", []) if history else []
    if pcpu_history:
        result["pcpu_now"] = pcpu_history[-1]
    recent = pcpu_history[-CPU_SAMPLES_REQUIRED:]
    result["cpu_sustained_high"] = (len(recent) >= CPU_SAMPLES_REQUIRED and
                                     all(v >= cpu_threshold for v in recent))

    if not result["progress_stale"]:
        result["verdict"] = "WORKING"
        return result
    if not result["cpu_sustained_high"]:
        result["verdict"] = "STALE_LOW_CPU"
        return result

    is_orphan = snapshot["ppid"] == 1
    no_tty = snapshot["tty_nr"] == 0
    result["verdict"] = "ENFORCE_ELIGIBLE" if (is_orphan and no_tty) else "WARN"
    return result


def evaluate_test_orphan(snapshot, exempt_pids=None, uptime_s=None,
                          reparented=None, cwd=None):
    """Classify a NON-agent process as a leaked test-run orphan. Notify-only:
    verdict is TEST_ORPHAN or NOT_CANDIDATE, never enforce-eligible -- these
    are reaped by the interactive notification action, never autonomously."""
    exempt_pids = exempt_pids or set()
    result = {
        "pid": snapshot["pid"], "comm": snapshot["comm"], "cmdline": snapshot["cmdline"],
        "ppid": snapshot["ppid"], "tty_nr": snapshot["tty_nr"], "kind": "leaked-test-orphan",
        "reason": None, "age_s": None, "verdict": "NOT_CANDIDATE",
    }
    if is_denylisted(snapshot["comm"], snapshot["cmdline"]):
        return result
    if snapshot["pid"] in exempt_pids:
        return result
    if is_agent_class(snapshot["cmdline"]):
        return result  # agents go through evaluate(), never here
    if snapshot["tty_nr"] != 0:
        return result  # a controlling terminal means it is interactive, not leaked

    if cwd is None:
        cwd = read_proc_cwd(snapshot["pid"])
    reason = test_orphan_reason(snapshot["cmdline"], cwd)
    if reason is None:
        return result
    result["reason"] = reason

    if reparented is None:
        reparented = is_reparented(snapshot["ppid"])
    if not reparented:
        return result  # parent still alive -> a live test, not an orphan

    if uptime_s is None:
        uptime_s = read_uptime_s()
    if uptime_s is not None:
        result["age_s"] = uptime_s - snapshot["starttime"] / HZ
    if result["age_s"] is None or result["age_s"] < TEST_ORPHAN_MIN_AGE_S:
        return result  # too young: a cleanup trap may still reap it

    result["verdict"] = "TEST_ORPHAN"
    return result


def evaluate_orphaned_sidecar(snapshot, exempt_pids=None, uptime_s=None,
                              reparented=None):
    """Classify a per-session helper daemon whose agent session is gone.
    Auto-reaped: a sidecar without its session has no function, so there is
    no decision for the user to make."""
    exempt_pids = exempt_pids or set()
    result = {
        "pid": snapshot["pid"], "comm": snapshot["comm"], "cmdline": snapshot["cmdline"],
        "ppid": snapshot["ppid"], "tty_nr": snapshot["tty_nr"], "kind": "orphaned-sidecar",
        "age_s": None, "verdict": "NOT_CANDIDATE",
    }
    if is_denylisted(snapshot["comm"], snapshot["cmdline"]):
        return result
    if snapshot["pid"] in exempt_pids:
        return result
    if not is_sidecar(snapshot["cmdline"]):
        return result

    if reparented is None:
        reparented = is_reparented(snapshot["ppid"])
    if not reparented:
        return result  # session still alive -> the sidecar is in use

    if uptime_s is None:
        uptime_s = read_uptime_s()
    if uptime_s is not None:
        result["age_s"] = uptime_s - snapshot["starttime"] / HZ
    if result["age_s"] is None or result["age_s"] < SIDECAR_MIN_AGE_S:
        return result

    result["verdict"] = "ORPHANED_SIDECAR"
    return result


def verify_sidecar_before_kill(pid, state):
    """Re-check the orphaned-sidecar criteria against the live process,
    refusing on pid reuse (starttime mismatch) or a criteria change."""
    snapshot = take_live_snapshot(pid)
    if snapshot is None:
        return None
    history = state.get("pids", {}).get(str(pid))
    if history is None or history.get("starttime") != snapshot["starttime"]:
        return None
    result = evaluate_orphaned_sidecar(snapshot, exempt_pids=dispatcher_exempt_pids())
    if result["verdict"] != "ORPHANED_SIDECAR":
        return None
    result["snapshot"] = snapshot
    return result


def orphan_runtime_trigger(root_pid):
    """(pid, cmdline) of the first process in root's tree running an allowlisted
    dev runtime with no controlling terminal, or None if the tree holds none."""
    for p in collect_descendants(root_pid):
        st = read_proc_stat(p)
        if st is None or st["tty_nr"] != 0:
            continue
        cmd = read_proc_cmdline(p)
        if is_agent_session(st["comm"], cmd) or is_denylisted(st["comm"], cmd):
            continue
        if is_orphan_runtime(cmd):
            return (p, cmd)
    return None


def evaluate_orphaned_runtime(root_snapshot, exempt_pids=None, uptime_s=None,
                              root_of=None, trigger=None):
    """Classify a process tree whose root lost its owner and which still runs a
    dev runtime. Auto-reaped at the ROOT: the tree serves nobody, and reaping the
    root tears down the descendants that a per-process check cannot reach."""
    exempt_pids = exempt_pids or set()
    pid = root_snapshot["pid"]
    result = {
        "pid": pid, "comm": root_snapshot["comm"], "cmdline": root_snapshot["cmdline"],
        "ppid": root_snapshot["ppid"], "tty_nr": root_snapshot["tty_nr"],
        "kind": "orphaned-runtime", "age_s": None, "trigger_pid": None,
        "trigger_cmdline": None, "verdict": "NOT_CANDIDATE",
    }
    if is_denylisted(root_snapshot["comm"], root_snapshot["cmdline"]):
        return result
    if pid in exempt_pids:
        return result
    if is_agent_session(root_snapshot["comm"], root_snapshot["cmdline"]):
        return result  # an orphaned agent session is class 1's call, not ours
    if root_snapshot["tty_nr"] != 0:
        return result  # a terminal started it -- never autonomously reaped

    if root_of is None:
        root_of = orphan_tree_root(pid)
    if root_of != pid:
        return result  # an owner appeared, or pid is not the top of its tree

    if uptime_s is None:
        uptime_s = read_uptime_s()
    if uptime_s is not None:
        result["age_s"] = uptime_s - root_snapshot["starttime"] / HZ
    if result["age_s"] is None or result["age_s"] < ORPHAN_RUNTIME_MIN_AGE_S:
        return result

    if trigger is None:
        trigger = orphan_runtime_trigger(pid)
    if trigger is None:
        return result
    result["trigger_pid"], result["trigger_cmdline"] = trigger
    result["verdict"] = "ORPHANED_RUNTIME"
    return result


def verify_orphaned_runtime_before_kill(pid, state):
    """Re-check the orphaned-runtime criteria against the live tree, refusing on
    pid reuse (starttime mismatch) or a criteria change since the scan."""
    snapshot = take_live_snapshot(pid)
    if snapshot is None:
        return None
    history = state.get("pids", {}).get(str(pid))
    if history is None or history.get("starttime") != snapshot["starttime"]:
        return None
    result = evaluate_orphaned_runtime(snapshot, exempt_pids=dispatcher_exempt_pids())
    if result["verdict"] != "ORPHANED_RUNTIME":
        return None
    result["snapshot"] = snapshot
    return result


_SESSION_UUID_RE = re.compile(r"/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/tasks/")


def child_stdout_path(child_pid):
    try:
        return os.readlink(f"/proc/{child_pid}/fd/1")
    except OSError:
        return None


def output_file_size(path, child_pid=None):
    """Size of the tool's output file; falls back through the child's mount
    namespace (/proc/<pid>/root) for tmpjail'd sessions."""
    if not path or not path.startswith("/"):
        return None
    try:
        return os.path.getsize(path)
    except OSError:
        pass
    if child_pid is None:
        return None
    try:
        return os.path.getsize(f"/proc/{child_pid}/root{path}")
    except OSError:
        return None


def transcript_path_for_child(child_pid):
    """A tool child's stdout targets .../<session-uuid>/tasks/<id>.output;
    that uuid names the session transcript. Sessions run under per-account
    CLAUDE_HOMEs too, and a crash-journal hook seeds tiny synthetic stubs at
    the same name -- search every home and prefer the largest file so a real
    transcript always wins over a stub."""
    target = child_stdout_path(child_pid)
    if target is None:
        return None
    m = _SESSION_UUID_RE.search(target)
    if not m:
        return None
    hits = []
    for pat in TRANSCRIPT_GLOBS:
        hits.extend(glob.glob(os.path.expanduser(pat.format(u=m.group(1)))))
    if not hits:
        return None
    def _size(p):
        try:
            return os.path.getsize(p)
        except OSError:
            return -1
    return max(hits, key=_size)


def _parse_event_ts(ts):
    if not ts:
        return None
    try:
        from datetime import datetime
        return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def read_turn_status(transcript_path, tail_bytes=1048576):
    """(status, blocked_since_epoch) from the transcript tail.
    'in-flight'   -- newest assistant event still ends in a tool_use: the turn
                     is waiting on a tool result and cannot advance;
                     blocked_since is that event's timestamp (immune to mtime
                     noise from queued messages/attachments).
    'advancing'   -- a newer tool_result or real user message exists.
    'no-evidence' -- transcript missing, or only stub/marker lines (a wedged
                     session that never produced a real event)."""
    if not transcript_path:
        return "no-evidence", None
    try:
        with open(transcript_path, "rb") as f:
            f.seek(0, 2)
            size = f.tell()
            f.seek(max(0, size - tail_bytes))
            lines = f.read().decode(errors="replace").split("\n")
    except OSError:
        return "no-evidence", None
    for line in reversed(lines):
        line = line.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
        except ValueError:
            continue
        t = obj.get("type")
        if t == "assistant":
            content = (obj.get("message") or {}).get("content")
            blocks = content if isinstance(content, list) else []
            in_flight = any(isinstance(b, dict) and b.get("type") == "tool_use" for b in blocks)
            return (("in-flight", _parse_event_ts(obj.get("timestamp"))) if in_flight
                    else ("advancing", None))
        if t == "user":
            content = (obj.get("message") or {}).get("content")
            has_tool_result = (isinstance(content, list) and
                               any(isinstance(b, dict) and b.get("type") == "tool_result" for b in content))
            if has_tool_result or not obj.get("isMeta"):
                return "advancing", None
    return "no-evidence", None


def find_tool_shell_children(agent_pid, child_map):
    """Direct shell children of an agent process (its Bash tool calls and any
    user-backgrounded tasks), oldest first, as (pid, stat) pairs."""
    out = []
    for c in child_map.get(agent_pid, []):
        st = read_proc_stat(c)
        if st is None or st["comm"] not in TOOL_SHELL_COMMS:
            continue
        out.append((c, st))
    out.sort(key=lambda p: p[1]["starttime"])
    return out


def subtree_cpu_ticks(root_pid, child_map):
    total = 0
    stack = [root_pid]
    seen = set()
    while stack:
        cur = stack.pop()
        if cur in seen:
            continue
        seen.add(cur)
        st = read_proc_stat(cur)
        if st is not None:
            total += st["utime"] + st["stime"]
        stack.extend(child_map.get(cur, []))
    return total


_DURATION_RE = re.compile(r"(\d+)([smhd]?)$")
_DURATION_MULT = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}


def parse_timeout_bound_s(cmdline):
    """Largest `timeout [flags] N[smhd]` bound present in a command, in seconds.
    Skips flag arguments (`-k 30`, `-s TERM`) so they are never read as the bound."""
    best = None
    for m in re.finditer(r"\btimeout\s+([^|;&()]{0,120})", cmdline or ""):
        toks = m.group(1).split()
        i = 0
        while i < len(toks):
            t = toks[i]
            if t in ("-k", "--kill-after", "-s", "--signal"):
                i += 2
                continue
            if t.startswith("-"):
                i += 1
                continue
            d = _DURATION_RE.fullmatch(t)
            if d:
                secs = int(d.group(1)) * _DURATION_MULT[d.group(2)]
                if best is None or secs > best:
                    best = secs
            break
    return best


def update_turn_history(entry, agent_pid, now, child_map):
    """Track the agent's in-flight tool-shell child inside its state entry,
    accumulating subtree-CPU%% samples. Resets on child identity change.
    When the transcript names the pending tool_use, the child whose start time
    matches it is tracked -- never an older user-backgrounded task."""
    children = find_tool_shell_children(agent_pid, child_map)
    if not children:
        entry["turn"] = None
        return None
    transcript_path = (entry.get("turn") or {}).get("transcript_path")
    if not transcript_path:
        transcript_path = transcript_path_for_child(children[0][0])
    status, blocked_since = read_turn_status(transcript_path)
    uptime = read_uptime_s()

    def wall_start(st):
        return (now - uptime + st["starttime"] / HZ) if uptime else None

    cpid, cst = children[0]
    if blocked_since is not None and uptime:
        cpid, cst = min(children,
                        key=lambda p: abs(wall_start(p[1]) - blocked_since))

    turn = entry.get("turn")
    if (not turn or turn.get("child_pid") != cpid
            or turn.get("child_start") != cst["starttime"]):
        turn = {"child_pid": cpid, "child_start": cst["starttime"],
                "child_cmdline": read_proc_cmdline(cpid),
                "pcpu_history": [], "_last": None}
    turn["transcript_path"] = transcript_path
    turn["turn_status"] = status
    turn["blocked_since_ts"] = blocked_since
    turn["child_wall_start"] = wall_start(cst)
    if not turn.get("output_path"):
        turn["output_path"] = child_stdout_path(cpid)
    osize = output_file_size(turn.get("output_path"), cpid)
    if turn.get("last_output_progress") is None or osize != turn.get("output_size"):
        turn["last_output_progress"] = now
    turn["output_size"] = osize
    ticks = subtree_cpu_ticks(cpid, child_map)
    last = turn.get("_last")
    if last is not None:
        dt = now - last["ts"]
        dticks = ticks - last["ticks"]
        if dt > 0 and dticks >= 0:
            pcpu = (dticks / HZ) / dt * 100.0
            turn["pcpu_history"] = (turn.get("pcpu_history", []) + [pcpu])[-HISTORY_MAX:]
    turn["_last"] = {"ts": now, "ticks": ticks}
    entry["turn"] = turn
    return turn


def evaluate_blocked_turn(snapshot, history, exempt_pids=None, uptime_s=None,
                           blocked_min_s=None, cpu_max=None):
    """Classify an agent whose turn is stuck on a low-CPU blocked tool call.
    Notify-only: verdict BLOCKED_TURN or NOT_CANDIDATE, never enforce-eligible.
    The remediation target is the CHILD (kills the blocked command so the turn
    resumes with a failed tool result), never the agent itself."""
    exempt_pids = exempt_pids or set()
    blocked_min_s = BLOCKED_TURN_MIN_S if blocked_min_s is None else blocked_min_s
    cpu_max = BLOCKED_CPU_MAX if cpu_max is None else cpu_max

    result = {
        "pid": snapshot["pid"], "comm": snapshot["comm"], "cmdline": snapshot["cmdline"],
        "ppid": snapshot["ppid"], "tty_nr": snapshot["tty_nr"], "kind": "blocked-turn",
        "child_pid": None, "child_cmdline": None, "child_age_s": None,
        "child_pcpu_now": None, "blocked_age_s": None, "timeout_bound_s": None,
        "turn_status": None, "output_quiet_s": None, "verdict": "NOT_CANDIDATE",
    }
    if (is_denylisted(snapshot["comm"], snapshot["cmdline"])
            or snapshot["pid"] in exempt_pids
            or not is_agent_class(snapshot["cmdline"])):
        return result

    turn = (history or {}).get("turn")
    if not turn:
        return result
    status = turn.get("turn_status")
    result["turn_status"] = status
    if status not in ("in-flight", "no-evidence"):
        return result
    result["child_pid"] = turn["child_pid"]
    result["child_cmdline"] = turn.get("child_cmdline")

    now = snapshot["now"]
    if uptime_s is None:
        uptime_s = read_uptime_s()
    if uptime_s is None:
        return result
    result["child_age_s"] = uptime_s - turn["child_start"] / HZ

    if status == "in-flight":
        blocked_since = turn.get("blocked_since_ts")
        if blocked_since:
            result["blocked_age_s"] = now - blocked_since
            if result["blocked_age_s"] < blocked_min_s:
                return result
            wall_start = turn.get("child_wall_start")
            if (wall_start is not None
                    and abs(wall_start - blocked_since) > BLOCKED_CORRELATE_S):
                return result  # child is not the pending tool call (bg task?)
        if result["child_age_s"] < blocked_min_s:
            return result
    else:
        # no-evidence: transcript exists but is a content-free stub. A child
        # whose session could not even be identified (no transcript_path) is
        # never killable -- is_agent_class false-positives land here.
        if not turn.get("transcript_path"):
            return result
        if result["child_age_s"] < BLOCKED_NOEVIDENCE_MIN_S:
            return result

    # harness-style stall rule: any output-file growth is progress and resets
    # the clock -- gate0 heartbeats keep healthy quiet runs (remote offload
    # waits at ~0%% local CPU) permanently out of reach
    last_prog = turn.get("last_output_progress")
    if last_prog is not None:
        result["output_quiet_s"] = now - last_prog
        quiet_bar = blocked_min_s if status == "in-flight" else BLOCKED_NOEVIDENCE_MIN_S
        if result["output_quiet_s"] < quiet_bar:
            return result

    bound = parse_timeout_bound_s(turn.get("child_cmdline"))
    result["timeout_bound_s"] = bound
    if bound is not None and result["child_age_s"] < bound + BLOCKED_TIMEOUT_SLACK_S:
        result["verdict"] = "BOUNDED_WAIT"
        return result

    recent = turn.get("pcpu_history", [])[-CPU_SAMPLES_REQUIRED:]
    if recent:
        result["child_pcpu_now"] = recent[-1]
    if len(recent) < CPU_SAMPLES_REQUIRED or any(v >= cpu_max for v in recent):
        return result

    result["verdict"] = "BLOCKED_TURN"
    return result


def verify_blocked_turn_before_kill(agent_pid, state):
    """Re-check blocked-turn criteria live; refuses on agent or child pid reuse."""
    snapshot = take_live_snapshot(agent_pid)
    if snapshot is None:
        return None
    history = state.get("pids", {}).get(str(agent_pid))
    if history is None or history.get("starttime") != snapshot["starttime"]:
        return None
    result = evaluate_blocked_turn(snapshot, history, exempt_pids=dispatcher_exempt_pids())
    if result["verdict"] != "BLOCKED_TURN":
        return None
    cst = read_proc_stat(result["child_pid"])
    if cst is None or cst["starttime"] != history["turn"]["child_start"]:
        return None
    turn = history["turn"]
    status, _ = read_turn_status(turn.get("transcript_path"))
    if status == "advancing":
        return None  # tool returned between scan and kill -- turn advancing
    if (turn.get("output_size") is not None
            and output_file_size(turn.get("output_path"), turn.get("child_pid")) != turn["output_size"]):
        return None  # output grew between scan and kill -- child is working
    result["snapshot"] = snapshot
    return result


def build_child_map():
    children = {}
    for d in glob.glob("/proc/[0-9]*"):
        try:
            pid = int(os.path.basename(d))
        except ValueError:
            continue
        st = read_proc_stat(pid)
        if st is None:
            continue
        children.setdefault(st["ppid"], []).append(pid)
    return children


def collect_descendants(pid):
    """pid plus every transitive child, leaves first, so a tree is torn down
    from the bottom up."""
    children = build_child_map()
    ordered = []
    stack = [pid]
    seen = set()
    while stack:
        cur = stack.pop()
        if cur in seen:
            continue
        seen.add(cur)
        ordered.append(cur)
        stack.extend(children.get(cur, []))
    ordered.reverse()
    return ordered


def verify_test_orphan_before_kill(pid, state):
    """Re-check the leaked-orphan criteria against the live process, refusing
    on pid reuse (starttime mismatch) or a criteria change since the scan."""
    snapshot = take_live_snapshot(pid)
    if snapshot is None:
        return None
    history = state.get("pids", {}).get(str(pid))
    if history is None or history.get("starttime") != snapshot["starttime"]:
        return None
    result = evaluate_test_orphan(snapshot, exempt_pids=dispatcher_exempt_pids())
    result["snapshot"] = snapshot
    return result


def reap_tree(pid, evidence):
    """SIGTERM the whole subtree, wait, SIGKILL survivors. Returns an outcome."""
    pids = collect_descendants(pid)
    for p in pids:
        try:
            os.kill(p, signal.SIGTERM)
        except ProcessLookupError:
            continue
        except PermissionError:
            log_event({"action": "reap_denied", "pid": p, "root": pid, "evidence": evidence})
    log_event({"action": "reap_SIGTERM", "root": pid, "tree": pids, "evidence": evidence})

    deadline = time.time() + KILL_WAIT_S
    while time.time() < deadline:
        if all(pid_gone_or_zombie(p) for p in pids):
            log_event({"action": "reaped", "root": pid, "tree": pids})
            return "reaped"
        time.sleep(0.5)

    survivors = [p for p in pids if not pid_gone_or_zombie(p)]
    for p in survivors:
        try:
            os.kill(p, signal.SIGKILL)
        except ProcessLookupError:
            continue
    log_event({"action": "reap_SIGKILL", "root": pid, "survivors": survivors})

    # SIGKILL is asynchronous; the kernel needs a moment to tear a process down
    # and leave it in Z until a subreaper reaps it. Poll briefly before verdict.
    grace = time.time() + 3
    while time.time() < grace:
        if all(pid_gone_or_zombie(p) for p in pids):
            log_event({"action": "reaped", "root": pid, "tree": pids})
            return "reaped"
        time.sleep(0.2)
    return "partial"


def classify_pid_live(pid, state, exempt_pids=None):
    snapshot = take_live_snapshot(pid)
    if snapshot is None:
        return None
    history = state.get("pids", {}).get(str(pid), {})
    return evaluate(snapshot, history, exempt_pids=exempt_pids)


def verify_live_before_kill(pid, state):
    """Re-check candidate criteria against the LIVE process right now.
    Refuses (returns None) on pid reuse or resumed progress."""
    snapshot = take_live_snapshot(pid)
    if snapshot is None:
        return None
    pid_key = str(pid)
    history = state.get("pids", {}).get(pid_key)
    if history is None or history.get("starttime") != snapshot["starttime"]:
        return None
    exempt = dispatcher_exempt_pids()
    result = evaluate(snapshot, history, exempt_pids=exempt)
    result["snapshot"] = snapshot
    return result


def refresh_flagged(state, results):
    """Keep state['flagged'] in sync with current WARN/ENFORCE_ELIGIBLE
    candidates. reaper-notifier watches this for NEW pids to notify on;
    it clears when a pid stops being a candidate (resumed, exited, denylisted)."""
    flagged = state.setdefault("flagged", {})
    now = time.time()
    seen = set()
    for r in results:
        if r["verdict"] in ("WARN", "ENFORCE_ELIGIBLE", "TEST_ORPHAN"):
            pid_key = str(r["pid"])
            seen.add(pid_key)
            existing = flagged.get(pid_key)
            first_ts = existing["first_flagged_ts"] if existing else now
            entry = dict(r)
            entry["first_flagged_ts"] = first_ts
            entry["last_seen_ts"] = now
            flagged[pid_key] = entry
    for pid_key in list(flagged.keys()):
        if pid_key not in seen:
            del flagged[pid_key]
    return flagged


def pid_alive(pid):
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def pid_gone_or_zombie(pid):
    """A SIGKILLed process is briefly a zombie (state Z) until its parent or a
    subreaper reaps it -- os.kill(pid, 0) still succeeds then, so treat Z as
    gone or a tree tear-down reports a false 'partial'."""
    try:
        with open(f"/proc/{pid}/stat", "rb") as f:
            raw = f.read().decode(errors="replace")
    except OSError:
        return True
    rp = raw.rfind(")")
    if rp == -1:
        return True
    rest = raw[rp + 2:].split()
    return (not rest) or rest[0] == "Z"


def log_event(event):
    os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
    event = dict(event)
    event["ts"] = time.time()
    with open(LOG_PATH, "a") as f:
        f.write(json.dumps(event, default=str) + "\n")


def kill_ladder(pid, evidence):
    """SIGTERM, wait, re-verify+SIGKILL. Returns the outcome string."""
    try:
        os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        log_event({"action": "kill_skip", "pid": pid, "reason": "already gone", "evidence": evidence})
        return "already_gone"
    except PermissionError:
        log_event({"action": "kill_denied", "pid": pid, "evidence": evidence})
        return "permission_denied"
    log_event({"action": "SIGTERM", "pid": pid, "evidence": evidence})

    deadline = time.time() + KILL_WAIT_S
    while time.time() < deadline:
        if not pid_alive(pid):
            log_event({"action": "terminated", "pid": pid, "evidence": evidence})
            return "terminated"
        time.sleep(0.5)

    if pid_alive(pid):
        try:
            os.kill(pid, signal.SIGKILL)
            log_event({"action": "SIGKILL", "pid": pid, "evidence": evidence})
            return "killed"
        except ProcessLookupError:
            log_event({"action": "terminated", "pid": pid, "evidence": evidence})
            return "terminated"
    return "terminated"