"""S3.1 semretrieval — load-bearing seams (SC2 cosine tie-break, SC5 MAF-free, SC8 no-network). ``semretrieval`` is the only numpy importer in the package and it sits on the MAF-free side of the D7 fault line. Three guards hold that line, each with a named detach point: 1. **Meta** — the module is registered in ``_MAF_FREE_MODULES``, so the direct-import AST guard (``test_okf_is_maf_free``) actually scans it. Without this the MAF-free claim is green-but-dead. 2. **Transitive + runtime** — the module body is exec'd standalone AND a ``rank()`` call is made, then ``sys.modules`` is checked. The AST guard only sees DIRECT imports and only at import time; a lazy ``from portfolio_optimiser.verdicts import similarity`` INSIDE ``rank`` would sail past both the AST guard and an import-only probe, and trip only this one. 3. **No network** — an AST sweep for network modules, since a real embeddings client is an extension point and must never be smuggled into the offline default. And one load-bearing behavioural seam (SC2): on a 500+ verdict base where two candidates are structurally INDISTINGUISHABLE, the cosine term is the only thing that can separate them. **Detach point: zero/remove the cosine term → CORRECT no longer top-1.** """ from __future__ import annotations import ast import subprocess import sys from pathlib import Path from portfolio_optimiser.semretrieval import ( FakeEmbedder, HybridRanker, StructuralRetriever, cosine, ) from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, VerdictStore, similarity _SEMRETRIEVAL = ( Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / "semretrieval.py" ) # The offline invariant: no verdict data and no proposal text may leave the machine. _NETWORK_ROOTS = {"socket", "urllib", "http", "requests", "httpx", "ftplib", "smtplib"} def test_semretrieval_registered_maf_free() -> None: """Meta: registered in the MAF-free guard list, so ``test_okf_is_maf_free`` scans it. Detach point: drop ``semretrieval.py`` from ``_MAF_FREE_MODULES`` → RED.""" from tests.test_okf import _MAF_FREE_MODULES assert "semretrieval.py" in _MAF_FREE_MODULES def test_semretrieval_import_and_rank_pull_in_no_maf_and_no_verdicts() -> None: """The genuine seam: exec the module body standalone, then actually RANK, then inspect ``sys.modules``. Ranking is the moment a lazy import would fire — ``StructuralRetriever``/``HybridRanker`` need the structural score, and the design INJECTS it as a callable precisely so that ``verdicts`` (which imports ``agent_framework``) stays out of this module's runtime graph. The candidates are duck-typed ``SimpleNamespace`` stand-ins, never real ``ProposalFeatures`` /``Verdict`` — building those would import ``verdicts`` in the probe itself and defeat the very assertion being made. Detach point: replace the injected ``similarity`` with ``from portfolio_optimiser.verdicts import similarity`` inside ``rank`` → RED.""" check = ( "import importlib.util, sys, types\n" f"spec = importlib.util.spec_from_file_location('semretrieval_standalone', {str(_SEMRETRIEVAL)!r})\n" "mod = importlib.util.module_from_spec(spec)\n" "spec.loader.exec_module(mod)\n" "\n" "def feat(codes, measure, saving, desc):\n" " return types.SimpleNamespace(affected_codes=frozenset(codes), measure_type=measure,\n" " claimed_saving_nok=saving, description=desc)\n" "\n" "query = feat(['05.2'], 'scope_reduction', 220000.0, 'query text')\n" "candidates = [\n" " types.SimpleNamespace(id='A', proposal_features=feat(['05.2'], 'scope_reduction', 200000.0, 'a')),\n" " types.SimpleNamespace(id='B', proposal_features=feat(['09.1'], 'rate_renegotiation', 50000.0, 'b')),\n" "]\n" "sim = lambda q, c: 1.0 if q.measure_type == c.measure_type else 0.0\n" "\n" "structural = mod.StructuralRetriever(sim).rank(query, candidates, 2)\n" "assert [v.id for v in structural] == ['A', 'B'], structural\n" "hybrid = mod.HybridRanker(mod.FakeEmbedder(), sim).rank(query, candidates, 2)\n" "assert len(hybrid) == 2, hybrid\n" "\n" "for name in ('agent_framework', 'mcp', 'portfolio_optimiser.verdicts', 'portfolio_optimiser'):\n" " assert name not in sys.modules, name + ' leaked into the semretrieval runtime graph'\n" ) result = subprocess.run([sys.executable, "-c", check], capture_output=True, text=True) assert result.returncode == 0, result.stderr def test_semretrieval_imports_no_network_modules() -> None: """SC8 — the offline default must not be able to reach the network. Detach point: add ``import urllib.request`` (or any ``_NETWORK_ROOTS`` member) to semretrieval.py → RED.""" tree = ast.parse(_SEMRETRIEVAL.read_text(encoding="utf-8")) imported_roots: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): imported_roots.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: imported_roots.add(node.module.split(".")[0]) leaked = imported_roots & _NETWORK_ROOTS assert not leaked, f"semretrieval imports network module(s): {sorted(leaked)}" # --- SC2: the cosine term is load-bearing on a structurally tied 500+ base --------------------- _TIED_CODES = frozenset({"05.2", "03.1"}) # CORRECT's id sorts AFTER the distractor's, so the structural ranking's `(-similarity, id)` key # puts the DISTRACTOR first. Only the cosine term can overturn that. The id doubles as a unique # grep marker: it appears nowhere else in the repo, so a stray match cannot fake a pass. _CORRECT_ID = "zz-s31-cosine-tiebreak-4b7e" _DISTRACTOR_ID = "aa-structural-tie-winner" _QUERY = ProposalFeatures( affected_codes=_TIED_CODES, measure_type="scope_reduction", claimed_saving_nok=220_000.0, description="reduce asphalt base course thickness on the school approach", ) def _tied_verdict(verdict_id: str, description: str) -> Verdict: """Structurally IDENTICAL to the query — same codes, measure and magnitude bucket. Only the (structurally ignored) description differs, which is exactly what makes these two candidates inseparable without a semantic signal.""" return Verdict( id=verdict_id, proposal_features=ProposalFeatures( affected_codes=_TIED_CODES, measure_type="scope_reduction", claimed_saving_nok=200_000.0, # same bucket [100k, 500k) as the query description=description, ), decision="approved", rationale=f"prior ruling reached via {verdict_id}", ) def _synthetic_base(n: int = 500) -> list[Verdict]: """Two structurally tied candidates plus ``n - 2`` strictly-lower-scoring fillers, so the ranking has to hold up at a realistic base size rather than on a three-item toy store.""" correct = _tied_verdict( _CORRECT_ID, # Tuned, and admittedly so: a sha256 projection carries no inherent semantics, so this # wording was selected because it lands nearer the query than the distractor does. The # proof of the seam is the detach control below, not the plausibility of this string. "shallower asphalt base layer along the school approach", ) distractor = _tied_verdict( _DISTRACTOR_ID, "unrelated administrative rebate on office cleaning contract", ) fillers = [ Verdict( id=f"filler-{i:04d}", proposal_features=ProposalFeatures( affected_codes=frozenset({f"9{i % 90:02d}.1"}), # no overlap -> jaccard 0 measure_type="rate_renegotiation", # no measure match claimed_saving_nok=5_000_000.0, # different magnitude bucket description=f"unrelated filler measure {i}", ), decision="rejected", rationale="filler", ) for i in range(n - 2) ] # Interleaved so a rank that accidentally preserved input order would not pass by luck. return [*fillers[: (n - 2) // 2], distractor, correct, *fillers[(n - 2) // 2 :]] def test_sc2_fixture_preconditions_hold() -> None: """The three properties the SC2 proof rests on, asserted rather than assumed: the pair is structurally tied, CORRECT loses the id tie-break, and cosine favours CORRECT.""" base = _synthetic_base() correct = next(v for v in base if v.id == _CORRECT_ID) distractor = next(v for v in base if v.id == _DISTRACTOR_ID) assert similarity(_QUERY, correct.proposal_features) == similarity( _QUERY, distractor.proposal_features ) assert _DISTRACTOR_ID < _CORRECT_ID # structural id tie-break favours the distractor embedder = FakeEmbedder() query_vector = embedder(_QUERY) assert cosine(query_vector, embedder(correct.proposal_features)) > cosine( query_vector, embedder(distractor.proposal_features) ) assert len(base) >= 500 def test_sc2_control_structural_ranking_puts_the_distractor_first() -> None: """CONTROL — without a semantic term the CORRECT verdict is unreachable. If this ever goes green with CORRECT on top, the positive below proves nothing.""" base = _synthetic_base() top = StructuralRetriever(similarity).rank(_QUERY, base, 3) assert top[0].id == _DISTRACTOR_ID assert top[0].id != _CORRECT_ID def test_sc2_control_hybrid_at_weight_zero_is_the_detach_point() -> None: """DETACH — zeroing the cosine weight collapses the hybrid onto the structural ranking and the CORRECT verdict falls back out of the top slot. This is the seam being load-bearing: remove the cosine contribution and the positive test below turns RED.""" base = _synthetic_base() detached = HybridRanker(FakeEmbedder(), similarity, weight=0.0).rank(_QUERY, base, 3) assert detached[0].id == _DISTRACTOR_ID assert detached[0].id != _CORRECT_ID def test_sc2_positive_cosine_breaks_the_structural_tie() -> None: """POSITIVE — with the cosine term active, the semantically closer verdict wins the tie it loses structurally, on a 500+ base.""" base = _synthetic_base() top = HybridRanker(FakeEmbedder(), similarity).rank(_QUERY, base, 3) assert top[0].id == _CORRECT_ID def test_sc2_positive_holds_through_the_store_seam() -> None: """The same result through the real entry point: a store with the hybrid installed must surface CORRECT, since that is the path ``run.py --semantic-retrieval`` drives.""" store = VerdictStore(verdicts=_synthetic_base()) store.retriever = HybridRanker(FakeEmbedder(), similarity) assert store.retrieve(_QUERY, k=3)[0].id == _CORRECT_ID # ...and the untouched default store still cannot see it. assert VerdictStore(verdicts=_synthetic_base()).retrieve(_QUERY, k=3)[0].id == _DISTRACTOR_ID