portfolio-optimiser/src/portfolio_optimiser/prepass.py

"""Consume an OKF consumption **pre-pass payload** — a cut of a knowledge base, declared.

The debate normally NAVIGATES its knowledge base (S2c): it is handed a pointer and four tools,
and it opens what it chooses. That works — measured 2026-09-06, a live model reached the price
form in three steps out of 630 concept documents — and it leaves an **undeclared cut**: nothing
in such a run says how many concepts were considered, how many withheld, or by which rule. The
OKF consumption contract SS 2.3 calls that a denominator failure dressed as an answer, and this
repository holds the same rule under its own names (``skipped_links``, ``unkeyed_verdicts``,
``cost_baseline_anchored`` all exist because a silent drop is a claim nobody can revise).

This module consumes the alternative: a payload produced ahead of the run by an external,
contract-conformant pre-pass, carrying the delivered excerpts AND the three denominators.

**po produces no payload and vendors no producer.** Contract SS 2.4 — "A conformant skill MAY be
handed a payload by any transport. The transport is not part of this contract." A subprocess
against the producer's checkout would bind this package to a path on one operator's machine and
die inside ``git archive HEAD``; a vendored copy would be the second copy of a ranking instrument
that ko-(p) forbids. So the payload is an INPUT FILE, in the ``--mandate`` / ``--ledger`` /
``--explore-config`` idiom, and what this module owns is the GATE in front of it.

**Two gates live inside the tool this seam withdraws**, and both are re-raised here on the
MOUNTED document rather than delegated to the producer's own rules: the verdict-layer refusal
(order 20260904T172353Z) and the SS 4.1a dimension scope. Delegating either would make one of
this repository's most recently gated invariants depend on a file an external caller supplied.

**Validation, NEVER repair** (``okf.write_concept_file``'s rule): a payload that does not hold is
refused by name and nothing is recounted, corrected or written.
"""

from __future__ import annotations

import hashlib
import json
import re
import unicodedata
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final

from pydantic import BaseModel, ConfigDict

from portfolio_optimiser import okf
from portfolio_optimiser.retrieval import PathSecurityError, safe_resolve

#: The payload revision this consumer understands, compared with ``==`` and never a prefix
#: (SS 8.2 exists so a reader can tell which revision it is holding). A future ``/2`` is refused by
#: name rather than consumed as if it were this one — the ``PORTFOLIO_OTEL`` rule: never a silent
#: fallback on an unknown declared value.
CONTRACT_REVISION = "okf-consumption/1"

#: The suffix a ``concept_id`` is missing. The producer's id is the bundle-relative path minus the
#: suffix (measured against real output, and against ``okf``'s own ``BundleFile.name``), so the
#: join back to a mounted document is this one concatenation.
CONCEPT_SUFFIX = ".md"


class PrepassRefused(ValueError):
    """A payload this run will not read from, named.

    A ``ValueError`` **subclass on purpose**: it must land on ``run.main``'s refusal tuple as
    ``run refused: ...`` and on the hosted flat's 400 arm, never on the crash channel — the
    ``okf.BundleIdMismatch`` / ``okf.CostBaselineDerivationError`` precedent. The caller supplied
    a file that does not hold; that is a refusal, not a failure of this program.
    """


# --- The payload (contract SS 8, plus po's declared superset) -------------------------------


class _Permissive(BaseModel):
    """SS 8: "Additional members are permitted and are not read by the checker."

    Measured against real producer output at revision ``54a0bc2``: excerpts carry ``rank`` and
    ``bundle_id_inherited`` beyond the members SS 8 fixes. A strict model would refuse conformant
    payloads, so every model here ignores what it does not name.
    """

    model_config = ConfigDict(extra="ignore")


class PrepassKnownPositive(_Permissive):
    """SS 7.4: the instrument's own proof that it can count."""

    case: str
    expected: int
    measured: int


class PrepassBudget(_Permissive):
    """SS 7: a limit, the unit it is counted in, the instrument that counted, and the spend."""

    unit: str
    instrument: str
    limit: int
    spent: int
    known_positive: PrepassKnownPositive


class PrepassBundle(_Permissive):
    """SS 3.3: the base's id and the **ref** — a fact about bytes, never a declared version."""

    bundle_id: str
    ref: str


class PrepassDenominators(_Permissive):
    """SS 5.1: the three counts, whose identity SS 5.2 requires to close."""

    considered: int
    withheld: int
    delivered: int


class PrepassSource(_Permissive):
    """SS 8's declared address for an excerpt: where the underlying source document lives.

    A fixed, singular member (never a growing family like ``source_*``) — named on purpose.
    """

    resource: str
    title: str | None = None


class PrepassExcerpt(_Permissive):
    """One delivered unit of bundle content.

    ``text``, ``text_sha256`` and ``concept_id`` are **po's declared superset of SS 8**, which
    names no content member at all — a payload can be fully conformant and carry nothing to read,
    and this seam's entire value is the content. Requiring them refuses that case by name instead
    of discovering it as an empty prompt.

    ``title``, ``req_number`` and ``sources`` are named fields because they are SS-8-declared,
    singular members — every payload has at most one of each. ``source_*`` locators
    (``source_element_id``, ``source_sha256``, and whatever a future producer adds) are NOT named
    here: measured (P3, order 20260908T141941Z), the producer's own message calls that a PREFIX
    RULE rather than a fixed allowlist — a real corpus already carries five such members where
    another carries two. Naming two of them would need a new field and a new render line for a
    third; ``extra="allow"`` plus :meth:`source_locators`' prefix scan needs neither. Every field
    here defaults to ``None`` (or, for ``sources``, absent) because most payloads in this
    repository predate P2's producer change and carry none of them — a required field would
    refuse every payload written before today.
    """

    model_config = ConfigDict(extra="allow")

    bundle_id: str
    concept_id: str
    sha256: str
    adjudication: str
    trust_tier: str
    text: str
    text_sha256: str
    title: str | None = None
    req_number: str | None = None
    sources: tuple[PrepassSource, ...] | None = None

    def source_locators(self) -> tuple[tuple[str, str], ...]:
        """Every ``source_*`` scalar the producer attached, sorted by key.

        Reads ``model_extra`` rather than a named field for each one — the prefix-rule reason
        this class's docstring gives. ``sources`` itself is excluded: it is a named field (a list,
        never a scalar) and does not land in ``model_extra`` in the first place, but the guard
        is explicit rather than relying on that.
        """
        extra = self.model_extra or {}
        return tuple(
            sorted(
                (key, str(value))
                for key, value in extra.items()
                if key.startswith("source_") and key != "sources"
            )
        )


class PrepassWithheld(_Permissive):
    """SS 5.3: a concept that was considered and not delivered, naming the rule that dropped it."""

    concept_id: str
    rule: str


class PrepassPayload(_Permissive):
    """One contract-SS-8 payload.

    ``question`` is REQUIRED although SS 8's example does not fix it: a cut computed for a
    different question, accepted in silence, would leave this run's artefacts unable to say which
    question produced the denominators they publish — the same undeclared claim the seam exists
    to remove, one level up.
    """

    contract: str
    bundle: PrepassBundle
    budget: PrepassBudget
    denominators: PrepassDenominators
    question: str
    excerpts: tuple[PrepassExcerpt, ...]
    withheld: tuple[PrepassWithheld, ...]


def load_prepass_payload(path: str) -> PrepassPayload:
    """Read one payload from disk, fail-fast.

    ``mandate.load_mandate``'s idiom exactly: ``FileNotFoundError`` for a path that is not there,
    ``ValueError`` (from ``json``) for bytes that are not JSON, ``ValidationError`` for a shape the
    model refuses. None of them is swallowed — a caller who asked for a declared cut and silently
    got ``None`` would have been answered by a downgraded order.
    """
    return PrepassPayload.model_validate(json.loads(Path(path).read_text(encoding="utf-8")))


# --- The shape gate ------------------------------------------------------------------------


def check_payload_shape(payload: PrepassPayload) -> None:
    """Refuse a payload whose own numbers do not hold. Reads no disk.

    Six refusals. Four are live against today's producer; two are DEFENSIVE and unwitnessed —
    ``spent > limit`` and a known-positive mismatch are both refused by the producer BEFORE it
    emits (``okf_consume.py`` raises ``budget_exceeded`` and ``instrument_unvalidated``), so no
    payload carrying them exists to test against. They stand for the reason ``budget_stop``'s
    portfolio arm stands: the contract states them as MUSTs about the emitted payload, and a
    consumer that trusts a producer to have checked has not checked.
    """
    if payload.contract != CONTRACT_REVISION:
        raise PrepassRefused(
            f"the payload declares contract {payload.contract!r}, and this consumer reads "
            f"{CONTRACT_REVISION!r}; refusing rather than reading an unknown revision as if it "
            "were this one"
        )
    counts = payload.denominators
    observed = counts.withheld + counts.delivered
    if counts.considered != observed:
        # The two numbers as FIELDS, never one sentence (BudgetExceeded's ko-(y) rule): "it does
        # not close" is not actionable, "9 declared against 5 observed" is.
        raise PrepassRefused(
            f"the denominators do not close: considered {counts.considered} declared, "
            f"{observed} observed as withheld {counts.withheld} + delivered {counts.delivered}; "
            "a count that does not close is not a denominator (SS 5.2)"
        )
    if len(payload.excerpts) != counts.delivered:
        raise PrepassRefused(
            f"the payload carries {len(payload.excerpts)} excerpts but declares "
            f"delivered {counts.delivered} (SS 8.1)"
        )
    if len(payload.withheld) != counts.withheld:
        # SS 8.1's SECOND half, and the one this whole seam is about: the withheld list IS the
        # declaration of what was not delivered.
        raise PrepassRefused(
            f"the payload carries {len(payload.withheld)} withheld entries but declares "
            f"withheld {counts.withheld} (SS 8.1)"
        )
    if payload.budget.spent > payload.budget.limit:
        raise PrepassRefused(
            f"the payload spent {payload.budget.spent} against a limit of {payload.budget.limit} "
            f"{payload.budget.unit}; exceeding the gate means the cut strategy is wrong for this "
            "bundle, which is a finding requiring a decision (SS 7.3)"
        )
    known = payload.budget.known_positive
    if known.expected != known.measured:
        raise PrepassRefused(
            f"the budget instrument did not reproduce its known positive {known.case!r}: "
            f"expected {known.expected}, measured {known.measured}; an instrument that has not "
            "reproduced a known figure has not been shown to count (SS 7.4)"
        )


# --- Binding the payload to the mounted base ------------------------------------------------


def concept_text(path: Path) -> str:
    """The concept body as the producer delivers it, re-derived from the mounted file.

    **Transcribed from the producer, and MEASURED — never guessed.** The rule is: take the text
    after the frontmatter block, NFC-normalise, and strip trailing whitespace per line (a
    spreadsheet render is padded to hundreds of trailing spaces per line, so unstripped, most of a
    budget goes on padding). The frontmatter split uses ``splitlines()``; measured 2026-09-07, a
    naive ``split("\\n")`` disagrees with the producer, and the disagreement is invisible until a
    real payload is checked against it — which is what
    ``test_concept_text_reproduces_the_producers_derivation`` does.
    """
    text = path.read_text(encoding="utf-8")
    lines = text.splitlines()
    body = text
    if lines and lines[0].strip() == "---":
        for offset, line in enumerate(lines[1:], start=2):
            if line.strip() == "---":
                body = "\n".join(lines[offset:])
                break
    return "\n".join(line.rstrip() for line in unicodedata.normalize("NFC", body).split("\n"))


def verify_against_bundle(
    payload: PrepassPayload,
    *,
    bundle_dir: str,
    resolved_id: okf.ResolvedBundleId,
    dimension: str | None = None,
) -> None:
    """Refuse a payload that is not this base's cut. Run AFTER :func:`check_payload_shape`.

    Five checks, each naming what it refused.

    1. **Identity** is the base's DECLARED id, never its mount (S7a-3): a base delivered under a
       directory name of its own is opened rather than refused, so comparing against the mount
       would refuse exactly the corpora the slack exists for.
    2. **The join** goes through ``safe_resolve``, the one fail-closed in-/out-of-bundle test in
       this repository. A ``concept_id`` is externally supplied, so it is untrusted input by
       definition — and ``PathSecurityError`` is a ``RuntimeError``, which would leave the CLI as
       a traceback and the hosted flat as a 500, so it is re-raised here as the refusal it is.
    3. **The file digest** catches a stale payload: ``sha256`` is the digest of the WHOLE mounted
       concept file. ``bundle.ref`` is deliberately NOT recomputed — ``sha256-tree`` is the
       producer's algorithm, and re-deriving it here would be the vendored copy this module exists
       without. **Honesty limit, stated:** the delivered documents are verified; the whole tree is
       not.
    4. **The delivered text must be re-derivable from the mounted file.** ``sha256`` digests the
       file while ``text`` is a DERIVED member, so a payload can carry a correct digest beside
       arbitrary text — and ``text`` is what enters the debate's task message, the conversation's
       highest-trust slot. Requiring equality means a payload **cannot deliver bytes the base does
       not hold**, which reduces the exposure from "external content in the prompt" to "bundle
       content in the prompt" — what ``read_file`` already does today.
    5. **The two withdrawn gates**, re-raised on the mounted DOCUMENT: the verdict layer (whose
       prior judgements reach a hypothesis only through the gated ExpeL fold) and SS 4.1a's
       dimension scope. The producer applies its own verdict-layer exclusion, and that is exactly
       why this is not delegated to it. The ``dimension`` arm is DEFENSIVE from the CLI, which
       refuses the two flags together; a library caller can still reach it (``budget_stop``'s
       precedent for an arm that stands without a live caller).
    """
    if payload.bundle.bundle_id != resolved_id.id:
        raise PrepassRefused(
            f"the payload declares bundle_id {payload.bundle.bundle_id!r} and this knowledge base "
            f"declares {resolved_id.id!r} (mounted at {resolved_id.mount!r}); a cut of another "
            "corpus is not this run's cut"
        )
    for excerpt in payload.excerpts:
        try:
            resolved = Path(safe_resolve(bundle_dir, excerpt.concept_id + CONCEPT_SUFFIX))
        except PathSecurityError as error:
            raise PrepassRefused(
                f"the payload names concept {excerpt.concept_id!r}, which does not resolve inside "
                f"the knowledge base: {error}"
            ) from error
        if not resolved.is_file():
            raise PrepassRefused(
                f"the payload delivers concept {excerpt.concept_id!r}, which this knowledge base "
                "does not hold"
            )
        if hashlib.sha256(resolved.read_bytes()).hexdigest() != excerpt.sha256:
            raise PrepassRefused(
                f"the sha256 the payload declares for {excerpt.concept_id!r} is not the digest of "
                "the mounted document; the payload was built against different bytes"
            )
        derived = concept_text(resolved)
        if derived != excerpt.text:
            raise PrepassRefused(
                f"the text the payload delivers for {excerpt.concept_id!r} is not the text of the "
                "mounted document; a payload may only deliver what the knowledge base holds"
            )
        if hashlib.sha256(derived.encode("utf-8")).hexdigest() != excerpt.text_sha256:
            raise PrepassRefused(
                f"the text_sha256 the payload declares for {excerpt.concept_id!r} is not the "
                "digest of the text it delivers"
            )
        if okf.declares_verdict_type(resolved):
            raise PrepassRefused(
                f"the payload delivers {excerpt.concept_id!r}, which the knowledge base declares "
                "as a verdict; prior judgements reach a hypothesis only through the gated "
                "experience fold, never as read context"
            )
        if dimension is not None and not _in_dimension(resolved, dimension):
            raise PrepassRefused(
                f"the payload delivers {excerpt.concept_id!r}, which is outside the dimension "
                f"{dimension!r} this run is scoped to"
            )


def _in_dimension(path: Path, dimension: str) -> bool:
    """The SS 4.1a predicate, applied to a mounted document.

    ``okf.in_dimension`` takes a ``BundleFile``, so this reads the one frontmatter key it reads
    and defers to it — never a second copy of the rule (ko-(p)).
    """
    frontmatter = okf.parse_frontmatter(path)
    return okf.in_dimension(
        okf.BundleFile(
            name=path.name,
            type=frontmatter.get("type", ""),
            frontmatter=frontmatter,
            body="",
        ),
        dimension,
    )


# --- The one admission gate, shared by every door that consumes a payload --------------------


def admit_payload(
    payload: PrepassPayload,
    *,
    bundle_dir: str,
    resolved_id: okf.ResolvedBundleId,
    dimension: str | None = None,
) -> None:
    """Everything that must hold before a payload may shape a run. Raises, or returns nothing.

    ONE copy, because there are now TWO doors onto this file — the debate's ``--prepass-payload``
    (the cut REPLACES the pointer and the tools are withdrawn) and the exploration's
    ``--prepass-seed`` (the cut is the STARTING POINT and the tools stay). The two arms differ in
    what they do with an admitted payload and in nothing at all about what makes one admissible,
    and two copies of an admission rule is the ko-(p) drift that would let one door accept what
    the other refuses.

    The three steps, in this order and for this reason: the shape gate reads no disk and so is
    free, the bundle check is the expensive one, and the empty-delivery refusal comes last because
    a payload that does not hold has not earned an interpretation of its own emptiness.

    **``delivered == 0`` is refused on BOTH arms**, and on the seeding arm that is a decision
    rather than an inheritance. Measured, it is reachable only when every concept failed to match
    lexically (the producer REFUSES the other empty case, where concepts matched and the budget
    admitted none), so it is evidence of ABSENCE for this question at this ref. On the seeding arm
    a caller might argue the tools are still there and the run could proceed — but it would then
    proceed as a PLAIN exploration while the operator had asked for a seeded one, which is the
    silently downgraded order ``load_mandate`` fail-fasts against.
    """
    check_payload_shape(payload)
    verify_against_bundle(
        payload, bundle_dir=bundle_dir, resolved_id=resolved_id, dimension=dimension
    )
    if not payload.excerpts:
        # Saying it beats two silent alternatives: an empty prompt, or falling through to
        # ``run_project``'s citation guard, whose message names ``docs_dir`` — ``None`` on this
        # path. SS 7.3's own posture: the skill stops and says so.
        raise PrepassRefused(
            f"the pre-pass delivered 0 of {payload.denominators.considered} concepts for the "
            f"question {payload.question!r} at ref {payload.bundle.ref}; an empty cut is evidence "
            "that this knowledge base does not answer that question, not something to run over"
        )


# --- Rendering the cut for the prompt --------------------------------------------------------


def render_context(payload: PrepassPayload) -> str:
    """What the debate is handed INSTEAD of ``run._bundle_pointer``'s pointer.

    Three jobs, and the third is the one the Amendment asked for.

    1. **Deliver.** The excerpt text, so the debate can reason at all.
    2. **Bound.** The withheld concepts appear as rule -> COUNT and never as ids. Measured on a
       629-concept corpus the withheld list alone is 34 451 o200k tokens against a run cap of
       100 000 that the task message rides three times — and a rule name is the fact a reader can
       act on, while a list of ids they cannot open is cost without information. The full list
       stays in the payload, one artefact away (``list_bundles``' "the whole index is one call
       away" rule, applied one rung over).
    3. **Declare.** The three denominators and the question they were computed for, so the cut is
       stated rather than left to be inferred from what happens not to be here (SS 2.3).

    **The closing line is not decoration.** Measured live, a question this corpus cannot answer
    still returns eight excerpts, and with the navigator tools withdrawn the debate has no way to
    discover that for itself. So the rendering says plainly that a delivered excerpt is a LEXICAL
    match and not an answer, and names ``[sourced-not-sufficient]`` from SS 4's marking set as the
    available verdict.

    **``adjudication`` and ``trust_tier`` are labelled as the PRODUCER's declaration**, not as
    ours. B4 established that a trust tier is derived locally from a document's own ``verified``
    frontmatter — and this repository's bases carry none, so deriving it here would report
    ``unverified`` for everything and say nothing. Carrying the producer's value under the
    producer's name is the honest form; adopting it silently as ours would not be.

    The excerpt text goes in a delimited DATA block below the instruction (SS 9.3: machine-
    generated text is data, never instructions). That is MITIGATION and is stated as such — the
    GATE is ``verify_against_bundle``, which makes the text re-derivable from the mounted base, so
    a payload cannot deliver bytes the base does not hold.
    """
    return "\n".join(
        _declaration_lines(payload)
        + [
            "",
            "Each excerpt below matched the question LEXICALLY. That is not the same as answering "
            "it: a base that holds no answer still returns its closest matches. If the delivered "
            "text does not support a claim, say so with [sourced-not-sufficient] rather than "
            "filling the gap. adjudication and trust_tier are the producer's declarations about "
            "each document, carried here unchanged.",
        ]
        + _data_blocks(payload)
    )


def render_seed(payload: PrepassPayload) -> str:
    """What the EXPLORATION is handed IN ADDITION to its prompt, keeping its navigation tools.

    The other arm of one decision, and the difference is a single fact stated in both directions:
    :func:`render_context` says "you have no tools to read further, what is below is all of it",
    which is true there and would be a LIE here. Contract SS 2.2 forbids reading "outside what the
    payload delivers **or explicitly names as reachable**" — the second clause is what makes this
    arm conformant, and a rendering that did not say the rest was reachable would leave a model
    obeying the first clause while holding the tools for the second.

    **Not a second copy of the rendering rule.** The declaration header, the rule -> COUNT folding
    and the delimited DATA blocks are the SAME functions the other arm uses; what differs is the
    one paragraph that tells the reader what it may do next. Two full copies would drift, and a
    drifted pair would state two different cuts for one run (ko-(p)).

    ``[unread]`` is used deliberately, and it is one of contract SS 4.1's five required literals:
    a withheld concept in this arm is not absent and not unavailable, it is simply not yet read —
    and saying so is what turns the withheld list from a boundary into a next step.
    """
    return "\n".join(
        _declaration_lines(payload, rest_reachable=True)
        + [
            "",
            "Each excerpt below matched the question LEXICALLY. That is not the same as answering "
            "it: a base that holds no answer still returns its closest matches. The withheld "
            "concepts are [unread], not absent — if the delivered text does not support a claim, "
            "OPEN THE BASE with your navigation tools rather than filling the gap, and reserve "
            "[sourced-not-sufficient] for a claim the base itself could not support. Report what "
            "you actually read. adjudication and trust_tier are the producer's declarations about "
            "each document, carried here unchanged.",
        ]
        + _data_blocks(payload)
    )


def _declaration_lines(payload: PrepassPayload, *, rest_reachable: bool = False) -> list[str]:
    """The header both renderings open with: the base, what the cut may be used for, the counts.

    ONE copy, because these lines ARE the declaration (SS 2.3) and two of them would be two
    answers to "what was this run's cut". Only the second line differs between the arms, and it
    differs on exactly the fact ``PrepassDeclaration.rest_reachable`` carries.
    """
    counts = payload.denominators
    rules = ", ".join(f"{rule} ({count})" for rule, count in withheld_rule_counts(payload))
    stance = (
        "You are reading a DECLARED CUT of that base as your STARTING POINT, not as a replacement "
        "for it. The rest of the base stays reachable with your navigation tools, and you are "
        "expected to use them when the cut does not carry what you need."
        if rest_reachable
        else "You are reading a DECLARED CUT of that base, not the base itself, and you have no "
        "tools to read further. What is below is all of it."
    )
    return [
        f"Knowledge base: {payload.bundle.bundle_id} (ref {payload.bundle.ref}).",
        "",
        stance,
        f"The cut was computed for this question: {payload.question}",
        f"Concepts considered: {counts.considered}. Withheld: {counts.withheld}. "
        f"Delivered below: {counts.delivered}.",
        f"Withheld by rule: {rules}." if rules else "Withheld by rule: none.",
    ]


def _excerpt_header(excerpt: PrepassExcerpt) -> str:
    """The BEGIN line for one excerpt: ``concept_id`` plus whatever the producer named it with.

    **Known-negative, load-bearing:** a P1-form excerpt (none of the five new fields) renders
    BYTE-IDENTICAL to before P3 — the loop below appends nothing, and the line is exactly the
    old ``(adjudication: ..., trust_tier: ...)`` form
    (``test_a_p1_form_payload_renders_the_header_exactly_as_before``). Carrying a field through
    must never change what an older payload renders as.

    ``req_number`` and ``title`` are what a person would cite; the address (``sources[0].resource``)
    and every ``source_*`` locator are what lets a claim be traced back to the document that
    produced it — measured (P2 SS 4) to be the exact thing modelled proposals cited a UUID instead
    of, because the UUID was the only identifier that reached the prompt.
    """
    fields = [f"adjudication: {excerpt.adjudication}", f"trust_tier: {excerpt.trust_tier}"]
    if excerpt.req_number is not None:
        fields.append(f"req_number: {excerpt.req_number}")
    if excerpt.title is not None:
        fields.append(f"title: {excerpt.title}")
    if excerpt.sources:
        fields.append(f"source: {excerpt.sources[0].resource}")
    for key, value in excerpt.source_locators():
        fields.append(f"{key}: {value}")
    return f"--- BEGIN DATA {excerpt.concept_id} ({', '.join(fields)}) ---"


#: Two or more. There is deliberately no threshold to tune: "a run of horizontal whitespace inside
#: a line is a column gap" is the whole rule, so no K can drift and no K has to be defended.
_PADDING_RUN_RE: Final = re.compile(r"(?<=\S)[ \t]{2,}")


def collapse_padding(text: str) -> str:
    r"""Drop a spreadsheet render's column PADDING on the way into the prompt, and nothing else.

    **Measured (finding 5, order 20260908T195801Z)** on K2's price schedule as
    ``concept_text`` delivers it -- 104 lines / 67 245 characters: 208 interior whitespace runs of
    two or more, **117 of them >= 100 characters and the longest 887**, together **56 806 of
    67 245 characters = 84.5 %** of the excerpt, over 72 of 104 lines. So a label and its amount
    reached the model hundreds of characters apart (``82`` / 593 spaces / a description / 250
    spaces / ``5647500.0``), and most of what the run paid for was a column width pandoc chose.

    **No judgement anywhere**, which is four separate promises and each is gated:

    * interior runs of spaces and tabs collapse to ONE space;
    * **LEADING whitespace is untouched** -- indentation is markdown structure (nested lists,
      indented code), not padding, and flattening it would rewrite documents rather than unpad
      them. That is what the ``(?<=\S)`` lookbehind buys;
    * **no newline is touched**, so the line count is invariant and no row is merged with its
      neighbour -- a naive ``re.sub(r"\s+", " ", text)`` turns a table into a paragraph;
    * **no non-whitespace character is touched**, so no number can change. The gate asserts the
      per-line sequence of non-whitespace characters is identical, not merely that the digits are
      still somewhere.

    **It runs in the RENDERING, never in :func:`concept_text`.** That function is the local
    re-derivation ``verify_against_bundle`` binds a payload's bytes to; collapsing there would make
    every payload ever written fail its own digest and would defeat the gate that stops a payload
    delivering bytes the base does not hold. This is a display transform applied AFTER that gate.

    **Honesty limit, stated.** Collapsing to a single space loses the CELL BOUNDARY: ``Post SUM``
    becomes indistinguishable from prose containing those two words. Inventing a delimiter to carry
    the boundary would be exactly the judgement this rule avoids, and no measurement says which
    delimiter a model reads better -- so the loss is recorded rather than papered over. That a model
    then USES the numbers is likewise not shown: it would take a paid run, which is a separate
    order.
    """
    return "\n".join(_PADDING_RUN_RE.sub(" ", line) for line in text.split("\n"))


def _data_blocks(payload: PrepassPayload) -> list[str]:
    """The delimited DATA blocks, one per delivered excerpt (SS 9.3), shared by both arms.

    :func:`collapse_padding` runs HERE, in the one renderer both arms share (kø-(p)): a fix in
    ``render_context`` alone would leave the exploration reading padding, and two copies of a
    display rule drift into two answers about one excerpt.
    """
    lines: list[str] = []
    for excerpt in payload.excerpts:
        lines += [
            "",
            _excerpt_header(excerpt),
            collapse_padding(excerpt.text),
            f"--- END DATA {excerpt.concept_id} ---",
        ]
    return lines


# --- The declaration -------------------------------------------------------------------------


@dataclass(frozen=True)
class PrepassDeclaration:
    """What a run says about the cut it was given: a RUN-level fact, not a per-candidate one.

    It reports the payload's OWN denominators verbatim, never a recount off the navigated bundle:
    the pre-pass may legitimately have considered a different set (it counts the verdict layer,
    ``Bundle.context_files`` does not), and two numbers for one fact is ko-(p).

    ``rest_reachable`` says which of the two arms consumed the cut, and it is REQUIRED WITHOUT A
    DEFAULT for ``ProvenanceStamp.cost_baseline_anchored``'s reason: **both defaults would lie.**
    ``False`` would let a run that kept its navigation tools publish a declaration claiming the cut
    was all it could read; ``True`` would let the arm that WITHDREW them claim the base stayed
    open. It is the contract's own distinction -- SS 2.2 forbids reading "outside what the payload
    delivers **or explicitly names as reachable**", so a conformant consumer may keep the base
    reachable, and the difference between the two readings is exactly what a reader of this
    declaration needs to know.
    """

    bundle_id: str
    ref: str
    question: str
    considered: int
    withheld: int
    delivered: int
    withheld_rules: tuple[tuple[str, int], ...]
    rest_reachable: bool
    unnamed_excerpts: tuple[str, ...] = ()


def withheld_rule_counts(payload: PrepassPayload) -> tuple[tuple[str, int], ...]:
    """rule -> COUNT, sorted. The ONE folding of the withheld list in this repository.

    The concept ids are deliberately NOT carried: measured on a 629-concept corpus the withheld
    list alone is 34 451 o200k tokens, and a rule name is the fact a reader can act on while a
    list of ids they cannot open is cost without information. The full list stays in the payload,
    one artefact away.

    It is a function of its own rather than a step inside ``declaration_of`` because THREE
    surfaces need it -- the declaration and both renderings -- and a second copy of "how the
    withheld list folds" would be free to disagree about the run it describes (ko-(p)).
    """
    counts: dict[str, int] = {}
    for entry in payload.withheld:
        counts[entry.rule] = counts.get(entry.rule, 0) + 1
    return tuple(sorted(counts.items()))


def unnamed_excerpts(payload: PrepassPayload) -> tuple[str, ...]:
    """Every delivered excerpt carrying no ``title``, BY CONCEPT ID, in payload order.

    okf's checker names this ``excerpt_unnamed`` (SS 8) and measured 20 of them across this
    repository's two K2 payloads. On po's side the same absence was SILENT: ``_excerpt_header``
    appends the ``title:`` field only when it is present and says nothing otherwise, so a payload
    whose producer named nothing rendered exactly like one that had nothing to add.

    Ids, never a count: "3 of 4 excerpts are unnamed" cannot be taken back to a producer, and
    "these three concepts are" can -- ``BudgetExceeded``'s ko-(y) rule one level down.

    **Absence only, mirroring okf's rule exactly.** An excerpt whose producer wrote ``title: ""``
    HAS been named, badly; reclassifying that would be repair, and the header already renders it
    as ``title: `` where an absent one renders nothing at all.
    """
    return tuple(excerpt.concept_id for excerpt in payload.excerpts if excerpt.title is None)


def declaration_of(payload: PrepassPayload, *, rest_reachable: bool) -> PrepassDeclaration:
    """The declaration a verified payload supports, for the arm that consumed it.

    ``rest_reachable`` is a REQUIRED keyword: see :class:`PrepassDeclaration`. The payload cannot
    supply it -- a cut does not know what its consumer did with the navigation tools -- so it is
    the caller's to state, and every caller states it.
    """
    return PrepassDeclaration(
        bundle_id=payload.bundle.bundle_id,
        ref=payload.bundle.ref,
        question=payload.question,
        considered=payload.denominators.considered,
        withheld=payload.denominators.withheld,
        delivered=payload.denominators.delivered,
        withheld_rules=withheld_rule_counts(payload),
        rest_reachable=rest_reachable,
        unnamed_excerpts=unnamed_excerpts(payload),
    )


def declaration_payload(declaration: PrepassDeclaration) -> Mapping[str, object]:
    """The declaration as a plain mapping, so the outbox stays framework-free."""
    return {
        "bundle_id": declaration.bundle_id,
        "ref": declaration.ref,
        "question": declaration.question,
        "considered": declaration.considered,
        "withheld": declaration.withheld,
        "delivered": declaration.delivered,
        "withheld_rules": [
            {"rule": rule, "count": count} for rule, count in declaration.withheld_rules
        ],
        # Which arm read the cut. An artefact that reported the denominators without saying
        # whether the consumer could still open the base would leave a reader unable to tell a
        # bounded run from a seeded one -- the same undeclared claim, one level up.
        "rest_reachable": declaration.rest_reachable,
        # Which delivered excerpts nobody named. It DEFAULTS to empty (the ``skipped_links``
        # half, never ``cost_baseline_anchored``'s): an empty list here is an honest POSITIVE
        # statement -- every delivered excerpt was named -- so omission cannot be mistaken for
        # a fact we failed to record.
        "unnamed_excerpts": list(declaration.unnamed_excerpts),
    }