portfolio-optimiser/tests/test_semretrieval.py

"""S3.1 — semantic retrieval unit tests (SC4 here; SC1/SC3/SC7 added by later steps).

The embedder must be a PURE function of the structured features: identical features yield a
byte-identical float64 vector in this process AND in a fresh process started with a different
``PYTHONHASHSEED`` — the cross-process leg is what rules out Python's salted builtin ``hash()``
(the exact failure mode that would make retrieval order drift between runs).

Components are non-negative, so ``cosine`` stays in ``[0, 1]`` and blends with the structural
score (also ``[0, 1]``) without a scale mismatch.

Pattern: tests/test_verdicts.py (determinism) + tests/test_retrieval.py (fixtures).
"""

from __future__ import annotations

import json
import subprocess
import sys
import types
from pathlib import Path

import numpy as np
import pytest
from pydantic import ValidationError

from portfolio_optimiser.semretrieval import (
    EMBED_DIM,
    Embedder,
    EmbedderConfig,
    FakeEmbedder,
    HybridRanker,
    StructuralRetriever,
    build_embedder,
    cosine,
    load_embedder_config,
    load_vector_store,
    save_vector_store,
)
from portfolio_optimiser.verdicts import (
    ProposalFeatures,
    Verdict,
    VerdictStore,
    similarity,
)

_SRC = Path(__file__).resolve().parents[1] / "src"


def _features(
    codes: frozenset[str] = frozenset({"05.2", "03.1"}),
    measure_type: str = "scope_reduction",
    saving: float = 220_000.0,
    description: str = "asphalt base course reduction near school",
) -> types.SimpleNamespace:
    """Duck-typed stand-in for ``ProposalFeatures`` — semretrieval never imports ``verdicts``
    at runtime, so the unit tests do not need the real (MAF-bound) type either."""
    return types.SimpleNamespace(
        affected_codes=codes,
        measure_type=measure_type,
        claimed_saving_nok=saving,
        description=description,
    )


def test_fake_embedder_is_pure() -> None:
    embedder = FakeEmbedder()
    assert np.array_equal(embedder(_features()), embedder(_features()))


def test_fake_embedder_has_fixed_float64_dimension() -> None:
    vec = FakeEmbedder()(_features())
    assert vec.shape == (EMBED_DIM,)
    assert vec.dtype == np.dtype("<f8")


def test_fake_embedder_is_insensitive_to_code_set_ordering() -> None:
    """``affected_codes`` is a set — the canonical string sorts it, so insertion order
    cannot leak into the vector."""
    a = FakeEmbedder()(_features(codes=frozenset({"05.2", "03.1"})))
    b = FakeEmbedder()(_features(codes=frozenset({"03.1", "05.2"})))
    assert np.array_equal(a, b)


def test_distinct_features_give_distinct_vectors() -> None:
    """SC4, NARROWED: "distinct" means distinct in the EMBEDDED fields — the structural triple.
    The description-only variant moved to the test below, which asserts the opposite."""
    embedder = FakeEmbedder()
    baseline = embedder(_features())
    for variant in (
        _features(codes=frozenset({"09.1"})),
        _features(measure_type="rate_renegotiation"),
        _features(saving=900_000.0),  # different magnitude bucket
    ):
        assert not np.array_equal(baseline, embedder(variant))


def test_features_differing_only_in_description_embed_identically() -> None:
    """SC4's narrowing, asserted POSITIVELY rather than left implicit — this is the same
    assertion the loop above used to carry, with the opposite sign.

    ``description`` is excluded from the embedding on purpose: ``verdicts.similarity`` already
    ignores text by design and ``verdicts._mint_id`` already hashes only the structural triple, so
    two proposals differing only in prose are ONE verdict as far as the rest of the system is
    concerned. Leaving the embedding as the sole place surface text still counted made the
    framework's own restatement of a query outrank real expert prose (defect P1).

    Detach point: put ``features.description`` back into ``_canonical_feature_string`` → RED."""
    embedder = FakeEmbedder()
    assert np.array_equal(
        embedder(_features()),
        embedder(_features(description="totally different wording")),
    )


def test_fake_embedder_components_are_non_negative() -> None:
    """Non-negative components keep ``cosine`` in [0, 1], matching the structural score's
    range — no scale mismatch when the two are blended by HybridRanker."""
    vec = FakeEmbedder()(_features())
    assert bool((vec >= 0.0).all())


def test_fake_embedder_is_bit_stable_across_processes(tmp_path: Path) -> None:
    """RED for any ``hash()``-based implementation: the child runs with a different
    ``PYTHONHASHSEED``, so a salted hash would produce different bytes."""
    embedder = FakeEmbedder()
    here = tmp_path / "here.npy"
    np.save(here, embedder(_features()))

    there = tmp_path / "there.npy"
    script = (
        "import types, numpy as np\n"
        "from portfolio_optimiser.semretrieval import FakeEmbedder\n"
        "f = types.SimpleNamespace(\n"
        "    affected_codes=frozenset({'05.2', '03.1'}),\n"
        "    measure_type='scope_reduction',\n"
        "    claimed_saving_nok=220000.0,\n"
        "    description='asphalt base course reduction near school',\n"
        ")\n"
        f"np.save({str(there)!r}, FakeEmbedder()(f))\n"
    )
    result = subprocess.run(
        [sys.executable, "-c", script],
        env={"PATH": "/usr/bin:/bin", "PYTHONPATH": str(_SRC), "PYTHONHASHSEED": "12345"},
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0, result.stderr
    assert here.read_bytes() == there.read_bytes()


def test_cosine_of_identical_unit_vectors_is_one() -> None:
    """Approximate by necessity: a float64 dot of a normalized vector with itself lands within
    an ulp or two of 1.0, never exactly on it. Ranking never depends on the exact value —
    ``HybridRanker`` rounds to 9 decimals and breaks remaining ties on ``id``."""
    vec = FakeEmbedder()(_features())
    assert cosine(vec, vec) == pytest.approx(1.0)


def test_cosine_is_bounded_to_unit_interval() -> None:
    embedder = FakeEmbedder()
    score = cosine(embedder(_features()), embedder(_features(codes=frozenset({"09.1"}))))
    assert 0.0 <= score <= 1.0


def test_cosine_of_zero_norm_is_zero() -> None:
    """A zero vector must never produce NaN — NaN in a sort key silently corrupts ranking."""
    zeros = np.zeros(EMBED_DIM, dtype="<f8")
    vec = FakeEmbedder()(_features())
    assert cosine(zeros, vec) == 0.0
    assert cosine(vec, zeros) == 0.0
    assert cosine(zeros, zeros) == 0.0


@pytest.mark.parametrize("bad_value", [np.nan, np.inf, -np.inf])
def test_cosine_refuses_a_non_finite_vector(bad_value: float) -> None:
    """The zero-norm guard tests ``norm == 0.0``, which a NaN or inf norm passes straight through:
    measured, ``cosine(unit, nan_vector)`` and ``cosine(unit, inf_vector)`` BOTH returned ``nan``.

    Refuse rather than repair, and deliberately NOT symmetric with the zero-norm branch above: a
    zero vector is a legitimate, handled state (``FakeEmbedder`` returns ``np.zeros`` by design),
    whereas a non-finite component only ever means the injected embedder is broken. Coercing it to
    ``0.0`` would launder that into "no semantic similarity" and let ranking proceed on a forged
    signal — the same reasoning that makes ``read_spend`` raise on corrupt content instead of
    reading it as zero."""
    vec = FakeEmbedder()(_features())
    bad = np.full(EMBED_DIM, bad_value, dtype="<f8")
    with pytest.raises(ValueError, match="non-finite"):
        cosine(vec, bad)
    with pytest.raises(ValueError, match="non-finite"):
        cosine(bad, vec)


# --- SC1: the Retriever seam is additive — the DEFAULT ranking is byte-identical to today ---

_QUERY = ProposalFeatures(
    affected_codes=frozenset({"05.2", "03.1"}),
    measure_type="scope_reduction",
    claimed_saving_nok=220_000,  # bucket [100k, 500k)
    description="asphalt base course reduction near school",
)


def _store_with_true_match_and_decoys() -> tuple[VerdictStore, str]:
    """Mirrors ``tests/test_verdicts.py::_store_with_true_match_and_decoys`` — the true match
    shares the STRUCTURED fields with the query but uses different wording; the decoys share
    surface text but differ structurally."""
    true_match = Verdict(
        id="TRUE",
        proposal_features=ProposalFeatures(
            affected_codes=frozenset({"05.2", "03.1"}),
            measure_type="scope_reduction",
            claimed_saving_nok=200_000,
            description="zzz totally unrelated wording alpha beta",
        ),
        decision="approved",
        rationale="prior scope reduction on the same codes was approved",
    )
    decoy_low = Verdict(
        id="DECOY-LOW",
        proposal_features=ProposalFeatures(
            affected_codes=frozenset({"09.1"}),
            measure_type="rate_renegotiation",
            claimed_saving_nok=50_000,
            description="asphalt base course reduction near school",
        ),
        decision="rejected",
        rationale="surface-text decoy",
    )
    decoy_high = Verdict(
        id="DECOY-HIGH",
        proposal_features=ProposalFeatures(
            affected_codes=frozenset({"21.2"}),
            measure_type="material_substitution",
            claimed_saving_nok=700_000,
            description="asphalt base course reduction extra words",
        ),
        decision="rejected",
        rationale="surface-text decoy",
    )
    return VerdictStore(verdicts=[decoy_low, true_match, decoy_high]), "TRUE"


def test_default_retriever_is_none() -> None:
    """Opt-in seam: a store built the ordinary way carries no retriever, so nothing about
    default retrieval changes unless a caller deliberately installs one."""
    store, _ = _store_with_true_match_and_decoys()
    assert store.retriever is None


def test_default_retrieve_matches_the_pre_seam_structural_sort() -> None:
    """SC1 — with no retriever installed, ``retrieve`` must reproduce the exact ordering the
    pre-seam inline sort produced: key ``(-similarity, id)``, sliced to k."""
    store, _ = _store_with_true_match_and_decoys()
    expected = [
        v.id
        for v in sorted(
            store.verdicts,
            key=lambda v: (-similarity(_QUERY, v.proposal_features), v.id),
        )[:3]
    ]
    assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected


def test_structural_retriever_is_the_wired_default_not_dead_code() -> None:
    """SC1 — the default path goes through ``StructuralRetriever``, so it is exercised code,
    not an unused parallel implementation that can silently rot."""
    store, _ = _store_with_true_match_and_decoys()
    direct = StructuralRetriever(similarity).rank(_QUERY, store.verdicts, 3)
    assert [h.id for h in store.retrieve(_QUERY, k=3)] == [v.id for v in direct]


def test_retrieve_still_rejects_non_positive_k() -> None:
    """The pre-seam ``k <= 0`` guard must survive the delegation rewrite
    (asserted independently by tests/test_verdicts.py::test_retrieve_rejects_non_positive_k)."""
    store, _ = _store_with_true_match_and_decoys()
    with pytest.raises(ValueError):
        store.retrieve(_QUERY, k=0)


def test_installed_retriever_actually_routes_retrieval() -> None:
    """Delegation proof: a sentinel retriever must receive the call and its result must be
    what ``retrieve`` returns — otherwise the seam is decorative."""
    store, _ = _store_with_true_match_and_decoys()
    calls: list[tuple[ProposalFeatures, int, int]] = []

    class _Sentinel:
        def rank(self, query: ProposalFeatures, candidates: list[Verdict], k: int) -> list[Verdict]:
            calls.append((query, len(candidates), k))
            return list(reversed(candidates))[:k]

    store.retriever = _Sentinel()
    hits = store.retrieve(_QUERY, k=2)
    assert calls == [(_QUERY, 3, 2)]
    assert [h.id for h in hits] == ["DECOY-HIGH", "TRUE"]


def test_hybrid_ranker_is_injectable_and_routes_retrieval() -> None:
    """A HybridRanker installed on the store must be the ranker ``retrieve`` delegates to."""
    store, _ = _store_with_true_match_and_decoys()
    ranker = HybridRanker(FakeEmbedder(), similarity)
    expected = [v.id for v in ranker.rank(_QUERY, store.verdicts, 3)]
    store.retriever = ranker
    assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected


def test_hybrid_ranker_at_weight_zero_equals_the_structural_ranking() -> None:
    """Weight 0 removes the semantic term entirely — the blend must then collapse onto the
    structural ordering. This is the control the Step-5 detach test relies on."""
    store, _ = _store_with_true_match_and_decoys()
    hybrid = HybridRanker(FakeEmbedder(), similarity, weight=0.0)
    structural = StructuralRetriever(similarity)
    assert [v.id for v in hybrid.rank(_QUERY, store.verdicts, 3)] == [
        v.id for v in structural.rank(_QUERY, store.verdicts, 3)
    ]


# --- The embedder config seam: a CLOSED registry, never an import path -------------------------


def test_build_embedder_returns_the_shipped_fake_for_the_known_type() -> None:
    """The registry's one shipped entry. Mirrors ``tests/test_notify.py``'s build_notifier tests."""
    built = build_embedder(EmbedderConfig(type="fake"))
    assert isinstance(built, FakeEmbedder)
    assert isinstance(built, Embedder)  # satisfies the runtime-checkable protocol


def test_build_embedder_rejects_an_unknown_type() -> None:
    """Closed vocabulary: an unknown type RAISES rather than falling back to a default, and the
    message names the seam so the constraint is discoverable at the failure."""
    with pytest.raises(ValueError, match="unknown embedder type"):
        build_embedder(EmbedderConfig(type="nope"))


def test_build_embedder_refuses_an_import_path_instead_of_importing_it() -> None:
    """SECURITY — the registry must never resolve a dotted ``module:Class`` string from config.

    Doing so would be arbitrary code execution at config-load time and would hand a config file the
    ability to import a module that opens a socket — routing around this module's no-network guard,
    which is scoped to ``semretrieval.py`` and blind to a third module by construction. The config
    is accepted as DATA and then refused; nothing is imported."""
    hostile = EmbedderConfig(type="evil_pkg.embedders:NetworkEmbedder")
    with pytest.raises(ValueError, match="unknown embedder type"):
        build_embedder(hostile)
    assert "evil_pkg" not in sys.modules


def test_embedder_config_rejects_an_empty_type() -> None:
    with pytest.raises(ValidationError):
        EmbedderConfig(type="")


def test_load_embedder_config_is_fail_fast_on_a_missing_file(tmp_path: Path) -> None:
    """Authoritative startup config — absence RAISES, in deliberate contrast to the tolerant
    verdict-inbox RAW layer which skips bad files."""
    with pytest.raises(FileNotFoundError):
        load_embedder_config(tmp_path / "nope.json")


def test_load_embedder_config_is_fail_fast_on_a_malformed_file(tmp_path: Path) -> None:
    bad = tmp_path / "embedder.json"
    bad.write_text('{"type": 17}', encoding="utf-8")
    with pytest.raises(ValidationError):
        load_embedder_config(bad)


def test_load_embedder_config_round_trips_a_valid_file(tmp_path: Path) -> None:
    good = tmp_path / "embedder.json"
    good.write_text('{"type": "fake"}', encoding="utf-8")
    assert isinstance(build_embedder(load_embedder_config(good)), FakeEmbedder)


# --- SC3 / SC7: the vector store is byte-deterministic and fails fast on a row/line mismatch ---


def test_vector_store_is_byte_identical_regardless_of_insertion_order(tmp_path: Path) -> None:
    """SC3 — the store sorts by id before embedding, so the same verdicts written in a different
    order must produce byte-identical artifacts. Pattern:
    tests/test_ledger.py::test_save_is_byte_identical_regardless_of_order."""
    store, _ = _store_with_true_match_and_decoys()
    forward, reverse = tmp_path / "forward", tmp_path / "reverse"

    save_vector_store(forward, store.verdicts, FakeEmbedder())
    save_vector_store(reverse, list(reversed(store.verdicts)), FakeEmbedder())

    assert (forward / "vectors.npy").read_bytes() == (reverse / "vectors.npy").read_bytes()
    assert (forward / "vectors.jsonl").read_bytes() == (reverse / "vectors.jsonl").read_bytes()


def test_vector_store_round_trips_ids_sorted(tmp_path: Path) -> None:
    store, _ = _store_with_true_match_and_decoys()
    save_vector_store(tmp_path, store.verdicts, FakeEmbedder())

    loaded = load_vector_store(tmp_path)
    assert loaded is not None
    ids, matrix = loaded
    assert ids == sorted(v.id for v in store.verdicts)
    assert matrix.shape == (len(ids), EMBED_DIM)
    assert matrix.dtype == np.dtype("<f8")

    embedder = FakeEmbedder()
    by_id = {v.id: v for v in store.verdicts}
    for row, verdict_id in enumerate(ids):
        assert np.array_equal(matrix[row], embedder(by_id[verdict_id].proposal_features))


def test_vector_store_jsonl_is_one_sorted_object_per_line(tmp_path: Path) -> None:
    """The id sidecar follows the repo's deterministic on-disk idiom: sorted keys, LF endings,
    trailing newline — so a diff of two runs is empty rather than noisy."""
    store, _ = _store_with_true_match_and_decoys()
    save_vector_store(tmp_path, store.verdicts, FakeEmbedder())

    raw = (tmp_path / "vectors.jsonl").read_bytes().decode("utf-8")
    assert raw.endswith("\n")
    assert "\r" not in raw
    assert [json.loads(line) for line in raw.splitlines()] == [
        {"id": v.id} for v in sorted(store.verdicts, key=lambda v: v.id)
    ]


def test_ranking_is_stable_over_twenty_repetitions() -> None:
    """The NFR asks for >=20 repetitions, not a single re-run: BLAS reduction order is the thing
    that would make this drift, and it is pinned at module import."""
    base, _ = _store_with_true_match_and_decoys()
    ranker = HybridRanker(FakeEmbedder(), similarity)
    first = [v.id for v in ranker.rank(_QUERY, base.verdicts, 3)]
    for _ in range(20):
        assert [v.id for v in ranker.rank(_QUERY, base.verdicts, 3)] == first


def test_missing_vector_store_loads_as_none(tmp_path: Path) -> None:
    """Tolerant on absence — the store is a rebuildable cache, so a caller with no store simply
    degrades to structural ranking rather than crashing."""
    assert load_vector_store(tmp_path / "never-written") is None
    assert load_vector_store(tmp_path) is None  # directory exists but holds no artifacts


def test_vector_store_row_line_mismatch_fails_fast(tmp_path: Path) -> None:
    """SC7 — a truncated sidecar would silently mis-map every vector to the wrong verdict id.
    That must raise, not rank garbage (mirrors SavingsLedger.load's fail-fast posture)."""
    store, _ = _store_with_true_match_and_decoys()
    save_vector_store(tmp_path, store.verdicts, FakeEmbedder())

    sidecar = tmp_path / "vectors.jsonl"
    kept = sidecar.read_text(encoding="utf-8").splitlines()[:-1]
    sidecar.write_text("".join(line + "\n" for line in kept), encoding="utf-8")

    with pytest.raises(ValueError, match="row"):
        load_vector_store(tmp_path)