feat(s31): Retriever seam — StructuralRetriever default + HybridRanker with injected similarity

This commit is contained in:
Kjell Tore Guttormsen 2026-07-25 06:16:23 +02:00
commit ae684c89d3
3 changed files with 218 additions and 9 deletions

View file

@ -21,7 +21,19 @@ from pathlib import Path
import numpy as np
import pytest
from portfolio_optimiser.semretrieval import EMBED_DIM, FakeEmbedder, cosine
from portfolio_optimiser.semretrieval import (
EMBED_DIM,
FakeEmbedder,
HybridRanker,
StructuralRetriever,
cosine,
)
from portfolio_optimiser.verdicts import (
ProposalFeatures,
Verdict,
VerdictStore,
similarity,
)
_SRC = Path(__file__).resolve().parents[1] / "src"
@ -131,3 +143,127 @@ def test_cosine_of_zero_norm_is_zero() -> None:
assert cosine(zeros, vec) == 0.0
assert cosine(vec, zeros) == 0.0
assert cosine(zeros, zeros) == 0.0
# --- SC1: the Retriever seam is additive — the DEFAULT ranking is byte-identical to today ---
_QUERY = ProposalFeatures(
affected_codes=frozenset({"05.2", "03.1"}),
measure_type="scope_reduction",
claimed_saving_nok=220_000, # bucket [100k, 500k)
description="asphalt base course reduction near school",
)
def _store_with_true_match_and_decoys() -> tuple[VerdictStore, str]:
"""Mirrors ``tests/test_verdicts.py::_store_with_true_match_and_decoys`` — the true match
shares the STRUCTURED fields with the query but uses different wording; the decoys share
surface text but differ structurally."""
true_match = Verdict(
id="TRUE",
proposal_features=ProposalFeatures(
affected_codes=frozenset({"05.2", "03.1"}),
measure_type="scope_reduction",
claimed_saving_nok=200_000,
description="zzz totally unrelated wording alpha beta",
),
decision="approved",
rationale="prior scope reduction on the same codes was approved",
)
decoy_low = Verdict(
id="DECOY-LOW",
proposal_features=ProposalFeatures(
affected_codes=frozenset({"09.1"}),
measure_type="rate_renegotiation",
claimed_saving_nok=50_000,
description="asphalt base course reduction near school",
),
decision="rejected",
rationale="surface-text decoy",
)
decoy_high = Verdict(
id="DECOY-HIGH",
proposal_features=ProposalFeatures(
affected_codes=frozenset({"21.2"}),
measure_type="material_substitution",
claimed_saving_nok=700_000,
description="asphalt base course reduction extra words",
),
decision="rejected",
rationale="surface-text decoy",
)
return VerdictStore(verdicts=[decoy_low, true_match, decoy_high]), "TRUE"
def test_default_retriever_is_none() -> None:
"""Opt-in seam: a store built the ordinary way carries no retriever, so nothing about
default retrieval changes unless a caller deliberately installs one."""
store, _ = _store_with_true_match_and_decoys()
assert store.retriever is None
def test_default_retrieve_matches_the_pre_seam_structural_sort() -> None:
"""SC1 — with no retriever installed, ``retrieve`` must reproduce the exact ordering the
pre-seam inline sort produced: key ``(-similarity, id)``, sliced to k."""
store, _ = _store_with_true_match_and_decoys()
expected = [
v.id
for v in sorted(
store.verdicts,
key=lambda v: (-similarity(_QUERY, v.proposal_features), v.id),
)[:3]
]
assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected
def test_structural_retriever_is_the_wired_default_not_dead_code() -> None:
"""SC1 — the default path goes through ``StructuralRetriever``, so it is exercised code,
not an unused parallel implementation that can silently rot."""
store, _ = _store_with_true_match_and_decoys()
direct = StructuralRetriever(similarity).rank(_QUERY, store.verdicts, 3)
assert [h.id for h in store.retrieve(_QUERY, k=3)] == [v.id for v in direct]
def test_retrieve_still_rejects_non_positive_k() -> None:
"""The pre-seam ``k <= 0`` guard must survive the delegation rewrite
(asserted independently by tests/test_verdicts.py::test_retrieve_rejects_non_positive_k)."""
store, _ = _store_with_true_match_and_decoys()
with pytest.raises(ValueError):
store.retrieve(_QUERY, k=0)
def test_installed_retriever_actually_routes_retrieval() -> None:
"""Delegation proof: a sentinel retriever must receive the call and its result must be
what ``retrieve`` returns otherwise the seam is decorative."""
store, _ = _store_with_true_match_and_decoys()
calls: list[tuple[ProposalFeatures, int, int]] = []
class _Sentinel:
def rank(self, query: ProposalFeatures, candidates: list[Verdict], k: int) -> list[Verdict]:
calls.append((query, len(candidates), k))
return list(reversed(candidates))[:k]
store.retriever = _Sentinel()
hits = store.retrieve(_QUERY, k=2)
assert calls == [(_QUERY, 3, 2)]
assert [h.id for h in hits] == ["DECOY-HIGH", "TRUE"]
def test_hybrid_ranker_is_injectable_and_routes_retrieval() -> None:
"""A HybridRanker installed on the store must be the ranker ``retrieve`` delegates to."""
store, _ = _store_with_true_match_and_decoys()
ranker = HybridRanker(FakeEmbedder(), similarity)
expected = [v.id for v in ranker.rank(_QUERY, store.verdicts, 3)]
store.retriever = ranker
assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected
def test_hybrid_ranker_at_weight_zero_equals_the_structural_ranking() -> None:
"""Weight 0 removes the semantic term entirely — the blend must then collapse onto the
structural ordering. This is the control the Step-5 detach test relies on."""
store, _ = _store_with_true_match_and_decoys()
hybrid = HybridRanker(FakeEmbedder(), similarity, weight=0.0)
structural = StructuralRetriever(similarity)
assert [v.id for v in hybrid.rank(_QUERY, store.verdicts, 3)] == [
v.id for v in structural.rank(_QUERY, store.verdicts, 3)
]