feat(context): S7 — D7 context seam: OKF navigation + gated ExpeL fold

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QdSfQdND84oeq2mbjueLTS
This commit is contained in:
Kjell Tore Guttormsen 2026-07-03 06:48:13 +02:00
commit 46f2f521f7
4 changed files with 730 additions and 0 deletions

View file

@ -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}"

View file

@ -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)

202
tests/test_okf.py Normal file
View file

@ -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)

View file

@ -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",
)