"""Cut an OKF bundle to one contract-conformant payload for one question. This is the **pre-pass** `docs/consumption-contract.md` SS 1 names: the deterministic program that reads the bundle, ranks its concepts, cuts them to a bounded set, and emits one payload. It decides nothing about the question being asked -- the skill does the judgement, this does the reading, the ranking and the cut (SS 2.1). **What it deliberately does not do, and the failure each refusal prevents:** - **It calls no model.** A model in the run path makes the same question at the same ref return different bytes, which is the whole property a declared cut buys over an emergent one. - **It opens no socket.** The repository requires an explicit per-run opt-in for network access; this command takes no such flag, so it can never reach one. - **It enumerates no directory.** SS 9.2 forbids that unless the named profile says the index is derived, and measured 2026-09-07, `entries_match_directory` is `True` for `STRICT_V1` alone -- for none of the profiles a segmented v0.2 bundle could have been built under. So the walk follows the INDEX TREE. That costs nothing: measured on the 629-concept K2 bundle, the index walk reaches exactly the set `rglob` finds. - **It imports nothing outside the standard library and this repository.** The package pins exactly one runtime dependency and a packaging test enforces it. It lives outside `src/`, so it never enters a wheel and no consumer's install surface changes because it exists -- the reason `tools/okf_contract_check.py` states for its own location. The entry point is `build_payload(...)`, with the CLI a thin `main()`, so lifting it into `src/` the day a consumer asks for a wheel-installed command is a move rather than a rewrite. """ from __future__ import annotations import hashlib import json import re import sys import unicodedata from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path from typing import Literal sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from llm_ingestion_okf.inbox import ( # noqa: E402 ADJUDICATION_ADJUDICATED, ADJUDICATION_PROPOSED, ADJUDICATION_STATES, ) from llm_ingestion_okf.materialize import parse_frontmatter # noqa: E402 from llm_ingestion_okf.profiles import SEGMENTED_OKF_V0_2, BundleProfile # noqa: E402 #: The profile whose index policy reads a faceted, per-directory index -- the #: shape both the in-repo golden bundle and the K2 corpus carry. Named as this #: instrument's default rather than hard-coded at each call site: a caller with #: a differently-shaped bundle passes its own. DEFAULT_PROFILE = SEGMENTED_OKF_V0_2 #: The algorithm the `ref` names, spelled in the value itself. SS 3.3 requires #: "a commit or equivalent content identity" and names no algorithm; a bundle #: that is not a git checkout has no commit, so the identity is computed. Naming #: the algorithm inline is what lets a consumer reproduce it without a document. REF_ALGORITHM = "sha256-tree" #: The one concept suffix this instrument reads, taken from the profile so a #: second literal cannot drift from it. CONCEPT_SUFFIX = DEFAULT_PROFILE.paths.concept_suffix def _walk_index_tree( bundle_root: Path, *, profile: BundleProfile ) -> tuple[tuple[str, ...], tuple[str, ...]]: """Every index file and every concept id the index tree reaches. One walk, two results, because the alternative is two walks that can disagree -- and a ref computed over a different set than the one that was read is an identity for something nobody consumed. """ index_name = profile.index.name suffix = profile.paths.concept_suffix indexes: set[str] = set() concepts: set[str] = set() pending: list[str] = [index_name] while pending: relative = pending.pop() if relative in indexes: continue indexes.add(relative) index_file = bundle_root / relative if not index_file.is_file(): continue parent = _parent_dir(relative) for line in index_file.read_text(encoding="utf-8").splitlines(): entry = profile.index.parse_entry(line) # `None` is curated prose, not a defect: the index is the one file # where this library writes beside somebody else's text. if entry is None: continue target = _join(parent, entry.target) if target is None: continue if target.rsplit("/", 1)[-1] == index_name: pending.append(target) elif target.endswith(suffix): concepts.add(target[: -len(suffix)]) return _byte_sorted(indexes), _byte_sorted(concepts) def _byte_sorted(values: set[str]) -> tuple[str, ...]: """Sorted by the UTF-8 encoding, never by the locale. A locale-dependent order makes byte-identical output a claim about the machine rather than about the bundle. """ return tuple(sorted(values, key=lambda value: value.encode("utf-8"))) def enumerate_concepts( bundle_root: Path, *, profile: BundleProfile = DEFAULT_PROFILE ) -> tuple[str, ...]: """Every concept id in the bundle, reached through the index tree. Ids are bundle-relative POSIX paths with the profile's concept suffix removed -- slash-preserving, so two same-named concepts in different documents stay two concepts. **Not `rglob`.** SS 9.2 forbids the consumer path from enumerating a directory unless the named profile says the index is derived. The index is safe to rely on here because Door B recomputes it as a projection over the whole bundle each round, which is what makes a rebuild from scratch equal an incremental update byte for byte. """ _, concepts = _walk_index_tree(bundle_root, profile=profile) return concepts def _parent_dir(relative: str) -> str: """The directory part of a bundle-relative POSIX path, or `""` at the root. Spelled here rather than as `PurePosixPath(...).parent` so the root case is `""` and not `"."` -- joining `"."` would put a `./` segment into every id at depth one, and two ids differing only by that segment are two ids for one concept. """ head, sep, _ = relative.rpartition("/") return head if sep else "" def _join(parent: str, target: str) -> str | None: """A relative index target resolved against the index's own directory. Returns `None` for a target that climbs above the bundle root or is absolute. A target escaping the root is refused rather than clamped: a clamped path names a real file the bundle never pointed at. """ if target.startswith("/"): return None parts: list[str] = parent.split("/") if parent else [] for segment in target.split("/"): if segment in ("", "."): continue if segment == "..": if not parts: return None parts.pop() continue parts.append(segment) return "/".join(parts) if parts else None def bundle_ref(bundle_root: Path, *, profile: BundleProfile = DEFAULT_PROFILE) -> str: """A content identity for the bundle: `sha256-tree:`. The digest is taken over the LF-joined, byte-sorted lines `\t` across every file the INDEX TREE reaches -- the indexes themselves and the concepts they name. **Reachable, not `rglob`, and the cost is stated rather than hidden.** A file in the directory that no index names is outside this identity. It is also outside what the pre-pass may read (SS 9.2), so an identity covering it would be an assertion about bytes this command is forbidden to look at. The property that matters holds: every byte that can reach a payload is inside the ref, so the ref moves whenever a delivered excerpt could. **Path and bytes, never mtime or size.** A digest over metadata would move on a copy and hold still on an edit that preserved length, which is the opposite of what an identity is for. """ indexes, concepts = _walk_index_tree(bundle_root, profile=profile) suffix = profile.paths.concept_suffix lines: list[bytes] = [] for relative in (*indexes, *(f"{concept}{suffix}" for concept in concepts)): path = bundle_root / relative if not path.is_file(): continue digest = hashlib.sha256(path.read_bytes()).hexdigest() lines.append(f"{relative}\t{digest}".encode()) joined = b"\n".join(sorted(lines)) return f"{REF_ALGORITHM}:{hashlib.sha256(joined).hexdigest()}" class ConsumeError(Exception): """A refusal this instrument can name. Carries a `code` for the same reason `okf_contract_check.Finding` does: one "invalid" verdict over a dozen defects is a diagnostic no caller can act on. """ def __init__(self, message: str, *, code: str) -> None: super().__init__(message) self.code = code #: The three states a CONSUMER must distinguish (SS 6.1), against the two the #: WIRE carries (`inbox.ADJUDICATION_STATES`). The difference is the whole #: point: `unknown` is not a value a producer writes, it is what the absence of #: the key means, and it is written explicitly here so a reader never has to #: infer it from a missing member. CONSUMER_ADJUDICATION_STATES = (*ADJUDICATION_STATES, "unknown") AdjudicationState = Literal["proposed", "adjudicated", "unknown"] @dataclass(frozen=True) class Concept: """One concept read off disk, with every state written rather than implied.""" path: Path concept_id: str bundle_id: str #: `True` when `bundle_id` came from the root index rather than the concept. #: Recorded rather than silently defaulted: SS 3.1 makes identity the #: `(bundle_id, concept_id)` tuple, so where the first half came from is #: part of what the payload is asserting. bundle_id_inherited: bool sha256: str okf_type: str title: str source_file: str adjudication: AdjudicationState #: `False` when the key was absent. `adjudication == "unknown"` already says #: so, but a separate flag keeps the two facts from being one inference. adjudication_present: bool frontmatter: Mapping[str, str] body: str def read_concept(path: Path, *, bundle_root: Path, root_bundle_id: str) -> Concept: """One concept file as a record. Reads; derives nothing about the question. `sha256` is the digest of the CONCEPT FILE (SS 3.2), never the `source_sha256` frontmatter key -- that one digests the source document the concept was extracted from, and conflating them would make the payload's content identity point at a PDF nobody in the chain reads. Both exist on every K2 concept, which is what makes the confusion available. """ frontmatter = parse_frontmatter(path) relative = path.relative_to(bundle_root).as_posix() #: The slash-preserving id: the bundle-relative path minus the suffix. This #: instrument's own choice, consistent with the RULE at `importer.py:244` #: and `inbox.py:1101-1102` -- but deliberately NOT `importer.import_slug`, #: which one line further down flattens the id to a single hyphenated #: segment. A flattened id fails a document-prefix match in a way that looks #: like a ranking miss rather than an id-format bug. concept_id = relative[: -len(CONCEPT_SUFFIX)] if relative.endswith(CONCEPT_SUFFIX) else relative raw = frontmatter.get("adjudication") if raw is None: adjudication: AdjudicationState = "unknown" elif raw == ADJUDICATION_PROPOSED: adjudication = "proposed" elif raw == ADJUDICATION_ADJUDICATED: adjudication = "adjudicated" else: raise ConsumeError( f"{relative} carries adjudication={raw!r}, outside the wire set " f"{ADJUDICATION_STATES}; refusing to map it to 'unknown', which " "would report 'we cannot tell whether it was judged' where the " "truth is that the bundle said something this consumer does not " "understand", code="adjudication_unknown_value", ) declared = frontmatter.get("bundle_id") return Concept( path=path, concept_id=concept_id, bundle_id=declared if declared else root_bundle_id, bundle_id_inherited=not declared, sha256=hashlib.sha256(path.read_bytes()).hexdigest(), okf_type=frontmatter.get("type", ""), title=frontmatter.get("title", ""), source_file=frontmatter.get("source_file", ""), adjudication=adjudication, adjudication_present=raw is not None, frontmatter=frontmatter, body=_body(path), ) def _body(path: Path) -> str: """The text after the frontmatter block, or the whole file when there is none.""" text = path.read_text(encoding="utf-8") lines = text.splitlines() if not lines or lines[0].strip() != "---": return text for offset, line in enumerate(lines[1:], start=2): if line.strip() == "---": return "\n".join(lines[offset:]) return text #: SS 6.2, from SPEC SS 5.3, lowest to highest. Imported from the checker's own #: constant would be circular (the checker is a separate tool); spelled here and #: held to the checker's set by the anti-drift test. TrustTier = Literal["unverified", "machine-confirmed", "human-reviewed"] #: The prefix that makes an actor a person. A PREFIX, never a substring: an #: actor id `bot/human:2` contains the literal and is a machine, and promoting #: it would be fabricated provenance produced by a matching bug. HUMAN_ACTOR_PREFIX = "human:" def trust_tier(verified_raw: str | None) -> TrustTier | None: """The tier `verified` implies, or `None` when the value cannot be read. Three inputs, three different facts, and collapsing any two is the defect: - `None` -- the key is ABSENT. SS 6.2: no `verified` key means `unverified`, and SS 6.3 forbids rejecting a concept for it. - `""` -- the key is PRESENT and this library cannot read it. Measured 2026-09-07: the line-oriented `parse_frontmatter` returns `''` for a block-form value and the full string for a flow one, so the two are distinguishable. Returning `None` here is not a tier; the caller withholds the concept under a named rule. Emitting `unverified` instead would assert a fact nobody measured, which is exactly what SS 6.4 forbids. - a flow value -- decoded, and the tier follows the actors. """ if verified_raw is None: return "unverified" if not verified_raw.strip(): return None entries = _parse_flow_mappings(verified_raw) if entries is None: return None actors: list[str] = [] for entry in entries: actor = entry.get("by") if not actor: raise ConsumeError( f"a `verified` entry names no `by` actor ({verified_raw!r}); " "refusing to tier it, because the tier IS a claim about who " "checked and there is nobody to name", code="verified_actorless", ) actors.append(actor) if not actors: return None if any(actor.startswith(HUMAN_ACTOR_PREFIX) for actor in actors): return "human-reviewed" return "machine-confirmed" def _parse_flow_mappings(value: str) -> list[dict[str, str]] | None: """A YAML flow sequence of flow mappings, or `None` when it is not one. Modelled on `structure._parse_flow_list` -- flow form only, because this library's standing rule is that a value it can write is a value it can read back. Not reused: that one splits on every comma, and `{ by: x, at: y }` carries a comma INSIDE a mapping, so it would return four fragments where there are two pairs. """ stripped = value.strip() if not (stripped.startswith("[") and stripped.endswith("]")): return None body = stripped[1:-1].strip() if not body: return [] mappings: list[dict[str, str]] = [] for chunk in _split_top_level(body, "{", "}"): item = chunk.strip() if not (item.startswith("{") and item.endswith("}")): return None pairs: dict[str, str] = {} for field in item[1:-1].split(","): key, separator, raw = field.partition(":") if separator: pairs[key.strip()] = raw.strip() mappings.append(pairs) return mappings def _split_top_level(body: str, opener: str, closer: str) -> list[str]: """Split on commas that are not inside a `{...}`.""" parts: list[str] = [] depth = 0 current: list[str] = [] for character in body: if character == opener: depth += 1 elif character == closer: depth -= 1 if character == "," and depth == 0: parts.append("".join(current)) current = [] continue current.append(character) parts.append("".join(current)) return [part for part in parts if part.strip()] # --- The budget instrument (SS 7) -------------------------------------------- #: SS 7.5 fixes no unit deliberately -- "a token is one encoder family's unit #: and fixing it would adopt one vendor's arithmetic as everyone's". This #: profile chooses utf-8 bytes of the EMITTED JSON, which the repository can #: count with no dependency at all. `tiktoken` would be runtime dependency #: number two behind a second optional extra plus a tokenizer-version fixture #: migration, bought to answer one comparison in its own unit. BUDGET_UNIT = "utf-8 bytes of emitted JSON" #: SS 7.1 requires the instrument to be NAMED, not merely used. The name is the #: function plus the one flag that changes its answer. BUDGET_INSTRUMENT = "okf_consume.measure (len of the ensure_ascii=False JSON encoding, utf-8)" #: Chosen, not derived, and the reason is a measurement rather than a taste: #: at 60 000 the largest realistic gold concept (101 313 B encoded) falls to the #: `over_budget_alone` pre-exclusion, so a CORRECT implementation would fail its #: own acceptance criteria. At 120 000 that concept fits with 18 424 B of #: headroom, and 3 of the K2 corpus's 629 concepts still cannot fit alone #: (4 at 60 000). A starting point to be moved by measurement. DEFAULT_LIMIT = 120_000 #: The known-positive artefact (SS 7.4). A SHIPPED file rather than the bundle #: under test, because a per-bundle known-positive can only be one of two #: useless things: a constant that is wrong for every bundle but one, or the #: instrument's own output, which makes `expected == measured` true by #: construction and the rule decorative. #: #: The coupling is stated rather than hidden: if this document's bytes move, the #: literal below goes stale and the pre-pass refuses until it is updated. That #: is the intended direction -- a stale known-positive is a loud failure, and #: the document is normative and not edited from this repository. KNOWN_POSITIVE_CASE = "docs/consumption-contract.md, encoded as a JSON string" #: `measure()`'s own answer for that file. Vacuous ALONE -- which is why the #: delta below exists. KNOWN_POSITIVE_EXPECTED = 10_349 #: The second, independent route. `wc -c` reports 10 060 raw bytes for the same #: file; the difference is this file's JSON quoting and escaping overhead. A #: reader can derive it without running `measure()` at all, and it moves the #: moment `measure()` changes what it counts -- which is what stops #: `expected == measured` from proving nothing. KNOWN_POSITIVE_ENCODING_DELTA = 289 _KNOWN_POSITIVE_PATH = Path(__file__).resolve().parents[1] / "docs" / "consumption-contract.md" def measure(value: str) -> int: """The cost of `value` in the unit the gate enforces. The ENCODED JSON form, because that is what the payload actually costs. A knapsack weighing `stat().st_size` while the gate measures this would let a cut computed as fitting be refused by the gate -- measured, the two differ by 7.1 % over the K2 corpus. """ return len(json.dumps(value, ensure_ascii=False).encode("utf-8")) def known_positive() -> tuple[str, int, int]: """The case, the figure expected of it, and the figure measured (SS 7.4). Takes no bundle argument on purpose: see `KNOWN_POSITIVE_CASE`. """ measured = measure(_KNOWN_POSITIVE_PATH.read_text(encoding="utf-8")) return KNOWN_POSITIVE_CASE, KNOWN_POSITIVE_EXPECTED, measured # --- Stage one: which documents are worth opening ----------------------------- #: The shortest token this instrument scores. Two characters in Norwegian are #: almost always a function word (`og`, `er`, `en`, `av`, `de`), and a matcher #: that scores them ranks every document equally. MIN_TOKEN_LENGTH = 3 #: How many leading characters two tokens must share to count as a match. #: #: THIS INSTRUMENT'S OWN CONSTANT, and a measurement rather than a preference. #: Token equality fails on Norwegian compounds: the question word `prisene` #: equals none of a price-form concept's `title`, `source_file` or path tokens. #: Plain substring containment does not save it either -- neither `prisene` nor #: `prissammenstilling` contains the other. A shared prefix does: `pris|ene` and #: `pris|sammenstilling` share 4. #: #: MEASURED HERE, 2026-09-07, over the 629-concept K2 corpus for the question #: token `prisene`: a 4-character floor matches **3** concepts -- over the #: concept id alone AND over title + `source_file` + id together, the same 3 -- #: and the price-form gold is among them. The plan this implements recorded 6 #: for the same measurement; 6 is not reproducible with this rule, and the #: number that is reproducible is the one carried here. A 3-character floor #: over-matches Norwegian function words. MIN_SHARED_PREFIX = 4 _TOKEN_SPLIT_RE = re.compile(r"[^0-9a-zà-öø-ÿ]+") def normalise(text: str) -> tuple[str, ...]: """Text as comparable tokens: NFC first, then casefold, then split. NFC FIRST is load-bearing and not tidiness. macOS hands filenames over decomposed, so `å` arrives as `a` + U+030A; the combining ring is not a word character, so an un-normalised split turns `årlig` into `a` and `rlig` and the term is silently lost. `æ` and `ø` have no canonical decomposition, so a test built on either passes while the bug is live -- which is why the known-positive for this function uses `å`. """ folded = unicodedata.normalize("NFC", text).casefold() return tuple(token for token in _TOKEN_SPLIT_RE.split(folded) if len(token) >= MIN_TOKEN_LENGTH) def tokens_match(left: str, right: str) -> bool: """Whether two tokens share a leading prefix of at least `MIN_SHARED_PREFIX`. Symmetric, and it degrades to equality for short tokens: two 4-character tokens match only if they are the same word. """ limit = min(len(left), len(right)) if limit < MIN_SHARED_PREFIX: return False shared = 0 while shared < limit and left[shared] == right[shared]: shared += 1 return shared >= MIN_SHARED_PREFIX def _overlap(question_tokens: Sequence[str], candidate: str) -> int: """How many of the question's tokens the candidate text answers to.""" candidate_tokens = normalise(candidate) return sum( 1 for token in question_tokens if any(tokens_match(token, other) for other in candidate_tokens) ) def document_scores( bundle_root: Path, question: str, *, profile: BundleProfile = DEFAULT_PROFILE ) -> dict[str, float]: """One score per top-level document, from the indexes and the path alone. A "document" is a top-level entry: a directory, or -- per the corpus's own shape -- a concept sitting at the root, which has no directory to inherit from and is therefore its own document. Measured on K2, 11 of 629 concepts are root-level, and they are exactly the 11 carrying neither `adjudication` nor `bundle_id`; scoring them as members of some parent would put one bug in three places. Reads the INDEX TREE only. No directory is enumerated here or anywhere else in this command (SS 9.2). """ question_tokens = normalise(question) indexes, concepts = _walk_index_tree(bundle_root, profile=profile) scores: dict[str, float] = {} for concept_id in concepts: document = concept_id.split("/", 1)[0] scores.setdefault(document, 0.0) scores[document] += float(_overlap(question_tokens, concept_id.replace("/", " "))) for relative in indexes: document = relative.split("/", 1)[0] if document == profile.index.name: continue scores.setdefault(document, 0.0) for line in (bundle_root / relative).read_text(encoding="utf-8").splitlines(): entry = profile.index.parse_entry(line) if entry is None: continue scores[document] += float(_overlap(question_tokens, entry.label)) return scores