solutions/10_capstone.py

"""Reference decision solution. Original synthetic data; no live actions."""
import hashlib
import json
import math
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "project"))
import opslab

REQUIRED = {"evidence_kind", "dataset", "route", "review", "calibration", "comparison",
            "monitor", "rollout", "maintenance", "drift", "manifest_changed",
            "rubric_review_complete", "follow_up_reviewed"}


def blocking_reasons(evidence: dict) -> list[str]:
    missing = sorted(REQUIRED - set(evidence))
    if missing:
        return [f"missing:{name}" for name in missing]
    incomplete = []
    for section in ("dataset", "route", "review", "comparison", "rollout"):
        key = "state" if section == "rollout" else "decision"
        if not isinstance(evidence[section], dict) or not isinstance(evidence[section].get(key), str):
            incomplete.append(section)
    calibration = evidence["calibration"]
    if not isinstance(calibration, dict) or not {"accept_precision", "false_acceptance_rate", "counts"} <= set(calibration):
        incomplete.append("calibration")
    elif not isinstance(calibration["counts"], dict) or type(calibration["counts"].get("FP")) is not int or calibration["counts"]["FP"] < 0:
        incomplete.append("calibration")
    else:
        for key in ("accept_precision", "false_acceptance_rate"):
            value = calibration[key]
            if value is not None and not (type(value) in (int, float) and 0 <= value <= 1 and math.isfinite(value)):
                incomplete.append("calibration")
    windows = evidence["monitor"]
    if not isinstance(windows, list) or any(not isinstance(row, dict) or
        any(type(row.get(key)) is not bool for key in ("operational_alert", "quality_review_required")) for row in windows):
        incomplete.append("monitor")
    changes = evidence["maintenance"]
    if not isinstance(changes, list) or any(not isinstance(row, dict) or not isinstance(row.get("timing"), str) or
        row["timing"] not in {"overdue", "due", "upcoming"} or not isinstance(row.get("kind"), str) for row in changes):
        incomplete.append("maintenance")
    drift = evidence["drift"]
    if not isinstance(drift, dict) or type(drift.get("reviewed")) is not bool or any(
        type(drift.get(key)) not in (int, float) or not 0 <= drift[key] <= 1 or not math.isfinite(drift[key])
        for key in ("distance", "threshold")):
        incomplete.append("drift")
    changed = evidence["manifest_changed"]
    if not isinstance(changed, list) or any(not isinstance(name, str) or not name for name in changed):
        incomplete.append("manifest_changed")
    if incomplete:
        return [f"incomplete:{name}" for name in sorted(set(incomplete))]
    reasons = []
    if evidence["evidence_kind"] != "synthetic_fixture_replay":
        reasons.append("unsupported_evidence_kind")
    for section in ("dataset", "route", "review", "comparison"):
        if evidence[section].get("decision") != "PASS":
            reasons.append(section)
    calibration = evidence["calibration"]
    if calibration.get("accept_precision") is None or calibration.get("false_acceptance_rate") is None:
        reasons.append("incomplete_calibration")
    elif calibration["counts"].get("FP") != 0:
        reasons.append("false_acceptances")
    if not evidence["monitor"]:
        reasons.append("missing_windows")
    if any(row["operational_alert"] for row in evidence["monitor"]):
        reasons.append("operational_alert")
    if any(row["quality_review_required"] for row in evidence["monitor"]):
        reasons.append("critical_quality_failure")
    if evidence["rollout"].get("state") != "COMPLETE_SIMULATION":
        reasons.append("rollout")
    if any(row["timing"] in {"overdue", "due"} and row["kind"] in {"policy", "provider"}
           for row in evidence["maintenance"]):
        reasons.append("unresolved_due_change")
    if evidence["drift"]["distance"] > evidence["drift"]["threshold"] and evidence["drift"].get("reviewed") is not True:
        reasons.append("unreviewed_mix_shift")
    if evidence["manifest_changed"]:
        reasons.append("manifest_mismatch")
    if evidence["rubric_review_complete"] is not True:
        reasons.append("rubric_review")
    if evidence["follow_up_reviewed"] is not True:
        reasons.append("incident_follow_up")
    return sorted(reasons)


def recommend(evidence: dict) -> dict:
    reasons = blocking_reasons(evidence)
    return {"decision": "BLOCK" if reasons else "PASS_SIMULATION", "blocking_reasons": reasons,
            "recovery_required": isinstance(evidence.get("rollout"), dict) and evidence["rollout"].get("state") == "RECOVER",
            "live_authorization": False}


def main() -> int:
    evidence = opslab.capstone_evidence(opslab.load())
    print(json.dumps({"recommendation": recommend(evidence), "evidence": evidence,
                      "decision_code_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest()},
                     indent=2, sort_keys=True, allow_nan=False))
    return 0  # Computation succeeded; inspect recommendation rather than exit code.


if __name__ == "__main__":
    raise SystemExit(main())