"""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}" )