"""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 os import subprocess import sys from pathlib import Path import numpy as np from portfolio_optimiser.semretrieval import ( FakeEmbedder, HybridRanker, StructuralRetriever, cosine, ) from portfolio_optimiser.verdicts import ( 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_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