llm-ingestion-okf/src/llm_ingestion_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 moved into the package on 2026-09-08 (O5).** It lived in `tools/` on the
argument that staying out of the wheel left no consumer's install surface
changed. The generated consumption skill made that argument cost more than it
bought: the skill's own check step named this file by absolute path into a
checkout, so the one command that tells a reader whether their payload
conforms was unreachable from an install. `okf check` is that command, on
PATH. `tools/okf_contract_check.py` remains as a thin wrapper for the
published reproduction blocks.
"""

from __future__ import annotations

import argparse
import json
import re
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")

#: SS 3.1 and SS 3.3, read back out of the skill's own prose. `okf skill` writes
#: the bundle it was generated for in one authored sentence, and this is the
#: pattern that reads it: there is no structured skill model, only its text.
#: Held to the generator by a test, because the two live in different files.
SKILL_IDENTITY = re.compile(r"for one bundle: `([^`<>]+)` at ref\s+`([^`<>]+)`")


def skill_identity(skill_text: str) -> tuple[str, str] | None:
    """The `(bundle_id, ref)` the skill declares, or `None` when it declares
    none a reader could act on. `None` is a finding, never a silent pass: the
    unfilled template's `<CORPUS>` and `<REF>` are placeholders, and the
    template's own rule is that a copy leaving one unfilled is not configured,
    it is unfinished."""
    match = SKILL_IDENTITY.search(skill_text)
    return (match.group(1), match.group(2)) if match else None


#: 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_bundle_identity(ctx: Context) -> list[Finding]:
    """SS 3.1 and SS 3.3: the skill and the payload must name one bundle.

    Added 2026-09-10 on a measurement this checker had published about itself
    since 2026-09-08 and not closed: it reported `conformant, 15 rules, 0
    findings` for a skill generated from one corpus against a payload assembled
    from another, for the unfilled template against that payload, and for a
    payload sharing the skill's `bundle_id` at a foreign `ref`.

    **Both halves are compared, and the `ref` half is the load-bearing one.**
    Three distinct builds on one machine were measured carrying the same
    `bundle_id`, so an id comparison would pass a stale skill -- exactly the
    case the generated skill warns about in its own words. SS 3.3: "a version
    is the producer's assertion; a ref is a fact about bytes".

    **It compares a DECLARED identity against a DECLARED identity** and never
    opens the bundle, so a payload misreporting its own `ref` passes here.
    Proving a ref against bytes is `okf consume --ref`'s job and needs a bundle
    path this command deliberately does not take.

    A payload that declares no identity at all is `rule_bundle_ref`'s defect,
    not this one's: restating it would report one hole twice.
    """
    declared = skill_identity(ctx.skill)
    if declared is None:
        return [
            Finding(
                "bundle_mismatch",
                "the skill declares no readable bundle identity, so no payload "
                "can be shown to belong to it; a `<PLACEHOLDER>` left unfilled "
                "is not an identity, and neither is its absence (SS 3.1, SS 3.3)",
            )
        ]
    skill_id, skill_ref = declared
    bundle = _mapping(ctx.payload.get("bundle"))
    payload_id, payload_ref = _text(bundle.get("bundle_id")), _text(bundle.get("ref"))
    disagreements = [
        f"{key} (skill {mine!r}, payload {theirs!r})"
        for key, mine, theirs in (
            ("bundle_id", skill_id, payload_id),
            ("ref", skill_ref, payload_ref),
        )
        if theirs and theirs != mine
    ]
    if disagreements:
        return [
            Finding(
                "bundle_mismatch",
                "the skill was generated for a bundle the payload does not "
                f"describe: {'; '.join(disagreements)}. Every number in the "
                "skill was measured against its own bundle's bytes (SS 3.1, "
                "SS 3.3)",
            )
        ]
    return [
        Finding(
            "bundle_mismatch",
            f"excerpt {position} names bundle {found!r}, which is not the "
            f"payload's {payload_id!r}; identity across bundles is the "
            "(bundle_id, concept_id) tuple, so an excerpt naming another "
            "bundle is another bundle's excerpt (SS 3.1)",
        )
        for position, raw in enumerate(_sequence(ctx.payload.get("excerpts")))
        if payload_id and (found := _text(_mapping(raw).get("bundle_id"))) and found != payload_id
    ]


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_named(ctx: Context) -> list[Finding]:
    """SS 8: every excerpt carries a `title`.

    Added 2026-09-08 on a measurement, not a preference: `portfolio-optimiser`
    ran three paid arms in which the pre-pass delivered the gold concept at rank
    1 of 8 on 3 of 3 bundles and the model answered correctly on 1 of 3, because
    the excerpt carried `concept_id` and `text` and nothing a reader could name
    the document by. A payload no answer can cite from is not conformant; the
    identity fields are what SS 3.1's tuple is FOR.
    """
    if not ctx.payload_is_mapping:
        return []
    return [
        Finding(
            "excerpt_unnamed",
            f"excerpt {position} carries no 'title'; an excerpt a reader cannot "
            "name is one an answer cannot cite, whatever its rank (SS 8)",
        )
        for position, raw in enumerate(_sequence(ctx.payload.get("excerpts")))
        if not _text(_mapping(raw).get("title"))
    ]


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_bundle_identity,
    rule_excerpt_source_marking,
    rule_excerpt_named,
    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())