"""stress-ng load generation for noisy-neighbor benchmarks.

stress-ng is a perturbation instrument, never a benchmark: it injects a known,

machine-independent load tier while a lane measures VCS commands under

contention, and it powers falsification CI (inject load, assert

``envwatch.environment_suspect`` fires).

Discipline (every command this module builds):

- ``--timeout`` is ALWAYS present — stress-ng's default run length is 24 hours.

- ``--temp-path`` is ALWAYS present and must live on the same filesystem as

  the benchmarked repos, otherwise the io tier is not actually contending.

- ``--seed`` is ALWAYS present so a tier is deterministic across runs.

- Load tiers are machine-independent shapes (half the cores at 50%%, two hdd

  workers), not absolute op counts.

- bogo-ops output is parsed ONLY to verify the load ran (``verify_ran``),

  never reported as a benchmark number.

stress-ng is an OPTIONAL external binary (the only non-stdlib dependency in

the harness, and it stays optional): when it is absent, ``LoadGenerator.start``

reports ``{"started": False, "reason": "stress_ng_missing"}`` and lanes emit

skip rows — a row may never claim load that was not applied (ADR-0002).

"""

from __future__ import annotations

import os

import re

import shutil

import subprocess

import tempfile

from pathlib import Path

from typing import Any

STRESS_NG = "stress-ng"

# Deterministic default for --seed; any fixed value works, it only has to be

# recorded and stable so two runs of the same tier perturb identically.

DEFAULT_SEED = 20260612

# Safety-net run cap. stop() is the normal end of a load window; the timeout

# only bounds orphaned generators (stress-ng's own default is 24 hours).

DEFAULT_TIMEOUT_S = 1800

# Machine-independent load tiers. "none" is the explicit no-load tier (rows

# still record the string "none" so the field is never ambiguous).

LOAD_TIERS: dict[str, dict[str, Any] | None] = {

    "none": None,

    "cpu50": {"cpu_load_pct": 50},

    "io": {"hdd_workers": 2, "hdd_bytes": "256M"},

    "mixed": {"cpu_load_pct": 50, "hdd_workers": 2, "hdd_bytes": "256M"},

}

def cpu_worker_count(cpu_count: int | None = None) -> int:

    """Half the cores, at least one — the machine-independent cpu tier width."""

    cores = cpu_count if cpu_count is not None else (os.cpu_count() or 1)

    return max(1, cores // 2)

def stress_ng_command(

    tier: str,

    *,

    timeout_s: int | float,

    temp_path: str | Path,

    seed: int = DEFAULT_SEED,

    cpu_count: int | None = None,

) -> list[str] | None:

    """argv for one load tier, or None for the "none" tier.

    Pure: no filesystem or process access, so the exact argument shape is

    unit-testable. argv[0] is the bare binary name; ``LoadGenerator`` resolves

    the real path (or reports the binary missing).

    """

    if tier not in LOAD_TIERS:

        raise ValueError(f"unknown load tier {tier!r}; known: {', '.join(sorted(LOAD_TIERS))}")

    spec = LOAD_TIERS[tier]

    if spec is None:

        return None

    if timeout_s <= 0:

        raise ValueError("timeout_s must be positive: stress-ng's default run length is 24h")

    command = [

        STRESS_NG,

        "--timeout",

        f"{int(timeout_s)}s",

        "--temp-path",

        str(temp_path),

        "--seed",

        str(int(seed)),

        "--metrics-brief",

    ]

    if "cpu_load_pct" in spec:

        command += ["--cpu", str(cpu_worker_count(cpu_count)), "--cpu-load", str(spec["cpu_load_pct"])]

    if "hdd_workers" in spec:

        command += ["--hdd", str(spec["hdd_workers"]), "--hdd-bytes", str(spec["hdd_bytes"])]

    return command

# A stress-ng metrics row: "stress-ng: metrc: [pid] cpu  20704  10.00 ..."

# (older releases emit the same row under "info:"). The stressor name is

# followed by integer bogo ops then the real-time column. Header lines

# ("stressor   bogo ops real time ...") and chatter ("dispatching hogs: 4 cpu")

# do not match because no integer directly follows the first word.

_BOGO_OPS_ROW = re.compile(

    r"^stress-ng:\s+(?:metrc|info):\s+\[\d+\]\s+([a-z][a-z0-9_-]*)\s+(\d+)\s+[0-9]+(?:\.[0-9]+)?\s",

    re.MULTILINE,

)

def verify_ran(stderr_text: str) -> bool:

    """Did the injected load actually run?

    True when at least one stressor reports positive bogo ops. The numbers are

    used ONLY as a did-it-run check; they are never recorded as a benchmark

    measurement.

    """

    return any(int(ops) > 0 for _name, ops in _BOGO_OPS_ROW.findall(stderr_text or ""))

def load_tier_unavailable_reason(detail: str) -> str:

    """Canonical skip reason for a tier the host cannot apply."""

    return f"load_tier_unavailable:{detail}"

def environment_boundary_fields(

    load_tier: str,

    before: dict[str, Any] | None,

    after: dict[str, Any] | None,

    *,

    cpu_count: int | None = None,

) -> dict[str, Any]:

    """Row fields for one measurement window's environment boundary.

    ``environment_suspect`` keeps the ADR-0002 distinction: None when the

    suspicion machinery did not run (a boundary sample is missing), [] when it

    ran and the window is clean, [reasons] when suspect. ``load_tier`` is

    always the explicit string ("none" included) so it is never ambiguous.

    """

    from oakbench.envwatch import environment_suspect

    suspect: list[str] | None = None

    if before is not None and after is not None:

        suspect = environment_suspect(before, after, cpu_count=cpu_count)

    return {

        "load_tier": load_tier,

        "environment_before": before,

        "environment_after": after,

        "environment_suspect": suspect,

    }

class LoadGenerator:

    """Owns one stress-ng subprocess for a measurement window.

    start() before the timed measurements, stop() after (both untimed).

    Honesty contract: start() never raises for an absent binary — it returns

    ``{"started": False, "reason": "stress_ng_missing"}`` so the lane can emit

    skip rows instead of claiming load that was not applied.

    """

    def __init__(

        self,

        tier: str,

        temp_path: str | Path,

        *,

        timeout_s: int | float = DEFAULT_TIMEOUT_S,

        seed: int = DEFAULT_SEED,

        cpu_count: int | None = None,

        stress_ng_path: str | None = None,

    ) -> None:

        if tier not in LOAD_TIERS:

            raise ValueError(f"unknown load tier {tier!r}; known: {', '.join(sorted(LOAD_TIERS))}")

        self.tier = tier

        self.temp_path = Path(temp_path)

        self.timeout_s = timeout_s

        self.seed = seed

        self.cpu_count = cpu_count

        self.stress_ng_path = stress_ng_path

        self._proc: subprocess.Popen[bytes] | None = None

        self._stderr_handle: Any = None

    def start(self) -> dict[str, Any]:

        command = stress_ng_command(

            self.tier,

            timeout_s=self.timeout_s,

            temp_path=self.temp_path,

            seed=self.seed,

            cpu_count=self.cpu_count,

        )

        if command is None:

            return {"started": False, "reason": "tier_none"}

        binary = self.stress_ng_path or shutil.which(STRESS_NG)

        if binary is None:

            return {"started": False, "reason": "stress_ng_missing"}

        self.temp_path.mkdir(parents=True, exist_ok=True)

        self._stderr_handle = tempfile.TemporaryFile()

        try:

            self._proc = subprocess.Popen(

                [binary, *command[1:]],

                cwd=self.temp_path,

                stdout=subprocess.DEVNULL,

                stderr=self._stderr_handle,

            )

        except OSError as error:

            self._close_stderr()

            return {"started": False, "reason": f"spawn_failed: {type(error).__name__}: {error}"}

        return {"started": True, "reason": None}

    def stop(self) -> dict[str, Any]:

        """Terminate the generator (untimed) and verify the load ran.

        ``verified`` is True only when stress-ng's own bogo-ops metrics confirm

        work happened; None when no process ever started (unmeasured, ADR-0002).

        """

        if self._proc is None:

            self._close_stderr()

            return {"stopped": False, "returncode": None, "verified": None, "stderr_tail": None}

        proc = self._proc

        self._proc = None

        proc.terminate()

        try:

            returncode = proc.wait(timeout=15)

        except subprocess.TimeoutExpired:

            proc.kill()

            returncode = proc.wait()

        stderr_text = self._read_stderr()

        self._close_stderr()

        return {

            "stopped": True,

            "returncode": returncode,

            "verified": verify_ran(stderr_text),

            "stderr_tail": stderr_text[-2000:],

        }

    def _read_stderr(self) -> str:

        if self._stderr_handle is None:

            return ""

        try:

            self._stderr_handle.flush()

            self._stderr_handle.seek(0)

            return self._stderr_handle.read().decode("utf-8", "replace")

        except (OSError, ValueError):

            return ""

    def _close_stderr(self) -> None:

        if self._stderr_handle is not None:

            try:

                self._stderr_handle.close()

            except OSError:

                pass

            self._stderr_handle = None

    def __enter__(self) -> "LoadGenerator":

        return self

    def __exit__(self, *exc_info: Any) -> None:

        self.stop()