229 lines
11 KiB
Python
229 lines
11 KiB
Python
"""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
|