"""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 argparse import hashlib import json import math 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.corpus import LOG_NAME # noqa: E402 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 ( # noqa: E402 RESERVED_OKF_TYPE, SEGMENTED_OKF_V0_2, BundleProfile, ) #: 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 name = target.rsplit("/", 1)[-1] if name == index_name: pending.append(target) elif name == LOG_NAME: # A run's own log -- reachable, but metadata, never a # concept. Counted as one it inflated the K2 walk to 630 # against a 629-concept bundle and let the log rank and cut # like real content (measured, S7 F2). # # THE PRODUCER NO LONGER WRITES THE LINK (2026-09-08, after a # consumer's navigator followed it and returned 630 where this # counts 629). This branch is not dead: every bundle built # between `95eb271` and that change carries it, including the # ones consumers are reading today. continue 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 #: The identifier the producer wrote, or `""` when there is no key. The #: number a lookup question is asked ON, and the one thing a reader needs to #: name the concept the answer rests on. req_number: str #: The SS 5.1 address entries, in either YAML form. sources: tuple[Mapping[str, str], ...] #: `True` when a `sources` key was there, whatever this reader made of it. #: With `sources == ()` that is the third state: present and unreadable. sources_present: bool #: The `LOCATOR_KEYS` this concept carries, values as written. Absent keys #: are absent, never `""`: SS 6.4 forbids reading the absence of a #: conditionally-written field as the negation of what it asserts. locators: Mapping[str, str] 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") entries, sources_present = read_sources(_frontmatter_lines(path)) 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, req_number=frontmatter.get("req_number", ""), sources=entries, sources_present=sources_present, locators={ key: value for key, value in frontmatter.items() if key.startswith(SOURCE_KEY_PREFIX) and value.strip() }, 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 locator keys O3 (`b6a8c8b`) writes on every SEGMENTED concept, in the #: order they are emitted so a reader comparing an excerpt against the concept #: file reads one sequence. The ADDRESS is SPEC SS 5.1's `sources`; these are #: this library's OWN top-level keys, because SS 5.1 has no field for a place #: within a resource and the guard rejects every route to putting one inside a #: `sources` entry. Their VALUES pass through as the frontmatter's own strings: #: `source_pages: [2, 27]` reaches the payload as `"[2, 27]"`, which is what #: makes an excerpt greppable against the file it came from. LOCATOR_KEYS = ( "source_pages", "source_sheet", "source_rows", "source_lines", "source_offset", ) #: What a locator key looks like to a reader that does not know the producer. #: The pass-through rule is this PREFIX and not `LOCATOR_KEYS`, which is a list #: of the producers someone thought of: measured 2026-09-08, 269 of 274 concepts #: in the N500 bundle carry `source_element_id`, a locator that repository chose #: under the same rule ("the key says what it indexes") and this library never #: writes. An allowlist drops it, and the excerpt then names a document without #: naming the place in it. A PREFIX, never a substring -- `resource_owner` #: contains the literal and is not a locator, and promoting it would be #: fabricated provenance produced by a matching bug. SOURCE_KEY_PREFIX = "source_" def _frontmatter_lines(path: Path) -> list[str]: """The raw lines between the two `---` fences, indentation intact. `parse_frontmatter` SKIPS indented lines on purpose -- a nested `title:` arriving later would SUBSTITUTE for the document's. That refusal is right for a flat mapping and it is why the block form has to be read from the raw lines instead. """ lines = path.read_text(encoding="utf-8").splitlines() if not lines or lines[0].strip() != "---": return [] block: list[str] = [] for line in lines[1:]: if line.strip() == "---": break block.append(line) return block def read_sources(lines: Sequence[str]) -> tuple[tuple[Mapping[str, str], ...], bool]: """The SS 5.1 address entries, and whether the key was there at all. Three states, kept apart because collapsing any two reports something nobody measured: `((), False)` the concept has no `sources` key; `((), True)` it has one this reader cannot decode; a non-empty tuple, the entries. BOTH YAML forms are read, and that is a measurement rather than a preference. Measured 2026-09-08: K2 writes the flow form on 629 of 629 concepts, the N500 bundle writes the block form on 270 of 270. A reader handling one form delivers the other bundle with no address at all -- and for N500 there is nothing else, because it carries zero locator keys. Reading the block form is not a licence to WRITE it: this library's line-oriented parser still cannot round-trip block lists, so the emission rule (flow only) is untouched. """ for position, line in enumerate(lines): if line[:1] in (" ", "\t") or not line.startswith("sources:"): continue value = line.partition(":")[2].strip() if value: flow = _parse_flow_mappings(value) if flow is None: return (), True return tuple(flow), True entries: list[dict[str, str]] = [] for nested in lines[position + 1 :]: if not nested.strip(): continue if nested[:1] not in (" ", "\t"): break item = nested.strip() if item.startswith("- "): entries.append({}) item = item[2:].strip() elif not entries: # An indented line before any `- ` opens no entry. Refused # rather than folded into one, which would invent an entry the # document does not have. return (), True key, separator, raw = item.partition(":") if not separator: return (), True entries[-1][key.strip()] = raw.strip() if not entries: return (), True return tuple(entries), True return (), False # --- 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 = 12_563 #: The second, independent route. `wc -c` reports 12 227 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 = 336 _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: a question's inflected noun #: equals none of the tokens in a concept's `title`, `source_file` or path when #: the concept spells the same subject as a compound. Plain substring #: containment does not save it either -- of `varene` and `varemottak`, neither #: contains the other. A shared prefix does: `vare|ne` and `vare|mottak` share 4. #: #: MEASURED 2026-09-07 over a 629-concept corpus, for one question's subject #: token: a 4-character floor matches **3** concepts -- over the concept id #: alone AND over title + `source_file` + id together, the same 3 -- and the #: gold concept 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à-öø-ÿ]+") #: Every dash a source spells an identifier's separator with, folded to the #: ASCII hyphen. NFC folds NONE of them, so `10.2—2` from a document viewer and #: `10.2-2` from a person typing the question are two different tokens until #: this table runs. The set is the Unicode dash block plus the minus sign. _DASH_TO_HYPHEN = str.maketrans(dict.fromkeys("‐‑‒–—―−", "-")) #: An identifier: NUMERIC groups joined by `.` or `-`, with an optional letter #: prefix that touches its digits without a separator (`R610.4`). #: #: THE LETTERS ARE THE POINT, and this pattern was narrowed by a measurement #: rather than written this way. A rule that joined alphanumeric groups across #: a separator swallowed a whole document slug -- `...bilag-3-6-premissrapport- #: akustikk` became ONE token because `3-6` sits inside it -- and that #: document's score for a question naming its subject fell from 0.735 to 0.0, #: taking a hit@8 row with it. Only digits may stand on either side of a #: separator, so a hyphenated word keeps its words. _IDENTIFIER_RE = re.compile(r"[a-zà-öø-ÿ]*[0-9]+(?:[.-][0-9]+)+") def normalise(text: str) -> tuple[str, ...]: """Text as comparable tokens: NFC, casefold, dash-fold, 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 `å`. IDENTIFIERS SURVIVE THE SPLIT. Splitting on every non-alphanumeric turns a requirement number into digit runs, and `MIN_TOKEN_LENGTH` then removes them: `Krav 10.2—2` reached the ranker as `krav` alone, a word every concept in a standards bundle carries, so the ranking became a corpus-wide tie and the named requirement was withheld `below_k` (measured 2026-09-08 on three bundles). The floor stays -- a bare `10` matches every page number in a corpus -- and the identifier is exempted from it rather than the floor lowered for everyone. """ folded = unicodedata.normalize("NFC", text).casefold().translate(_DASH_TO_HYPHEN) tokens: list[str] = [] position = 0 for match in _IDENTIFIER_RE.finditer(folded): tokens.extend(_split(folded[position : match.start()])) tokens.append(match.group()) position = match.end() tokens.extend(_split(folded[position:])) return tuple(tokens) def _split(folded: str) -> list[str]: """The generic split, on text already folded by `normalise`.""" return [token for token in _TOKEN_SPLIT_RE.split(folded) if len(token) >= MIN_TOKEN_LENGTH] def is_identifier(token: str) -> bool: """Whether a token is a NUMBER a document is known by, rather than a word. The whole token, never a part of one: `normalise` emits an identifier as one token, so a full match is what "this token is an identifier" means. A bare number is not one -- `2023` has no separator, and every page number in a corpus would become an identifier if it were. """ return _IDENTIFIER_RE.fullmatch(token) is not None 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. **AN IDENTIFIER MATCHES BY EQUALITY ALONE**, and that is a defect fix measured on the case it costs most rather than a preference. The prefix rule was measured for Norwegian compounds, where `vare|ne` and `vare|mottak` share a stem; a requirement number has no stem, and four leading characters of `3.3.1-13` are four leading characters of every requirement in section 3.3. MEASURED 2026-09-08 on a 446-concept bundle: the unique identifier `3.3.1-13` reached **135** concepts under the prefix rule and **1** under equality, which made the rarity weight rank a common adjective as more informative than the number naming the document (`docs/2026-09-08-sjeldenhetsvekt.md` SS 3). `54a0bc2` SS 1 named this class -- "`df` measured over the colliding matcher measures collision breadth, not rarity" -- and this is that sentence applied to the identifier itself. """ if is_identifier(left) or is_identifier(right): # No floor, either: `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, so a document known by a short number was # unreachable by that number. return left == right 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 #: One declared vocabulary family, spelled once: within it, any term answers to #: any other. OFF by default and reachable only through `--cost-vocabulary`. #: #: WHAT IT IS FOR, and the failure it addresses. Measured 2026-09-08 over a #: 629-concept corpus: a mandate-shaped question about cost ranked that #: corpus's one priced table 249th of 269 lexical candidates, because the #: question said `kostnadsbesparelser` and the document said `pris` -- two #: words with no shared prefix. Its title score and its document score were #: both 0. No value of `k` closes a gap in the VOCABULARY. #: #: WHAT IT IS NOT. Each member must be at least `MIN_SHARED_PREFIX` characters #: or it can never match anything (`sum` is 3 and does not match `Summen`; it #: was dropped for that reason, not by taste). The bridge is symmetric and #: needs a family term on BOTH sides, so it can widen a cost question towards a #: cost document and never towards an arbitrary one. Norwegian, and stated as #: such: a corpus in another language gets nothing from it. #: #: HONESTY, measured rather than asserted: on that corpus the whole effect #: rests on `kost` and `pris`. Removing either returns the priced table to rank #: 249; removing any other member moves it not at all, and three members match #: nothing in that corpus. They are kept because dropping a term for being #: absent from ONE corpus fits the list to that corpus. COST_VOCABULARY = ( "beløp", "budsjett", "enhet", "honorar", "kost", "kroner", "mengde", "pris", "utgift", "vederlag", ) def in_cost_vocabulary(token: str) -> bool: """Whether one token belongs to the declared family, by the same prefix rule.""" return any(tokens_match(token, member) for member in COST_VOCABULARY) def question_uses_cost_vocabulary(question: str) -> bool: """Whether the QUESTION opens the bridge. The gate is the question, never the flag. A question naming no term in the family gets byte-identical bytes with the flag set, which is what keeps the flag a widening of one question class rather than a second ranker. """ return any(in_cost_vocabulary(token) for token in normalise(question)) def searchable_text(concepts: Sequence["Concept"]) -> list[str]: """The text a concept is scored against, one string per concept. The same two fields the ranker's two lexical signals read -- title plus id, and body -- joined, so a `df` counted here is a `df` over exactly what a hit can be scored on. Counting rarity over one field and matching on another would weight a token by how rare it is somewhere it is not read. """ return [ f"{concept.title} {concept.concept_id.replace('/', ' ')} {concept.body}" for concept in concepts ] def rarity_weights(question_tokens: Sequence[str], corpus: Sequence[str]) -> dict[str, float]: """What one hit on each question token is worth, from the bundle alone. `log(N / df)`: `N` concepts, and `df` the number of them bearing the token under the SAME prefix rule a hit is scored with. No constant is set by hand and no class of token is declared anywhere -- a word every concept carries weighs exactly `log(1) == 0` of itself, and an identifier one concept carries takes the corpus's maximum of itself. **What this is NOT.** `54a0bc2` swept smoothed IDF as a GATE -- a threshold below which a concept is withheld -- and falsified it: no threshold zeroed both known-negatives while any positive question still reached its gold document. That result stands and is not re-litigated here. This is the other use: an ordering inside the candidate set, with the gate untouched and `lexical` still a count. A ranking cannot withhold anything, so the failure mode that refuted the gate has no counterpart here. **`df` is measured over the colliding matcher, and so measures collision breadth as well as rarity** (`54a0bc2` § 1: every `brann*` compound shares four leading characters). Inherited deliberately rather than fixed here: the weight must agree with the matcher it weights, and changing the matcher is a different change with its own measurement. One pass over the corpus. A token borne by no concept takes the weight of a token borne by one -- it is never consumed, because a token that matches nothing is never a hit. """ total = len(corpus) if total == 0: return {token: 0.0 for token in question_tokens} counts = {token: 0 for token in question_tokens} for text in corpus: candidate_tokens = normalise(text) for token in counts: if any(tokens_match(token, other) for other in candidate_tokens): counts[token] += 1 return { token: math.log(total / count) if count else math.log(total) for token, count in counts.items() } def lookup_hits(concepts: Sequence["Concept"], question: str) -> tuple[str, ...]: """The concepts a question NAMES, rather than the ones it describes. A question carrying an identifier that sits VERBATIM in a concept's title or id is a lookup, not a search: the reader already knows which document they want and is spelling its number. Returns those concepts' ids, byte sorted so several holders of one number arrive in a declared order, and the EMPTY tuple whenever the question carries no identifier -- which is what makes this rule invisible to every question that is not a lookup. **It reads the text the title-and-id signal reads, and no frontmatter key list is declared.** Measured 2026-09-08 on three real bundles: of the 1 846 concepts carrying a `req_number`, the identifier in that key is ALSO in the title on **1 846** of them, and on **0** does the key carry an identifier the title lacks. A key list would therefore have bought nothing here and would have been a constant no measurement asked for. A bundle whose identifiers live only in frontmatter is not served by this rule, and that is stated rather than guessed at. **Verbatim after `normalise`, so the three spellings of one identifier are one lookup** (`_DASH_TO_HYPHEN`) -- but `.` and `-` are NOT interchangeable, so a question spelling `1.10` does not find a document whose id spells it `1-10`. Measured and left open. """ identifiers = {token for token in normalise(question) if is_identifier(token)} if not identifiers: return () return tuple( sorted( concept.concept_id for concept in concepts if identifiers & set(normalise(f"{concept.title} {concept.concept_id.replace('/', ' ')}")) ) ) def _overlap( question_tokens: Sequence[str], candidate: str, *, cost_vocabulary: bool = False, weights: Mapping[str, float] | None = None, ) -> float: """What the candidate text answers of the question. A COUNT when `weights` is None -- one per question token the candidate answers to, which is what every caller got before rarity weighting existed and what the cut still reads. With `weights`, the sum of those tokens' rarity weights instead. """ candidate_tokens = normalise(candidate) bridged = cost_vocabulary and any(in_cost_vocabulary(token) for token in candidate_tokens) return sum( 1 if weights is None else weights.get(token, 1.0) for token in question_tokens if any(tokens_match(token, other) for other in candidate_tokens) or (bridged and in_cost_vocabulary(token)) ) def document_scores( bundle_root: Path, question: str, *, profile: BundleProfile = DEFAULT_PROFILE, cost_vocabulary: bool = False, weights: Mapping[str, float] | None = None, ) -> dict[str, float]: """One score per top-level document, from the indexes and the paths 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. **The score is a DENSITY, not a sum, and that is a correction rather than a preference.** A sum over a document's units grows with the number of units, so a large document outscores a small one on size alone. Measured on K2 for the price question: the competition document sums to 6.0 over 79 concepts (0.076 each) and the price document to 2.0 over 1 (2.0), so the sum ranks the larger document three times higher while the density ranks the smaller one twenty-six times higher. A prior that grows with size is measuring size. **Stated because it bears on how the hit@k number should be read:** this defect was found by running the one question whose gold was confirmed independently, and fixing it therefore happened with that answer visible. The fix is justified by the scoring function's own arithmetic rather than by the answer -- but the ranker is not blind to that one case, and the measurement document says so beside the number. Reads the INDEX TREE only. No directory is enumerated here or anywhere else in this command (SS 9.2). """ question_tokens = normalise(question) bridge = cost_vocabulary and question_uses_cost_vocabulary(question) indexes, concepts = _walk_index_tree(bundle_root, profile=profile) totals: dict[str, float] = {} units: dict[str, int] = {} def record(document: str, overlap: float) -> None: totals[document] = totals.get(document, 0.0) + float(overlap) units[document] = units.get(document, 0) + 1 for concept_id in concepts: record( concept_id.split("/", 1)[0], _overlap( question_tokens, concept_id.replace("/", " "), cost_vocabulary=bridge, weights=weights, ), ) for relative in indexes: document = relative.split("/", 1)[0] if document == profile.index.name: continue for line in (bundle_root / relative).read_text(encoding="utf-8").splitlines(): entry = profile.index.parse_entry(line) if entry is None: continue record( document, _overlap(question_tokens, entry.label, cost_vocabulary=bridge, weights=weights), ) return {document: totals[document] / units[document] for document in totals} # --- Stage two: which concepts inside those documents ------------------------- #: Reciprocal Rank Fusion's smoothing constant. Lifted IN METHOD from the #: reference implementation this repository read (`wiki-advise/rank.mjs:82`), #: not in code and not as a quality claim: that engine's recall figures were #: measured on an LLM candidate-generation pass this pre-pass deliberately does #: not have, so its numbers say nothing about this one. #: #: RRF is chosen because it consumes RANKS ONLY. There is no score #: normalisation to get wrong, no weight to tune against a test set, and the #: fusion is invariant to any monotone transform of the individual signals. RRF_K = 60 def concept_scores( concepts: Sequence[Concept], question: str, document_score: Mapping[str, float], *, cost_vocabulary: bool = False, weights: Mapping[str, float] | None = None, lookup: bool = True, ) -> list[tuple[Concept, float, int]]: """Every concept, ordered best first, fused from three signals by RRF. The signals: (1) the question against the concept's title and the segments of its id, (2) the question against the body, (3) the stage-one score of the document the concept belongs to. **Ties break lexicographically by `concept_id`, at both the per-signal sort and the fused sort.** Declared rather than inherited from dict insertion order, so reproducibility is a property of the input SET and not of the order it happened to arrive in. **No float reaches the payload.** These scores order the cut; only ranks and whole byte counts are emitted. `lookup=False` isolates the FUSION from the lookup partition below it, and exists because two stages sharing one output cannot otherwise be measured apart: the tests that state what the rarity weight does to the fusion, and the harness that measures a lookup's effect, both need the fusion's own order. No caller in the run path sets it and the CLI does not expose it. The third element of each tuple is the concept's OWN lexical overlap -- signals 1 and 2 only, with the document prior excluded. The cut needs it separately: a concept that answers nothing in the question, sitting in a document that does, is a GUESS, and a guess is the one thing a declared cut must not deliver. """ question_tokens = normalise(question) bridge = cost_vocabulary and question_uses_cost_vocabulary(question) titles = { concept.concept_id: f"{concept.title} {concept.concept_id.replace('/', ' ')}" for concept in concepts } signals: list[dict[str, float]] = [ { concept.concept_id: float( _overlap( question_tokens, titles[concept.concept_id], cost_vocabulary=bridge, weights=weights, ) ) for concept in concepts }, { concept.concept_id: float( _overlap(question_tokens, concept.body, cost_vocabulary=bridge, weights=weights) ) for concept in concepts }, { concept.concept_id: document_score.get(concept.concept_id.split("/", 1)[0], 0.0) for concept in concepts }, ] fused: dict[str, float] = {concept.concept_id: 0.0 for concept in concepts} for signal in signals: # Sort by score descending, then by id ascending -- the declared # tie-break, applied before a rank is ever read. order = sorted(signal, key=lambda key: (-signal[key], key)) for position, concept_id in enumerate(order, start=1): fused[concept_id] += 1.0 / (RRF_K + position) lexical = ( { concept.concept_id: int(signals[0][concept.concept_id] + signals[1][concept.concept_id]) for concept in concepts } if weights is None # A COUNT even when the signals are weighted. The cut reads this, and a # word every concept carries weighs zero: were `lexical` the weighted # sum, a concept matching only that word would fall to # `no_lexical_match` -- turning a ranking change into the GATE # `54a0bc2` falsified. The gate is a different axis and stays where it # was, at the price of one more pass over the same two fields. else { concept.concept_id: int( _overlap(question_tokens, titles[concept.concept_id], cost_vocabulary=bridge) + _overlap(question_tokens, concept.body, cost_vocabulary=bridge) ) for concept in concepts } ) by_id = {concept.concept_id: concept for concept in concepts} ranked_ids = sorted(fused, key=lambda key: (-fused[key], key)) named = set(lookup_hits(concepts, question)) if lookup else set() if named: # THE LOOKUP LANDS BEFORE THE FUSION'S OUTPUT IS READ, and it is a # partition rather than a fourth signal. The form was chosen by # measurement, not by preference: a fourth RRF signal was simulated on # the same three bundles first and put the named concept at rank # **26 / 15 / 19** of 446 / 1 133 / 270 -- none of them delivered. RRF # consumes RANKS ONLY, so any single signal contributes at most # `1/(RRF_K + 1)` however certain it is, and a concept the question # NAMES cannot outbid three signals that merely describe it # (`docs/2026-09-08-sjeldenhetsvekt.md` SS 4 predicted exactly this). # # STABLE: the named concepts keep the order the fusion gave them, and # so does everything else, so nothing here depends on dict order. ranked_ids = [key for key in ranked_ids if key in named] + [ key for key in ranked_ids if key not in named ] return [ (by_id[concept_id], fused[concept_id], lexical[concept_id]) for concept_id in ranked_ids ] # --- The cut (SS 5, SS 7, SS 9.1) --------------------------------------------- #: Every rule this instrument may drop a concept under, spelled once. CLOSED: #: a drop with no rule is the silent cut SS 5.3 exists to forbid, and a rule #: invented at the drop site is a vocabulary no consumer can be held to. WITHHOLDING_RULES = ( "verdict_layer_excluded", "verified_unreadable", "no_lexical_match", "over_budget_alone", "below_k", "over_budget_after_knapsack", ) #: The knapsack's weight granularity, in bytes. Bucketing keeps the DP table #: small; bucketing UP the item and DOWN the capacity keeps the error one-sided, #: so the pack may under-deliver by a bucket and can never over-spend. WEIGHT_BUCKET = 500 def excerpt_for(concept: Concept) -> dict[str, object] | None: """One concept as a payload excerpt, or `None` when it cannot be tiered. Carries the SS 8 members plus three the contract permits and this profile needs: - **`text`** -- the concept body. SS 8 names no content member, but SS 1 defines an excerpt as "one delivered unit of bundle CONTENT", and without a body the budget gate would measure a two-kilobyte skeleton while the skill went to the bundle itself, breaking SS 2.2. - **`text_sha256`** -- the digest of the delivered bytes. `sha256` is the digest of the WHOLE concept file (SS 3.2), so without this second digest a consumer holds an identity that cannot verify what it was handed. - **`bundle_id_inherited`** -- whether the first half of the SS 3.1 identity tuple came from the concept or from the root index. And, since C1 step 0, the keys that let a reader NAME what it is citing. `title` is unconditional (SS 8, this revision); `req_number`, `sources` and each locator are written only when the concept carries them, because a key with an empty value asserts that the producer wrote one. po measured 2026-09-08 (`e7ffe9e`) that the pre-pass delivered the gold concept at rank 1 on three bundles while the model could not name it: the ranking found the document and the delivery dropped the key. Trailing whitespace is stripped per line: a spreadsheet render is padded to hundreds of trailing spaces per line, and unstripped, most of a budget goes on padding. """ tier = trust_tier(concept.frontmatter.get("verified")) if tier is None: return None text = "\n".join( line.rstrip() for line in unicodedata.normalize("NFC", concept.body).split("\n") ) excerpt: dict[str, object] = { "bundle_id": concept.bundle_id, "concept_id": concept.concept_id, "sha256": concept.sha256, "adjudication": concept.adjudication, "trust_tier": tier, "bundle_id_inherited": concept.bundle_id_inherited, "title": concept.title, } if concept.req_number: excerpt["req_number"] = concept.req_number if concept.sources: excerpt["sources"] = [dict(entry) for entry in concept.sources] elif concept.sources_present: # The third state, written rather than silently dropped: the concept # HAS an address and this reader could not decode it. excerpt["sources_unreadable"] = True excerpt.update(concept.locators) excerpt["text_sha256"] = hashlib.sha256(text.encode("utf-8")).hexdigest() excerpt["text"] = text return excerpt def excerpt_weight(excerpt: Mapping[str, object]) -> int: """What this excerpt costs by the gate's own instrument. The ENCODED JSON object, never `stat().st_size`: measured, the two differ by 7.1 % over the K2 corpus, and a cut computed as fitting would then be refused by a gate measuring different bytes. """ return len(json.dumps(excerpt, ensure_ascii=False).encode("utf-8")) def knapsack(items: Sequence[tuple[float, int]], *, capacity: int) -> tuple[int, ...]: """The exact 0/1 knapsack: indices of the highest-value subset that fits. Exact rather than greedy-by-density, which has an unbounded approximation factor -- and over at most `k` items the DP is microseconds, so the approximation buys nothing. Deterministic: a subset replaces the incumbent only on a STRICT improvement, so equal-value subsets resolve to the one built from earlier items, and the caller sorts the pool by `concept_id` first. """ best: list[tuple[float, tuple[int, ...]]] = [(0.0, ()) for _ in range(capacity + 1)] for index, (value, weight) in enumerate(items): if weight > capacity: continue for room in range(capacity, weight - 1, -1): candidate_value = best[room - weight][0] + value if candidate_value > best[room][0]: best[room] = (candidate_value, (*best[room - weight][1], index)) return max(best, key=lambda entry: entry[0])[1] def cut( ranked: Sequence[tuple[Concept, float, int]], *, k: int, limit: int, reserve_top_rank: bool = False, ) -> tuple[tuple[dict[str, object], ...], tuple[tuple[str, str], ...], tuple[str, int] | None]: """The ranked concepts split into delivered excerpts, named drops, and the reservation that was made, if any. **The partition is the invariant, not a consequence.** Every considered concept lands in exactly one of the two, so `considered == withheld + delivered` closes by construction rather than by a count computed twice. Exclusions run before the pack, each naming its rule, because "it did not fit" and "it could never be delivered" are different facts about the cut. **A concept answering nothing in the question is withheld, never ranked into the top k as filler.** Without that rule a question with no answer in the bundle still returns eight excerpts -- a confident guess wearing a denominator -- and hit@k over such a ranker measures the corpus's size rather than the ranker. **`reserve_top_rank` (default off) buys the highest-ranked candidate its bytes before the pack runs.** The DP maximises a SUM of fused scores, so a single candidate costing a large share of the budget loses to enough small ones no matter how far ahead it ranks -- measured, a top-ranked excerpt worth 56.5 % of the budget is evicted as soon as the shortlist holds enough alternatives, which makes `k` a dial that can remove the one concept a question was asked about. The reservation makes rank one a floor rather than a bid, and the pack fills what is left. It runs AFTER the `over_budget_alone` pre-exclusion, never before: a reservation for an excerpt the budget can never hold would deliver bytes the gate refuses. """ withheld: list[tuple[str, str]] = [] candidates: list[tuple[Concept, float, dict[str, object], int]] = [] for concept, score, lexical in ranked: # SS 9.1: a TYPE check at every level, never a path filter. Case-folded # against the profile's own constant, because `type: Log` (capital L) # really occurs in the corpus. if concept.okf_type.casefold() == RESERVED_OKF_TYPE: withheld.append((concept.concept_id, "verdict_layer_excluded")) continue if lexical == 0: withheld.append((concept.concept_id, "no_lexical_match")) continue excerpt = excerpt_for(concept) if excerpt is None: withheld.append((concept.concept_id, "verified_unreadable")) continue weight = excerpt_weight(excerpt) if weight > limit: withheld.append((concept.concept_id, "over_budget_alone")) continue candidates.append((concept, score, excerpt, weight)) for concept, _, _, _ in candidates[k:]: withheld.append((concept.concept_id, "below_k")) shortlist = candidates[:k] # The DP POOL is sorted by `concept_id`, so which of two equal-value subsets # wins is a property of the input set rather than of the order the ranker # happened to emit. The OUTPUT is not: excerpts come back in fused-rank # order, because the rank is what a hit@k measurement reads, and an id-sorted # list would silently turn "position in the payload" into a different number # from "position in the ranking". pool = sorted(shortlist, key=lambda entry: entry[0].concept_id) reserved: tuple[str, int] | None = None room = limit if reserve_top_rank and shortlist: # `shortlist` is in fused-rank order, so its first entry IS the # top-ranked candidate -- not the heaviest, and not the first by id. top = shortlist[0] reserved = (top[0].concept_id, top[3]) room = limit - top[3] pool = [entry for entry in pool if entry[0] is not top[0]] capacity = room // WEIGHT_BUCKET packed = { id(pool[index][0]) for index in knapsack( tuple((score, -(-weight // WEIGHT_BUCKET)) for _, score, _, weight in pool), capacity=capacity, ) } if reserved is not None: packed.add(id(shortlist[0][0])) delivered: list[dict[str, object]] = [] for concept, _, excerpt, _ in shortlist: if id(concept) in packed: delivered.append({**excerpt, "rank": len(delivered) + 1}) else: withheld.append((concept.concept_id, "over_budget_after_knapsack")) withheld.sort() return tuple(delivered), tuple(withheld), reserved # --- The payload (SS 8) ------------------------------------------------------- #: SS 8.2: present so a reader can tell which revision it is holding. CONTRACT_REVISION = "okf-consumption/1" #: `--k` caps the DELIVERED set. The budget is the gate; this is a second, #: cheaper bound so a question matching half the corpus does not run a #: 300-item DP to discover the same answer. DEFAULT_K = 8 def root_bundle_id_of(bundle_root: Path, *, profile: BundleProfile = DEFAULT_PROFILE) -> str: """The `bundle_id` the root index declares, or a refusal naming which half of SS 3.1's identity tuple is missing. Shared with the skill generator, so the two agree on what makes a directory a readable bundle.""" root_index = bundle_root / profile.index.name if not root_index.is_file(): raise ConsumeError( f"{bundle_root} carries no {profile.index.name}, so there is no index " "tree to walk and no way to read the bundle without enumerating a " "directory, which SS 9.2 forbids", code="bundle_unreadable", ) declared = parse_frontmatter(root_index).get("bundle_id", "") if not declared: raise ConsumeError( f"{root_index} declares no `bundle_id`; identity across bundles is " "the (bundle_id, concept_id) tuple (SS 3.1) and half of it is missing", code="bundle_id_missing", ) return declared def build_payload( bundle_root: Path, *, question: str, k: int = DEFAULT_K, limit: int = DEFAULT_LIMIT, profile: BundleProfile = DEFAULT_PROFILE, cost_vocabulary: bool = False, reserve_top_rank: bool = False, rarity_weight: bool = False, ) -> dict[str, object]: """One bundle plus one question, cut to one contract-conformant payload. Pure with respect to the clock and the network: the same `(bundle_root, question, k, limit, cost_vocabulary, reserve_top_rank, rarity_weight)` at the same bytes returns the same object, every time. """ case, expected, measured = known_positive() if expected != measured: # SS 7.4: the instrument reports NONE of its own numbers until it has # reproduced a known figure. Refusing here rather than emitting a # payload with a failing known-positive is the difference between an # instrument that has been shown to count and one that merely says so. raise ConsumeError( f"the budget instrument's known-positive ({case}) expected {expected} " f"and measured {measured}; refusing to report any figure until the " "two agree", code="instrument_unvalidated", ) root_bundle_id = root_bundle_id_of(bundle_root, profile=profile) concept_ids = enumerate_concepts(bundle_root, profile=profile) concepts = [ read_concept( bundle_root / f"{concept_id}{profile.paths.concept_suffix}", bundle_root=bundle_root, root_bundle_id=root_bundle_id, ) for concept_id in concept_ids ] weights = ( rarity_weights(normalise(question), searchable_text(concepts)) if rarity_weight else None ) ranked = concept_scores( concepts, question, document_scores( bundle_root, question, profile=profile, cost_vocabulary=cost_vocabulary, weights=weights, ), cost_vocabulary=cost_vocabulary, weights=weights, ) matched = sum(1 for _, _, lexical in ranked if lexical > 0) delivered, withheld, reserved = cut(ranked, k=k, limit=limit, reserve_top_rank=reserve_top_rank) spent = sum(excerpt_weight(excerpt) for excerpt in delivered) if matched and not delivered: # SS 7.3: a finding requiring a decision, never something to retry # narrower. Distinguished from the honest empty result BY THE # DENOMINATOR: concepts answered this question and the budget admitted # none of them, which is a statement about the limit, not about the # bundle. raise ConsumeError( f"{matched} concept(s) answered this question and the {limit}-byte " f"budget admitted none of them; the cut strategy is wrong for this " "bundle at this limit -- refusing rather than emitting an empty " "payload that would read as 'nothing was found'", code="budget_admits_nothing", ) if spent > limit: # Structurally unreachable while the knapsack bucket arithmetic is # one-sided, and kept because SS 7.3 is a MUST about the emitted # payload rather than about the algorithm that produced it. raise ConsumeError(f"spent ({spent}) exceeds limit ({limit})", code="budget_exceeded") return { "contract": CONTRACT_REVISION, "bundle": { "bundle_id": root_bundle_id, "ref": bundle_ref(bundle_root, profile=profile), }, "budget": { "unit": BUDGET_UNIT, "instrument": BUDGET_INSTRUMENT, "limit": limit, "spent": spent, "known_positive": { "case": case, "expected": expected, "measured": measured, # The second, independent route (`wc -c` on the same file), so # `expected == measured` is not the only thing standing between # a broken instrument and a green gate. "raw_bytes": len(_KNOWN_POSITIVE_PATH.read_bytes()), "encoding_delta": KNOWN_POSITIVE_ENCODING_DELTA, }, # Present only when a reservation was made, because a cut whose # strategy changed without saying so is the silent cut SS 5.3 # forbids -- and absent otherwise, so the default payload keeps # every byte it had. **( {"reserved": {"concept_id": reserved[0], "bytes": reserved[1]}} if reserved is not None else {} ), }, "denominators": { "considered": len(concepts), "withheld": len(withheld), "delivered": len(delivered), }, "question": question, "excerpts": list(delivered), "withheld": [{"concept_id": concept_id, "rule": rule} for concept_id, rule in withheld], } def serialise(payload: Mapping[str, object]) -> str: """The payload as the bytes that are actually emitted. `ensure_ascii=False` is load-bearing rather than cosmetic: the default inflates this corpus by 7.1 % (1 950 745 -> 2 089 391 B), so a gate measuring one form and a knapsack weighing the other disagree by more than the headroom. LF only, one trailing newline. """ return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=False) + "\n" # --- The CLI ------------------------------------------------------------------ def parse_args(argv: list[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("bundle", type=Path, help="the OKF bundle directory to read") parser.add_argument("--question", required=True, help="the question to cut the bundle for") parser.add_argument( "--k", type=int, default=DEFAULT_K, help=f"cap on delivered excerpts (default {DEFAULT_K})" ) parser.add_argument( "--limit", type=int, default=DEFAULT_LIMIT, help=f"budget in {BUDGET_UNIT} (default {DEFAULT_LIMIT})", ) parser.add_argument( "--cost-vocabulary", action="store_true", help=( "let the declared cost/price/quantity vocabulary bridge a question " "and a document that share no word. OFF by default; a question " "naming no term in that vocabulary is unaffected either way" ), ) parser.add_argument( "--reserve-top-rank", action="store_true", help=( "give the highest-ranked candidate its bytes before the budget is " "packed, so a large top-ranked excerpt is not out-summed by small " "ones. OFF by default; a candidate that alone exceeds the budget is " "still refused" ), ) parser.add_argument( "--rarity-weight", action="store_true", help=( "weight each lexical hit by log(N/df) over the bundle's own " "concepts instead of counting it as one. OFF by default, and the " "default is a MEASUREMENT rather than a preference: measured on " "four corpora it moved one gold rank 9->8, left one at 35 and made " "one 96->103 worse. See docs/2026-09-08-sjeldenhetsvekt.md" ), ) parser.add_argument("--out", type=Path, default=None, help="write here instead of stdout") parser.add_argument( "--ref", default=None, help=( "assert the bundle's content identity. NOT an override: the identity " "is computed regardless and a mismatch refuses, because labelling a " "payload with an identity its bytes do not have is what SS 3.3 exists " "to prevent" ), ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: """Three exit codes, not two. **0** a payload was written, **1** the run happened and refused, **2** the run did not happen. Collapsing 2 into 1 would report an unread bundle as a failed cut -- two findings with different owners under one number. """ args = parse_args(argv) if not args.bundle.is_dir(): print(f"okf_consume: FAILED - {args.bundle} is not a directory", file=sys.stderr) return 2 try: payload = build_payload( args.bundle, question=args.question, k=args.k, limit=args.limit, cost_vocabulary=args.cost_vocabulary, reserve_top_rank=args.reserve_top_rank, rarity_weight=args.rarity_weight, ) except ConsumeError as error: print(f"okf_consume: FAILED - {error}", file=sys.stderr) return 1 except OSError as error: print(f"okf_consume: FAILED - the bundle could not be read: {error}", file=sys.stderr) return 2 except UnicodeDecodeError as error: print(f"okf_consume: FAILED - the bundle is not utf-8: {error}", file=sys.stderr) return 2 computed = payload["bundle"] assert isinstance(computed, dict) if args.ref is not None and args.ref != computed["ref"]: # Refuses BEFORE writing: a payload on disk under an asserted ref that # the bytes contradict is worse than no payload. print( f"okf_consume: FAILED - the bundle's identity is {computed['ref']}, " f"not the asserted {args.ref}; refusing to write a payload the " "caller would label with an identity its bytes do not have", file=sys.stderr, ) return 1 text = serialise(payload) if args.out is None: sys.stdout.write(text) else: args.out.write_text(text, encoding="utf-8", newline="\n") return 0 if __name__ == "__main__": raise SystemExit(main())