"""Shared stall detectors for process trees and cgroup sampling.

`stall-guard` keeps the existing process-tree detector, while `pids-guard`
uses the same accumulator logic for cgroup counters.
"""

import os
import time


def accumulate(samples, keys):
    """Per-signal cumulative sum of positive deltas only.

    Raw counters can decrease when tasks exit or logs rotate. Only increases are
    treated as progress.
    """
    cumulative = {key: [0.0] * len(samples) for key in keys}
    for i in range(1, len(samples)):
        for key in keys:
            rise = samples[i][key] - samples[i - 1][key]
            cumulative[key][i] = cumulative[key][i - 1] + (rise if rise > 0 else 0.0)
    return cumulative


def _read_counter(path):
    try:
        with open(path) as handle:
            return int(handle.read().strip())
    except (OSError, ValueError):
        return None


def _read_cpu_usecs(path):
    try:
        with open(path) as handle:
            for line in handle:
                if line.startswith("usage_usec"):
                    return int(line.split()[1])
    except (OSError, ValueError):
        return None
    return None


SIGNALS = ("cpu", "io", "pids")


def _read_proc_io(raw_pids):
    """Sum rchar+wchar over the cgroup's tasks, or None when nothing was readable.

    /proc/<pid>/io is ptrace-gated: a bwrap-confined agent tree denies it to its
    own uid. A denied read is not evidence of zero IO, so a cgroup whose tasks
    are all unreadable reports the signal as absent rather than as a flat zero.
    """
    total = 0
    tasks = 0
    readable = 0
    for raw_pid in raw_pids:
        try:
            pid = int(raw_pid.strip())
        except ValueError:
            continue
        if pid <= 0:
            continue
        tasks += 1
        try:
            with open(f"/proc/{pid}/io") as handle:
                for line in handle:
                    name, _, value = line.partition(":")
                    if name == "rchar" or name == "wchar":
                        total += int(value)
            readable += 1
        except (OSError, ValueError):
            continue
    if tasks and not readable:
        return None
    return total


def unavailable_signals(samples):
    """Signals that failed to produce a real observation in at least one sample."""
    return tuple(
        key for key in SIGNALS if any(sample.get(key) is None for sample in samples)
    )


def sample_cgroup(path):
    """Read cgroup counters for a single cgroup directory.

    Returns None when required counters cannot be read. `io` is None when the
    kernel refused every per-task read — see _read_proc_io.
    """
    procs_path = os.path.join(path, "cgroup.procs")
    try:
        with open(procs_path) as handle:
            raw_pids = handle.read().splitlines()
    except OSError:
        return None

    cpu = _read_cpu_usecs(os.path.join(path, "cpu.stat"))
    if cpu is None:
        return None

    pids = _read_counter(os.path.join(path, "pids.current"))
    if pids is None:
        return None

    mem = _read_counter(os.path.join(path, "memory.current"))
    if mem is None:
        return None

    io = _read_proc_io(raw_pids)

    return {
        "t": time.time(),
        "cpu": cpu,
        "io": io,
        "pids": pids,
        "mem": mem,
    }


def classify_cgroup(samples, idle_window_s):
    """Return a reason string when all three counters are flat for the window.

    Refuses to answer while any signal is absent: an unobserved counter is
    constant, and a constant counter can never dissent from the others.
    """
    if len(samples) < 2:
        return None
    if unavailable_signals(samples):
        return None
    cumulative = accumulate(samples, SIGNALS)
    last = samples[-1]

    def anchor_index(window_s):
        window_start = last["t"] - window_s
        found = None
        for i, sample in enumerate(samples):
            if sample["t"] > window_start:
                break
            found = i
        return found

    index = anchor_index(idle_window_s)
    if index is None:
        return None

    if all(cumulative[key][index] == cumulative[key][-1] for key in SIGNALS):
        idle_for = last["t"] - samples[index]["t"]
        return f"no progress on cpu+io+pids for {idle_for:.0f}s"
    return None