"""S5.1 HITL — load-bearing detach seams (målbilde §3 / §7). Each test goes RED when the guarded mechanism is removed; a bare happy-path pass would not catch the regression. Seams pinned here: - the outbox↔inbox **id-join** (a landed verdict must clear the queue); - **causality control** (a pre-judged proposal is never listed); - **inbox-predicate parity** with ``load_verdicts_from_dir`` (wrong decision + malformed features are skipped, so ``pending`` never counts as judged a file the real loader would drop); - (Step 3) **route classification** detach; - (Step 5) **transitive import-graph** MAF-freedom probe. """ from __future__ import annotations import json from pathlib import Path from portfolio_optimiser.ir import AffectedItem, SavingsProposal from portfolio_optimiser.outbox import write_outbox from portfolio_optimiser.provenance import Citation, ProvenanceStamp from portfolio_optimiser.retrieval import TextSpan from portfolio_optimiser.validator import ValidatedProposal from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, write_verdict from portfolio_optimiser import hitl _PROVENANCE = ProvenanceStamp( citations=[Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="hi")], model="synthetic", role="proposer", validator_decision="validated", token_usage=8, ) def _write_proposal(outbox: Path, run_id: str, *, verdict_id: str, codes: list[str] | None = None) -> None: items = [AffectedItem(code=c, quantity=1000.0, unit_cost=1.0) for c in (codes or ["05.2"])] proposal = SavingsProposal( project_id="P1", measure="LED-retrofit", affected_items=items, claimed_saving_nok=200.0 ) write_outbox( str(outbox), run_id, outcome=ValidatedProposal( proposal=proposal, p10=100.0, p50=150.0, p90=200.0, nominal_feasible=180.0 ), provenance=_PROVENANCE, checker_verdict="approve", verdict_id=verdict_id, ) def _write_raw_verdict(inbox: Path, verdict_id: str, payload: dict) -> None: """Hand-write an inbox ``{id}.json`` — used to construct decisions/shapes ``write_verdict`` cannot emit (e.g. ``approved_with_adjustment``, or malformed features).""" inbox.mkdir(parents=True, exist_ok=True) (inbox / f"{verdict_id}.json").write_text(json.dumps(payload), encoding="utf-8") # --- id-join (the flagship seam) ------------------------------------------------------------------ def test_verdict_landing_removes_from_pending(tmp_path: Path) -> None: """LOAD-BEARING: a proposal is pending until a verdict with the EXACT SAME id lands in the inbox. The id is the shared literal passed to ``write_outbox(verdict_id=X)`` and ``Verdict(id=X)`` — not a re-hash of features. Detach the ``verdict_id ∉ inbox id-set`` predicate (return all outbox proposals) → the landed verdict never clears the queue → RED.""" outbox = tmp_path / "outbox" inbox = tmp_path / "inbox" _write_proposal(outbox, "run-1", verdict_id="SHARED-VID") assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"] write_verdict( str(inbox), Verdict( id="SHARED-VID", proposal_features=ProposalFeatures( affected_codes=frozenset({"05.2"}), measure_type="scope_reduction", claimed_saving_nok=200.0, ), decision="approved", rationale="expert reviewed (test)", ), ) assert hitl.pending(str(outbox), str(inbox)) == [] def test_pending_ignores_prejudged_proposal(tmp_path: Path) -> None: """CAUSALITY CONTROL: a proposal whose verdict is ALREADY in the inbox at first read is never listed — proving the removal above is caused by the id-join, not incidental.""" outbox = tmp_path / "outbox" inbox = tmp_path / "inbox" _write_proposal(outbox, "run-1", verdict_id="SHARED-VID") write_verdict( str(inbox), Verdict( id="SHARED-VID", proposal_features=ProposalFeatures( affected_codes=frozenset({"05.2"}), measure_type="scope_reduction", claimed_saving_nok=200.0, ), decision="approved", rationale="expert reviewed (test)", ), ) assert hitl.pending(str(outbox), str(inbox)) == [] # --- inbox-predicate parity with load_verdicts_from_dir ------------------------------------------- def test_inbox_idset_skips_wrong_decision(tmp_path: Path) -> None: """An otherwise-fully-valid inbox verdict (all required keys, well-formed proposal_features) whose ONLY defect is ``decision="approved_with_adjustment"`` (outside the binary run-path vocabulary) does NOT clear pending — parity with ``load_verdicts_from_dir`` skipping it. Being otherwise valid, dropping the decision filter is the ONLY reason it would skip → genuinely load-bearing.""" outbox = tmp_path / "outbox" inbox = tmp_path / "inbox" _write_proposal(outbox, "run-1", verdict_id="SHARED-VID") _write_raw_verdict( inbox, "SHARED-VID", { "id": "SHARED-VID", "decision": "approved_with_adjustment", "rationale": "adjusted", "proposal_features": { "affected_codes": ["05.2"], "measure_type": "scope_reduction", "claimed_saving_nok": 200.0, "description": "", }, }, ) assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"] # --- route classification (Step 3) ---------------------------------------------------------------- def test_route_classification_is_load_bearing(tmp_path: Path) -> None: """LOAD-BEARING: ``_matches`` genuinely classifies. Detach it (route everything to ``entries[0]``) → both assertions flip RED: an unmatched proposal would stop being UNROUTABLE, and a two-entry match would tie-break to ``entries[0]`` instead of the sorted-first id. ``entries[0]`` is deliberately NOT the sorted-first id, so the tie-break assertion is sensitive to the detach.""" outbox = tmp_path / "outbox" inbox = tmp_path / "inbox" config = hitl.RoutingConfig( entries=[ hitl.RoutingEntry(id="z-vei", allowed_code_prefixes=frozenset({"07"}), expert="Zeta"), hitl.RoutingEntry(id="a-energi", allowed_code_prefixes=frozenset({"05"}), expert="Alpha"), ] ) _write_proposal(outbox, "run-A", verdict_id="vA", codes=["99.9"]) # matches nothing _write_proposal(outbox, "run-C", verdict_id="vC", codes=["05.1", "07.1"]) # matches BOTH routed = {r.pending.run_id: r for r in hitl.route(str(outbox), str(inbox), config)} assert routed["run-A"].expert is None # detach → routed to Zeta → RED assert routed["run-C"].ambiguous is True assert routed["run-C"].dimension_id == "a-energi" # sorted-first, NOT entries[0] z-vei assert routed["run-C"].expert == "Alpha" def test_inbox_idset_skips_malformed_features(tmp_path: Path) -> None: """An inbox verdict with all four top-level keys but ``proposal_features`` MISSING ``measure_type`` (which ``verdict_from_dict`` reads → would raise) is skipped, so it does NOT clear pending — parity with the real loader dropping it.""" outbox = tmp_path / "outbox" inbox = tmp_path / "inbox" _write_proposal(outbox, "run-1", verdict_id="SHARED-VID") _write_raw_verdict( inbox, "SHARED-VID", { "id": "SHARED-VID", "decision": "approved", "rationale": "ok", "proposal_features": {"affected_codes": ["05.2"], "claimed_saving_nok": 200.0}, }, ) assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"]