atlas-iac/scripts/ops/hermes_handoff_rules.py

#!/usr/bin/env python3
"""Shared primitives and dispatch for handoff acceptance evaluators.

Evaluators are registered by name so the check catalog stays declarative: an
entry says *what* it asserts, not how to plumb it. The dispatch here is the one
place that turns an unexpected exception into ``NOT_RUN`` — a harness bug must
never be indistinguishable from a passing release.
"""

from __future__ import annotations

import json
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any

from hermes_handoff_exec import Outcome
from hermes_handoff_model import NOT_RUN, STATUSES, CheckSpec

DENIAL_MARKERS = (
    "forbidden",
    "is not allowed",
    "unauthorized",
    "permission denied",
    "cannot get",
    "cannot list",
    "cannot create",
    "cannot delete",
    "cannot patch",
    "not permitted",
    "gitea api returned http 403",
)


@dataclass
class Evaluation:
    """A classification plus the small facts that justify it."""

    status: str
    reason: str
    evidence: dict[str, Any] = field(default_factory=dict)


Evaluator = Callable[[CheckSpec, dict[str, Outcome]], Evaluation]
EVALUATORS: dict[str, Evaluator] = {}


def evaluator(name: str) -> Callable[[Evaluator], Evaluator]:
    """Register an evaluator under the rule name catalog entries use."""

    def register(function: Evaluator) -> Evaluator:
        EVALUATORS[name] = function
        return function

    return register


def missing_steps(
    spec: CheckSpec, outcomes: Mapping[str, Outcome]
) -> Evaluation | None:
    """Return NOT_RUN when a required step never produced usable evidence."""
    for step in spec.steps:
        if step.optional:
            continue
        outcome = outcomes.get(step.key)
        if outcome is None:
            return Evaluation(NOT_RUN, f"step {step.key} was never executed")
        if outcome.truncated:
            return Evaluation(NOT_RUN, f"step {step.key} output was truncated")
        if not outcome.ran:
            return Evaluation(NOT_RUN, f"step {step.key} did not run: {outcome.error}")
    return None


def looks_denied(outcome: Outcome) -> bool:
    """Report whether a real request was refused by policy rather than failing."""
    if outcome.ok:
        return False
    haystack = outcome.combined.lower()
    return any(marker in haystack for marker in DENIAL_MARKERS)


def step_keys(spec: CheckSpec, kind: str) -> tuple[str, ...]:
    """Return the keys of every step of one kind."""
    return tuple(step.key for step in spec.steps if step.kind == kind)


def is_list(value: Any) -> bool:
    """Report whether a value is a JSON array rather than a string."""
    return isinstance(value, Sequence) and not isinstance(value, (str, bytes))


def exact_equal(actual: Any, expected: Any) -> bool:
    """Compare JSON values without Python's bool/int type confusion."""
    return type(actual) is type(expected) and actual == expected


def dotted(payload: Any, path: str) -> Any:
    """Return a value addressed by a dotted path.

    ``a.b[0].c`` indexes; ``a[].b`` maps the remainder of the path over a list,
    which is how the catalog records one field from every pull request or every
    Flux object without a bespoke parser per check.
    """
    current = payload
    segments = [segment for segment in path.split(".") if segment]
    for position, raw in enumerate(segments):
        name, bracket, indexes = raw.partition("[")
        if name:
            if not isinstance(current, Mapping) or name not in current:
                raise KeyError(path)
            current = current[name]
        if not bracket:
            continue
        for chunk in indexes.rstrip("]").split("]["):
            if not is_list(current):
                raise KeyError(path)
            if chunk == "":
                remainder = ".".join(segments[position + 1 :])
                return [
                    dotted(item, remainder) if remainder else item for item in current
                ]
            offset = int(chunk)
            if offset >= len(current):
                raise KeyError(path)
            current = current[offset]
    return current


def strict_json(text: str) -> Any:
    """Decode strict JSON, rejecting constants and duplicate object keys."""
    text = text.strip()
    if not text:
        raise ValueError("empty output")

    def reject_constant(value: str) -> None:
        raise ValueError(f"non-finite JSON constant {value}")

    def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
        result: dict[str, Any] = {}
        for key, value in pairs:
            if key in result:
                raise ValueError(f"duplicate JSON key {key!r}")
            result[key] = value
        return result

    return json.loads(
        text,
        strict=True,
        parse_constant=reject_constant,
        object_pairs_hook=unique_object,
    )


def parse_json(outcome: Outcome) -> Any:
    """Parse one outcome as strict JSON."""
    return strict_json(outcome.stdout)


def lines(outcome: Outcome) -> list[str]:
    """Return non-empty, stripped stdout lines."""
    return [line.strip() for line in outcome.stdout.splitlines() if line.strip()]


def parsed_step(
    spec: CheckSpec, outcomes: dict[str, Outcome]
) -> tuple[Any, str] | Evaluation:
    """Return the parsed payload of ``expect['step']`` or the NOT_RUN to report."""
    key = spec.expect["step"]
    outcome = outcomes[key]
    if not outcome.ok:
        return Evaluation(NOT_RUN, f"step {key} failed (rc={outcome.returncode})")
    try:
        return (parse_json(outcome), key)
    except ValueError as exc:
        return Evaluation(NOT_RUN, f"step {key} did not return JSON: {exc}")


def evaluate(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Dispatch to the named evaluator, converting any surprise into NOT_RUN."""
    rule = EVALUATORS.get(spec.rule)
    if rule is None:
        return Evaluation(NOT_RUN, f"unknown evaluator: {spec.rule}")
    truncated = sorted(key for key, outcome in outcomes.items() if outcome.truncated)
    if truncated:
        return Evaluation(
            NOT_RUN, f"truncated evidence from step(s): {', '.join(truncated)}"
        )
    try:
        result = rule(spec, outcomes)
    except Exception as exc:  # noqa: BLE001 - a harness bug must never read as PASS
        return Evaluation(
            NOT_RUN, f"evaluator {spec.rule} raised {type(exc).__name__}: {exc}"
        )
    if result.status not in STATUSES:
        return Evaluation(
            NOT_RUN, f"evaluator {spec.rule} returned unknown status {result.status!r}"
        )
    return result