resource-control/tools/basis.py

#!/usr/bin/env python3
"""Evidence basis: how a value was obtained, and how far it can be trusted.

The vocabulary is **owned by ITC-GOV** and read from
`infospace/models/governance/evidence-basis.yaml` in info-tech-canon. This
module does not vendor it. resource-control originated the concept, filed it as
demand, and now consumes the canon's version of it — drift in either repository
fails here rather than diverging quietly.

The central rule is that a derived value is only as strong as the weakest
*tier* among its inputs. Without it, arithmetic launders assumptions: EUR 30.00
of monthly labour reads like a measurement when it is another repository's
estimate of hours multiplied by a rate we chose.
"""

from __future__ import annotations

import json
import sys
from functools import lru_cache
from pathlib import Path

CATALOG_PATHS = (
    Path("/home/worsch/info-tech-canon/infospace/models/governance/evidence-basis.yaml"),
    Path("../info-tech-canon/infospace/models/governance/evidence-basis.yaml"),
)


@lru_cache(maxsize=1)
def load_catalog(paths: tuple = CATALOG_PATHS) -> dict:
    """Read the ITC-GOV EvidenceBasis catalog."""
    import yaml

    for path in paths:
        if path.exists():
            raw = yaml.safe_load(path.read_text())
            tiers = {tier["id"]: tier for tier in raw["tiers"]}
            bases = {b["id"]: b for b in raw["bases"]}
            # Membership comes from tiers[].members, which is authoritative.
            # The coarser bases[].tier label is not usable as a tier id: it
            # reads "evidenced" for invoiced/measured/quoted while the tier
            # list splits those across "observed" and "quoted".
            tier_of = {
                member: tier["id"]
                for tier in raw["tiers"]
                for member in tier["members"]
            }
            # A basis in no tier (derived) has no resolved tier by design.
            tier_of.update({b: None for b in bases if b not in tier_of})
            return {
                "version": raw["canon"]["version"],
                "canon_version": raw["canon"]["canon_version"],
                "source": str(path),
                "bases": bases,
                "order": [b["id"] for b in raw["bases"]],
                "tiers": tiers,
                "tier_of": tier_of,
                "tier_rank": {t["id"]: t["rank"] for t in raw["tiers"]},
                "evidenced_tiers": {t["id"] for t in raw["tiers"] if t.get("evidenced")},
                "grades": {g["id"]: g["when"] for g in raw["decision_grades"]},
            }
    raise FileNotFoundError(
        "ITC-GOV EvidenceBasis catalog not found; is info-tech-canon checked out?"
    )


def _catalog(catalog: dict | None = None) -> dict:
    return catalog or load_catalog()


def bases(catalog: dict | None = None) -> list[str]:
    return list(_catalog(catalog)["order"])


def tier_of(basis: str, catalog: dict | None = None) -> str:
    """The resolved strength tier of a basis.

    `derived` deliberately has none: resolve it against its inputs first.
    """
    cat = _catalog(catalog)
    if basis not in cat["tier_of"]:
        raise ValueError(f"unknown evidence basis {basis!r}")
    tier = cat["tier_of"][basis]
    if tier is None:
        raise ValueError(
            f"{basis!r} has no resolved tier; resolve it against derived_from first"
        )
    return tier


def rank(basis: str, catalog: dict | None = None) -> int:
    """Tier rank; lower is stronger. Members of one tier share a rank."""
    cat = _catalog(catalog)
    return cat["tier_rank"][tier_of(basis, cat)]


def weakest(names, catalog: dict | None = None) -> str:
    """The weakest basis in a collection, compared by tier.

    Ties within a tier resolve by catalog order so the result is deterministic
    without implying a strength difference the canon does not assert.
    """
    cat = _catalog(catalog)
    names = list(names)
    if not names:
        return "unknown"
    order = cat["order"]
    return min(names, key=lambda b: (-rank(b, cat), order.index(b)))


def strongest(names, catalog: dict | None = None) -> str:
    cat = _catalog(catalog)
    names = list(names)
    if not names:
        return "unknown"
    order = cat["order"]
    return min(names, key=lambda b: (rank(b, cat), order.index(b)))


def is_evidenced(basis: str, catalog: dict | None = None) -> bool:
    """True when the value asserts an observed or contracted fact.

    `quoted` is weaker than `observed` for propagation and still counts as
    evidenced for a decision grade — the canon separates those two uses.
    """
    cat = _catalog(catalog)
    return tier_of(basis, cat) in cat["evidenced_tiers"]


def resolve(value: dict, catalog: dict | None = None) -> str:
    """Effective basis of a value, propagating through derivation."""
    cat = _catalog(catalog)
    basis = value.get("basis", "unknown")
    if basis not in cat["bases"]:
        raise ValueError(f"unknown evidence basis {basis!r}")
    if basis != "derived":
        return basis
    inputs = value.get("derived_from") or []
    if not inputs:
        raise ValueError("a derived value must record derived_from")
    return weakest([resolve(item, cat) for item in inputs], cat)


def validate_value(value: dict, catalog: dict | None = None) -> None:
    """Enforce the ITC-GOV rules: unknown-is-not-zero and derived-names-inputs."""
    cat = _catalog(catalog)
    basis = value.get("basis")
    if basis not in cat["bases"]:
        raise ValueError(f"unknown evidence basis {basis!r}")
    if basis == "derived" and not value.get("derived_from"):
        raise ValueError("a derived value must record derived_from")
    if basis != "derived" and value.get("derived_from"):
        raise ValueError("only a derived value may record derived_from")
    if basis == "unknown":
        if value.get("value") is not None:
            raise ValueError("a value with basis unknown must not carry a quantity")
        if not value.get("gap"):
            raise ValueError("an unknown value must name the gap and its owner")
    elif value.get("value") is None:
        raise ValueError("a known basis must carry a quantity; use basis unknown instead")
    proxy_for = value.get("proxy_for")
    if proxy_for is not None and not str(proxy_for).strip():
        raise ValueError("proxy_for must name the quantity actually wanted")
    for item in value.get("derived_from") or []:
        validate_value(item, cat)


def profile(values, catalog: dict | None = None) -> dict:
    """Summarize a set of values for decision review."""
    cat = _catalog(catalog)
    resolved, proxies = [], []
    for value in values:
        validate_value(value, cat)
        resolved.append(resolve(value, cat))
        if value.get("proxy_for"):
            proxies.append({"name": value.get("name"), "proxy_for": value["proxy_for"]})
    counts: dict[str, int] = {}
    for basis in resolved:
        counts[basis] = counts.get(basis, 0) + 1
    evidenced = [b for b in resolved if is_evidenced(b, cat)]
    weakest_basis = weakest(resolved, cat)
    return {
        "count": len(resolved),
        "by_basis": {k: counts[k] for k in cat["order"] if k in counts},
        "weakest": weakest_basis,
        "weakest_tier": tier_of(weakest_basis, cat) if resolved else "unknown",
        "evidenced": len(evidenced),
        "evidenced_ratio": round(len(evidenced) / len(resolved), 4) if resolved else None,
        "proxies": proxies,
    }


# Decision grade per tier, from the catalog's decision_grades.
_GRADE_BY_TIER = {
    "observed": "evidenced",
    "quoted": "evidenced",
    "projected": "projected",
    "judgement": "indicative",
    "unknown": "insufficient",
}
_GRADE_NOTE = {
    "evidenced": "every load-bearing value is observed, invoiced, or quoted",
    "projected": "the conclusion rests on values projected from observations",
    "indicative": "the conclusion is no stronger than an estimated or assumed value",
    "insufficient": "at least one load-bearing value is unknown",
}


def decision_grade(values, catalog: dict | None = None) -> dict:
    """Grade a decision by the weakest tier it actually rests on.

    A conclusion is not stronger than its weakest load-bearing input, however
    precise the arithmetic between them looks.
    """
    cat = _catalog(catalog)
    summary = profile(values, cat)
    tier = summary["weakest_tier"]
    grade = _GRADE_BY_TIER[tier]
    note = _GRADE_NOTE[grade]
    if summary["proxies"]:
        note += f"; {len(summary['proxies'])} value(s) measure a proxy rather than the quantity named"
    return {**summary, "grade": grade, "note": note}


def main() -> int:
    if len(sys.argv) != 2:
        print(f"usage: {sys.argv[0]} VALUES.json", file=sys.stderr)
        return 2
    payload = json.loads(Path(sys.argv[1]).read_text())
    values = payload["values"] if isinstance(payload, dict) else payload
    print(json.dumps(decision_grade(values), indent=2))
    return 0


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