#!/usr/bin/env python3
"""pids-exhaustion killer + early warning + desktop self-heal.

A saturated pids.max denies fork() to every process in the cgroup, so nothing
inside it can run the command that would fix it. Crossing the limit must
therefore KILL the offending workload rather than freeze everyone sharing the
cgroup: a killed runaway is recoverable in seconds, a wedged cgroup is not.
`--watch` samples the agent-owned scopes and writes cgroup.kill on any that
crosses KILL_PCT of its effective cap. The kill path is pure file I/O — no
subprocess, no fork — so it still works while the machine is out of pids.

Only scopes provably under agent.slice/agent-seat.slice/build.slice and named
by a known agent launcher are killable; everything else, human.slice and
terminal scopes above all, is refused. Unrecognised means not killed.

2026-08-07: a fork bomb saturated app.slice's pids.max=8000. systemd-oomd never
fired — it watches memory/swap PSI, never pids.max. Nothing else was watching
either, so the freeze was silent until the compositor itself could not fork.
This closes that gap: sample pids.current/pids.max on the slices agent work
can saturate, and fire once on the rising edge past a high-water mark — never
every tick (that trains the user to ignore it).

A second, independent check runs every tick regardless of threshold: known
desktop-session processes (compositor, panel, file-manager-desktop) are only
safe from slice-wide pids exhaustion while they live in the login session
scope (session-N.scope), which carries no pids cap. If one is ever found
inside a capped slice — e.g. a manual `cinnamon --replace` recovery landed in
app.slice by systemd-run's default placement — it is reparented back via a
raw cgroup.procs move (no signal, no restart, no visual disturbance) before
it can be caught in the next saturation event.
"""
import json
import os
import re
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "lib"))
from pids_cgroup import (  # noqa: E402
    KILLABLE_SLICES, is_killable, is_killable_slice, same_chain,
)
from stall_detect import classify_cgroup, sample_cgroup, unavailable_signals

UID = os.getuid()
CG_ROOT = f"/sys/fs/cgroup/user.slice/user-{UID}.slice"
USER_SVC = f"{CG_ROOT}/user@{UID}.service"

# Slices sampled for the pressure WARNING. Wider than KILLABLE_SLICES: app.slice
# is watched but never killed, because it holds the user's terminal scopes.
WATCH_SLICES = ["app.slice", "agent.slice", "agent-seat.slice", "unsafe.slice", "build.slice"]

HIGH_PCT = float(os.environ.get("PIDS_HIGH_PCT", 80))
CLEAR_PCT = float(os.environ.get("PIDS_CLEAR_PCT", 65))

KILL_PCT = float(os.environ.get("PIDS_KILL_PCT", 60))
WATCH_INTERVAL = float(os.environ.get("PIDS_WATCH_INTERVAL", 0.1))
# Ticks between two runs of the diagnostic arm (loginctl, /proc walk, notify),
# which uses subprocesses and must never sit in front of a kill.
SLOW_EVERY = int(os.environ.get("PIDS_SLOW_EVERY", 150))
PIDS_STALL_ENABLE = os.environ.get("PIDS_STALL_ENABLE", "1") != "0"
PIDS_STALL_SAMPLE_S = float(os.environ.get("PIDS_STALL_SAMPLE_S", "30"))
PIDS_STALL_IDLE_WINDOW_S = float(os.environ.get("PIDS_STALL_IDLE_WINDOW_S", "1800"))
# Observation gate only, and relative to the machine rather than a fixed byte
# count: a scope too small to matter is not worth sampling every tick.
PIDS_STALL_FLOOR_FRACTION = float(os.environ.get("PIDS_STALL_FLOOR_FRACTION", "0.05"))
PIDS_STALL_MIN_PIDS = int(os.environ.get("PIDS_STALL_MIN_PIDS", "2"))
# /proc/pressure/memory "some avg60": the share of the last minute in which some
# task was stalled waiting on memory. A stalled cgroup on an unpressured machine
# is consuming nothing from anyone, so it is not a runaway.
PIDS_STALL_PRESSURE_PCT = float(os.environ.get("PIDS_STALL_PRESSURE_PCT", "10"))
# A killed cgroup keeps reporting tasks while they exit; without this every tick
# would re-kill and re-log it.
KILL_COOLDOWN_S = float(os.environ.get("PIDS_KILL_COOLDOWN", 10))
# Not cgroup paths, so they can share the stall arm's cooldown table without colliding.
NO_LEDGER_KEY = "\x00no-ledger"
PRESSURE_KEY = "\x00no-pressure"
BLIND_KEY = "\x00blind:"
# Diagnostics repeat every tick until the condition clears; the log is the record
# of record and must stay readable.
STALL_LOG_EVERY_S = float(os.environ.get("PIDS_STALL_LOG_EVERY", "300"))

# Desktop alerts are opt-in: this runs unattended and its events log is the
# record of record.
NOTIFY = os.environ.get("PIDS_GUARD_NOTIFY") == "1"

# Child cgroups listed alongside a pressure warning.
HOLDERS_TOP_N = int(os.environ.get("PIDS_HOLDERS_TOP", 5))

# Desktop-session processes that must live in the login session scope
# (session-N.scope, uncapped) rather than any user@.service slice.
DESKTOP_PROCS = {
    "cinnamon", "cinnamon-session-binary", "cinnamon-launcher",
    "nemo-desktop", "xapp-sn-watcher", "cinnamon-settings-daemon",
    "cinnamon-killer-daemon",
}

STATE_DIR = os.path.join(
    os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state")), "pids-guard"
)
STATE_FILE = os.path.join(STATE_DIR, "state.json")
LOG = os.path.join(STATE_DIR, "events.log")
STALL_STATE_FILE = os.path.join(STATE_DIR, "stall-state.json")
os.makedirs(STATE_DIR, exist_ok=True)


# Resolved per call, not at import: the tests point it at a temp ledger.
def _ledger_sessions_dir():
    base = os.environ.get(
        "PIDS_STALL_LEDGER_DIR",
        os.environ.get("AGENT_SESSIONS_DIR", "~/.local/state/agent-sessions"),
    )
    return os.path.join(os.path.expanduser(base), "sessions")


def logline(s):
    with open(LOG, "a") as f:
        f.write(time.strftime("%F %T") + "  " + s + "\n")


def notify(title, body, urgency="critical"):
    if not NOTIFY:
        return
    try:
        subprocess.run(["notify-send", "-u", urgency, title, body], timeout=10)
    except (OSError, subprocess.SubprocessError):
        pass


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


def save_state(state):
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump(state, f)
    os.replace(tmp, STATE_FILE)


def _load_stall_state():
    try:
        with open(STALL_STATE_FILE) as f:
            raw = json.load(f)
    except (OSError, ValueError):
        return {}
    if not isinstance(raw, dict):
        return {}
    cleaned = {}
    for path, samples in raw.items():
        if not isinstance(path, str) or not isinstance(samples, list):
            continue
        tidy = []
        for sample in samples:
            if not isinstance(sample, dict):
                continue
            if not all(k in sample for k in ("t", "cpu", "io", "pids", "mem")):
                continue
            try:
                tidy.append({
                    "t": float(sample["t"]),
                    "cpu": int(sample["cpu"]),
                    "io": int(sample["io"]),
                    "pids": int(sample["pids"]),
                    "mem": int(sample["mem"]),
                })
            except (TypeError, ValueError):
                continue
        if tidy:
            cleaned[path] = tidy
    return cleaned


def _save_stall_state(state):
    tmp = STALL_STATE_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump(state, f)
    os.replace(tmp, STALL_STATE_FILE)


def _read_cgroup_procs(path):
    try:
        with open(os.path.join(path, "cgroup.procs")) as f:
            procs = []
            for line in f:
                try:
                    procs.append(int(line.strip()))
                except ValueError:
                    continue
            return procs
    except OSError:
        return None


def _comm_for_pid(pid):
    try:
        with open(f"/proc/{pid}/comm") as f:
            return f.read().strip()
    except OSError:
        return None


def _truncate_history(samples, window_s, now_t):
    cutoff = now_t - (window_s + PIDS_STALL_SAMPLE_S)
    while samples and samples[0]["t"] < cutoff:
        samples.pop(0)


def _log_throttled(recent, now, key, message):
    if now - recent.get(key, -STALL_LOG_EVERY_S) < STALL_LOG_EVERY_S:
        return
    recent[key] = now
    logline(message)


def _mem_total_bytes():
    try:
        with open("/proc/meminfo") as f:
            for line in f:
                if line.startswith("MemTotal:"):
                    return int(line.split()[1]) * 1024
    except (OSError, ValueError, IndexError):
        return None
    return None


def _stall_floor_bytes():
    total = _mem_total_bytes()
    if total is None:
        return None
    return int(total * PIDS_STALL_FLOOR_FRACTION)


def _memory_pressure_pct():
    """`some avg60` from /proc/pressure/memory, or None when it cannot be read."""
    try:
        with open("/proc/pressure/memory") as f:
            for line in f:
                if not line.startswith("some"):
                    continue
                for field in line.split()[1:]:
                    name, _, value = field.partition("=")
                    if name == "avg60":
                        return float(value)
    except (OSError, ValueError):
        return None
    return None


def _stalled_cgroup_history(history, path, sample, now_t=None):
    if now_t is None:
        now_t = time.time()
    samples = history.get(path, [])
    # A guard that was stopped must not wake up and read the gap as flatness.
    if samples and now_t - samples[-1]["t"] > 3.0 * PIDS_STALL_SAMPLE_S:
        samples = []
    samples.append(sample)
    _truncate_history(samples, PIDS_STALL_IDLE_WINDOW_S, now_t)
    history[path] = samples


def _owner_session_pids():
    """Pids of live owner-launched sessions, or None when the ledger cannot answer.

    None and an empty set are different verdicts: empty means the ledger is healthy
    and no owner session is running; None means it holds no live record at all, so
    it cannot vouch for any cgroup.
    """
    owner_pids = set()
    live = 0
    ledger_sessions_dir = _ledger_sessions_dir()
    try:
        entries = os.listdir(ledger_sessions_dir)
    except OSError:
        return None

    for entry in entries:
        if not entry.endswith(".json"):
            continue
        try:
            with open(os.path.join(ledger_sessions_dir, entry)) as f:
                record = json.load(f)
        except (OSError, ValueError):
            continue
        if record.get("finishedAt") is not None:
            continue
        pid = record.get("pid")
        if not isinstance(pid, int):
            continue
        live += 1
        if record.get("launchedBy") == "user":
            owner_pids.add(pid)
    return owner_pids if live else None


def _has_user_owner(path, owners=None):
    procs = _read_cgroup_procs(path)
    if procs is None:
        return False
    if owners is None:
        owners = _owner_session_pids() or set()
    return any(pid in owners for pid in procs)


def _comms_for_path(path, limit=5):
    procs = _read_cgroup_procs(path)
    if procs is None:
        return []
    comms = []
    for pid in procs:
        if len(comms) >= limit:
            break
        name = _comm_for_pid(pid)
        if name:
            comms.append(name)
    return comms


def read_int(path):
    try:
        with open(path) as f:
            v = f.read().strip()
        return None if v == "max" else int(v)
    except (OSError, ValueError):
        return None


def slice_holders(slice_dir, top_n=HOLDERS_TOP_N):
    """Direct child cgroups of slice_dir holding tasks, biggest first.

    Direct children only: pids.current is aggregated over descendants, so a
    deeper walk would count the same tasks once per ancestor. Filesystem reads
    only — this runs while fork() is failing, when subprocesses cannot start.
    """
    holders = []
    try:
        entries = os.listdir(slice_dir)
    except OSError:
        return []
    for name in entries:
        n = read_int(os.path.join(slice_dir, name, "pids.current"))
        if n:
            holders.append((name, n))
    holders.sort(key=lambda h: (-h[1], h[0]))
    return holders[:top_n]


_diffuse_log_at = {}
DIFFUSE_LOG_EVERY_S = float(os.environ.get("PIDS_DIFFUSE_LOG_EVERY", 60))
# Minimum share of a saturated slice's DIRECT children's pids (every immediate
# child cgroup, killable or not -- pids.current is aggregated over descendants,
# so summing direct children rather than all depths avoids double-counting)
# that the largest killable child must hold before it is blamed. A slice can be
# packed mostly with cgroups this guard cannot or should not kill (e.g. a
# session-admission wait triple); measuring share against the slice's raw cur
# still credits the victim with the whole slice even when it is a sliver of
# the real population, so direct-children total is what "share" means here.
# Below this floor the slice is "diffuse saturation" -- many small cgroups,
# no single runaway -- and killing the largest killable one frees nothing: a
# new scope replaces it within seconds and a different victim is picked next
# tick, forever, while the pids that actually fill the slice go untouched. A
# real single-scope runaway clears this bar easily (the six-scopes-at-half-cap
# example below each hold ~16.7% of the slice's direct-children total).
MIN_VICTIM_SHARE_PCT = float(os.environ.get("PIDS_KILL_MIN_VICTIM_SHARE_PCT", 10))


def kill_targets(nodes, pct=None, self_rel=None):
    """Killable cgroups at or past `pct` of their cap, deepest first.

    An ancestor is dropped when a descendant also qualifies: killing the
    nested scope that actually holds the runaway spares its parent. A node
    sharing a chain with `self_rel` is dropped so the guard cannot kill itself.

    A slice can saturate while every scope inside it stays under its own
    threshold — six scopes at half of a 1024 cap fill a 3072 slice, and then
    all six are fork-denied with nothing over the line. So a killable slice
    past the threshold contributes its largest scope as a target -- unless that
    largest scope fails MIN_VICTIM_SHARE_PCT, in which case the slice is
    diffusely saturated and nothing is killed (see DIFFUSE-SATURATION log).
    """
    if pct is None:
        pct = KILL_PCT

    def over(n):
        return (n.get("cap") and n.get("cur") is not None
                and n["cur"] * 100.0 >= pct * n["cap"])

    def eligible(n):
        return (is_killable(n.get("rel")) and n.get("cur") is not None
                and not (self_rel and same_chain(n["rel"], self_rel)))

    hits = [n for n in nodes if over(n) and eligible(n)]
    targets = [
        n for n in hits
        if not any(o["rel"].startswith(n["rel"] + "/") for o in hits)
    ]
    for n in targets:
        n["why"] = f"crossed {pct:.0f}% of its cap"

    for s in nodes:
        if not (is_killable_slice(s.get("rel")) and over(s)):
            continue
        if any(t["rel"].startswith(s["rel"] + "/") for t in targets):
            continue
        prefix = s["rel"] + "/"
        direct_depth = prefix.count("/")
        direct_children = [n for n in nodes
                            if n["rel"].startswith(prefix)
                            and n["rel"].count("/") == direct_depth
                            and n.get("cur") is not None]
        accounted = sum(n["cur"] for n in direct_children)
        inside = [n for n in nodes
                  if n["rel"].startswith(prefix) and eligible(n)]
        if inside:
            biggest = max(inside, key=lambda n: (n["cur"], n["rel"]))
            share_pct = 100.0 * biggest["cur"] / accounted if accounted else 0.0
            if share_pct < MIN_VICTIM_SHARE_PCT:
                now = time.monotonic()
                if now - _diffuse_log_at.get(s["rel"], -DIFFUSE_LOG_EVERY_S) >= DIFFUSE_LOG_EVERY_S:
                    _diffuse_log_at[s["rel"]] = now
                    logline(
                        f"DIFFUSE-SATURATION {s['rel']} {s['cur']}/{s['cap']} "
                        f"largest scope {biggest['rel']} holds {biggest['cur']} of "
                        f"{accounted} accounted-for pids ({share_pct:.1f}%, floor "
                        f"{MIN_VICTIM_SHARE_PCT:.0f}%) -- no single scope to blame, "
                        f"nothing killed"
                    )
                continue
            biggest["why"] = (f"largest scope in {s['rel'].rsplit('/', 1)[-1]}, "
                              f"itself at {s['cur']}/{s['cap']}")
            targets.append(biggest)
    return targets


def scan_nodes(user_svc=None, slices=KILLABLE_SLICES):
    """Every cgroup under `slices` with its pids count and effective cap.

    The effective cap is the tightest pids.max on the path from the slice down,
    because that is the value the kernel enforces on the node's own forks.
    """
    if user_svc is None:
        user_svc = USER_SVC
    rel_root = user_svc[len("/sys/fs/cgroup"):] if user_svc.startswith("/sys/fs/cgroup") else ""
    nodes = []

    def walk(path, rel, cap):
        own = read_int(os.path.join(path, "pids.max"))
        if own is not None:
            cap = own if cap is None else min(cap, own)
        cur = read_int(os.path.join(path, "pids.current"))
        if cur is not None:
            nodes.append({"path": path, "rel": rel, "cur": cur, "cap": cap})
        try:
            entries = os.listdir(path)
        except OSError:
            return
        for name in entries:
            child = os.path.join(path, name)
            if os.path.isdir(child):
                walk(child, rel + "/" + name, cap)

    for slice_name in slices:
        d = os.path.join(user_svc, slice_name)
        if os.path.isdir(d):
            walk(d, rel_root + "/" + slice_name, None)
    return nodes


def kill_cgroup(path):
    """Write cgroup.kill. Pure file I/O: no fork, so it works out of pids."""
    try:
        with open(os.path.join(path, "cgroup.kill"), "w") as f:
            f.write("1")
        return True
    except OSError:
        return False


def self_cgroup():
    try:
        with open("/proc/self/cgroup") as f:
            for line in f:
                if line.startswith("0::"):
                    return line[3:].strip() or None
    except OSError:
        pass
    return None


def enforce_kills(recent, now=None, user_svc=None, self_rel=None):
    """Kill every over-threshold agent scope. Returns the paths killed."""
    if now is None:
        now = time.monotonic()
    killed = []
    for n in kill_targets(scan_nodes(user_svc), self_rel=self_rel):
        if now - recent.get(n["rel"], -KILL_COOLDOWN_S) < KILL_COOLDOWN_S:
            continue
        recent[n["rel"]] = now
        ok = kill_cgroup(n["path"])
        logline(
            f"{'KILL' if ok else 'KILL-FAILED'} {n['rel']} pids {n['cur']}/{n['cap']} "
            f"{n.get('why', 'over threshold')}"
        )
        if ok:
            killed.append(n["path"])
    return killed


def enforce_stall(recent, history, now=None, user_svc=None):
    if not PIDS_STALL_ENABLE:
        return
    if now is None:
        now = time.monotonic()
    now_t = time.time()

    floor = _stall_floor_bytes()
    if floor is None:
        floor = 0
    nodes = [n for n in scan_nodes(user_svc) if is_killable(n.get("rel"), uid=UID)]
    by_path = {n["path"]: n for n in nodes}
    active = set(by_path)

    for path in list(history):
        if path not in active:
            history.pop(path, None)

    for n in nodes:
        path = n["path"]
        # Gate on two single-file reads before sample_cgroup, which walks
        # /proc/<pid>/io for every task and so costs the most exactly when the
        # 100ms kill arm can least afford to wait for it.
        mem = read_int(os.path.join(path, "memory.current"))
        pids = read_int(os.path.join(path, "pids.current"))
        if mem is None or pids is None or mem < floor or pids < PIDS_STALL_MIN_PIDS:
            history.pop(path, None)
            continue
        sample = sample_cgroup(path)
        if sample is None:
            history.pop(path, None)
            continue
        _stalled_cgroup_history(history, path, sample, now_t)

    owners = _owner_session_pids()
    if owners is None:
        # The ledger is the only thing that marks an owner session, and an owner
        # session lives in the same agent.slice this arm kills from. History keeps
        # accumulating; only the kill decision waits for a ledger that can answer.
        if now - recent.get(NO_LEDGER_KEY, -KILL_COOLDOWN_S) >= KILL_COOLDOWN_S:
            recent[NO_LEDGER_KEY] = now
            logline("STALL-SKIP-NO-LEDGER")
        return

    candidates = []
    for path, samples in list(history.items()):
        rel = by_path[path]["rel"]
        absent = unavailable_signals(samples)
        if absent:
            # Never a silent no-op: the arm reports which signal it cannot see
            # rather than degrading into a narrower rule that still fires.
            _log_throttled(recent, now, f"{BLIND_KEY}{rel}",
                           f"STALL-BLIND {rel} unreadable={'+'.join(absent)}")
            continue
        reason = classify_cgroup(samples, PIDS_STALL_IDLE_WINDOW_S)
        if reason is None:
            continue
        if _has_user_owner(path, owners):
            logline(f"STALL-SKIP-OWNER {rel} {reason}")
            continue
        candidates.append((path, rel, samples[-1], reason))

    if not candidates:
        return

    # A stalled cgroup that is not taking anything from anyone is not a runaway:
    # it is an agent blocked on a remote build. Only scarcity makes it one.
    pressure = _memory_pressure_pct()
    if pressure is None or pressure < PIDS_STALL_PRESSURE_PCT:
        shown = "unreadable" if pressure is None else f"{pressure:.2f}"
        _log_throttled(recent, now, PRESSURE_KEY,
                       f"STALL-SKIP-NO-PRESSURE stalled={len(candidates)} "
                       f"mem_pressure={shown} need={PIDS_STALL_PRESSURE_PCT}")
        return

    # Only the largest holder: killing a small stalled scope costs the owner work
    # without returning the memory the machine is short of.
    path, rel, sample, reason = max(candidates, key=lambda c: c[2]["mem"])
    if now - recent.get(rel, -KILL_COOLDOWN_S) < KILL_COOLDOWN_S:
        return
    recent[rel] = now
    comm = ", ".join(_comms_for_path(path, limit=5)) or "-"
    ok = kill_cgroup(path)
    logline(
        f"{('STALL-KILL' if ok else 'STALL-KILL-FAILED')} {rel} mem={sample['mem']} "
        f"pids={sample['pids']} mem_pressure={pressure:.2f} reason={reason} comm={comm}"
    )


def _slice_entry(raw):
    if isinstance(raw, dict):
        return dict(raw)
    if isinstance(raw, str):
        return {"band": raw}
    return {}


def check_pids_pressure(state):
    changed = False
    slices = state.setdefault("slices", {})
    for slice_name in WATCH_SLICES:
        d = os.path.join(USER_SVC, slice_name)
        cur = read_int(os.path.join(d, "pids.current"))
        cap = read_int(os.path.join(d, "pids.max"))
        if cur is None or cap is None or cap == 0:
            continue  # slice absent, or uncapped: nothing to watch
        pct = 100.0 * cur / cap
        entry = _slice_entry(slices.get(slice_name))
        band = entry.get("band", "ok")
        prev_cap = entry.get("cap")
        # Percent-of-cap is meaningless across a cap change: on 2026-08-07 the
        # cap went 8000 -> 24000 and the count 7879 -> 11985, which read as a
        # 98% -> 50% "recovery" while the load grew 52%.
        if band == "warn" and prev_cap is not None and prev_cap != cap:
            logline(
                f"CAP-CHANGE {slice_name} pids.max {prev_cap} -> {cap} at pids {cur}; "
                "band reset, not a recovery"
            )
            band, entry = "ok", {}
            changed = True
        # A warn band with no recorded count (legacy state) adopts this sample,
        # so clearing still requires a real fall rather than a smaller ratio.
        warn_cur = None
        if band == "warn":
            warn_cur = entry.get("warn_cur")
            if warn_cur is None:
                warn_cur = cur

        if pct >= HIGH_PCT and band == "ok":
            band, warn_cur = "warn", cur
            changed = True
            msg = f"{slice_name} pids {cur}/{cap} ({pct:.0f}%) crossed {HIGH_PCT:.0f}% high-water mark"
            logline("WARN " + msg)
            holders = slice_holders(d)
            logline(
                f"HOLDERS {slice_name} "
                + (", ".join(f"{n}={c}" for n, c in holders) if holders else "(none)")
            )
            notify("pids-guard: exhaustion risk", msg)
        elif band == "warn" and pct <= CLEAR_PCT and cur < warn_cur:
            logline(
                f"CLEAR {slice_name} pids {cur}/{cap} ({pct:.0f}%) back under "
                f"{CLEAR_PCT:.0f}%, down from the {warn_cur} that raised it"
            )
            band, warn_cur = "ok", None
            changed = True

        slices[slice_name] = {"band": band, "cap": cap, "warn_cur": warn_cur}
    return changed


def graphical_session_scope():
    """Login session id for the active seat0 graphical session, or None."""
    try:
        out = subprocess.run(
            ["loginctl", "list-sessions", "--no-legend"],
            capture_output=True, text=True, timeout=5,
        ).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    for line in out.splitlines():
        sid = line.split()[0] if line.split() else None
        if not sid:
            continue
        try:
            props = subprocess.run(
                ["loginctl", "show-session", sid, "-p", "Seat", "-p", "Class", "-p", "Type"],
                capture_output=True, text=True, timeout=5,
            ).stdout
        except (OSError, subprocess.SubprocessError):
            continue
        kv = dict(p.split("=", 1) for p in props.splitlines() if "=" in p)
        if kv.get("Seat") == "seat0" and kv.get("Class") == "user" and kv.get("Type") in ("x11", "wayland"):
            return sid
    return None


def proc_name(pid):
    try:
        with open(f"/proc/{pid}/comm") as f:
            return f.read().strip()
    except OSError:
        return None


def proc_cgroup(pid):
    try:
        with open(f"/proc/{pid}/cgroup") as f:
            return f.read().strip().split(":", 2)[-1]
    except (OSError, IndexError):
        return None


def try_reparent(sid, pid):
    """Best-effort move of pid into session-{sid}.scope. Returns True on success.

    Two mechanisms, in order:
      1. systemd's own AttachProcessesToUnit D-Bus call — the sanctioned path,
         but systemd refuses it for any unit without Delegate=yes, which login
         session scopes are not.
      2. a raw cgroup.procs write — works at the kernel level regardless of
         systemd's delegation policy, but the file is root:root/nobody-owned
         for a non-delegated scope, so an unprivileged caller gets EACCES.
    Neither works for an unprivileged process against a login session scope;
    both are attempted anyway so this keeps working the day either changes
    (e.g. a future systemd marks session scopes delegated) without a rewrite.
    """
    try:
        r = subprocess.run(
            ["busctl", "call", "--quiet", "org.freedesktop.systemd1",
             "/org/freedesktop/systemd1", "org.freedesktop.systemd1.Manager",
             "AttachProcessesToUnit", "ssau", f"session-{sid}.scope", "", "1", str(pid)],
            capture_output=True, text=True, timeout=5,
        )
        if r.returncode == 0:
            return True
    except (OSError, subprocess.SubprocessError):
        pass
    target_procs = os.path.join(CG_ROOT, f"session-{sid}.scope", "cgroup.procs")
    try:
        with open(target_procs, "w") as f:
            f.write(str(pid))
        return True
    except OSError:
        return False


def misplaced_edges(current, previous):
    """Rising edges of the misplaced set, keyed by (process, cgroup) — never pid.

    A misplaced desktop process that respawns keeps its name and cgroup but gets a
    new pid every time; a pid-keyed identity therefore re-fires on every respawn.
    Returns (edges_to_notify, new_state); pairs absent from `current` are dropped,
    so a pair that clears and later returns fires again.
    """
    cur = sorted({tuple(p) for p in current})
    prev = {tuple(p) for p in previous}
    return [p for p in cur if p not in prev], [list(p) for p in cur]


def self_heal_desktop_placement(state):
    """Detect (and reparent where privilege allows) a misplaced desktop process.

    Reparenting a login session scope requires root — see try_reparent(). As an
    unprivileged user-level timer this can only DETECT the hazard reliably; it
    still attempts the move so it self-heals for free the moment it ever runs
    with the needed privilege. Either way the event fires once per (process,
    cgroup) pair, never every tick.
    """
    sid = graphical_session_scope()
    if sid is None:
        return  # no active graphical session right now; nothing to protect
    still_misplaced = []
    pids_by_pair = {}
    for entry in os.listdir("/proc"):
        if not entry.isdigit():
            continue
        pid = int(entry)
        name = proc_name(pid)
        if name not in DESKTOP_PROCS:
            continue
        cg = proc_cgroup(pid)
        if cg is None or re.search(rf"/session-{re.escape(sid)}\.scope$", cg):
            continue  # already correctly placed
        if try_reparent(sid, pid):
            msg = f"moved {name} pid={pid} from {cg} to session-{sid}.scope"
            logline("HEAL " + msg)
            notify("pids-guard: desktop process reparented", msg, urgency="normal")
        else:
            still_misplaced.append((name, cg))
            pids_by_pair.setdefault((name, cg), pid)

    edges, new_state = misplaced_edges(still_misplaced, state.get("misplaced", []))
    state["misplaced"] = new_state
    state.pop("healed", None)
    for name, cg in edges:
        msg = (
            f"{name} pid={pids_by_pair[(name, cg)]} is misplaced in {cg} (not "
            f"session-{sid}.scope) and cannot be self-healed without root — a fork "
            f"bomb in that cgroup can freeze the desktop again. Fix needs either a "
            f"root cgroup move or a logout/login."
        )
        logline("MISPLACED " + msg)
        notify("pids-guard: desktop process misplaced", msg)


def diagnostics(state):
    """Warn/heal arm. Isolated so a failure here can never stop a kill."""
    try:
        check_pids_pressure(state)
        self_heal_desktop_placement(state)
        save_state(state)
    except Exception as err:  # noqa: BLE001 - the kill loop outranks every diagnostic
        logline(f"DIAG-ERROR {type(err).__name__}: {err}")


def run_stall_arm(recent, history, now=None, user_svc=None):
    try:
        enforce_stall(recent, history, now=now, user_svc=user_svc)
    except Exception as err:  # noqa: BLE001 - stall exceptions never stop kill loop
        logline(f"STALL-ERROR {type(err).__name__}: {err}")


def run_stall_once():
    if not PIDS_STALL_ENABLE:
        return
    recent = {}
    history = _load_stall_state()
    now = time.monotonic()
    run_stall_arm(recent, history, now=now, user_svc=USER_SVC)
    _save_stall_state(history)


def watch():
    state = load_state()
    self_rel = self_cgroup()
    recent = {}
    # Separate from `recent`: a stall decision must never enter the pids arm's
    # cooldown table and suppress a cap kill on the same cgroup.
    stall_recent = {}
    stall_history = {}
    last_stall = time.monotonic()
    tick = 0
    while True:
        now = time.monotonic()
        enforce_kills(recent, self_rel=self_rel)
        if PIDS_STALL_ENABLE and now - last_stall >= PIDS_STALL_SAMPLE_S:
            run_stall_arm(stall_recent, stall_history, now=now, user_svc=USER_SVC)
            last_stall = now
        cutoff = time.monotonic() - KILL_COOLDOWN_S
        for table in (recent, stall_recent):
            for rel in [r for r, t in table.items() if t < cutoff]:
                del table[rel]
        if tick % SLOW_EVERY == 0:
            diagnostics(state)
        tick += 1
        time.sleep(WATCH_INTERVAL)


def main():
    if "--watch" in sys.argv[1:]:
        watch()
        return 0
    if "--stall-once" in sys.argv[1:]:
        run_stall_once()
        return 0
    state = load_state()
    enforce_kills({}, self_rel=self_cgroup())
    diagnostics(state)
    return 0


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