portfolio-optimiser/tests/test_parse_failure_capture_loadbearing.py

"""Load-bearing: the model's RAW reply must survive a parse failure (Fase 1b, funn 1).

The gap, measured on the project's first live run (``docs/2026-08-14-fase1b-forste-levende-kjoring.md``):
``generate._fetch_parsed`` caught ``(ValidationError, ValueError, TypeError)`` and ``continue``-d.
The text the model actually produced — the thing that was PAID for and the only evidence of WHY it
did not parse — was dropped on the floor. The live run burned all twelve rounds on parse failures
and died with ``BudgetExceeded``, and no artefact anywhere held a single character of what the model
said. Every further paid run would have been a guess.

**WHERE the text surfaces is decided by a measurement, not by symmetry with Step 5.**
``BudgetExceeded`` is raised by ``meter.tick_round()`` INSIDE ``_fetch_parsed``, so it propagates out
of ``generate_via_llm`` before any value is returned, and (without a mandate) nothing catches it —
``run.py``'s only ``except BudgetExceeded`` is the mandate arm. So:

* a typed RETURN VALUE (``GenerationResult.parse_failures``, mirroring Step 5's ``refinements``) is
  blind to exactly the one path the capture exists for: the returning path is the path where the run
  already succeeded in parsing something;
* an outbox artefact written AFTER the run (``write_outbox``, run.py step 7) is blind for the same
  reason — ``run_project`` never reaches it.

The seam is therefore a CALLER-OWNED SINK plus an artefact written in a ``finally``:
``run_project`` owns the ``list[ParseFailure]``, hands it to ``generate_via_llm`` beside ``meter``,
and writes ``{run_id}-parse-failures.json`` however the evaluation ended. The sink mirrors ``meter``
exactly — a caller-owned accumulator the loop mutates, whose contents the caller still holds after an
exception. Step 5's "a returned value cannot be silently lost by a caller that forgets to pass a
collector" governs a value that REACHES the caller; here it does not, and copying that rule blindly
would have rebuilt the defect one layer up.

Five tests, load-bearing as a set. Each detach point is RED on its own:

* T1 — the budget-exhausted run (the live 1b shape) still writes the artefact
  (RED when the capture is detached, when the write is not in a ``finally``, or when run.py does not
  hand the sink over);
* T2 — the sink receives the reply VERBATIM at the ``generate_via_llm`` level
  (RED when the capture is detached OR when the text is truncated/summarised on the way in);
* T3 — the CONTROL: a run whose replies all parse writes NO artefact
  (RED on a "write it always" implementation, which would make the artefact's presence meaningless);
* T4 — a run that COMPLETES writes the artefact too
  (RED on an implementation that only writes from the exception arm — the mirror of T1);
* T5 — the marker gate: the sentinel is ABSENT from the bundle, so a positive assert on it cannot be
  satisfied by bundle context leaking into a prompt (repo rule for bundle-driven tests).

They drive the CANONICAL ``ScriptedChatClient`` through its ``reply_selector`` seam (S2.5
consolidation) and key on the generation prompt's own instruction line, so the debate's replies —
which are never parsed — cannot be mistaken for a capture.
"""

from __future__ import annotations

import json
from collections.abc import Callable
from pathlib import Path

import pytest

from portfolio_optimiser.budget import Budget, BudgetExceeded, TokenMeter
from portfolio_optimiser.generate import ParseFailure, generate_via_llm
from portfolio_optimiser.reference_domain import load_reference_projects
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.verdicts import VerdictStore

BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
_PROJECT_ID = "BYGG-KONTOR-NORD"

#: The instruction line ``generate._build_messages`` puts in EVERY generation prompt and nowhere
#: else — the one identifier that separates a generation call from a debate turn.
_GENERATION_MARK = "Respond with ONLY a JSON object"

#: A reply shaped like what a chatty model actually returns: prose, a fenced block, newlines and
#: non-ASCII — NOT JSON. The hex sentinel is what the positive asserts key on; it is gated absent
#: from the bundle by T5, so bundle context cannot satisfy them. Deliberately >200 characters, so a
#: truncating capture (``text[:80]``) is measurable rather than merely suspected.
_MALFORMED = (
    "Certainly! Here is my analysis of the project.\n\n"
    "PARSE-FAIL-SENTINEL-7f3a9c: the måling below is prose, not the requested object.\n\n"
    "```\n"
    "measure: Behovsstyrt belysning\n"
    "claimed_saving_nok: about 30 000 NOK (approx.)\n"
    "```\n\n"
    "Let me know if you would like me to format this as JSON instead!"
)
_SENTINEL = "PARSE-FAIL-SENTINEL-7f3a9c"

#: BYGG-KONTOR-NORD: affected total 300000 x 1.0 -> degenerate Monte Carlo P90 = 90000, so a claim
#: of 30000 validates (same arithmetic as test_a5_per_approach_artifacts_loadbearing).
_VALID_REPLY = (
    '{"measure":"Behovsstyrt belysning i fellesarealer","affected_items":'
    '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],'
    '"claimed_saving_nok":30000}'
)


def _factory(
    select: Callable[[str, str], str],
) -> Callable[[str], ScriptedChatClient]:
    def factory(role: str) -> ScriptedChatClient:
        return ScriptedChatClient(role=role, reply_selector=select, default_reply="ok")

    return factory


def _always_malformed(blob: str, _role: str) -> str:
    return _MALFORMED if _GENERATION_MARK in blob else "ok"


def _malformed_then_valid(failures: int) -> Callable[[str, str], str]:
    """Fail to parse ``failures`` times, then answer with a proposal that validates. The counter is
    keyed on the GENERATION prompt only, so debate turns do not consume it."""
    seen = {"n": 0}

    def _select(blob: str, _role: str) -> str:
        if _GENERATION_MARK not in blob:
            return "ok"
        seen["n"] += 1
        return _MALFORMED if seen["n"] <= failures else _VALID_REPLY

    return _select


async def _run(
    select: Callable[[str, str], str],
    outbox_dir: Path,
    *,
    run_id: str,
    max_rounds: int,
) -> RunResult:
    result = await run_project(
        _PROJECT_ID,
        "local",
        docs_dir=str(BUNDLE_DIR),
        bundle_dir=str(BUNDLE_DIR),
        verdict_input=_VERDICT_INPUT,
        store=VerdictStore(verdicts=[]),
        client_factory=_factory(select),
        outbox_dir=str(outbox_dir),
        run_id=run_id,
        # The ROUND cap is the bound under test; tokens are left effectively unbounded so a failure
        # here can only be the round ledger (the live 1b shape: "rounds limit=12 observed=13").
        meter=TokenMeter(Budget(max_tokens=10**9, max_rounds=max_rounds)),
    )
    assert isinstance(result, RunResult)
    return result


def _artefact(outbox_dir: Path, run_id: str) -> Path:
    return outbox_dir / f"{run_id}-parse-failures.json"


def _failures(outbox_dir: Path, run_id: str) -> list[dict[str, str]]:
    payload = json.loads(_artefact(outbox_dir, run_id).read_text(encoding="utf-8"))
    assert payload["run_id"] == run_id
    failures = payload["parse_failures"]
    assert isinstance(failures, list)
    return failures


# --------------------------------------------------------------------------------------------
# T1 — the live 1b shape: the run dies inside the generation loop, and the text survives anyway.
# --------------------------------------------------------------------------------------------


async def test_budget_exhausted_run_still_writes_the_raw_text(tmp_path: Path) -> None:
    """The measured 1b failure, reproduced offline: every reply is unparseable, the round ledger is
    exhausted inside ``_fetch_parsed``, and ``BudgetExceeded`` leaves ``run_project`` as an
    exception. The artefact must exist ANYWAY — this is the path a return value cannot reach.

    RED when the capture is detached, when the write is moved out of the ``finally``, or when
    ``run.py`` stops handing the sink to ``generate_via_llm``."""
    outbox_dir = tmp_path / "outbox"
    run_id = "run-parse-exhausted"

    with pytest.raises(BudgetExceeded) as excinfo:
        await _run(_always_malformed, outbox_dir, run_id=run_id, max_rounds=3)
    # Prove the event this test claims actually happened, and that it is the ROUND ledger — an
    # assert on the artefact alone could not tell an exhausted run from a completed one.
    assert excinfo.value.kind == "rounds"

    failures = _failures(outbox_dir, run_id)
    # max_rounds=3 -> ticks 1..3 each fetch a reply and fail to parse; tick 4 raises. So exactly
    # three replies were paid for, and exactly three must be recoverable.
    assert len(failures) == 3
    assert all(_SENTINEL in f["text"] for f in failures)
    assert all(f["error"] for f in failures), (
        "the parse error itself must be recorded, not just why"
    )


# --------------------------------------------------------------------------------------------
# T2 — the capture itself, at the generate level: VERBATIM, not summarised.
# --------------------------------------------------------------------------------------------


async def test_the_sink_receives_the_reply_verbatim(tmp_path: Path) -> None:
    """``generate_via_llm`` appends to the caller's sink at the moment the parse fails, and the text
    is the model's reply BYTE-FOR-BYTE. A capture that truncates, strips or summarises would leave
    the operator reading a paraphrase of the evidence.

    RED when the capture is detached, and RED when the text is shortened on the way in."""
    # A reference project rather than a hand-built one: this test never reaches the validator (no
    # reply ever parses), so the project only has to be a real one the prompt can name.
    project = load_reference_projects()[0]
    client = ScriptedChatClient(reply_selector=_always_malformed, role="proposer")
    sink: list[ParseFailure] = []

    with pytest.raises(BudgetExceeded):
        await generate_via_llm(
            client,
            project,
            "",
            TokenMeter(Budget(max_tokens=10**9, max_rounds=2)),
            max_attempts=3,
            parse_failures=sink,
        )

    assert len(sink) == 2
    assert [f.text for f in sink] == [_MALFORMED, _MALFORMED], "the reply must arrive verbatim"


# --------------------------------------------------------------------------------------------
# T3 — the CONTROL: no parse failure, no artefact.
# --------------------------------------------------------------------------------------------


async def test_a_run_whose_replies_parse_writes_no_artefact(tmp_path: Path) -> None:
    """A clean run must leave the outbox exactly as it was before this seam existed. Without this
    control an implementation that writes the file unconditionally passes T1 and T4, and the
    artefact's PRESENCE would then say nothing about whether anything failed to parse.

    RED when the ``if parse_failures`` guard is dropped."""
    outbox_dir = tmp_path / "outbox"
    run_id = "run-parse-clean"

    result = await _run(_malformed_then_valid(0), outbox_dir, run_id=run_id, max_rounds=8)
    assert result.outcome is not None

    assert not _artefact(outbox_dir, run_id).exists()
    # The pre-existing artefacts are untouched — the addition is inert on the clean path.
    # ``-debate.json`` (S2c) is written on EVERY bundle-path run that has an outbox,
    # unlike this artefact whose PRESENCE is its signal: a debate that opened nothing is
    # the S2c regression itself, so it must be readable rather than inferred from a
    # file that is not there.
    written = sorted(p.name for p in outbox_dir.iterdir())
    assert written == [
        f"{run_id}-debate.json",
        f"{run_id}-outcome.json",
        f"{run_id}-proposal.json",
        f"{run_id}-runconfig.json",
    ]


# --------------------------------------------------------------------------------------------
# T4 — the mirror of T1: a run that COMPLETES writes the artefact too.
# --------------------------------------------------------------------------------------------


async def test_completed_run_writes_the_artefact_too(tmp_path: Path) -> None:
    """One unparseable reply, then a proposal that validates: the run completes normally and the
    discarded first reply is still recoverable. Without this, an implementation that writes only
    from the exception arm would pass T1 and silently lose every parse failure on runs that
    eventually succeeded — the common case once prompting improves.

    RED when the write happens only on the exception path."""
    outbox_dir = tmp_path / "outbox"
    run_id = "run-parse-recovered"

    result = await _run(_malformed_then_valid(1), outbox_dir, run_id=run_id, max_rounds=8)
    assert result.outcome is not None

    failures = _failures(outbox_dir, run_id)
    assert len(failures) == 1
    assert _SENTINEL in failures[0]["text"]


# --------------------------------------------------------------------------------------------
# T5 — the marker gate.
# --------------------------------------------------------------------------------------------


def test_the_sentinel_is_absent_from_the_bundle() -> None:
    """Repo rule for bundle-driven tests: a marker asserted as PRESENT must be absent from the
    bundle, or bundle context reaching the prompt could satisfy the assert on its own."""
    hits = [
        path.name
        for path in BUNDLE_DIR.rglob("*")
        if path.is_file() and _SENTINEL in path.read_text(encoding="utf-8", errors="ignore")
    ]
    assert hits == []