FabledScribe/tests/test_divergence_meaning_gate.py

"""A divergence prompt may be silenced by MEANING, never by silence (#4208).

WHAT THIS IS ABOUT. #4204 gave the divergence check a structural gate: a canon
is only urged on a shape whose form could plausibly BE it. That silenced one of
the five false prompts it was filed for. The other four are `def` helpers in a
directory whose canon is an `async def` service unit — callables beside a
callable — and no refinement of `shape_form` ever separates them, because they
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.

So the lever has to be meaning, and the ledger already holds one reading of it:
the proposer's semantic arm embeds each definition's own BODY against canon.
What it did not do was record its misses. A hit became `proposal_basis =
"semantic"`; a miss left the row indistinguishable from a row nobody had looked
at yet. `flag_divergence` could therefore ask the proposer "do you agree this is
the canon?" but never "did you check, and did you find it is not?".

THE WHOLE RISK IS IN THE NEGATIVE. A miss is only evidence if the arm actually
formed an opinion, and there are three ways for it to come back empty that look
identical from the outside:

    body too thin to embed   -> no opinion
    no allowed canon to test -> no opinion
    result set was truncated -> no opinion (the canon may be at N+1)
    compared, nothing above the floor -> EVIDENCE

Only the last may silence a prompt. Reading any of the others as a negative is
how "I cannot tell" turns into "I checked" — the exact failure #4204 was opened
on, and the one `FORM_UNKNOWN` already guards against everywhere else in this
module: not knowing must make a check QUIETER, never more confident.

These tests pin the report contract that carries that distinction. The
end-to-end behaviour — a conclusive miss silencing a real prompt while #2793's
acceptance case still raises — is in
tests/test_integration_shape_classify.py, because it needs real rows.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, patch

import pytest

from scribe.services.shape_ledger import (
    _SEMANTIC_LIMIT, BASIS_NO_SEMANTIC_MATCH, _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 miss that IS evidence ────────────────────────────────────────────


async def test_a_short_result_set_is_a_conclusive_miss() -> None:
    """Fewer hits than asked for means the limit was not binding: everything
    above the floor came back, and the canon was not among it. That is a fact
    about the corpus, not an artefact of where the list was cut."""
    mock = _hits((0.91, OTHER))
    report: dict = {}
    with _patch(mock):
        found = await _semantic_canon(1, BODY, {CANON}, report=report)
    assert found is None
    assert report.get("conclusive") is True


async def test_an_empty_result_set_is_also_conclusive() -> None:
    """Nothing cleared the floor at all — the strongest form of the miss."""
    report: dict = {}
    with _patch(_hits()):
        assert await _semantic_canon(1, BODY, {CANON}, report=report) is None
    assert report.get("conclusive") is True


# ── the three misses that are NOT ────────────────────────────────────────


async def test_a_full_result_set_may_have_been_truncated() -> None:
    """The case that makes `_SEMANTIC_LIMIT` load-bearing rather than a tuning
    knob. The search returns the top N above the floor; when it returns
    exactly N, an allowed canon can be sitting at N+1 and this same silence
    would mean nothing. Reading it as a negative would silence real
    divergences in direct proportion to how many snippets the operator has —
    a check that quietly weakens as the corpus grows, which is the worst
    possible failure mode for a guard nobody is watching."""
    mock = _hits(*[(0.9, OTHER + i) for i in range(_SEMANTIC_LIMIT)])
    report: dict = {}
    with _patch(mock):
        assert await _semantic_canon(1, BODY, {CANON}, report=report) is None
    assert "conclusive" not in report


async def test_a_body_too_thin_to_embed_forms_no_opinion() -> None:
    """And does not spend an embedding finding that out."""
    mock = _hits()
    report: dict = {}
    with _patch(mock):
        assert await _semantic_canon(1, TOO_THIN, {CANON}, report=report) is None
    assert "conclusive" not in report
    mock.assert_not_awaited()


async def test_no_allowed_canon_means_nothing_was_compared() -> None:
    """An empty allowed set is not "the canons all missed" — there were none
    to miss. Distinct because the language-family gate (#2871) empties this
    set routinely: a Vue body simply has no Python canon to be compared to."""
    mock = _hits()
    report: dict = {}
    with _patch(mock):
        assert await _semantic_canon(1, BODY, set(), report=report) is None
    assert "conclusive" not in report
    mock.assert_not_awaited()


# ── a hit is a proposal, not a miss ──────────────────────────────────────


async def test_a_hit_returns_the_canon_and_claims_no_miss() -> None:
    mock = _hits((0.88, CANON))
    report: dict = {}
    with _patch(mock):
        found = await _semantic_canon(1, BODY, {CANON}, report=report)
    assert found == (CANON, 0.88)
    assert "conclusive" not in report


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. Pinned because if it did,
    the short-list case above would start reporting conclusive misses for
    bodies that DO have a canon."""
    mock = _hits((0.95, OTHER), (0.83, CANON))
    report: dict = {}
    with _patch(mock):
        found = await _semantic_canon(1, BODY, {CANON}, report=report)
    assert found == (CANON, 0.83)
    assert "conclusive" not in report


# ── the contract callers depend on ───────────────────────────────────────


async def test_a_caller_that_passes_no_report_still_gets_an_answer() -> None:
    """The existing test double is an `AsyncMock(return_value=None)` that
    never touches a report. Absence of the key must therefore mean "cannot
    tell" at every call site — so a stub, an older caller, or an arm that
    threw all default to asking the question rather than excusing it."""
    with _patch(_hits()):
        assert await _semantic_canon(1, BODY, {CANON}) is None


@pytest.mark.parametrize("value", ["semantic", "symbol", "reference", "derive"])
def test_the_miss_basis_is_not_one_of_the_proposal_bases(value: str) -> None:
    """It shares a column with them and must not collide: every reader keys
    "is there a proposal" on `proposed_snippet_id`, but `confirm_shape_proposals`
    filters BY basis, and a collision there would mean confirming a miss as
    though it were a match."""
    assert BASIS_NO_SEMANTIC_MATCH != value


# ── the cap spends itself on the rows its verdict can act on ─────────────


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 the gate shipped: a version
    bump queued 1,533 rows, 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 reached it. The
    gate silenced nothing because it never got to look."""
    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"]