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, hostname / name, ip.
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


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")
            return guests
    raise SystemExit(
        f"no authoritative report_only_guests block found in {contract_path}"
    )


def _keys(entry: dict) -> set[str]:
    """Guest/host identity keys for an exclusion entry."""
    out: set[str] = set()
    for field in ("guest", "id", "vmid", "hostname", "name", "ip"):
        value = entry.get(field)
        if value is not None and str(value).strip():
            out.add(str(value).strip().lower())
    return out


def _entry_keys(scan_entry: dict) -> set[str]:
    out: set[str] = set()
    for field in ("id", "guest", "vmid", "hostname", "name", "ip"):
        value = scan_entry.get(field)
        if value is not None and str(value).strip():
            out.add(str(value).strip().lower())
    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 = _entry_keys(entry)
        match = next((e for e, keys in excluded if keys & scan_keys), None)
        target = next(
            (entry.get(k) for k in ("id", "guest", "vmid", "hostname", "name", "ip")
             if entry.get(k) not in (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|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())