resource-control/tools/capability.py

#!/usr/bin/env python3
"""Validate capability requirements and provisions against the live ITC-CAP catalog.

resource-control does not copy the canon. It reads `capabilities.yaml` from
info-tech-canon and checks that every capability id, profile, quality dimension,
maturity level, resource class, and predicate this repository uses is one the
canon actually declares. A drift in either repository fails here rather than
silently producing a record the canon would reject.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import basis as basis_module

CANON_PATHS = (
    Path("/home/worsch/info-tech-canon/infospace/models/capability/capabilities.yaml"),
    Path("../info-tech-canon/infospace/models/capability/capabilities.yaml"),
)
PREDICATES = frozenset({"not_in", "in", "equals", "lte", "gte"})


def _parse_yaml(text: str) -> dict:
    try:
        import yaml
    except ModuleNotFoundError:  # pragma: no cover - yaml is present in this env
        raise RuntimeError("PyYAML is required to read the canon catalog")
    return yaml.safe_load(text)


def load_canon(paths=CANON_PATHS) -> dict:
    for path in paths:
        if path.exists():
            catalog = _parse_yaml(path.read_text())
            capabilities = {
                cap["id"]: cap
                for domain in catalog["domains"]
                for cap in domain.get("capabilities", [])
            }
            return {
                "version": catalog["canon"]["version"],
                "canon_version": catalog["canon"]["canon_version"],
                "capabilities": capabilities,
                "resource_classes": {rc["id"]: rc for rc in catalog["resource_classes"]},
                "maturity_levels": {m["id"] if isinstance(m, dict) else m for m in catalog["maturity_levels"]},
                "source": str(path),
            }
    raise FileNotFoundError("ITC-CAP catalog not found; is info-tech-canon checked out?")


def validate_requirement(requirement: dict, canon: dict) -> None:
    cap_id = requirement["capability"]
    capability = canon["capabilities"].get(cap_id)
    if capability is None:
        raise ValueError(f"unknown capability {cap_id}")

    profile = requirement.get("profile")
    if profile and profile not in (capability.get("profiles") or []):
        raise ValueError(f"{cap_id} does not declare profile {profile!r}")

    maturity = requirement.get("minimum_maturity")
    if maturity and maturity not in canon["maturity_levels"]:
        raise ValueError(f"unknown maturity level {maturity!r}")

    dimensions = set(capability.get("quality_dimensions") or [])
    # CAP-R9: a target key or constraint dimension must be declared on the
    # capability. This is what stops a consumer inventing quality vocabulary.
    for key in (requirement.get("targets") or {}):
        if key not in dimensions:
            raise ValueError(f"{cap_id} does not declare quality dimension {key!r}")
    for constraint in requirement.get("constraints") or []:
        if constraint["dimension"] not in dimensions:
            raise ValueError(f"{cap_id} does not declare quality dimension {constraint['dimension']!r}")
        if constraint["predicate"] not in PREDICATES:
            raise ValueError(f"predicate {constraint['predicate']!r} is not in the closed set")
        if not constraint.get("of"):
            raise ValueError("a constraint must name what it applies to")


def validate_provision(provision: dict, canon: dict) -> None:
    cap_id = provision["capability"]
    capability = canon["capabilities"].get(cap_id)
    if capability is None:
        raise ValueError(f"unknown capability {cap_id}")
    if provision["maturity"] not in canon["maturity_levels"]:
        raise ValueError(f"unknown maturity level {provision['maturity']!r}")
    profile = provision.get("profile")
    if profile and profile not in (capability.get("profiles") or []):
        raise ValueError(f"{cap_id} does not declare profile {profile!r}")

    hooks = set(capability.get("evidence_hooks") or [])
    for item in provision.get("evidence") or []:
        if item["hook"] not in hooks:
            raise ValueError(f"{cap_id} does not declare evidence hook {item['hook']!r}")
        basis_module.rank(item["basis"])

    seen = set()
    for row in provision.get("consumes") or []:
        klass = row["class"]
        if klass not in canon["resource_classes"]:
            raise ValueError(f"unknown resource class {klass!r}")
        if klass in seen:
            raise ValueError(f"duplicate consumption row for class {klass}")
        seen.add(klass)

        declared = canon["resource_classes"][klass]
        unit = row["quantity"]["unit"]
        if unit != declared["native_unit"]:
            raise ValueError(
                f"class {klass} native unit is {declared['native_unit']!r}, got {unit!r}"
            )
        # Unknown is recorded as unknown, never as zero.
        basis_module.validate_value({
            "basis": row["basis"],
            "value": row["quantity"]["value"],
            "gap": row.get("gap"),
            "derived_from": row.get("derived_from"),
        })
        if row.get("supply") and row["supply"] not in {"internal", "external"}:
            raise ValueError(f"supply must be internal or external, got {row['supply']!r}")

    validate_uses_provisions(provision, canon)


def validate_uses_provisions(provision: dict, canon: dict) -> None:
    """CAP-R11: relying on another provision is a relationship, not consumption.

    A `depends_on` entry MUST correspond to a `depends_on` declared between the
    two capabilities in the catalog, so the provision graph stays checkable
    against the capability graph rather than free-form.
    """
    capability = canon["capabilities"][provision["capability"]]
    for entry in provision.get("uses_provisions") or []:
        target = entry["capability"]
        if target not in canon["capabilities"]:
            raise ValueError(f"unknown capability {target} in uses_provisions")
        relation = entry["relation"]
        if relation not in {"depends_on", "may_use"}:
            raise ValueError(f"relation must be depends_on or may_use, got {relation!r}")
        if not entry.get("provider"):
            raise ValueError(f"uses_provisions entry for {target} must name a provider")
        declared = set(capability.get(relation) or [])
        if relation == "depends_on" and target not in declared:
            raise ValueError(
                f"{provision['capability']} does not declare depends_on {target} in the catalog"
            )
        if relation == "may_use" and target not in declared:
            # SHOULD, not MUST — surfaced rather than fatal.
            entry.setdefault("_note", f"catalog does not declare may_use {target}")


def evidence_coverage(provision: dict, canon: dict) -> dict:
    """Which of the capability's declared evidence hooks this provision satisfies."""
    capability = canon["capabilities"][provision["capability"]]
    declared = list(capability.get("evidence_hooks") or [])
    supplied = {item["hook"] for item in provision.get("evidence") or []}
    return {
        "declared": declared,
        "supplied": sorted(supplied),
        "missing": sorted(set(declared) - supplied),
        "complete": not (set(declared) - supplied),
    }


def consumption_profile(provision: dict) -> dict:
    values = [
        {
            "name": f"{row['class']}:{row.get('name', '')}",
            "basis": row["basis"],
            "value": row["quantity"]["value"],
            "gap": row.get("gap"),
            "derived_from": row.get("derived_from"),
            "proxy_for": row.get("proxy_for"),
        }
        for row in provision.get("consumes") or []
    ]
    return basis_module.profile(values)


def review(record: dict, canon: dict) -> dict:
    for requirement in record.get("requires") or []:
        validate_requirement(requirement, canon)
    for provision in record.get("provisions") or []:
        validate_provision(provision, canon)

    # Does each requirement actually have a provision that meets it?
    provisions = {p["capability"]: p for p in record.get("provisions") or []}
    met = []
    for requirement in record.get("requires") or []:
        provision = provisions.get(requirement["capability"])
        wanted = requirement.get("minimum_maturity")
        if provision is None:
            met.append({"capability": requirement["capability"], "status": "unprovided"})
            continue
        levels = sorted(canon["maturity_levels"])
        satisfied = levels.index(provision["maturity"]) >= levels.index(wanted) if wanted else True
        met.append({
            "capability": requirement["capability"],
            "required": wanted,
            "provided": provision["maturity"],
            "status": "met" if satisfied else "below_requirement",
        })

    return {
        "record_id": record["record_id"],
        "canon": {k: canon[k] for k in ("version", "canon_version", "source")},
        "requirements": met,
        "provisions": [
            {
                "capability": provision["capability"],
                "maturity": provision["maturity"],
                "evidence": evidence_coverage(provision, canon),
                "consumption": consumption_profile(provision),
                "uses_provisions": [
                    {"capability": u["capability"], "relation": u["relation"], "provider": u["provider"]}
                    for u in provision.get("uses_provisions") or []
                ],
            }
            for provision in record.get("provisions") or []
        ],
        "alternatives_grade": (
            basis_module.decision_grade(record["modelled_alternatives"]["values"])
            if record.get("modelled_alternatives") else None
        ),
    }


def main() -> int:
    paths = sys.argv[1:] or sorted(str(p) for p in Path("data/capability").glob("*.json"))
    canon = load_canon()
    reports = [review(json.loads(Path(p).read_text()), canon) for p in paths]
    print(json.dumps(reports, indent=2))
    return 0


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