llm-ingestion-okf/tests/test_contract_check.py

"""The consumption contract, checked rather than described.

`docs/consumption-contract.md` states what a consumption skill and its pre-pass
payload MUST carry. Prose has no test, and this repository has already paid for
that once: a published promise without a test goes false without anyone
noticing. So the contract's mechanically checkable half is checked here.

Two disciplines the checker itself is held to:

- **A known-positive, shipped.** The template and its example payload must
  PASS. Without that arm a checker that refuses everything is green on every
  negative case -- the harness-lies-red failure, one level down.
- **One mutation per arm, each with its own code.** A single "invalid" verdict
  over fourteen different defects is a diagnostic no caller can act on, so each
  mutation asserts the code it produces, not merely that something failed.

The last test is the anti-drift gate: every literal the checker enforces must
appear in the contract document. Two copies of a closed set drift, and the
copy nobody reads is the one that goes wrong.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT / "tools"))

import okf_contract_check  # noqa: E402
import okf_skill  # noqa: E402

TEMPLATE = PROJECT_ROOT / "skills" / "okf-consume-template" / "SKILL.md"
EXAMPLE = TEMPLATE.parent / "references" / "example-payload.json"
CONTRACT = PROJECT_ROOT / "docs" / "consumption-contract.md"


def load_example() -> dict[str, Any]:
    payload: dict[str, Any] = json.loads(EXAMPLE.read_text(encoding="utf-8"))
    return payload


def declaring(skill_text: str, payload: dict[str, Any]) -> str:
    """The template with the one hole `bundle_mismatch` reads filled in.

    The template alone can no longer be a conformant skill: `<CORPUS>` and
    `<REF>` are placeholders, and an identity the checker cannot read is a
    finding by design. The sentence is the generator's own, taken from it
    rather than copied, so the two cannot drift apart.
    """
    bundle = payload["bundle"]
    return skill_text.replace(
        okf_skill.TEMPLATE_HEADER,
        okf_skill.identity_line(bundle["bundle_id"], bundle["ref"]) + ".",
        1,
    )


def instantiated() -> str:
    return declaring(TEMPLATE.read_text(encoding="utf-8"), load_example())


def instantiated_file(tmp_path: Path) -> Path:
    target = tmp_path / "SKILL.md"
    target.write_text(instantiated(), encoding="utf-8")
    return target


def codes(skill_text: str, payload: Any) -> list[str]:
    return [finding.code for finding in okf_contract_check.check(skill_text, payload).findings]


# --- The known-positive, which every negative arm depends on ----------------


def test_an_instantiated_skill_and_its_own_example_payload_are_conformant() -> None:
    """The CONTROL. A checker that cannot pass anything proves nothing below.

    It is the INSTANTIATED skill, not the template: since 2026-09-10 a skill
    declaring no readable bundle identity is `bundle_mismatch`, measured on
    the template itself in `test_bundle_identity.py`."""
    report = okf_contract_check.check(instantiated(), load_example())
    assert report.findings == ()


def test_the_report_carries_its_denominators() -> None:
    """ "Conformant" without a denominator is unmeasured, per the contract's own
    section 5. The report says how many rules ran over how many units."""
    report = okf_contract_check.check(instantiated(), load_example())
    assert report.rules_evaluated == len(okf_contract_check.RULES)
    assert report.rules_evaluated > 0
    assert report.excerpts_examined == 3
    assert report.withheld_examined == 2
    assert report.withheld_total == 2
    rendered = report.render()
    assert str(report.rules_evaluated) in rendered
    assert "3" in rendered and "2" in rendered


# --- Payload mutations, one per arm -----------------------------------------


def test_broken_denominator_identity_is_named() -> None:
    payload = load_example()
    payload["denominators"]["considered"] = 6
    assert "denominator_identity" in codes(instantiated(), payload)


def test_delivered_count_disagreeing_with_the_list_is_named() -> None:
    payload = load_example()
    payload["excerpts"] = payload["excerpts"][:2]
    assert "denominator_list_mismatch" in codes(instantiated(), payload)


def test_a_missing_adjudication_key_is_not_absence() -> None:
    """The contract's sharpest rule: `unknown` is written, never omitted."""
    payload = load_example()
    del payload["excerpts"][0]["adjudication"]
    assert "state_missing" in codes(instantiated(), payload)


def test_an_adjudication_value_outside_the_closed_set_is_named() -> None:
    payload = load_example()
    payload["excerpts"][0]["adjudication"] = "absent"
    assert "state_not_in_set" in codes(instantiated(), payload)


def test_a_trust_tier_outside_the_closed_set_is_named() -> None:
    payload = load_example()
    payload["excerpts"][1]["trust_tier"] = "trusted"
    assert "state_not_in_set" in codes(instantiated(), payload)


@pytest.mark.parametrize("key", ["bundle_id", "concept_id", "sha256"])
def test_an_excerpt_missing_its_source_marking_is_named(key: str) -> None:
    payload = load_example()
    del payload["excerpts"][0][key]
    assert "source_marking_incomplete" in codes(instantiated(), payload)


def test_a_bundle_without_a_ref_is_named() -> None:
    payload = load_example()
    payload["bundle"]["ref"] = ""
    assert "ref_missing" in codes(instantiated(), payload)


def test_a_withheld_concept_without_a_rule_is_an_undeclared_cut() -> None:
    payload = load_example()
    del payload["withheld"]["nearest"][0]["rule"]
    assert "cut_undeclared" in codes(instantiated(), payload)


def test_spending_over_the_limit_is_the_gate_firing() -> None:
    payload = load_example()
    payload["budget"]["spent"] = payload["budget"]["limit"] + 1
    assert "budget_exceeded" in codes(instantiated(), payload)


def test_an_instrument_that_missed_its_known_positive_is_named() -> None:
    payload = load_example()
    payload["budget"]["known_positive"]["measured"] = 10405
    assert "instrument_unvalidated" in codes(instantiated(), payload)


def test_a_budget_without_an_instrument_is_named() -> None:
    payload = load_example()
    payload["budget"]["instrument"] = ""
    assert "budget_undeclared" in codes(instantiated(), payload)


def test_a_payload_naming_no_contract_revision_is_named() -> None:
    payload = load_example()
    del payload["contract"]
    assert "contract_unversioned" in codes(instantiated(), payload)


def test_a_payload_that_is_not_a_mapping_is_refused_not_crashed() -> None:
    assert "payload_invalid" in codes(instantiated(), ["not", "a", "mapping"])


# --- Skill mutations --------------------------------------------------------


def test_a_missing_required_section_is_named() -> None:
    text = instantiated().replace("## Denominators", "## Counts")
    assert "skill_section_missing" in codes(text, load_example())


def test_a_missing_required_marking_is_named() -> None:
    text = instantiated().replace("[sourced-not-sufficient]", "[thin]")
    assert "skill_marking_missing" in codes(text, load_example())


def test_a_translated_marking_literal_does_not_count() -> None:
    """One literal string, no variants, no translations."""
    text = instantiated().replace("[unverifiable-from-bundle]", "[ikke-verifiserbar-fra-bundle]")
    assert "skill_marking_missing" in codes(text, load_example())


def test_a_skill_omitting_an_adjudication_value_is_named() -> None:
    text = instantiated().replace("`unknown`", "`missing`")
    assert "skill_state_missing" in codes(text, load_example())


# --- The command line -------------------------------------------------------


def test_the_command_exits_zero_on_an_instantiated_pair(tmp_path: Path) -> None:
    exit_code = okf_contract_check.main(
        ["--skill", str(instantiated_file(tmp_path)), "--payload", str(EXAMPLE)]
    )
    assert exit_code == 0


def test_the_command_exits_one_on_a_non_conformant_payload(tmp_path: Path) -> None:
    payload = load_example()
    payload["denominators"]["considered"] = 6
    broken = tmp_path / "payload.json"
    broken.write_text(json.dumps(payload), encoding="utf-8")
    assert (
        okf_contract_check.main(
            ["--skill", str(instantiated_file(tmp_path)), "--payload", str(broken)]
        )
        == 1
    )


def test_the_command_separates_could_not_run_from_non_conformant(tmp_path: Path) -> None:
    """Exit 2, never 1: "the check did not run" and "the check failed" are
    different outcomes, and collapsing them is the fourth face of the
    verification law."""
    missing = tmp_path / "absent.json"
    assert okf_contract_check.main(["--skill", str(TEMPLATE), "--payload", str(missing)]) == 2
    unreadable = tmp_path / "bad.json"
    unreadable.write_text("{not json", encoding="utf-8")
    assert okf_contract_check.main(["--skill", str(TEMPLATE), "--payload", str(unreadable)]) == 2


def test_the_module_runs_as_a_script(tmp_path: Path) -> None:
    result = subprocess.run(
        [
            sys.executable,
            str(PROJECT_ROOT / "tools" / "okf_contract_check.py"),
            "--skill",
            str(instantiated_file(tmp_path)),
            "--payload",
            str(EXAMPLE),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0, result.stderr
    assert "conformant" in result.stdout


# --- The anti-drift gate ----------------------------------------------------


def test_every_literal_the_checker_enforces_is_in_the_contract() -> None:
    """The checker and the document state one closed set each. If they can
    disagree, the copy nobody reads is the one that goes wrong."""
    contract = CONTRACT.read_text(encoding="utf-8")
    for literal in (
        *okf_contract_check.REQUIRED_MARKINGS,
        *okf_contract_check.ADJUDICATION_STATES,
        *okf_contract_check.TRUST_TIERS,
    ):
        assert literal in contract, literal
    for heading in okf_contract_check.REQUIRED_SECTIONS:
        assert heading.lower() in contract.lower(), heading


def test_the_contract_document_is_reachable_from_the_template() -> None:
    assert "docs/consumption-contract.md" in TEMPLATE.read_text(encoding="utf-8")