prose-contracts/scripts/disk-gc-plan.py

#!/usr/bin/env python3
"""disk-gc-plan — turn a fleet disk scan into the GC action plan.

This is the executable side of `disk-gc-threat-response.prose.md`. It exists so the
report-only gate is enforced by code that can be tested, rather than by prose the
executor might misread.

THE HARD GATE: guests listed in the contract's `report_only_guests` block are
DETECT-AND-REPORT-ONLY at EVERY level (AMBER, RED, CRITICAL). This tool will never
emit a `gc-executor` action for one, so no GC command can be constructed for it.

The gate is keyed on GUEST identity — guest id, hostname, or IP — never on an agent
name. An agent-name marker can silently miss the guest it lives on; a guest marker
cannot.

The authoritative exclusion list lives in the contract itself (the fenced ```yaml
block containing `report_only_guests:`). This tool reads it from there so there is
only ever one copy.

Usage:
    disk-gc-plan.py --scan scan.json            # [{"id":111,"usage_pct":84}, ...]
    cat scan.json | disk-gc-plan.py             # same, via stdin
    disk-gc-plan.py --scan scan.json --json     # machine-readable plan

Scan entries may carry any of: id / guest / vmid / ct / ctid, hostname / name, ip.
A threshold-crossing entry with no recognizable identity is reported, never GC'd.
Exit codes: 0 ok, 1 usage/parse error.
"""
from __future__ import annotations

import argparse
import json
import pathlib
import re
import sys

import yaml

REPO = pathlib.Path(__file__).resolve().parent.parent
DEFAULT_CONTRACT = REPO / "disk-gc-threat-response.prose.md"

AMBER, RED, CRITICAL = 75, 85, 95

IDENTITY_FIELDS = ("id", "guest", "vmid", "ct", "ctid", "hostname", "name", "ip")
IDENTITY_TYPE_PREFIX = re.compile(r"^(?:lxc|qemu)/")
UNIDENTIFIED_REASON = "unidentified target - refusing to schedule GC"


def load_report_only_guests(contract_path: pathlib.Path) -> list[dict]:
    """Read the authoritative report_only_guests block out of the contract.

    The contract carries it as a fenced ```yaml block. Parsing the declared,
    machine-readable block is the intended interface — the contract owns the list.
    """
    text = contract_path.read_text(encoding="utf-8")
    for block in re.findall(r"```yaml\n(.*?)```", text, re.S):
        if "report_only_guests:" in block:
            data = yaml.safe_load(block)
            guests = data.get("report_only_guests") or []
            if not isinstance(guests, list):
                raise SystemExit("report_only_guests must be a list")
            if not guests:
                raise SystemExit(
                    "report_only_guests is empty or missing - refusing to plan GC "
                    "without the report-only gate"
                )
            for guest in guests:
                if not isinstance(guest, dict) or not _keys(guest):
                    raise SystemExit(
                        "report_only_guests entry has no recognizable identity key "
                        f"(expected one of: {', '.join(IDENTITY_FIELDS)}): {guest!r}"
                    )
            return guests
    raise SystemExit(
        f"no authoritative report_only_guests block found in {contract_path}"
    )


def _canonical_number(number: float) -> str:
    if float(number).is_integer():
        return str(int(number))
    return str(number).strip().lower()


def _normalize_identity(value: object) -> str:
    """Canonicalise a guest identity so differently-encoded ids compare equal:
    numeric and numeric-string ids collapse to an integer string, Proxmox
    type prefixes and leading zeros are stripped, and hostnames/IPs are only
    trimmed and lowercased."""
    if isinstance(value, bool):
        return str(value).strip().lower()
    if isinstance(value, (int, float)):
        return _canonical_number(float(value))
    text = str(value).strip().lower()
    text = IDENTITY_TYPE_PREFIX.sub("", text)
    try:
        return _canonical_number(float(text))
    except ValueError:
        return text


def _keys(entry: dict) -> set[str]:
    """Guest/host identity keys, shared by exclusions and scan entries so the two
    sides of the gate can never key on different fields."""
    out: set[str] = set()
    for field in IDENTITY_FIELDS:
        value = entry.get(field)
        if value is None:
            continue
        key = _normalize_identity(value)
        if key:
            out.add(key)
    return out


def level_for(pct: float) -> str | None:
    if pct >= CRITICAL:
        return "CRITICAL"
    if pct >= RED:
        return "RED"
    if pct >= AMBER:
        return "AMBER"
    return None


def build_plan(scan: list[dict], report_only: list[dict]) -> list[dict]:
    excluded = [(e, _keys(e)) for e in report_only]
    plan: list[dict] = []
    for entry in scan:
        pct = entry.get("usage_pct")
        if pct is None:
            continue
        level = level_for(float(pct))
        if level is None:
            continue  # GREEN: log only, no action
        scan_keys = _keys(entry)
        target = next(
            (entry.get(k) for k in IDENTITY_FIELDS if entry.get(k) not in (None, "")),
            "?",
        )
        if not scan_keys:
            plan.append({
                "target": target,
                "level": level,
                "pct": float(pct),
                "action": "report-only",
                "reason": UNIDENTIFIED_REASON,
            })
            continue
        match = next((e for e, keys in excluded if keys & scan_keys), None)
        if match is not None:
            plan.append({
                "target": target,
                "level": level,
                "pct": float(pct),
                "action": "report-only",
                "reason": match.get("reason", "").strip(),
            })
        else:
            plan.append({
                "target": target,
                "level": level,
                "pct": float(pct),
                "action": "gc-executor",
            })
    plan.sort(key=lambda row: row["pct"], reverse=True)
    return plan


def main() -> int:
    ap = argparse.ArgumentParser(description="Plan disk GC actions with the report-only gate.")
    ap.add_argument("--scan", help="JSON file: list of {id|ct|hostname|ip, usage_pct}")
    ap.add_argument("--contract", default=str(DEFAULT_CONTRACT))
    ap.add_argument("--json", action="store_true", help="emit the plan as JSON")
    args = ap.parse_args()

    raw = pathlib.Path(args.scan).read_text() if args.scan else sys.stdin.read()
    try:
        scan = json.loads(raw)
    except json.JSONDecodeError as exc:
        print(f"invalid scan JSON: {exc}", file=sys.stderr)
        return 1
    if not isinstance(scan, list):
        print("scan must be a JSON list", file=sys.stderr)
        return 1

    report_only = load_report_only_guests(pathlib.Path(args.contract))
    plan = build_plan(scan, report_only)

    if args.json:
        print(json.dumps(plan, indent=2))
        return 0

    if not plan:
        print("no threats (nothing at or above 75%)")
        return 0
    for row in plan:
        if row["action"] == "report-only":
            print(f"  {row['target']} {row['level']} {row['pct']}% -> REPORT-ONLY (no GC) — {row['reason']}")
        else:
            print(f"  {row['target']} {row['level']} {row['pct']}% -> gc-executor")
    return 0


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