A Validation-Model Decision Matrix in Python

#!/usr/bin/env python3
"""Select an escalating backup-validation model from an artifact profile.

The tool scores a JSON artifact profile against a data-encoded decision matrix
and prints the mandated validation model plus a machine-readable justification
that a DR drill orchestrator can dispatch on.

Validation models, cheapest to most expensive:
    manifest-checksum   re-hash artifacts against a signed manifest
    structural-scan     open the artifact and walk its internal page structure
    full-restore        restore into a sandbox and run smoke tests

Exit codes (consumed by the orchestrator):
    0  a model was selected and fits the RTO budget -> proceed with the drill
    1  the mandated model cannot complete within RTO -> escalate for a decision
    2  usage / configuration error -> abort
"""
import json
import sys
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Dict, List, Tuple


class Model(IntEnum):
    """Escalation order is the enum value; lower is cheaper."""

    MANIFEST_CHECKSUM = 1
    STRUCTURAL_SCAN = 2
    FULL_RESTORE = 3

    @property
    def label(self) -> str:
        return self.name.lower().replace("_", "-")


# --- Policy encoded as data, not control flow ---------------------------------

TIER_SCORE: Dict[str, int] = {"tier1": 50, "tier2": 30, "tier3": 10}

# (inclusive upper bound in minutes, points) evaluated in order, first match wins
RTO_BANDS: Tuple[Tuple[int, int], ...] = ((15, 25), (60, 15), (240, 8))
RPO_BANDS: Tuple[Tuple[int, int], ...] = ((5, 15), (60, 8), (1440, 3))

BLAST_POINTS_PER_SYSTEM = 3
BLAST_CAP = 21  # never let a wide fan-out alone dominate the tier signal

# Ordered high-to-low: the first row whose min_score <= score fires.
@dataclass(frozen=True)
class Band:
    rule_id: str
    min_score: int
    model: Model
    rationale: str


SCORE_BANDS: Tuple[Band, ...] = (
    Band("R1", 80, Model.FULL_RESTORE,
         "high criticality and a tight recovery envelope justify a live restore"),
    Band("R2", 45, Model.STRUCTURAL_SCAN,
         "moderate criticality warrants opening the artifact, not just hashing it"),
    Band("R3", 0, Model.MANIFEST_CHECKSUM,
         "low criticality is adequately covered by a signed-manifest re-hash"),
)

# GB per minute each model sustains on the validation fleet (capacity estimate).
THROUGHPUT_GB_PER_MIN: Dict[Model, float] = {
    Model.MANIFEST_CHECKSUM: 40.0,
    Model.STRUCTURAL_SCAN: 8.0,
    Model.FULL_RESTORE: 1.5,
}


@dataclass
class Profile:
    criticality: str
    rto_minutes: int
    rpo_minutes: int
    size_gb: float
    blast_radius: int

    @classmethod
    def from_dict(cls, raw: dict) -> "Profile":
        try:
            prof = cls(
                criticality=str(raw["criticality"]).lower(),
                rto_minutes=int(raw["rto_minutes"]),
                rpo_minutes=int(raw["rpo_minutes"]),
                size_gb=float(raw["size_gb"]),
                blast_radius=int(raw["blast_radius"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise ValueError(f"invalid profile: {exc}") from exc
        if prof.criticality not in TIER_SCORE:
            raise ValueError(f"unknown criticality tier {prof.criticality!r}")
        if prof.size_gb < 0 or prof.rto_minutes <= 0:
            raise ValueError("size_gb must be >= 0 and rto_minutes > 0")
        return prof


@dataclass
class Decision:
    model: Model
    score: int
    rule_id: str
    estimated_minutes: float
    fits_rto: bool
    reasons: List[str] = field(default_factory=list)


def _band_points(value: int, bands: Tuple[Tuple[int, int], ...]) -> int:
    for upper, points in bands:
        if value <= upper:
            return points
    return 0


def score_profile(prof: Profile) -> Tuple[int, List[str]]:
    """Convert a profile into an integer criticality score with a paper trail."""
    reasons: List[str] = []
    tier_pts = TIER_SCORE[prof.criticality]
    reasons.append(f"tier {prof.criticality} -> +{tier_pts}")

    rto_pts = _band_points(prof.rto_minutes, RTO_BANDS)
    reasons.append(f"RTO {prof.rto_minutes}m -> +{rto_pts}")

    rpo_pts = _band_points(prof.rpo_minutes, RPO_BANDS)
    reasons.append(f"RPO {prof.rpo_minutes}m -> +{rpo_pts}")

    blast_pts = min(prof.blast_radius * BLAST_POINTS_PER_SYSTEM, BLAST_CAP)
    reasons.append(f"blast radius {prof.blast_radius} -> +{blast_pts}")

    return tier_pts + rto_pts + rpo_pts + blast_pts, reasons


def select_band(score: int) -> Band:
    for band in SCORE_BANDS:
        if score >= band.min_score:
            return band
    return SCORE_BANDS[-1]  # R3 has min_score 0 and is the guaranteed fallback


def estimate_minutes(model: Model, size_gb: float) -> float:
    return size_gb / THROUGHPUT_GB_PER_MIN[model]


def decide(prof: Profile) -> Decision:
    """Score, select, then de-escalate non-critical artifacts that miss the RTO."""
    score, reasons = score_profile(prof)
    band = select_band(score)
    model = band.model
    reasons.append(f"score {score} matched {band.rule_id}: {band.rationale}")

    # Size makes an expensive model infeasible; de-escalate all but tier1, which
    # must keep its mandated depth and be escalated to a human instead.
    while True:
        minutes = estimate_minutes(model, prof.size_gb)
        if minutes <= prof.rto_minutes:
            return Decision(model, score, band.rule_id, round(minutes, 2), True, reasons)
        if prof.criticality != "tier1" and model > Model.MANIFEST_CHECKSUM:
            cheaper = Model(model - 1)
            reasons.append(
                f"{model.label} est {minutes:.1f}m exceeds RTO {prof.rto_minutes}m "
                f"-> de-escalate to {cheaper.label}")
            model = cheaper
            continue
        reasons.append(
            f"{model.label} est {minutes:.1f}m exceeds RTO {prof.rto_minutes}m "
            f"and cannot be de-escalated -> escalate")
        return Decision(model, score, band.rule_id, round(minutes, 2), False, reasons)


def main(argv: List[str]) -> int:
    if len(argv) != 2:
        print("usage: select_validation_model.py <profile.json>", file=sys.stderr)
        return 2
    try:
        raw = json.loads(open(argv[1], "r", encoding="utf-8").read())
        prof = Profile.from_dict(raw)
    except (OSError, json.JSONDecodeError, ValueError) as exc:
        print(f"config error: {exc}", file=sys.stderr)
        return 2

    decision = decide(prof)
    json.dump(
        {
            "model": decision.model.label,
            "score": decision.score,
            "rule_id": decision.rule_id,
            "estimated_minutes": decision.estimated_minutes,
            "fits_rto": decision.fits_rto,
            "justification": decision.reasons,
        },
        sys.stdout,
        indent=2,
    )
    sys.stdout.write("\n")
    return 0 if decision.fits_rto else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))

$ python3 select_validation_model.py profile.json
{
  "model": "full-restore",
  "score": 98,
  "rule_id": "R1",
  "estimated_minutes": 8.0,
  "fits_rto": true,
  "justification": [
    "tier tier1 -> +50",
    "RTO 30m -> +15",
    "RPO 5m -> +15",
    "blast radius 6 -> +18",
    "score 98 matched R1: high criticality and a tight recovery envelope justify a live restore"
  ]
}
$ echo $?
0