327 lines
16 KiB
Python
327 lines
16 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 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 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"
|
|
"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 ---------------------
|
|
#
|
|
# 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
|