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 unicodedata
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path

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 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.
    """

    bundle_id: str
    concept_id: str
    sha256: str
    adjudication: str
    trust_tier: str
    text: str
    text_sha256: str


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,
    )


# --- 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.
    """
    counts = payload.denominators
    rules = ", ".join(f"{rule} ({count})" for rule, count in declaration_of(payload).withheld_rules)
    lines = [
        f"Knowledge base: {payload.bundle.bundle_id} (ref {payload.bundle.ref}).",
        "",
        "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.",
        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.",
        "",
        "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.",
    ]
    for excerpt in payload.excerpts:
        lines += [
            "",
            f"--- BEGIN DATA {excerpt.concept_id} "
            f"(adjudication: {excerpt.adjudication}, trust_tier: {excerpt.trust_tier}) ---",
            excerpt.text,
            f"--- END DATA {excerpt.concept_id} ---",
        ]
    return "\n".join(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).
    """

    bundle_id: str
    ref: str
    question: str
    considered: int
    withheld: int
    delivered: int
    withheld_rules: tuple[tuple[str, int], ...]


def declaration_of(payload: PrepassPayload) -> PrepassDeclaration:
    """The declaration a verified payload supports.

    ``withheld_rules`` is rule -> COUNT, sorted. 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.
    """
    counts: dict[str, int] = {}
    for entry in payload.withheld:
        counts[entry.rule] = counts.get(entry.rule, 0) + 1
    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=tuple(sorted(counts.items())),
    )


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
        ],
    }