"""Step-1 experience fold (ExpeL) — LOAD-BEARING (method-spec §3 Step 1, §11). The seam this file keeps alive: a prior verdict reaches the next hypothesis prompt ONLY via seed → store → structural retrieval → fold. The realization marker test is RED when the fold is detached; the empty-store control proves the fold is what changes the outcome signal. """ from __future__ import annotations import json from pathlib import Path from typing import Any import pytest from portfolio_optimiser_claude.experience import ( CandidateFeatures, VerdictRecord, VerdictStore, fold_experience, mint_verdict_id, seed_store_from_bundle, similarity, ) from portfolio_optimiser_claude.ir import load_validator_input BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" BASE_PROMPT = "Propose exactly one cost-saving measure for this project." def _features( codes: frozenset[str] = frozenset({"A", "B"}), measure_type: str = "M", claimed: float = 30_000.0, ) -> CandidateFeatures: return CandidateFeatures( affected_codes=codes, measure_type=measure_type, claimed_saving_nok=claimed ) def _record( verdict_id: str, features: CandidateFeatures, decision: str = "approved", rationale: str = "fine", ) -> VerdictRecord: return VerdictRecord( verdict_id=verdict_id, decision=decision, rationale=rationale, features=features ) @pytest.fixture(scope="module") def golden_surface() -> dict[str, Any]: raw: dict[str, Any] = json.loads((BUNDLE / "golden.json").read_text(encoding="utf-8")) surface: dict[str, Any] = raw["learning_surface"] return surface @pytest.fixture() def seeded_store() -> VerdictStore: store = VerdictStore() seed_store_from_bundle(store, BUNDLE) return store class TestSeedFromBundle: """§3 Step 1: every navigable ``type: verdict`` file becomes a store entry.""" def test_shared_bundle_seeds_one_verdict(self, seeded_store: VerdictStore) -> None: assert len(seeded_store) == 1 def test_seed_carries_decision_and_learning_fields( self, seeded_store: VerdictStore, golden_surface: dict[str, Any] ) -> None: # The rationale is the carrier of the learning signal: description plus # the structured learning fields, anchored to the golden learning surface. features = CandidateFeatures.from_proposal(load_validator_input(BUNDLE)) (record,) = seeded_store.retrieve(features, k=1) assert record.decision == "approved_with_adjustment" marker = ( f"[realiseringsgrad={golden_surface['realization_rate']}; " f"forventet_faktisk_NOK={golden_surface['expected_actual_saving_nok']}]" ) assert marker in record.rationale assert "ekspert-dom" in record.rationale # the description prose survives def test_seed_decision_defaults_to_approved(self, tmp_path: Path) -> None: _make_micro_bundle( tmp_path, index_links=["v.md"], verdict="---\ntype: verdict\ntitle: V\n---\nBody." ) store = VerdictStore() seed_store_from_bundle(store, tmp_path) features = CandidateFeatures.from_proposal(load_validator_input(tmp_path)) (record,) = store.retrieve(features, k=1) assert record.decision == "approved" def test_unlinked_verdict_file_is_unreachable(self, tmp_path: Path) -> None: # §6: navigation follows only index cross-links — an unlinked verdict # file must not seed the store. _make_micro_bundle( tmp_path, index_links=[], verdict="---\ntype: verdict\ntitle: V\n---\nBody." ) store = VerdictStore() seed_store_from_bundle(store, tmp_path) assert len(store) == 0 class TestStructuralRanking: """§3 Step 1: ranking is structural, never textual — frozen weights.""" def test_identical_features_score_one(self) -> None: assert similarity(_features(), _features()) == pytest.approx(1.0) def test_weights_decompose_per_spec(self) -> None: base = _features() # Disjoint code sets, same type, same magnitude bucket → 0.25 + 0.15. disjoint = _features(codes=frozenset({"X"})) assert similarity(base, disjoint) == pytest.approx(0.40) # Same codes, different type, same bucket → 0.60 + 0.15. other_type = _features(measure_type="OTHER") assert similarity(base, other_type) == pytest.approx(0.75) # Same codes, same type, different magnitude bucket → 0.60 + 0.25. other_bucket = _features(claimed=150_000.0) assert similarity(base, other_bucket) == pytest.approx(0.85) def test_jaccard_of_two_empty_sets_is_one(self) -> None: a = _features(codes=frozenset()) b = _features(codes=frozenset()) assert similarity(a, b) == pytest.approx(1.0) def test_magnitude_buckets_are_the_frozen_edges(self) -> None: # [0, 1e5), [1e5, 5e5), [5e5, 1e6), [1e6, ∞) — boundary values fall right. for claimed, other, same in ( (99_999.0, 5_000.0, True), (99_999.0, 100_000.0, False), (100_000.0, 499_999.0, True), (500_000.0, 999_999.0, True), (999_999.0, 1_000_000.0, False), (1_000_000.0, 9e9, True), ): got = similarity(_features(claimed=claimed), _features(claimed=other)) assert got == pytest.approx(1.0 if same else 0.85), (claimed, other) def test_surface_text_never_contributes(self) -> None: # Two records differing ONLY in rationale text rank identically — # deterministic tie-break is by verdict id ascending. store = VerdictStore() store.add(_record("bbb", _features(), rationale="LED retrofit — perfect match text")) store.add(_record("aaa", _features(), rationale="unrelated prose")) ranked = store.retrieve(_features(), k=2) assert [r.verdict_id for r in ranked] == ["aaa", "bbb"] def test_top_k_by_similarity(self) -> None: store = VerdictStore() store.add(_record("far", _features(codes=frozenset({"X"}), measure_type="OTHER"))) store.add(_record("near", _features())) assert [r.verdict_id for r in store.retrieve(_features(), k=1)] == ["near"] def test_k_must_be_positive(self) -> None: store = VerdictStore() with pytest.raises(ValueError): store.retrieve(_features(), k=0) class TestStoreSemantics: """§4.2: the in-memory store is FIRST-write-wins per id (idempotent merges).""" def test_first_write_wins_per_id(self) -> None: store = VerdictStore() store.add(_record("same-id", _features(), rationale="first")) store.add(_record("same-id", _features(), rationale="second")) assert len(store) == 1 (record,) = store.retrieve(_features(), k=1) assert record.rationale == "first" class TestIdMinting: """§4.2: the id keys on the candidate measure, not the verdict event.""" def test_structurally_identical_candidates_share_an_id(self) -> None: a = _features(codes=frozenset({"B", "A"})) b = _features(codes=frozenset({"A", "B"})) minted = mint_verdict_id(a) assert minted == mint_verdict_id(b) assert len(minted) == 16 assert set(minted) <= set("0123456789abcdef") def test_number_formatting_participates_in_the_hash(self) -> None: # 30000 vs 30000.0 differ — the reason loaded ids are kept verbatim. as_int = _features(claimed=30_000) as_float = _features(claimed=30_000.0) assert mint_verdict_id(as_int) != mint_verdict_id(as_float) class TestExperienceFold: """LOAD-BEARING (§11): the realization marker reaches the prompt via the fold.""" def test_realization_marker_reaches_the_hypothesis_prompt( self, seeded_store: VerdictStore, golden_surface: dict[str, Any] ) -> None: # Seed → store → structural retrieval → fold: the expert's realization # correction must reach the next hypothesis prompt. RED when the fold is # detached (nothing prepended) or seeding drops the learning fields. features = CandidateFeatures.from_proposal(load_validator_input(BUNDLE)) prompt = fold_experience(seeded_store, features, BASE_PROMPT, k=3) assert f"realiseringsgrad={golden_surface['realization_rate']}" in prompt assert prompt.endswith(BASE_PROMPT) # prepended, never replacing the task def test_fold_lines_carry_id_decision_and_rationale(self, seeded_store: VerdictStore) -> None: features = CandidateFeatures.from_proposal(load_validator_input(BUNDLE)) (record,) = seeded_store.retrieve(features, k=1) prompt = fold_experience(seeded_store, features, BASE_PROMPT, k=1) for part in (record.verdict_id, record.decision, record.rationale): assert part in prompt def test_empty_store_control_changes_the_outcome_signal( self, seeded_store: VerdictStore, golden_surface: dict[str, Any] ) -> None: # Control (§11): with an empty store the prompt is the unchanged base — # no marker, no verdict lines. The fold is what makes the difference. features = CandidateFeatures.from_proposal(load_validator_input(BUNDLE)) unfolded = fold_experience(VerdictStore(), features, BASE_PROMPT, k=3) folded = fold_experience(seeded_store, features, BASE_PROMPT, k=3) assert unfolded == BASE_PROMPT assert f"realiseringsgrad={golden_surface['realization_rate']}" not in unfolded assert folded != unfolded def _make_micro_bundle(tmp_path: Path, index_links: list[str], verdict: str) -> None: links = " ".join(f"[{name}]({name})" for name in index_links) (tmp_path / "index.md").write_text( f"---\ntype: index\ntitle: Micro\n---\nSummary. {links}\n", encoding="utf-8" ) (tmp_path / "v.md").write_text(verdict, encoding="utf-8") (tmp_path / "validator-input.json").write_text( json.dumps( { "project_id": "P1", "measure": "M", "affected_items": [{"code": "A", "quantity": 10, "unit_cost": 100.0}], "claimed_saving_nok": 300, "assumptions": {}, } ), encoding="utf-8", )