portfolio-optimiser/tests/test_prepass_seed_door_loadbearing.py

"""Load-bearing gate for ``--prepass-seed`` — the pre-pass as the exploration's STARTING POINT.

Q5 = B (operator decision 2026-09-07). Arm A stands: ``--prepass-payload`` hands the DEBATE a
declared cut, withdraws the four navigator tools, and is still refused together with ``--explore``
for exactly the reason it was — an exploration reads the whole base with the very tools the
payload withdraws, so the run as a whole would read far outside the cut it declares.

This is the OTHER arm, and it is a different flag rather than a loosening of that refusal: the
cut seeds the exploration's task message and the tools STAY. Contract § 2.2 is what makes it
conformant — a skill may not read "outside what the payload delivers **or explicitly names as
reachable**" — and ``PrepassDeclaration.rest_reachable`` is what makes the two readings tellable
apart afterwards.

Four properties, and each arm is paired with the control that stops it being satisfied by a run
that did nothing:

(a) the flag is accepted only together with ``--explore``, and refused BY NAME everywhere it
    would otherwise be silently dropped;
(b) the delivered text lands in the exploration's STARTING POINT, and the navigator tools are
    still there — asserted by making a scripted navigator actually call two of them;
(c) the declaration says the cut was a starting point rather than a replacement, on stdout AND in
    ``{run_id}-exploration.json``, with arm A's ``rest_reachable: false`` as the paired control;
(d) without the flag every byte of today's behaviour stands.
"""

from __future__ import annotations

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

import pytest

import portfolio_optimiser.simulation as simulation_module
import portfolio_optimiser.run as run_module
from portfolio_optimiser import prepass
from portfolio_optimiser.run import main

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"
DECLARED_ID = "bygg-energi-mikro-fixture"

#: Written INTO the mounted base and then into the payload's excerpt, so its only route to a
#: prompt is the seed itself: the scripted sink records ``message.text`` only, and a tool RESULT
#: is ``contents`` — so a navigator that opened the same document could not put it there.
_EXCERPT_SENTINEL = "SENTINEL-I-ET-LEVERT-UTDRAG"

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


def _ledger(*, satisfied: bool, speaker: str = "navigator") -> str:
    return json.dumps(
        {
            "is_request_satisfied": {"reason": "r", "answer": satisfied},
            "is_in_loop": {"reason": "r", "answer": False},
            "is_progress_being_made": {"reason": "r", "answer": True},
            "next_speaker": {"reason": "r", "answer": speaker},
            "instruction_or_question": {"reason": "r", "answer": "open a base"},
        }
    )


#: One entry per stage the orchestrator asks for. A single constant answers all five with the
#: first round's ledger and concludes after one round having given nobody a turn (MAJOR-1's own
#: measurement), which would make the tool arm below vacuous.
_MANAGER_STAGES = [
    "FACTS: the base is anchored.",
    "PLAN: - let the navigator open a base",
    _ledger(satisfied=False),
    _ledger(satisfied=True),
    "FINAL: the exploration is done.",
]
_HYPOTHESIS = "HYPOTHESIS: " + json.dumps({"label": "Night setback", "rationale": "y"})
_NAVIGATING_SCRIPT = [
    {"call": "list_bundles"},
    {"call": "read_bundle", "args": {"bundle_id": DECLARED_ID}},
    "NAVIGATOR: read the index.",
]


# --- fixtures ---------------------------------------------------------------------------------


def _base(tmp_path: Path, *, sentinel: bool = False) -> str:
    """A copy of the shipped base DECLARING its own id, mounted under a different name (S7a-3)."""
    root = tmp_path / "mounted-under-another-name"
    if not root.exists():
        shutil.copytree(SHIPPED_BASE, root)
        index = root / "index.md"
        lines = index.read_text(encoding="utf-8").split("\n")
        lines.insert(1, f"bundle_id: {DECLARED_ID}")
        index.write_text("\n".join(lines), encoding="utf-8")
    if sentinel:
        raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
        path = root / (raw["excerpts"][0]["concept_id"] + ".md")
        if _EXCERPT_SENTINEL not in path.read_text(encoding="utf-8"):
            path.write_text(
                path.read_text(encoding="utf-8") + f"\n\n{_EXCERPT_SENTINEL}\n", encoding="utf-8"
            )
    return str(root)


def _redigest(raw: dict[str, Any], root: Path, *, only_first: bool = False) -> None:
    """Bring the payload's byte claims back in line with the base as it now stands.

    Needed because ``verify_against_bundle`` refuses a stale digest FIRST: an arm that edits a
    concept file and then expects some LATER refusal would be green off the digest check instead,
    which is how ``test_a_base_that_cannot_say_what_it_is_refused_at_this_door`` was measured
    green for the wrong reason (found by M13).
    """
    import hashlib

    for excerpt in raw["excerpts"][:1] if only_first else raw["excerpts"]:
        path = root / (excerpt["concept_id"] + ".md")
        excerpt["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
        excerpt["text"] = prepass.concept_text(path)
        excerpt["text_sha256"] = hashlib.sha256(excerpt["text"].encode("utf-8")).hexdigest()


def _payload_file(
    tmp_path: Path, *, sentinel: bool = False, redigest: bool = False, **mutate: Any
) -> str:
    """The payload as delivered, re-digested against the mounted base where an arm edits it."""
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    if sentinel:
        _redigest(raw, Path(_base(tmp_path, sentinel=True)), only_first=True)
    if redigest:
        _redigest(raw, Path(_base(tmp_path)))
    raw.update(mutate)
    name = "payload" + ("-sentinel" if sentinel else "") + ("-redigested" if redigest else "")
    path = tmp_path / f"{name}.json"
    path.write_text(json.dumps(raw), encoding="utf-8")
    return str(path)


def _config_file(tmp_path: Path, **overrides: Any) -> str:
    path = tmp_path / f"exploration{'-review' if overrides else ''}.json"
    path.write_text(
        json.dumps(
            {
                "max_rounds": 4,
                "max_tokens": 200_000,
                "max_stall_count": 2,
                "max_reset_count": 1,
                "max_plan_revisions": 0,
                "enable_plan_review": False,
                **overrides,
            }
        ),
        encoding="utf-8",
    )
    return str(path)


def _replies_file(tmp_path: Path, *, navigating: bool = False) -> str:
    path = tmp_path / f"replies{'-nav' if navigating else ''}.json"
    path.write_text(
        json.dumps(
            {
                "proposer": _PROPOSER_REPLY,
                "checker": "VERDICT: APPROVE",
                "manager": list(_MANAGER_STAGES),
                "navigator": list(_NAVIGATING_SCRIPT) if navigating else "NAVIGATOR: read it.",
                "hypothesiser": _HYPOTHESIS,
            }
        ),
        encoding="utf-8",
    )
    return str(path)


def _explore_argv(tmp_path: Path, *extra: str, navigating: bool = False, run_id: str) -> list[str]:
    """An argv the CLI ACCEPTS — the control every refusal arm below is measured against."""
    return [
        PROJECT_ID,
        "--docs-dir",
        _base(tmp_path),
        "--bundle-dir",
        _base(tmp_path),
        "--explore",
        "Finn den billigste besparelsen",
        "--explore-config",
        _config_file(tmp_path),
        "--scripted-replies",
        _replies_file(tmp_path, navigating=navigating),
        "--outbox-dir",
        str(tmp_path / "outbox"),
        "--run-id",
        run_id,
        *extra,
    ]


def _blob(messages: Any) -> str:
    """Text PLUS function calls and results — the corrected S7a-2 instrument.

    ``ScriptedChatClient``'s own sink records ``message.text`` alone, which measures a prompt
    carrying a tool RESULT at zero characters. That is precisely the difference between a
    navigator that reached the base and one holding tools over an empty index, so the sink is not
    enough here.
    """
    parts: list[str] = []
    for message in messages:
        text = getattr(message, "text", "") or ""
        if text:
            parts.append(text)
        for content in getattr(message, "contents", ()) or ():
            for attribute in ("result", "arguments"):
                value = getattr(content, attribute, None)
                if value is not None:
                    parts.append(str(value))
    return "\n".join(parts)


def _prompts(monkeypatch: pytest.MonkeyPatch) -> list[str]:
    """Every prompt the scripted roles were handed, in order.

    ``run.main`` imports ``scripted_factory`` INSIDE the function and discards the sink it passes,
    so the module attribute is the seam. Each client's instance ``_inner_get_response`` is then
    REBOUND to a recorder — rebinding rather than subclassing, because
    ``tests/test_scripted_client_consolidation`` keeps a registry of every site that DEFINES that
    method. One list is shared by every role, which is what lets this see the manager's very first
    prompt: the task message the seed rides in.
    """
    sink: list[str] = []
    real = simulation_module.scripted_factory

    def wrapper(replies: Any, _discarded: Any) -> Any:
        inner = real(replies, [])

        def build(role: str) -> Any:
            client = inner(role)
            original = client._inner_get_response

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

            client._inner_get_response = recording
            return client

        return build

    monkeypatch.setattr(simulation_module, "scripted_factory", wrapper)
    return sink


def _artefact(tmp_path: Path, run_id: str) -> dict[str, Any]:
    path = tmp_path / "outbox" / f"{run_id}-exploration.json"
    assert path.exists(), "the exploration artefact must be written even when the run failed"
    return json.loads(path.read_text(encoding="utf-8"))


def _refuse_model(monkeypatch: pytest.MonkeyPatch) -> None:
    """Any model client construction becomes a failure, so a refusal that fired AFTER the spend is
    distinguishable from one that fired before it. At the exit code the two look identical."""

    def refuse(profile: Any) -> Any:  # pragma: no cover - reached only by a regression
        raise AssertionError("a model client was built despite a refusal")

    monkeypatch.setattr(run_module, "_default_factory", refuse)


# --- (b) the cut lands in the starting point, and the tools stay ------------------------------


def test_the_delivered_text_reaches_the_explorations_first_prompt(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The seed IS the starting point, not a flag that was parsed and dropped (the F4 class)."""
    sink = _prompts(monkeypatch)
    rc = main(
        _explore_argv(
            tmp_path,
            "--prepass-seed",
            _payload_file(tmp_path, sentinel=True),
            run_id="seeded",
        )
    )
    assert rc == 0, rc
    assert sink, "no scripted role was ever asked anything"
    assert _EXCERPT_SENTINEL in sink[0], (
        "the delivered excerpt must be in the FIRST prompt — a cut that arrives later is not a "
        "starting point"
    )


def test_without_the_flag_the_first_prompt_carries_no_cut(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """(d) The control. Without it the arm above is satisfied by a run that built no prompt."""
    sink = _prompts(monkeypatch)
    _base(tmp_path, sentinel=True)
    rc = main(_explore_argv(tmp_path, run_id="unseeded"))
    assert rc == 0, rc
    assert sink, "no scripted role was ever asked anything"
    assert _EXCERPT_SENTINEL not in " ".join(sink)


def test_the_seed_tells_the_model_it_may_read_past_the_cut(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The tools being ATTACHED is half the property; the model being told it has them is the
    other. ``render_context`` opens with "you have no tools to read further. What is below is all
    of it" — true on that arm, and a LIE on this one. A rendering that carried it here would leave
    a model obeying a boundary the run does not have, and every other arm in this file would stay
    green: the DATA blocks, the denominators and the tool list are identical between the two.
    """
    sink = _prompts(monkeypatch)
    rc = main(_explore_argv(tmp_path, "--prepass-seed", _payload_file(tmp_path), run_id="stance"))
    assert rc == 0, rc
    assert sink, "no scripted role was ever asked anything"
    first = sink[0]
    assert "STARTING POINT" in first and "stays reachable" in first, first[:400]
    assert "you have no tools to read further" not in first, (
        "that sentence belongs to --prepass-payload, where it is true; here it forbids exactly "
        "what this arm exists to allow"
    )


def test_a_base_that_cannot_say_what_it_is_refused_at_this_door(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """This door OPENS a base, so it carries the declared-id agreement gate itself (S7a-3: every
    opening door gets its own call and its own mutation). Two concepts naming two corpora make a
    base that cannot be the one a cut is OF, and the refusal must land before the first model
    call rather than inside a tool three rounds later."""
    _refuse_model(monkeypatch)
    sink = _prompts(monkeypatch)
    root = Path(_base(tmp_path))
    for concept, declared in (
        ("tiltak-led-retrofit.md", "a-second-corpus"),
        ("metode-ipmvp-a.md", "a-third-corpus"),
    ):
        path = root / concept
        lines = path.read_text(encoding="utf-8").split("\n")
        lines.insert(1, f"bundle_id: {declared}")
        path.write_text("\n".join(lines), encoding="utf-8")

    # TWO concepts, and the payload RE-DIGESTED against the edited base. Both are load-bearing:
    # ``assert_declared_ids_agree`` reads concepts only, so one declaration is a base that agrees
    # with itself; and a stale digest is refused earlier, which would make this arm green off a
    # different gate entirely (measured — M13 turned the one-concept version red).
    rc = main(
        _explore_argv(
            tmp_path, "--prepass-seed", _payload_file(tmp_path, redigest=True), run_id="split"
        )
    )

    assert rc == 1
    err = capsys.readouterr().err
    assert "cannot say what it is" in err, err
    assert sink == [], "the refusal must land before the first model call, not after it"


#: A phrase from the base's ``index.md`` heading, plus the structural key only a catalogue entry
#: carries. Neither is in any delivered excerpt — ``index.md`` is not a concept file — so their
#: only route into a prompt is a ``list_bundles`` RESULT, which is the discriminator between a
#: tool that reached the base and one holding four tools over an empty index.
#:
#: **ASCII on purpose.** A first attempt used "rask smaaskala-testing og validering" spelled with
#: the Norwegian letter, and the arm was red against a working implementation: the tool result is
#: serialised into the prompt with ``\uXXXX`` escapes, so the probe could not match. That is the
#: S2c measurement repeated ("a fravaer that was instrument error, not fact") — the probe is the
#: first thing to falsify, not the code.
_INDEX_ONLY = "Bygg-energi mikro-eksempel"
_CATALOGUE_KEY = '"index_excerpt"'


def test_the_navigator_tools_survive_the_seed(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """(b) second half, and the whole difference from ``--prepass-payload``: a scripted navigator
    still READS PAST the seed with a cut in play.

    **Asserted on the RESULT, not on the record of the call.** ``tool_calls`` alone was measured
    green against the mutation this arm exists for (M3: build the workflow with no bases when a
    seed is present, i.e. collapse this arm into the other one). The recorder appends BEFORE
    ``call_next``, so a tool that raised ``unknown knowledge base`` leaves a byte-identical trace
    — "a tool was called" and "the base was open" are different facts, and only the second one is
    this arm. So the call record is kept AND the base's own index text must come back.
    """
    sink = _prompts(monkeypatch)
    rc = main(
        _explore_argv(
            tmp_path,
            "--prepass-seed",
            _payload_file(tmp_path),
            navigating=True,
            run_id="seeded-nav",
        )
    )
    assert rc == 0, rc
    calls = [c["name"] for c in _artefact(tmp_path, "seeded-nav")["tool_calls"]]
    assert calls[:2] == ["list_bundles", "read_bundle"], (
        "a seeded exploration must still be able to read past its seed — that is the ONE thing "
        f"that separates this arm from --prepass-payload; got {calls!r}"
    )
    assert _INDEX_ONLY not in sink[0] and _CATALOGUE_KEY not in sink[0], (
        "control: if the seed itself carried this text the assertion below would prove nothing"
    )
    assert any(_INDEX_ONLY in blob and _CATALOGUE_KEY in blob for blob in sink[1:]), (
        "the tools must reach the BASE, not merely exist: a navigator holding four tools over an "
        "empty index calls them, is recorded, and learns nothing"
    )


def test_the_cut_never_enters_the_commissions_objective(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    """The seed rides the TASK MESSAGE, never ``prompt``: ``_finish`` builds ``Mandate.objective``
    from the prompt and ``announce`` prints it back to the person who wrote it, so a cut folded in
    there would make the commission unreadable and would follow it into every artefact."""
    rc = main(
        _explore_argv(
            tmp_path, "--prepass-seed", _payload_file(tmp_path, sentinel=True), run_id="obj"
        )
    )
    assert rc == 0, rc
    out = capsys.readouterr().out
    assert "Finn den billigste besparelsen" in out, "the objective must still be announced"
    assert _EXCERPT_SENTINEL not in out


# --- (c) the declaration says STARTING POINT, and arm A says the opposite ----------------------


def test_the_notice_says_the_rest_stayed_reachable(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    rc = main(_explore_argv(tmp_path, "--prepass-seed", _payload_file(tmp_path), run_id="notice"))
    assert rc == 0, rc
    out = capsys.readouterr().out
    assert "SEEDED" in out and "reachable" in out, out


def test_the_artefact_declares_the_cut_and_that_it_was_a_starting_point(tmp_path: Path) -> None:
    """(c) The artefact is the half a terminal cannot carry: § 2.3 asks the consumer to DECLARE
    its cut, and a reader of ``{run_id}-exploration.json`` has to be able to tell a seeded run
    from a bounded one without having watched it."""
    rc = main(_explore_argv(tmp_path, "--prepass-seed", _payload_file(tmp_path), run_id="declared"))
    assert rc == 0, rc
    declared = _artefact(tmp_path, "declared")["prepass"]
    assert declared is not None, "a seeded run must declare its cut where a machine can read it"
    assert declared["rest_reachable"] is True
    assert (declared["considered"], declared["withheld"], declared["delivered"]) == (5, 1, 4)
    assert declared["bundle_id"] == DECLARED_ID
    assert declared["question"]


def test_the_replacing_arm_declares_the_opposite(tmp_path: Path) -> None:
    """The PAIRED CONTROL for the arm above: a constant ``rest_reachable: true`` would satisfy it.
    Arm A withdraws the tools, so its declaration must say the rest was NOT reachable — measured
    end to end through ``--prepass-payload``'s own artefact, not asserted on a constructor."""
    rc = main(
        [
            PROJECT_ID,
            "--docs-dir",
            _base(tmp_path),
            "--bundle-dir",
            _base(tmp_path),
            "--scripted-replies",
            _replies_file(tmp_path),
            "--outbox-dir",
            str(tmp_path / "outbox"),
            "--run-id",
            "replaced",
            "--prepass-payload",
            _payload_file(tmp_path),
        ]
    )
    assert rc == 0, rc
    body = json.loads((tmp_path / "outbox" / "replaced-prepass.json").read_text(encoding="utf-8"))[
        "prepass"
    ]
    assert body["rest_reachable"] is False


def test_without_the_flag_the_artefact_declares_no_cut(tmp_path: Path) -> None:
    """(d) ``None``, not an empty declaration: no cut was given is an honest positive statement,
    and inventing a zero-cut would say a pre-pass ran."""
    rc = main(_explore_argv(tmp_path, run_id="plain"))
    assert rc == 0, rc
    assert _artefact(tmp_path, "plain")["prepass"] is None


# --- (a) the refusals, each with an rc-0 control ----------------------------------------------


def test_it_requires_explore(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    _refuse_model(monkeypatch)
    rc = main(
        [
            PROJECT_ID,
            "--docs-dir",
            _base(tmp_path),
            "--bundle-dir",
            _base(tmp_path),
            "--scripted-replies",
            _replies_file(tmp_path),
            "--prepass-seed",
            _payload_file(tmp_path),
        ]
    )
    assert rc == 1
    err = capsys.readouterr().err
    assert "--prepass-seed" in err and "--explore" in err, err


def test_the_same_argv_with_explore_is_accepted(tmp_path: Path) -> None:
    """The rc-0 control for the arm above."""
    assert (
        main(_explore_argv(tmp_path, "--prepass-seed", _payload_file(tmp_path), run_id="ok")) == 0
    )


def test_the_two_arms_are_refused_together_by_name(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """Two opposite arms of one decision. The message must name BOTH flags: falling through to
    "--prepass-payload and --explore cannot be combined" names neither of the two the operator
    actually put in conflict, and an arm asserting on the token ``--prepass-payload`` alone would
    be green against exactly that fall-through (økt 57's rule)."""
    _refuse_model(monkeypatch)
    rc = main(
        _explore_argv(
            tmp_path,
            "--prepass-seed",
            _payload_file(tmp_path),
            "--prepass-payload",
            _payload_file(tmp_path),
            run_id="both",
        )
    )
    assert rc == 1
    err = capsys.readouterr().err
    assert "two opposite arms" in err, err
    assert "--prepass-seed" in err and "--prepass-payload" in err, err


def test_the_replacing_arm_is_still_refused_with_explore_in_the_same_words(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """(d) M32/F4 STANDS. B is a new flag, never a loosening of that refusal — and the wording is
    the one only that refusal produces."""
    _refuse_model(monkeypatch)
    rc = main(_explore_argv(tmp_path, "--prepass-payload", _payload_file(tmp_path), run_id="a"))
    assert rc == 1
    assert "cannot be combined" in capsys.readouterr().err


def test_it_is_refused_in_portfolio_mode_by_name(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    _refuse_model(monkeypatch)
    rc = main(
        [
            "--portfolio",
            "--scripted-replies",
            _replies_file(tmp_path),
            "--prepass-seed",
            _payload_file(tmp_path),
        ]
    )
    assert rc == 1
    assert "--portfolio" in capsys.readouterr().err


def test_portfolio_mode_without_the_flag_is_accepted(tmp_path: Path) -> None:
    """The rc-0 control for the arm above."""
    assert main(["--portfolio", "--scripted-replies", _replies_file(tmp_path)]) == 0


def _ledger_file(tmp_path: Path) -> str:
    """A JSON **ARRAY**. An object is refused by the ledger loader itself, which would make the
    report arm below red for the wrong reason (measured in økt 89)."""
    path = tmp_path / "ledger.json"
    path.write_text(json.dumps([]), encoding="utf-8")
    return str(path)


def test_it_is_refused_in_report_mode(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """Report mode returns ABOVE every dispatch, so an omission here is a silent DROP."""
    _refuse_model(monkeypatch)
    rc = main(
        [
            "--report",
            "--ledger",
            _ledger_file(tmp_path),
            "--prepass-seed",
            _payload_file(tmp_path),
        ]
    )
    assert rc == 1
    assert "--report" in capsys.readouterr().err


def test_report_mode_without_the_flag_is_accepted(tmp_path: Path) -> None:
    """The rc-0 control for the arm above."""
    assert main(["--report", "--ledger", _ledger_file(tmp_path)]) == 0


def test_it_is_refused_with_a_checkpoint_dir_by_name(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """A parked leg declares the cut in ``{run_id}-exploration.json``; the RESUMED leg runs in a
    process that never saw the payload and overwrites the same file with ``prepass: null``. A
    declaration that evaporates halfway is worse than one refused, and it would do so silently."""
    _refuse_model(monkeypatch)
    rc = main(
        [
            *_explore_argv(tmp_path, run_id="parked"),
            "--explore-config",
            _config_file(tmp_path, enable_plan_review=True, max_plan_revisions=2),
            "--checkpoint-dir",
            str(tmp_path / "checkpoints"),
            "--prepass-seed",
            _payload_file(tmp_path),
        ]
    )
    assert rc == 1
    err = capsys.readouterr().err
    assert "--prepass-seed" in err and "--checkpoint-dir" in err, err


def test_a_checkpoint_dir_without_the_flag_parks_and_returns_zero(tmp_path: Path) -> None:
    """The rc-0 control: the asynchronous door is fine on its own, so the refusal above is about
    the pair rather than about the argv being broken."""
    rc = main(
        [
            *_explore_argv(tmp_path, run_id="parked-ok"),
            "--explore-config",
            _config_file(tmp_path, enable_plan_review=True, max_plan_revisions=2),
            "--checkpoint-dir",
            str(tmp_path / "checkpoints"),
        ]
    )
    assert rc == 0, rc


# --- the admission gate is the SAME one, at this door too --------------------------------------


def test_a_cut_of_another_base_is_refused_before_any_model_call(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """The seeding door OPENS a base, so it carries the admission gate itself rather than trusting
    the tool-side one to fire later. Measured on calls, never on the exit code: a refusal after
    the spend and one before it are the same rc."""
    _refuse_model(monkeypatch)
    sink = _prompts(monkeypatch)
    rc = main(
        _explore_argv(
            tmp_path,
            "--prepass-seed",
            _payload_file(tmp_path, bundle={"bundle_id": "a-different-corpus", "ref": "x"}),
            run_id="other",
        )
    )
    assert rc == 1
    assert "run refused:" in capsys.readouterr().err
    assert sink == [], "the refusal must land before the first model call, not after it"


def test_an_empty_delivery_is_refused_on_this_arm_too(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    """Delivered 0 is evidence of ABSENCE for this question at this ref. The tools are still
    there, so a caller could argue the run should proceed — but it would proceed as a PLAIN
    exploration while the operator asked for a seeded one, which is a silently downgraded order."""
    _refuse_model(monkeypatch)
    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
    withheld = [{"concept_id": e["concept_id"], "rule": "below_k"} for e in raw["excerpts"]]
    empty = _payload_file(
        tmp_path,
        excerpts=[],
        withheld=raw["withheld"] + withheld,
        denominators={
            "considered": raw["denominators"]["considered"],
            "withheld": raw["denominators"]["considered"],
            "delivered": 0,
        },
    )
    rc = main(_explore_argv(tmp_path, "--prepass-seed", empty, run_id="empty"))
    assert rc == 1
    assert "delivered 0" in capsys.readouterr().err


def test_a_missing_seed_file_refuses_without_a_traceback(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    _refuse_model(monkeypatch)
    rc = main(_explore_argv(tmp_path, "--prepass-seed", str(tmp_path / "nope.json"), run_id="miss"))
    assert rc == 1
    assert "run refused:" in capsys.readouterr().err


# --- the hosted surface, deliberately untouched -----------------------------------------------


def test_the_hosted_surface_refuses_the_field_without_being_edited() -> None:
    """MAJOR-4 / S7b precedent: the field enters NONE of the three sets, so the generic
    ``unknown field(s)`` 400 already answers it and Fase 4e's two halves stand."""
    from portfolio_optimiser import hosting

    assert "prepass_seed" not in hosting._ALLOWED_FIELDS
    with pytest.raises(ValueError, match="unknown field"):
        hosting._run_kwargs({"project_id": PROJECT_ID, "prepass_seed": "/x.json"})


# --- the README block --------------------------------------------------------------------------


def _readme_block(flag: str) -> str:
    """The prose block for ONE flag, extracted rather than substring-matched: ``--portfolio`` and
    ``--report`` occur all over the README, so a file-wide search cannot tell a documented refusal
    from an unrelated mention."""
    readme = (Path(__file__).parent.parent / "README.md").read_text(encoding="utf-8")
    start = readme.index(f"(`{flag}`)")
    end = readme.find("\n  **", start)
    return readme[start : end if end != -1 else len(readme)]


def test_the_readme_documents_the_flag_and_every_partner_refusal() -> None:
    block = _readme_block("--prepass-seed")
    for partner in (
        "--explore",
        "--prepass-payload",
        "--portfolio",
        "--report",
        "--checkpoint-dir",
    ):
        assert partner in block, partner
    # Customer-facing terminology: never "OKF bundle" on a published surface.
    assert "OKF bundle" not in block


def test_the_extractor_finds_a_block_that_has_existed_since_f4() -> None:
    """The known-positive control: an extractor that silently finds nothing would make the arm
    above green against a README with no block at all."""
    assert "--explore" in _readme_block("--plan-review")