"""The verdict layer is REFUSED by ``read_file``, however the path was found. The gap S2c stated as an honesty limit and left open (``docs/2026-09-04-s2c-debatt-k2.md``, closing section): ``read_file`` is path-addressable to the ``type: verdict`` layer. No listing names it — ``Bundle.context_files`` drops it at every level, so neither ``read_bundle`` nor ``read_dir`` nor ``bundle_context`` ever mentions a verdict — but a GUESSED path reached one, and reaching one that way bypasses the gated ExpeL fold that is the only sanctioned route from a prior expert judgement into a hypothesis (målbilde §4/§5). MEASURED before this work: ``read_file`` on the fixture base's ``verdict-led-fro.md`` returned all 2 883 characters of it. The property was inherited from S7a-3 and, since S2c handed the DEBATE the navigator's four tools, became reachable from the pipeline as well — which is why the rule lives in ONE place, the tool, rather than in two callers. Rendering cannot hold it: S2c measured that a filter in the listing while ``read_file`` still serves the bytes is "a filter in name only", and prompt text cannot hold it either, because a model-chosen path is untrusted input by construction. Arms, each with a named detach point: * **(1) the exploration path refuses.** ``navigator_tools`` called directly, ``dimension=None`` — the exploration's own call. CONTROL: an ordinary concept file in the SAME base still reads, so the refusal is caused by the layer and not by an unreachable fixture. * **(2) the debate path refuses.** The same tool as ``run_project`` builds it, driven through the real seam, plus a BEHAVIOURAL run whose proposer manuscript asks for the verdict by path. * **(3) the gated route still works.** The seed verdict's realization signal still reaches the hypothesis prompt through the ExpeL fold — the refusal closes the ungated door, not the door. * **(4) a refused read is RECORDED, never silent.** The call stands in ``debate_tool_calls`` and in ``{run_id}-debate.json`` with its path, while the verdict's body reaches NO prompt. Recording is gated on its own: MEASURED, moving the recorder's append after ``call_next`` empties the trace for exactly this call, because a refused invocation never returns. * **(5) the rule reads the DOCUMENT, not the walk.** A verdict file no index links to — never navigated, therefore absent from ``Bundle.verdicts`` — is refused too. "However the path was found" is the whole point: gating on the navigation would leave the guessed path to an unlinked judgement open, which is the defect in its own disguise. * **(6) the ladder is intact.** ``index.md`` and ordinary concept files still read whole. """ from __future__ import annotations import json import shutil from collections.abc import Callable from pathlib import Path from typing import Any from agent_framework import BaseChatClient from portfolio_optimiser import okf from portfolio_optimiser.explore import VerdictLayerRefused, navigator_tools from portfolio_optimiser.run import run_project from portfolio_optimiser.simulation import ScriptedChatClient, scripted_factory from portfolio_optimiser.validator import ValidatedProposal from portfolio_optimiser.verdicts import seed_store_from_bundle BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" _PID = "BYGG-KONTOR-NORD" _VERDICT_FILE = "verdict-led-fro.md" _CONCEPT_FILE = "metode-ipmvp-a.md" _VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"} _VALID_REPLY = ( '{"measure":"LED-retrofit","affected_items":' '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],' '"claimed_saving_nok":30000}' ) _CHECKER_REPLY = "Reasoning holds.\nVERDICT: APPROVE" #: Written into a COPY of the base, into the verdict body alone. A sentinel of our own rather than #: a phrase lifted out of the shipped file: the ExpeL fold legitimately renders parts of a verdict #: into the generation prompt, so a probe made of the file's own prose could not tell the ungated #: read from the gated one. _VERDICT_SENTINEL = "VERDICT-LEAK-SENTINEL-q4w5e6" #: The exact ``_verdict_rationale`` fewshot string (test_step1_expel_loadbearing's marker). _REALIZATION_SIGNAL = "realiseringsgrad=0.82" def _tools(bundle_dir: str, dimension: str | None = None) -> dict[str, Any]: return {t.name: t for t in navigator_tools([bundle_dir], dimension=dimension)} async def _invoke(tool: Any, **arguments: Any) -> str: """A tool's answer as text (S2c's helper: ``invoke`` returns ``[Content]``, and a test that stringified the list would compare object reprs and pass against anything).""" return "".join(getattr(c, "text", "") or "" for c in await tool.invoke(arguments=arguments)) def _base_with_a_marked_verdict(tmp_path: Path) -> str: """A copy of the fixture base whose verdict body carries ``_VERDICT_SENTINEL``.""" copy = tmp_path / "bundle" shutil.copytree(BUNDLE_DIR, copy) verdict = copy / _VERDICT_FILE verdict.write_text( verdict.read_text(encoding="utf-8") + f"\n{_VERDICT_SENTINEL}\n", encoding="utf-8" ) return str(copy) def _base_with_an_unlinked_verdict(tmp_path: Path) -> str: """A copy of the fixture base plus a ``type: verdict`` file NO index links to, so navigation never reaches it and it is absent from ``Bundle.verdicts``.""" copy = tmp_path / "unlinked" shutil.copytree(BUNDLE_DIR, copy) (copy / "verdict-orphan.md").write_text( "---\ntype: verdict\ntitle: Orphan judgement\ndecision: approved\n---\n\n" f"{_VERDICT_SENTINEL}\n", encoding="utf-8", ) return str(copy) def _recording_factory( sink: list[str], *, script: dict[str, Any] | None = None ) -> Callable[[str], BaseChatClient]: """S2c's recording factory: every prompt blob (text PLUS function call/result contents) lands in ``sink``. ``Message.text`` alone measures a context-bearing prompt at a few characters.""" def factory(role: str) -> BaseChatClient: if script is not None and role in script: client: BaseChatClient = scripted_factory(script, [])(role) else: client = ScriptedChatClient( _CHECKER_REPLY if role == "checker" else _VALID_REPLY, role=role ) original = client._inner_get_response # type: ignore[attr-defined] def recording(*, messages, options, stream=False, **kwargs): # type: ignore[no-untyped-def] 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 attr in ("result", "arguments"): value = getattr(content, attr, None) if value: parts.append(str(value)) sink.append("\n".join(parts)) return original(messages=messages, options=options, stream=stream, **kwargs) client._inner_get_response = recording # type: ignore[attr-defined,method-assign] return client return factory # ------------------------------------------------------------------- (1) the exploration path async def test_the_exploration_path_refuses_a_verdict_document(tmp_path) -> None: """LOAD-BEARING (1): the exploration's own call — ``dimension=None``, which admits every dimension — still refuses the verdict layer. Detach point: remove the gate from ``read_file`` → RED.""" bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id tools = _tools(bundle_dir) answer = await _invoke(tools["read_file"], bundle_id=bundle_id, path=_VERDICT_FILE) assert answer.startswith(f"REFUSED ({VerdictLayerRefused.__name__})") # F99-D3 made the refusal a RETURN VALUE, so the property the gate exists for has to be # asserted on that value: the reason travels, the bytes never do. assert _VERDICT_SENTINEL not in answer async def test_an_ordinary_document_in_the_same_base_still_reads(tmp_path) -> None: """CAUSALITY CONTROL for (1): a concept file in the SAME base reads whole, so the refusal above is caused by the layer rather than by a fixture nothing can open.""" bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id body = await _invoke(_tools(bundle_dir)["read_file"], bundle_id=bundle_id, path=_CONCEPT_FILE) assert "IPMVP" in body, f"the control document did not read: {body[:120]!r}" async def test_no_listing_names_the_verdict_either(tmp_path) -> None: """The OTHER half of "no listing names it": ``read_bundle`` still hides the layer, so the two rungs agree. Without this the gate could be green while the listing advertised the document it then refuses — the disagreement S2c's ``in_dimension`` rule exists to prevent.""" bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id listing = await _invoke(_tools(bundle_dir)["read_bundle"], bundle_id=bundle_id) assert _VERDICT_FILE not in listing # ------------------------------------------------------------------------ (2) the debate path async def test_the_debate_path_refuses_a_verdict_document(tmp_path, monkeypatch) -> None: """LOAD-BEARING (2): the tool ``run_project`` hands the debate refuses the same document. Built through the REAL seam rather than by calling ``navigator_tools`` again: the two callers share one construction, and a test that rebuilt the tools itself would prove nothing about what the pipeline actually gave its agents (S2c arm (e)'s own form). Detach point: remove the gate from ``read_file`` → RED.""" import portfolio_optimiser.run as run_module bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id captured: list[list[Any]] = [] original = run_module.fresh_workflow def spy(*args: Any, **kwargs: Any) -> Any: captured.append(list(kwargs.get("tools") or [])) return original(*args, **kwargs) monkeypatch.setattr(run_module, "fresh_workflow", spy) sink: list[str] = [] await run_project( _PID, "local", docs_dir=bundle_dir, bundle_dir=bundle_dir, verdict_input=_VERDICT_INPUT, client_factory=_recording_factory(sink), ) assert captured, "the debate was never built" read_file = next(t for t in captured[0] if getattr(t, "name", "") == "read_file") answer = await _invoke(read_file, bundle_id=bundle_id, path=_VERDICT_FILE) assert answer.startswith(f"REFUSED ({VerdictLayerRefused.__name__})") assert _VERDICT_SENTINEL not in answer # --------------------------------------------------------------- (3) the gated route survives async def test_the_gated_expel_route_still_reaches_the_hypothesis( make_recording_client_factory, ) -> None: """LOAD-BEARING (3): the sanctioned door is untouched — the seed verdict's realization signal still reaches the generation prompt through the Step-1 ExpeL fold. This is what separates a gate from a wall. A refusal that also closed the fold would look identical on arms (1), (2) and (4) and would have removed the loop's whole learning path.""" store = seed_store_from_bundle(str(BUNDLE_DIR)) assert store.verdicts, "precondition: the bundle seeds exactly one verdict" factory, recorded = make_recording_client_factory(_VALID_REPLY) result = await run_project( _PID, "local", docs_dir=str(BUNDLE_DIR), bundle_dir=str(BUNDLE_DIR), verdict_input=_VERDICT_INPUT, store=store, client_factory=factory, ) assert isinstance(result.outcome, ValidatedProposal) gen_prompts = [p for p in recorded if "SavingsProposal" in p] assert gen_prompts, "the generation call must have happened" assert any(_REALIZATION_SIGNAL in p for p in gen_prompts), ( "the prior verdict no longer reaches the hypothesis prompt — the refusal closed the GATED " "route as well, which is a wall rather than a gate" ) # ------------------------------------------------------------------ (4) the refusal is traced async def test_a_refused_read_is_recorded_and_leaks_nothing(tmp_path) -> None: """LOAD-BEARING (4): the debate asks for the verdict by path; the read is refused, the CALL is in the trace with its path, and the verdict's body reaches no prompt. Both halves are needed and each has its own detach point. Without the recording an operator reading ``{run_id}-debate.json`` after a paid run cannot tell a run that tried the ungated door from one that never did — MEASURED: moving the recorder's append after ``call_next`` empties the trace for exactly a refused call, because the invocation never returns. Without the leak probe the gate could be recording refusals while the bytes left anyway.""" bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id outbox_dir = tmp_path / "outbox" sink: list[str] = [] result = await run_project( _PID, "local", docs_dir=bundle_dir, bundle_dir=bundle_dir, verdict_input=_VERDICT_INPUT, outbox_dir=str(outbox_dir), run_id="verdict-gate", client_factory=_recording_factory( sink, script={ "proposer": [ {"call": "read_file", "args": {"bundle_id": bundle_id, "path": _VERDICT_FILE}}, _VALID_REPLY, ], "checker": _CHECKER_REPLY, }, ), ) leaking = [i for i, prompt in enumerate(sink) if _VERDICT_SENTINEL in prompt] assert not leaking, ( f"the verdict body reached prompt(s) {leaking} — a guessed path still walks around the " "gated ExpeL fold" ) observed = [(c.name, c.bundle_id, c.path) for c in result.debate_tool_calls] assert ("read_file", bundle_id, _VERDICT_FILE) in observed, ( f"the refused read left no trace: {observed} — a refusal nobody can see is a refusal " "nobody can audit" ) payload = json.loads((outbox_dir / "verdict-gate-debate.json").read_text(encoding="utf-8")) identity = [{k: c[k] for k in ("name", "bundle_id", "path")} for c in payload["tool_calls"]] assert {"name": "read_file", "bundle_id": bundle_id, "path": _VERDICT_FILE} in identity, ( f"the artefact does not carry the refused call: {payload['tool_calls']}" ) # ------------------------------------------------------ (5) the document, not the walk, decides async def test_an_unlinked_verdict_is_refused_too(tmp_path) -> None: """LOAD-BEARING (5): a verdict file navigation never reached is refused as well. "However the path was found" is the order's own wording and the reason the rule reads the RESOLVED DOCUMENT's frontmatter rather than looking the path up among the navigated files. An implementation gated on ``Bundle.verdicts`` passes arms (1), (2) and (4) and still serves this one — the defect wearing the fix's clothes. Detach point: gate on the walk instead of the document → RED here and nowhere else.""" bundle_dir = _base_with_an_unlinked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id assert not any(f.name == "verdict-orphan.md" for f in okf.navigate_bundle(bundle_dir).files), ( "precondition: the orphan must be OUTSIDE the walk, or this arm proves nothing" ) answer = await _invoke( _tools(bundle_dir)["read_file"], bundle_id=bundle_id, path="verdict-orphan.md" ) assert answer.startswith(f"REFUSED ({VerdictLayerRefused.__name__})") assert _VERDICT_SENTINEL not in answer # ------------------------------------------------------------------------ (6) ladder intact async def test_the_index_still_reads_whole(tmp_path) -> None: """The ladder's top rung is untouched: ``index.md`` is navigation, never a judgement, and ``read_file(id, 'index.md')`` is the disclosure level ``list_bundles``' excerpt points at. Detach point: gate on the complement of ``context_files`` (which drops index files too) → RED.""" bundle_dir = _base_with_a_marked_verdict(tmp_path) bundle_id = okf.reconcile_bundle_id(bundle_dir).id whole = await _invoke(_tools(bundle_dir)["read_file"], bundle_id=bundle_id, path="index.md") assert "progressiv disclosure" in whole.lower()