resource-control/tools/optimization.py

#!/usr/bin/env python3
"""Evaluate resource optimization cases.

The evaluator is fail-closed: any decision field that is unknown makes the
affected comparison unknown rather than optimistic. A case only reaches a
recommendation when every field the decision template requires is present for
the baseline and for the alternative being compared.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

CASE_TYPES = {
    "rightsizing", "consolidation", "commitment", "renewal",
    "migration", "retirement", "provider_switch",
}
DECISION_STATES = {
    "blocked_on_evidence", "proposed", "approved", "rejected", "superseded",
}
COST_FIELDS = (
    "recurring_infrastructure_eur_month",
    "recurring_internal_labor_eur_month",
    "recurring_external_labor_eur_month",
)
# Fields the decision template requires before any recommendation is made.
REQUIRED_FOR_DECISION = COST_FIELDS + ("one_time_eur", "exit_path")

# A saving below this monthly threshold is not material enough to act on.
MATERIALITY_EUR_MONTH = 5.0


def recurring_total(option: dict) -> float | None:
    """Monthly recurring cost, or None when any component is unknown."""
    values = [option[field] for field in COST_FIELDS]
    if any(value is None for value in values):
        return None
    return round(sum(values), 2)


def missing_fields(option: dict) -> list[str]:
    missing = [field for field in REQUIRED_FOR_DECISION if option[field] is None]
    if not option["utilization"]:
        missing.append("utilization")
    if not option["service_constraints"]:
        missing.append("service_constraints")
    return sorted(missing)


def utilization_ratios(option: dict) -> dict:
    """Used-over-provisioned per metric; None where either side is unknown."""
    ratios = {}
    for metric, pair in option["utilization"].items():
        provisioned = pair["provisioned"]["value"]
        used = pair["used"]["value"]
        if provisioned in (None, 0) or used is None:
            ratios[metric] = None
        else:
            ratios[metric] = round(used / provisioned, 4)
    return ratios


def compare_option(baseline: dict, alternative: dict) -> dict:
    """Compare one alternative against the baseline on the decision fields."""
    if alternative.get("excluded"):
        # The deciding authority set this option aside rather than wait for
        # evidence that was not going to arrive. It no longer blocks the case,
        # and the reason travels with the decision.
        return {
            "option_id": alternative["option_id"],
            "label": alternative["label"],
            "verdict": "excluded",
            "exclusion_reason": alternative.get("exclusion_reason"),
            "blocking_evidence": [],
        }

    blocking = sorted(set(
        [f"baseline.{field}" for field in missing_fields(baseline)]
        + [f"{alternative['option_id']}.{field}" for field in missing_fields(alternative)]
        + [f"baseline.unknown:{item}" for item in baseline["unknowns"]]
        + [f"{alternative['option_id']}.unknown:{item}" for item in alternative["unknowns"]]
    ))

    base_recurring = recurring_total(baseline)
    alt_recurring = recurring_total(alternative)
    monthly_delta = None
    monthly_saving = None
    if base_recurring is not None and alt_recurring is not None:
        monthly_delta = round(alt_recurring - base_recurring, 2)
        monthly_saving = round(-monthly_delta, 2)

    one_time = alternative["one_time_eur"]
    payback_months = None
    payback_note = None
    if monthly_saving is None or one_time is None:
        payback_note = "unknown: incomplete cost evidence"
    elif monthly_saving <= 0:
        payback_note = "never: the alternative does not reduce recurring cost"
    elif one_time == 0:
        payback_months = 0.0
        payback_note = "immediate: no one-time cost"
    else:
        payback_months = round(one_time / monthly_saving, 1)

    # Failure domains the alternative removes, and ones it newly introduces.
    base_domains = set(baseline["failure_domains"])
    alt_domains = set(alternative["failure_domains"])

    if blocking:
        verdict = "blocked_on_evidence"
    elif monthly_saving is not None and monthly_saving > MATERIALITY_EUR_MONTH:
        verdict = "recommend"
    elif monthly_delta is not None and abs(monthly_delta) <= MATERIALITY_EUR_MONTH:
        verdict = "no_material_change"
    else:
        verdict = "reject"

    return {
        "option_id": alternative["option_id"],
        "label": alternative["label"],
        "verdict": verdict,
        "baseline_recurring_eur_month": base_recurring,
        "alternative_recurring_eur_month": alt_recurring,
        "monthly_delta_eur": monthly_delta,
        "monthly_saving_eur": monthly_saving,
        "one_time_eur": one_time,
        "payback_months": payback_months,
        "payback_note": payback_note,
        "baseline_utilization": utilization_ratios(baseline),
        "alternative_utilization": utilization_ratios(alternative),
        "uncertainty": alternative["uncertainty"]["level"],
        "failure_domains_removed": sorted(base_domains - alt_domains),
        "failure_domains_added": sorted(alt_domains - base_domains),
        "exit_path_known": alternative["exit_path"] is not None,
        "blocking_evidence": blocking,
    }


def validate_case(case: dict) -> None:
    if case["schema_version"] != "0.1":
        raise ValueError("unsupported optimization-case schema version")
    if case["record_scope"] not in {"operational", "illustrative"}:
        raise ValueError("record_scope must be operational or illustrative")
    if case["case_type"] not in CASE_TYPES:
        raise ValueError(f"unknown case_type {case['case_type']}")
    if not case["case_id"].startswith("opt:"):
        raise ValueError("case_id must start with 'opt:'")
    if not case["resource_ids"]:
        raise ValueError("a case must name at least one resource")
    if any(not rid.startswith("resource:") for rid in case["resource_ids"]):
        raise ValueError("resource_ids must be portfolio resource identifiers")
    if not case["evidence"]:
        raise ValueError("a case must cite evidence")

    decision = case["decision"]
    if decision["state"] not in DECISION_STATES:
        raise ValueError(f"unknown decision state {decision['state']}")

    option_ids = [case["baseline"]["option_id"]] + [a["option_id"] for a in case["alternatives"]]
    if len(set(option_ids)) != len(option_ids):
        raise ValueError("option identifiers must be unique within a case")

    for alternative in case["alternatives"]:
        if alternative.get("excluded") and not alternative.get("exclusion_reason"):
            raise ValueError(
                f"excluded option {alternative['option_id']} must record an exclusion_reason"
            )

    report = evaluate(case)
    blocked = any(r["verdict"] == "blocked_on_evidence" for r in report["comparisons"])

    # A case cannot be approved while its own evidence is incomplete, and an
    # approval must name the authority that gave it.
    if decision["state"] in {"approved", "rejected"}:
        if decision["approver"] is None or decision["approved_on"] is None:
            raise ValueError("a decided case must record approver and approved_on")
    if decision["state"] == "approved":
        if blocked:
            raise ValueError("cannot approve a case with blocking evidence gaps")
        if decision["recommended_option_id"] not in option_ids:
            raise ValueError("approved case must recommend a known option")
    if decision["state"] == "blocked_on_evidence" and not blocked:
        raise ValueError("case is marked blocked but every decision field is known")
    if decision["state"] == "proposed" and blocked:
        raise ValueError("case has blocking evidence gaps and cannot be proposed")


def evaluate(case: dict) -> dict:
    baseline = case["baseline"]
    comparisons = [compare_option(baseline, alt) for alt in case["alternatives"]]
    recommended = [c for c in comparisons if c["verdict"] == "recommend"]
    recommended.sort(key=lambda c: (c["payback_months"] is None, c["payback_months"]))
    return {
        "case_id": case["case_id"],
        "case_type": case["case_type"],
        "record_scope": case["record_scope"],
        "review_period": case["review_period"],
        "baseline": {
            "option_id": baseline["option_id"],
            "label": baseline["label"],
            "recurring_eur_month": recurring_total(baseline),
            "utilization": utilization_ratios(baseline),
        },
        "comparisons": comparisons,
        "best_option_id": recommended[0]["option_id"] if recommended else None,
        "decision_state": case["decision"]["state"],
    }


def main() -> int:
    paths = sys.argv[1:] or sorted(str(p) for p in Path("data/optimization").glob("*.json"))
    if not paths:
        print("no optimization cases found", file=sys.stderr)
        return 2
    reports = []
    for path in paths:
        case = json.loads(Path(path).read_text())
        validate_case(case)
        reports.append(evaluate(case))
    print(json.dumps(reports, indent=2))
    return 0


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