From 46f2f521f73541e14b89d71cddc0ce58f2f072ee Mon Sep 17 00:00:00 2001 From: Kjell Tore Guttormsen Date: Fri, 3 Jul 2026 06:48:13 +0200 Subject: [PATCH] =?UTF-8?q?feat(context):=20S7=20=E2=80=94=20D7=20context?= =?UTF-8?q?=20seam:=20OKF=20navigation=20+=20gated=20ExpeL=20fold?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 1 of the loop, built from method-spec §3 alone: - okf.py (pure stdlib): frontmatter parse, deterministic/tolerant/boundary-checked index navigation, bundle_context rendering with type:verdict exclusion - experience.py: CandidateFeatures from the IR projection, §4.2 id minting, structural ranking (0.60·Jaccard + 0.25·type + 0.15·magnitude bucket), first-write-wins store, bundle seeding with the realization marker, fold-before-generation (empty retrieval → base unchanged) Load-bearing (§11), each proved RED on detach: verdict-layer exclusion, fold, seed learning fields, agent-toolkit import guard. 76/76 without an API key; ruff + mypy --strict clean. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QdSfQdND84oeq2mbjueLTS --- src/portfolio_optimiser_claude/experience.py | 168 +++++++++++++ src/portfolio_optimiser_claude/okf.py | 112 +++++++++ tests/test_okf.py | 202 +++++++++++++++ tests/test_step1_expel_loadbearing.py | 248 +++++++++++++++++++ 4 files changed, 730 insertions(+) create mode 100644 src/portfolio_optimiser_claude/experience.py create mode 100644 src/portfolio_optimiser_claude/okf.py create mode 100644 tests/test_okf.py create mode 100644 tests/test_step1_expel_loadbearing.py diff --git a/src/portfolio_optimiser_claude/experience.py b/src/portfolio_optimiser_claude/experience.py new file mode 100644 index 0000000..8e7ad3f --- /dev/null +++ b/src/portfolio_optimiser_claude/experience.py @@ -0,0 +1,168 @@ +"""The experience seam (ExpeL-style): store, structural retrieval, fold (§3 Step 1). + +Prior verdicts reach the hypothesis prompt ONLY through this seam: bundle seeding → +in-memory store → structural retrieval → fold-before-generation. Ranking is +structural, never textual — surface text must not contribute to similarity. The +rationale is the carrier of the learning signal: the fold is what lets an expert's +realization-rate correction reach the next hypothesis. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass + +from pathlib import Path + +from .ir import SavingsProposal, load_validator_input +from .okf import navigate_bundle + +_VERDICT_TYPE = "verdict" +_DEFAULT_SEED_DECISION = "approved" +# §3 Step 1: frozen similarity weights and magnitude bucket edges. +_JACCARD_WEIGHT = 0.60 +_MEASURE_TYPE_WEIGHT = 0.25 +_MAGNITUDE_WEIGHT = 0.15 +_MAGNITUDE_BUCKET_EDGES = (1e5, 5e5, 1e6) + + +@dataclass(frozen=True) +class CandidateFeatures: + """The structural features retrieval ranks over (§4.2 ``proposal_features``).""" + + affected_codes: frozenset[str] + measure_type: str + claimed_saving_nok: float + + @classmethod + def from_proposal(cls, proposal: SavingsProposal) -> CandidateFeatures: + # The IR projection carries no separate measure-type field; its ``measure`` + # string is the candidate's measure type at this level (§3 Step 1: the + # query key is read from the IR projection, before any proposal exists). + return cls( + affected_codes=frozenset(item.code for item in proposal.affected_items), + measure_type=proposal.measure, + claimed_saving_nok=proposal.claimed_saving_nok, + ) + + +@dataclass(frozen=True) +class VerdictRecord: + """One store entry: id (the learning-loop key), decision, rationale, features.""" + + verdict_id: str + decision: str + rationale: str + features: CandidateFeatures + + +def mint_verdict_id(features: CandidateFeatures) -> str: + """First 16 hex chars of SHA-256 over the canonical feature JSON (§4.2). + + Raw JSON number formatting participates in the hash (30000 vs 30000.0 differ), + which is why a LOADED verdict's id is kept verbatim — never re-minted. + """ + canonical = json.dumps( + { + "affected_codes": sorted(features.affected_codes), + "claimed_saving_nok": features.claimed_saving_nok, + "measure_type": features.measure_type, + }, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] + + +def _magnitude_bucket(claimed_saving_nok: float) -> int: + # Buckets [0, 1e5), [1e5, 5e5), [5e5, 1e6), [1e6, ∞) over the claimed saving. + return sum(1 for edge in _MAGNITUDE_BUCKET_EDGES if claimed_saving_nok >= edge) + + +def similarity(a: CandidateFeatures, b: CandidateFeatures) -> float: + """Structural similarity — surface text never contributes (§3 Step 1).""" + if not a.affected_codes and not b.affected_codes: + jaccard = 1.0 + else: + union = a.affected_codes | b.affected_codes + jaccard = len(a.affected_codes & b.affected_codes) / len(union) + return ( + _JACCARD_WEIGHT * jaccard + + _MEASURE_TYPE_WEIGHT * (a.measure_type == b.measure_type) + + _MAGNITUDE_WEIGHT + * (_magnitude_bucket(a.claimed_saving_nok) == _magnitude_bucket(b.claimed_saving_nok)) + ) + + +class VerdictStore: + """In-memory verdict store — FIRST-write-wins per id (idempotent merges, §4.2).""" + + def __init__(self) -> None: + self._records: dict[str, VerdictRecord] = {} + + def __len__(self) -> int: + return len(self._records) + + def add(self, record: VerdictRecord) -> None: + self._records.setdefault(record.verdict_id, record) + + def retrieve(self, features: CandidateFeatures, k: int) -> list[VerdictRecord]: + """Top-k by structural similarity, ties broken by verdict id ascending.""" + if k <= 0: + raise ValueError(f"retrieval k must be positive, got {k}") + ranked = sorted( + self._records.values(), + key=lambda record: (-similarity(record.features, features), record.verdict_id), + ) + return ranked[:k] + + +def seed_store_from_bundle(store: VerdictStore, bundle_dir: Path) -> int: + """Seed the store from the bundle's navigable ``type: verdict`` files (§3 Step 1). + + Entries are keyed on the bundle's candidate features, read from the IR + projection (fail-fast, required input). The rationale is built from the + ``description`` frontmatter plus, when present, the structured learning fields. + Returns the number of verdict files seeded. + """ + features = CandidateFeatures.from_proposal(load_validator_input(bundle_dir)) + seeded = 0 + for concept in navigate_bundle(bundle_dir): + if concept.type != _VERDICT_TYPE: + continue + rationale = concept.frontmatter.get("description", "") + realization_rate = concept.frontmatter.get("realization_rate") + expected_actual = concept.frontmatter.get("expected_actual_saving_nok") + if realization_rate is not None and expected_actual is not None: + learning = ( + f"[realiseringsgrad={realization_rate}; forventet_faktisk_NOK={expected_actual}]" + ) + rationale = f"{rationale} {learning}".strip() + store.add( + VerdictRecord( + verdict_id=mint_verdict_id(features), + decision=concept.frontmatter.get("decision", _DEFAULT_SEED_DECISION), + rationale=rationale, + features=features, + ) + ) + seeded += 1 + return seeded + + +def fold_experience( + store: VerdictStore, features: CandidateFeatures, base_context: str, k: int +) -> str: + """Prepend the retrieved prior verdicts to the generation context (§3 Step 1). + + One line per verdict — id, decision, rationale. An empty retrieval returns the + base context unchanged (the empty-store control, §11). + """ + retrieved = store.retrieve(features, k) + if not retrieved: + return base_context + lines = "\n".join( + f"- {record.verdict_id} [{record.decision}]: {record.rationale}" for record in retrieved + ) + return f"Prior expert verdicts (most similar first):\n{lines}\n\n{base_context}" diff --git a/src/portfolio_optimiser_claude/okf.py b/src/portfolio_optimiser_claude/okf.py new file mode 100644 index 0000000..284f7b7 --- /dev/null +++ b/src/portfolio_optimiser_claude/okf.py @@ -0,0 +1,112 @@ +"""OKF bundle navigation and read-context rendering (method-spec §3 Step 1). + +The read-context is built by NAVIGATING the bundle with progressive disclosure — +never by stuffing the whole bundle (or keyword-retrieved chunks) into the prompt. +Navigation starts at ``index.md`` and follows its intra-bundle cross-links; broken +or bundle-escaping links are tolerated (skipped, never raised — the OKF robustness +rule), while a missing ``index.md`` is an error (no entry point). ``type: verdict`` +files are EXCLUDED from rendering: prior verdicts reach the hypothesis prompt ONLY +via the gated experience fold (see ``experience``), never via context rendering. + +Pure stdlib by design — the context seam imports no agent toolkit (§11). +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + +_INDEX_FILENAME = "index.md" +_VERDICT_TYPE = "verdict" +# (reference) the intra-bundle cross-link pattern, per §3 Step 1. +_CROSSLINK_PATTERN = re.compile(r"\]\(([^)]+\.md)\)") + + +@dataclass(frozen=True) +class ConceptFile: + """One parsed OKF concept file: frontmatter (``type`` required) plus body.""" + + path: Path + frontmatter: dict[str, str] + body: str + + @property + def type(self) -> str: + return self.frontmatter["type"] + + @property + def title(self) -> str: + return self.frontmatter.get("title", self.path.stem) + + +def _strip_matching_quotes(value: str) -> str: + if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: + return value[1:-1] + return value + + +def parse_concept_file(path: Path) -> ConceptFile: + """Parse frontmatter (leading ``---`` block, line-oriented ``key: value``) + body. + + The single required field is ``type``; unknown fields are preserved as strings. + """ + lines = path.read_text(encoding="utf-8").splitlines() + if not lines or lines[0].strip() != "---": + raise ValueError(f"{path.name}: missing frontmatter block") + frontmatter: dict[str, str] = {} + body_start = len(lines) + for i, line in enumerate(lines[1:], start=1): + if line.strip() == "---": + body_start = i + 1 + break + key, sep, value = line.partition(":") + if sep: + frontmatter[key.strip()] = _strip_matching_quotes(value.strip()) + if "type" not in frontmatter: + raise ValueError(f"{path.name}: frontmatter lacks the required field 'type'") + return ConceptFile( + path=path, frontmatter=frontmatter, body="\n".join(lines[body_start:]).strip() + ) + + +def navigate_bundle(bundle_dir: Path) -> list[ConceptFile]: + """Navigate from ``index.md`` — deterministic order: index first, links first-seen. + + Targets containing a path separator are out-of-bundle and skipped; resolution is + boundary-checked against the bundle directory (fail-closed); broken links are + skipped, never raised. Repeated links are de-duplicated. + """ + index_path = bundle_dir / _INDEX_FILENAME + if not index_path.is_file(): + raise FileNotFoundError( + f"bundle has no entry point: missing {_INDEX_FILENAME} in {bundle_dir}" + ) + index = parse_concept_file(index_path) + bundle_root = bundle_dir.resolve() + concepts = [index] + seen = {_INDEX_FILENAME} + for target in _CROSSLINK_PATTERN.findall(index_path.read_text(encoding="utf-8")): + if "/" in target or "\\" in target or target in seen: + continue + seen.add(target) + resolved = (bundle_dir / target).resolve() + if not resolved.is_relative_to(bundle_root) or not resolved.is_file(): + continue + concepts.append(parse_concept_file(resolved)) + return concepts + + +def bundle_context(bundle_dir: Path) -> str: + """Render the read-context: index body, then ``## {type}: {title}`` sections. + + Empty sections are dropped. ``type: verdict`` files are excluded — the verdict + layer must never leak into the read-context (§3 Step 1, load-bearing §11). + """ + index, *concepts = navigate_bundle(bundle_dir) + sections = [index.body] if index.body else [] + for concept in concepts: + if concept.type == _VERDICT_TYPE or not concept.body: + continue + sections.append(f"## {concept.type}: {concept.title}\n\n{concept.body}") + return "\n\n".join(sections) diff --git a/tests/test_okf.py b/tests/test_okf.py new file mode 100644 index 0000000..6d5ac97 --- /dev/null +++ b/tests/test_okf.py @@ -0,0 +1,202 @@ +"""OKF navigation + read-context rendering (method-spec §3 Step 1, §11). + +Load-bearing seams proved here: + +- **Verdict-layer exclusion:** the realization signal must NEVER appear in the + rendered read-context — prior verdicts reach the prompt ONLY via the gated + experience fold. The test is RED if ``type: verdict`` files leak into rendering. +- **Context-seam purity (import guard):** the navigation/context module imports + no agent toolkit — and stays pure stdlib (D7-portable by design). +""" + +from __future__ import annotations + +import ast +import sys +from pathlib import Path + +import pytest + +from portfolio_optimiser_claude.okf import ConceptFile, bundle_context, navigate_bundle + +BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" +SRC_PKG = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser_claude" + + +def _write(path: Path, text: str) -> None: + path.write_text(text, encoding="utf-8") + + +def _make_bundle(tmp_path: Path, index_body: str, files: dict[str, str]) -> Path: + _write(tmp_path / "index.md", f"---\ntype: index\ntitle: Test\n---\n{index_body}\n") + for name, text in files.items(): + _write(tmp_path / name, text) + return tmp_path + + +class TestNavigation: + """§3 Step 1: navigate from index.md — deterministic, tolerant, boundary-checked.""" + + def test_missing_index_is_an_error(self, tmp_path: Path) -> None: + # A bundle has no entry point without index.md. + with pytest.raises(FileNotFoundError): + navigate_bundle(tmp_path) + + def test_order_is_index_first_then_first_seen_deduplicated(self, tmp_path: Path) -> None: + bundle = _make_bundle( + tmp_path, + "See [b](b.md) then [a](a.md) then [b again](b.md).", + { + "a.md": "---\ntype: project\ntitle: A\n---\nBody A.", + "b.md": "---\ntype: reference\ntitle: B\n---\nBody B.", + }, + ) + names = [c.path.name for c in navigate_bundle(bundle)] + assert names == ["index.md", "b.md", "a.md"] + + def test_out_of_bundle_and_escaping_targets_are_skipped(self, tmp_path: Path) -> None: + # A target containing a path separator is out-of-bundle — skipped, never + # raised; ../-escapes never resolve outside the bundle (fail-closed). + outside = tmp_path / "outside.md" + _write(outside, "---\ntype: project\ntitle: Outside\n---\nSecret.") + bundle_dir = tmp_path / "bundle" + bundle_dir.mkdir() + bundle = _make_bundle( + bundle_dir, + "Links: [up](../outside.md), [sub](sub/inner.md), [ok](a.md).", + {"a.md": "---\ntype: project\ntitle: A\n---\nBody A."}, + ) + names = [c.path.name for c in navigate_bundle(bundle)] + assert names == ["index.md", "a.md"] + + def test_broken_link_is_tolerated_never_raised(self, tmp_path: Path) -> None: + bundle = _make_bundle( + tmp_path, + "Links: [gone](missing.md), [ok](a.md).", + {"a.md": "---\ntype: project\ntitle: A\n---\nBody A."}, + ) + names = [c.path.name for c in navigate_bundle(bundle)] + assert names == ["index.md", "a.md"] + + def test_shared_bundle_navigates_all_linked_concepts(self) -> None: + # Integration on the shared example bundle: every index-linked file is + # reached exactly once; the out-of-bundle ../../README.md link is skipped. + names = [c.path.name for c in navigate_bundle(BUNDLE)] + assert names[0] == "index.md" + assert len(names) == len(set(names)) + assert set(names) == { + "index.md", + "bygg-kontor-nord.md", + "tiltak-led-retrofit.md", + "metode-ipmvp-a.md", + "kilder-realiseringsgap.md", + "verdict-led-fro.md", + } + + +class TestFrontmatter: + """§3 Step 1: leading ``---`` block, line-oriented ``key: value``; ``type`` required.""" + + def test_missing_type_is_an_error(self, tmp_path: Path) -> None: + _write(tmp_path / "index.md", "---\ntitle: No type\n---\nBody.") + with pytest.raises(ValueError, match="type"): + navigate_bundle(tmp_path) + + def test_unknown_fields_preserved_and_quotes_stripped(self, tmp_path: Path) -> None: + _write( + tmp_path / "index.md", + '---\ntype: index\ntitle: "Quoted title"\ncustom_field: kept\n---\nBody.', + ) + index = navigate_bundle(tmp_path)[0] + assert isinstance(index, ConceptFile) + assert index.type == "index" + assert index.title == "Quoted title" + assert index.frontmatter["custom_field"] == "kept" + + +class TestRendering: + """§3 Step 1: index body first, then ``## {type}: {title}`` sections.""" + + def test_sections_are_typed_and_titled_after_index_body(self, tmp_path: Path) -> None: + bundle = _make_bundle( + tmp_path, + "The summary.\n\nSee [a](a.md).", + {"a.md": "---\ntype: project\ntitle: Building A\n---\nProject body."}, + ) + context = bundle_context(bundle) + assert context.startswith("The summary.") + assert "## project: Building A" in context + assert "Project body." in context + + def test_empty_sections_are_dropped(self, tmp_path: Path) -> None: + bundle = _make_bundle( + tmp_path, + "Summary. See [empty](empty.md).", + {"empty.md": "---\ntype: reference\ntitle: Empty\n---\n"}, + ) + context = bundle_context(bundle) + assert "## reference: Empty" not in context + + +class TestVerdictLayerExclusion: + """LOAD-BEARING (§3 Step 1, §11): verdicts never leak via the read-context.""" + + def test_navigable_verdict_is_excluded_from_rendering(self, tmp_path: Path) -> None: + # Control + seam in one: the verdict file IS navigable (so exclusion is + # doing real work), yet its unique marker never reaches the read-context. + # RED if rendering stops gating on ``type: verdict``. + marker = "UNIQUE-REALIZATION-MARKER-0.82" + bundle = _make_bundle( + tmp_path, + "Summary. See [v](v.md) and [a](a.md).", + { + "v.md": f"---\ntype: verdict\ntitle: Seed\n---\nSignal: {marker}.", + "a.md": "---\ntype: project\ntitle: A\n---\nBody A.", + }, + ) + navigated = [c.path.name for c in navigate_bundle(bundle)] + assert "v.md" in navigated + context = bundle_context(bundle) + assert marker not in context + assert "## verdict" not in context + assert "## project: A" in context + + def test_shared_bundle_context_carries_no_realization_signal(self) -> None: + # The seed verdict's learning signal (realization rate 0.82, expected + # actual 24 600 NOK) must be absent from the rendered context — it may + # reach the prompt ONLY via the gated experience fold. + context = bundle_context(BUNDLE) + assert "## verdict" not in context + for leak in ("0.82", "0,82", "24600", "24 600", "realization_rate"): + assert leak not in context + # ...while the non-verdict concept layers ARE rendered: + assert "## project:" in context + assert "## hypothesis:" in context + assert "## methodology:" in context + assert "## reference:" in context + + +def _imported_module_names(module_path: Path) -> set[str]: + tree = ast.parse(module_path.read_text(encoding="utf-8")) + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + names.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + names.add(node.module.split(".")[0]) + return names + + +class TestContextSeamPurity: + """LOAD-BEARING (§11): the context seam imports no agent toolkit.""" + + @pytest.mark.parametrize("module", ["okf.py", "experience.py"]) + def test_context_seam_never_imports_an_agent_toolkit(self, module: str) -> None: + names = _imported_module_names(SRC_PKG / module) + assert not names & {"claude_agent_sdk", "anthropic"} + + def test_okf_is_pure_stdlib(self) -> None: + # D7-portable by design: navigation/rendering depends on nothing beyond + # the standard library (relative imports would show as level > 0). + names = _imported_module_names(SRC_PKG / "okf.py") + assert names <= set(sys.stdlib_module_names) diff --git a/tests/test_step1_expel_loadbearing.py b/tests/test_step1_expel_loadbearing.py new file mode 100644 index 0000000..ec577f3 --- /dev/null +++ b/tests/test_step1_expel_loadbearing.py @@ -0,0 +1,248 @@ +"""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", + )