Kjell Tore Guttormsen
37547fe292
The context sets, the packaged knowledge bases and the example bundles are replaced by one fictitious example set about IT operations in an invented organisation: three context sets (serverrom-2027, driftsavtale-2027 and the two-base drift-og-avtale-2027), two synthetic knowledge bases under src/portfolio_optimiser/data/kunnskapsbaser and two example bundles under src/portfolio_optimiser/data/bundles. Numbers, codes and structural values in tests and fixtures are kept; names, ids and wording change. Dated measurement documents that only recorded runs on the replaced material are deleted. Gate figures measured on the new set are not comparable with earlier ones. The exclusion gate from the previous commit is green: 0 tracked files hit outside the shared/ subtree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
679 lines
34 KiB
Python
679 lines
34 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 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 = "lisens"
|
|
_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 licence scope reduction was accepted at the head office",
|
|
"proposal_features": {
|
|
**verdict_to_dict(minted)["proposal_features"],
|
|
"description": "smaller licence tier across the head office",
|
|
},
|
|
}
|
|
)
|
|
|
|
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]}"
|
|
)
|
|
|
|
|
|
# --- The non-finite guard: an injected embedder must not be able to corrupt the sort key --------
|
|
#
|
|
# ``cosine``'s docstring claims its guard is load-bearing because "a NaN reaching the ranking sort
|
|
# key would corrupt ordering silently rather than failing loudly". The guard tested ``norm == 0.0``
|
|
# only, which a NaN or inf norm passes straight through — so the claim was prose, not behaviour.
|
|
# ``FakeEmbedder`` cannot emit a non-finite component, but ``Embedder`` is a documented CODE-LEVEL
|
|
# extension point (``docs/extending.md``), and an injected client is exactly what this seam exists
|
|
# to accept. That makes the reachable caller the injected one, NOT the shipped fake.
|
|
|
|
|
|
class _NaNEmbedder:
|
|
"""An injected embedder that has gone wrong on ONE input — the realistic failure, since a
|
|
client that returned NaN for everything would be caught by the first smoke test."""
|
|
|
|
def __init__(self, poison: str) -> None:
|
|
self._poison = poison
|
|
self._fake = FakeEmbedder()
|
|
|
|
def __call__(self, features: ProposalFeatures) -> np.ndarray:
|
|
if features.description == self._poison:
|
|
return np.full(len(self._fake(features)), np.nan, dtype="<f8")
|
|
return self._fake(features)
|
|
|
|
|
|
def _verdict(vid: str, description: str) -> Verdict:
|
|
return Verdict(
|
|
id=vid,
|
|
proposal_features=ProposalFeatures(
|
|
affected_codes=frozenset({"05.2"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=200_000,
|
|
description=description,
|
|
),
|
|
decision="approved",
|
|
rationale="prior scope reduction on the same codes was approved",
|
|
)
|
|
|
|
|
|
def test_hybrid_rank_refuses_a_non_finite_embedding() -> None:
|
|
"""Detach point: drop the non-finite check in ``cosine`` -> ``rank`` returns a silently
|
|
corrupted order instead of raising -> RED.
|
|
|
|
Refusal is the contract, not coercion-to-zero: a broken embedder must not be laundered into
|
|
"no semantic similarity" while ranking proceeds on the forged signal."""
|
|
query = ProposalFeatures(
|
|
affected_codes=frozenset({"05.2"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=200_000,
|
|
description="query prose",
|
|
)
|
|
candidates = [_verdict("A", "poisoned"), _verdict("B", "fine"), _verdict("C", "also fine")]
|
|
ranker = HybridRanker(_NaNEmbedder("poisoned"), similarity)
|
|
|
|
try:
|
|
ranked = ranker.rank(query, candidates, k=3)
|
|
except ValueError as exc:
|
|
assert "non-finite" in str(exc)
|
|
else:
|
|
raise AssertionError(
|
|
"rank() returned a NaN-contaminated ordering instead of refusing: "
|
|
f"{[v.id for v in ranked]}"
|
|
)
|
|
|
|
|
|
def test_a_nan_sort_key_is_input_order_dependent() -> None:
|
|
"""CONTROL — green with or without the guard. Its job is to pin the PREMISE the guard rests
|
|
on, so the guard above cannot later be dismissed as defensive noise.
|
|
|
|
``HybridRanker``'s docstring promises ``id`` "makes the result independent of input order even
|
|
when two candidates score identically". Measured: a single NaN score breaks that promise —
|
|
NaN compares False against everything, so ``sorted`` leaves it wherever the input put it. Six
|
|
permutations of the same three candidates yielded FOUR distinct orderings."""
|
|
scored = [("A", float("nan")), ("B", 0.9), ("C", 0.5)]
|
|
orderings = {
|
|
tuple(item[0] for item in sorted(perm, key=lambda t: (-round(t[1], 9), t[0])))
|
|
for perm in itertools.permutations(scored)
|
|
}
|
|
assert len(orderings) > 1, (
|
|
"a NaN score no longer perturbs the ranking key — the premise behind "
|
|
"test_hybrid_rank_refuses_a_non_finite_embedding has changed, re-measure it"
|
|
)
|