"""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 TIE on the structural score, the cosine term is the only thing that can separate them. **Detach point: zero/remove the cosine term → CORRECT no longer top-1.** The pair ties on equal Jaccard over DIFFERENT code sets rather than by being structurally identical, so both candidates are constructible through ``capture_verdict`` — the path the framework actually mints on. Plus one regression guard (defect P1): a framework-minted and an externally-authored verdict that are structurally identical must embed identically, so the system's own echo of the query can no longer outrank genuine expert prose. """ from __future__ import annotations import ast import itertools import os import subprocess import sys from pathlib import Path import numpy as np from portfolio_optimiser.semretrieval import ( SEMANTIC_WEIGHT_DEFAULT, FakeEmbedder, HybridRanker, StructuralRetriever, cosine, ) from portfolio_optimiser.verdicts import ( _W_MAGNITUDE, ProposalFeatures, Verdict, VerdictStore, capture_verdict, similarity, verdict_from_dict, verdict_to_dict, ) _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" "# The vector-store functions are exercised BEFORE the sweep too: they are public API and a\n" "# lazy import placed inside either one would otherwise never run under this probe.\n" "import tempfile\n" "tmp = tempfile.mkdtemp()\n" "assert mod.load_vector_store(tmp) is None, 'a directory with no artifacts must load as None'\n" "mod.save_vector_store(tmp, candidates, mod.FakeEmbedder())\n" "loaded = mod.load_vector_store(tmp)\n" "assert loaded is not None, 'the store round trip returned None'\n" "ids, matrix = loaded\n" "assert ids == ['A', 'B'], ids\n" "assert matrix.shape == (2, mod.EMBED_DIM), matrix.shape\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_blas_thread_pins_are_set_before_numpy_is_imported() -> None: """SC3 GUARD — the determinism claim's one untested half, until now. Two properties, both checked in a FRESH subprocess that execs the module standalone: 1. all four thread variables are set to ``"1"`` after the module body runs; 2. ``numpy`` is absent from ``sys.modules`` BEFORE the module's env writes execute — the ordering the whole mechanism depends on, since every backend latches its variable by the time the numpy import completes. ``VECLIB_MAXIMUM_THREADS`` is the one that BITES on this build (numpy links Accelerate); the other three cover OpenBLAS/MKL/OpenMP deployments. Before it was added this block pinned nothing at all here, and no test could tell — ``grep -rn 'OPENBLAS\\|OMP_NUM\\|MKL_NUM' tests/`` returned zero hits. Detach point: move any ``os.environ.setdefault`` below the ``import numpy`` line → RED.""" check = ( "import importlib.util, sys\n" "assert 'numpy' not in sys.modules, 'numpy was imported before the probe started'\n" f"spec = importlib.util.spec_from_file_location('semretrieval_pins', {str(_SEMRETRIEVAL)!r})\n" "mod = importlib.util.module_from_spec(spec)\n" "\n" "# Trip-wire: record whether numpy was already imported at the moment the env writes ran.\n" "import os\n" "seen = {}\n" "real_setdefault = os.environ.setdefault\n" "def spy(key, value):\n" " seen.setdefault(key, 'numpy' in sys.modules)\n" " return real_setdefault(key, value)\n" "os.environ.setdefault = spy\n" "try:\n" " spec.loader.exec_module(mod)\n" "finally:\n" " os.environ.setdefault = real_setdefault\n" "\n" "for var in ('VECLIB_MAXIMUM_THREADS', 'OPENBLAS_NUM_THREADS', 'OMP_NUM_THREADS',\n" " 'MKL_NUM_THREADS'):\n" " assert os.environ.get(var) == '1', var + ' is ' + repr(os.environ.get(var))\n" " assert var in seen, var + ' was never pinned by the module body'\n" " assert seen[var] is False, var + ' was pinned AFTER numpy was imported'\n" "assert 'numpy' in sys.modules, 'the module never imported numpy — probe is not exercising it'\n" ) env = { k: v for k, v in os.environ.items() if not k.endswith(("_NUM_THREADS", "_MAXIMUM_THREADS")) } result = subprocess.run([sys.executable, "-c", check], capture_output=True, text=True, env=env) assert result.returncode == 0, result.stderr def test_semretrieval_is_the_sole_numpy_importer() -> None: """The PREMISE the pin above rests on, guarded — it was prose until now. ``semretrieval.py`` can pin the BLAS thread variables at import time only because it is the first and only place numpy enters the package. A second importer elsewhere under ``src/`` would very likely be imported FIRST (``run.py``, ``verdicts.py`` and friends load long before the opt-in retrieval seam), numpy would latch its backend's thread count before these ``setdefault`` calls ever run, and the pin would become a silent no-op in the real process — while ``test_blas_thread_pins_are_set_before_numpy_is_imported`` stayed green, because that probe execs THIS module standalone in a fresh interpreter and can never observe the collision. AST, not substring: a docstring mentioning numpy must not trip the guard, exactly as ``test_okf_is_maf_free`` reasons about the MAF-free claim. Deliberate limit, stated rather than oversold: this covers ``src/`` code, which is what the premise actually claims. A third-party dependency importing numpy transitively is out of scope — the pins are defence-in-depth for byte-identical artifacts, never the basis of the ranking guarantee (see the module docstring). Detach point: add ``import numpy`` to any other module under ``src/portfolio_optimiser/`` → RED, while the rest of the suite stays green.""" package = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" importers: set[str] = set() for path in sorted(package.glob("*.py")): roots: set[str] = set() for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): if isinstance(node, ast.Import): roots.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: roots.add(node.module.split(".")[0]) if "numpy" in roots: importers.add(path.name) assert importers == {"semretrieval.py"}, ( f"numpy is imported by {sorted(importers)}; the import-time BLAS pin in semretrieval.py " "is only viable while that module is the SOLE numpy importer. Another importer loaded " "earlier makes the pin a no-op in the real process and no existing test can see it." ) 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.""" from tests.test_okf import _dynamic_import_targets tree = ast.parse(_SEMRETRIEVAL.read_text(encoding="utf-8")) imported_roots: set[str] = set() dynamic: list[str] = [] 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]) elif isinstance(node, ast.Call): dynamic += _dynamic_import_targets(node) leaked = imported_roots & _NETWORK_ROOTS assert not leaked, f"semretrieval imports network module(s): {sorted(leaked)}" # RATCHET (green today): this walk sees only static imports, so a single # ``importlib.import_module("httpx")`` would defeat the whole guard. Refuse dynamic imports # outright — a computed target cannot be judged statically. The ``importlib.util`` used by the # probes in this file lives inside the probe STRINGS, not in the module under test. assert dynamic == [], ( f"semretrieval performs dynamic import(s) {dynamic} — the no-network guard is a STATIC " "check and cannot see through them" ) # --- SC2: the cosine term is load-bearing on a structurally tied 500+ base --------------------- # # The pair ties on the STRUCTURAL score while carrying DIFFERENT cost codes — equal Jaccard # against the query (1/3 each), same measure, same magnitude bucket => similarity 0.6 == 0.6. # That matters: the previous fixture tied by being structurally IDENTICAL and separated only by # hand-written prose, a shape the framework cannot mint (``_mint_id`` ignores description, so # both candidates would share one id and ``VerdictStore.add`` is first-write-wins). Differing in # codes makes the pair genuinely constructible through ``capture_verdict`` — the real minting # path — so the proof now runs on data the system can actually produce. # # Every feature set carries ``description == measure_type``, which is exactly what both live # minting paths emit (``run._features_of`` sets ``description=proposal.measure``; # ``verdicts._features_from_ir`` sets ``description=ir["measure"]``). _MEASURE = "asfalt" _QUERY_CODES = frozenset({"05.1", "05.2"}) _CORRECT_CODES = frozenset({"05.1", "07.4"}) _DISTRACTOR_CODES = frozenset({"05.2", "09.8"}) # Unique grep markers: these strings appear nowhere else in the repo, so a stray match cannot # fake a pass. They ride in the RATIONALE (not the id) because ids are now minted content # hashes — and because ``format_fewshot`` puts the rationale into the prompt, which is the # channel the CLI-level proof downstream depends on. _CORRECT_MARKER = "zz-s31-cosine-tiebreak-4b7e" _DISTRACTOR_MARKER = "aa-structural-tie-winner" _QUERY = ProposalFeatures( affected_codes=_QUERY_CODES, measure_type=_MEASURE, claimed_saving_nok=220_000.0, description=_MEASURE, ) def _tied_verdict(codes: frozenset[str], decision: str, marker: str) -> Verdict: """Structurally TIED with the query — equal Jaccard, same measure, same magnitude bucket — but on a different code set, so the canonical embedding string differs and cosine has something to separate. Minted through ``capture_verdict``, never hand-assigned.""" return capture_verdict( ProposalFeatures( affected_codes=codes, measure_type=_MEASURE, claimed_saving_nok=200_000.0, # same bucket [100k, 500k) as the query description=_MEASURE, ), decision, f"prior ruling reached via {marker}", ) # Minted ids, resolved once so the ordering property below is stated in terms of real values. _CORRECT_ID = _tied_verdict(_CORRECT_CODES, "approved", _CORRECT_MARKER).id _DISTRACTOR_ID = _tied_verdict(_DISTRACTOR_CODES, "rejected", _DISTRACTOR_MARKER).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.""" # Not tuned prose any more: the two candidates differ in their CODE SETS, and the cosine # ordering over a sha256 projection is deterministic but semantically arbitrary. The proof of # the seam is the detach control below — that removing the cosine term flips the ranking — # never the plausibility of either candidate. correct = _tied_verdict(_CORRECT_CODES, "approved", _CORRECT_MARKER) distractor = _tied_verdict(_DISTRACTOR_CODES, "rejected", _DISTRACTOR_MARKER) 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: """Every property the SC2 proof rests on, asserted rather than assumed: the pair is structurally tied on DIFFERENT code sets, both ids are minted by ``capture_verdict``, CORRECT loses the id tie-break, and cosine favours CORRECT. The code-set assertion is the one that keeps this fixture honest — a pair that tied by being structurally identical could not be minted at all, since ``_mint_id`` would collapse them onto one id and the store would keep only the first.""" 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) # Tied structurally, but NOT structurally identical. assert similarity(_QUERY, correct.proposal_features) == similarity( _QUERY, distractor.proposal_features ) assert correct.proposal_features.affected_codes != distractor.proposal_features.affected_codes # The framework-minted shape: description carries the measure, on every feature set. for features in (_QUERY, correct.proposal_features, distractor.proposal_features): assert features.description == features.measure_type == _MEASURE # Ids are content hashes of the features, not hand-written — re-minting reproduces them. assert correct.id != distractor.id assert correct.id == capture_verdict(correct.proposal_features, "approved", "re-mint").id assert distractor.id == capture_verdict(distractor.proposal_features, "rejected", "re-mint").id 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_minted_and_authored_verdicts_embed_identically_on_a_structural_tie() -> None: """REGRESSION (defect P1) — a framework-minted verdict and an externally-authored one that are structurally identical must now embed IDENTICALLY. Before ``description`` left the embedding, the framework minted every feature set with ``description == measure`` while a human/persona verdict arrived through the inbox carrying real prose. On a mixed store the system's own echo of the query scored ~1.0 and genuine expert prose ~0.72, so the hybrid pushed real rulings BELOW the framework's restatement of its own question — the inverse of the feature's purpose, and the same self-contamination the Step-8 promotion gate exists to prevent. Detach point: restore ``features.description`` to ``_canonical_feature_string`` → RED.""" features = ProposalFeatures( affected_codes=_CORRECT_CODES, measure_type=_MEASURE, claimed_saving_nok=200_000.0, description=_MEASURE, # the shape run._features_of / _features_from_ir emit ) minted = capture_verdict(features, "approved", "framework-captured ruling") # The same proposal as an expert would file it: identical structure, human prose. authored = verdict_from_dict( { **verdict_to_dict(minted), "rationale": "the base course reduction was accepted on the school approach", "proposal_features": { **verdict_to_dict(minted)["proposal_features"], "description": "shallower asphalt base layer along the school approach", }, } ) embedder = FakeEmbedder() query_vector = embedder(_QUERY) # The load-bearing claim: the two embed to the SAME vector, bit for bit. assert np.array_equal( embedder(minted.proposal_features), embedder(authored.proposal_features) ), "prose still leaks into the embedding — the framework's own echo can outrank expert text" # ...and therefore rank identically. Compared at the ranking key's own tolerance, not on raw # equality: ``np.dot`` over two separately allocated but bit-identical vectors can differ by # one ulp, because the reduction path varies with buffer alignment. That is precisely why # ``HybridRanker`` orders on ``(-round(score, 9), id)`` rather than the raw score — asserting # bit-equality here would pin a property the system deliberately does not rely on. assert round(cosine(query_vector, embedder(minted.proposal_features)), 9) == round( cosine(query_vector, embedder(authored.proposal_features)), 9 ) 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 # --- The other half of the contract: the hybrid must PRESERVE a real structural gap ------------ # # SC2 above proves the cosine term can break a structural TIE. Nothing above proves it stops # there. The two claims are opposites at the margin, and only one of them was ever pinned: an # arbitrary sha256 term with enough weight does not merely settle ties, it overturns genuine # orderings. The blend weight is the only thing that bounds how large a gap it can overturn, and # until this test that bound rested on a comment rather than on a guard. # # The fixture family is fully enumerated and RNG-free — the repo's byte-determinism posture rules # out a flaky property test, so the sweep is exhaustive over a small vocabulary instead. _ORDER_CODES = ("A10", "A20", "B10", "B20") _ORDER_MEASURES = ("energy", "material") # One amount per magnitude bucket boundary that matters: [0, 1e5), [1e5, 5e5), [5e5, 1e6). _ORDER_AMOUNTS = (5e4, 2e5, 7e5) # The structural score is a sum of three float terms, so a gap that is mathematically EXACTLY one # ``_W_MAGNITUDE`` step can subtract to 0.1499999999999999. Comparing strictly would silently drop # those pairs — and they are real contract violations: admitting them raises the qualifying count # from 71 820 to 73 980 and the adverse count at ``w=0.5`` from 7 to 9. A guard errs toward # sensitivity, so the threshold carries a float tolerance rather than pretending the arithmetic is # exact. _GAP_TOLERANCE = 1e-9 def _order_features(codes: frozenset[str], measure: str, amount: float) -> ProposalFeatures: """``description == measure_type`` — the shape both live minting paths emit.""" return ProposalFeatures( affected_codes=codes, measure_type=measure, claimed_saving_nok=amount, description=measure, ) def _order_family() -> list[ProposalFeatures]: """4 codes (subsets of size 1-2) x 2 measures x 3 magnitude buckets = 60 feature sets.""" code_sets = [ frozenset(combo) for size in (1, 2) for combo in itertools.combinations(_ORDER_CODES, size) ] return [ _order_features(codes, measure, amount) for codes in code_sets for measure in _ORDER_MEASURES for amount in _ORDER_AMOUNTS ] def test_hybrid_preserves_structural_order_across_one_magnitude_step() -> None: """CONTRACT — over a structural gap of one ``_W_MAGNITUDE`` step the hybrid must not overturn the ordering. Asserted at the SHIPPED default weight, so the constant itself is what is pinned. Two deliberately split assertions: 1. **Named anchor, through the real API.** query ``{A10}/energy/200k``; better ``{A10,B20}/energy/50k`` (structural 0.5500); worse ``{A20}/energy/200k`` (structural 0.4000). The gap is exactly one ``_W_MAGNITUDE`` step, and ``worse`` shares NO cost code with the query while ``better`` does. Human-readable, instant, and it exercises ``HybridRanker.rank`` rather than a re-implementation of the score. 2. **Exhaustive breadth sweep** over the n=60 family, scoring ``w * cosine + (1 - w) * structural`` directly against one cached embedding per feature set. ``rank()`` re-embeds every candidate on every call, which measured 6,1 s over this family versus 1,1 s cached — too expensive for a 37 s suite, and the anchor above already covers the real ranking path. MEASURED BASIS (this family, this embedder, ``EMBED_DIM=64``): 73 980 qualifying triples; **9 adverse at ``w=0.5``**, and **0 at every weight from 0.45 down to 0.05**. So this test is genuinely RED at 0.5 and green at the shipped 0.25 — it is not vacuous, and it is not fixture-tuned to the exact value either, since the whole corridor below 0.45 passes. THE TRAP, quantified rather than warned about: a 3x2x2 family (n=24) yields **0** adverse triples and passes SILENTLY at ``w=0.5``. The family size is therefore load-bearing and must not be shrunk — a test that cannot fail proves nothing, which is the exact defect class this test was written to close. Detach point: restore ``SEMANTIC_WEIGHT_DEFAULT = 0.5`` -> RED.""" embedder = FakeEmbedder() # 1. Named anchor, through the shipped ranking path. query = _order_features(frozenset({"A10"}), "energy", 200_000.0) better = capture_verdict( _order_features(frozenset({"A10", "B20"}), "energy", 50_000.0), "approved", "shares a cost code with the query", ) worse = capture_verdict( _order_features(frozenset({"A20"}), "energy", 200_000.0), "rejected", "shares no cost code with the query", ) # The anchor's structural values are asserted, not assumed — if ``similarity``'s weights ever # move, this test must fail loudly rather than quietly stop testing a one-step gap. structural_better = similarity(query, better.proposal_features) structural_worse = similarity(query, worse.proposal_features) assert round(structural_better, 4) == 0.5500 assert round(structural_worse, 4) == 0.4000 assert round(structural_better - structural_worse, 9) == round(_W_MAGNITUDE, 9) top = HybridRanker(embedder, similarity).rank(query, [better, worse], 2) assert top[0].id == better.id, ( "the hybrid ranked a candidate sharing NO cost code with the query above one that does, " f"across a full {_W_MAGNITUDE} structural step, on nothing but sha256 cosine noise " f"(weight={SEMANTIC_WEIGHT_DEFAULT})" ) # 2. Exhaustive breadth sweep, one cached embedding per feature set. family = _order_family() assert len(family) == 60, "the family size is load-bearing — n=24 passes this test vacuously" vectors = [embedder(features) for features in family] qualifying = 0 adverse: list[str] = [] for query_index, query_features in enumerate(family): structural = [similarity(query_features, candidate) for candidate in family] for i, j in itertools.combinations(range(len(family)), 2): if query_index in (i, j): continue high, low = (i, j) if structural[i] >= structural[j] else (j, i) if structural[high] - structural[low] < _W_MAGNITUDE - _GAP_TOLERANCE: continue qualifying += 1 score_high = ( SEMANTIC_WEIGHT_DEFAULT * cosine(vectors[query_index], vectors[high]) + (1.0 - SEMANTIC_WEIGHT_DEFAULT) * structural[high] ) score_low = ( SEMANTIC_WEIGHT_DEFAULT * cosine(vectors[query_index], vectors[low]) + (1.0 - SEMANTIC_WEIGHT_DEFAULT) * structural[low] ) # Compared at the ranking key's own tolerance — ``HybridRanker`` orders on # ``(-round(score, 9), id)``, so anything finer than that is not an inversion the # ranker can express. if round(score_low, 9) > round(score_high, 9): adverse.append( f"query={sorted(query_features.affected_codes)}/{query_features.measure_type}" f"/{query_features.claimed_saving_nok:.0f} " f"better={sorted(family[high].affected_codes)} ({structural[high]:.4f}) " f"lost to worse={sorted(family[low].affected_codes)} ({structural[low]:.4f})" ) # Anti-vacuity floor: if a future edit narrows the vocabulary, this fails before the contract # assertion below can pass for the wrong reason. assert qualifying > 70_000, ( f"only {qualifying} qualifying triples — the family no longer expresses the contract " "(measured basis: 73 980)" ) assert adverse == [], ( f"{len(adverse)} structural orderings overturned by the semantic term at " f"weight={SEMANTIC_WEIGHT_DEFAULT}; first three: {adverse[:3]}" )