import importlib.util
import os
import signal
import subprocess
import sys
import time
from pathlib import Path

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

spec = importlib.util.spec_from_loader("stall_guard", importlib.machinery.SourceFileLoader("stall_guard", str(BIN)))
stall_guard = importlib.util.module_from_spec(spec)
spec.loader.exec_module(stall_guard)
classify = stall_guard.classify


def trace(points):
    """points: list of (t, out, cpu, io, beat)"""
    return [{"t": t, "out": o, "cpu": c, "io": i, "beat": b} for t, o, c, i, b in points]


# --- detection arm: genuinely stuck runs MUST fire ---

def test_all_signals_flat_fires():
    samples = trace([(t, 100, 50, 900, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is not None


def test_epoll_blocked_run_fires():
    # a process waiting forever on a socket: no output, no cpu, no io
    samples = trace([(0, 10, 5, 80, 0.0)] + [(t, 10, 5, 80, 0.0) for t in range(60, 1500, 60)])
    assert classify(samples, 900) is not None


# --- false-positive arm: legitimate work MUST stay quiet ---

def test_silent_compile_burning_cpu_is_quiet():
    # no output for 20 minutes, but cpu climbs — a long tsc, not a hang
    samples = trace([(t, 100, 50 + t, 900, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is None


def test_network_wait_moving_bytes_is_quiet():
    # rchar grows while nothing prints — a slow download
    samples = trace([(t, 100, 50, 900 + t * 10, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is None


def test_declared_heartbeat_keeps_a_queue_wait_quiet():
    # blocked on a build lock, but the waiter declares itself alive
    samples = trace([(t, 100, 50, 900, 1000.0 + t) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is None


def test_output_still_growing_is_quiet():
    samples = trace([(t, 100 + t, 50, 900, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is None


# --- guards against the detector itself being wrong ---

def test_short_history_never_fires():
    samples = trace([(0, 10, 5, 80, 0.0), (60, 10, 5, 80, 0.0)])
    assert classify(samples, 900) is None


def test_window_must_be_fully_covered():
    # only 120s of history: cannot conclude a 900s idle window
    samples = trace([(t, 10, 5, 80, 0.0) for t in range(0, 180, 60)])
    assert classify(samples, 900) is None


def test_unreadable_counters_do_not_fire_alone():
    # cpu and io unreadable for the whole run (all zero) must not read as "flat
    # and therefore stalled" — output still moving keeps it quiet
    samples = trace([(t, 100 + t, 0, 0, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900) is None


def test_recovery_after_flat_period_is_quiet():
    flat = [(t, 100, 50, 900, 0.0) for t in range(0, 600, 60)]
    moving = [(t, 100 + t, 50 + t, 900 + t, 0.0) for t in range(600, 1500, 60)]
    assert classify(trace(flat + moving), 900) is None


def test_decreasing_counter_cannot_suppress_detection():
    # observed in real cdx traces: summed cpu FALLS when a child exits. Comparing
    # raw endpoints reads that as "not flat" and silently disables the detector.
    samples = trace([(t, 100, 50, 900, 0.0) for t in range(0, 600, 60)])
    samples += trace([(t, 100, 40, 900, 0.0) for t in range(600, 1500, 60)])
    assert classify(samples, 900) is not None


def test_decrease_then_regrowth_is_progress():
    # a truncated-and-refilled log is real progress, not a stall
    samples = trace([(t, 100 + t, 50, 900, 0.0) for t in range(0, 600, 60)])
    samples += trace([(t, t - 590, 50, 900, 0.0) for t in range(600, 1500, 60)])
    assert classify(samples, 900) is None


# --- ceiling arm: an ACTIVE hang that keeps one signal moving must still fire ---

def test_spin_loop_fires_on_the_output_ceiling():
    # cpu climbs forever, nothing is ever printed: the AND rule alone never fires
    samples = trace([(t, 100, 50 + t, 900 + t, 0.0) for t in range(0, 5000, 60)])
    assert classify(samples, 900) is None
    assert "ceiling" in classify(samples, 900, 3600)


def test_ceiling_stays_quiet_while_output_moves():
    samples = trace([(t, 100 + t, 50 + t, 900 + t, 0.0) for t in range(0, 9000, 60)])
    assert classify(samples, 900, 3600) is None


def test_ceiling_needs_the_full_window_covered():
    samples = trace([(t, 100, 50 + t, 900 + t, 0.0) for t in range(0, 1200, 60)])
    assert classify(samples, 900, 3600) is None


# --- end-to-end ---

def run_guard(argv, timeout=90):
    """Run stall-guard in its own session; a timeout kills the whole tree, never orphans it."""
    proc = subprocess.Popen(
        argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True
    )
    try:
        out, err = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        os.killpg(proc.pid, signal.SIGKILL)
        out, err = proc.communicate()
        raise AssertionError(f"stall-guard did not exit within {timeout}s\n{err}")
    return subprocess.CompletedProcess(argv, proc.returncode, out, err)


def test_live_stalled_child_is_killed_loudly(tmp_path):
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest", "--idle-window", "4",
         "--sample-interval", "1", "--kill-grace", "1", "--enforce",
         "--trace", str(tmp_path / "t.jsonl"), "--log", str(tmp_path / "o.log"),
         "--", "sleep", "120"],
    )
    assert result.returncode == stall_guard.STALL_EXIT, result.stderr
    assert "STALL DETECTED" in result.stderr
    assert "FAILED LOUDLY" in result.stderr
    assert "desktop notification suppressed" in result.stderr, (
        "the gate did not reach the subprocess; this suite is popping the session"
    )


def test_live_working_child_is_not_killed(tmp_path):
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest-ok", "--idle-window", "4",
         "--sample-interval", "1", "--enforce",
         "--trace", str(tmp_path / "t2.jsonl"), "--log", str(tmp_path / "o2.log"),
         "--", "sh", "-c", "for i in $(seq 12); do echo tick $i; sleep 0.5; done"],
    )
    assert result.returncode == 0, result.stderr
    assert "STALL DETECTED" not in result.stderr


def test_live_spin_loop_is_killed_on_the_ceiling(tmp_path):
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest-spin", "--idle-window", "3600",
         "--output-ceiling", "5", "--sample-interval", "1", "--kill-grace", "1", "--enforce",
         "--trace", str(tmp_path / "t4.jsonl"), "--log", str(tmp_path / "o4.log"),
         "--", "timeout", "60", "sh", "-c", "while :; do :; done"],
    )
    assert result.returncode == stall_guard.STALL_EXIT, result.stderr
    assert "no output for" in result.stderr


def test_live_grandchild_in_its_own_session_is_killed(tmp_path):
    marker = tmp_path / "grandchild.alive"
    grandchild = tmp_path / "grandchild.py"
    grandchild.write_text(
        f"import pathlib, time\n"
        f"pathlib.Path({str(marker)!r}).touch()\n"
        f"time.sleep(120)\n"
    )
    parent = tmp_path / "parent.py"
    parent.write_text(
        f"import subprocess, sys, time\n"
        f"subprocess.Popen([sys.executable, {str(grandchild)!r}], start_new_session=True)\n"
        f"time.sleep(120)\n"
    )
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest-tree", "--idle-window", "4",
         "--sample-interval", "1", "--kill-grace", "1", "--enforce",
         "--trace", str(tmp_path / "t5.jsonl"), "--log", str(tmp_path / "o5.log"),
         "--", sys.executable, str(parent)],
    )
    assert result.returncode == stall_guard.STALL_EXIT, result.stderr
    assert marker.exists(), "grandchild never started; the test proves nothing"
    time.sleep(1)
    survivors = subprocess.run(
        ["pgrep", "-f", str(grandchild)], capture_output=True, text=True
    ).stdout.split()
    assert not survivors, f"grandchild survived the kill: {survivors}"


def test_wall_notice_warns_loudly_without_killing(tmp_path):
    # the chatty-poll class: output keeps moving, so no progress rule can fire.
    # The advisory must surface it while letting the run finish.
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest-slow", "--idle-window", "3600",
         "--wall-notice", "2", "--sample-interval", "1", "--enforce",
         "--trace", str(tmp_path / "t6.jsonl"), "--log", str(tmp_path / "o6.log"),
         "--", "sh", "-c", "for i in $(seq 8); do echo waiting for lock...; sleep 1; done"],
    )
    assert result.returncode == 0, result.stderr
    assert "SLOW:" in result.stderr
    assert "STALL DETECTED" not in result.stderr
    assert result.stderr.count("SLOW:") == 1, "advisory must fire once, not every sample"


def test_observe_mode_reports_without_killing(tmp_path):
    result = run_guard(
        [sys.executable, str(BIN), "run", "--key", "selftest-observe", "--idle-window", "4",
         "--sample-interval", "1",
         "--trace", str(tmp_path / "t3.jsonl"), "--log", str(tmp_path / "o3.log"),
         "--", "sleep", "8"],
    )
    assert result.returncode == 0, result.stderr
    assert "observe mode" in result.stderr
    assert "would have failed" in result.stderr


# --- the notification gate: silent unless a CLI entry point opted in ---

def _record_notify_send(monkeypatch):
    calls = []
    monkeypatch.setattr(stall_guard.subprocess, "run", lambda argv, **kw: calls.append(argv))
    return calls


def test_notify_reaches_the_desktop_under_the_gate(monkeypatch):
    calls = _record_notify_send(monkeypatch)
    monkeypatch.setenv("STALL_GUARD_NOTIFY", "1")
    stall_guard.notify("stuck", "no progress")
    assert calls and calls[0][0] == "notify-send"


def test_notify_is_silent_without_the_gate(monkeypatch):
    calls = _record_notify_send(monkeypatch)
    monkeypatch.delenv("STALL_GUARD_NOTIFY", raising=False)
    stall_guard.notify("stuck", "no progress")
    assert not calls


def test_notify_is_silent_under_an_explicit_zero(monkeypatch):
    calls = _record_notify_send(monkeypatch)
    monkeypatch.setenv("STALL_GUARD_NOTIFY", "0")
    stall_guard.notify("stuck", "no progress")
    assert not calls