import importlib.machinery
import importlib.util
import os
import tempfile
from pathlib import Path

import pytest

BIN = Path(__file__).resolve().parents[1] / "bin" / "pids-guard"

# STATE_DIR and the events log are resolved at import; without this the suite
# appends test records to the running guard's log, which is its record of record.
_STATE = tempfile.mkdtemp(prefix="pids-guard-test-state-")
os.environ["XDG_STATE_HOME"] = _STATE

spec = importlib.util.spec_from_loader(
    "pids_guard", importlib.machinery.SourceFileLoader("pids_guard", str(BIN))
)
pids_guard = importlib.util.module_from_spec(spec)
spec.loader.exec_module(pids_guard)
misplaced_edges = pids_guard.misplaced_edges
check_pids_pressure = pids_guard.check_pids_pressure


# --- dedupe arm: a respawning process MUST NOT re-notify ---

def test_first_sighting_fires():
    edges, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    assert edges == [("cinnamon", "/app.slice")]
    assert state == [["cinnamon", "/app.slice"]]


def test_same_pair_second_tick_is_quiet():
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    edges, _ = misplaced_edges([("cinnamon", "/app.slice")], state)
    assert edges == []


def test_respawn_with_new_pid_is_quiet():
    # the C210 defect: identity keyed by pid re-fired on every respawn
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    for _ in range(50):
        edges, state = misplaced_edges([("cinnamon", "/app.slice")], state)
        assert edges == []


def test_state_survives_json_round_trip():
    import json
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    edges, _ = misplaced_edges([("cinnamon", "/app.slice")], json.loads(json.dumps(state)))
    assert edges == []


# --- detection arm: genuine new hazards MUST still fire ---

def test_new_cgroup_for_same_process_fires():
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    edges, _ = misplaced_edges([("cinnamon", "/build.slice")], state)
    assert edges == [("cinnamon", "/build.slice")]


def test_second_process_fires():
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    edges, _ = misplaced_edges(
        [("cinnamon", "/app.slice"), ("nemo-desktop", "/app.slice")], state
    )
    assert edges == [("nemo-desktop", "/app.slice")]


def test_cleared_pair_is_pruned_and_refires_on_return():
    _, state = misplaced_edges([("cinnamon", "/app.slice")], [])
    edges, state = misplaced_edges([], state)
    assert edges == [] and state == []
    edges, _ = misplaced_edges([("cinnamon", "/app.slice")], state)
    assert edges == [("cinnamon", "/app.slice")]


# --- pressure arm: hysteresis bands, no flapping ---

def test_pressure_band_fires_once_then_clears(tmp_path, monkeypatch):
    d = tmp_path / "app.slice"
    d.mkdir()
    monkeypatch.setattr(pids_guard, "USER_SVC", str(tmp_path))
    monkeypatch.setattr(pids_guard, "WATCH_SLICES", ["app.slice"])
    fired = []
    monkeypatch.setattr(pids_guard, "notify", lambda *a, **k: fired.append(a))
    monkeypatch.setattr(pids_guard, "logline", lambda *a: None)

    def sample(cur):
        (d / "pids.current").write_text(str(cur))
        (d / "pids.max").write_text("8000")

    state = {}
    sample(7900)  # 98% — over the 80% high-water mark
    check_pids_pressure(state)
    sample(7900)
    check_pids_pressure(state)
    assert len(fired) == 1, "warn must fire on the rising edge only"

    sample(5000)  # 62% — under the 65% clear mark
    check_pids_pressure(state)
    assert state["slices"]["app.slice"]["band"] == "ok"

    sample(7900)
    check_pids_pressure(state)
    assert len(fired) == 2, "a fresh crossing after a clear must fire again"


# --- cap-change arm: a bigger denominator is not a recovery ---


@pytest.fixture
def slice_probe(tmp_path, monkeypatch):
    """Fake app.slice cgroup dir + captured log lines, wired into the module."""
    d = tmp_path / "app.slice"
    d.mkdir()
    monkeypatch.setattr(pids_guard, "USER_SVC", str(tmp_path))
    monkeypatch.setattr(pids_guard, "WATCH_SLICES", ["app.slice"])
    monkeypatch.setattr(pids_guard, "notify", lambda *a, **k: None)
    logged = []
    monkeypatch.setattr(pids_guard, "logline", logged.append)

    def sample(cur, cap):
        (d / "pids.current").write_text(str(cur))
        (d / "pids.max").write_text(str(cap))

    return d, sample, logged


def test_cap_raise_is_not_reported_as_clear(slice_probe):
    # the 2026-08-07 defect: 7879/8000 (98%) WARN then 11985/24000 (50%) CLEAR,
    # declaring recovery while the task count grew 52%
    _, sample, logged = slice_probe
    state = {}
    sample(7879, 8000)
    check_pids_pressure(state)
    logged.clear()
    sample(11985, 24000)
    check_pids_pressure(state)
    assert not any(line.startswith("CLEAR") for line in logged)
    assert any(line.startswith("CAP-CHANGE") for line in logged)


def test_cap_raise_rearms_so_a_new_crossing_fires(slice_probe):
    _, sample, logged = slice_probe
    state = {}
    sample(7879, 8000)
    check_pids_pressure(state)
    sample(11985, 24000)
    check_pids_pressure(state)
    logged.clear()
    sample(20000, 24000)  # 83% of the new cap
    check_pids_pressure(state)
    assert any(line.startswith("WARN") for line in logged)


def test_clear_requires_the_count_to_fall(slice_probe):
    _, sample, logged = slice_probe
    state = {"slices": {"app.slice": "warn"}}  # legacy band with no recorded count
    sample(5200, 8000)  # 65% — at the clear mark, but never observed falling
    check_pids_pressure(state)
    assert not any(line.startswith("CLEAR") for line in logged)
    sample(5000, 8000)
    check_pids_pressure(state)
    assert any(line.startswith("CLEAR") for line in logged)


def test_warn_band_persists_while_the_count_climbs(slice_probe):
    _, sample, logged = slice_probe
    state = {}
    sample(7000, 8000)
    check_pids_pressure(state)
    for cur in (7200, 7500, 7900):
        sample(cur, 8000)
        check_pids_pressure(state)
    assert [line.split()[0] for line in logged].count("WARN") == 1
    assert state["slices"]["app.slice"]["band"] == "warn"


# --- attribution arm: which child cgroup holds the pids ---


def test_warn_logs_the_top_holders(slice_probe):
    d, sample, logged = slice_probe
    for child, n in (("claude-a.scope", 5000), ("build-b.scope", 2000), ("idle.scope", 0)):
        (d / child).mkdir()
        (d / child / "pids.current").write_text(str(n))
    sample(7900, 8000)
    check_pids_pressure(state={})
    holders = [line for line in logged if line.startswith("HOLDERS")]
    assert holders == ["HOLDERS app.slice claude-a.scope=5000, build-b.scope=2000"]


def test_holders_reconcile_with_the_slice_total(slice_probe):
    d, _, _ = slice_probe
    for child, n in (("a.scope", 300), ("b.scope", 200)):
        (d / child).mkdir()
        (d / child / "pids.current").write_text(str(n))
    (d / "pids.current").write_text("543")
    holders = pids_guard.slice_holders(str(d))
    assert sum(n for _, n in holders) <= int((d / "pids.current").read_text())


def test_holders_are_capped_at_top_n(slice_probe):
    d, _, _ = slice_probe
    for i in range(12):
        (d / f"s{i}.scope").mkdir()
        (d / f"s{i}.scope" / "pids.current").write_text(str(100 + i))
    holders = pids_guard.slice_holders(str(d), top_n=3)
    assert [n for _, n in holders] == [111, 110, 109]


# --- notification arm: alerts stay opt-in ---


def test_notify_is_silent_unless_opted_in(monkeypatch):
    calls = []
    monkeypatch.setattr(pids_guard.subprocess, "run", lambda *a, **k: calls.append(a))
    monkeypatch.setattr(pids_guard, "NOTIFY", False)
    pids_guard.notify("t", "b")
    assert calls == []
    monkeypatch.setattr(pids_guard, "NOTIFY", True)
    pids_guard.notify("t", "b")
    assert len(calls) == 1


# --- kill arm: membership, fail-closed ---

UID = pids_guard.UID
AGENT = f"/user.slice/user-{UID}.slice/user@{UID}.service/agent.slice"
BUILD = f"/user.slice/user-{UID}.slice/user@{UID}.service/build.slice"
APP = f"/user.slice/user-{UID}.slice/user@{UID}.service/app.slice"
HUMAN = f"/user.slice/user-{UID}.slice/user@{UID}.service/human.slice"
UNSAFE = f"/user.slice/user-{UID}.slice/user@{UID}.service/unsafe.slice"
is_killable = pids_guard.is_killable
is_killable_slice = pids_guard.is_killable_slice
kill_targets = pids_guard.kill_targets


@pytest.mark.parametrize("rel", [
    f"{AGENT}/confine-agent-2253113-11147.scope",
    f"{AGENT}/run-p3523119-i11906443.scope",
    f"{BUILD}/confine-build-1-2.scope",
    f"/user.slice/user-{UID}.slice/user@{UID}.service/agent-seat.slice/agent-seat-s01.scope",
    f"{UNSAFE}/unsafe-cdx-1754600000000000000-12345.scope",
])
def test_agent_scopes_are_killable(rel):
    assert is_killable(rel) is True


@pytest.mark.parametrize("rel", [
    HUMAN,
    f"{HUMAN}/human-session-tmux.service",
    f"/user.slice/user-{UID}.slice/session-364.scope",
    f"{APP}/app-org.gnome.Terminal.slice/vte-spawn-4013e8e0-5d6f-4b42-9b70-5afc5fb77c9d.scope",
    APP,
    AGENT,                                        # the slice itself: kills every sibling agent
    BUILD,
    f"{AGENT}/confine-agent-2253113-11147.scope/human.slice",
    f"{AGENT}/vte-spawn-abc.scope",               # a human component anywhere in the chain
    f"{AGENT}/mystery.scope",                     # no agent launcher produces this name
    f"{AGENT}/nested.slice",                      # a slice is never a target
    f"/user.slice/user-999.slice/user@999.service/agent.slice/confine-agent-1-2.scope",
    "/machine.slice/machine-dangerlab.slice",
    "",
    None,
    42,
])
def test_everything_else_is_refused(rel):
    assert is_killable(rel) is False


def test_unsafe_slice_is_a_slice_target():
    assert is_killable(UNSAFE) is False
    assert is_killable_slice(UNSAFE) is True


# Captured from this workstation on 2026-08-08 while the owner's own sessions were
# running. uid is pinned so the decision is exercised rather than short-circuited by a
# prefix mismatch when the suite runs under a different uid.
@pytest.mark.parametrize("rel", [
    "/user.slice/user-1000.slice/session-364.scope",
    "/user.slice/user-1000.slice/user@1000.service/app.slice/app-org.gnome.Terminal.slice"
    "/vte-spawn-5602b846-0ac1-4640-89d2-dfc2da8230bd.scope",
    "/user.slice/user-1000.slice/user@1000.service/human.slice/rescue.scope-like",
    "/user.slice/user-1000.slice/user@1000.service/human.slice/human-session-tmux.service",
    "/user.slice/user-1000.slice/user@1000.service/%h",
    "/user.slice/user-1000.slice/user@1000.service/app.slice",
    "/user.slice/user-1000.slice/user@1000.service/unsafe.slice",
])
def test_live_owner_session_cgroups_are_refused(rel):
    assert is_killable(rel, uid=1000) is False


def test_live_unsafe_hatch_scope_is_killable():
    assert is_killable(
        "/user.slice/user-1000.slice/user@1000.service/unsafe.slice"
        "/unsafe-cdx-1754600000000000000-12345.scope",
        uid=1000,
    ) is True


def test_kill_fires_at_the_threshold_not_at_saturation():
    node = {"rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 615, "cap": 1024}
    assert kill_targets([node], pct=60) == [node]
    assert kill_targets([{**node, "cur": 614}], pct=60) == []


def test_normal_agent_peak_is_never_killed():
    # measured 2026-08-07: the busiest healthy confine-agent scope peaked at 276
    node = {"rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 276, "cap": 1024}
    assert kill_targets([node], pct=60) == []


def test_a_saturated_human_scope_is_never_killed():
    nodes = [
        {"rel": f"{APP}/app-org.gnome.Terminal.slice/vte-spawn-a.scope", "cur": 24000, "cap": 24000},
        {"rel": HUMAN, "cur": 9999, "cap": 10000},
        {"rel": f"/user.slice/user-{UID}.slice/session-364.scope", "cur": 4096, "cap": 4096},
    ]
    assert kill_targets(nodes, pct=60) == []


def test_deepest_qualifying_scope_wins():
    parent = {"rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 900, "cap": 1024}
    child = {"rel": f"{AGENT}/confine-agent-1-2.scope/confine-build-3-4.scope", "cur": 900, "cap": 1024}
    assert kill_targets([parent, child], pct=60) == [child]


def test_guard_never_kills_its_own_chain():
    own = f"{AGENT}/confine-agent-1-2.scope"
    node = {"rel": own, "cur": 1000, "cap": 1024}
    assert kill_targets([node], pct=60, self_rel=own) == []
    assert kill_targets([node], pct=60, self_rel=own + "/inner") == []


def test_uncapped_scope_is_not_a_target():
    node = {"rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 99999, "cap": None}
    assert kill_targets([node], pct=60) == []


def test_saturated_slice_kills_its_largest_scope_when_no_scope_qualifies():
    # six agent scopes at half of a 1024 cap fill a 3072 slice: none is over its
    # own threshold, yet all six are about to be fork-denied.
    slice_node = {"rel": AGENT, "cur": 3072, "cap": 3072}
    scopes = [
        {"rel": f"{AGENT}/confine-agent-{i}-1.scope", "cur": 512 + i, "cap": 1024}
        for i in range(6)
    ]
    got = kill_targets([slice_node] + scopes, pct=60)
    assert [n["rel"] for n in got] == [f"{AGENT}/confine-agent-5-1.scope"]
    assert "agent.slice" in got[0]["why"]


def test_saturated_slice_is_never_itself_a_target():
    slice_node = {"rel": BUILD, "cur": 1024, "cap": 1024}
    scope = {"rel": f"{BUILD}/confine-build-1-2.scope", "cur": 256, "cap": 512}
    got = kill_targets([slice_node, scope], pct=60)
    assert [n["rel"] for n in got] == [f"{BUILD}/confine-build-1-2.scope"]


def test_saturated_slice_adds_nothing_when_a_scope_already_qualifies():
    slice_node = {"rel": AGENT, "cur": 3072, "cap": 3072}
    hot = {"rel": f"{AGENT}/confine-agent-1-1.scope", "cur": 1000, "cap": 1024}
    cold = {"rel": f"{AGENT}/confine-agent-2-1.scope", "cur": 100, "cap": 1024}
    assert [n["rel"] for n in kill_targets([slice_node, hot, cold], pct=60)] == [hot["rel"]]


def test_saturated_slice_holding_only_unkillable_children_kills_nothing():
    slice_node = {"rel": AGENT, "cur": 3072, "cap": 3072}
    mystery = {"rel": f"{AGENT}/mystery.scope", "cur": 3000, "cap": 3072}
    assert kill_targets([slice_node, mystery], pct=60) == []


def test_saturated_slice_never_targets_the_guards_own_scope():
    own = f"{AGENT}/confine-agent-9-9.scope"
    slice_node = {"rel": AGENT, "cur": 3072, "cap": 3072}
    mine = {"rel": own, "cur": 900, "cap": 1024}
    other = {"rel": f"{AGENT}/confine-agent-1-1.scope", "cur": 300, "cap": 1024}
    got = kill_targets([slice_node, mine, other], pct=60, self_rel=own)
    assert [n["rel"] for n in got] == [other["rel"]]


def test_saturated_human_slice_is_not_a_saturation_signal():
    slice_node = {"rel": HUMAN, "cur": 10000, "cap": 10000}
    inner = {"rel": f"{HUMAN}/human-session-tmux.service", "cur": 9000, "cap": 10000}
    assert kill_targets([slice_node, inner], pct=60) == []


def test_saturated_slice_with_only_tiny_scopes_kills_nothing():
    # Diffuse saturation: hundreds of small, legitimate scopes fill the slice;
    # no single one is a runaway, so killing the largest of them (a sliver of
    # the total) frees nothing and would just repeat on a fresh scope forever.
    slice_node = {"rel": AGENT, "cur": 2800, "cap": 3072}
    scopes = [
        {"rel": f"{AGENT}/confine-agent-{i}-1.scope", "cur": 4, "cap": 1024}
        for i in range(600)
    ]
    assert kill_targets([slice_node] + scopes, pct=60) == []


def test_saturated_slice_diffuse_saturation_is_logged():
    slice_node = {"rel": AGENT, "cur": 100, "cap": 3072}
    scopes = [
        {"rel": f"{AGENT}/confine-agent-{i}-1.scope", "cur": 1, "cap": 1024}
        for i in range(100)
    ]
    kill_targets([slice_node] + scopes, pct=1)
    with open(pids_guard.LOG) as f:
        log = f.read()
    assert "DIFFUSE-SATURATION" in log
    assert AGENT in log


def test_saturated_slice_share_measured_against_accounted_pids_not_raw_slice():
    # Only a fraction of the slice's raw total is trackable/killable here (the
    # rest is untracked or unkillable); the victim's share must be judged
    # against what IS accounted for, not the slice's full cur.
    slice_node = {"rel": AGENT, "cur": 3072, "cap": 3072}
    scope = {"rel": f"{AGENT}/confine-agent-1-1.scope", "cur": 300, "cap": 1024}
    got = kill_targets([slice_node, scope], pct=60)
    assert [n["rel"] for n in got] == [scope["rel"]]


def test_saturated_unsafe_slice_targets_largest_unsafe_scope():
    slice_node = {"rel": UNSAFE, "cur": 2048, "cap": 2048}
    scopes = [
        {"rel": f"{UNSAFE}/unsafe-cdx-1754600000000000000-111.scope", "cur": 400, "cap": 1024},
        {"rel": f"{UNSAFE}/unsafe-cdx-1754600000000000000-222.scope", "cur": 390, "cap": 1024},
    ]
    got = kill_targets([slice_node, *scopes], pct=60)
    assert [n["rel"] for n in got] == [scopes[0]["rel"]]
    assert "unsafe.slice" in got[0]["why"]


# --- kill arm: the filesystem walk and the write ---


def _fake_tree(tmp_path, cur, cap, name="confine-agent-1-2.scope", parent_cap="max"):
    svc = tmp_path / f"user.slice/user-{UID}.slice/user@{UID}.service"
    d = svc / "agent.slice" / name
    d.mkdir(parents=True)
    (svc / "agent.slice" / "pids.max").write_text(parent_cap)
    (svc / "agent.slice" / "pids.current").write_text(str(cur))
    (d / "pids.max").write_text(str(cap))
    (d / "pids.current").write_text(str(cur))
    (d / "cgroup.kill").write_text("")
    return svc, d


def test_scan_reports_the_tightest_cap_on_the_path(tmp_path, monkeypatch):
    svc, d = _fake_tree(tmp_path, cur=100, cap=4096, parent_cap="1024")
    monkeypatch.setattr(pids_guard, "CG_ROOT", "/sys/fs/cgroup")
    nodes = pids_guard.scan_nodes(user_svc=str(svc))
    leaf = [n for n in nodes if n["path"] == str(d)][0]
    assert leaf["cap"] == 1024, "an ancestor's tighter cap is what the kernel enforces"


def test_enforce_writes_cgroup_kill_for_a_runaway(tmp_path, monkeypatch):
    svc, d = _fake_tree(tmp_path, cur=900, cap=1024)
    monkeypatch.setattr(pids_guard, "logline", lambda *a: None)
    monkeypatch.setattr(pids_guard, "USER_SVC", str(svc))
    monkeypatch.setattr(
        pids_guard, "scan_nodes",
        lambda user_svc=None, slices=None: [
            {"path": str(d), "rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 900, "cap": 1024}
        ],
    )
    killed = pids_guard.enforce_kills({})
    assert killed == [str(d)]
    assert (d / "cgroup.kill").read_text() == "1"


def test_enforce_does_not_rewrite_inside_the_cooldown(tmp_path, monkeypatch):
    svc, d = _fake_tree(tmp_path, cur=900, cap=1024)
    monkeypatch.setattr(pids_guard, "logline", lambda *a: None)
    monkeypatch.setattr(
        pids_guard, "scan_nodes",
        lambda user_svc=None, slices=None: [
            {"path": str(d), "rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 900, "cap": 1024}
        ],
    )
    recent = {}
    assert pids_guard.enforce_kills(recent, now=100.0) == [str(d)]
    assert pids_guard.enforce_kills(recent, now=105.0) == []
    assert pids_guard.enforce_kills(recent, now=200.0) == [str(d)]


def test_kill_path_uses_no_subprocess(monkeypatch):
    monkeypatch.setattr(
        pids_guard.subprocess, "run",
        lambda *a, **k: pytest.fail("the kill path must not fork: it runs when fork() is denied"),
    )
    assert pids_guard.kill_cgroup("/nonexistent/cgroup") is False


def test_diagnostic_failure_cannot_stop_a_kill(monkeypatch):
    logged = []
    monkeypatch.setattr(pids_guard, "logline", logged.append)
    monkeypatch.setattr(pids_guard, "check_pids_pressure", lambda s: (_ for _ in ()).throw(OSError("EAGAIN")))
    pids_guard.diagnostics({})
    assert any(line.startswith("DIAG-ERROR") for line in logged)


# --- the two killers must agree: pids-guard (python) and kill-guard (node) ---

KILL_GUARD = (
    Path(__file__).resolve().parents[2]
    / "workstation" / "claude" / "lib" / "kill-guard.mjs"
)


def _kill_guard_allows(rel):
    import subprocess
    r = subprocess.run(
        ["node", str(KILL_GUARD), f"echo 1 > /sys/fs/cgroup{rel}/cgroup.kill"],
        capture_output=True, text=True, timeout=30,
    )
    return r.returncode == 0


@pytest.mark.skipif(not KILL_GUARD.exists(), reason="kill-guard.mjs not in this checkout")
@pytest.mark.parametrize("rel", [
    HUMAN,
    f"{HUMAN}/human-session-tmux.service",
    f"/user.slice/user-{UID}.slice",
])
def test_a_target_pids_guard_refuses_is_also_refused_by_kill_guard(rel):
    assert is_killable(rel) is False
    assert _kill_guard_allows(rel) is False


@pytest.mark.skipif(not KILL_GUARD.exists(), reason="kill-guard.mjs not in this checkout")
def test_a_target_pids_guard_kills_is_not_a_human_session():
    rel = f"{AGENT}/confine-agent-2253113-11147.scope"
    assert is_killable(rel) is True
    assert _kill_guard_allows(rel) is True

# --- stall arm (cgroup-based) behavior contracts ---


def test_stall_state_retains_recent_samples_only(monkeypatch):
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)

    history = {}
    samples = [
        {"t": 100.0, "cpu": 10, "io": 20, "pids": 2, "mem": 0},
        {"t": 101.0, "cpu": 10, "io": 20, "pids": 2, "mem": 0},
    ]
    pids_guard._stalled_cgroup_history(history, "/scope", samples[0], 100.0)
    pids_guard._stalled_cgroup_history(history, "/scope", samples[1], 150.0)
    assert history["/scope"] == [
        {"t": 101.0, "cpu": 10, "io": 20, "pids": 2, "mem": 0},
    ]


def test_stall_sample_gap_aborts_history(monkeypatch):
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)

    history = {
        "/scope": [
            {"t": 100.0, "cpu": 10, "io": 20, "pids": 2, "mem": 0},
        ]
    }
    pids_guard._stalled_cgroup_history(history, "/scope", {"t": 140.0, "cpu": 10, "io": 20, "pids": 2, "mem": 0}, 140.0)
    assert len(history["/scope"]) == 1
    assert history["/scope"][0]["t"] == 140.0


def test_stall_none_sample_drops_history(tmp_path, monkeypatch):
    cgroup = tmp_path / "scope"
    history = {
        str(cgroup): [
            {"t": 1.0, "cpu": 10, "io": 20, "pids": 3, "mem": 3000000000},
        ],
    }
    monkeypatch.setattr(
        pids_guard,
        "scan_nodes",
        lambda user_svc=None, slices=None: [
            {"path": str(cgroup), "rel": f"{AGENT}/confine-agent-1-2.scope", "cur": 3, "cap": 1024},
        ],
    )
    monkeypatch.setattr(pids_guard, "sample_cgroup", lambda path: None)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert history == {}


def test_stall_removed_paths_drop_history(monkeypatch):
    history = {"/no-longer-there": [{"t": 1.0, "cpu": 10, "io": 20, "pids": 3, "mem": 3000000000}]}
    monkeypatch.setattr(pids_guard, "scan_nodes", lambda user_svc=None, slices=None: [])
    pids_guard.enforce_stall({}, history, now=100.0, user_svc="/")
    assert history == {}


def _healthy_ledger(tmp_path, monkeypatch):
    """A ledger holding a live record that is not the owner's — the arm may act."""
    sessions = tmp_path / "ledger" / "sessions"
    sessions.mkdir(parents=True, exist_ok=True)
    (sessions / "dispatch.json").write_text(
        '{"pid":777777,"finishedAt":null,"launchedBy":"agent"}'
    )
    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(tmp_path / "ledger"))


def _stall_node(tmp_path, monkeypatch, rel, sample):
    cgroup = tmp_path / "scope"
    cgroup.mkdir(exist_ok=True)
    (cgroup / "cgroup.procs").write_text("")
    (cgroup / "cgroup.kill").write_text("")
    (cgroup / "memory.current").write_text(str(sample["mem"]))
    (cgroup / "pids.current").write_text(str(sample["pids"]))
    monkeypatch.setattr(
        pids_guard,
        "scan_nodes",
        lambda user_svc=None, slices=None: [
            {"path": str(cgroup), "rel": rel, "cur": sample["pids"], "cap": 1024},
        ],
    )
    monkeypatch.setattr(pids_guard, "sample_cgroup", lambda path: dict(sample))
    return cgroup


def _under_pressure(monkeypatch, pct=42.0):
    monkeypatch.setattr(pids_guard, "_memory_pressure_pct", lambda: pct)


@pytest.mark.parametrize("sample, why", [
    ({"t": 1.0, "cpu": 10, "io": 20, "pids": 4, "mem": 1}, "below the memory floor"),
    ({"t": 1.0, "cpu": 10, "io": 20, "pids": 1, "mem": 1 << 40}, "a lone process is a wait"),
])
def test_stall_candidate_gates_exclude(tmp_path, monkeypatch, sample, why):
    _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", sample)

    def _refuse(path):
        raise AssertionError("sample_cgroup walked /proc before the cheap gates ran")

    monkeypatch.setattr(pids_guard, "sample_cgroup", _refuse)
    history = {}
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert history == {}, why


def test_stall_never_samples_a_cgroup_pids_guard_refuses(tmp_path, monkeypatch):
    _stall_node(
        tmp_path, monkeypatch,
        f"/user.slice/user-{pids_guard.UID}.slice/user@{pids_guard.UID}.service/human.slice/x.scope",
        {"t": 1.0, "cpu": 10, "io": 20, "pids": 9, "mem": 1 << 40},
    )
    history = {}
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert history == {}


def test_stall_kill_writes_cgroup_kill(tmp_path, monkeypatch):
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)
    _healthy_ledger(tmp_path, monkeypatch)
    _under_pressure(monkeypatch)
    flat = {"t": 90.0, "cpu": 10, "io": 20, "pids": 4, "mem": 1 << 40}
    cgroup = _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", flat)
    history = {str(cgroup): [dict(flat, t=t) for t in (0.0, 30.0, 60.0)]}
    monkeypatch.setattr(pids_guard.time, "time", lambda: 90.0)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert (cgroup / "cgroup.kill").read_text() == "1"


@pytest.mark.parametrize("pressure, why", [
    (None, "an unreadable pressure reading is not evidence of scarcity"),
    (0.0, "an idle machine has nothing to reclaim"),
    (9.99, "below the threshold is still not scarcity"),
])
def test_stall_never_kills_without_memory_pressure(tmp_path, monkeypatch, pressure, why):
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)
    _healthy_ledger(tmp_path, monkeypatch)
    monkeypatch.setattr(pids_guard, "_memory_pressure_pct", lambda: pressure)
    flat = {"t": 90.0, "cpu": 10, "io": 20, "pids": 4, "mem": 1 << 40}
    cgroup = _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", flat)
    history = {str(cgroup): [dict(flat, t=t) for t in (0.0, 30.0, 60.0)]}
    monkeypatch.setattr(pids_guard.time, "time", lambda: 90.0)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert (cgroup / "cgroup.kill").read_text() == "", why


def test_stall_never_kills_when_a_signal_is_unreadable(tmp_path, monkeypatch):
    """A bwrap-confined tree denies /proc/<pid>/io; absent must not read as flat."""
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)
    _healthy_ledger(tmp_path, monkeypatch)
    _under_pressure(monkeypatch)
    flat = {"t": 90.0, "cpu": 10, "io": None, "pids": 4, "mem": 1 << 40}
    cgroup = _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", flat)
    history = {str(cgroup): [dict(flat, t=t) for t in (0.0, 30.0, 60.0)]}
    monkeypatch.setattr(pids_guard.time, "time", lambda: 90.0)
    logged = []
    monkeypatch.setattr(pids_guard, "logline", logged.append)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert (cgroup / "cgroup.kill").read_text() == ""
    assert any(line.startswith("STALL-BLIND") and "io" in line for line in logged)


@pytest.mark.parametrize("state", ["missing", "no-live-record"])
def test_stall_never_kills_when_the_ledger_cannot_name_the_owner(tmp_path, monkeypatch, state):
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)
    root = tmp_path / "ledger"
    if state == "no-live-record":
        sessions = root / "sessions"
        sessions.mkdir(parents=True)
        (sessions / "done.json").write_text(
            '{"pid":999,"finishedAt":"2026-08-01T01:00:00Z","launchedBy":"user"}'
        )
        (sessions / "bad.json").write_text("{")
    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(root))
    flat = {"t": 90.0, "cpu": 10, "io": 20, "pids": 4, "mem": 1 << 40}
    cgroup = _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", flat)
    history = {str(cgroup): [dict(flat, t=t) for t in (0.0, 30.0, 60.0)]}
    monkeypatch.setattr(pids_guard.time, "time", lambda: 90.0)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert (cgroup / "cgroup.kill").read_text() == ""


def test_stall_kill_is_vetoed_by_a_user_owned_session(tmp_path, monkeypatch):
    ledger = tmp_path / "ledger" / "sessions"
    ledger.mkdir(parents=True)
    (ledger / "run.json").write_text(
        '{"schemaVersion":1,"ledgerId":"run","pid":424242,"runtime":"claude",'
        '"startedAt":"2026-08-01T00:00:00Z","finishedAt":null,"launchedBy":"user"}'
    )
    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(tmp_path / "ledger"))
    monkeypatch.setattr(pids_guard, "PIDS_STALL_IDLE_WINDOW_S", 60)
    monkeypatch.setattr(pids_guard, "PIDS_STALL_SAMPLE_S", 10)
    flat = {"t": 90.0, "cpu": 10, "io": 20, "pids": 4, "mem": 1 << 40}
    cgroup = _stall_node(tmp_path, monkeypatch, f"{AGENT}/confine-agent-1-2.scope", flat)
    (cgroup / "cgroup.procs").write_text("424242\n")
    history = {str(cgroup): [dict(flat, t=t) for t in (0.0, 30.0, 60.0)]}
    monkeypatch.setattr(pids_guard.time, "time", lambda: 90.0)
    pids_guard.enforce_stall({}, history, now=100.0, user_svc=str(tmp_path))
    assert (cgroup / "cgroup.kill").read_text() == ""


@pytest.mark.parametrize("record", [
    '{"pid":111,"finishedAt":null,"launchedBy":null}',
    '{"pid":111,"finishedAt":null}',
    '{"pid":111,"finishedAt":"2026-08-01T01:00:00Z","launchedBy":"user"}',
    '{"pid":111,"finishedAt":null,"launchedBy":"agent"}',
    '{"broken',
])
def test_owner_veto_does_not_fire_without_an_explicit_user_launch(tmp_path, monkeypatch, record):
    cgroup = tmp_path / "cg"
    cgroup.mkdir()
    (cgroup / "cgroup.procs").write_text("111\n")
    ledger = tmp_path / "ledger" / "sessions"
    ledger.mkdir(parents=True)
    (ledger / "run.json").write_text(record)
    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(tmp_path / "ledger"))
    assert pids_guard._has_user_owner(str(cgroup)) is False


def test_owner_veto_skips_user_owned_session_records(tmp_path, monkeypatch):
    cgroup = tmp_path / "cg"
    cgroup.mkdir()
    (cgroup / "cgroup.procs").write_text("111\n222\n")

    ledger = tmp_path / "ledger" / "sessions"
    ledger.mkdir(parents=True)
    (ledger / "run.json").write_text(
        '{"schemaVersion":1,"ledgerId":"run","pid":111,"runtime":"codex","'
        'title":"x","startedAt":"2026-08-01T00:00:00Z","finishedAt":null,"launchedBy":"user"}\n'
    )
    (ledger / "other.json").write_text(
        '{"schemaVersion":1,"ledgerId":"other","pid":222,"runtime":"codex","'
        'title":"y","startedAt":"2026-08-01T00:00:00Z","finishedAt":"2026-08-01T01:00:00Z","launchedBy":"user"}\n'
    )
    (ledger / "bad.json").write_text("{")

    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(tmp_path / "ledger"))
    assert pids_guard._has_user_owner(str(cgroup)) is True


def test_owner_veto_reads_ledger_from_env_override(tmp_path, monkeypatch):
    cgroup = tmp_path / "cg"
    cgroup.mkdir()
    (cgroup / "cgroup.procs").write_text("111\n")

    ledger_root = tmp_path / "sessions-root"
    ledger = ledger_root / "sessions"
    ledger.mkdir(parents=True)
    (ledger / "run.json").write_text(
        '{"schemaVersion":1,"ledgerId":"run","pid":111,"runtime":"codex",'
        '"title":"x","startedAt":"2026-08-01T00:00:00Z","finishedAt":null,"launchedBy":"user"}'
    )

    monkeypatch.setenv("PIDS_STALL_LEDGER_DIR", str(ledger_root))
    assert pids_guard._has_user_owner(str(cgroup)) is True