open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity
llm-ingestion-okf/tests/test_okf_consume.py
Kjell Tore Guttormsen 17c49fc04b feat(consume): give every excerpt the name and the address an answer must cite 
The pre-pass delivered the right concept and the answer could not name it.
Measured by portfolio-optimiser 2026-09-08 over three paid arms: the gold
concept came back at rank 1 of 8 on 3 of 3 bundles, and the model answered
correctly on 1 of 3, because a delivered excerpt carried `concept_id`, body
text and nothing the document is known by. The previous session measured the
same gap from the other side: the provenance it had just written into every
concept did not reach the payload at all.

`excerpt_for` now carries `title` unconditionally, and `req_number`, the SPEC
5.1 address `sources` and each locator key (`source_pages`, `source_sheet`,
`source_rows`, `source_lines`, `source_offset`) when the concept has them. A key
the producer did not write stays absent: an empty value would assert that they
wrote an empty one, which is the contract's 6.4 failure.

`sources` is read in BOTH YAML forms, on a measurement rather than a taste. K2
writes the flow form on 629 of 629 concepts; the largest N-bundle writes the
block form on 270 of 270 and carries no locator key at all, so a flow-only
reader delivers that bundle with no address whatsoever. Reading the block form
is not a licence to write it - the emission rule is untouched, because the
line-oriented parser still cannot round-trip a block list. A `sources` value
this reader cannot decode is named (`sources_unreadable`), never dropped into
the same silence as an absent one.

Contract 8 gains the requirement and the checker gains its code
(`excerpt_unnamed`, 15 rules now, was 14): an excerpt a reader cannot name is
one an answer cannot cite, whatever its rank. `req_number`, `sources` and the
locators are SHOULD, not MUST - they are conditional on the producer, and a
bundle whose concepts carry no identifier cannot deliver one.

K2 controls, same question and same k, before against a frozen copy of the tool
at b6a8c8b: the RANKING does not move - the same 8 concept ids in the same
order, identical `text_sha256`, identical `withheld`, identical denominators
(629 = 621 + 8). The FIELD is what moved: payload 108 877 -> 111 744 B
(+2.63 %), budget spent 18 606 -> 20 907 (+287.6 B per excerpt), excerpt
members 9 -> 15, 83 changed lines. The contract document's own bytes moved with
8, so the budget instrument's known-positive moves with it: 10 349 -> 12 049
measured, 10 060 -> 11 719 raw, delta 289 -> 330.

New fixture `tests/fixtures/consume-provenance`: the two address forms and a
concept carrying neither address nor identifier. Purpose-built, because the two
real bundles are complementary and neither exercises both forms.

Suite 1347 (1339 before), ruff clean, mypy src clean.

Co-Authored-By: Claude <claude-opus-5>
2026-09-08 15:01:41 +02:00

2230 lines
99 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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 re
import socket
import subprocess
import sys
import unicodedata
from collections.abc import Mapping
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_consume_measure # 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_index_walk_excludes_a_linked_log_from_concept_navigation(tmp_path: Path) -> None:
"""A linked `log.md` is bundle metadata, not a concept -- measured on K2 (S7 F2).
`link_log_in_root_index` (`corpus.py`, `95eb271`) linked a run's own log from
the root index so a reader entering at `index.md` could reach it. That link
makes the log reachable by the same walk this instrument uses to enumerate
concepts, and a walk that does not distinguish "linked" from "concept"
counts it as a 630th concept on a 629-concept bundle -- exactly what the S7
acid test measured, with the log then ranked and cut like real content.
THE PRODUCER NO LONGER WRITES THAT LINK (2026-09-08), so the fixture writes
it here instead. The exclusion stays and is not dead code: every bundle
built between `95eb271` and that removal carries the link, including the
ones consumers are reading today, and this instrument must count 629 on
those too.
"""
from llm_ingestion_okf.corpus import LOG_NAME
copy = tmp_path / "bundle"
_copy_bundle(GOLDEN, copy)
(copy / LOG_NAME).write_text("# Corpus run history\n\nN = 3\n", encoding="utf-8")
index_path = copy / "index.md"
index_path.write_text(
index_path.read_text(encoding="utf-8") + f"- [Corpus run history]({LOG_NAME})\n",
encoding="utf-8",
newline="",
)
index = index_path.read_text(encoding="utf-8")
assert "](log.md)" in index, (
"the fixture must actually link the log for this control to mean anything"
)
found = okf_consume.enumerate_concepts(copy)
assert found == (
"krav/1-1/foerste-krav",
"krav/1-2/andre-krav",
"veiledning",
)
assert not any(concept.endswith("log") for concept in found)
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_normalise_holds_an_identifier_number_as_one_token() -> None:
# MEASURED 2026-09-08 over three vegnormal bundles (446, 1133 and 270
# concepts): `_TOKEN_SPLIT_RE` shatters `10.2-2` into `10`, `2`, `2` and
# `MIN_TOKEN_LENGTH` then drops every piece, so a question naming a
# requirement number reaches the ranker carrying only the word `krav` --
# which every concept in such a bundle also carries. BOTH mechanisms
# participate: the split destroys the number, the floor removes the
# remains. The gold requirement was `below_k` in three of three.
assert "10.2-2" in okf_consume.normalise("Krav 10.2\u20142")
assert okf_consume.normalise("3.3.1\u201413") == ("3.3.1-13",)
assert okf_consume.normalise("2.9.2\u201412") == ("2.9.2-12",)
assert okf_consume.normalise("R610.4") == ("r610.4",)
assert okf_consume.normalise("4.2.1") == ("4.2.1",)
def test_an_identifiers_three_spellings_of_its_separator_normalise_alike() -> None:
# One requirement number arrives as an em dash from the source viewer, an
# en dash from a converter and a plain hyphen from a person typing the
# question. NFC folds NONE of the three, so a rule that does not fold them
# finds the number only in the spelling it was asked with.
hyphen = okf_consume.normalise("10.2-2")
assert okf_consume.normalise("10.2\u20142") == hyphen
assert okf_consume.normalise("10.2\u20132") == hyphen
assert hyphen == ("10.2-2",)
def test_the_identifier_rule_leaves_the_noise_floor_it_was_added_under() -> None:
# The known-negative, and the whole reason `MIN_TOKEN_LENGTH` exists: a
# bare short number matches every page number, row count and year in a
# corpus, and a matcher that scores them ranks every document equally.
assert okf_consume.normalise("10") == ()
assert okf_consume.normalise("2") == ()
assert okf_consume.normalise("er en pris i et skjema") == ("pris", "skjema")
# A hyphenated WORD is not an identifier -- no digit stands on either side
# of the separator -- so it splits exactly as it always did.
assert okf_consume.normalise("skole-anbudet") == ("skole", "anbudet")
# And a separator this rule does not claim leaves its token set untouched:
# the K2 corpus spells standards this way.
assert okf_consume.normalise("NS3935:2019") == ("ns3935", "2019")
assert okf_consume.normalise("TEK 17") == ("tek",)
def test_an_identifier_inside_a_slug_does_not_swallow_the_words_around_it() -> None:
# MEASURED, and the reason the rule joins DIGIT groups rather than
# alphanumeric ones. A first version joined alphanumeric groups across a
# separator; a corpus document's slug then became ONE token, because a
# `3-6` sits inside it, and that document's score for a question naming its
# subject fell from 0.735 to 0.0 -- one hit@8 row lost, on a question
# carrying no identifier at all. The slug below has that shape and is not
# the corpus's (SS "consumer content stays at form level"). The identifier
# is ADDED here; nothing is taken away.
tokens = okf_consume.normalise("rapport-iv-vedlegg-3-6-grunnforhold-akustikk")
assert "akustikk" in tokens
assert "grunnforhold" in tokens
assert "vedlegg" in tokens
assert "3-6" in tokens
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/pristabell")
krav_later = [c.concept_id for c, _, _ in dropped].index("krav/pristabell")
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/pristabell"
# --- 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/pristabell"
# --- 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"])
#: The gold set is LOCAL-ONLY: it names corpus documents, which never reach a
#: tracked file here. The test reads it rather than restating it, so this file
#: carries the assertion and not the answer key.
GOLD_SET = PROJECT_ROOT / ".claude/projects/2026-09-07-okf-consume-prepass/hit-at-k-questions.json"
@requires_k2
@pytest.mark.skipif(not GOLD_SET.is_file(), reason=f"the local gold set is absent ({GOLD_SET})")
def test_every_gold_document_in_the_local_set_is_reached_or_named_as_a_miss() -> None:
# SC5 and SC6 together, run against the answer key rather than a literal.
# Row 1's gold is the one confirmed by a signal from outside this
# repository -- a live model reached that document unprompted in three
# navigation steps on 2026-09-06 -- and its gold document holds exactly one
# concept, so it is also the one concept-granularity row.
spec = json.loads(GOLD_SET.read_text(encoding="utf-8"))
questions = spec["questions"]
assert len(questions) >= 5, "fewer than five questions is not the measurement"
hits = 0
for entry in questions:
payload = okf_consume.build_payload(K2_BUNDLE, question=entry["question"])
excerpts = payload["excerpts"]
assert isinstance(excerpts, list)
if okf_consume_measure.hit_rank(excerpts, entry["gold_document"]) is not None:
hits += 1
# The published bar, and the published number. A regression that drops a
# row goes red here rather than in a document nobody re-runs.
assert hits == 5, f"hit@8 moved: {hits} of {len(questions)}"
# --- Step 9: the CLI ----------------------------------------------------------
TOOL = PROJECT_ROOT / "tools" / "okf_consume.py"
def _run(*args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, str(TOOL), *args], capture_output=True, text=True, check=False
)
def test_two_runs_of_the_same_arguments_produce_byte_identical_stdout() -> None:
first = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?")
second = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?")
assert first.returncode == 0, first.stderr
assert first.stdout == second.stdout
assert first.stdout
def test_the_module_reaches_no_clock() -> None:
# Determinism is a property of the code, not only of two runs that happened
# to land in the same second.
source = TOOL.read_text(encoding="utf-8")
for forbidden in ("datetime.now", "time.time", "utcnow", "time.monotonic"):
assert forbidden not in source
def test_exit_zero_one_and_two_are_each_reached_by_a_distinct_real_condition() -> None:
ok = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?")
assert ok.returncode == 0
refused = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?", "--limit", "100")
assert refused.returncode == 1
assert "budget" in refused.stderr
absent = _run(str(FIXTURE / "does-not-exist"), "--question", "Hva som helst her")
assert absent.returncode == 2
def test_a_matching_ref_passes_and_a_mismatching_one_refuses_and_writes_nothing(
tmp_path: Path,
) -> None:
# SS 3.3: `--ref` is an ASSERTION. An override would let a caller label a
# payload with an identity its bytes do not have, which is the one thing
# that paragraph exists to prevent.
real = okf_consume.bundle_ref(FIXTURE)
out = tmp_path / "payload.json"
good = _run(
str(FIXTURE),
"--question",
"Hvordan skal prisene fylles ut?",
"--ref",
real,
"--out",
str(out),
)
assert good.returncode == 0
assert json.loads(out.read_text(encoding="utf-8"))["bundle"]["ref"] == real
missing = tmp_path / "never-written.json"
bad = _run(
str(FIXTURE),
"--question",
"Hvordan skal prisene fylles ut?",
"--ref",
"sha256-tree:0000",
"--out",
str(missing),
)
assert bad.returncode == 1
assert not missing.exists()
def test_out_writes_exactly_the_bytes_stdout_produced(tmp_path: Path) -> None:
out = tmp_path / "payload.json"
piped = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?")
written = _run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?", "--out", str(out))
assert written.returncode == 0
assert out.read_text(encoding="utf-8") == piped.stdout
def test_every_top_level_import_is_stdlib_or_this_repository() -> None:
# SC4, narrowed with the measurement that forced it: importing any library
# primitive pulls `socket`/`ssl`/`urllib` transitively, because Door A
# legitimately needs them. Reachability is not use. The honest guarantee is
# no THIRD-PARTY dependency plus no network call, and the second half is
# asserted below.
source = TOOL.read_text(encoding="utf-8")
imported = set(re.findall(r"^(?:from|import) ([a-zA-Z_][\w.]*)", source, re.MULTILINE))
for module in imported:
root = module.split(".")[0]
assert root in sys.stdlib_module_names or root == "llm_ingestion_okf", root
def test_no_socket_is_opened_during_a_real_run(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[object] = []
def refuse(*args: object, **kwargs: object) -> None:
calls.append(args)
raise AssertionError("the pre-pass opened a socket")
monkeypatch.setattr(socket, "socket", refuse)
monkeypatch.setattr(socket, "create_connection", refuse)
# The guard proven able to fire, before its silence counts as evidence.
with pytest.raises(AssertionError):
socket.socket()
calls.clear()
okf_consume.build_payload(FIXTURE, question="Hvordan skal prisene fylles ut?")
assert calls == []
def test_the_payload_written_by_the_cli_passes_the_checker(tmp_path: Path) -> None:
out = tmp_path / "payload.json"
assert (
_run(str(FIXTURE), "--question", "Hvordan skal prisene fylles ut?", "--out", str(out))
).returncode == 0
checked = subprocess.run(
[
sys.executable,
str(PROJECT_ROOT / "tools" / "okf_contract_check.py"),
"--skill",
str(TEMPLATE),
"--payload",
str(out),
],
capture_output=True,
text=True,
check=False,
)
assert checked.returncode == 0, checked.stdout
assert "0 findings" in checked.stdout
# --- Step 10: the instantiated skill -----------------------------------------
SKILL = PROJECT_ROOT / "skills" / "okf-consume" / "SKILL.md"
PLACEHOLDER_RE = re.compile(r"<[A-Z][A-Z_]{2,}(?::.*?)?>", re.DOTALL)
def test_the_placeholder_scan_finds_them_in_the_template_before_its_zero_counts() -> None:
# The known-positive, run FIRST. The obvious check is blind: a
# line-oriented `<[A-Z_]*>` cannot match `<EXTENSION_MARKINGS: …>`,
# `<CONDITIONAL_FIELDS: …>` or `<COST_SCALING: …>`, each of which spans
# lines. Measured: the naive pattern reports 17 against 20 real occurrences.
template = TEMPLATE.read_text(encoding="utf-8")
naive = re.findall(r"<[A-Z_]*>", template)
thorough = PLACEHOLDER_RE.findall(template)
assert len(thorough) >= 20
assert len(thorough) > len(naive), "the scan is no better than the blind one"
def test_the_instantiated_skill_has_no_placeholder_left() -> None:
assert PLACEHOLDER_RE.findall(SKILL.read_text(encoding="utf-8")) == []
def test_the_instantiated_skill_carries_every_required_section() -> None:
text = SKILL.read_text(encoding="utf-8")
for section in okf_contract_check.REQUIRED_SECTIONS:
assert f"## {section}" in text
def test_the_instantiated_skill_carries_every_marking_and_state_literal() -> None:
text = SKILL.read_text(encoding="utf-8")
for marking in okf_contract_check.REQUIRED_MARKINGS:
assert marking in text, marking
for state in (*okf_contract_check.ADJUDICATION_STATES, *okf_contract_check.TRUST_TIERS):
assert f"`{state}`" in text, state
def test_every_rule_the_pre_pass_can_emit_is_named_in_the_skill() -> None:
# The anti-drift gate. A rule the pre-pass emits and the skill does not
# explain is a `withheld` entry no reader can act on, and the copy nobody
# reads is the one that goes wrong.
text = SKILL.read_text(encoding="utf-8")
for rule in okf_consume.WITHHOLDING_RULES:
assert rule in text, rule
def test_the_skill_and_a_real_payload_pass_the_checker_together() -> None:
payload = _payload()
assert okf_contract_check.check(SKILL.read_text(encoding="utf-8"), payload).findings == ()
def test_the_shipped_example_payload_is_current_and_regenerates_byte_for_byte() -> None:
# A shipped artefact that has drifted from the tool that made it is worse
# than none: it documents a shape the code no longer emits.
shipped = (SKILL.parent / "references" / "example-payload.json").read_text(encoding="utf-8")
regenerated = okf_consume.serialise(
okf_consume.build_payload(GOLDEN, question="Hva sier veiledningen om krav?")
)
assert shipped == regenerated
@requires_k2
def test_no_corpus_document_name_reaches_any_file_this_work_tracks() -> None:
# CLAUDE.md's public-file rule. The pattern is DERIVED from the corpus's own
# top-level document names at run time rather than hand-picked, so it covers
# every document rather than the six someone thought of -- and so this
# tracked file carries no corpus name of its own.
documents = sorted(
{concept_id.split("/", 1)[0] for concept_id in okf_consume.enumerate_concepts(K2_BUNDLE)}
)
assert len(documents) > 30, "too few documents to be the real corpus"
leak = re.compile("|".join(re.escape(name) for name in documents), re.IGNORECASE)
# The known-positive, first: the pattern must be shown able to find before
# its zero counts as a measurement.
control = (K2_BUNDLE / "index.md").read_text(encoding="utf-8")
assert leak.findall(control), "the pattern cannot find; the zeros below would mean nothing"
tracked = [
SKILL,
SKILL.parent / "references" / "README.md",
SKILL.parent / "references" / "example-payload.json",
PROJECT_ROOT / "tools" / "okf_consume.py",
PROJECT_ROOT / "tools" / "okf_consume_measure.py",
PROJECT_ROOT / "tests" / "test_okf_consume.py",
PROJECT_ROOT / "docs" / "2026-09-07-okf-konsumskill-maaling.md",
PROJECT_ROOT / "docs" / "2026-09-08-blindsone-below-k-k2.md",
PROJECT_ROOT / "docs" / "2026-09-08-blindsone-laas2-budsjett-k2.md",
PROJECT_ROOT / "docs" / "2026-09-08-prisform-og-loggen-k2.md",
PROJECT_ROOT / "docs" / "2026-09-08-kravnummer-tokenisering.md",
PROJECT_ROOT / "docs" / "2026-09-08-sjeldenhetsvekt.md",
PROJECT_ROOT / "README.md",
PROJECT_ROOT / "CLAUDE.md",
]
for path in tracked:
assert leak.findall(path.read_text(encoding="utf-8")) == [], path
def test_the_readme_consume_section_states_the_rule_count_the_code_emits() -> None:
# A published number must have a test that goes red when it goes false.
readme = (PROJECT_ROOT / "README.md").read_text(encoding="utf-8")
assert readme.count("## Consume") == 1
assert len(okf_consume.WITHHOLDING_RULES) == 6
assert "closed set of six" in readme
assert "tools/okf_consume.py" in readme
# --- Step 11: the measurement scorer -----------------------------------------
def test_a_gold_at_rank_three_is_a_hit_at_eight_and_a_miss_at_two() -> None:
excerpts = [
{"concept_id": "other-doc/a", "rank": 1},
{"concept_id": "other-doc/b", "rank": 2},
{"concept_id": "gold-doc/c", "rank": 3},
]
assert okf_consume_measure.hit_rank(excerpts, "gold-doc") == 3
assert okf_consume_measure.hit_rank(excerpts[:2], "gold-doc") is None
def test_the_document_prefix_test_does_not_match_a_longer_document_name() -> None:
# Without the trailing slash, a gold document `bilag-7` would count a
# concept in `bilag-70` as a hit -- a false positive that inflates the
# headline number and looks like a correct answer.
excerpts = [{"concept_id": "bilag-70/a", "rank": 1}]
assert okf_consume_measure.hit_rank(excerpts, "bilag-7") is None
assert okf_consume_measure.hit_rank([{"concept_id": "bilag-7", "rank": 1}], "bilag-7") == 1
def test_the_analytic_chance_baseline_is_exact_for_a_single_gold_concept() -> None:
# One gold concept in a 629-concept corpus, drawn 8: exactly 8/629, which
# is derivable by hand and pins the closed form.
assert okf_consume_measure.chance_analytic(1, 629, 8) == pytest.approx(8 / 629)
assert okf_consume_measure.chance_analytic(629, 629, 8) == 1.0
assert okf_consume_measure.chance_analytic(0, 629, 8) == 0.0
def test_the_empirical_baseline_reproduces_the_analytic_one_on_a_fixed_seed() -> None:
# Two routes to one number. The agreement is about half a percentage point
# at 20 000 trials -- asserted at that tolerance rather than at the three
# decimal places the plan claimed, because the looser figure is the one
# that is true.
for gold in (1, 5, 11, 20, 43, 49):
analytic = okf_consume_measure.chance_analytic(gold, 629, 8)
empirical = okf_consume_measure.chance_empirical(gold, 629, 8)
assert abs(analytic - empirical) < 0.01, gold
def test_document_sizes_counts_root_level_concepts_as_their_own_document() -> None:
sizes = okf_consume_measure.document_sizes(("a/1", "a/2", "b/1", "root-concept"))
assert sizes == {"a": 2, "b": 1, "root-concept": 1}
def test_the_measurement_instrument_names_no_corpus_document() -> None:
# It is tracked and public; the answer key is an input, never a constant.
source = (PROJECT_ROOT / "tools" / "okf_consume_measure.py").read_text(encoding="utf-8")
leak = re.compile(r"del-ii-bilag|del-i-vedlegg|del-i-konkurranse|prisskjema|stange", re.I)
assert leak.findall(source) == []
# --- Step 12: the declared cost vocabulary, behind a flag ---------------------
#
# Measured 2026-09-08 on the K2 corpus (`docs/2026-09-08-blindsone-below-k-k2.md`):
# a mandate-shaped cost question ranks the corpus's one priced table 249th of
# 269 lexical candidates, because its title, its id and its document index
# entries carry none of the question's tokens. The gap is a VOCABULARY gap --
# the question says `kostnadsbesparelser`, the document says `pris` -- and no
# amount of `k` closes it. The fixture below reproduces that gap synthetically:
# `krav/pristabell` is `no_lexical_match` for a question about `kostnader`.
def test_a_cost_question_reaches_no_price_concept_without_the_flag() -> None:
# The known-negative this whole step is measured against. Without it, the
# flag's effect below would have no denominator.
payload = _payload(question="Hvor kan vi kutte kostnader?")
counts, withheld = payload["denominators"], payload["withheld"]
assert isinstance(counts, dict) and isinstance(withheld, list)
assert counts["delivered"] == 0
assert {"concept_id": "krav/pristabell", "rule": "no_lexical_match"} in withheld
def test_the_cost_vocabulary_flag_bridges_a_question_and_a_document_that_share_no_word() -> None:
payload = okf_consume.build_payload(
FIXTURE, question="Hvor kan vi kutte kostnader?", cost_vocabulary=True
)
excerpts = payload["excerpts"]
assert isinstance(excerpts, list)
assert "krav/pristabell" in [excerpt["concept_id"] for excerpt in excerpts]
def test_the_flag_is_off_by_default_and_the_default_payload_is_byte_identical() -> None:
# The library's standing promise to a consumer: a new parameter is
# keyword-only with a default, and the default bytes do not move.
question = "Hvor kan vi kutte kostnader?"
off = okf_consume.serialise(okf_consume.build_payload(FIXTURE, question=question))
explicit = okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question, cost_vocabulary=False)
)
assert off == explicit
def test_the_flag_changes_nothing_when_the_question_names_no_such_term() -> None:
# The GATE is the question, never the flag: a question with no cost term
# gets byte-identical bytes whether the flag is set or not.
question = "Hvor ofte er den årlige kontrollen?"
off = okf_consume.serialise(okf_consume.build_payload(FIXTURE, question=question))
on = okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question, cost_vocabulary=True)
)
assert off == on
def test_the_bridge_needs_a_vocabulary_term_on_both_sides() -> None:
# A one-sided bridge would make every cost question match every document,
# which is the confident guess `no_lexical_match` exists to forbid.
assert okf_consume.in_cost_vocabulary("kostnadsbesparelser")
assert okf_consume.in_cost_vocabulary("prissammenstilling")
assert not okf_consume.in_cost_vocabulary("kontrollen")
tokens = okf_consume.normalise("kostnader")
assert okf_consume._overlap(tokens, "aarlig kontroll", cost_vocabulary=True) == 0
assert okf_consume._overlap(tokens, "prisene fylles ut", cost_vocabulary=True) == 1
# And the bridge carries the vocabulary term ALONE: a question's unrelated
# tokens do not ride along on it. Without this the widening would be
# "everything matches a price document", not "cost words do".
mixed = okf_consume.normalise("kostnader kontrollen")
assert okf_consume._overlap(mixed, "prisene fylles ut", cost_vocabulary=True) == 1
# The gate is the question. Asserted directly, because the per-token test
# above holds even when the gate is stuck open.
assert okf_consume.question_uses_cost_vocabulary("Hvor kan vi kutte kostnader?")
assert not okf_consume.question_uses_cost_vocabulary("Hvor ofte er den årlige kontrollen?")
def test_every_vocabulary_member_is_long_enough_to_ever_match() -> None:
# `tokens_match` needs MIN_SHARED_PREFIX characters, so a shorter member is
# dead code that reads as coverage. Measured: `sum` (3) never matches
# `Summen` and was dropped for that reason.
assert okf_consume.COST_VOCABULARY
for member in okf_consume.COST_VOCABULARY:
assert len(member) >= okf_consume.MIN_SHARED_PREFIX, member
assert member == member.casefold(), member
assert list(okf_consume.COST_VOCABULARY) == sorted(okf_consume.COST_VOCABULARY)
def test_the_vocabulary_is_one_list_and_names_no_corpus_document() -> None:
source = (PROJECT_ROOT / "tools" / "okf_consume.py").read_text(encoding="utf-8")
assert source.count("COST_VOCABULARY = (") == 1
leak = re.compile(r"del-ii-bilag|del-i-vedlegg|prisskjema|prissammenstilling|stange", re.I)
assert leak.findall(source) == []
def test_the_cli_exposes_the_flag_and_omitting_it_reproduces_the_default_bytes() -> None:
question = "Hvordan skal prisene fylles ut?"
plain = _run(str(FIXTURE), "--question", question)
assert plain.returncode == 0
flagged = _run(str(FIXTURE), "--question", question, "--cost-vocabulary")
assert flagged.returncode == 0
assert plain.stdout == okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question)
)
assert flagged.stdout == okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question, cost_vocabulary=True)
)
# --- Step 13: the budget reserved for the top-ranked candidate, behind a flag -
#: The shape the corpus measurement found (`docs/2026-09-08-blindsone-below-k-k2.md`
#: SS 3): one top-ranked concept costing more than half the budget, and enough
#: small ones that their SUM of fused scores out-values it. Synthetic and
#: general -- no corpus path, no corpus byte constant, no corpus document name.
EVICTION_QUESTION = "Hva staar i den store tabellen om kontroll?"
EVICTION_LIMIT = 12_000
EVICTION_TOP = "stor/tabell"
_EVICTION_FRONTMATTER = (
"---\ntype: reference\ntitle: {title}\nsource_file: {title}.md\n"
"source_sha256: {digest}\ningested_at: 2026-09-01T00:00:00Z\n"
"adjudication: proposed\nbundle_id: eviction-fixture\n"
"verified: [{{ by: process:okf-check, at: 2026-09-01T00:00:00Z }}]\n---\n\n"
)
def _eviction_bundle(root: Path, *, smalls: int = 12, fat_small_lines: int = 9) -> Path:
# `fat_small_lines` makes ONE lower-ranked concept the heaviest in the
# bundle, so "the top-ranked candidate" and "the largest excerpt" can be
# told apart by a test rather than coinciding by accident.
(root / "stor").mkdir(parents=True)
(root / "smaa").mkdir(parents=True)
(root / "index.md").write_text(
"---\nokf_version: 0.2\nbundle_id: eviction-fixture\n---\n\n"
"- [stor (index)](stor/index.md)\n- [smaa (index)](smaa/index.md)\n",
encoding="utf-8",
)
(root / "stor" / "index.md").write_text(
"- [Stor tabell om kontroll](tabell.md) — adjudication: proposed\n", encoding="utf-8"
)
(root / "stor" / "tabell.md").write_text(
_EVICTION_FRONTMATTER.format(title="Stor tabell om kontroll", digest="1" * 64)
+ "## Stor tabell om kontroll\n\n"
+ "Kontrollen av den store tabellen foelger tabellen rad for rad.\n" * 105,
encoding="utf-8",
)
entries = []
for number in range(1, smalls + 1):
name = f"notat-{number:02d}"
entries.append(f"- [Notat om kontroll {number:02d}]({name}.md) — adjudication: proposed\n")
lines = fat_small_lines if number == smalls else 9
(root / "smaa" / f"{name}.md").write_text(
_EVICTION_FRONTMATTER.format(title=f"Notat om kontroll {number:02d}", digest="2" * 64)
+ f"## Notat om kontroll {number:02d}\n\n"
+ "Notatet gjelder kontroll av ett enkelt punkt.\n" * lines,
encoding="utf-8",
)
(root / "smaa" / "index.md").write_text("".join(entries), encoding="utf-8")
return root
def _eviction_payload(
root: Path, *, k: int = 16, limit: int = EVICTION_LIMIT, reserve_top_rank: bool = False
) -> dict[str, object]:
return okf_consume.build_payload(
root,
question=EVICTION_QUESTION,
k=k,
limit=limit,
reserve_top_rank=reserve_top_rank,
)
def _top_candidate(root: Path) -> str:
# Computed the long way -- through the ranker, not read off the payload --
# so "the top-ranked candidate" in the assertions below is not whatever the
# cut happened to deliver first.
profile = okf_consume.DEFAULT_PROFILE
concepts = [
okf_consume.read_concept(
root / f"{concept_id}.md", bundle_root=root, root_bundle_id="eviction-fixture"
)
for concept_id in okf_consume.enumerate_concepts(root, profile=profile)
]
ranked = okf_consume.concept_scores(
concepts, EVICTION_QUESTION, okf_consume.document_scores(root, EVICTION_QUESTION)
)
return next(concept.concept_id for concept, _, lexical in ranked if lexical > 0)
def test_the_knapsack_evicts_the_top_ranked_candidate_that_costs_half_the_budget(
tmp_path: Path,
) -> None:
# The known-positive for everything below: without it, a delivered top rank
# with the flag on would prove nothing, because nothing would have been
# shown to remove it.
root = _eviction_bundle(tmp_path / "bundle")
assert _top_candidate(root) == EVICTION_TOP
# The DEFAULT path, called without the new argument, so this control runs
# and can be believed before the rule exists.
payload = okf_consume.build_payload(
root, question=EVICTION_QUESTION, k=16, limit=EVICTION_LIMIT
)
excerpts = payload["excerpts"]
assert isinstance(excerpts, list)
weights = {
excerpt["concept_id"]: okf_consume.excerpt_weight(excerpt)
for excerpt in excerpts
if isinstance(excerpt, dict)
}
assert EVICTION_TOP not in weights
assert (
dict(
(entry["concept_id"], entry["rule"])
for entry in payload["withheld"] # type: ignore[union-attr]
)[EVICTION_TOP]
== "over_budget_after_knapsack"
)
# The shape itself, stated as numbers rather than assumed: the top candidate
# fits ALONE and still loses, which is what makes this a budget question.
top = _eviction_bundle_top_weight(root)
assert top <= EVICTION_LIMIT
assert top > EVICTION_LIMIT // 2
def _eviction_bundle_top_weight(root: Path) -> int:
concept = okf_consume.read_concept(
root / f"{EVICTION_TOP}.md", bundle_root=root, root_bundle_id="eviction-fixture"
)
excerpt = okf_consume.excerpt_for(concept)
assert excerpt is not None
return okf_consume.excerpt_weight(excerpt)
def test_reserving_the_top_rank_delivers_the_candidate_the_knapsack_evicted(
tmp_path: Path,
) -> None:
root = _eviction_bundle(tmp_path / "bundle")
payload = _eviction_payload(root, reserve_top_rank=True)
excerpts = payload["excerpts"]
assert isinstance(excerpts, list)
assert excerpts[0]["concept_id"] == EVICTION_TOP
assert excerpts[0]["rank"] == 1
assert EVICTION_TOP not in {
entry["concept_id"]
for entry in payload["withheld"] # type: ignore[union-attr]
}
def test_the_reservation_is_off_by_default_and_the_default_payload_is_byte_identical(
tmp_path: Path,
) -> None:
root = _eviction_bundle(tmp_path / "bundle")
default = okf_consume.serialise(
okf_consume.build_payload(root, question=EVICTION_QUESTION, k=16, limit=EVICTION_LIMIT)
)
explicit_off = okf_consume.serialise(_eviction_payload(root, reserve_top_rank=False))
assert default == explicit_off
assert '"reserved"' not in default
question = "Hvordan skal prisene fylles ut?"
assert okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question)
) == okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question, reserve_top_rank=False)
)
def test_a_top_candidate_that_alone_exceeds_the_budget_is_still_refused_by_name(
tmp_path: Path,
) -> None:
# The control the reservation must not break: `over_budget_alone` is a
# PRE-exclusion, and a reservation that ran before it would deliver an
# excerpt the budget can never hold.
root = _eviction_bundle(tmp_path / "bundle")
limit = _eviction_bundle_top_weight(root) - 1
payload = _eviction_payload(root, limit=limit, reserve_top_rank=True)
rules = dict(
(entry["concept_id"], entry["rule"])
for entry in payload["withheld"] # type: ignore[union-attr]
)
assert rules[EVICTION_TOP] == "over_budget_alone"
spent = payload["budget"]["spent"] # type: ignore[index]
assert isinstance(spent, int)
assert spent <= limit
def test_the_reservation_never_spends_more_than_the_budget(tmp_path: Path) -> None:
root = _eviction_bundle(tmp_path / "bundle")
for limit in (EVICTION_LIMIT, EVICTION_LIMIT + 3_000, EVICTION_LIMIT * 2):
payload = _eviction_payload(root, limit=limit, reserve_top_rank=True)
spent = payload["budget"]["spent"] # type: ignore[index]
assert isinstance(spent, int)
assert spent <= limit
def test_the_reservation_displaces_lower_ranked_excerpts_under_the_rule_that_exists(
tmp_path: Path,
) -> None:
# The cost of the rule, asserted rather than described: reserving the top
# rank buys its bytes from the excerpts the knapsack preferred, and they
# leave under a rule already in the closed set.
root = _eviction_bundle(tmp_path / "bundle")
without = _eviction_payload(root)
with_reservation = _eviction_payload(root, reserve_top_rank=True)
assert len(with_reservation["excerpts"]) < len(without["excerpts"]) # type: ignore[arg-type]
displaced = {
entry["concept_id"]
for entry in with_reservation["withheld"] # type: ignore[union-attr]
if entry["rule"] == "over_budget_after_knapsack"
}
delivered_before = {
excerpt["concept_id"]
for excerpt in without["excerpts"] # type: ignore[union-attr]
}
assert displaced & delivered_before
assert {
entry["rule"]
for entry in with_reservation["withheld"] # type: ignore[union-attr]
} <= set(okf_consume.WITHHOLDING_RULES)
def test_the_payload_declares_which_concept_the_reservation_took_and_what_it_cost(
tmp_path: Path,
) -> None:
root = _eviction_bundle(tmp_path / "bundle")
budget = _eviction_payload(root, reserve_top_rank=True)["budget"]
assert isinstance(budget, dict)
assert budget["reserved"] == {
"concept_id": _top_candidate(root),
"bytes": _eviction_bundle_top_weight(root),
}
spent = budget["spent"]
assert isinstance(spent, int)
assert spent >= _eviction_bundle_top_weight(root)
def test_the_reservation_is_paid_once_and_does_not_bid_for_its_own_bytes(
tmp_path: Path,
) -> None:
# A reserved excerpt left in the pack's pool competes for the budget it has
# already been given, and wins it back from the next candidate. Visible
# only where the reserved excerpt would out-value what the remaining room
# can hold: two candidates, and room enough for the reserved one twice.
root = _eviction_bundle(tmp_path / "bundle")
payload = _eviction_payload(root, k=2, limit=15_000, reserve_top_rank=True)
delivered = [
excerpt["concept_id"]
for excerpt in payload["excerpts"] # type: ignore[union-attr]
]
assert delivered == [EVICTION_TOP, "smaa/notat-01"]
def test_the_reservation_names_the_fused_top_and_not_the_heaviest_candidate(
tmp_path: Path,
) -> None:
# A bundle whose HEAVIEST excerpt is a LOWER-ranked one. A reservation
# reading weight where it should read rank reserves the wrong concept, and
# this is the only place the two come apart.
root = _eviction_bundle(tmp_path / "bundle", fat_small_lines=400)
heaviest = max(
okf_consume.enumerate_concepts(root),
key=lambda concept_id: okf_consume.excerpt_weight(
okf_consume.excerpt_for(
okf_consume.read_concept(
root / f"{concept_id}.md",
bundle_root=root,
root_bundle_id="eviction-fixture",
)
)
or {}
),
)
assert heaviest != EVICTION_TOP
budget = _eviction_payload(root, limit=EVICTION_LIMIT * 3, reserve_top_rank=True)["budget"]
assert isinstance(budget, dict)
assert budget["reserved"]["concept_id"] == EVICTION_TOP # type: ignore[index]
def test_a_payload_carrying_a_reservation_still_passes_the_checker(tmp_path: Path) -> None:
# SS 8 permits additional members; a declaration the checker refuses would
# buy honesty at the price of conformance.
root = _eviction_bundle(tmp_path / "bundle")
report = okf_contract_check.check(
TEMPLATE.read_text(encoding="utf-8"), _eviction_payload(root, reserve_top_rank=True)
)
assert report.findings == ()
def test_the_cli_exposes_the_reservation_and_omitting_it_reproduces_the_default_bytes(
tmp_path: Path,
) -> None:
root = _eviction_bundle(tmp_path / "bundle")
common = (
str(root),
"--question",
EVICTION_QUESTION,
"--k",
"16",
"--limit",
str(EVICTION_LIMIT),
)
plain = _run(*common)
assert plain.returncode == 0
reserved = _run(*common, "--reserve-top-rank")
assert reserved.returncode == 0
assert plain.stdout == okf_consume.serialise(_eviction_payload(root))
assert reserved.stdout == okf_consume.serialise(_eviction_payload(root, reserve_top_rank=True))
assert plain.stdout != reserved.stdout
# --- Step 12: the rarity weight (O2b) -----------------------------------------
#: The situation the previous session measured and could not close, recreated
#: small: every concept carries the common word, exactly one carries the
#: identifier, and the identifier is worth the same as the common word because
#: `_overlap` counts. Synthetic rather than borrowed from a bundle, because a
#: fixture that is a corpus measures that corpus.
RARITY_QUESTION = "Hva krever Krav 10.2-2 i N500? Gjengi det sentrale vilkåret."
RARITY_GOLD = "krav/c-29"
def _synthetic(concept_id: str, title: str, body: str) -> okf_consume.Concept:
return okf_consume.Concept(
path=Path(concept_id),
concept_id=concept_id,
bundle_id="rarity-fixture",
bundle_id_inherited=False,
sha256="0" * 64,
okf_type="Krav",
title=title,
source_file="synthetic.md",
adjudication="unknown",
adjudication_present=False,
req_number="",
sources=(),
sources_present=False,
locators={},
frontmatter={},
body=body,
)
def _rarity_corpus(*, identifier_in_body: bool) -> list[okf_consume.Concept]:
"""29 concepts bearing the common words, one bearing the identifier.
The gold's `concept_id` sorts LAST, so nothing but the score can lift it:
on a tie the declared tie-break puts it at the bottom.
`identifier_in_body` is the whole difference between the two shapes the
real corpora turned out to have, and it decides whether the weight can do
anything at all -- see the two tests below.
"""
common = "Kravet i N500 gjengir det sentrale vilkåret for anlegget."
concepts = [
_synthetic(f"krav/c-{index:02d}", f"Krav 3.{index}-1 Alminnelig bestemmelse", common)
for index in range(29)
]
body = (
"Krav 10.2-2 stiller vilkår om anlegget."
if identifier_in_body
else "Tekniske bygg stiller vilkår om anlegget."
)
concepts.append(_synthetic(RARITY_GOLD, "Krav 10.2-2 Tekniske bygg", body))
return concepts
def _rank_of(ranked: list[tuple[okf_consume.Concept, float, int]], concept_id: str) -> int:
return [concept.concept_id for concept, _, _ in ranked].index(concept_id) + 1
def _signal_rank(
concepts: list[okf_consume.Concept], question: str, gold: str, weights: object
) -> int:
"""Where the gold sits on the title-and-id signal alone, by the ranker's own rule."""
tokens = okf_consume.normalise(question)
scores = {
concept.concept_id: okf_consume._overlap(
tokens,
f"{concept.title} {concept.concept_id.replace('/', ' ')}",
weights=weights, # type: ignore[arg-type]
)
for concept in concepts
}
order = sorted(scores, key=lambda key: (-scores[key], key))
return order.index(gold) + 1
def test_counting_leaves_the_identifier_worth_no_more_than_the_common_word() -> None:
# The known-positive for the weight: without it, a rank of 1 with the
# weight on would prove nothing, because nothing would have been shown to
# hold the gold down in the first place.
# `lookup=False` throughout: this fixture puts the identifier in the
# gold's TITLE, which the lookup partition (Step 14) answers at rank one.
# The claim here is about the FUSION, so the fusion is what is read.
for identifier_in_body in (False, True):
concepts = _rarity_corpus(identifier_in_body=identifier_in_body)
ranked = okf_consume.concept_scores(concepts, RARITY_QUESTION, {}, lookup=False)
assert _rank_of(ranked, RARITY_GOLD) == 18
lexical = {concept.concept_id: value for concept, _, value in ranked}
assert lexical["krav/c-00"] == 6
def test_weighting_a_hit_by_its_rarity_lifts_the_identifier_into_the_cut() -> None:
# The arm where the weight can act: the identifier is in the body too, so
# weighting REORDERS the body signal. This is the shape N200 and the K2
# price sheet have, and both moved (withheld -> delivered rank 8; candidate
# rank 10 -> 2).
concepts = _rarity_corpus(identifier_in_body=True)
weights = okf_consume.rarity_weights(
okf_consume.normalise(RARITY_QUESTION), okf_consume.searchable_text(concepts)
)
ranked = okf_consume.concept_scores(concepts, RARITY_QUESTION, {}, weights=weights)
assert _rank_of(ranked, RARITY_GOLD) <= 8
def test_a_rarity_weight_cannot_move_a_signal_the_gold_already_leads() -> None:
"""The falsification, kept as a test so it cannot quietly stop being true.
RRF consumes RANKS ONLY -- that is the documented reason it was chosen. So
a weight changes the fused order only where it changes some signal's ORDER.
When the identifier is in the title alone, the gold already leads the title
signal by counting, the weight makes its lead larger and its RANK identical,
and the two signals that hold it down cannot see the identifier at all.
Measured on the real bundles: N500 35 -> 35, N100 96 -> 103.
"""
concepts = _rarity_corpus(identifier_in_body=False)
weights = okf_consume.rarity_weights(
okf_consume.normalise(RARITY_QUESTION), okf_consume.searchable_text(concepts)
)
# The weight DOES do its job on the score: the gold's title is worth more
# than every other title, by more than counting made it worth.
plain = okf_consume._overlap(
okf_consume.normalise(RARITY_QUESTION), f"{concepts[-1].title} krav c-29"
)
lifted = okf_consume._overlap(
okf_consume.normalise(RARITY_QUESTION), f"{concepts[-1].title} krav c-29", weights=weights
)
assert lifted > plain
# And the rank it buys is the rank it already had, so the fusion sees none of it.
assert _signal_rank(concepts, RARITY_QUESTION, RARITY_GOLD, None) == 1
assert _signal_rank(concepts, RARITY_QUESTION, RARITY_GOLD, weights) == 1
ranked = okf_consume.concept_scores(
concepts, RARITY_QUESTION, {}, weights=weights, lookup=False
)
assert _rank_of(ranked, RARITY_GOLD) == 18
def test_the_weight_leaves_the_lexical_count_a_count_so_the_cut_is_untouched() -> None:
# SS: the gate is a different axis from the ranking, and a token every
# concept carries weighs exactly zero. Were `lexical` the weighted sum, a
# concept matching only that token would become `no_lexical_match` -- the
# gate `54a0bc2` falsified rarity weighting FOR.
concepts = _rarity_corpus(identifier_in_body=True)
weights = okf_consume.rarity_weights(
okf_consume.normalise(RARITY_QUESTION), okf_consume.searchable_text(concepts)
)
plain = {
c.concept_id: v for c, _, v in okf_consume.concept_scores(concepts, RARITY_QUESTION, {})
}
weighted = {
c.concept_id: v
for c, _, v in okf_consume.concept_scores(concepts, RARITY_QUESTION, {}, weights=weights)
}
assert plain == weighted
def _equal_frequency_corpus() -> list[okf_consume.Concept]:
"""16 concepts over three words, each word borne by exactly 8 of them.
Equal `df` is the condition under which the weight provably cannot reorder
anything: every hit is multiplied by the same positive number, and RRF
consumes ranks. Concepts carry one word or all three, so the order being
preserved is a real order and not a single tie.
"""
words = ("vilkåret", "anlegget", "kravet")
bodies = [" ".join(words)] * 4 + [words[0]] * 4 + [words[1]] * 4 + [words[2]] * 4
return [
_synthetic(f"felles/e-{index:02d}", f"Bestemmelse {index}", f"Denne teksten nevner {body}.")
for index, body in enumerate(bodies)
]
def test_a_question_of_equally_common_words_comes_back_in_the_order_it_had() -> None:
concepts = _equal_frequency_corpus()
question = "Hva sier vilkåret om anlegget og kravet?"
texts = okf_consume.searchable_text(concepts)
weights = okf_consume.rarity_weights(okf_consume.normalise(question), texts)
# The three borne words weigh the same; `hva` and `sier` are borne by no
# concept at all, and their weight is never consumed because they never hit.
assert len({round(weights[word], 12) for word in ("vilkåret", "anlegget", "kravet")}) == 1
plain = [c.concept_id for c, _, _ in okf_consume.concept_scores(concepts, question, {})]
weighted = [
c.concept_id
for c, _, _ in okf_consume.concept_scores(concepts, question, {}, weights=weights)
]
assert weighted == plain
assert len(set(plain)) == 16
def test_a_word_every_concept_carries_weighs_exactly_nothing() -> None:
concepts = _rarity_corpus(identifier_in_body=True)
weights = okf_consume.rarity_weights(
okf_consume.normalise(RARITY_QUESTION), okf_consume.searchable_text(concepts)
)
assert weights["krav"] == 0.0
assert weights["10.2-2"] > weights["n500"] > 0.0
def test_a_question_token_no_concept_carries_is_weighted_but_never_consumed() -> None:
concepts = _rarity_corpus(identifier_in_body=True)
weights = okf_consume.rarity_weights(("kabelbroer",), okf_consume.searchable_text(concepts))
assert weights["kabelbroer"] > 0.0
assert okf_consume._overlap(("kabelbroer",), concepts[0].body, weights=weights) == 0.0
def test_the_weight_is_computed_from_the_bundle_and_not_from_a_constant() -> None:
# Same question, two corpora, two different weights for the same token:
# the number comes from the bundle in hand or it comes from nowhere.
small = okf_consume.rarity_weights(("n500",), ["N500 gjelder", "noe helt annet"])
large = okf_consume.rarity_weights(("n500",), ["N500 gjelder"] + ["noe helt annet"] * 9)
assert large["n500"] > small["n500"]
def test_the_weight_is_off_by_default_and_the_default_payload_is_unmoved() -> None:
# The decision recorded as a test rather than as a sentence: measured on
# four corpora the weight delivers one gold and costs another seven rank
# positions, so it ships OFF, and OFF has to mean the bytes that were
# already published.
plain = okf_consume.build_payload(FIXTURE, question="Hvordan skal prisene fylles ut?")
explicit = okf_consume.build_payload(
FIXTURE, question="Hvordan skal prisene fylles ut?", rarity_weight=False
)
assert okf_consume.serialise(plain) == okf_consume.serialise(explicit)
def test_the_cli_exposes_the_weight_and_omitting_it_reproduces_the_default_bytes() -> None:
question = "Hva krever Krav 10.2-2 om sentrale vilkår?"
plain = _run(str(FIXTURE), "--question", question)
weighted = _run(str(FIXTURE), "--question", question, "--rarity-weight")
assert plain.returncode == 0, plain.stderr
assert weighted.returncode == 0, weighted.stderr
assert plain.stdout == okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question)
)
assert weighted.stdout == okf_consume.serialise(
okf_consume.build_payload(FIXTURE, question=question, rarity_weight=True)
)
def test_the_weight_reaches_the_document_prior_and_not_only_the_concept_signals() -> None:
# One df table for the bundle, used wherever the question is scored against
# its text. A weight applied to two of three fused signals would be a
# second ranker rather than one statistic.
question = "Hvordan skal prisene fylles ut?"
weights = okf_consume.rarity_weights(
okf_consume.normalise(question), okf_consume.searchable_text(_fixture_concepts())
)
plain = okf_consume.document_scores(FIXTURE, question)
weighted = okf_consume.document_scores(FIXTURE, question, weights=weights)
assert plain.keys() == weighted.keys()
assert plain != weighted
def test_build_payload_hands_the_same_weights_to_the_document_prior(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""ONE df table for the bundle, reaching every stage that scores text.
Asserted on the call rather than on an output, and that is the point: on
all three real bundles the document prior is a two-document density whose
ORDER the weight does not change, so a version passing the weights to the
concept signals alone ranks identically there -- measured, 103/8/35 either
way. The commitment is still that one statistic reaches every stage, and a
commitment no output can distinguish has to be checked where it is made.
"""
seen: list[object] = []
original = okf_consume.document_scores
def spy(*args: object, **kwargs: object) -> dict[str, float]:
seen.append(kwargs.get("weights"))
return original(*args, **kwargs) # type: ignore[arg-type]
monkeypatch.setattr(okf_consume, "document_scores", spy)
question = "Hvordan skal prisene fylles ut?"
okf_consume.build_payload(FIXTURE, question=question)
okf_consume.build_payload(FIXTURE, question=question, rarity_weight=True)
assert seen[0] is None
expected = okf_consume.rarity_weights(
okf_consume.normalise(question), okf_consume.searchable_text(_fixture_concepts())
)
assert seen[1] == expected
def test_the_weight_reaches_the_title_signal_and_reorders_it() -> None:
# The gold's title answers TWO question tokens and every other title
# answers THREE, so counting puts the gold behind all of them; only the
# rarity of the identifier can turn that around, and only if the weight
# reaches the title-and-id signal. Bodies are identical, so the body signal
# decides nothing. This is the N200 shape, where the gold's title-signal
# rank moved 8 -> 4 and the gold went from withheld to delivered.
body = "Denne bestemmelsen gjelder for anlegget."
concepts = [
_synthetic(f"krav/t-{index:02d}", f"Krav 3.{index}-1 N500 gjengir bestemmelsen", body)
for index in range(29)
]
concepts.append(_synthetic("krav/t-29", "Krav 10.2-2 Tekniske bygg", body))
question = "Hva krever Krav 10.2-2 i N500? Gjengi det sentrale vilkåret."
weights = okf_consume.rarity_weights(
okf_consume.normalise(question), okf_consume.searchable_text(concepts)
)
assert _signal_rank(concepts, question, "krav/t-29", None) == 30
assert _signal_rank(concepts, question, "krav/t-29", weights) == 1
# `lookup=False`: the gold's title bears the identifier, so the lookup
# partition answers this question at rank one. The fusion is the claim.
plain = okf_consume.concept_scores(concepts, question, {}, lookup=False)
weighted = okf_consume.concept_scores(concepts, question, {}, weights=weights, lookup=False)
# Fused: 30 -> 18. The title signal is reordered from last to first and the
# fused rank moves by twelve, not to one -- the other two signals still
# cannot see the identifier. That gap IS the finding of this session, and
# the numbers are here so a change to either half shows up as a diff.
assert _rank_of(plain, "krav/t-29") == 30
assert _rank_of(weighted, "krav/t-29") == 18
# --- Step 14: exact identifier matching and the lookup signal (O2c) -----------
#: The lookup fixture: 300 concepts that all carry the word a standards corpus
#: repeats on every page, one of which also carries the identifier. Synthetic
#: rather than borrowed, and 300 rather than 30 so a rank of one is a claim
#: about the rule and not about a small corpus.
LOOKUP_QUESTION = "Hva krever Krav 3.3.1-13? Gjengi det sentrale vilkåret."
LOOKUP_GOLD = "krav/c-299"
LOOKUP_SECOND = "krav/z-000"
def _lookup_corpus(*, second_holder: bool = False) -> list[okf_consume.Concept]:
"""299 neighbours numbered `3.3.1-<n>`, plus the concept the question names.
The neighbours share the identifier's first four characters, which is
exactly what made a unique requirement number read as 135-of-446 common on
a real bundle. The gold's `concept_id` sorts LAST, so nothing but the rule
can lift it.
"""
body = "Kravet gjengir det sentrale vilkåret for anlegget."
concepts = [
_synthetic(f"krav/c-{index:03d}", f"Krav 3.3.1-{index + 20} Alminnelig krav", body)
for index in range(299)
]
concepts.append(_synthetic(LOOKUP_GOLD, "Krav 3.3.1-13 Tekniske bygg", body))
if second_holder:
# The second holder's id sorts AFTER the gold's and its title answers
# four question tokens more, so the fusion ranks it FIRST while byte
# order ranks it second. The two orders disagree on purpose: a lookup
# that re-sorted its hits by id instead of keeping the fused order
# would otherwise be indistinguishable from one that keeps it.
concepts.append(
_synthetic(LOOKUP_SECOND, "Krav 3.3.1-13 Gjengi det sentrale vilkåret", body)
)
return concepts
def test_an_identifier_matches_its_own_spelling_and_no_neighbouring_number() -> None:
# The defect, at the case it costs most: `3.3.1-13` and `3.3.1-14` share
# four leading characters, so the prefix rule called them a match and every
# requirement number beginning `3.3.` counted as a hit.
assert okf_consume.tokens_match("3.3.1-13", "3.3.1-13") is True
assert okf_consume.tokens_match("3.3.1-13", "3.3.1-14") is False
assert okf_consume.tokens_match("3.3.1-13", "3.3.2-13") is False
# Symmetric, like the rule it replaces.
assert okf_consume.tokens_match("3.3.1-14", "3.3.1-13") is False
def test_a_norwegian_compound_still_matches_on_the_shared_prefix() -> None:
# The known-positive. `MIN_SHARED_PREFIX` exists because Norwegian
# compounds do not match token-exactly, and an identifier rule that also
# narrowed words would buy one lookup by losing every compound.
assert okf_consume.tokens_match("brannsikring", "brannvern") is True
assert okf_consume.tokens_match("prisene", "prissammenstilling") is True
assert okf_consume.tokens_match("varene", "varemottak") is True
assert okf_consume.tokens_match("brann", "bygg") is False
def test_a_short_identifier_becomes_reachable_because_equality_has_no_floor() -> None:
# `MIN_SHARED_PREFIX` made a three-character identifier match NOTHING, not
# even itself: measured on a 629-concept bundle, `9.2` reached 0 concepts
# under the matcher while sitting verbatim in one title.
assert okf_consume.tokens_match("9.2", "9.2") is True
assert okf_consume.tokens_match("9.2", "9.3") is False
# The floor still stands for words, which is what it was measured for.
assert okf_consume.tokens_match("veg", "veg") is False
def test_the_identifier_rule_makes_a_unique_number_unique_in_the_document_frequency() -> None:
# The df is what the rarity weight reads, and 135-of-446 was the reason a
# unique requirement number weighed less than a common adjective.
concepts = _lookup_corpus()
corpus = okf_consume.searchable_text(concepts)
identifier = "3.3.1-13"
under_matcher = sum(
1
for text in corpus
if any(okf_consume.tokens_match(identifier, other) for other in okf_consume.normalise(text))
)
assert under_matcher == 1
def test_is_identifier_names_numbers_and_not_words() -> None:
assert okf_consume.is_identifier("3.3.1-13") is True
assert okf_consume.is_identifier("10.2-2") is True
assert okf_consume.is_identifier("r610.4") is True
assert okf_consume.is_identifier("brannsikring") is False
assert okf_consume.is_identifier("krav") is False
# A bare number is not an identifier: it has no separator, and every page
# number in a corpus would become one.
assert okf_consume.is_identifier("2023") is False
# THE WHOLE TOKEN, never a part of one. No token `normalise` emits can tell
# a whole-token rule from a substring rule -- the generic split keeps
# neither `.` nor `-`, so a split token can never contain an identifier --
# which is why the commitment is asserted on the function rather than on an
# output it happens to leave unchanged.
assert okf_consume.is_identifier("bilag3.3-1x") is False
def test_a_question_naming_an_identifier_delivers_the_concept_that_bears_it() -> None:
# The lookup: 300 concepts all answer `krav`, one is NAMED. Rank one, not
# rank eight -- a question that names a concept is not a search.
concepts = _lookup_corpus()
ranked = okf_consume.concept_scores(concepts, LOOKUP_QUESTION, {})
assert _rank_of(ranked, LOOKUP_GOLD) == 1
def test_two_concepts_bearing_the_same_identifier_both_reach_the_top() -> None:
concepts = _lookup_corpus(second_holder=True)
ranked = okf_consume.concept_scores(concepts, LOOKUP_QUESTION, {})
top = [concept.concept_id for concept, _, _ in ranked[:2]]
assert sorted(top) == sorted([LOOKUP_GOLD, LOOKUP_SECOND])
def test_the_lookup_is_invisible_to_a_question_that_names_no_identifier() -> None:
# The KNOWN-NEGATIVE. Every published control question on the consumer
# corpus carries zero identifiers, so this is the property that lets the
# rule ship on by default.
concepts = _lookup_corpus()
question = "Hvordan skal det sentrale vilkåret oppfylles?"
assert okf_consume.lookup_hits(concepts, question) == ()
ranked = okf_consume.concept_scores(concepts, question, {})
order = [concept.concept_id for concept, _, _ in ranked]
assert order == sorted(order)
def test_a_question_without_an_identifier_never_reads_the_corpus(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# The early return is a COST commitment, not a semantic one: an empty
# identifier set intersects to nothing anyway, so no ranking distinguishes
# the guard from its absence. What it does buy is the corpus never being
# tokenised for a question that cannot be a lookup -- measured, the lookup
# pass costs 0.024 s over 1 133 concepts when it does run.
concepts = _lookup_corpus()
calls: list[str] = []
original = okf_consume.normalise
monkeypatch.setattr(
okf_consume, "normalise", lambda text: (calls.append(text), original(text))[1]
)
assert okf_consume.lookup_hits(concepts, "Hvordan oppfylles vilkåret?") == ()
assert calls == ["Hvordan oppfylles vilkåret?"]
def test_an_identifier_no_concept_bears_changes_nothing_and_empties_nothing() -> None:
concepts = _lookup_corpus()
question = "Hva krever Krav 9.9.9-99? Gjengi det sentrale vilkåret."
assert okf_consume.lookup_hits(concepts, question) == ()
ranked = okf_consume.concept_scores(concepts, question, {})
assert len(ranked) == len(concepts)
order = [concept.concept_id for concept, _, _ in ranked]
assert order == sorted(order)
def test_the_three_spellings_of_one_identifier_are_one_lookup() -> None:
concepts = _lookup_corpus()
for dash in ("-", "–", "—"):
question = f"Hva krever Krav 3.3.1{dash}13? Gjengi det sentrale vilkåret."
assert okf_consume.lookup_hits(concepts, question) == (LOOKUP_GOLD,)
def test_the_lookup_reads_the_text_the_title_signal_reads() -> None:
# Measured on three real bundles: `req_number` carries an identifier that
# is ALSO in the title on 1 846 of 1 846 concepts that have the key, so a
# frontmatter key list buys nothing and is not declared.
identified = _synthetic("krav/only-in-id", "Alminnelig krav", "Kravet gjelder anlegget.")
concepts = [*_lookup_corpus()[:10], identified]
assert okf_consume.lookup_hits(concepts, "Hva krever Krav 3.3.1-13?") == ()
in_id = _synthetic("krav/3.3.1-13", "Alminnelig krav", "Kravet gjelder anlegget.")
assert okf_consume.lookup_hits([*concepts, in_id], "Hva krever Krav 3.3.1-13?") == (
"krav/3.3.1-13",
)
def test_the_lookup_keeps_the_fused_order_among_the_concepts_it_lifts() -> None:
# Determinism with several hits: the lifted concepts keep the order the
# fusion gave them, which is itself declared down to the id tie-break.
concepts = _lookup_corpus(second_holder=True)
ranked = okf_consume.concept_scores(concepts, LOOKUP_QUESTION, {})
hits = okf_consume.lookup_hits(concepts, LOOKUP_QUESTION)
# `lookup_hits` reports in BYTE order, and the two orders disagree here on
# purpose: the second holder's id sorts second and its fusion rank is better.
assert hits == (LOOKUP_GOLD, LOOKUP_SECOND)
lifted = [concept.concept_id for concept, _, _ in ranked[: len(hits)]]
assert lifted == [LOOKUP_SECOND, LOOKUP_GOLD]
assert [concept.concept_id for concept, _, _ in ranked[len(hits) :]] == sorted(
concept.concept_id for concept in concepts if concept.concept_id not in hits
)
# --- Step 13: the excerpt carries the key the question is asked ON (C1 step 0) -
#: A bundle purpose-built for one measurement: the two SS 5.1 `sources` forms
#: and a concept carrying neither address nor locator. It exists because the
#: two real bundles are complementary and neither exercises both forms --
#: measured 2026-09-08, K2 writes `sources` in FLOW form on 629 of 629 concepts
#: and carries `req_number` on 0, while the N500 bundle writes BLOCK form on 270
#: of 270, carries `req_number` on 269 and carries no locator key at all.
PROVENANCE = PROJECT_ROOT / "tests" / "fixtures" / "consume-provenance"
PROVENANCE_QUESTION = "Hva krever Krav 10.2-2?"
def _provenance_excerpts() -> dict[str, Mapping[str, object]]:
# Every concept, not only the delivered ones: the cut is a different
# question and would make this measurement depend on a ranking.
excerpts: dict[str, Mapping[str, object]] = {}
for concept_id in okf_consume.enumerate_concepts(PROVENANCE):
excerpt = okf_consume.excerpt_for(
okf_consume.read_concept(
PROVENANCE / f"{concept_id}.md",
bundle_root=PROVENANCE,
root_bundle_id="provenance-fixture",
)
)
assert excerpt is not None
excerpts[concept_id] = excerpt
return excerpts
def test_every_delivered_excerpt_names_the_concept_the_answer_would_cite() -> None:
# po measured 2026-09-08 (`e7ffe9e`) that the pre-pass delivers the gold
# concept at rank 1 and the model still cannot name what it is citing: the
# excerpt carried `concept_id` and `text` and no `title`. The ranking found
# the document; the delivery dropped the key.
payload = okf_consume.build_payload(PROVENANCE, question=PROVENANCE_QUESTION)
delivered = payload["excerpts"]
assert isinstance(delivered, list)
assert delivered, "the cut delivered nothing, so the assertion below is vacuous"
for excerpt in delivered:
concept = okf_consume.read_concept(
PROVENANCE / f"{excerpt['concept_id']}.md", # type: ignore[index]
bundle_root=PROVENANCE,
root_bundle_id="provenance-fixture",
)
assert excerpt["title"] == concept.title # type: ignore[index]
assert excerpt["title"] # type: ignore[index]
def test_an_excerpt_carries_the_identifier_the_question_spells_and_never_invents_one() -> None:
excerpts = _provenance_excerpts()
assert excerpts["blokkform-krav"]["req_number"] == "Krav 10.2-2"
# A concept without the key gets no key. Emitting `""` would assert that the
# producer wrote an empty identifier, which is SS 6.4's failure.
assert "req_number" not in excerpts["flytform-side"]
assert "req_number" not in excerpts["uten-adresse"]
def test_a_flow_form_address_and_its_locator_reach_the_excerpt() -> None:
excerpt = _provenance_excerpts()["flytform-side"]
assert excerpt["sources"] == [
{"resource": "Fiktivt grunnlag.pdf", "title": "Fiktivt grunnlag.pdf"}
]
assert excerpt["source_pages"] == "[4, 6]"
assert excerpt["source_offset"] == "[120, 480]"
assert "source_lines" not in excerpt
def test_a_block_form_address_reaches_the_excerpt_rather_than_becoming_an_empty_string() -> None:
# `parse_frontmatter` SKIPS indented lines, so the block form arrives as
# `''` -- present and unreadable. Measured 2026-09-08: 270 of 270 concepts
# in the largest N-bundle write it that way, so a reader that only handles
# the flow form delivers that bundle with no address at all.
excerpt = _provenance_excerpts()["blokkform-krav"]
assert excerpt["sources"] == [
{
"resource": "https://example.invalid/api/nisosts/1?languageCode=nb",
"title": "Fiktiv normal 2024",
}
]
def test_a_concept_with_no_address_and_no_locator_carries_neither_key() -> None:
excerpt = _provenance_excerpts()["uten-adresse"]
for key in ("sources", "sources_unreadable", *okf_consume.LOCATOR_KEYS):
assert key not in excerpt, key
def test_an_unreadable_address_is_named_rather_than_dropped(tmp_path: Path) -> None:
# The third state. A `sources` value this reader cannot decode is neither an
# address nor an absence, and collapsing it into either reports a fact
# nobody measured.
root = tmp_path / "bundle"
_copy_bundle(PROVENANCE, root)
target = root / "uten-adresse.md"
target.write_text(
target.read_text(encoding="utf-8").replace(
"bundle_id: provenance-fixture",
"bundle_id: provenance-fixture\nsources: not-a-list",
),
encoding="utf-8",
)
concept = okf_consume.read_concept(
target, bundle_root=root, root_bundle_id="provenance-fixture"
)
excerpt = okf_consume.excerpt_for(concept)
assert excerpt is not None
assert excerpt["sources_unreadable"] is True
assert "sources" not in excerpt
def test_the_golden_bundle_excerpt_carries_the_address_and_locator_o3_wrote() -> None:
payload = okf_consume.build_payload(GOLDEN, question="Hva sier veiledningen om krav?")
excerpts = payload["excerpts"]
assert isinstance(excerpts, list)
by_id = {entry["concept_id"]: entry for entry in excerpts} # type: ignore[index,union-attr]
first = by_id["krav/1-1/foerste-krav"]
assert first["title"] == "Foerste krav"
assert first["sources"] == [{"resource": "veiledning.md", "title": "veiledning.md"}]
assert first["source_lines"] == "[5, 8]"
assert first["source_offset"] == "[94, 176]"
def test_the_checker_refuses_an_excerpt_that_cannot_be_named() -> None:
# The SS 8 update, with its own code: a payload whose excerpts carry no
# `title` is the defect po measured, and a checker that passes it certifies
# a payload a model cannot cite from.
payload = json.loads((SKILL.parent / "references" / "example-payload.json").read_text("utf-8"))
assert okf_contract_check.check(SKILL.read_text(encoding="utf-8"), payload).findings == ()
del payload["excerpts"][0]["title"]
codes = [f.code for f in okf_contract_check.check(SKILL.read_text("utf-8"), payload).findings]
assert codes == ["excerpt_unnamed"]
Reference in a new issue View git blame Copy permalink