portfolio-optimiser/tests/test_verdict_layer_refusal_loadbearing.py

"""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"))
    assert {"name": "read_file", "bundle_id": bundle_id, "path": _VERDICT_FILE} in payload[
        "tool_calls"
    ], 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()