475 lines
19 KiB
Python
475 lines
19 KiB
Python
#!/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)
|