"""The consumption pre-pass, checked rather than described. `tools/okf_consume.py` cuts an OKF bundle to one contract-conformant payload for one question. Three disciplines this suite is held to, all of them the house pattern rather than new inventions: - **Every zero carries a control.** A count of nothing is a measurement whose query must first be shown capable of finding. The placeholder scan runs against the template (known-positive) before its zero on the filled copy is believed; the index walk is controlled against the `rglob` the contract forbids the consumer path from using; the socket guard is fired directly before its silence during a real run counts as evidence. - **One mutation per rule.** `tests/test_contract_check.py` establishes the shape: assert the code a defect produces, never merely that something failed. - **The corpus is never a test dependency.** K2 lives outside the repository. Every test here runs against `examples/.../expected-bundle` (3 concepts) or `tests/fixtures/consume-bundle` (synthetic, carrying the states the real corpus has zero of). The corpus-conditional arm skips with its denominator named, so a skip cannot read as a pass. """ from __future__ import annotations import hashlib import json import os import sys import unicodedata from pathlib import Path import pytest PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT / "tools")) import okf_consume # noqa: E402 import okf_contract_check # noqa: E402 from llm_ingestion_okf.materialize import parse_frontmatter # noqa: E402 TEMPLATE = PROJECT_ROOT / "skills" / "okf-consume-template" / "SKILL.md" GOLDEN = PROJECT_ROOT / "examples" / "ingest-golden-segmented-okf-v0-2" / "expected-bundle" # --- Step 1: the walk and the ref --------------------------------------------- def test_the_index_walk_finds_every_concept_and_no_index_or_log() -> None: found = okf_consume.enumerate_concepts(GOLDEN) assert found == ( "krav/1-1/foerste-krav", "krav/1-2/andre-krav", "veiledning", ) assert not any(concept.endswith("index") or concept.endswith("log") for concept in found) def test_the_index_walk_is_complete_against_the_method_the_contract_forbids() -> None: # SS 9.2 forbids the CONSUMER path from enumerating a directory. Using the # forbidden method here, in a test, is what proves the permitted one loses # nothing -- an absence with no control is not a measurement. by_rglob = { path.relative_to(GOLDEN).with_suffix("").as_posix() for path in GOLDEN.rglob("*.md") if path.name not in ("index.md", "log.md") } assert by_rglob, "the control found nothing, so it cannot certify the walk" assert set(okf_consume.enumerate_concepts(GOLDEN)) == by_rglob def test_a_concept_reachable_only_through_a_nested_index_is_still_found() -> None: # `krav/1-1/foerste-krav` is three levels down and named in no root entry. root_entries = (GOLDEN / "index.md").read_text(encoding="utf-8") assert "foerste-krav" not in root_entries, "the fixture no longer exercises nesting" assert "krav/1-1/foerste-krav" in okf_consume.enumerate_concepts(GOLDEN) def test_the_ref_is_stable_across_calls_and_names_its_algorithm() -> None: first = okf_consume.bundle_ref(GOLDEN) assert first == okf_consume.bundle_ref(GOLDEN) assert first.startswith("sha256-tree:") def test_the_ref_moves_when_one_concept_byte_moves(tmp_path: Path) -> None: copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) target = copy / "veiledning.md" target.write_text(target.read_text(encoding="utf-8") + "x", encoding="utf-8") assert okf_consume.bundle_ref(copy) != before def test_the_ref_does_not_move_when_only_mtimes_move(tmp_path: Path) -> None: copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) for path in sorted(copy.rglob("*")): if path.is_file(): os.utime(path, (0, 0)) assert okf_consume.bundle_ref(copy) == before def _copy_bundle(source: Path, target: Path) -> None: for path in sorted(source.rglob("*")): if path.is_file(): destination = target / path.relative_to(source) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(path.read_bytes()) def test_the_ref_covers_the_indexes_too_since_the_walk_reads_them(tmp_path: Path) -> None: # The docstring claims every byte that can reach a payload is inside the # ref. An index byte can: it decides which concepts are reachable at all. copy = tmp_path / "bundle" _copy_bundle(GOLDEN, copy) before = okf_consume.bundle_ref(copy) nested = copy / "krav" / "1-1" / "index.md" nested.write_text(nested.read_text(encoding="utf-8") + "\nfritekst\n", encoding="utf-8") assert okf_consume.bundle_ref(copy) != before # --- Step 2: one concept, read into a record ---------------------------------- PROPOSED_CONCEPT = "krav/1-1/foerste-krav" ROOT_BUNDLE_ID = "b-golden-segmented-okf-v0-2" def _read(concept_id: str, root: Path = GOLDEN) -> okf_consume.Concept: return okf_consume.read_concept( root / f"{concept_id}.md", bundle_root=root, root_bundle_id=ROOT_BUNDLE_ID ) def test_a_concept_carrying_adjudication_reads_that_value() -> None: assert _read(PROPOSED_CONCEPT).adjudication == "proposed" def test_a_concept_carrying_no_adjudication_key_reads_as_unknown(tmp_path: Path) -> None: # SS 6.1: `unknown` is written EXPLICITLY. "Not judged" and "we cannot tell # whether it was judged" are different facts, and only one is about the # concept. root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace("adjudication: proposed\n", ""), encoding="utf-8", ) concept = _read(PROPOSED_CONCEPT, root) assert concept.adjudication == "unknown" assert concept.adjudication_present is False def test_an_adjudication_value_outside_the_wire_set_is_refused_by_name(tmp_path: Path) -> None: # Mapping an unrecognised value to `unknown` would report "we cannot tell" # where the truth is "the bundle said something this consumer does not # understand" -- a defect laundered into a state. root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace("adjudication: proposed", "adjudication: seen"), encoding="utf-8", ) with pytest.raises(okf_consume.ConsumeError) as raised: _read(PROPOSED_CONCEPT, root) assert raised.value.code == "adjudication_unknown_value" def test_the_digest_is_of_the_concept_file_and_is_not_the_source_sha256() -> None: concept = _read(PROPOSED_CONCEPT) on_disk = hashlib.sha256((GOLDEN / f"{PROPOSED_CONCEPT}.md").read_bytes()).hexdigest() assert concept.sha256 == on_disk assert len(concept.sha256) == 64 frontmatter = parse_frontmatter(GOLDEN / f"{PROPOSED_CONCEPT}.md") assert frontmatter["source_sha256"] != concept.sha256 def test_the_concept_id_keeps_its_slashes_where_import_slug_would_flatten_them() -> None: # `importer.import_slug` flattens one line below the rule this id follows. # A flattened id fails a document-prefix match in a way that looks like a # ranking miss rather than an id-format bug. assert _read(PROPOSED_CONCEPT).concept_id == "krav/1-1/foerste-krav" def test_bundle_id_falls_back_to_the_root_index_and_says_that_it_did(tmp_path: Path) -> None: root = tmp_path / "bundle" _copy_bundle(GOLDEN, root) target = root / f"{PROPOSED_CONCEPT}.md" target.write_text( target.read_text(encoding="utf-8").replace(f"bundle_id: {ROOT_BUNDLE_ID}\n", ""), encoding="utf-8", ) concept = _read(PROPOSED_CONCEPT, root) assert concept.bundle_id == ROOT_BUNDLE_ID assert concept.bundle_id_inherited is True assert _read(PROPOSED_CONCEPT).bundle_id_inherited is False # --- Step 3: trust_tier, and the refusal to tier what cannot be read ---------- FIXTURE = PROJECT_ROOT / "tests" / "fixtures" / "consume-bundle" def test_the_fixture_bundle_carries_what_the_real_corpus_has_none_of() -> None: # The control on every assertion below. K2 has 0 `verified:` keys, 0 # `type: verdict` and 0 `adjudication: adjudicated` over 629 concepts, so a # fixture missing any of them would make its tests pass over an empty set. text = "\n".join(path.read_text(encoding="utf-8") for path in sorted(FIXTURE.rglob("*.md"))) assert "type: verdict" in text assert "adjudication: adjudicated" in text assert "by: human:" in text assert "by: process:" in text def test_no_verified_key_reads_as_unverified() -> None: # SS 6.3: a concept carrying no trust frontmatter is still consumable. assert okf_consume.trust_tier(None) == "unverified" def test_a_human_actor_reads_as_human_reviewed() -> None: assert okf_consume.trust_tier("[{ by: human:ktg, at: 2026-09-01T00:00:00Z }]") == ( "human-reviewed" ) def test_a_process_actor_reads_as_machine_confirmed() -> None: assert okf_consume.trust_tier("[{ by: process:okf-check, at: 2026-09-01T00:00:00Z }]") == ( "machine-confirmed" ) def test_the_human_test_is_a_prefix_and_never_a_substring() -> None: # `bot/human:2` is a MACHINE actor whose id contains the string `human:`. # A substring test would promote it to the highest tier -- fabricated # provenance produced by a matching bug. assert okf_consume.trust_tier("[{ by: bot/human:2 }]") == "machine-confirmed" def test_an_entry_naming_no_actor_is_refused_rather_than_tiered() -> None: with pytest.raises(okf_consume.ConsumeError) as raised: okf_consume.trust_tier("[{ at: 2026-09-01T00:00:00Z }]") assert raised.value.code == "verified_actorless" def test_a_block_form_verified_is_not_a_tier_at_all() -> None: # Measured 2026-09-07: this library's line-oriented `parse_frontmatter` # returns `''` for a block-form `verified:` and the full string for a flow # one, so PRESENT-BUT-UNREADABLE is distinguishable from ABSENT. Emitting # `unverified` here would assert a fact nobody measured (SS 6.4). assert okf_consume.trust_tier("") is None def test_the_block_form_case_is_real_in_the_fixture_and_not_only_in_the_unit_test() -> None: # The control: without this, `trust_tier("") is None` could be true of a # string no bundle ever produces. frontmatter = parse_frontmatter(FIXTURE / "dyp" / "nivaa" / "blokkform-verifisert.md") assert frontmatter["verified"] == "" assert okf_consume.trust_tier(frontmatter["verified"]) is None # --- Step 4: the budget instrument ------------------------------------------- CONTRACT = PROJECT_ROOT / "docs" / "consumption-contract.md" def test_measure_counts_bytes_and_not_characters() -> None: # The exact conflation the brief records itself making once: a chars/token # ratio quoted where a bytes/token one was needed. `æøå` is three # characters and six bytes, and the two only differ outside ASCII. assert okf_consume.measure("æøå") == len('"æøå"'.encode()) assert okf_consume.measure("æøå") != len("æøå") def test_measure_counts_the_encoded_form_the_payload_actually_costs() -> None: # `json.dumps` defaults to `ensure_ascii=True`, which inflates this corpus # by 7.1 %. A gate measuring one form while the knapsack weighs the other # disagrees by more than the headroom. norwegian = "årlig kontroll av anlegget" assert okf_consume.measure(norwegian) == len( json.dumps(norwegian, ensure_ascii=False).encode("utf-8") ) assert okf_consume.measure(norwegian) < len( json.dumps(norwegian, ensure_ascii=True).encode("utf-8") ) def test_the_known_positive_is_reproduced_by_the_gates_own_instrument() -> None: case, expected, measured = okf_consume.known_positive() assert case assert expected == measured, "SS 7.4: the instrument has not been shown to count" def test_the_known_positive_is_not_the_raw_byte_count_of_the_same_file() -> None: # Validating one instrument while gating with another is the SS 7.4 failure # the rule exists to prevent. The delta is derivable by a second, wholly # independent route (`wc -c`) and moves the moment `measure` changes what it # counts -- which is what keeps `expected == measured` from being vacuous. _, expected, _ = okf_consume.known_positive() raw = len(CONTRACT.read_bytes()) assert expected != raw assert expected - raw == okf_consume.KNOWN_POSITIVE_ENCODING_DELTA def test_the_default_limit_admits_a_concept_the_size_of_the_price_form() -> None: # Measured during planning: at the drafted 60 000 B default the SC6 gold # concept (101 313 B encoded) falls to the "cannot fit alone" pre-exclusion, # so SC1 and SC6 were mutually unsatisfiable on a CORRECT implementation. assert okf_consume.DEFAULT_LIMIT >= 101_313 def test_the_budget_unit_and_instrument_are_named_rather_than_implied() -> None: assert "byte" in okf_consume.BUDGET_UNIT assert "ensure_ascii=False" in okf_consume.BUDGET_INSTRUMENT # --- Step 5: stage-one document ranking -------------------------------------- def test_normalise_is_nfc_stable_on_the_one_letter_that_decomposes() -> None: # `NFD("å")` is `a` + U+030A, and the combining ring is not `\w`, so an # un-normalised split returns `["a", "rlig"]`. `æ` and `ø` have NO canonical # decomposition, so a test built on `miljø` passes while the bug is live -- # the known-positive here MUST use `å`. composed = unicodedata.normalize("NFC", "årlig kontroll") decomposed = unicodedata.normalize("NFD", "årlig kontroll") assert composed != decomposed, "the control is broken: the two forms are identical" assert okf_consume.normalise(decomposed) == okf_consume.normalise(composed) assert "årlig" in okf_consume.normalise(decomposed) def test_normalise_drops_tokens_under_three_characters() -> None: assert okf_consume.normalise("er en pris i et skjema") == ("pris", "skjema") def test_two_tokens_match_on_a_shared_prefix_of_four_and_not_of_three() -> None: # "Stem-substring" is not an implementable rule: neither `prisene` nor # `prissammenstilling` contains the other. Shared prefix does the work -- # `pris|ene` and `pris|sammenstilling` share 4. A 3-character floor # over-matches Norwegian function words. assert okf_consume.tokens_match("prisene", "prissammenstilling") assert okf_consume.tokens_match("kontrollen", "kontroll") assert not okf_consume.tokens_match("pris", "pri") assert not okf_consume.tokens_match("krav", "kraft") def test_a_question_naming_a_directorys_subject_ranks_that_directory_first() -> None: scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") assert scores, "no document scored, so 'ranks first' would measure nothing" assert max(scores, key=lambda key: (scores[key], key)) == "krav" def test_a_question_about_a_different_subject_ranks_a_different_directory() -> None: # The control on the test above: without it, a scorer returning "krav" # unconditionally would pass. scores = okf_consume.document_scores(FIXTURE, "Hva er omfanget og formaalet?") assert max(scores, key=lambda key: (scores[key], key)) == "scope" def test_curated_prose_in_an_index_is_ignored_rather_than_scored(tmp_path: Path) -> None: root = tmp_path / "bundle" _copy_bundle(FIXTURE, root) index = root / "krav" / "index.md" index.write_text( "Denne mappen handler om priser og prissammenstilling.\n\n" + index.read_text(encoding="utf-8"), encoding="utf-8", ) assert ( okf_consume.DEFAULT_PROFILE.index.parse_entry( "Denne mappen handler om priser og prissammenstilling." ) is None ) assert okf_consume.document_scores(root, "Hvordan skal prisene fylles ut?") == ( okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") ) def test_document_scores_are_identical_across_two_calls() -> None: question = "Hvordan skal prisene fylles ut?" assert okf_consume.document_scores(FIXTURE, question) == okf_consume.document_scores( FIXTURE, question ) # --- Step 6: stage-two concept ranking, fused by RRF -------------------------- def _fixture_concepts() -> list[okf_consume.Concept]: return [ okf_consume.read_concept( FIXTURE / f"{concept_id}.md", bundle_root=FIXTURE, root_bundle_id="consume-fixture", ) for concept_id in okf_consume.enumerate_concepts(FIXTURE) ] def test_concepts_tying_on_every_signal_come_back_in_concept_id_order() -> None: concepts = _fixture_concepts() # A question matching nothing makes every signal identical, so the ONLY # thing left deciding the order is the declared tie-break. ranked = okf_consume.concept_scores(concepts, "zzzz qqqq", {}) ids = [concept.concept_id for concept, _, _ in ranked] assert ids == sorted(ids) def test_reversing_the_input_order_does_not_change_the_output_order() -> None: concepts = _fixture_concepts() forward = [c.concept_id for c, _, _ in okf_consume.concept_scores(concepts, "zzzz qqqq", {})] backward = [ c.concept_id for c, _, _ in okf_consume.concept_scores(list(reversed(concepts)), "zzzz qqqq", {}) ] assert forward == backward def test_a_concept_in_a_high_scoring_document_outranks_an_equally_lexical_one() -> None: concepts = _fixture_concepts() question = "Hvordan skal prisene fylles ut?" lifted = okf_consume.concept_scores(concepts, question, {"krav": 10.0, "dyp": 0.0}) dropped = okf_consume.concept_scores(concepts, question, {"krav": 0.0, "dyp": 10.0}) krav_first = [c.concept_id for c, _, _ in lifted].index("krav/prissammenstilling") krav_later = [c.concept_id for c, _, _ in dropped].index("krav/prissammenstilling") assert krav_first < krav_later def test_the_ranked_order_is_identical_across_two_calls() -> None: concepts = _fixture_concepts() scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") first = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) second = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) assert [c.concept_id for c, _, _ in first] == [c.concept_id for c, _, _ in second] def test_the_price_concept_leads_on_the_price_question_in_the_fixture() -> None: concepts = _fixture_concepts() scores = okf_consume.document_scores(FIXTURE, "Hvordan skal prisene fylles ut?") ranked = okf_consume.concept_scores(concepts, "Hvordan skal prisene fylles ut?", scores) assert ranked[0][0].concept_id == "krav/prissammenstilling" # --- Step 7: the cut ---------------------------------------------------------- def _cut_fixture( question: str = "Hvordan skal prisene fylles ut?", k: int = 8, limit: int | None = None ) -> tuple[list[dict[str, object]], list[tuple[str, str]], int]: concepts = _fixture_concepts() scores = okf_consume.document_scores(FIXTURE, question) ranked = okf_consume.concept_scores(concepts, question, scores) delivered, withheld = okf_consume.cut( ranked, k=k, limit=okf_consume.DEFAULT_LIMIT if limit is None else limit ) return list(delivered), list(withheld), len(ranked) def test_the_fixture_has_both_a_verdict_concept_and_a_delivered_one() -> None: # The control on every count below: neither zero may come from an empty # fixture. delivered, withheld, considered = _cut_fixture() assert delivered, "nothing was delivered, so 'excluded' would measure nothing" assert withheld, "nothing was withheld, so the rules would measure nothing" assert considered == len(okf_consume.enumerate_concepts(FIXTURE)) def test_a_verdict_concept_is_withheld_by_rule_and_reaches_no_excerpt() -> None: delivered, withheld, _ = _cut_fixture() rules = dict(withheld) assert rules["dyp/nivaa/alminnelig-notat"] == "verdict_layer_excluded" assert all(excerpt["concept_id"] != "dyp/nivaa/alminnelig-notat" for excerpt in delivered) def test_the_verdict_exclusion_is_a_type_check_and_never_a_path_filter() -> None: # One question reaching BOTH: a `type: reference` file NAMED # `verdict-lookalike` is delivered, and a `type: verdict` file under an # ordinary name at depth 3 is withheld. A path filter gets both backwards, # and neither half of this is measured unless both concepts match. delivered, withheld, _ = _cut_fixture(question="Hva sier notatet om stifilter og typesjekk?") delivered_ids = {excerpt["concept_id"] for excerpt in delivered} rules = dict(withheld) assert "krav/verdict-lookalike" in delivered_ids assert "dyp/nivaa/alminnelig-notat" not in delivered_ids assert rules["dyp/nivaa/alminnelig-notat"] == "verdict_layer_excluded" assert rules["dyp/nivaa/alminnelig-notat"] != "no_lexical_match" def test_a_capital_l_log_type_does_not_crash_the_reader() -> None: # `type: Log` really occurs in the K2 corpus. Case handling is a test here # rather than an accident. delivered, withheld, _ = _cut_fixture() seen = {excerpt["concept_id"] for excerpt in delivered} | {cid for cid, _ in withheld} assert "krav/loggnotat" in seen def test_a_concept_whose_verified_cannot_be_read_is_withheld_by_name() -> None: # SS 6.2 requires a tier on every excerpt and SS 6.4 forbids reading absence # as negation. Emitting `unverified` for an unreadable value asserts a fact # nobody measured. # The question must REACH the concept: a relevance drop fires first, and a # `no_lexical_match` here would prove nothing about tiering. _, withheld, _ = _cut_fixture(question="Hva staar i blokkform?") assert dict(withheld)["dyp/nivaa/blokkform-verifisert"] == "verified_unreadable" def test_delivered_and_withheld_partition_the_considered_set() -> None: delivered, withheld, considered = _cut_fixture() delivered_ids = {excerpt["concept_id"] for excerpt in delivered} withheld_ids = {concept_id for concept_id, _ in withheld} assert delivered_ids & withheld_ids == set() assert len(delivered_ids) + len(withheld_ids) == considered assert delivered_ids | withheld_ids == set(okf_consume.enumerate_concepts(FIXTURE)) def test_every_withheld_entry_names_a_rule_from_the_closed_set() -> None: _, withheld, _ = _cut_fixture() assert {rule for _, rule in withheld} <= set(okf_consume.WITHHOLDING_RULES) def test_a_concept_larger_than_the_limit_is_excluded_by_name_before_the_dp() -> None: # Named as a RULE rather than left as a packing artefact: "it did not fit" # and "it could never fit" are different facts about the cut. _, withheld, _ = _cut_fixture(limit=200) rules = {rule for _, rule in withheld} assert "over_budget_alone" in rules assert "over_budget_after_knapsack" not in rules def test_concepts_ranked_beyond_k_are_withheld_as_below_k() -> None: # A question matching TWO concepts, so that capping at one leaves a real # `below_k` drop rather than an empty set. matching = "kontroll av prisene" _, wide, _ = _cut_fixture(question=matching, k=8) assert "below_k" not in {rule for _, rule in wide} _, narrow, _ = _cut_fixture(question=matching, k=1) assert "below_k" in {rule for _, rule in narrow} def test_the_exact_knapsack_beats_greedy_by_density() -> None: # Greedy takes the densest item first and is then unable to fit either of # the two that together are worth more. Greedy-by-density has an unbounded # approximation factor; an exact DP over at most `k` items is microseconds. items = ((10.0, 6), (7.0, 5), (7.0, 5)) chosen = okf_consume.knapsack(items, capacity=10) assert sorted(chosen) == [1, 2] assert sum(items[index][0] for index in chosen) == 14.0 def test_the_knapsack_is_deterministic_over_equal_value_subsets() -> None: items = ((5.0, 5), (5.0, 5), (5.0, 5)) assert okf_consume.knapsack(items, capacity=10) == okf_consume.knapsack(items, capacity=10) def test_excerpts_come_back_in_rank_order_and_carry_that_rank() -> None: # An id-sorted payload would turn "position in the payload" into a # different number from "position in the ranking", and hit@k reads the # second one. delivered, _, _ = _cut_fixture() assert [excerpt["rank"] for excerpt in delivered] == list(range(1, len(delivered) + 1)) assert delivered[0]["concept_id"] == "krav/prissammenstilling" # --- Step 8: the payload ------------------------------------------------------ def _payload( root: Path = FIXTURE, question: str = "Hvordan skal prisene fylles ut?", **kwargs: object ) -> dict[str, object]: return okf_consume.build_payload(root, question=question, **kwargs) # type: ignore[arg-type] def test_the_payload_passes_the_checker_against_the_template_with_zero_findings() -> None: report = okf_contract_check.check(TEMPLATE.read_text(encoding="utf-8"), _payload()) assert report.findings == () def test_the_payload_carries_every_section_eight_member() -> None: payload = _payload() assert payload["contract"] == "okf-consumption/1" assert set(payload) >= { "contract", "bundle", "budget", "denominators", "excerpts", "withheld", } bundle = payload["bundle"] assert isinstance(bundle, dict) assert bundle["bundle_id"] == "consume-fixture" assert str(bundle["ref"]).startswith("sha256-tree:") def test_spent_is_the_cost_of_the_delivered_set_and_not_of_the_whole_payload() -> None: # SS 7.2 verbatim: "what the DELIVERED SET spent by that same instrument". # Measured on K2 at k=8, the whole-payload reading puts a 628-entry # `withheld` list (81 565 B) plus one gold excerpt (101 576 B) against a # 120 000 B limit -- so a CORRECT implementation would exit 1 and fail its # own SC1 and SC6. payload = _payload() budget = payload["budget"] excerpts = payload["excerpts"] assert isinstance(budget, dict) and isinstance(excerpts, list) assert budget["spent"] == sum(okf_consume.excerpt_weight(e) for e in excerpts) def test_spent_moves_when_an_excerpt_moves_and_holds_when_withheld_grows() -> None: # The property that distinguishes SS 7.2's reading from the whole-payload # one, asserted rather than described. matching = "kontroll av prisene" wide = _payload(question=matching, k=8) narrow = _payload(question=matching, k=1) wide_budget, narrow_budget = wide["budget"], narrow["budget"] wide_counts, narrow_counts = wide["denominators"], narrow["denominators"] assert isinstance(wide_budget, dict) and isinstance(narrow_budget, dict) assert isinstance(wide_counts, dict) and isinstance(narrow_counts, dict) assert narrow_counts["withheld"] > wide_counts["withheld"] assert narrow_budget["spent"] < wide_budget["spent"] def test_the_counts_and_the_lists_are_two_statements_of_one_fact() -> None: payload = _payload() counts, excerpts, withheld = payload["denominators"], payload["excerpts"], payload["withheld"] assert isinstance(counts, dict) and isinstance(excerpts, list) and isinstance(withheld, list) assert counts["delivered"] == len(excerpts) assert counts["withheld"] == len(withheld) assert counts["considered"] == counts["delivered"] + counts["withheld"] def test_every_excerpt_digest_recomputes_from_the_named_concepts_bytes() -> None: payload = _payload() excerpts = payload["excerpts"] assert isinstance(excerpts, list) and excerpts for excerpt in excerpts: on_disk = FIXTURE / f"{excerpt['concept_id']}.md" assert excerpt["sha256"] == hashlib.sha256(on_disk.read_bytes()).hexdigest() assert ( excerpt["text_sha256"] == hashlib.sha256(str(excerpt["text"]).encode("utf-8")).hexdigest() ) def test_every_concept_id_keeps_the_slash_import_slug_would_have_removed() -> None: payload = _payload() excerpts = payload["excerpts"] assert isinstance(excerpts, list) assert any("/" in str(excerpt["concept_id"]) for excerpt in excerpts) def test_the_budget_refuses_rather_than_narrowing_when_the_cut_cannot_fit() -> None: # SS 7.3: exceeding the gate is a finding requiring a decision, never # something to retry narrower. with pytest.raises(okf_consume.ConsumeError) as raised: _payload(limit=100) assert raised.value.code == "budget_admits_nothing" def test_the_serialised_payload_is_lf_only_and_ends_in_exactly_one_newline() -> None: text = okf_consume.serialise(_payload()) assert "\r" not in text assert text.endswith("}\n") assert not text.endswith("}\n\n") def test_the_serialised_payload_does_not_escape_norwegian_letters() -> None: # `ensure_ascii=True` inflates this corpus by 7.1 %, which is more than the # headroom the gate leaves. text = okf_consume.serialise(_payload(question="Hvor ofte er den årlige kontrollen?")) assert "\\u00e5" not in text def test_a_question_with_no_answer_returns_a_measured_empty_set_not_a_guess() -> None: # The order's known-negative control, and the reason the `no_lexical_match` # rule exists: a ranker that always returns its top eight scores well on # every positive question and is useless. The emptiness must be POSITIVE -- # every considered concept named in `withheld` under a rule, so the identity # still closes and the skill can say "measured, nothing cleared the bar" # rather than "nothing was found". payload = _payload(question="Hva er reglene for sveising av titan i vakuum?") counts, excerpts, withheld = payload["denominators"], payload["excerpts"], payload["withheld"] assert isinstance(counts, dict) and isinstance(excerpts, list) and isinstance(withheld, list) assert excerpts == [] assert counts["delivered"] == 0 assert ( counts["withheld"] == counts["considered"] == len(okf_consume.enumerate_concepts(FIXTURE)) ) assert {entry["rule"] for entry in withheld} == {"no_lexical_match", "verdict_layer_excluded"} # And the control: the SAME payload builder returns a non-empty set for a # question this bundle does answer, so the zero is a measurement. answered = _payload() answered_excerpts = answered["excerpts"] assert isinstance(answered_excerpts, list) and answered_excerpts def test_the_empty_payload_still_passes_the_checker() -> None: payload = _payload(question="Hva er reglene for sveising av titan i vakuum?") assert okf_contract_check.check(TEMPLATE.read_text(encoding="utf-8"), payload).findings == () # --- Corpus-conditional arms -------------------------------------------------- K2_BUNDLE = Path.home() / "corpora" / "okf-telling-20260829" / "K2-bundle-20260903" K2_CONCEPTS = 629 K2_PROPOSED = 618 K2_KEYLESS = 11 requires_k2 = pytest.mark.skipif( not K2_BUNDLE.is_dir(), reason=( f"the K2 corpus is not present at {K2_BUNDLE}. NOT MEASURED, not zero: " f"this arm covers a denominator of {K2_CONCEPTS} concepts, of which " f"{K2_PROPOSED} carry `adjudication: proposed` and {K2_KEYLESS} carry no " "`adjudication` key at all. A skip here is an unmeasured denominator, " "never a pass." ), ) @requires_k2 def test_the_eleven_keyless_k2_concepts_come_back_unknown_over_a_stated_denominator() -> None: # SS 6.1's third state, on real data rather than on a fixture. The 11 are # asserted as ONE named set: measured, the concepts carrying no # `adjudication` are EXACTLY those carrying no `bundle_id`, so three # independent counts would share one blind spot. root_bundle_id = parse_frontmatter(K2_BUNDLE / "index.md")["bundle_id"] concepts = [ okf_consume.read_concept( K2_BUNDLE / f"{concept_id}.md", bundle_root=K2_BUNDLE, root_bundle_id=root_bundle_id, ) for concept_id in okf_consume.enumerate_concepts(K2_BUNDLE) ] assert len(concepts) == K2_CONCEPTS unknown = {c.concept_id for c in concepts if c.adjudication == "unknown"} inherited = {c.concept_id for c in concepts if c.bundle_id_inherited} proposed = [c for c in concepts if c.adjudication == "proposed"] assert len(proposed) == K2_PROPOSED assert len(unknown) == K2_KEYLESS assert unknown == inherited, "the two sets diverged; the fallback is no longer one fact" assert all(c.bundle_id == root_bundle_id for c in concepts if c.bundle_id_inherited) # `adjudicated` has denominator ZERO on this corpus. Stated, not implied. assert [c for c in concepts if c.adjudication == "adjudicated"] == [] @requires_k2 def test_spent_is_the_delivered_set_where_the_whole_payload_reading_would_refuse() -> None: # The regression guard, with figures RE-MEASURED here rather than carried # from the plan: the plan predicted 101 576 B for this excerpt and 188 758 B # for the payload, both taken before per-line trailing-whitespace stripping # landed. What this build actually produces is recorded instead. payload = okf_consume.build_payload(K2_BUNDLE, question="Hvordan skal prisene fylles ut?") budget, excerpts = payload["budget"], payload["excerpts"] assert isinstance(budget, dict) and isinstance(excerpts, list) whole_payload = len(okf_consume.serialise(payload).encode("utf-8")) assert whole_payload > int(budget["limit"]), ( "the guard measures nothing: the whole payload already fits, so the two " "readings of SS 7.2 cannot be told apart on this case" ) assert int(budget["spent"]) <= int(budget["limit"]) @requires_k2 def test_the_price_form_gold_is_delivered_for_the_price_question() -> None: # SC6. The one question whose gold is confirmed by a signal from outside # this repository: a live model reached that directory unprompted in three # navigation steps on 2026-09-06. payload = okf_consume.build_payload(K2_BUNDLE, question="Hvordan skal prisene fylles ut?") excerpts = payload["excerpts"] assert isinstance(excerpts, list) ids = [str(excerpt["concept_id"]) for excerpt in excerpts] assert "del-ii-bilag-7-prisskjema/prissammenstilling-sheet-1" in ids