llm-ingestion-okf/tools/okf_contract_check.py

"""Check a consumption skill and one pre-pass payload against the contract.

`docs/consumption-contract.md` is normative prose. Prose has no test, so this
command reads the mechanically checkable half of it: the payload shape (SS 3, 5,
6, 7, 8) and the fixed vocabulary a conformant `SKILL.md` must carry (SS 4, 6).

**What it cannot see, said here rather than implied.** The division of labour
(SS 2) and the prohibitions (SS 9) are properties of a RUN -- whether an agent
went looking for context the pre-pass withheld, whether a retrieval tool was
pointed at the verdict layer. No static check reaches them. Conformance here is
the floor, never the proof, and a green run means the payload carries what a
claim would have to rest on -- not that the claim was made honestly.

**Every rule has its own code.** One "invalid" verdict over fourteen different
defects is a diagnostic no caller can act on. The codes are the contract's
paragraphs made addressable.

**The report carries denominators**, because the contract requires them of its
consumers and a checker exempting itself would be stating the rule it breaks:
how many rules ran, over how many excerpts and withheld entries.

Exit codes are three, not two: 0 conformant, 1 non-conformant, 2 the check did
not run. Collapsing 2 into 1 would report an unread file as a failed check.

It lives outside `src/`, so it never enters a wheel and no consumer's install
surface changes because it exists.
"""

from __future__ import annotations

import argparse
import json
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

#: SS 4.1. Five literals, spelled exactly. `[unverifiable-from-bundle]` is one
#: literal string -- no variants, no translations -- so this is a membership
#: test on the literal, never on a translated concept name.
REQUIRED_MARKINGS = (
    "extracted",
    "derived",
    "[unverifiable-from-bundle]",
    "[unread]",
    "[sourced-not-sufficient]",
)

#: SS 6.1. A closed set of three, and the third is a real state: `unknown` says
#: the concept carries no `adjudication` key, which is what an older bundle
#: looks like. Collapsing it into `proposed` or into absence reports "not
#: judged" where the truth is "we cannot tell", and only one of those is a fact
#: about the concept.
ADJUDICATION_STATES = ("proposed", "adjudicated", "unknown")

#: SS 6.2, from SPEC SS 5.3: derived from `verified`, lowest to highest.
TRUST_TIERS = ("unverified", "machine-confirmed", "human-reviewed")

#: The level-2 headings a conformant skill carries. Checked by name because the
#: alternative -- searching the prose for the obligation -- passes on any
#: document that happens to use the words.
REQUIRED_SECTIONS = (
    "Pre-pass",
    "Division of labour",
    "Markings",
    "States",
    "Budget",
    "Denominators",
    "Prohibitions",
)


@dataclass(frozen=True)
class Finding:
    """One contract paragraph, unmet, named by its code."""

    code: str
    message: str


@dataclass(frozen=True)
class Report:
    """Findings plus the denominators they were measured over."""

    findings: tuple[Finding, ...]
    rules_evaluated: int
    excerpts_examined: int
    withheld_examined: int

    def render(self) -> str:
        denominator = (
            f"{self.rules_evaluated} rules over {self.excerpts_examined} excerpts "
            f"and {self.withheld_examined} withheld entries"
        )
        if not self.findings:
            return f"conformant: {denominator}, 0 findings"
        lines = [f"NOT conformant: {denominator}, {len(self.findings)} findings"]
        lines += [f"  {finding.code}: {finding.message}" for finding in self.findings]
        return "\n".join(lines)


@dataclass(frozen=True)
class Context:
    """What every rule reads. `payload` is empty when the payload is not a
    mapping at all, so each rule stays a total function over its input."""

    skill: str
    payload: Mapping[str, Any]
    payload_is_mapping: bool


def _mapping(value: object) -> Mapping[str, Any]:
    return value if isinstance(value, Mapping) else {}


def _sequence(value: object) -> Sequence[Any]:
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
        return value
    return ()


def _text(value: object) -> str:
    return value if isinstance(value, str) else ""


def _whole(value: object) -> int | None:
    # `bool` is an `int` in Python and would make `True` a budget. Excluded by
    # name rather than trusted not to arrive.
    return value if isinstance(value, int) and not isinstance(value, bool) else None


# --- Payload rules -----------------------------------------------------------


def rule_payload_shape(ctx: Context) -> list[Finding]:
    if ctx.payload_is_mapping:
        return []
    return [Finding("payload_invalid", "the payload is not a JSON object (SS 8)")]


def rule_contract_version(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping or _text(ctx.payload.get("contract")):
        return []
    return [
        Finding(
            "contract_unversioned",
            "the payload names no `contract` revision, so a reader cannot tell "
            "which revision it is holding (SS 8.2)",
        )
    ]


def rule_bundle_ref(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    bundle = _mapping(ctx.payload.get("bundle"))
    findings = []
    if not _text(bundle.get("ref")):
        findings.append(
            Finding(
                "ref_missing",
                "the payload names no bundle `ref`; a version is the producer's "
                "assertion, a ref is a fact about bytes (SS 3.3)",
            )
        )
    if not _text(bundle.get("bundle_id")):
        findings.append(Finding("ref_missing", "the payload names no `bundle.bundle_id` (SS 3.1)"))
    return findings


def rule_excerpt_source_marking(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    findings = []
    for position, raw in enumerate(_sequence(ctx.payload.get("excerpts"))):
        excerpt = _mapping(raw)
        for key in ("bundle_id", "concept_id", "sha256"):
            if not _text(excerpt.get(key)):
                findings.append(
                    Finding(
                        "source_marking_incomplete",
                        f"excerpt {position} carries no {key!r}; identity across "
                        "bundles is the (bundle_id, concept_id) tuple with the "
                        "digest of the bytes it was taken from (SS 3.1, SS 3.2)",
                    )
                )
    return findings


def rule_excerpt_states(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    findings = []
    closed: tuple[tuple[str, tuple[str, ...]], ...] = (
        ("adjudication", ADJUDICATION_STATES),
        ("trust_tier", TRUST_TIERS),
    )
    for position, raw in enumerate(_sequence(ctx.payload.get("excerpts"))):
        excerpt = _mapping(raw)
        for key, allowed in closed:
            if key not in excerpt:
                findings.append(
                    Finding(
                        "state_missing",
                        f"excerpt {position} carries no {key!r}; the third state is "
                        "written explicitly, never omitted -- omission collapses "
                        '"we cannot tell" into a fact about the concept (SS 6.1)',
                    )
                )
                continue
            value = _text(excerpt.get(key))
            if value not in allowed:
                findings.append(
                    Finding(
                        "state_not_in_set",
                        f"excerpt {position} has {key}={value!r}, outside the closed "
                        f"set {allowed} (SS 6)",
                    )
                )
    return findings


def rule_denominator_identity(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    counts = _mapping(ctx.payload.get("denominators"))
    values = {key: _whole(counts.get(key)) for key in ("considered", "withheld", "delivered")}
    missing = sorted(key for key, value in values.items() if value is None)
    if missing:
        return [
            Finding(
                "denominator_identity",
                f"the payload reports no whole-number {', '.join(missing)}; a count "
                "that is not reported is unmeasured, not zero (SS 5.1)",
            )
        ]
    considered = values["considered"]
    withheld = values["withheld"]
    delivered = values["delivered"]
    assert considered is not None and withheld is not None and delivered is not None
    if considered != withheld + delivered:
        return [
            Finding(
                "denominator_identity",
                f"considered ({considered}) != withheld ({withheld}) + delivered "
                f"({delivered}); a count that does not close is not a denominator "
                "(SS 5.2)",
            )
        ]
    return []


def rule_denominator_lists(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    counts = _mapping(ctx.payload.get("denominators"))
    findings = []
    for key, member in (("delivered", "excerpts"), ("withheld", "withheld")):
        declared = _whole(counts.get(key))
        if declared is None:
            continue
        actual = len(_sequence(ctx.payload.get(member)))
        if declared != actual:
            findings.append(
                Finding(
                    "denominator_list_mismatch",
                    f"denominators.{key} is {declared} but {member} holds {actual}; "
                    "the count and the list are two statements of one fact (SS 8.1)",
                )
            )
    return findings


def rule_withheld_rules(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    findings = []
    for position, raw in enumerate(_sequence(ctx.payload.get("withheld"))):
        entry = _mapping(raw)
        for key in ("concept_id", "rule"):
            if not _text(entry.get(key)):
                findings.append(
                    Finding(
                        "cut_undeclared",
                        f"withheld entry {position} carries no {key!r}; a visible "
                        "drop is worth more than a silent one (SS 5.3)",
                    )
                )
    return findings


def rule_budget_declared(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    budget = _mapping(ctx.payload.get("budget"))
    findings = []
    for key in ("unit", "instrument"):
        if not _text(budget.get(key)):
            findings.append(
                Finding(
                    "budget_undeclared",
                    f"the budget names no {key!r}; a number without its instrument "
                    "and unit is not a measurement (SS 7.1)",
                )
            )
    limit = _whole(budget.get("limit"))
    if limit is None or limit <= 0:
        findings.append(
            Finding(
                "budget_undeclared",
                '"bounded" without a bound is a denominator failure in prose (SS 7.1)',
            )
        )
    if _whole(budget.get("spent")) is None:
        findings.append(Finding("budget_undeclared", "the budget reports no `spent` (SS 7.2)"))
    return findings


def rule_budget_gate(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    budget = _mapping(ctx.payload.get("budget"))
    limit = _whole(budget.get("limit"))
    spent = _whole(budget.get("spent"))
    if limit is None or spent is None or spent <= limit:
        return []
    return [
        Finding(
            "budget_exceeded",
            f"spent ({spent}) exceeds limit ({limit}); the cut strategy is wrong "
            "for this bundle, which is a finding requiring a decision and never "
            "a retry with a narrower question (SS 7.3)",
        )
    ]


def rule_instrument_validated(ctx: Context) -> list[Finding]:
    if not ctx.payload_is_mapping:
        return []
    known = _mapping(_mapping(ctx.payload.get("budget")).get("known_positive"))
    expected = _whole(known.get("expected"))
    measured = _whole(known.get("measured"))
    if expected is None or measured is None or expected <= 0 or not _text(known.get("case")):
        return [
            Finding(
                "instrument_unvalidated",
                "the budget carries no usable `known_positive` (case, expected, "
                "measured); an instrument that has not reproduced a known figure "
                "has not been shown to count (SS 7.4)",
            )
        ]
    if expected != measured:
        return [
            Finding(
                "instrument_unvalidated",
                f"the known-positive expected {expected} and the instrument "
                f"measured {measured} (SS 7.4)",
            )
        ]
    return []


# --- Skill rules -------------------------------------------------------------


def rule_skill_sections(ctx: Context) -> list[Finding]:
    return [
        Finding(
            "skill_section_missing",
            f"the skill carries no `## {section}` section (SS 8 of this checker's "
            "reading; the headings are fixed so they can be checked by name)",
        )
        for section in REQUIRED_SECTIONS
        if f"## {section}" not in ctx.skill
    ]


def rule_skill_markings(ctx: Context) -> list[Finding]:
    return [
        Finding(
            "skill_marking_missing",
            f"the skill does not carry the required marking {marking!r} verbatim "
            "(SS 4.1); one literal string, no variants, no translations",
        )
        for marking in REQUIRED_MARKINGS
        if marking not in ctx.skill
    ]


def rule_skill_states(ctx: Context) -> list[Finding]:
    return [
        Finding(
            "skill_state_missing",
            f"the skill does not name the state {state!r}, so its consumer cannot "
            "be held to the closed set (SS 6)",
        )
        for state in (*ADJUDICATION_STATES, *TRUST_TIERS)
        if f"`{state}`" not in ctx.skill
    ]


#: Every rule, in report order. `len(RULES)` is the denominator the report
#: quotes: "how many rules ran" is the number that makes "0 findings" mean
#: something.
RULES: tuple[Callable[[Context], list[Finding]], ...] = (
    rule_payload_shape,
    rule_contract_version,
    rule_bundle_ref,
    rule_excerpt_source_marking,
    rule_excerpt_states,
    rule_denominator_identity,
    rule_denominator_lists,
    rule_withheld_rules,
    rule_budget_declared,
    rule_budget_gate,
    rule_instrument_validated,
    rule_skill_sections,
    rule_skill_markings,
    rule_skill_states,
)


def check(skill_text: str, payload: object) -> Report:
    """Run every rule. No rule short-circuits another: a caller fixing one
    defect should not discover a second only on the next run."""
    is_mapping = isinstance(payload, Mapping)
    ctx = Context(
        skill=skill_text,
        payload=payload if isinstance(payload, Mapping) else {},
        payload_is_mapping=is_mapping,
    )
    findings: list[Finding] = []
    for rule in RULES:
        findings.extend(rule(ctx))
    return Report(
        findings=tuple(findings),
        rules_evaluated=len(RULES),
        excerpts_examined=len(_sequence(ctx.payload.get("excerpts"))),
        withheld_examined=len(_sequence(ctx.payload.get("withheld"))),
    )


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("--skill", type=Path, required=True, help="the SKILL.md to check")
    parser.add_argument("--payload", type=Path, required=True, help="one pre-pass payload (JSON)")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        skill_text = args.skill.read_text(encoding="utf-8")
    except OSError as exc:
        print(f"could not read the skill: {exc}")
        return 2
    try:
        payload = json.loads(args.payload.read_text(encoding="utf-8"))
    except OSError as exc:
        print(f"could not read the payload: {exc}")
        return 2
    except json.JSONDecodeError as exc:
        print(f"the payload is not readable JSON: {exc}")
        return 2
    report = check(skill_text, payload)
    print(report.render())
    return 1 if report.findings else 0


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