portfolio-optimiser/tests/test_prepass_render_loadbearing.py

"""Load-bearing gate for what a declared cut LOOKS LIKE in the prompt (order 20260907T080223Z).

Two things are gated here, and they pull in opposite directions:

- The rendering must carry enough that the debate can reason and DECLARE — the delivered text,
  the three denominators, and the rules that dropped everything else.
- It must stay bounded. Measured on a 629-concept corpus, the whole payload is 41 097 o200k
  tokens and its withheld list alone is 34 451; the debate's task message rides in three calls,
  so a rendering that carried the payload verbatim would cost ~123 000 against a run cap of
  100 000 that has already fired live.

The ceiling therefore lives HERE and not in `prepass.py` — `_CATALOGUE_EXCERPT_CHARS`' own rule:
a bound imported from the implementation moves with it, and widening it is exactly the regression
this file exists to catch.
"""

from __future__ import annotations

import hashlib
import json
import shutil
from pathlib import Path
from typing import Any

from portfolio_optimiser import prepass

FIXTURE = Path(__file__).parent / "fixtures" / "prepass" / "bygg-energi-mikro-fixture.payload.json"
SHIPPED_BASE = Path(__file__).parent.parent / "shared" / "examples" / "bygg-energi-mikro"

#: What the rendering may add ON TOP of the delivered text, in CHARACTERS.
#:
#: **The OVERHEAD is bounded, not the total, and that is the whole point.** The delivered text is
#: already bounded by the producer's own budget gate (SS 7.3, re-checked by `check_payload_shape`),
#: so an absolute ceiling would either refuse a legitimately large cut or be so loose it caught
#: nothing. What po must guarantee is the shape of the cost: O(delivered text) + O(1), never
#: O(corpus) — which is exactly what regresses if the 621-entry withheld list creeps back in.
#:
#: Characters, not tokens, for MAJOR-3's reason: `tiktoken` is not a dependency of this package,
#: and a gate that skips when an optional package is missing is a gate that can be silently
#: absent. Measured 2026-09-07 with a validated instrument, the byte/token ratio is not one
#: number — 2,50 for a withheld list, 2,90 for ordinary Norwegian concept prose, 17,90 for a
#: padded spreadsheet render — so converting a token budget into this bound would import an
#: arithmetic that does not hold across the corpus.
#:
#: The bound lives HERE and never in `prepass.py`: `_CATALOGUE_EXCERPT_CHARS`' rule, because a
#: bound imported from the implementation moves with it, and widening it is the regression.
_CEILING_OVERHEAD_CHARS = 2_000

_DELIVERED_SENTINEL = "SENTINEL-LEVERT-UTDRAG"
_WITHHELD_SENTINEL = "SENTINEL-TILBAKEHOLDT-KONSEPT"


def _raw() -> dict[str, Any]:
    return json.loads(FIXTURE.read_text(encoding="utf-8"))


def _marked_base(tmp_path: Path) -> tuple[str, prepass.PrepassPayload]:
    """A base where a DELIVERED concept and a WITHHELD one each carry their own sentinel.

    Two sentinels, not one: with a single marker the "the withheld text never reaches the prompt"
    assertion could be satisfied by a rendering that carries nothing at all, and the "the
    delivered text does reach it" assertion by one that carries everything.
    """
    root = tmp_path / "marked"
    shutil.copytree(SHIPPED_BASE, root)
    index = root / "index.md"
    lines = index.read_text(encoding="utf-8").split("\n")
    lines.insert(1, "bundle_id: bygg-energi-mikro-fixture")
    index.write_text("\n".join(lines), encoding="utf-8")

    raw = _raw()
    delivered = raw["excerpts"][0]
    withheld_id = raw["withheld"][0]["concept_id"]

    for concept_id, sentinel in (
        (delivered["concept_id"], _DELIVERED_SENTINEL),
        (withheld_id, _WITHHELD_SENTINEL),
    ):
        path = root / (concept_id + ".md")
        path.write_text(path.read_text(encoding="utf-8") + f"\n\n{sentinel}\n", encoding="utf-8")

    delivered["sha256"] = hashlib.sha256(
        (root / (delivered["concept_id"] + ".md")).read_bytes()
    ).hexdigest()
    delivered["text"] = prepass.concept_text(root / (delivered["concept_id"] + ".md"))
    delivered["text_sha256"] = hashlib.sha256(delivered["text"].encode("utf-8")).hexdigest()
    return str(root), prepass.PrepassPayload.model_validate(raw)


# --- what the rendering carries -------------------------------------------------------------


def test_the_rendering_carries_the_delivered_text(tmp_path: Path) -> None:
    _, payload = _marked_base(tmp_path)
    assert _DELIVERED_SENTINEL in prepass.render_context(payload)


def test_the_rendering_carries_nothing_from_a_withheld_concept(tmp_path: Path) -> None:
    """The half that matters. A withheld concept was withheld deliberately (SS 2.2)."""
    _, payload = _marked_base(tmp_path)
    assert _WITHHELD_SENTINEL not in prepass.render_context(payload)


def test_the_withheld_sentinel_really_is_in_the_base(tmp_path: Path) -> None:
    """The causality control for the arm above: without it, that assertion is satisfied by a
    sentinel that was never written anywhere."""
    bundle_dir, payload = _marked_base(tmp_path)
    withheld_id = payload.withheld[0].concept_id
    body = (Path(bundle_dir) / (withheld_id + ".md")).read_text(encoding="utf-8")
    assert _WITHHELD_SENTINEL in body


def test_the_rendering_names_the_withheld_rules_and_never_the_withheld_ids() -> None:
    """Rule -> COUNT. A rule name is a fact a reader can act on; a list of ids they cannot open
    is cost without information, and on a real corpus it is 34 451 tokens of it."""
    payload = prepass.load_prepass_payload(str(FIXTURE))
    rendering = prepass.render_context(payload)
    # The HEADER — everything po itself authors, before the first delivered document. A whole-
    # rendering substring assert would be red for the wrong reason the moment a delivered concept
    # happens to CITE a withheld one, which in a cross-linked base is ordinary (measured: it does).
    header = rendering.split("--- BEGIN DATA")[0]
    assert "verdict_layer_excluded" in header
    for entry in payload.withheld:
        assert entry.concept_id not in header


def test_the_rendering_carries_the_three_denominators_and_the_question() -> None:
    payload = prepass.load_prepass_payload(str(FIXTURE))
    rendering = prepass.render_context(payload)
    for number in (
        payload.denominators.considered,
        payload.denominators.withheld,
        payload.denominators.delivered,
    ):
        assert str(number) in rendering
    assert payload.question in rendering
    assert payload.bundle.bundle_id in rendering


def test_the_rendering_says_a_delivered_excerpt_is_not_an_answer() -> None:
    """Measured live on a 629-concept corpus: a question the base cannot answer still returns
    eight excerpts. A rendering that presented them as "what the base says about this" would be
    handing the debate a confident guess wearing a denominator."""
    rendering = prepass.render_context(prepass.load_prepass_payload(str(FIXTURE)))
    assert "[sourced-not-sufficient]" in rendering


# --- the bound ------------------------------------------------------------------------------


def _overhead(payload: prepass.PrepassPayload) -> int:
    """What the rendering costs beyond the delivered text itself."""
    return len(prepass.render_context(payload)) - sum(len(e.text) for e in payload.excerpts)


def test_the_rendering_adds_only_a_bounded_overhead_to_the_delivered_text() -> None:
    payload = prepass.load_prepass_payload(str(FIXTURE))
    assert _overhead(payload) <= _CEILING_OVERHEAD_CHARS, _overhead(payload)


def test_the_overhead_stays_bounded_when_the_corpus_is_large() -> None:
    """The arm that makes the bound mean something.

    The checked-in fixture withholds ONE concept, so a rendering that dumped every withheld id
    would pass the arm above on it. This one synthesises the corpus-scale case — 620 withheld
    concepts, K2's own order of magnitude — and requires the overhead to be UNCHANGED in order of
    magnitude, which only a rule -> count rendering can do.
    """
    raw = _raw()
    raw["withheld"] = [
        {
            "concept_id": f"del-ii/{n:03d}/et-ganske-langt-konseptnavn-som-i-et-ekte-korpus",
            "rule": "no_lexical_match",
        }
        for n in range(620)
    ] + raw["withheld"]
    raw["denominators"]["withheld"] = len(raw["withheld"])
    raw["denominators"]["considered"] = len(raw["withheld"]) + raw["denominators"]["delivered"]
    payload = prepass.PrepassPayload.model_validate(raw)
    prepass.check_payload_shape(payload)
    assert _overhead(payload) <= _CEILING_OVERHEAD_CHARS, _overhead(payload)


def test_the_withheld_list_alone_is_far_over_the_overhead_ceiling() -> None:
    """The flat control: without it, the arm above could be green because the synthetic list is
    somehow cheap rather than because the rendering summarises it."""
    withheld = [
        {
            "concept_id": f"del-ii/{n:03d}/et-ganske-langt-konseptnavn-som-i-et-ekte-korpus",
            "rule": "no_lexical_match",
        }
        for n in range(620)
    ]
    assert len(json.dumps(withheld, ensure_ascii=False)) > 5 * _CEILING_OVERHEAD_CHARS


# --- the declaration -------------------------------------------------------------------------


def test_the_declaration_reports_the_payloads_own_denominators() -> None:
    """Never a recount off the navigated bundle: the pre-pass counts the verdict layer and
    `Bundle.context_files` does not, so two numbers for one fact would be kø-(p)."""
    payload = prepass.load_prepass_payload(str(FIXTURE))
    declaration = prepass.declaration_of(payload)
    assert declaration.considered == payload.denominators.considered == 5
    assert declaration.withheld == payload.denominators.withheld == 1
    assert declaration.delivered == payload.denominators.delivered == 4
    assert declaration.ref == payload.bundle.ref
    assert declaration.question == payload.question


def test_the_declaration_counts_the_withheld_rules_without_naming_the_concepts() -> None:
    payload = prepass.load_prepass_payload(str(FIXTURE))
    declaration = prepass.declaration_of(payload)
    assert declaration.withheld_rules == (("verdict_layer_excluded", 1),)
    assert sum(count for _, count in declaration.withheld_rules) == declaration.withheld


def test_the_declaration_payload_is_a_plain_mapping() -> None:
    """The outbox stays framework-free (`explore.trace_payload`'s rule)."""
    declaration = prepass.declaration_of(prepass.load_prepass_payload(str(FIXTURE)))
    body = prepass.declaration_payload(declaration)
    assert json.loads(json.dumps(body))["withheld_rules"] == [
        {"rule": "verdict_layer_excluded", "count": 1}
    ]