portfolio-optimiser/tests/test_plan_review_progress_line_loadbearing.py

"""The plan review never shows an expert the word ``None`` and calls it progress.

``PlanReviewRequest.current_progress`` and ``ParkedExploration.current_progress`` were both built
with ``str(review.current_progress)``. The value is a ``MagenticProgressLedger | None``, and in
every run measured so far it is ``None`` (``docs/2026-09-02-misjonsreview-v2.md``, PM addendum 4) —
so ``str()`` produced the four characters ``None``, the renderer's ``if ... .strip()`` guard found
them truthy, and the terminal printed:

    --- progress so far ---
    None

That is worse than saying nothing. An expert being asked to SIGN a plan was shown a section that
looks like content and is not, and the same four characters were written into
``{run_id}-plan-review.json`` — the one thing that crosses the process boundary in the U12
asynchronous door, where nobody can ask what it meant.

Two halves, and each is asserted so the other cannot carry it:

(a) ``_progress_text`` is the ONE place a possibly-absent ledger becomes text: ``None`` becomes the
    EMPTY string, never ``"None"``. It sits beside ``_plan_text`` and is applied at BOTH
    construction sites — the synchronous request (``explore.py``, F4's terminal door) and the
    parked question (U12's file). The data layer says "absent" by being absent, which is what
    ``Bundle.skipped``'s empty-tuple rule and ``cost_baseline_notice``'s omission both do;
(b) the TERMINAL says it in words. This is the one place the order asks for a line rather than an
    omission, and the reason is the surface's own job: silence at a signing gate is exactly what
    this repo refuses when it makes EOF an error rather than an approval. An expert must be able
    to tell "there is no progress yet" from "the section was dropped for some reason".

The control arm is what keeps (b) from being a blanket replacement: with a REAL ledger, the real
ledger is shown, not the placeholder.
"""

from __future__ import annotations

import io
import json
from pathlib import Path
from typing import Any

from portfolio_optimiser import explore as ex


class _Ledger:
    """Stands in for ``MagenticProgressLedger``: something with a readable ``str``."""

    def __str__(self) -> str:  # pragma: no cover - exercised through the renderer
        return "LEDGER-SENTINEL: two documents read"


def _render(current_progress: str) -> str:
    out = io.StringIO()
    reviewer = ex.terminal_plan_reviewer(stream_in=io.StringIO("approve\n"), stream_out=out)
    reviewer(
        ex.PlanReviewRequest(
            index=0, plan="a plan", current_progress=current_progress, is_stalled=False
        )
    )
    return out.getvalue()


def test_an_absent_ledger_becomes_the_empty_string_not_the_word_none() -> None:
    """(a) The ONE conversion. ``str(None)`` is a lie the data layer must never carry."""
    assert ex._progress_text(None) == "", (
        "an absent progress ledger must be absent, not the four characters 'None'"
    )
    assert ex._progress_text(_Ledger()) == "LEDGER-SENTINEL: two documents read", (
        "a ledger that EXISTS must still be rendered — this is not a blanket blank"
    )


def test_the_terminal_says_there_is_no_ledger_yet_in_words() -> None:
    """(b) An expert asked to sign must be able to tell 'none yet' from 'section dropped'."""
    out = _render("")

    assert "None" not in out, f"the reviewer must never be shown a repr of absence: {out!r}"
    assert "progress so far" in out
    assert "no progress ledger yet" in out, (
        "silence at a signing gate is what this repo refuses elsewhere (EOF is an error, never an "
        f"approval); the absence must be stated: {out!r}"
    )


def test_a_real_ledger_is_shown_verbatim_and_not_replaced() -> None:
    """The CONTROL for (b): the placeholder must not swallow a ledger that exists."""
    out = _render("LEDGER-SENTINEL: two documents read")

    assert "LEDGER-SENTINEL: two documents read" in out
    assert "no progress ledger yet" not in out, (
        "a run WITH progress must show it — a placeholder that always fires proves nothing"
    )


def test_both_construction_sites_go_through_the_one_conversion() -> None:
    """(a), at the sites. ``str(...)`` at either place puts 'None' back — in the terminal on one
    side, and in the parked question file on the other, which is the only thing U12's asynchronous
    door carries across the process boundary.
    """
    source = Path(ex.__file__).read_text(encoding="utf-8")

    assert "str(review.current_progress)" not in source, (
        "str() on a MagenticProgressLedger | None is the defect; both sites must use _progress_text"
    )
    assert source.count("current_progress=_progress_text(review.current_progress)") == 2, (
        "both the synchronous request and the parked question must use the one conversion"
    )


def test_the_parked_question_file_carries_absence_as_absence(tmp_path: Path) -> None:
    """The data layer, end to end: what crosses the process boundary must not say 'None'."""
    parked = ex.ParkedExploration(
        prompt="p",
        request_id="r",
        checkpoint_id="c",
        index=0,
        plan="a plan",
        current_progress=ex._progress_text(None),
        is_stalled=False,
        bundle_dirs=(),
        contract=ex.ExplorationContract(
            max_rounds=1,
            max_tokens=1,
            max_stall_count=0,
            max_reset_count=1,
            max_plan_revisions=0,
            enable_plan_review=True,
        ),
        ledger=(),
        plan_reviews=(),
        hypotheses=(),
        tokens_spent=0,
        replans=0,
    )

    payload: dict[str, Any] = ex.parked_payload(parked)

    assert payload["current_progress"] == ""
    assert "None" not in json.dumps(payload["current_progress"])


# ---------------------------------------------------------------------------------------------
# The BEHAVIOURAL witnesses, one per construction site.
#
# Measured while gating this: reverting BOTH sites to the bare ``str()`` left only the
# source-inspection arm above red — a lint, not a gate, because the arms before it construct a
# ``PlanReviewRequest`` themselves and so never enter either site. Each site therefore gets a
# witness that drives the REAL door and reads the surface an expert reads: the terminal for F4's
# synchronous door, the parked question FILE for U12's asynchronous one, which is the only thing
# that crosses the process boundary.
# ---------------------------------------------------------------------------------------------

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

_REPLIES = {
    "proposer": json.dumps(
        {
            "measure": "LED-retrofit",
            "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000, "unit_cost": 1.0}],
            "claimed_saving_nok": 30000,
        }
    ),
    "checker": "VERDICT: APPROVE",
    "manager": json.dumps(
        {
            "is_request_satisfied": {"reason": "r", "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"},
        }
    ),
    "navigator": "NAVIGATOR: read the index.",
    "hypothesiser": "HYPOTHESIS: "
    + json.dumps(
        {"label": "Night setback", "rationale": "y", "requirement": None, "why_none": "scripted"}
    ),
}


def _files(tmp_path: Path) -> tuple[str, str]:
    config = tmp_path / "exploration.json"
    config.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,
            }
        ),
        encoding="utf-8",
    )
    replies = tmp_path / "replies.json"
    replies.write_text(json.dumps(_REPLIES), encoding="utf-8")
    return str(config), str(replies)


def _argv(tmp_path: Path, *extra: str) -> list[str]:
    config, replies = _files(tmp_path)
    return [
        _PID,
        "--docs-dir",
        str(_BUNDLE_DIR),
        "--bundle-dir",
        str(_BUNDLE_DIR),
        "--explore",
        "Find the cheapest saving.",
        "--explore-config",
        config,
        "--scripted-replies",
        replies,
        "--outbox-dir",
        str(tmp_path / "outbox"),
        "--run-id",
        _RUN_ID,
        *extra,
    ]


def _progress_section(text: str) -> str:
    """The line the surface printed UNDER the progress header — the one the defect corrupted."""
    lines = text.splitlines()
    assert "--- progress so far ---" in lines, f"the section must be shown at all: {text!r}"
    return lines[lines.index("--- progress so far ---") + 1]


def test_the_synchronous_door_shows_absence_in_words_end_to_end(
    tmp_path, capsys, monkeypatch
) -> None:
    """Site 1, driven for real: F4's terminal door, with a run whose ledger is genuinely ``None``.

    Detach point: ``current_progress=str(review.current_progress)`` at the request site → the line
    under the header is the word ``None`` (RED).
    """
    import sys as _sys

    from portfolio_optimiser import run

    monkeypatch.setattr(_sys, "stdin", io.StringIO("approve\napprove\napprove\n"))

    run.main(_argv(tmp_path, "--plan-review"))

    line = _progress_section(capsys.readouterr().out)
    assert line == "(no progress ledger yet)", (
        f"an expert signing a plan must be told there is no ledger yet, not shown 'None': {line!r}"
    )


def test_the_parked_question_file_shows_absence_as_absence_end_to_end(tmp_path) -> None:
    """Site 2, driven for real: U12's asynchronous door. The file is the ONLY thing that crosses
    the process boundary, so 'None' written here is a question nobody can interrogate.

    Detach point: ``current_progress=str(review.current_progress)`` at the park site → the file
    carries the four characters ``None`` (RED).
    """
    from portfolio_optimiser import run

    run.main(_argv(tmp_path, "--checkpoint-dir", str(tmp_path / "checkpoints")))

    question = json.loads(
        (tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
    )
    assert question["current_progress"] == "", (
        "the parked question must carry absence as absence, never as the string 'None': "
        f"{question['current_progress']!r}"
    )