portfolio-optimiser/tests/test_async_plan_review_loadbearing.py

"""U12 + asynchronous U13 (plan § D.2 row 3) — a plan review a human answers over DAYS.

F4 (økt 63) made "still spørsmål, be om svar, bruke svarene" reachable from the CLI, but only
SYNCHRONOUSLY: ``terminal_plan_reviewer`` blocks the loop on a human at a terminal, so the answer
has to arrive while the process is alive. Målbilde §3's time-scale is the other one — the expert
answers days later, in a process that never saw the run — and that is impossible without carrying
the workflow's state to disk.

**The measured trap this row is built around** (§ F row A4, and confirmed here against the
INSTALLED source rather than the plan's prose): ``FileCheckpointStorage.list_checkpoints``
(``_workflows/_checkpoint.py:386-388``) swallows a deserialisation failure into a
``logger.warning`` and returns an EMPTY list. Without ``MagenticPlanReviewRequest`` and
``MagenticPlanReviewResponse`` in ``allowed_checkpoint_types``, a resume therefore fails as an
ABSENCE — "nothing to resume" — not as an error. A test asserting "the listing is empty, so there
is nothing to resume" would be GREEN against exactly that defect, which is why every test here
asserts that the resume DID something instead.

**The discriminator, in both halves.** A door that writes a question file and a resume that reads
an answer file both pass "the expert was asked" while failing the målbilde. So the goal test drives
the F4 T1 shape across TWO process boundaries: ``revise`` written into an inbox on day 1 must reach
the manager, make it replan, and produce a SECOND question about the NEW plan — which an
always-approve resume, or one that discards the answer, cannot produce.

The witness is the artefacts, never scraped stdout: ``{run_id}-plan-review.json`` is the question
and ``{run_id}-exploration.json`` is the record of what was decided.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest

from portfolio_optimiser import explore as ex
from portfolio_optimiser import hitl, run

_REPO = Path(__file__).resolve().parents[1]
_BUNDLE_DIR = _REPO / "shared" / "examples" / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_RUN_ID = "async-review"

#: See ``tests/test_plan_review_cli_door_loadbearing.py`` for the measurement behind this: MAF
#: composes the task-ledger plan ``Message`` around the manager's answer verbatim, and ``Message``
#: has no ``__str__``, so ``str(review.plan)`` renders ``<…Message object at 0x…>``. The sentinel
#: can therefore appear only where the plan's TEXT was taken. Restated here rather than imported,
#: because this file already carries its own copy of the whole scripted-reply block: the two
#: scenarios park and resume differently and are meant to be readable apart.
_PLAN_SENTINEL = "PLAN-TEXT-SENTINEL-4b1e"

#: The defect asserted NEGATIVELY — any default object repr on a surface an expert reads, not only
#: this one.
_REPR_LEAK = " object at 0x"

_PROPOSER_REPLY = json.dumps(
    {
        "measure": "LED-retrofit",
        "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000, "unit_cost": 1.0}],
        "claimed_saving_nok": 30000,
    }
)
_MANAGER_REPLY = json.dumps(
    {
        "is_request_satisfied": {"reason": _PLAN_SENTINEL, "answer": True},
        "is_in_loop": {"reason": "r", "answer": False},
        "is_progress_being_made": {"reason": "r", "answer": True},
        "next_speaker": {"reason": "r", "answer": "hypothesiser"},
        "instruction_or_question": {"reason": "r", "answer": "go"},
    }
)
_REPLIES = {
    "proposer": _PROPOSER_REPLY,
    "checker": "VERDICT: APPROVE",
    "manager": _MANAGER_REPLY,
    "navigator": "NAVIGATOR: read the index.",
    "hypothesiser": "HYPOTHESIS: " + json.dumps({"label": "Night setback", "rationale": "y"}),
}

_FEEDBACK = "Also test night setback on the ventilation."


# ---------------------------------------------------------------------------------------------
# Fixture plumbing: the two operator surfaces, driven exactly as an operator would drive them
# ---------------------------------------------------------------------------------------------


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


def _replies_file(tmp_path: Path) -> str:
    path = tmp_path / "replies.json"
    path.write_text(json.dumps(_REPLIES), encoding="utf-8")
    return str(path)


def _park_argv(tmp_path: Path, **config: Any) -> list[str]:
    """Day 1: explore with the ASYNCHRONOUS door armed. No reviewer at this terminal."""
    return [
        _PID,
        "--docs-dir",
        str(_BUNDLE_DIR),
        "--bundle-dir",
        str(_BUNDLE_DIR),
        "--explore",
        "Find the cheapest saving.",
        "--explore-config",
        _config_file(tmp_path, **config),
        "--scripted-replies",
        _replies_file(tmp_path),
        "--outbox-dir",
        str(tmp_path / "outbox"),
        "--checkpoint-dir",
        str(tmp_path / "checkpoints"),
        "--run-id",
        _RUN_ID,
    ]


def _resume_argv(tmp_path: Path) -> list[str]:
    """Day N: a process that never saw the run, resuming from the checkpoint and the answer."""
    return [
        _PID,
        "--docs-dir",
        str(_BUNDLE_DIR),
        "--bundle-dir",
        str(_BUNDLE_DIR),
        "--scripted-replies",
        _replies_file(tmp_path),
        "--outbox-dir",
        str(tmp_path / "outbox"),
        "--checkpoint-dir",
        str(tmp_path / "checkpoints"),
        "--review-inbox",
        str(tmp_path / "review-inbox"),
        "--resume",
        _RUN_ID,
    ]


def _question(tmp_path: Path) -> dict[str, Any]:
    path = tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json"
    assert path.exists(), "a parked exploration must leave the question where an expert can read it"
    return json.loads(path.read_text(encoding="utf-8"))


def _artefact(tmp_path: Path) -> 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 parked"
    return json.loads(path.read_text(encoding="utf-8"))


def _answer(tmp_path: Path, **payload: Any) -> None:
    """The expert's side of the loop: a file dropped into the review inbox, days later."""
    inbox = tmp_path / "review-inbox"
    inbox.mkdir(parents=True, exist_ok=True)
    body = {"run_id": _RUN_ID, "request_id": _question(tmp_path)["request_id"], **payload}
    (inbox / f"{_RUN_ID}-plan-review-answer.json").write_text(json.dumps(body), encoding="utf-8")


def _resume_in_a_fresh_process(tmp_path: Path) -> subprocess.CompletedProcess[str]:
    """The resume runs in its OWN interpreter, because an in-process resume would prove nothing.

    The whole claim of U12 is that the only thing crossing the boundary is what is on disk. The
    subprocess precedent is ``spikes/e_magentic_resume.py`` and ``test_hosting_loadbearing.py``.
    """
    return subprocess.run(
        [sys.executable, "-m", "portfolio_optimiser.run", *_resume_argv(tmp_path)],
        capture_output=True,
        text=True,
        cwd=str(_REPO),
    )


# ---------------------------------------------------------------------------------------------
# 1. THE GOAL — asked on day 1, answered on day N, and the answer USED
# ---------------------------------------------------------------------------------------------


def test_an_answer_written_days_later_reaches_the_manager_and_produces_a_new_question(
    tmp_path, capsys
) -> None:
    """T1: the whole row, across two process boundaries.

    Day 1 parks at the plan review. The expert writes ``revise`` into the inbox. A FRESH
    interpreter resumes from the checkpoint alone — and because the feedback reached the manager,
    the manager replans and asks AGAIN about the NEW plan.

    RED against a resume that discards the answer, and RED against one that always approves: both
    yield a single review and no second question. RED against an in-process-only door: there is no
    checkpoint for the child to resume from.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err

    first = _question(tmp_path)
    assert first["index"] == 0
    assert first["request_id"], "without the request id the answer can never be routed back"

    _answer(tmp_path, decision="revise", feedback=_FEEDBACK)
    completed = _resume_in_a_fresh_process(tmp_path)
    assert completed.returncode == 0, completed.stderr

    second = _question(tmp_path)
    assert second["index"] == 1, (
        "a revision must produce a SECOND question about the replanned plan — a resume that "
        "discarded the answer would finish, or park again on the SAME review"
    )
    assert second["request_id"] != first["request_id"], (
        "the replanned review is a new request; reusing the old id would route the next answer "
        "into a request the orchestrator has already retired"
    )

    reviews = _artefact(tmp_path)["plan_reviews"]
    assert [r["decision"] for r in reviews] == ["revise"], (
        "the record must carry the decision the expert actually made"
    )
    assert reviews[0]["feedback"] == _FEEDBACK, (
        "what a human told the loop is worth nothing paraphrased"
    )


def test_the_answer_is_carried_all_the_way_to_a_completed_exploration(tmp_path, capsys) -> None:
    """T2: the control for T1 — the door does not only re-ask, it can also FINISH.

    A gate that could only ever park again would be a hang wearing a loop's clothes. Answering
    ``approve`` must let the exploration conclude and the pipeline run, leaving NO open question.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    _answer(tmp_path, decision="approve")

    completed = _resume_in_a_fresh_process(tmp_path)
    assert completed.returncode in (0, 1), completed.stderr
    assert "Traceback" not in completed.stderr, completed.stderr

    artefact = _artefact(tmp_path)
    assert artefact["completed"] is True, (
        "an approved review must let the exploration conclude, not park again"
    )
    assert [r["decision"] for r in artefact["plan_reviews"]] == ["approve"]
    assert (
        hitl.pending_plan_reviews(str(tmp_path / "outbox"), str(tmp_path / "review-inbox")) == []
    ), "a concluded exploration leaves no question waiting for anybody"


# ---------------------------------------------------------------------------------------------
# 2. THE MEASURED TRAP — a checkpoint that cannot be read back
# ---------------------------------------------------------------------------------------------


def test_the_parked_checkpoint_can_actually_be_read_back(tmp_path, capsys) -> None:
    """T3: the trap, asserted POSITIVELY.

    ``list_checkpoints`` turns a blocked deserialisation into an empty list, so the failure mode
    is silence. This asserts the opposite of silence: the id the question file names resolves to a
    checkpoint that loads. RED the moment ``_ALLOWED_CHECKPOINT_TYPES`` stops naming both types.
    """
    import asyncio

    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    checkpoint_id = _question(tmp_path)["checkpoint_id"]
    assert checkpoint_id, "parking on an unreadable checkpoint is an unanswerable question"

    storage = ex.checkpoint_storage(str(tmp_path / "checkpoints"))
    loaded = asyncio.run(storage.load(checkpoint_id))
    assert loaded.checkpoint_id == checkpoint_id


def test_a_park_with_no_readable_checkpoint_refuses_instead_of_writing_a_dead_question(
    tmp_path, capsys, monkeypatch
) -> None:
    """T4: fail LOUDLY where the framework fails silently.

    If the listing comes back empty there is nothing to resume from, and writing the question
    anyway would hand an expert a review whose answer can never be applied — the fourth face of
    the verification law, built into our own surface. Simulated by emptying the allow-list, which
    is exactly what produces an empty listing in the installed source.

    It leaves as a RAISE, not an rc-1 refusal, and that is the consistent call rather than a
    softer one: argv was fine and the loop had already spent, so this is the run failing — the
    same channel ``BudgetExceeded`` and an unreadable marked hypothesis use. What the door owes is
    that it fails LOUDLY where the framework fails silently, and that no dead question is left
    behind for somebody to answer into the void.
    """
    monkeypatch.setattr(ex, "_ALLOWED_CHECKPOINT_TYPES", ())

    with pytest.raises(ex.CheckpointUnreadable):
        run.main(_park_argv(tmp_path))

    capsys.readouterr()
    assert not (tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").exists(), (
        "a question nobody can answer must not be written at all"
    )


# ---------------------------------------------------------------------------------------------
# 3. FAIL-CLOSED ON THE EXPERT'S OWN INPUT (the F4 rule, on a file instead of a terminal)
# ---------------------------------------------------------------------------------------------


def test_an_answer_outside_the_vocabulary_is_refused_never_read_as_a_sign_off(
    tmp_path, capsys
) -> None:
    """T5: the closed vocabulary survives the move from stdin to a file.

    ``terminal_plan_reviewer`` re-asks anything it does not recognise; a file cannot be re-asked,
    so the only honest answer is a refusal. Reading it as approval would sign a plan nobody signed.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    _answer(tmp_path, decision="looks fine to me")

    completed = _resume_in_a_fresh_process(tmp_path)

    assert completed.returncode == 1, completed.stdout
    assert "looks fine to me" in completed.stderr, completed.stderr
    assert _artefact(tmp_path)["plan_reviews"] == [], (
        "an unreadable answer must not be recorded as a decision"
    )


def test_a_revision_with_nothing_to_revise_is_refused(tmp_path, capsys) -> None:
    """T6: ``revise`` without feedback is the same defect wearing a valid token.

    ``PlanReviewDecision.revise`` refuses an empty revision at the library door too, and that is
    exactly why the assertion here is on the SHAPE of the failure rather than on the exit code:
    measured, removing the inbox guard still gives rc 1 and still puts the word "revise" on
    stderr — as a TRACEBACK out of the library. A test that stopped at those two facts could not
    tell a fail-closed door from an unhandled exception, so it asserts the structured refusal.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    _answer(tmp_path, decision="revise", feedback="   ")

    completed = _resume_in_a_fresh_process(tmp_path)

    assert completed.returncode == 1, completed.stdout
    assert "Traceback" not in completed.stderr, completed.stderr
    assert "run refused" in completed.stderr, completed.stderr
    assert "revise" in completed.stderr.lower(), completed.stderr


def test_an_answer_to_a_different_review_is_refused_never_applied_to_this_one(
    tmp_path, capsys
) -> None:
    """T7: staleness is a refusal, not a silent misapplication.

    Two reviews of one run share a file name, so the answer names the ``request_id`` it answers.
    An answer left over from the previous round must not be applied to the current question — that
    would sign off a plan the expert never saw.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    _answer(tmp_path, decision="approve")
    inbox = tmp_path / "review-inbox" / f"{_RUN_ID}-plan-review-answer.json"
    stale = json.loads(inbox.read_text(encoding="utf-8"))
    stale["request_id"] = "a-request-from-last-week"
    inbox.write_text(json.dumps(stale), encoding="utf-8")

    completed = _resume_in_a_fresh_process(tmp_path)

    assert completed.returncode == 1, completed.stdout
    assert "a-request-from-last-week" in completed.stderr, completed.stderr


def test_a_resume_with_no_answer_yet_refuses_before_spending_anything(tmp_path, capsys) -> None:
    """T8: "not answered yet" is the normal state of this door, and it must be cheap.

    The hoist rule from økt 57: a refusal that fires AFTER the model calls is indistinguishable
    from one that fires before, by exit code alone — so this asserts that NOTHING was spent.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    before = json.loads(
        (tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
    )

    completed = _resume_in_a_fresh_process(tmp_path)

    assert completed.returncode == 1, completed.stdout
    after = json.loads(
        (tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
    )
    assert after == before, (
        "a resume with no answer must not touch the run at all — an unchanged question file is "
        "what proves the exploration was never restarted"
    )


# ---------------------------------------------------------------------------------------------
# 4. THE PENDING REGISTRY (hitl.py) — who is still waiting on whom
# ---------------------------------------------------------------------------------------------


def test_a_parked_review_is_pending_until_its_own_answer_lands(tmp_path, capsys) -> None:
    """T9: ``hitl.pending_plan_reviews`` is the machine-readable "still waiting", mirroring
    ``hitl.pending`` for proposals: an outbox question whose answer is not yet in the inbox.

    The join is on ``request_id``, so an answer to a DIFFERENT review leaves the question pending
    rather than quietly clearing it — the same fail-closed rule the resume applies.
    """
    outbox, inbox = str(tmp_path / "outbox"), str(tmp_path / "review-inbox")
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err

    waiting = hitl.pending_plan_reviews(outbox, inbox)
    assert [p.run_id for p in waiting] == [_RUN_ID]

    # BLOCKER-1, on the ASYNCHRONOUS door. ``assert waiting[0].plan`` was truthiness, and a repr is
    # truthy: the register said an expert could answer while what it showed them was
    # ``<…Message object at 0x…>``. Both surfaces the expert actually reads are asserted — the
    # question file ``explore.py:1478`` writes, and the register that joins it to the inbox.
    question = _question(tmp_path)
    assert _PLAN_SENTINEL in question["plan"], (
        "the parked question is the ONLY thing crossing the process boundary; an expert answering "
        "days later can read nothing else, so it must carry the plan as text"
    )
    assert _REPR_LEAK not in question["plan"], (
        f"no default object repr may stand in for the plan: {question['plan'][:80]!r}"
    )
    assert _PLAN_SENTINEL in waiting[0].plan, (
        "an expert cannot answer a review that does not show them the plan — and a repr shows "
        "them nothing they can judge"
    )

    _answer(tmp_path, decision="approve")
    inbox_file = tmp_path / "review-inbox" / f"{_RUN_ID}-plan-review-answer.json"
    wrong = json.loads(inbox_file.read_text(encoding="utf-8"))
    wrong["request_id"] = "someone-elses-review"
    inbox_file.write_text(json.dumps(wrong), encoding="utf-8")
    assert hitl.pending_plan_reviews(outbox, inbox) == waiting, (
        "an answer to another review must not clear this one"
    )

    _answer(tmp_path, decision="approve")
    assert hitl.pending_plan_reviews(outbox, inbox) == []


# ---------------------------------------------------------------------------------------------
# 5. THE BUDGET MUST SPAN THE SUSPENSION, NOT RESTART WITH IT
# ---------------------------------------------------------------------------------------------


def test_a_resumed_exploration_does_not_get_a_fresh_budget(tmp_path, capsys) -> None:
    """T10: the hole a park would otherwise open.

    Both budget channels live in the process: a fresh ``TokenMeter`` and an empty ledger mean a
    resumed exploration could spend its whole cap AGAIN, once per park — unbounded consumption
    behind guards that all look satisfied (the S3.4 class). The suspended state therefore carries
    what was already spent, and the resume starts from it.

    RED when the carry-over is detached: the resumed run then reports a spend of its own calls
    only, and the ledger restarts at round 1.
    """
    assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
    parked = ex.load_parked(_question(tmp_path))
    assert parked.tokens_spent > 0, (
        "the parked run made model calls; a zero here would make the assertion below vacuous"
    )

    _answer(tmp_path, decision="approve")
    assert _resume_in_a_fresh_process(tmp_path).returncode in (0, 1)

    artefact = _artefact(tmp_path)
    assert artefact["tokens_spent"] > parked.tokens_spent, (
        "the resumed exploration must add to the day-1 spend, never start over from zero"
    )


def test_the_revision_cap_is_counted_across_the_suspension_not_restarted_by_it(
    tmp_path, capsys
) -> None:
    """T11: the hole the carry-over closes, and the reason it exists at all.

    A revise costs two manager calls, emits no ledger and consumes no round (§ F, A3), so
    ``max_plan_revisions`` is the ONLY bound on it. With ``max_plan_revisions=1`` a second revision
    must be refused — and refused on the strength of what the FIRST process did, which only the
    carried ``plan_reviews`` can say.

    RED when the carry-over is detached: every leg then counts zero prior revisions, the cap never
    binds, and the run parks a third time. That mutation left the entire suite green before this
    test existed, which is exactly the vacuous-gate class it was written against.
    """
    assert run.main(_park_argv(tmp_path, max_plan_revisions=1)) == 0, capsys.readouterr().err

    _answer(tmp_path, decision="revise", feedback=_FEEDBACK)
    assert _resume_in_a_fresh_process(tmp_path).returncode == 0
    assert _question(tmp_path)["index"] == 1, "the first revision must be applied"

    _answer(tmp_path, decision="revise", feedback="And once more.")
    second = _resume_in_a_fresh_process(tmp_path)

    assert second.returncode in (0, 1), second.stderr
    assert "Traceback" not in second.stderr, second.stderr
    artefact = _artefact(tmp_path)
    assert artefact["stop"] == "plan_revisions_exhausted", (
        "the second revision is over the cap and must STOP the exploration, never be sent"
    )
    assert (
        hitl.pending_plan_reviews(str(tmp_path / "outbox"), str(tmp_path / "review-inbox")) == []
    ), "a stopped exploration must not leave a third question waiting for anybody"


def test_the_review_history_survives_every_leg_not_just_the_last(tmp_path, capsys) -> None:
    """T16: the carried ``plan_reviews`` are the RECORD, and the record is the only witness.

    Written because the mutation that detaches ``trace.plan_reviews.extend(parked.plan_reviews)``
    left the ENTIRE suite green (measured, økt 64): the revision cap counts
    ``parked.plan_reviews`` DIRECTLY, so it binds either way, and the first two legs cannot tell
    the difference — a park with an empty carried history and one with none look identical until
    there are two reviews to carry. An unmeasured seam is this repo's recurring defect class, so
    the gate is the THIRD leg, where the two implementations finally diverge.

    RED when the carry-over is detached: the artefact then records only the review the LAST
    process saw — the day-1 revision vanishes from the run's own history — and the next question
    is numbered 1 again, so two distinct reviews of one run share an index.
    """
    second_feedback = "And check the pumps while you are at it."
    assert run.main(_park_argv(tmp_path, max_plan_revisions=2)) == 0, capsys.readouterr().err
    assert _question(tmp_path)["index"] == 0

    _answer(tmp_path, decision="revise", feedback=_FEEDBACK)
    assert _resume_in_a_fresh_process(tmp_path).returncode == 0
    assert _question(tmp_path)["index"] == 1, "the first revision must be applied"

    _answer(tmp_path, decision="revise", feedback=second_feedback)
    third = _resume_in_a_fresh_process(tmp_path)
    assert third.returncode == 0, third.stderr

    assert _question(tmp_path)["index"] == 2, (
        "the third question is the third review of this run — a resume that dropped the carried "
        "history would number it 1 again, and two different plans would share one index"
    )
    reviews = _artefact(tmp_path)["plan_reviews"]
    assert [r["feedback"] for r in reviews] == [_FEEDBACK, second_feedback], (
        "the record must carry every decision the expert made, not only the most recent one: "
        "what a human told the loop on day 1 is not superseded by what they said on day 2"
    )


def test_what_the_first_process_found_survives_into_the_resumed_mandate(tmp_path) -> None:
    """T12: the other half of the carry-over — the loop's own findings.

    A plan review can fire mid-run after a stall, and everything the loop found before it would be
    lost if the mandate were minted from only what the resuming process observed. Driven through a
    CRAFTED parked state (the ``budget_stop`` precedent) because the scripted manager never stalls,
    so the mid-run park has no reachable path today — but the carry-over that serves it does.

    RED when the ledger and hypotheses are dropped on resume: the mandate then names only what the
    second leg saw, and the returned ledger restarts at the resumed round.
    """
    import asyncio
    import dataclasses

    from portfolio_optimiser.simulation import scripted_factory

    factory = scripted_factory(_REPLIES, [])
    contract = ex.load_exploration_contract(_config_file(tmp_path))
    checkpoints = str(tmp_path / "checkpoints")

    with pytest.raises(ex.PlanReviewParked) as caught:
        asyncio.run(
            ex.explore(
                "Find the cheapest saving.",
                contract=contract,
                bundle_dirs=(str(_BUNDLE_DIR),),
                client_factory=factory,
                checkpoint_dir=checkpoints,
            )
        )

    carried = dataclasses.replace(
        caught.value.parked,
        hypotheses=("HYPOTHESIS: " + json.dumps({"label": "Carried", "rationale": "found first"}),),
        ledger=(
            ex.LedgerEntry(
                round_index=1,
                is_request_satisfied=False,
                is_in_loop=False,
                is_progress_being_made=True,
                next_speaker="hypothesiser",
                instruction_or_question="keep going",
                speaker_known=True,
            ),
        ),
    )
    result = asyncio.run(
        ex.resume_exploration(
            carried,
            ex.PlanReviewDecision.approve(),
            checkpoint_dir=checkpoints,
            client_factory=factory,
        )
    )

    assert "Carried" in [a.label for a in result.mandate.approaches], (
        "a hypothesis the FIRST process found must reach the mandate the second one mints"
    )
    assert result.ledger_log[0].instruction_or_question == "keep going", (
        "the resumed ledger must continue the suspended run's, not restart it"
    )


# ---------------------------------------------------------------------------------------------
# 6. THE CLI PARTITION — two doors onto one review, and the preconditions
# ---------------------------------------------------------------------------------------------


def test_the_two_review_doors_are_refused_together(tmp_path, capsys) -> None:
    """T11: ``--plan-review`` and ``--checkpoint-dir`` are two sources of one answer.

    Refused rather than ranked, for the reason ``--explore`` + ``--mandate`` is: silently
    preferring one would mean an operator who asked for the asynchronous door got the synchronous
    one, and found out by being blocked at a terminal.
    """
    rc = run.main([*_park_argv(tmp_path), "--plan-review"])

    err = capsys.readouterr().err
    assert rc == 1
    assert "--plan-review" in err and "--checkpoint-dir" in err, err


def test_an_asynchronous_door_without_somewhere_to_put_the_question_is_refused_early(
    tmp_path, capsys
) -> None:
    """T12: the økt-57 hoist. The question artefact IS the door — without ``--outbox-dir`` and
    ``--run-id`` the exploration would spend a full budget and then have nowhere to say what it
    was waiting for. Refused before the first model call; the empty checkpoint dir is what proves
    nothing ran.
    """
    argv = [a for a in _park_argv(tmp_path) if a not in ("--outbox-dir", str(tmp_path / "outbox"))]

    rc = run.main(argv)

    err = capsys.readouterr().err
    assert rc == 1
    assert "--outbox-dir" in err and "--checkpoint-dir" in err, err
    assert not (tmp_path / "checkpoints").exists() or not list(
        (tmp_path / "checkpoints").iterdir()
    ), "the refusal must fire before the exploration starts, not after it has paid"


def test_a_review_with_neither_door_is_still_refused_and_says_which_two_exist(
    tmp_path, capsys
) -> None:
    """T13: the F4 refusal is WIDENED, never weakened.

    ``enable_plan_review`` with no reviewer at all is still a hang. The message must now name BOTH
    doors, because an operator told only about ``--plan-review`` cannot find the asynchronous one.
    """
    argv = [
        a
        for a in _park_argv(tmp_path)
        if a not in ("--checkpoint-dir", str(tmp_path / "checkpoints"))
    ]

    rc = run.main(argv)

    err = capsys.readouterr().err
    assert rc == 1
    assert "--plan-review" in err and "--checkpoint-dir" in err, err


@pytest.mark.parametrize(
    "drop,expected",
    [
        (("--checkpoint-dir",), "--checkpoint-dir"),
        (("--review-inbox",), "--review-inbox"),
    ],
)
def test_a_resume_names_the_coordinate_it_is_missing(
    tmp_path, capsys, drop: tuple[str, ...], expected: str
) -> None:
    """T14: a resume needs the checkpoints AND the inbox. Refused by NAME rather than falling
    through to a generic message, so an operator is told which of the two to add.
    """
    argv = _resume_argv(tmp_path)
    for flag in drop:
        i = argv.index(flag)
        del argv[i : i + 2]

    rc = run.main(argv)

    err = capsys.readouterr().err
    assert rc == 1
    assert expected in err, err


def test_resume_and_explore_are_refused_together(tmp_path, capsys) -> None:
    """T15: two sources of one exploration. ``--resume`` continues the exploration recorded in the
    parked file; ``--explore`` starts a new one. Merging them would silently drop one prompt.
    """
    argv = [
        *_resume_argv(tmp_path),
        "--explore",
        "A different question entirely.",
        "--explore-config",
        _config_file(tmp_path),
    ]

    rc = run.main(argv)

    err = capsys.readouterr().err
    assert rc == 1
    assert "--resume" in err and "--explore" in err, err