FabledScribe/tests/test_stamps_to_review.py

"""The review surface: rows whose evidence no longer meets the bar, and
canons whose own rows no longer agree what they are (#4204, #4208).

WHAT THESE GUARD, and it is a property rather than an output shape. The
ledger was poisoned by a machine writing permanent classifications that
nobody read — 32 rows on one project, 334 on another. The correction is NOT
a machine that un-writes them; that is the same mistake pointed the other
way. It is a reader that puts the evidence in front of a judge.

So the tests below assert two things hardest:

  * a judgment is NEVER listed as weak, however old and whatever its score
    would have been — an agent's decision is a different kind of evidence,
    not a worse one;
  * the listing changes nothing, which is checked by giving it rows and
    asserting the rows come back untouched.

The round-trip test exists because the score used to live only inside a
prose sentence. Writer and parser are now one constant apart, and a change
to one that the other does not follow is exactly the kind of silent rot the
hook's own history is made of.
"""


import pytest

from scribe.services.shape_ledger import (
    _RESEMBLE_MIN, _RESEMBLE_REASON, stamp_score, stamps_to_review,
)


class _Row:
    """A ledger row, as the review reader touches one."""

    def __init__(self, *, path="src/a.py", symbol="f", kind="sym",
                 status="instance", snippet_id=1, reason=None,
                 classified_by="hook", signature="async def f():",
                 vanished_at=None, classified_at=None):
        self.path, self.symbol, self.kind = path, symbol, kind
        self.status, self.snippet_id, self.reason = status, snippet_id, reason
        self.classified_by, self.signature = classified_by, signature
        self.vanished_at, self.classified_at = vanished_at, classified_at


def _stamped(score, **kw):
    return _Row(reason=_RESEMBLE_REASON.format(sid=1, score=score), **kw)


# ── the writer and the parser are one constant apart ──────────────────────

@pytest.mark.parametrize("score", [0.0, 0.5, 0.69, 0.77, 0.8, 0.95, 1.0])
def test_every_score_the_hook_can_write_reads_back(score) -> None:
    written = _RESEMBLE_REASON.format(sid=4204, score=score)
    assert stamp_score(written) == pytest.approx(round(score, 2))


def test_the_real_reasons_from_the_poisoned_ledgers_parse() -> None:
    """Verbatim from Portal and Scribe rows, so the parser is pinned to text
    that actually exists in the database rather than to text it generates."""
    assert stamp_score("hook: pulled #3283; payload resembles it (0.69)") == 0.69
    assert stamp_score("hook: pulled #3283; payload resembles it (0.77)") == 0.77
    assert stamp_score("hook: pulled #2860; payload resembles it (0.68)") == 0.68


@pytest.mark.parametrize("reason", [
    None, "", "   ",
    "hook: pulled #3461; payload references `icon-btn`",   # the OTHER evidence
    "an agent wrote prose here",
    "hook: pulled #1; payload resembles it",               # no score
])
def test_anything_that_is_not_a_score_reads_as_no_score(reason) -> None:
    """None means "not written on a score", never "scored zero" — the two
    must not collapse, because one is a judgment and the other is weak."""
    assert stamp_score(reason) is None


# ── weak: only hook stamps, only below the floor ──────────────────────────

def test_a_stamp_below_the_floor_is_listed_with_its_evidence() -> None:
    out = stamps_to_review([_stamped(0.69)])
    assert out["weak_count"] == 1
    row = out["weak"][0]
    assert row["score"] == 0.69
    assert row["form"] == "async-fn"
    assert row["signature"] == "async def f():"
    assert out["floor"] == _RESEMBLE_MIN


def test_a_stamp_at_or_above_the_floor_is_not_listed() -> None:
    assert stamps_to_review([_stamped(_RESEMBLE_MIN)])["weak_count"] == 0
    assert stamps_to_review([_stamped(0.95)])["weak_count"] == 0


@pytest.mark.parametrize("by", ["agent", "audit", "import", "mechanical"])
def test_a_judgment_is_never_weak_however_it_would_have_scored(by) -> None:
    """The measured case: 302 of Scribe's 334 rows under one canon were
    `audit`-classified with no score at all. Listing those as weak would
    invite an agent to withdraw the only judgments in the ledger that were
    ever actually made by one."""
    rows = [_Row(classified_by=by, reason=None),
            _Row(classified_by=by,
                 reason=_RESEMBLE_REASON.format(sid=1, score=0.10))]
    assert stamps_to_review(rows)["weak_count"] == 0


def test_an_unclassified_or_vanished_row_is_not_listed() -> None:
    assert stamps_to_review([_stamped(0.1, status="unclassified")])["weak_count"] == 0
    assert stamps_to_review([_stamped(0.1, vanished_at="gone")])["weak_count"] == 0


def test_the_weakest_evidence_is_listed_first() -> None:
    out = stamps_to_review([
        _stamped(0.77, symbol="c"), _stamped(0.68, symbol="a"),
        _stamped(0.71, symbol="b"),
    ])
    assert [r["symbol"] for r in out["weak"]] == ["a", "b", "c"]


def test_top_bounds_the_listing_but_not_the_count() -> None:
    out = stamps_to_review([_stamped(0.7, symbol=f"s{i}") for i in range(9)], top=3)
    assert out["weak_count"] == 9 and len(out["weak"]) == 3


# ── incoherent: a canon whose own members disagree ────────────────────────

def _member(sig, sid=7, **kw):
    return _Row(snippet_id=sid, signature=sig, classified_by="audit", **kw)


def test_a_canon_whose_members_agree_is_not_listed() -> None:
    rows = [_member("async def a():"), _member("async def b():"),
            _member("async def c():"), _member("async def d():")]
    assert stamps_to_review(rows)["incoherent_count"] == 0


def test_a_canon_whose_members_are_all_different_things_is_listed() -> None:
    """Portal's #3283, in miniature: a class, a getter, a test and a binding
    recorded as instances of one pattern."""
    rows = [_member("class SessionAbsent(RuntimeError):"),
            _member("def build_channel() -> str:"),
            _member("async def attach(self) -> None:"),
            _member("MAX = 10")]
    out = stamps_to_review(rows)
    assert out["incoherent_count"] == 1
    entry = out["incoherent"][0]
    assert entry["snippet_id"] == 7 and entry["judged"] == 4
    assert set(entry["forms"]) == {"type", "fn", "async-fn", "binding"}
    assert len(entry["sample"]) == 4


def test_too_few_readable_members_says_nothing_either_way() -> None:
    """Two rows that disagree is not an incoherent canon, it is two rows.
    The floor is the same one the density check uses."""
    rows = [_member("class A:"), _member("def b():")]
    assert stamps_to_review(rows)["incoherent_count"] == 0


def test_unreadable_signatures_do_not_manufacture_incoherence() -> None:
    rows = [_member(""), _member("   "), _member("???"), _member("")]
    assert stamps_to_review(rows)["incoherent_count"] == 0


def test_a_canon_carrying_weak_stamps_is_ranked_above_one_that_is_not() -> None:
    clean = [_member("class A:", sid=1), _member("def b():", sid=1),
             _member("MAX = 1", sid=1), _member("async def c():", sid=1)]
    dirty = [_Row(snippet_id=2, signature=s, classified_by="hook",
                  reason=_RESEMBLE_REASON.format(sid=2, score=0.70))
             for s in ("class D:", "def e():", "MAX = 2", "async def f():")]
    out = stamps_to_review(clean + dirty)
    assert out["incoherent_count"] == 2
    assert out["incoherent"][0]["snippet_id"] == 2
    assert out["incoherent"][0]["weak_rows"] == 4
    assert out["incoherent"][1]["weak_rows"] == 0


# ── the property that matters most: it is a reader ────────────────────────

def test_the_review_changes_nothing_about_the_rows_it_reads() -> None:
    """Asserted rather than assumed. The whole design claim of this surface
    is that it reports and never acts; a future refactor that "helpfully"
    resets a status here would pass every other test in this file."""
    rows = [_stamped(0.69), _member("class A:"), _member("def b():"),
            _member("MAX = 1"), _member("async def c():")]
    before = [(r.status, r.snippet_id, r.classified_by, r.reason) for r in rows]
    stamps_to_review(rows)
    after = [(r.status, r.snippet_id, r.classified_by, r.reason) for r in rows]
    assert before == after


def test_the_service_carries_no_machinery_for_bulk_withdrawal() -> None:
    """A structural guard (rule 167) on the design decision, not on output.
    If someone adds an auto-retire path to the ledger, this fails and the
    conversation about whether a machine may un-judge in bulk happens again
    — deliberately, rather than in a diff nobody read."""
    import scribe.services.shape_ledger as ledger

    body = open(ledger.__file__).read()
    for banned in ("def retire_weak", "def auto_unclassify", "def bulk_withdraw"):
        assert banned not in body, banned