FabledGrimoire/tests/test_divergence_meaning_gate.py

"""A semantic MISS is not evidence, and the divergence check never reads one.

WHAT THIS IS ABOUT. The divergence check's structural gate separates a type
from a callable canon. `def` helpers beside an `async def` service canon —
callables beside a callable — differ from the readout's acceptance case (a
hand-rolled sync `confirmDanger` where an async confirm helper is canon) only
by the JOB they do. A signature does not carry a job.

Meaning cannot withdraw the prompt either. Judged instances of a service-unit
canon score 0.68–0.71 against it at best, most below 0.66; helpers score
0.66–0.75 against snippets unrelated to them. No floor separates the two, so a
"compared, nothing cleared the floor" miss fires for nearly every body,
silencing the real divergence exactly as it silences the helper. So the prompt
asks and the judge answers, and the arm proposes at the write-path hint's
floor: one sized for proposals nobody reads proposes almost nothing.

These tests pin what stayed: the arm proposes on a hit and says nothing on a
miss, the divergence check does not consult it, and the capped pass reads new
shapes first so the proposals it CAN make land where the check looks.
"""
from __future__ import annotations

import ast
import inspect
import textwrap
from unittest.mock import AsyncMock, patch

from grimoire.services import shape_ledger
from grimoire.services.shape_ledger import _semantic_canon

# Comfortably over WRITEPATH_MIN_CODE_CHARS (48 non-whitespace characters), so
# these tests exercise the comparison rather than the substance guard. A real
# helper body, quoted rather than invented.
BODY = (
    "def is_registered(source: str) -> bool:\n"
    "    return source in _REGISTRY and _REGISTRY[source].enabled\n"
)
TOO_THIN = "def f():\n    pass\n"

CANON = 2860          # the allowed canon, as a caller would pass it
OTHER = 9999          # a snippet that is not in the allowed set


class _FakeNote:
    """Only `.id` is read off a hit."""

    def __init__(self, note_id: int) -> None:
        self.id = note_id


def _hits(*hits: tuple[float, int]) -> AsyncMock:
    return AsyncMock(return_value=[(score, _FakeNote(nid)) for score, nid in hits])


def _patch(mock: AsyncMock):
    return patch("grimoire.services.embeddings.semantic_search_notes", mock)


# ── the arm proposes on a hit ────────────────────────────────────────────


async def test_a_hit_returns_the_canon() -> None:
    with _patch(_hits((0.88, CANON))):
        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.88, "semantic")


async def test_an_allowed_canon_below_the_top_hit_still_wins() -> None:
    """The scan is over the whole result set, so a disallowed snippet ranking
    first does not hide an allowed one behind it."""
    with _patch(_hits((0.95, OTHER), (0.83, CANON))):
        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.83, "semantic")


async def test_a_miss_is_none_and_nothing_more() -> None:
    with _patch(_hits((0.91, OTHER))):
        assert await _semantic_canon(1, BODY, {CANON}) is None


async def test_a_body_too_thin_to_embed_spends_no_embedding() -> None:
    mock = _hits()
    with _patch(mock):
        assert await _semantic_canon(1, TOO_THIN, {CANON}) is None
    mock.assert_not_awaited()


async def test_no_allowed_canon_spends_no_search() -> None:
    """The language-family gate empties the allowed set routinely."""
    mock = _hits()
    with _patch(mock):
        assert await _semantic_canon(1, BODY, set()) is None
    mock.assert_not_awaited()


async def test_the_arm_proposes_at_the_write_path_floor() -> None:
    """It asks the write-path hint's question of the same documents, so it
    uses that surface's floor — not a stricter private one that yields no
    proposals for the judge to weigh."""
    from grimoire.services.plugin_context import WRITEPATH_DEFAULT_THRESHOLD

    mock = _hits()
    with _patch(mock):
        await _semantic_canon(1, BODY, {CANON})
    assert mock.await_args.kwargs["threshold"] == WRITEPATH_DEFAULT_THRESHOLD


# ── ...and its silence reaches nothing ───────────────────────────────────


def test_the_divergence_check_does_not_read_the_proposal_basis() -> None:
    """The retired gate keyed on `proposal_basis`. Asserted on the function's
    structure: any attribute read of it inside `flag_divergence` is
    the miss being consulted again, whatever the constant is called."""
    tree = ast.parse(textwrap.dedent(inspect.getsource(shape_ledger.flag_divergence)))
    reads = [n for n in ast.walk(tree)
             if isinstance(n, ast.Attribute) and n.attr == "proposal_basis"]
    assert not reads, (
        "flag_divergence reads proposal_basis — a semantic miss is 'cannot "
        "tell' at the arm's floor, never 'not the canon'"
    )


def test_no_miss_basis_is_stored() -> None:
    """A miss basis sharing the column with real bases is a thing
    `confirm_shape_proposals(basis=…)` could filter on and confirm."""
    assert not hasattr(shape_ledger, "BASIS_NO_SEMANTIC_MATCH")


# ── the cap spends itself where proposals matter ─────────────────────────


class _AgedRow:
    def __init__(self, name: str, status: str, created) -> None:
        self.name, self.status, self.created_at = name, status, created


def test_the_capped_pass_reads_new_shapes_first() -> None:
    """Measured on the first live refresh after a version bump: 1,533 rows
    queued, the cap read 150 of them in row order, and every shape NEW since
    the previous refresh — the only rows `flag_divergence` acts on, and the
    highest ids — was flagged before the arm could propose for it."""
    from datetime import datetime, timedelta, timezone

    from grimoire.services.shape_ledger import _semantic_priority

    t = datetime(2026, 9, 22, tzinfo=timezone.utc)
    rows = [
        _AgedRow("scoped_new", "scoped", t),
        _AgedRow("old", "unclassified", t - timedelta(days=30)),
        _AgedRow("undated", "unclassified", None),
        _AgedRow("new", "unclassified", t),
        _AgedRow("scoped_old", "scoped", t - timedelta(days=30)),
    ]
    order = [r.name for r in sorted(rows, key=_semantic_priority)]
    assert order == ["new", "old", "undated", "scoped_new", "scoped_old"]