eval_gate_hook.py

"""Git-hook regression gate — `soup eval gate install` (v0.55.0 Part D).

Generates a portable pre-push hook that runs `soup eval against` against
a baseline run id and blocks the push if any of:

* task accuracy
* refusal rate
* format validity
* p95 latency

regress past the configured thresholds. Threshold checks use a
paired-bootstrap CI so single-outlier rows do not flip the gate.

Public surface
--------------
- Frozen dataclass: ``GateThresholds``, ``RegressionVerdict``.
- Pure functions: ``render_pre_push_hook``, ``write_pre_push_hook``,
  ``paired_bootstrap_ci``, ``decide_regression``.
"""

from __future__ import annotations

import math
import os
import random
import re
import shlex
import stat
import tempfile
import types
from collections.abc import Mapping, Sequence
from dataclasses import dataclass

from soup_cli.utils.paths import is_under_cwd

_MAX_FILE_BYTES = 64 * 1024  # hooks are tiny — 64 KiB plenty
_MIN_BOOTSTRAP_SAMPLES = 100
_MAX_BOOTSTRAP_SAMPLES = 100_000
_DEFAULT_BOOTSTRAP_SAMPLES = 1000
_DEFAULT_CI_LEVEL = 0.95

# Regex on the run id — alphanumeric + ``-_`` only; mirrors v0.26.0
# registry name policy.
_RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_\-]{0,127}$")


@dataclass(frozen=True)
class GateThresholds:
    """Per-metric regression tolerance for the pre-push gate.

    Each tolerance is the *minimum acceptable* delta vs the baseline.
    Negative means a regression of that magnitude is still acceptable.
    Positive thresholds tighten the gate (require an improvement).
    """

    task_accuracy: float = -0.02
    refusal_rate: float = -0.05
    format_validity: float = -0.02
    p95_latency_ms: float = 100.0  # latency: lower-is-better — see semantics

    def __post_init__(self) -> None:
        # Every threshold must be a finite, real number. Pydantic-style
        # bool rejection (bool is a subclass of int — Python policy).
        for fld in (
            "task_accuracy", "refusal_rate", "format_validity",
            "p95_latency_ms",
        ):
            value = getattr(self, fld)
            if isinstance(value, bool):
                raise TypeError(f"{fld} must be float, got bool")
            if not isinstance(value, (int, float)):
                raise TypeError(
                    f"{fld} must be a number, got {type(value).__name__}"
                )
            if not math.isfinite(float(value)):
                raise ValueError(f"{fld} must be finite")


@dataclass(frozen=True)
class RegressionVerdict:
    """Output of :func:`decide_regression`.

    ``regressed`` is True iff any metric breached its tolerance after
    factoring in the paired-bootstrap CI. ``offenders`` names every
    metric that breached.
    """

    regressed: bool
    offenders: tuple[str, ...]
    ci_lower: float
    ci_upper: float
    delta_mean: float


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------

def _require_finite(value: object, *, field_name: str) -> float:
    if isinstance(value, bool):
        raise TypeError(f"{field_name} must be float, got bool")
    if not isinstance(value, (int, float)):
        raise TypeError(
            f"{field_name} must be a number, got {type(value).__name__}"
        )
    f = float(value)
    if not math.isfinite(f):
        raise ValueError(f"{field_name} must be finite")
    return f


def _validate_run_id(value: object) -> str:
    if isinstance(value, bool):
        raise TypeError("run_id must be str, got bool")
    if not isinstance(value, str):
        raise TypeError(
            f"run_id must be str, got {type(value).__name__}"
        )
    if not _RUN_ID_RE.match(value):
        raise ValueError(
            "run_id must be alphanumeric + '_-' (1-128 chars)"
        )
    return value


def _validate_thresholds(value: object) -> GateThresholds:
    if isinstance(value, GateThresholds):
        return value
    raise TypeError("thresholds must be a GateThresholds")


def _validate_bootstrap_samples(value: object) -> int:
    if isinstance(value, bool):
        raise TypeError("n_samples must be int, got bool")
    if not isinstance(value, int):
        raise TypeError(
            f"n_samples must be int, got {type(value).__name__}"
        )
    if value < _MIN_BOOTSTRAP_SAMPLES or value > _MAX_BOOTSTRAP_SAMPLES:
        raise ValueError(
            f"n_samples must be in "
            f"[{_MIN_BOOTSTRAP_SAMPLES}, {_MAX_BOOTSTRAP_SAMPLES}]"
        )
    return value


def _validate_ci_level(value: object) -> float:
    f = _require_finite(value, field_name="ci_level")
    if f <= 0.0 or f >= 1.0:
        raise ValueError("ci_level must be in (0.0, 1.0)")
    return f


# ---------------------------------------------------------------------------
# Paired bootstrap
# ---------------------------------------------------------------------------

def paired_bootstrap_ci(
    baseline: Sequence[float],
    candidate: Sequence[float],
    *,
    n_samples: int = _DEFAULT_BOOTSTRAP_SAMPLES,
    ci_level: float = _DEFAULT_CI_LEVEL,
    seed: int = 0,
) -> tuple[float, float, float]:
    """Paired-bootstrap (lower, upper, mean) of ``candidate - baseline``.

    Standard paired-sample bootstrap with replacement at the row level
    (preserves correlation between baseline/candidate). Deterministic
    given ``seed``.
    """
    if isinstance(baseline, (str, bytes)) or not isinstance(baseline, Sequence):
        raise TypeError("baseline must be a sequence of floats")
    if isinstance(candidate, (str, bytes)) or not isinstance(candidate, Sequence):
        raise TypeError("candidate must be a sequence of floats")
    if len(baseline) != len(candidate):
        raise ValueError(
            f"baseline ({len(baseline)}) and candidate "
            f"({len(candidate)}) lengths must match"
        )
    if len(baseline) == 0:
        raise ValueError("baseline must be non-empty")
    base_floats = [
        _require_finite(v, field_name="baseline[i]") for v in baseline
    ]
    cand_floats = [
        _require_finite(v, field_name="candidate[i]") for v in candidate
    ]
    n_samples = _validate_bootstrap_samples(n_samples)
    ci_level = _validate_ci_level(ci_level)
    if isinstance(seed, bool) or not isinstance(seed, int):
        raise TypeError("seed must be int")
    if seed < 0 or seed > 2**31 - 1:
        raise ValueError("seed must be non-negative int < 2**31")

    rng = random.Random(seed)
    n = len(base_floats)
    deltas = [c - b for b, c in zip(base_floats, cand_floats)]
    mean_delta = sum(deltas) / n
    means: list = []
    for _ in range(n_samples):
        sample_sum = 0.0
        for _ in range(n):
            idx = rng.randrange(n)
            sample_sum += deltas[idx]
        means.append(sample_sum / n)
    means.sort()
    alpha = (1.0 - ci_level) / 2.0
    lo_idx = max(0, int(alpha * n_samples))
    hi_idx = min(n_samples - 1, int((1.0 - alpha) * n_samples))
    return means[lo_idx], means[hi_idx], mean_delta


# ---------------------------------------------------------------------------
# Regression decision
# ---------------------------------------------------------------------------

# Mapping: metric → (tolerance attr name, direction).
# direction = +1 means "higher is better" (regression when ci_upper < tol)
# direction = -1 means "lower is better"  (regression when ci_lower > tol)
_METRIC_DIRECTION: Mapping[str, int] = types.MappingProxyType(
    {
        "task_accuracy": +1,
        "refusal_rate": +1,
        "format_validity": +1,
        "p95_latency_ms": -1,
    }
)


def decide_regression(
    metric: str,
    baseline: Sequence[float],
    candidate: Sequence[float],
    thresholds: GateThresholds,
    *,
    n_samples: int = _DEFAULT_BOOTSTRAP_SAMPLES,
    seed: int = 0,
) -> RegressionVerdict:
    """Decide whether ``metric`` regressed past the configured tolerance.

    Uses the paired-bootstrap 95 % CI of the delta. Higher-is-better
    metrics regress when the *upper* CI bound is still worse than the
    tolerance; lower-is-better metrics regress when the *lower* CI
    bound is still worse.
    """
    if isinstance(metric, bool) or not isinstance(metric, str):
        raise TypeError("metric must be str")
    if metric not in _METRIC_DIRECTION:
        raise ValueError(
            f"unknown metric {metric!r}; allowed: "
            + ", ".join(sorted(_METRIC_DIRECTION))
        )
    _validate_thresholds(thresholds)
    tol = getattr(thresholds, metric)
    direction = _METRIC_DIRECTION[metric]
    lo, hi, mean = paired_bootstrap_ci(
        baseline, candidate, n_samples=n_samples, seed=seed
    )
    regressed = False
    if direction > 0:
        # higher-is-better metric: regression iff the upper CI bound is
        # *still* below the tolerance (i.e. even the optimistic estimate
        # is bad).
        regressed = hi < tol
    else:
        # lower-is-better metric: regression iff the lower CI bound is
        # still above the tolerance (i.e. even the pessimistic estimate
        # is bad).
        regressed = lo > tol
    return RegressionVerdict(
        regressed=regressed,
        offenders=(metric,) if regressed else (),
        ci_lower=lo,
        ci_upper=hi,
        delta_mean=mean,
    )


# ---------------------------------------------------------------------------
# Hook script rendering
# ---------------------------------------------------------------------------

_HOOK_TEMPLATE = """#!/usr/bin/env bash
# Generated by `soup eval gate-install` (v0.55.0) — do not edit by hand.
# Pre-push regression gate: blocks the push when `soup eval against`
# detects a regression vs the baseline run id.
set -euo pipefail

BASELINE_RUN_ID={baseline_run_id}
GATE_SUITE={gate_suite}

CANDIDATE_RUN_ID="${{SOUP_CANDIDATE_RUN_ID:-}}"
if [ -z "$CANDIDATE_RUN_ID" ]; then
    echo "[soup] SOUP_CANDIDATE_RUN_ID not set; skipping regression gate." >&2
    exit 0
fi

soup eval against "$BASELINE_RUN_ID" --candidate "$CANDIDATE_RUN_ID" \\
    --suite "$GATE_SUITE" --json-only \\
    || {{
        echo "[soup] pre-push gate blocked: regression vs $BASELINE_RUN_ID" >&2
        exit 1
    }}

exit 0
"""


def _safe_shell_quote(value: str) -> str:
    """Wrapper around ``shlex.quote`` with a control-char rejection prelude.

    Project security policy mandates ``shlex.quote`` for shell-script
    generation. The control-char guard is defence-in-depth — validated
    callers already reject NUL / newline / tab, but the helper itself
    must remain safe to call on raw user-controlled strings.
    """
    if any(ord(ch) < 0x20 for ch in value):
        raise ValueError("value contains control characters")
    return shlex.quote(value)


def render_pre_push_hook(
    *,
    baseline_run_id: str,
    suite_path: str,
) -> str:
    """Render the pre-push hook script body — no I/O, deterministic."""
    rid = _validate_run_id(baseline_run_id)
    if isinstance(suite_path, bool) or not isinstance(suite_path, str):
        raise TypeError("suite_path must be str")
    if not suite_path:
        raise ValueError("suite_path must be non-empty")
    if "\x00" in suite_path:
        raise ValueError("suite_path must not contain NUL")
    if "\n" in suite_path or "\r" in suite_path:
        raise ValueError("suite_path must be a single line")
    if len(suite_path) > 4096:
        raise ValueError("suite_path exceeds 4096 characters")
    if not is_under_cwd(suite_path):
        raise ValueError("suite_path must stay under cwd")
    return _HOOK_TEMPLATE.format(
        baseline_run_id=_safe_shell_quote(rid),
        gate_suite=_safe_shell_quote(suite_path),
    )


def write_pre_push_hook(
    *,
    baseline_run_id: str,
    suite_path: str,
    hook_path: str = ".git/hooks/pre-push",
    overwrite: bool = False,
) -> str:
    """Write the rendered hook to ``hook_path`` with cwd + TOCTOU guards.

    Returns the path written. Refuses to overwrite an existing file
    unless ``overwrite=True``.
    """
    body = render_pre_push_hook(
        baseline_run_id=baseline_run_id, suite_path=suite_path
    )
    if isinstance(hook_path, bool) or not isinstance(hook_path, str):
        raise TypeError("hook_path must be str")
    if not hook_path:
        raise ValueError("hook_path must be non-empty")
    # Single explicit bool guard: bool is a subclass of int, so an
    # ``isinstance(..., bool)`` check is the only one that distinguishes
    # ``True``/``False`` from ``1``/``"yes"``/etc. (review fix —
    # eliminates the previous redundant double-branch).
    if not isinstance(overwrite, bool):
        raise TypeError("overwrite must be bool")
    # Cwd containment for the destination — operators may also pass
    # ``.git/hooks/pre-push`` so we go through the shared helper.
    if "\x00" in hook_path:
        raise ValueError("hook_path must not contain NUL")
    if not is_under_cwd(hook_path):
        raise ValueError("hook_path must stay under cwd")
    if os.path.lexists(hook_path):
        try:
            st = os.lstat(hook_path)
        except OSError as exc:
            raise ValueError(
                f"hook_path unreadable: {type(exc).__name__}"
            ) from exc
        if stat.S_ISLNK(st.st_mode):
            raise ValueError(
                "hook_path must not be a symlink (TOCTOU defence)"
            )
        if not overwrite:
            raise ValueError(
                "hook already exists; pass overwrite=True to replace it"
            )
    if len(body.encode("utf-8")) > _MAX_FILE_BYTES:
        raise ValueError("rendered hook exceeds 64 KiB cap")
    parent = os.path.dirname(os.path.abspath(hook_path)) or "."
    os.makedirs(parent, exist_ok=True)
    # Atomic write — same idiom as the rest of v0.55.0.
    fd, tmp = tempfile.mkstemp(prefix=".soup-pre-push.", dir=parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(body)
        os.replace(tmp, hook_path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    # POSIX executable bit so git can launch the hook directly.
    if os.name == "posix":
        try:
            mode = os.stat(hook_path).st_mode
            os.chmod(hook_path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
        except OSError:
            pass
    return hook_path