atlas-iac/scripts/ops/hermes_handoff_json_rules.py

#!/usr/bin/env python3
"""JSON-shaped evaluators for the handoff acceptance harness.

These rules cover the evidence that arrives as structured output — Kubernetes
objects, forge pull requests, provider health files, and the two append-only
evidence logs the platform keeps. They share the fail-closed contract of every
other rule: an absent path, an unparsable payload, or an empty result set is
``NOT_RUN``, never a quiet pass.
"""

from __future__ import annotations

import datetime as dt
import math
import re
from typing import Any

from hermes_handoff_exec import Outcome
from hermes_handoff_model import FAIL, NOT_RUN, PASS, CheckSpec
from hermes_handoff_rules import (
    Evaluation,
    dotted,
    evaluator,
    exact_equal,
    is_list,
    missing_steps,
    parsed_step,
    strict_json,
)

_ROUTE_PART_RE = re.compile(r"[a-z0-9][a-z0-9_.-]{0,63}\Z")
_FLUX_NAME_RE = re.compile(
    r"[a-z0-9]([-a-z0-9.]*[a-z0-9])?/[a-z0-9]([-a-z0-9.]*[a-z0-9])?\Z"
)


def _timestamp(value: Any) -> dt.datetime:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("timestamp is not a non-empty string")
    observed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
    if observed.tzinfo is None:
        raise ValueError("timestamp has no timezone")
    return observed.astimezone(dt.timezone.utc)


def _now(value: Any) -> dt.datetime:
    if not isinstance(value, dt.datetime) or value.tzinfo is None:
        raise ValueError("run clock is not timezone-aware")
    return value.astimezone(dt.timezone.utc)


@evaluator("json_field")
def evaluate_json_field(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Compare dotted JSON paths of one step's output against expected values."""
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    parsed = parsed_step(spec, outcomes)
    if isinstance(parsed, Evaluation):
        return parsed
    payload, key = parsed
    evidence: dict[str, Any] = {}
    mismatches: list[str] = []
    for path, expected in spec.expect["fields"].items():
        try:
            actual = dotted(payload, path)
        except (KeyError, ValueError, IndexError):
            return Evaluation(
                NOT_RUN, f"path {path} is absent from step {key}", evidence
            )
        evidence[path] = actual
        allowed = expected if isinstance(expected, list) else [expected]
        if not any(exact_equal(actual, item) for item in allowed):
            mismatches.append(f"{path}={actual!r} expected one of {allowed!r}")
    if mismatches:
        return Evaluation(FAIL, "; ".join(mismatches), evidence)
    return Evaluation(PASS, "observed fields match the release contract", evidence)


@evaluator("json_numeric")
def evaluate_json_numeric(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Compare numeric JSON paths against inclusive minimum/maximum bounds."""
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    parsed = parsed_step(spec, outcomes)
    if isinstance(parsed, Evaluation):
        return parsed
    payload, key = parsed
    evidence: dict[str, Any] = {}
    problems: list[str] = []
    for path, bounds in spec.expect["fields"].items():
        try:
            actual = dotted(payload, path)
        except (KeyError, ValueError, IndexError):
            return Evaluation(
                NOT_RUN, f"path {path} is absent from step {key}", evidence
            )
        if (
            not isinstance(actual, (int, float))
            or isinstance(actual, bool)
            or not math.isfinite(float(actual))
        ):
            return Evaluation(
                NOT_RUN, f"path {path} is not numeric in step {key}", evidence
            )
        evidence[path] = actual
        if "min" in bounds and actual < bounds["min"]:
            problems.append(f"{path}={actual} below minimum {bounds['min']}")
        if "max" in bounds and actual > bounds["max"]:
            problems.append(f"{path}={actual} above maximum {bounds['max']}")
    if problems:
        return Evaluation(FAIL, "; ".join(problems), evidence)
    return Evaluation(PASS, "numeric fields are within the expected bounds", evidence)


@evaluator("json_recent")
def evaluate_json_recent(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Pass when a timestamp field is fresh enough to be evidence about now.

    Stale provider or routing evidence is worse than none: it describes a state
    the release no longer has. A timestamp older than the allowed age is a FAIL,
    and one the harness cannot parse is NOT_RUN.
    """
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    parsed = parsed_step(spec, outcomes)
    if isinstance(parsed, Evaluation):
        return parsed
    payload, key = parsed
    try:
        now = _now(spec.expect["now"])
    except (TypeError, ValueError) as exc:
        return Evaluation(NOT_RUN, f"invalid run clock: {exc}")
    evidence: dict[str, Any] = {}
    stale: list[str] = []
    for path, max_age in spec.expect["fields"].items():
        try:
            raw = dotted(payload, path)
        except (KeyError, ValueError, IndexError):
            return Evaluation(
                NOT_RUN, f"path {path} is absent from step {key}", evidence
            )
        try:
            observed = _timestamp(raw)
        except (TypeError, ValueError):
            return Evaluation(
                NOT_RUN, f"path {path} is not a timezone-aware ISO timestamp", evidence
            )
        age = (now - observed).total_seconds()
        evidence[path] = {"age_seconds": int(age)}
        if age < 0:
            stale.append(f"{path} is {abs(int(age))}s in the future")
        elif age > max_age:
            stale.append(f"{path} is {int(age)}s old, limit {max_age}s")
    if stale:
        return Evaluation(FAIL, "; ".join(stale), evidence)
    return Evaluation(PASS, "observed timestamps are fresh", evidence)


@evaluator("all_of_field")
def evaluate_all_of_field(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Pass when every selected element of a JSON array matches expected fields."""
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    parsed = parsed_step(spec, outcomes)
    if isinstance(parsed, Evaluation):
        return parsed
    payload, key = parsed
    try:
        elements = dotted(payload, spec.expect.get("array", ""))
    except (KeyError, ValueError, IndexError):
        return Evaluation(NOT_RUN, f"array path is absent from step {key}")
    if not is_list(elements):
        return Evaluation(NOT_RUN, f"step {key} did not return an array")
    key_field = spec.expect.get("key_field", "")
    wanted = spec.expect.get("keys")
    exempt = set(spec.expect.get("exempt_keys", ()))
    seen: list[Any] = []
    problems: list[str] = []
    evidence: dict[str, Any] = {}
    for element in elements:
        identity = dotted(element, key_field) if key_field else None
        if wanted is not None and identity not in wanted:
            continue
        if identity in exempt:
            continue
        seen.append(identity)
        for path in spec.expect.get("non_empty", ()):
            try:
                value = dotted(element, path)
            except (KeyError, ValueError, IndexError):
                value = None
            if value in (None, "", [], {}):
                problems.append(f"{identity}: {path} is empty")
        for path, expected in spec.expect.get("fields", {}).items():
            try:
                actual = dotted(element, path)
            except (KeyError, ValueError, IndexError):
                actual = None
            allowed = expected if isinstance(expected, list) else [expected]
            if not any(exact_equal(actual, item) for item in allowed):
                problems.append(
                    f"{identity}: {path}={actual!r} expected one of {allowed!r}"
                )
    evidence["examined"] = seen[:32]
    if wanted is not None:
        absent = [item for item in wanted if item not in seen]
        if absent:
            return Evaluation(
                NOT_RUN, f"expected keys absent from step {key}: {absent}", evidence
            )
    elif not seen:
        return Evaluation(NOT_RUN, f"step {key} returned nothing to examine", evidence)
    if problems:
        return Evaluation(FAIL, "; ".join(problems[:8]), evidence)
    return Evaluation(
        PASS, f"{len(seen)} element(s) match the release contract", evidence
    )


@evaluator("json_record")
def evaluate_json_record(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Capture named JSON paths as evidence; pass when all of them resolve.

    Recording is a release requirement in its own right — the runbook has to
    name the exact heads it was verified against — so an unreachable source is a
    NOT_RUN rather than an empty section.
    """
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    parsed = parsed_step(spec, outcomes)
    if isinstance(parsed, Evaluation):
        return parsed
    payload, key = parsed
    evidence: dict[str, Any] = {}
    for label, path in spec.expect["record"].items():
        try:
            evidence[label] = dotted(payload, path)
        except (KeyError, ValueError, IndexError):
            return Evaluation(
                NOT_RUN, f"path {path} is absent from step {key}", evidence
            )
    return Evaluation(PASS, "state recorded", evidence)


@evaluator("routing_evidence")
def evaluate_routing_evidence(
    spec: CheckSpec, outcomes: dict[str, Outcome]
) -> Evaluation:
    """Summarise a Switchyard routing log tail into provider/effort coverage.

    Route identifiers are ``<lane>/<provider>/<family>/<effort>``, so provider,
    model family, effort tier, and lane all come from the same field. Fallback
    and failure evidence is a separate ``fallback_reason`` on the record that
    was retried. Every row must be complete strict JSON; a clipped or malformed
    line makes the evidence unusable rather than being silently discarded.
    """
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    key = spec.expect["step"]
    outcome = outcomes[key]
    if not outcome.ok:
        return Evaluation(NOT_RUN, f"step {key} failed (rc={outcome.returncode})")
    records: list[dict[str, Any]] = []
    routes: list[list[str]] = []
    stamps_by_dimension: dict[tuple[str, str], dt.datetime] = {}
    now_value = spec.expect.get("now")
    try:
        now = _now(now_value) if now_value is not None else None
    except (TypeError, ValueError) as exc:
        return Evaluation(NOT_RUN, f"invalid routing run clock: {exc}")
    for row_number, line in enumerate(outcome.stdout.splitlines(), 1):
        line = line.strip()
        if not line:
            continue
        try:
            record = strict_json(line)
        except ValueError as exc:
            return Evaluation(
                NOT_RUN, f"routing row {row_number} is malformed JSON: {exc}"
            )
        if not isinstance(record, dict):
            return Evaluation(NOT_RUN, f"routing row {row_number} is not an object")
        try:
            observed = _timestamp(record.get("ts"))
        except (TypeError, ValueError) as exc:
            return Evaluation(
                NOT_RUN, f"routing row {row_number} has invalid timestamp: {exc}"
            )
        if now is not None and observed > now:
            return Evaluation(FAIL, f"routing row {row_number} has a future timestamp")
        model = record.get("model")
        if not isinstance(model, str) or not model:
            return Evaluation(NOT_RUN, f"routing row {row_number} has no model")
        route = model.split("/")
        if len(route) == 4 and all(_ROUTE_PART_RE.fullmatch(part) for part in route):
            routes.append(route)
            for label, value in (
                ("provider", route[1]),
                ("effort", route[3]),
                ("lane", route[0]),
            ):
                key_dimension = (label, value)
                stamps_by_dimension[key_dimension] = max(
                    observed, stamps_by_dimension.get(key_dimension, observed)
                )
        elif record.get("tier") != "classifier":
            return Evaluation(
                NOT_RUN, f"routing row {row_number} has malformed route {model!r}"
            )
        fallback = record.get("fallback_reason")
        if fallback is not None and (
            not isinstance(fallback, str) or not fallback.strip()
        ):
            return Evaluation(
                NOT_RUN, f"routing row {row_number} has malformed fallback_reason"
            )
        record["_observed"] = observed
        records.append(record)
    fallbacks = [record for record in records if record.get("fallback_reason")]
    stamps = sorted(record["_observed"] for record in records)
    evidence: dict[str, Any] = {
        "records": len(records),
        "routed": len(routes),
        "lanes": sorted({route[0] for route in routes}),
        "providers": sorted({route[1] for route in routes}),
        "families": sorted({route[2] for route in routes}),
        "efforts": sorted({route[3] for route in routes}),
        "fallback_events": len(fallbacks),
        "fallback_reasons": sorted(
            {str(record["fallback_reason"]) for record in fallbacks}
        ),
        "latest_ts": stamps[-1].isoformat() if stamps else "",
    }
    if not records:
        return Evaluation(
            NOT_RUN, "routing log tail contained no parsable records", evidence
        )
    missing = [
        f"{label}:{item}"
        for label, wanted in (
            ("provider", spec.expect.get("providers", ())),
            ("effort", spec.expect.get("efforts", ())),
            ("lane", spec.expect.get("lanes", ())),
        )
        for item in wanted
        if item not in evidence[f"{label}s"]
    ]
    if spec.expect.get("require_fallback_evidence") and not fallbacks:
        missing.append("fallback:none-observed")
    max_age = spec.expect.get("max_age_seconds")
    if max_age and now:
        ages: dict[str, int] = {}
        for label, wanted in (
            ("provider", spec.expect.get("providers", ())),
            ("effort", spec.expect.get("efforts", ())),
            ("lane", spec.expect.get("lanes", ())),
        ):
            for item in wanted:
                stamp = stamps_by_dimension.get((label, item))
                if stamp is None:
                    continue
                age = int((now - stamp).total_seconds())
                ages[f"{label}:{item}"] = age
                if age > max_age:
                    missing.append(
                        f"freshness:{label}:{item}:{age}s-exceeds-{max_age}s"
                    )
        evidence["age_seconds"] = ages
    if missing:
        return Evaluation(
            FAIL, f"routing evidence missing {', '.join(missing)}", evidence
        )
    return Evaluation(PASS, "routing evidence covers the required lanes", evidence)


@evaluator("flux_health")
def evaluate_flux_health(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation:
    """Classify a compact Flux projection into suspensions and unhealthy objects.

    Each row carries name, generation, observed generation, suspension, and all
    condition states. A non-suspended object must be current, explicitly
    ``Ready=True``, not ``Healthy=False/Unknown``, and not reconciling. Suspended
    objects are judged against the exact expected-suspension list.
    """
    blocked = missing_steps(spec, outcomes)
    if blocked:
        return blocked
    key = spec.expect["step"]
    outcome = outcomes[key]
    if not outcome.ok:
        return Evaluation(NOT_RUN, f"step {key} failed (rc={outcome.returncode})")
    expected = set(spec.expect.get("expected_suspensions", ()))
    suspended: list[str] = []
    unhealthy: list[str] = []
    total = 0
    for row_number, line in enumerate(outcome.stdout.splitlines(), 1):
        if not line.strip():
            continue
        columns = line.split("\t")
        if len(columns) != 5:
            return Evaluation(
                NOT_RUN, f"Flux row {row_number} has {len(columns)} columns"
            )
        name, generation, observed_generation, suspend, conditions = columns
        name = name.strip()
        if not _FLUX_NAME_RE.fullmatch(name):
            return Evaluation(
                NOT_RUN, f"Flux row {row_number} has malformed object name"
            )
        try:
            current_generation = int(generation)
            reconciled_generation = int(observed_generation)
        except ValueError:
            return Evaluation(
                NOT_RUN, f"Flux row {row_number} has malformed generations"
            )
        if current_generation <= 0 or current_generation != reconciled_generation:
            return Evaluation(
                FAIL, f"Flux row {row_number} has stale observedGeneration"
            )
        if suspend.strip().lower() not in {"", "false", "true"}:
            return Evaluation(
                NOT_RUN, f"Flux row {row_number} has malformed suspension state"
            )
        total += 1
        if suspend.strip().lower() == "true":
            suspended.append(name)
            continue
        states: dict[str, str] = {}
        for pair in filter(None, conditions.strip().strip(",").split(",")):
            if pair.count("=") != 1:
                return Evaluation(
                    NOT_RUN, f"Flux row {row_number} has malformed condition"
                )
            condition, state = pair.split("=", 1)
            if (
                not condition
                or state not in {"True", "False", "Unknown"}
                or condition in states
            ):
                return Evaluation(
                    NOT_RUN, f"Flux row {row_number} has malformed condition state"
                )
            states[condition] = state
        healthy = states.get("Healthy")
        if (
            states.get("Ready") != "True"
            or healthy in {"False", "Unknown"}
            or states.get("Reconciling") == "True"
        ):
            unhealthy.append(name)
    unexpected = sorted(set(suspended) - expected)
    evidence = {
        "total": total,
        "suspended": sorted(suspended),
        "unexpected_suspensions": unexpected,
        "unhealthy": sorted(unhealthy),
    }
    if not total:
        return Evaluation(NOT_RUN, f"step {key} returned no Flux objects", evidence)
    problems = []
    if unexpected:
        problems.append(f"unexpected suspensions: {', '.join(unexpected)}")
    if unhealthy:
        problems.append(f"unhealthy: {', '.join(sorted(unhealthy)[:8])}")
    if problems:
        return Evaluation(FAIL, "; ".join(problems), evidence)
    return Evaluation(PASS, f"{total} Flux objects reconciled as expected", evidence)