portfolio-optimiser-claude/tests/test_step7_async_loop_loadbearing.py

"""Step-7 async file loop — LOAD-BEARING (method-spec §3 Step 7, §5, §11).

The seam this file keeps alive: a verdict dropped into the inbox FOLDER after
Run A reaches Run B's hypothesis prompt via a FRESH store (merge → fold) — the
system is fully resumable across runs separated in time, with no live-session
assumption. The empty-inbox control proves the inbox is what changes the
outcome signal. Role split (§5): the system READS the inbox; the expert writes
it — the merge path never persists anything back.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest
from _scripted import ScriptedClient, reply
from pydantic import ValidationError

from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.contracts import TerminationContract
from portfolio_optimiser_claude.experience import (
    CandidateFeatures,
    VerdictRecord,
    VerdictStore,
    fold_experience,
    mint_verdict_id,
)
from portfolio_optimiser_claude.inbox import (
    InboxLimitError,
    ProposalFeatures,
    VerdictDocument,
    load_inbox,
    merge_inbox_into_store,
    write_verdict,
)
from portfolio_optimiser_claude.loop import ModelReply, run_project

BASE_CONTEXT = "Project context: office building, LED retrofit candidate."
MARKER = "realiseringsgrad=0.79"
RATIONALE = f"Godkjent med korreksjon: i drift realiseres ~79% ({MARKER})."

BASE_PROPOSAL = {
    "project_id": "p1",
    "measure": "led-retrofit",
    "affected_items": [{"code": "E01", "quantity": 100, "unit_cost": 1000}],
    "claimed_saving_nok": 25000,
    "assumptions": {},
}
ADJUSTED_PROPOSAL = dict(BASE_PROPOSAL, claimed_saving_nok=19750)


def _meter() -> BudgetMeter:
    return BudgetMeter(TerminationContract(max_rounds=50, max_tokens=100_000))


def _features(
    codes: frozenset[str] = frozenset({"E01"}),
    measure_type: str = "led-retrofit",
    claimed: float = 25000.0,
) -> CandidateFeatures:
    return CandidateFeatures(
        affected_codes=codes, measure_type=measure_type, claimed_saving_nok=claimed
    )


def _document(
    features: CandidateFeatures | None = None,
    decision: str = "approved",
    rationale: str = RATIONALE,
) -> VerdictDocument:
    return VerdictDocument.from_candidate(
        features or _features(),
        decision=decision,
        rationale=rationale,
        description="LED retrofit for one office building",
    )


class TestVerdictDocument:
    """§4.2: the verdict file model — minted on authoring, VERBATIM on load."""

    def test_from_candidate_mints_the_spec_id(self) -> None:
        features = _features()
        assert _document(features).id == mint_verdict_id(features)

    def test_affected_codes_are_emitted_sorted(self) -> None:
        doc = _document(_features(codes=frozenset({"B", "A", "C"})))
        assert doc.proposal_features.affected_codes == ["A", "B", "C"]

    def test_to_record_keeps_a_loaded_id_verbatim(self) -> None:
        # A loaded id is NEVER re-minted (raw number formatting could diverge).
        doc = VerdictDocument(
            id="feedfeedfeedfeed",
            decision="approved",
            rationale=RATIONALE,
            proposal_features=ProposalFeatures(
                affected_codes=["E01"],
                measure_type="led-retrofit",
                claimed_saving_nok=25000.0,
                description="surface text",
            ),
        )
        record = doc.to_record()
        assert isinstance(record, VerdictRecord)
        assert record.verdict_id == "feedfeedfeedfeed"
        assert record.verdict_id != mint_verdict_id(doc.features())

    def test_description_is_excluded_from_minting(self) -> None:
        a = _document().model_copy(deep=True)
        b = VerdictDocument.from_candidate(
            _features(), decision="approved", rationale=RATIONALE, description="other text"
        )
        assert a.id == b.id


class TestAuthoringPrimitive:
    """§5: one verdict per file, ``{id}.json``, written deterministically."""

    def test_write_creates_the_directory_and_names_by_id(self, tmp_path: Path) -> None:
        inbox = tmp_path / "nested" / "inbox"
        doc = _document()
        path = write_verdict(inbox, doc)
        assert path == inbox / f"{doc.id}.json"
        assert path.is_file()

    def test_write_is_deterministic_sorted_keys_two_space_indent(self, tmp_path: Path) -> None:
        doc = _document()
        path = write_verdict(tmp_path, doc)
        text = path.read_text(encoding="utf-8")
        assert text == json.dumps(json.loads(text), sort_keys=True, indent=2)
        assert write_verdict(tmp_path, doc).read_text(encoding="utf-8") == text

    def test_disk_layer_is_last_write_wins_per_file(self, tmp_path: Path) -> None:
        write_verdict(tmp_path, _document(rationale="first"))
        write_verdict(tmp_path, _document(rationale="second"))
        assert len(list(tmp_path.iterdir())) == 1
        (loaded,) = load_inbox(tmp_path)
        assert loaded.rationale == "second"


class TestTolerantLoad:
    """§5: the raw layer is written out of band — skip, never raise."""

    def test_missing_folder_yields_zero_verdicts(self, tmp_path: Path) -> None:
        assert load_inbox(tmp_path / "no-such-inbox") == []

    def test_non_json_broken_and_incomplete_files_are_skipped(self, tmp_path: Path) -> None:
        write_verdict(tmp_path, _document())
        (tmp_path / "notes.txt").write_text("not a verdict", encoding="utf-8")
        (tmp_path / "broken.json").write_text("{not json", encoding="utf-8")
        incomplete = _document().model_dump()
        del incomplete["rationale"]
        (tmp_path / "incomplete.json").write_text(json.dumps(incomplete), encoding="utf-8")
        (tmp_path / "subdir.json").mkdir()
        loaded = load_inbox(tmp_path)
        assert len(loaded) == 1
        assert loaded[0].rationale == RATIONALE

    def test_files_are_processed_sorted_by_filename(self, tmp_path: Path) -> None:
        late = _document(_features(codes=frozenset({"Z9"})))
        early = _document(_features(codes=frozenset({"A1"})))
        for doc in (late, early):
            write_verdict(tmp_path, doc)
        assert [d.id for d in load_inbox(tmp_path)] == sorted([late.id, early.id])


class TestFileLayerVocabulary:
    """C2.5 (C-F7): the file layer polices the §4.2 decision set — SKIP, never raise."""

    def test_an_unknown_decision_never_reaches_the_store(self, tmp_path: Path) -> None:
        # Fixture = the review's RUN C-F7 proof: before C2.5 this string
        # entered the store and its rationale folded into the next prompt.
        doc = _document().model_copy(update={"decision": "hva-som-helst"})
        (tmp_path / f"{doc.id}.json").write_text(json.dumps(doc.model_dump()), encoding="utf-8")
        store = VerdictStore()
        assert merge_inbox_into_store(store, tmp_path) == 0
        assert len(store) == 0

    def test_the_full_file_layer_vocabulary_loads(self, tmp_path: Path) -> None:
        # Control: exactly the §4.2 set {approved, rejected,
        # approved_with_adjustment} passes the file layer.
        decisions = ("approved", "rejected", "approved_with_adjustment")
        for index, decision in enumerate(decisions):
            write_verdict(
                tmp_path, _document(_features(codes=frozenset({f"C{index}"})), decision=decision)
            )
        assert {d.decision for d in load_inbox(tmp_path)} == set(decisions)


class TestInboxCaps:
    """C2.5: configurable caps fail FAST with a precise error — never a silent cut."""

    def test_rationale_over_the_cap_fails_fast_with_a_precise_error(self, tmp_path: Path) -> None:
        doc = _document(rationale="x" * 201)
        write_verdict(tmp_path, doc)
        with pytest.raises(InboxLimitError) as err:
            load_inbox(tmp_path, max_rationale_chars=200)
        message = str(err.value)
        assert doc.id in message
        assert "201" in message
        assert "200" in message

    def test_rationale_at_the_cap_loads(self, tmp_path: Path) -> None:
        write_verdict(tmp_path, _document(rationale="x" * 200))
        assert len(load_inbox(tmp_path, max_rationale_chars=200)) == 1

    def test_more_files_than_the_cap_fails_fast(self, tmp_path: Path) -> None:
        for index in range(3):
            write_verdict(tmp_path, _document(_features(codes=frozenset({f"C{index}"}))))
        with pytest.raises(InboxLimitError) as err:
            load_inbox(tmp_path, max_files=2)
        message = str(err.value)
        assert "3" in message
        assert "2" in message

    def test_merge_threads_the_caps_through(self, tmp_path: Path) -> None:
        write_verdict(tmp_path, _document(rationale="x" * 201))
        with pytest.raises(InboxLimitError):
            merge_inbox_into_store(VerdictStore(), tmp_path, max_rationale_chars=200)


class TestIdGrammar:
    """C2.5 (R-6): the id grammar is policed BEFORE ``write_verdict`` touches disk."""

    def test_a_traversal_id_is_rejected_at_construction(self) -> None:
        # The security agent's RUN R-6 proof: '../../escaped' used to
        # construct fine and write outside the inbox.
        with pytest.raises(ValidationError):
            VerdictDocument(
                id="../../escaped",
                decision="approved",
                rationale=RATIONALE,
                proposal_features=ProposalFeatures(
                    affected_codes=["E01"],
                    measure_type="led-retrofit",
                    claimed_saving_nok=25000.0,
                    description="surface text",
                ),
            )

    def test_write_verdict_refuses_an_escaping_id_writing_nothing(self, tmp_path: Path) -> None:
        # model_copy(update=...) bypasses model validation — the write seam
        # must fail closed on its own (detach the grammar here → a file lands
        # OUTSIDE the inbox → red).
        inbox = tmp_path / "inbox"
        doc = _document().model_copy(update={"id": "../../escaped"})
        with pytest.raises(ValueError):
            write_verdict(inbox, doc)
        assert not (tmp_path / "escaped.json").exists()
        assert not inbox.exists()

    def test_a_loaded_file_with_an_escaping_id_is_skipped(self, tmp_path: Path) -> None:
        # Tolerant load stays tolerant: a hostile id is SKIPPED, never raised.
        doc = _document().model_copy(update={"id": "../../escaped"})
        (tmp_path / "escaped.json").write_text(json.dumps(doc.model_dump()), encoding="utf-8")
        assert load_inbox(tmp_path) == []


class TestMergeIntoStore:
    """§5: merge, never replace — first-write-wins per id, idempotent, read-only."""

    def test_merge_ingests_and_repeats_idempotently(self, tmp_path: Path) -> None:
        write_verdict(tmp_path, _document())
        store = VerdictStore()
        assert merge_inbox_into_store(store, tmp_path) == 1
        assert merge_inbox_into_store(store, tmp_path) == 1
        assert len(store) == 1

    def test_passed_in_store_verdicts_survive_the_merge(self, tmp_path: Path) -> None:
        # Cross-project threading: existing entries survive; on an id collision
        # the store's FIRST write wins over the inbox file.
        inbox_doc = _document(rationale="from the inbox")
        write_verdict(tmp_path, inbox_doc)
        store = VerdictStore()
        other = VerdictRecord(
            verdict_id="0000000000000000",
            decision="approved",
            rationale="other project",
            features=_features(codes=frozenset({"X"})),
        )
        colliding = VerdictRecord(
            verdict_id=inbox_doc.id,
            decision="approved",
            rationale="already threaded",
            features=_features(),
        )
        store.add(other)
        store.add(colliding)
        merge_inbox_into_store(store, tmp_path)
        assert len(store) == 2
        ranked = store.retrieve(_features(), k=2)
        assert {r.rationale for r in ranked} == {"already threaded", "other project"}

    def test_merge_never_writes_to_the_inbox(self, tmp_path: Path) -> None:
        # Role split (§3 Step 7, unwaivable): the system READS the inbox.
        write_verdict(tmp_path, _document())
        before = {p.name: p.read_bytes() for p in tmp_path.iterdir()}
        merge_inbox_into_store(VerdictStore(), tmp_path)
        after = {p.name: p.read_bytes() for p in tmp_path.iterdir()}
        assert after == before


def _run_a_client() -> ScriptedClient:
    return ScriptedClient(
        replies=[
            reply("debate reasoning"),
            reply("VERDICT: APPROVE"),
            reply(json.dumps(BASE_PROPOSAL)),
        ]
    )


def _run_b_script(role: str, prompt: str) -> ModelReply:
    # Prompt-SENSITIVE (flip proof): the adjusted candidate appears ONLY when the
    # expert's realization marker reached the prompt via merge → fold.
    if role == "checker":
        return reply("VERDICT: APPROVE")
    if "JSON object" in prompt:
        return reply(json.dumps(ADJUSTED_PROPOSAL if MARKER in prompt else BASE_PROPOSAL))
    return reply("debate reasoning")


def _run_b(inbox_dir: Path) -> tuple[ScriptedClient, float]:
    # Run B is a SEPARATE, later run: FRESH store, no state carried from Run A.
    store = VerdictStore()
    merge_inbox_into_store(store, inbox_dir)
    features = _features()
    context = fold_experience(store, features, BASE_CONTEXT, k=3)
    client = ScriptedClient(script=_run_b_script)
    result = run_project(client, context, meter=_meter(), max_debate_rounds=3)
    return client, result.proposal.claimed_saving_nok


class TestAsyncFileLoopLoadBearing:
    """LOAD-BEARING (§11): the dropped verdict reaches Run B's prompt and flips it."""

    def test_verdict_dropped_after_run_a_reaches_run_b_via_a_fresh_store(
        self, tmp_path: Path
    ) -> None:
        inbox = tmp_path / "inbox"
        run_a = run_project(_run_a_client(), BASE_CONTEXT, meter=_meter(), max_debate_rounds=3)
        assert run_a.proposal.claimed_saving_nok == 25000
        # Days later, the expert judges Run A's candidate and drops a verdict
        # file — the run itself never wrote it (role split).
        judged = CandidateFeatures.from_proposal(run_a.proposal)
        write_verdict(inbox, _document(judged))
        client_b, claimed_b = _run_b(inbox)
        assert any(MARKER in prompt for prompt in client_b.prompts("proposer"))
        assert claimed_b == 19750  # the flip: the marker informed Run B's candidate

    def test_empty_inbox_control_keeps_run_b_at_the_base_outcome(self, tmp_path: Path) -> None:
        # Control (§11): an empty inbox → no marker in any prompt, no flip.
        inbox = tmp_path / "inbox"
        inbox.mkdir()
        client_b, claimed_b = _run_b(inbox)
        # Positive control: prompts were actually issued. `all()` over an empty list is
        # vacuously True, so without this the control would pass for a run that never
        # prompted the proposer at all.
        assert client_b.prompts("proposer")
        assert all(MARKER not in prompt for prompt in client_b.prompts("proposer"))
        assert claimed_b == 25000

    def test_missing_inbox_folder_control_behaves_like_empty(self, tmp_path: Path) -> None:
        client_b, claimed_b = _run_b(tmp_path / "never-created")
        assert claimed_b == 25000