llm-ingestion-okf/tests/test_bundle_identity.py

"""Which bundle was this skill generated for? The checker had no rule for it.

Measured 2026-09-10 on this repository's HEAD, before this file existed: `okf
check` reported `conformant: 15 rules over 8 excerpts and 438 withheld entries,
0 findings` for a skill generated from one corpus against a payload assembled
from a different one -- and the same line for the UNFILLED template against
that payload, and the same line again when only the `ref` was foreign. Three
forms, one hole: nothing compared the identity the skill declares with the
identity the payload declares.

**The `ref` half is what makes this a rule rather than an id check.** Three
distinct builds on this machine carry the same `bundle_id`, so `bundle_id` does
not identify the bytes. The contract says so in SS 3.3 -- "a version is the
producer's assertion; a ref is a fact about bytes" -- and the generated skill
says it about itself: "If the bundle moves, the ref moves with it and this file
is stale -- regenerate". A rule comparing ids alone would pass a stale skill.

**An identity the rule cannot read is a finding, never a silent pass.** That is
the template arm: `<CORPUS>` and `<REF>` left in place are not an identity, and
the template's own rule says a copy leaving a placeholder unfilled "is not
configured, it is unfinished".

Two bundles already tracked here supply the arms, so no fixture carries a
sentence from any corpus: the segmented golden bundle and the provenance
fixture differ in `bundle_id` and in `ref`.

**What this rule does NOT do**, stated here rather than implied: it compares a
DECLARED identity against a DECLARED identity and never opens the bundle. A
payload that misreports its own `ref` passes. Proving a ref against bytes is
`okf consume --ref`'s job and needs a bundle path this checker deliberately
does not take.
"""

from __future__ import annotations

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

import pytest

from llm_ingestion_okf import contract_check, skill

PROJECT_ROOT = Path(__file__).resolve().parents[1]
TEMPLATE = PROJECT_ROOT / "skills" / "okf-consume-template" / "SKILL.md"

#: Two bundles that differ in both halves of the identity. One bundle would let
#: an id comparison masquerade as an identity comparison.
BUNDLE_A = PROJECT_ROOT / "examples" / "ingest-golden-segmented-okf-v0-2" / "expected-bundle"
BUNDLE_B = PROJECT_ROOT / "tests" / "fixtures" / "consume-provenance"


def instantiate(bundle: Path) -> tuple[str, dict[str, Any]]:
    """One generated skill and the payload generated beside it, in memory."""
    text, payload = skill.render(bundle, out=bundle.parent / "unwritten")
    return text, dict(payload)


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


def run(tmp_path: Path, skill_text: str, payload: Any) -> int:
    """The exit code, taken from `main` directly. A pipe would report the
    status of the last command in it, which is not the checker's."""
    skill_file = tmp_path / "SKILL.md"
    payload_file = tmp_path / "payload.json"
    skill_file.write_text(skill_text, encoding="utf-8")
    payload_file.write_text(json.dumps(payload), encoding="utf-8")
    return contract_check.main(["--skill", str(skill_file), "--payload", str(payload_file)])


# --- The control, which every arm below depends on ---------------------------


def test_the_two_bundles_declare_different_identities() -> None:
    """The known-positive for the arms: if these agreed, every mismatch below
    would be measuring nothing."""
    _, payload_a = instantiate(BUNDLE_A)
    _, payload_b = instantiate(BUNDLE_B)
    assert payload_a["bundle"]["bundle_id"] != payload_b["bundle"]["bundle_id"]
    assert payload_a["bundle"]["ref"] != payload_b["bundle"]["ref"]


def test_a_skill_and_its_own_payload_stay_conformant(tmp_path: Path) -> None:
    """KNOWN-POSITIVE. A rule that refuses everything is green on every
    negative arm above it."""
    text, payload = instantiate(BUNDLE_A)
    assert contract_check.check(text, payload).findings == ()
    assert run(tmp_path, text, payload) == 0


# --- The three measured forms ------------------------------------------------


def test_a_skill_for_another_bundle_is_a_finding(tmp_path: Path) -> None:
    """Row A: different `bundle_id` AND different `ref`."""
    text, _ = instantiate(BUNDLE_A)
    _, payload = instantiate(BUNDLE_B)
    assert codes(text, payload) == ["bundle_mismatch"]
    assert run(tmp_path, text, payload) == 1


def test_the_same_id_at_another_ref_is_a_finding(tmp_path: Path) -> None:
    """Row C, the arm that separates a whole rule from half of one: the
    `bundle_id` agrees and only the `ref` is foreign."""
    text, payload = instantiate(BUNDLE_A)
    _, other = instantiate(BUNDLE_B)
    payload["bundle"] = dict(payload["bundle"], ref=other["bundle"]["ref"])
    assert codes(text, payload) == ["bundle_mismatch"]
    assert run(tmp_path, text, payload) == 1


def test_an_unfilled_template_is_a_finding(tmp_path: Path) -> None:
    """Row B: `<CORPUS>` and `<REF>` are not an identity. An identity the rule
    cannot read must fail, or the template passes again and the rule is a
    comparison nobody reaches."""
    _, payload = instantiate(BUNDLE_A)
    template = TEMPLATE.read_text(encoding="utf-8")
    assert "<CORPUS>" in template and "<REF>" in template
    assert "bundle_mismatch" in codes(template, payload)
    assert run(tmp_path, template, payload) == 1


# --- The same question one level down ----------------------------------------


def test_an_excerpt_from_another_bundle_is_a_finding() -> None:
    """SS 3.1's identity is the (bundle_id, concept_id) tuple, so an excerpt
    naming a bundle the payload does not is the same defect one level down.
    Measured on the two real payloads reproduced this round: 0 of 16 excerpts
    disagreed, so this arm is built rather than found."""
    text, payload = instantiate(BUNDLE_A)
    excerpts = [dict(excerpt) for excerpt in payload["excerpts"]]
    assert excerpts, "the known-positive: no excerpts would make the zero meaningless"
    excerpts[0]["bundle_id"] = "some-other-bundle"
    payload["excerpts"] = excerpts
    assert codes(text, payload) == ["bundle_mismatch"]


# --- The rule does not reach past what it is given ----------------------------


def test_a_payload_declaring_no_identity_is_left_to_its_own_rule(tmp_path: Path) -> None:
    """No rule short-circuits another, and none restates another either: a
    payload carrying no `bundle` is `ref_missing`'s defect, not this rule's.
    The count is the known-negative's, written down rather than assumed."""
    text, _ = instantiate(BUNDLE_A)
    found = codes(text, {})
    assert "bundle_mismatch" not in found
    assert "ref_missing" in found
    assert len(found) == 9
    assert run(tmp_path, text, {}) == 1


def test_the_rule_count_is_the_denominator_the_report_quotes() -> None:
    """The nevner moves 15 -> 16, and every consumer quoting "15 rules" is
    quoting a number that has changed."""
    text, payload = instantiate(BUNDLE_A)
    report = contract_check.check(text, payload)
    assert report.rules_evaluated == len(contract_check.RULES) == 16
    assert "16 rules" in report.render()


def test_the_generator_writes_an_identity_the_checker_can_read() -> None:
    """The two halves are in different files, so the coupling gets a test: the
    sentence `okf skill` writes is the sentence the rule parses."""
    for bundle in (BUNDLE_A, BUNDLE_B):
        text, payload = instantiate(bundle)
        read = contract_check.skill_identity(text)
        assert read == (payload["bundle"]["bundle_id"], payload["bundle"]["ref"])


@pytest.mark.parametrize("bundle", (BUNDLE_A, BUNDLE_B), ids=lambda path: path.name)
def test_the_console_script_reports_the_mismatch(bundle: Path, tmp_path: Path) -> None:
    """Through the installed entry point, because that is what a consumer
    runs, and its exit code is read directly rather than through a pipe."""
    text, _ = instantiate(bundle)
    other = BUNDLE_B if bundle == BUNDLE_A else BUNDLE_A
    _, payload = instantiate(other)
    skill_file = tmp_path / "SKILL.md"
    payload_file = tmp_path / "payload.json"
    skill_file.write_text(text, encoding="utf-8")
    payload_file.write_text(json.dumps(payload), encoding="utf-8")
    result = subprocess.run(
        [
            sys.executable,
            str(PROJECT_ROOT / "tools" / "okf_contract_check.py"),
            "--skill",
            str(skill_file),
            "--payload",
            str(payload_file),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 1, result.stdout
    assert "bundle_mismatch" in result.stdout