portfolio-optimiser/tests/test_prepass_run_seam_loadbearing.py

"""Load-bearing gate for the pre-pass seam inside ``run_project`` (order 20260907T080223Z).

Given a verified payload, the bundle arm hands the debate a DECLARED CUT instead of
``_bundle_pointer``'s pointer, and WITHDRAWS the four navigator tools. Without one, every byte of
today's behaviour stands — which is the control every arm here is paired against.

**The withdrawal is asserted on the tool list passed to ``fresh_workflow``, never on
``RunResult.debate_tool_calls``.** Measured: that trace is ALREADY empty with all four tools
attached, because a ``ScriptedChatClient`` returns text and never emits a function call. An arm
written on it cannot distinguish the two implementations at all.
"""

from __future__ import annotations

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

import pytest

import portfolio_optimiser.run as run_module
from portfolio_optimiser.budget import BudgetExceeded
from portfolio_optimiser import okf, prepass
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.mcp_tools import McpServerConfig
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.verdicts import VerdictStore, seed_store_from_bundle

FIXTURE = Path(__file__).parent / "fixtures" / "prepass" / "bygg-energi-mikro-fixture.payload.json"
SHIPPED_BASE = Path(__file__).parent.parent / "shared" / "examples" / "bygg-energi-mikro"
PROJECT_ID = "BYGG-KONTOR-NORD"
NAVIGATOR_TOOLS = {"list_bundles", "read_bundle", "read_dir", "read_file"}

_LADDER = "read it with your tools"
_EXCERPT_SENTINEL = "SENTINEL-I-ET-LEVERT-UTDRAG"

_PROPOSAL = json.dumps(
    {
        "project_id": PROJECT_ID,
        "measure": "energy_efficiency",
        "claimed_saving_nok": 30000,
        "affected_items": [
            {"code": "ENERGI-TOTAL-EL", "quantity": 120000.0, "unit_cost": 1.25},
        ],
        "assumptions": {},
    }
)


def _prompt_blob(messages: Any) -> str:
    """Text PLUS function calls and results (the corrected S7a-2 instrument). ``.text`` alone
    measures a context-bearing prompt at a few characters."""
    parts: list[str] = []
    for message in messages:
        for content in getattr(message, "contents", []) or []:
            for attribute in ("text", "arguments", "result"):
                value = getattr(content, attribute, None)
                if value is not None:
                    parts.append(str(value))
    return " ".join(parts)


def _recording_factory(sink: list[str]) -> Any:
    """A scripted client whose instance ``_inner_get_response`` is REBOUND to a recorder.

    Rebinding rather than subclassing is deliberate: ``tests/test_scripted_client_consolidation``
    keeps a registry of every site that DEFINES that method, and a new definition here would go
    red there. This is the shape the two existing prompt gates use.
    """

    def factory(role: str) -> Any:
        client = ScriptedChatClient(_PROPOSAL, role=role)
        original = client._inner_get_response

        async def recording(*args: Any, **kwargs: Any) -> Any:
            messages = kwargs.get("messages") or (args[0] if args else [])
            sink.append(_prompt_blob(messages))
            return await original(*args, **kwargs)

        client._inner_get_response = recording  # type: ignore[method-assign]
        return client

    return factory


def _base(tmp_path: Path, *, sentinel: bool = False) -> tuple[str, prepass.PrepassPayload]:
    """A copy of the shipped base declaring its own id, with a matching payload.

    The MOUNT differs from the DECLARATION (S7a-3's slack case), so the identity check cannot be
    satisfied by a mount comparison.
    """
    root = tmp_path / "mounted-under-another-name"
    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 = json.loads(FIXTURE.read_text(encoding="utf-8"))
    if sentinel:
        first = raw["excerpts"][0]
        path = root / (first["concept_id"] + ".md")
        path.write_text(
            path.read_text(encoding="utf-8") + f"\n\n{_EXCERPT_SENTINEL}\n", encoding="utf-8"
        )
        first["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
        first["text"] = prepass.concept_text(path)
        first["text_sha256"] = hashlib.sha256(first["text"].encode("utf-8")).hexdigest()
    return str(root), prepass.PrepassPayload.model_validate(raw)


def _docs(tmp_path: Path) -> str:
    """The road path's data source, unused on the bundle arm but required by the signature."""
    d = tmp_path / "docs"
    d.mkdir(exist_ok=True)
    (d / "cost.txt").write_text("Energitiltak i kontorbygg.", encoding="utf-8")
    return str(d)


async def _run(bundle_dir: str, **kwargs: Any) -> tuple[Any, list[str], list[list[Any]]]:
    """Drive the bundle arm offline, capturing both the prompts and the tool list."""
    sink: list[str] = []
    captured: list[list[Any]] = []
    original = run_module.fresh_workflow

    def spy(*args: Any, **spy_kwargs: Any) -> Any:
        captured.append(list(spy_kwargs.get("tools") or []))
        return original(*args, **spy_kwargs)

    run_module.fresh_workflow = spy  # type: ignore[assignment]
    try:
        result = await run_project(
            PROJECT_ID,
            bundle_dir=bundle_dir,
            docs_dir=_docs(Path(bundle_dir).parent),
            store=VerdictStore(verdicts=[]),
            client_factory=_recording_factory(sink),
            max_rounds=2,
            **kwargs,
        )
    finally:
        run_module.fresh_workflow = original  # type: ignore[assignment]
    return result, sink, captured


# --- the rendering replaces the pointer -----------------------------------------------------


async def test_a_payload_replaces_the_pointer_in_the_debate(tmp_path: Path) -> None:
    bundle_dir, payload = _base(tmp_path, sentinel=True)
    _, sink, _ = await _run(bundle_dir, prepass_payload=payload)
    joined = " ".join(sink)
    assert _EXCERPT_SENTINEL in joined
    assert _LADDER not in joined


async def test_without_a_payload_the_pointer_stands(tmp_path: Path) -> None:
    """The control. Without it the arm above is satisfied by a run that built no prompt at all."""
    bundle_dir, _ = _base(tmp_path, sentinel=True)
    _, sink, _ = await _run(bundle_dir)
    joined = " ".join(sink)
    assert _LADDER in joined
    assert _EXCERPT_SENTINEL not in joined


# --- the tools are withdrawn, and only they -------------------------------------------------


async def test_a_payload_withdraws_the_four_navigator_tools(tmp_path: Path) -> None:
    bundle_dir, payload = _base(tmp_path)
    _, _, captured = await _run(bundle_dir, prepass_payload=payload)
    assert captured, "the debate was never built"
    assert {getattr(t, "name", "") for t in captured[0]} & NAVIGATOR_TOOLS == set()


async def test_without_a_payload_the_debate_gets_exactly_those_four(tmp_path: Path) -> None:
    """The paired control, asserting the EXACT set: `not any(...)` alone is also what an empty
    list produces, and an empty list is what a run that never built produces."""
    bundle_dir, _ = _base(tmp_path)
    _, _, captured = await _run(bundle_dir)
    assert captured, "the debate was never built"
    assert {getattr(t, "name", "") for t in captured[0]} == NAVIGATOR_TOOLS


async def test_a_configured_mcp_tool_survives_the_withdrawal(tmp_path: Path) -> None:
    """Distinguishes "withdraw the navigator tools" from ``debate_tools = []``. The MCP append
    sits BELOW the fork on purpose."""

    class _FakeMcpTool:
        name = "an_external_tool"

        async def __aenter__(self) -> "_FakeMcpTool":
            return self

        async def __aexit__(self, *exc: Any) -> None:
            return None

    bundle_dir, payload = _base(tmp_path)
    original = run_module.build_mcp_tools
    run_module.build_mcp_tools = lambda servers: [_FakeMcpTool()]  # type: ignore[assignment]
    try:
        _, _, captured = await _run(
            bundle_dir,
            prepass_payload=payload,
            mcp_servers=(
                McpServerConfig(
                    name="prisregister",
                    transport="http",
                    url="https://intern.example/mcp",
                    allowed_tools=("an_external_tool",),
                    timeout_seconds=15,
                ),
            ),
        )
    finally:
        run_module.build_mcp_tools = original  # type: ignore[assignment]
    names = {getattr(t, "name", "") for t in captured[0]}
    assert "an_external_tool" in names
    assert names & NAVIGATOR_TOOLS == set()


# --- what else the fork must get right ------------------------------------------------------


async def test_the_citations_are_the_delivered_concepts(tmp_path: Path) -> None:
    """A stamp citing every navigable concept for a proposal that saw a subset of them re-creates
    the undeclared claim this seam removes. Each snippet stays exact by construction.

    **The payload must deliver a STRICT SUBSET, and that is measured rather than assumed.** The
    shipped base navigates to four non-verdict concepts and the checked-in payload delivers all
    four, so on the unmodified fixture "cite the delivered ones" and "cite everything navigable"
    produce the SAME set — and an arm built on it stays green against a run that kept
    ``bundle_citations``. One excerpt is therefore moved to ``withheld`` first.
    """
    bundle_dir, _ = _base(tmp_path)
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    dropped = raw["excerpts"].pop()
    raw["withheld"].append({"concept_id": dropped["concept_id"], "rule": "below_k"})
    raw["denominators"]["delivered"] -= 1
    raw["denominators"]["withheld"] += 1
    payload = prepass.PrepassPayload.model_validate(raw)

    navigable = {f.name for f in okf.navigate_bundle(bundle_dir).context_files}
    delivered = {e.concept_id + ".md" for e in payload.excerpts}
    assert delivered < navigable, "the fixture cannot distinguish the two implementations"

    result, _, _ = await _run(bundle_dir, prepass_payload=payload)
    assert isinstance(result, RunResult)
    assert {c.file for c in result.provenance.citations} == delivered

    bodies = {f.name: f.body for f in okf.navigate_bundle(bundle_dir).context_files}
    for citation in result.provenance.citations:
        body = bodies[citation.file]
        assert citation.snippet == body[citation.locator.start_index : citation.locator.end_index]


async def test_a_payload_run_reaches_an_outcome(tmp_path: Path) -> None:
    """Nothing else here drives the fork past the debate; without this arm the citation guard at
    ``run.py:1015``, the checker gate and the outbox path are all unexercised on the new path."""
    bundle_dir, payload = _base(tmp_path)
    result, _, _ = await _run(bundle_dir, prepass_payload=payload)
    assert isinstance(result, RunResult)
    assert result.provenance.validator_decision in {"validated", "rejected"}


async def test_the_gated_expel_fold_still_reaches_the_hypothesis(tmp_path: Path) -> None:
    """GATE, not wall. A prior judgement must still reach the hypothesis prompt through the
    ExpeL fold — otherwise an implementation that simply refuses everything verdict-shaped
    passes every other arm here while having removed the loop's whole learning path."""
    bundle_dir, payload = _base(tmp_path)
    store = seed_store_from_bundle(bundle_dir)
    assert store.verdicts, "the shipped base no longer seeds a verdict"

    sink: list[str] = []
    await run_project(
        PROJECT_ID,
        bundle_dir=bundle_dir,
        docs_dir=_docs(Path(bundle_dir).parent),
        store=store,
        client_factory=_recording_factory(sink),
        max_rounds=2,
        prepass_payload=payload,
    )
    assert any("0.82" in prompt for prompt in sink), "the ExpeL fold no longer reaches generation"


# --- refusals -------------------------------------------------------------------------------


async def test_a_payload_for_another_base_refuses_without_building_a_debate(
    tmp_path: Path,
) -> None:
    bundle_dir, _ = _base(tmp_path)
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    raw["bundle"]["bundle_id"] = "a-different-corpus"
    with pytest.raises(prepass.PrepassRefused):
        await _run(bundle_dir, prepass_payload=prepass.PrepassPayload.model_validate(raw))


async def test_a_refused_payload_never_falls_back_to_the_pointer(tmp_path: Path) -> None:
    """``load_mandate``'s rule: a caller who asked for a declared cut and got a navigating run
    instead was answered by a silently downgraded order."""
    bundle_dir, _ = _base(tmp_path)
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    raw["bundle"]["bundle_id"] = "a-different-corpus"
    sink: list[str] = []
    with pytest.raises(prepass.PrepassRefused):
        await run_project(
            PROJECT_ID,
            bundle_dir=bundle_dir,
            docs_dir=_docs(Path(bundle_dir).parent),
            store=VerdictStore(verdicts=[]),
            client_factory=_recording_factory(sink),
            max_rounds=2,
            prepass_payload=prepass.PrepassPayload.model_validate(raw),
        )
    assert sink == [], "the run made model calls despite refusing the payload"


async def test_a_payload_that_delivers_nothing_is_refused_by_name(tmp_path: Path) -> None:
    """Measured: ``delivered == 0`` is only reachable when every concept failed to match, which
    IS evidence of absence for this question at this ref. Saying so beats falling through to
    ``run.py:1015``'s ``no citable content in docs_dir`` — a surface that is ``None`` here."""
    bundle_dir, _ = _base(tmp_path)
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    raw["withheld"] = [
        {"concept_id": e["concept_id"], "rule": "no_lexical_match"} for e in raw["excerpts"]
    ] + raw["withheld"]
    raw["excerpts"] = []
    raw["denominators"]["withheld"] = len(raw["withheld"])
    raw["denominators"]["delivered"] = 0
    sink: list[str] = []
    with pytest.raises(prepass.PrepassRefused) as excinfo:
        await run_project(
            PROJECT_ID,
            bundle_dir=bundle_dir,
            docs_dir=_docs(Path(bundle_dir).parent),
            store=VerdictStore(verdicts=[]),
            client_factory=_recording_factory(sink),
            max_rounds=2,
            prepass_payload=prepass.PrepassPayload.model_validate(raw),
        )
    message = str(excinfo.value)
    assert "0" in message and raw["question"] in message
    assert sink == []


async def test_a_payload_without_a_bundle_dir_is_refused(tmp_path: Path) -> None:
    """The road path has no base for the payload to agree with."""
    _, payload = _base(tmp_path)
    with pytest.raises(prepass.PrepassRefused, match="bundle"):
        await run_project(
            PROJECT_ID,
            docs_dir=_docs(tmp_path),
            store=VerdictStore(verdicts=[]),
            client_factory=_recording_factory([]),
            max_rounds=2,
            prepass_payload=payload,
        )


# --- the declaration on the result ----------------------------------------------------------


async def test_the_run_carries_the_declaration(tmp_path: Path) -> None:
    bundle_dir, payload = _base(tmp_path)
    result, _, _ = await _run(bundle_dir, prepass_payload=payload)
    assert isinstance(result, RunResult)
    assert result.prepass is not None
    assert result.prepass.ref == payload.bundle.ref
    assert result.prepass.delivered == payload.denominators.delivered


async def test_a_run_without_a_payload_carries_none(tmp_path: Path) -> None:
    bundle_dir, _ = _base(tmp_path)
    result, _, _ = await _run(bundle_dir)
    assert isinstance(result, RunResult)
    assert result.prepass is None


async def test_the_dry_run_report_carries_the_declaration(tmp_path: Path) -> None:
    """The dry-run cut sits BELOW the fork, so a dry run can honestly report the cut it was
    given — unlike ``unkeyed_verdicts``, which is resolved above it and could only report zero."""
    bundle_dir, payload = _base(tmp_path)
    report = await run_project(
        PROJECT_ID,
        bundle_dir=bundle_dir,
        docs_dir=_docs(Path(bundle_dir).parent),
        store=VerdictStore(verdicts=[]),
        client_factory=_recording_factory([]),
        max_rounds=2,
        live_dry_run=True,
        prepass_payload=payload,
    )
    assert isinstance(report, run_module.DryRunReport)
    assert report.prepass is not None
    assert report.prepass.considered == payload.denominators.considered


# --- the declaration leaves the run ---------------------------------------------------------


def test_the_notice_is_omitted_without_a_payload() -> None:
    """Omission, never an empty row. The golden transcript is the independent, pre-existing
    witness: a renderer that always returned a line would print in the demo and go red there."""
    assert run_module.prepass_notice(None) is None


def test_the_notice_reports_the_cut_when_there_was_one() -> None:
    declaration = prepass.declaration_of(prepass.load_prepass_payload(str(FIXTURE)))
    line = run_module.prepass_notice(declaration)
    assert line is not None
    assert "4 of 5" in line
    assert "verdict_layer_excluded (1)" in line
    assert declaration.ref in line


def test_a_withheld_rule_is_counted_and_the_concept_is_never_named() -> None:
    """ "We held 620 back under this rule" is the fact; naming them is 34 451 tokens of cost."""
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    raw["withheld"] = [
        {"concept_id": f"hemmelig-konsept-{n}", "rule": "no_lexical_match"} for n in range(20)
    ] + raw["withheld"]
    raw["denominators"]["withheld"] = len(raw["withheld"])
    raw["denominators"]["considered"] = len(raw["withheld"]) + raw["denominators"]["delivered"]
    declaration = prepass.declaration_of(prepass.PrepassPayload.model_validate(raw))
    line = run_module.prepass_notice(declaration)
    assert line is not None and "no_lexical_match (20)" in line
    assert "hemmelig-konsept-0" not in line
    body = prepass.declaration_payload(declaration)
    assert "hemmelig-konsept-0" not in json.dumps(body)


async def test_a_payload_run_writes_the_artefact_beside_the_debate_trace(tmp_path: Path) -> None:
    """SC15's actual qualification, observed on ONE run: an empty ``tool_calls`` next to a
    prepass artefact is a withdrawal; an empty one alone is the S2c regression."""
    bundle_dir, payload = _base(tmp_path)
    outbox_dir = tmp_path / "outbox"
    result, _, _ = await _run(
        bundle_dir, prepass_payload=payload, outbox_dir=str(outbox_dir), run_id="r1"
    )
    debate = json.loads((outbox_dir / "r1-debate.json").read_text(encoding="utf-8"))
    cut = json.loads((outbox_dir / "r1-prepass.json").read_text(encoding="utf-8"))
    assert debate["tool_calls"] == []
    assert cut["prepass"]["ref"] == payload.bundle.ref
    assert cut["prepass"]["question"] == payload.question
    assert isinstance(result, RunResult) and result.prepass is not None
    assert cut["prepass"]["delivered"] == result.prepass.delivered


async def test_a_run_without_a_payload_writes_no_prepass_artefact(tmp_path: Path) -> None:
    """The write-iff-offered rule, with three pre-existing witnesses for the same shape."""
    bundle_dir, _ = _base(tmp_path)
    outbox_dir = tmp_path / "outbox"
    await _run(bundle_dir, outbox_dir=str(outbox_dir), run_id="r1")
    assert not (outbox_dir / "r1-prepass.json").exists()
    assert (outbox_dir / "r1-debate.json").exists()


async def test_a_budget_stop_inside_the_debate_still_leaves_the_declaration(
    tmp_path: Path,
) -> None:
    """The ``finally`` is the point: the run that most needs the evidence is the one a cap cut."""
    bundle_dir, payload = _base(tmp_path)
    outbox_dir = tmp_path / "outbox"
    with pytest.raises(BudgetExceeded):
        await _run(
            bundle_dir,
            prepass_payload=payload,
            outbox_dir=str(outbox_dir),
            run_id="r1",
            max_tokens=1,
        )
    cut = json.loads((outbox_dir / "r1-prepass.json").read_text(encoding="utf-8"))
    assert cut["prepass"]["delivered"] == payload.denominators.delivered