tegwick
711c4e1149
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
84 lines
3.7 KiB
Python
84 lines
3.7 KiB
Python
"""Runbook maturity: earned from run history, independent of any one environment, bound to a pack revision.
|
|
|
|
draft valid, no review and no clean rehearsal at the current revision
|
|
reviewed a review is recorded for the current revision
|
|
rehearsed at least one clean run in a rehearsal environment at the current revision
|
|
proven reviewed, rehearsed, and clean in at least two environments by at least two operators
|
|
stale nothing holds at the current revision, but a higher level was reached at an earlier one
|
|
|
|
A run is *clean* when it succeeded with no deviation (no override, waiver or push-through).
|
|
Runs without an environment context are not attributable and never count.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import yaml
|
|
|
|
from . import run as R
|
|
from .pack import REVIEW_FILE, Pack, digest
|
|
|
|
LEVELS = ("draft", "reviewed", "rehearsed", "proven")
|
|
|
|
|
|
class ReviewError(Exception):
|
|
pass
|
|
|
|
|
|
def _reviews(pack: Pack) -> list[dict]:
|
|
f = pack.path.parent / REVIEW_FILE
|
|
if not f.exists():
|
|
return []
|
|
return list((yaml.safe_load(f.read_text()) or {}).get("reviews", []))
|
|
|
|
|
|
def add_review(pack: Pack, by: str, note: str = "") -> dict:
|
|
"""Record that `by` reviewed the pack at its current revision."""
|
|
if not by.strip() or len(by.split()) != 1:
|
|
raise ReviewError("reviewer must be one word (name or handle)")
|
|
try:
|
|
R._clean(by, "reviewer"); R._clean(note, "review note")
|
|
except R.RunError as e:
|
|
raise ReviewError(str(e)) from e
|
|
entry = {"by": by.strip(), "at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"digest": digest(pack), "note": note.strip()}
|
|
revs = _reviews(pack) + [entry]
|
|
(pack.path.parent / REVIEW_FILE).write_text(yaml.safe_dump({"reviews": revs}, sort_keys=False))
|
|
return entry
|
|
|
|
|
|
def _env_key(ctx: dict) -> str:
|
|
return f"{ctx.get('project', '-')}/{ctx.get('target', '-')}"
|
|
|
|
|
|
def _evidence(pack: Pack, dig: str, runs: list[R.Run]) -> dict:
|
|
clean = [r for r in runs if r.data.get("pack_digest") == dig and r.data.get("outcome") == "succeeded"
|
|
and not r.data.get("deviations") and r.data.get("context", {}).get("environment")
|
|
and R.seal_state(r) != "broken"] # a record changed after closing is not evidence
|
|
reviewed = any(rv.get("digest") == dig for rv in _reviews(pack))
|
|
rehearsal = [r for r in clean if r.data["context"]["environment"] == "rehearsal"]
|
|
envs = {_env_key(r.data["context"]) for r in clean}
|
|
ops = {r.data["operator"] for r in clean}
|
|
proven = reviewed and bool(rehearsal) and len(envs) >= 2 and len(ops) >= 2
|
|
level = "proven" if proven else "rehearsed" if rehearsal else "reviewed" if reviewed else "draft"
|
|
return {"level": level, "reviewed": reviewed, "clean_runs": len(clean), "rehearsal_runs": len(rehearsal),
|
|
"environments": sorted(envs), "operators": sorted(ops)}
|
|
|
|
|
|
def maturity(pack: Pack) -> dict:
|
|
"""{'level', 'stale_from', 'digest', 'evidence'} for the pack's current revision."""
|
|
runs = [R.load(pack, rid) for rid in R.list_runs(pack)]
|
|
dig = digest(pack)
|
|
now = _evidence(pack, dig, runs)
|
|
stale_from = None
|
|
if now["level"] == "draft":
|
|
# a higher level reached at an earlier revision means the maturity has expired
|
|
older = {r.data.get("pack_digest") for r in runs} | {rv.get("digest") for rv in _reviews(pack)}
|
|
best = "draft"
|
|
for d in older - {dig, None}:
|
|
lv = _evidence(pack, d, runs)["level"]
|
|
if LEVELS.index(lv) > LEVELS.index(best):
|
|
best = lv
|
|
if best != "draft":
|
|
stale_from = best
|
|
return {"level": "stale" if stale_from else now["level"], "stale_from": stale_from, "digest": dig, "evidence": now}
|