FabledScribe/tests/test_divergence_meaning_gate.py

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

WHAT THIS IS ABOUT. #4204 gave the divergence check a structural gate, which
silenced one of the five false prompts it was filed for. The other four are
`def` helpers beside an `async def` service canon — callables beside a
callable — and differ from #2793'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.

#4208 tried meaning: store the proposer's semantic "compared, nothing cleared
the floor" and let it withdraw the prompt. Measured live on 2026-09-22 it
cannot discriminate. Judged instances of the 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. The then-floor of 0.8 was reached by nothing, so
the "miss" fired for nearly every body — the real divergence silenced exactly
as the helper was. The gate was removed; the prompt asks and the judge
answers. The same measurement moved the arm's floor down to the write-path
hint's: 0.8 was sized for proposals nobody reads, and it proposed 0 of 150.

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 scribe.services import shape_ledger
from scribe.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. One of
# #4204's four survivors, 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("scribe.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 (#2871) 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 (#4208)."""
    from scribe.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 (rule 167): 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 (#4208), 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 scribe.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"]