llm-ingestion-okf/tools/okf_consume.py

"""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 (  # 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
            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:<hex>`.

    The digest is taken over the LF-joined, byte-sorted lines
    `<posix-relative-path>\t<sha256 of the file bytes>` 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


# --- 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],
) -> list[tuple[Concept, float]]:
    """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.
    """
    question_tokens = normalise(question)
    signals: list[dict[str, float]] = [
        {
            concept.concept_id: float(
                _overlap(question_tokens, f"{concept.title} {concept.concept_id.replace('/', ' ')}")
            )
            for concept in concepts
        },
        {
            concept.concept_id: float(_overlap(question_tokens, concept.body))
            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)
    by_id = {concept.concept_id: concept for concept in concepts}
    ranked_ids = sorted(fused, key=lambda key: (-fused[key], key))
    return [(by_id[concept_id], fused[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",
    "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.

    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")
    )
    return {
        "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,
        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
        "text": text,
    }


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]], *, k: int, limit: int
) -> tuple[tuple[dict[str, object], ...], tuple[tuple[str, str], ...]]:
    """The ranked concepts split into delivered excerpts and named drops.

    **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.
    """
    withheld: list[tuple[str, str]] = []
    candidates: list[tuple[Concept, float, dict[str, object], int]] = []
    for concept, score 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
        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)
    capacity = limit // WEIGHT_BUCKET
    packed = {
        id(pool[index][0])
        for index in knapsack(
            tuple((score, -(-weight // WEIGHT_BUCKET)) for _, score, _, weight in pool),
            capacity=capacity,
        )
    }
    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)