portfolio-optimiser/tests/test_plan_review_cli_door_loadbearing.py

"""F4 (docs/2026-08-25-fable-misjonsreview.md) — "be om svar, bruke svarene" must be reachable
from an OPERATOR surface, not only from the library API.

Before this, ``explore(..., plan_reviewer=...)`` (``explore.py:849``) was the sole door onto the
synchronous plan review, and BOTH operator surfaces refused ``enable_plan_review`` outright
(``run.py:1975-1988``, ``hosting.py:167-172``) — measured against the source, not taken from the
review's prose. Målbilde's "still spørsmål, be om svar, bruke svarene" was therefore unreachable
by anyone who was not importing the package.

**The gate is the SECOND half of that phrase.** A door that prints the plan, reads a line and
throws it away passes "the operator was asked" and fails the målbilde — this repo's vacuous-gate
class, eight times over. Every test here is therefore built so an always-approve reviewer is RED:
the discriminator is that a ``revise`` reaches the manager, the manager replans, and the operator
is asked AGAIN about the NEW plan (the T15 shape from ``test_explore_loadbearing.py``, lifted to
the CLI), with the feedback recorded VERBATIM.

The witness is ``{run_id}-exploration.json`` rather than scraped stdout: ``trace_payload``
(``explore.py:285-293``) already carries the decisions, the feedback and the order, and it is
written from a ``finally`` — so it survives the one run that most needs it, the one a cap or an
unanswered review cut short.
"""

from __future__ import annotations

import io
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 hosting, run

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

_PROPOSER_REPLY = json.dumps(
    {
        "measure": "LED-retrofit",
        "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000, "unit_cost": 1.0}],
        "claimed_saving_nok": 30000,
    }
)
#: A string that exists ONLY inside the manager's scripted reply, so it reaches a surface exactly
#: when that surface rendered the plan's TEXT. MAF composes the task-ledger plan ``Message`` around
#: the manager's answer verbatim (measured), and ``Message`` has no ``__str__``
#: (``type(Message).__str__ is object.__str__``, measured) — so ``str(review.plan)`` yields
#: ``<agent_framework._types.Message object at 0x…>``, in which the sentinel cannot appear. It sits
#: in a ``reason`` field because nothing reads those: the ledger's ANSWERS drive the loop and are
#: asserted elsewhere, so keying the sentinel to them would change the run it is measuring.
_PLAN_SENTINEL = "PLAN-TEXT-SENTINEL-4b1e"

#: The shape of the defect itself, asserted NEGATIVELY. Stronger than the sentinel's absence: it
#: catches ANY object rendered by its default repr onto a surface a human reads, not only this one.
_REPR_LEAK = " object at 0x"

_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", "requirement": None, "why_none": "scripted"}
    ),
}

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


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 _argv(tmp_path: Path, *, plan_review: bool = True, **config: Any) -> list[str]:
    argv = [
        _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"),
        "--run-id",
        _RUN_ID,
    ]
    if plan_review:
        argv.append("--plan-review")
    return argv


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 failed"
    return json.loads(path.read_text(encoding="utf-8"))


def _stdin(monkeypatch: pytest.MonkeyPatch, text: str) -> None:
    monkeypatch.setattr(sys, "stdin", io.StringIO(text))


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


def test_an_operator_answer_typed_at_the_cli_reaches_the_manager_and_is_asked_again(
    tmp_path, capsys, monkeypatch
) -> None:
    """T1: the whole point. ``revise`` typed at the CLI reaches the manager, the manager replans,
    and the operator is asked to sign off on the NEW plan.

    RED against an always-approve reviewer (the vacuous door): that yields ONE review and
    ``["approve"]``, so both the count and the order fail. RED against a reviewer that reads the
    line and discards it: the feedback assertion fails and the manager is never asked twice.

    Detach point: drop the ``plan_reviewer=`` wiring in ``run.py`` → the CLI refuses instead
    (there is no reviewer), so this never runs at all.
    """
    _stdin(monkeypatch, f"revise {_FEEDBACK}\napprove\n")

    rc = run.main(_argv(tmp_path))

    err = capsys.readouterr().err
    assert "Traceback" not in err, err
    assert rc in (0, 1), f"expected a clean exit, got rc={rc} stderr={err!r}"

    reviews = _artefact(tmp_path)["plan_reviews"]
    assert [r["decision"] for r in reviews] == ["revise", "approve"], (
        "a revision must produce a SECOND review, not resume silently — an always-approve door "
        "gives ['approve']"
    )
    assert reviews[0]["feedback"] == _FEEDBACK, (
        "what a human told the loop is worth nothing paraphrased"
    )
    # BLOCKER-1: the two recorded plans come from two DIFFERENT lines — the revise branch writes
    # its own copy and the approve branch writes another. ``!= ""`` was green against a repr, so
    # reverting either line alone stayed green; the sentinel is what separates them.
    assert _PLAN_SENTINEL in reviews[0]["plan"], (
        "the REVISE record must carry the plan as text — an expert reading the artefact to see "
        "what they revised cannot use an object repr"
    )
    assert _PLAN_SENTINEL in reviews[1]["plan"], (
        "the second review must show the replanned plan as TEXT, not as a repr of the message "
        "that carried it"
    )


def test_the_operator_is_shown_the_plan_and_the_answer_vocabulary(
    tmp_path, capsys, monkeypatch
) -> None:
    """T2: a review nobody can read is a review nobody can answer. The prompt carries the plan
    AS TEXT, the progress and the two words that answer it.

    Detach point: print only "plan review?" → RED. The control for T1: T1 proves the answer is
    used, this proves the question was askable.

    **Sharpened for BLOCKER-1.** The old form asserted ``reviews[0]["plan"][:40] in out``, which
    an object repr satisfies just as well as the plan does — the same string was shown and stored,
    so the two agreed while both were unreadable. What discriminates is content that can only come
    from ``Message.text``: the sentinel, plus the negative that no default repr reached either
    surface. Mutation: revert ``explore.py:1395`` → the operator is shown a repr (stdout RED);
    revert ``:1404`` → the artefact stores one (artefact RED).
    """
    _stdin(monkeypatch, "approve\napprove\napprove\n")

    run.main(_argv(tmp_path))

    out = capsys.readouterr().out
    artefact = _artefact(tmp_path)
    reviews = artefact["plan_reviews"]
    assert reviews, "the run must actually have reached a plan review"
    assert reviews[0]["plan"][:40] in out, (
        "the operator must be shown the plan they are signing off"
    )
    assert _PLAN_SENTINEL in out, (
        "an operator asked to sign a plan must be shown the PLAN; a repr of the message that "
        "carried it is a signature given blind"
    )
    assert _PLAN_SENTINEL in reviews[0]["plan"], (
        "the artefact is the record of what was approved, so it must hold the plan as text too"
    )
    assert _REPR_LEAK not in out, (
        f"no default object repr may reach a surface a human reads; found in stdout: "
        f"{out[max(0, out.find(_REPR_LEAK) - 60) : out.find(_REPR_LEAK) + 20]!r}"
    )
    assert _REPR_LEAK not in json.dumps(artefact), (
        "nor may one be written into the artefact that outlives the run"
    )
    assert "approve" in out and "revise" in out, "the prompt must name the vocabulary it accepts"


def test_an_unrecognised_answer_is_asked_again_never_taken_as_a_sign_off(
    tmp_path, capsys, monkeypatch
) -> None:
    """T3: fail-closed on the operator's own input. Anything outside the vocabulary is re-asked;
    it is never read as approval, and never as a revision either.

    Detach point: treat any non-``revise`` line as approve → the first line ("yes please") would
    sign the plan off and the recorded decision would still be ``approve``, so the count of
    prompts is what discriminates: RED here, green there.
    """
    _stdin(monkeypatch, f"yes please\n\nrevise {_FEEDBACK}\napprove\n")

    run.main(_argv(tmp_path))

    out = capsys.readouterr().out
    reviews = _artefact(tmp_path)["plan_reviews"]
    assert [r["decision"] for r in reviews] == ["revise", "approve"], (
        "the junk line and the blank line must both be re-asked, not consumed as decisions"
    )
    assert reviews[0]["feedback"] == _FEEDBACK
    assert out.count("PLAN REVIEW") == 2, (
        "two REVIEWS were answered; a third prompt would mean a junk line was consumed as one"
    )


def test_end_of_input_never_becomes_an_approval_and_the_evidence_still_lands(
    tmp_path, monkeypatch
) -> None:
    """T4: the silence that must not be read as a yes.

    A pipe that ends — or an operator who walks away — leaves the review unanswered. Reading that
    as approval would let an autonomous loop run on a plan no human signed, which is the exact
    thing the door exists to prevent, and it would do so invisibly. It raises instead.

    The artefact is asserted TOO, and that is the load-bearing half: the write is in a ``finally``
    (``run.py``), so the run that failed still leaves a record of what the operator was asked and
    what they had answered so far. ``completed: false`` is what says the run never finished
    (``trace_payload``'s required field — an absent ``stop`` cannot say it).

    Detach point: return ``PlanReviewDecision.approve()`` at EOF → no exception, ``completed``
    true, and the loop runs on an unsigned plan (RED on all three).
    """
    _stdin(monkeypatch, "")

    with pytest.raises(ex.PlanReviewInputError):
        run.main(_argv(tmp_path))

    artefact = _artefact(tmp_path)
    assert artefact["completed"] is False
    assert artefact["plan_reviews"] == [], "nothing was decided, so nothing may be recorded"


# ---------------------------------------------------------------------------------------------
# 2. The refusals — every combination that would silently drop the flag
# ---------------------------------------------------------------------------------------------


def test_plan_review_without_an_exploration_is_refused_by_name(tmp_path, capsys) -> None:
    """T5: there is no plan to review without an exploration. Refused rather than loaded and
    dropped — the ``--explore-config`` precedent, verbatim.

    Detach point: accept it silently → RED (rc 0, flag ignored).
    """
    rc = run.main([_PID, "--docs-dir", str(_BUNDLE_DIR), "--plan-review", "--live-dry-run"])

    assert rc == 1
    assert "--plan-review" in capsys.readouterr().err


def test_plan_review_against_a_config_that_asks_for_no_review_is_refused(tmp_path, capsys) -> None:
    """T6: a reviewer nobody will ever call. ``explore()`` refuses this too, but as an
    ``ExplorationError`` — a ``RuntimeError``, outside ``main()``'s refusal tuple — so it would
    leave as a traceback instead of the rc-1 line. Hoisted here for that reason alone.

    The assertion names wording UNIQUE to this branch: after F4 the CLI has two refusals
    containing ``enable_plan_review``, and asserting on the shared token is this repo's
    "assert never on wording two branches share" defect (measured in økt 57).

    Detach point: leave it to ``explore()`` → RED (traceback, not rc 1).
    """
    rc = run.main(_argv(tmp_path, enable_plan_review=False, max_plan_revisions=0))

    assert rc == 1
    assert "no review is ever requested" in capsys.readouterr().err


def test_a_review_with_no_reviewer_still_refuses_and_now_names_the_door(tmp_path, capsys) -> None:
    """T7: the opposite half — the config asks for a review and no ``--plan-review`` was given.

    The refusal SURVIVES F4 (a run must never stop at a review nobody can answer), but its wording
    was a claim the surface made about itself: "the synchronous door is the library API" stopped
    being true the moment this CLI grew one. It now names the flag.

    Detach point: leave the old wording → RED. Same class as the Fase 3 credential claim.
    """
    rc = run.main(_argv(tmp_path, plan_review=False))

    assert rc == 1
    err = capsys.readouterr().err
    assert "no reviewer was offered" in err
    assert "--plan-review" in err, "the refusal must name the door that answers it"


def test_plan_review_belongs_to_single_project_mode(tmp_path, capsys) -> None:
    """T8: the documented partition. ``--explore`` is single-project-only and ``--plan-review``
    answers its review, so a ``--portfolio --plan-review`` argv has to hear which flag is wrong.

    The assertion names ``--portfolio`` rather than ``--plan-review``: the refusal below it
    (``--plan-review requires --explore``) names ``--plan-review`` too, and asserting on the shared
    token would pass against no partition entry at all — the økt-57 mutation, verbatim.

    Detach point: leave it out of ``single_only`` → RED (the message names --explore only, or the
    run falls through to the requires-refusal).
    """
    rc = run.main(["--portfolio", "--plan-review"])

    assert rc == 1
    assert "--portfolio" in capsys.readouterr().err


def test_plan_review_is_refused_in_report_mode(tmp_path, capsys) -> None:
    """T9: ``--report`` returns BEFORE every exploration refusal, so a flag missing from
    ``report_forbidden`` is silently dropped rather than refused — the reason that list enumerates
    every distinguishable flag in the first place.

    Detach point: leave it out of ``report_forbidden`` → rc 0 and a printed report, the flag gone
    without a word (RED).
    """
    ledger = tmp_path / "ledger.json"
    ledger.write_text(json.dumps({"entries": []}), encoding="utf-8")

    rc = run.main(["--report", "--ledger", str(ledger), "--plan-review"])

    assert rc == 1
    assert "mode-exclusive" in capsys.readouterr().err


@pytest.mark.asyncio
async def test_the_hosted_surface_still_refuses_and_names_the_cli_door() -> None:
    """T10: hosting keeps its refusal — a synchronous review would block the HTTP request on a
    reviewer that does not exist, and it would block the event loop that answers ``/readiness``
    while doing it. What changes is the honesty of the message: there is now an operator door,
    and the refusal says where.

    Detach point: leave the message pointing only at the library API → RED (the same claim-drift
    class as the Fase 3 credential line).
    """
    with pytest.raises(ValueError) as excinfo:
        await hosting.invoke(
            {
                "project_id": _PID,
                "docs_dir": str(_BUNDLE_DIR),
                "verdict_input": {"decision": "approved", "rationale": "expert reviewed"},
                "profile": "local",
                "bundle_dir": str(_BUNDLE_DIR),
                "explore_prompt": "p",
                "explore_contract": {
                    "max_rounds": 4,
                    "max_tokens": 200_000,
                    "max_stall_count": 2,
                    "max_reset_count": 1,
                    "max_plan_revisions": 2,
                    "enable_plan_review": True,
                },
            }
        )

    assert "--plan-review" in str(excinfo.value)


# ---------------------------------------------------------------------------------------------
# 3. A real argv, in a real process (the P4 precedent)
# ---------------------------------------------------------------------------------------------


def test_the_flag_answers_a_review_from_a_real_argv(tmp_path) -> None:
    """T11: in-process tests patch ``sys.stdin`` and call ``main()`` directly, so neither proves
    the flag exists on the parsed command line or that a real pipe reaches the reviewer. A child
    process settles both — the same reason the demo's stderr and the hosting shim are measured in
    a subprocess rather than with ``capsys``.

    Detach point: never add the argparse flag → the child exits 2 with an argparse usage error
    (RED).
    """
    argv = _argv(tmp_path)
    proc = subprocess.run(
        [sys.executable, "-m", "portfolio_optimiser.run", *argv],
        input=f"revise {_FEEDBACK}\napprove\n",
        capture_output=True,
        text=True,
        cwd=_REPO,
    )

    assert "unrecognized arguments" not in proc.stderr, proc.stderr
    assert "Traceback" not in proc.stderr, proc.stderr
    reviews = _artefact(tmp_path)["plan_reviews"]
    assert [r["decision"] for r in reviews] == ["revise", "approve"]
    assert reviews[0]["feedback"] == _FEEDBACK


def test_the_reviewer_reads_the_stream_that_exists_when_it_is_asked(monkeypatch) -> None:
    """T12: the streams are resolved at CALL time, not when the reviewer is built.

    A factory that captured ``sys.stdin`` at construction would answer from whatever stream
    happened to be installed when ``run.py`` built the reviewer — before the loop, before anything
    was asked. Nothing above catches that (every other test here installs its stream first), so the
    claim would be prose. This asks the question the other way round: build FIRST, swap AFTER.

    Detach point: resolve the streams in ``terminal_plan_reviewer``'s body instead of inside
    ``review`` → RED (the reviewer reads the stream that is gone).
    """
    reviewer = ex.terminal_plan_reviewer()
    monkeypatch.setattr(sys, "stdin", io.StringIO(f"revise {_FEEDBACK}\n"))
    monkeypatch.setattr(sys, "stdout", io.StringIO())

    decision = reviewer(
        ex.PlanReviewRequest(index=0, plan="a plan", current_progress="", is_stalled=False)
    )

    assert decision.feedback == _FEEDBACK