feat(snippets): drift check — verify a snippet still matches its source

A recorded snippet points at a repo · path · symbol that WILL rot: files
move, symbols get renamed, implementations diverge from the copy stored
here. Nothing detected any of it, so a record degraded silently from
"canonical reference" to "confidently wrong" — worse than no record, since
it is surfaced with the same authority either way.

WHERE THE CHECK RUNS. Agent-side, which the task flagged as the design
question to settle first. Scribe has no checkout of the operator's repos
and must not acquire one: giving the server repo access would make every
install a credential problem and break instance-agnosticism (rule #115).
The agent already has the working tree, so it does the comparing; the
server remembers the verdict, makes it queryable, and knows when it has
expired. New MCP tool verify_snippet teaches the four-step procedure and
records the result; a REST endpoint mirrors it so the UI can clear a
marker after a manual fix.

WHY THE VERDICT CARRIES A CODE HASH. A verdict describes the code it was
checked against. Invalidating it on edit means deciding which edits count
— a when_to_use tweak shouldn't void a code check, a rewrite must — which
is fiddly and easy to get subtly wrong, and easy for a new write path to
forget entirely. Stamping the verdict with a hash sidesteps all of it: one
whose code_sha no longer matches is self-evidently expired, computed at
read time, no invalidation branch to maintain.

That makes "expired" the interesting filter case. It is not `drifted`
(nothing was found wrong) and not `unverified` (a check did happen), yet
it plainly needs looking at — so `verification=attention` covers both. To
keep that one index-served predicate rather than a post-filter that would
make the pagination total a lie, data now also mirrors the CURRENT code's
fingerprint as data.code_sha, and a jsonpath compares the two fields
within the row. The filter is implemented in both dialects, SQL and
Python, for the same reason the location filter is: the semantic arm's
candidates arrive already fetched.

A merge deliberately carries no verdict forward — the survivor's code is a
union of several sources, so no prior check describes it, and unverified
is the honest answer.

UI: a danger-toned drift badge on each card (an actively misleading record
outranks a merely unused one), and a "Needs attention" filter. Its empty
state says plainly that never-verified snippets don't appear there —
otherwise `attention` would mean "everything" on day one and be useless as
a worklist.

Refs #2086

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

tests/test_snippet_drift_check.py

@@ -0,0 +1,251 @@
+"""Tests for the drift check (#2086) — does a snippet still match its source?
+
+The check itself runs agent-side (Scribe has no checkout), so what's testable
+here is everything around it: the verdict's shape, the rule that a verdict
+expires when the code it blessed is edited away, and the two dialects of the
+filter predicate — SQL for the browse/keyword arms, Python for the semantic
+arm's already-fetched candidates.
+"""
+from types import SimpleNamespace
+
+import pytest
+
+from scribe.services import knowledge as knowledge_svc
+from scribe.services import snippets as snippets_svc
+from scribe.services.snippets import (
+    code_sha,
+    compose_data,
+    compose_verification,
+    verification_view,
+)
+
+
+def _note(data=None, body=""):
+    return SimpleNamespace(data=data, body=body, title="n — w", tags=["snippet"])
+
+
+# --- the verdict record ---------------------------------------------------
+
+
+def test_unknown_status_is_rejected_not_stored():
+    """A typo'd status would be stored happily and then never match any filter —
+    invisible rot in the thing built to make rot visible."""
+    with pytest.raises(ValueError):
+        compose_verification(status="broekn", checked_code_sha="abc")
+
+
+@pytest.mark.parametrize("status", ["ok", "missing", "moved", "changed"])
+def test_every_documented_status_is_accepted(status):
+    out = compose_verification(status=status, checked_code_sha="abc")
+    assert out["status"] == status
+    assert out["code_sha"] == "abc"
+    assert out["checked_at"]
+
+
+def test_blank_detail_and_path_are_omitted_not_stored_empty():
+    """Keeps `data` sparse, and keeps a containment match from tripping on ""."""
+    out = compose_verification(status="ok", checked_code_sha="abc")
+    assert "detail" not in out and "path" not in out
+
+
+# --- expiry by edit -------------------------------------------------------
+
+
+def test_code_sha_ignores_reformatting_but_not_meaning():
+    """A verdict must not expire because an editor stripped trailing whitespace,
+    and must expire when the code actually changes."""
+    assert code_sha("def f():\n    return 1") == code_sha("def f():   \n    return 1\n")
+    assert code_sha("def f():\n    return 1") != code_sha("def f():\n    return 2")
+
+
+def test_verdict_stays_current_while_code_is_unchanged():
+    code = "def f():\n    return 1"
+    fields = {"code": code, "verification": compose_verification(
+        status="ok", checked_code_sha=code_sha(code)
+    )}
+    view = verification_view(_note(), fields)
+    assert view["status"] == "ok"
+    assert view["current"] is True
+    assert view["needs_attention"] is False
+
+
+def test_an_ok_verdict_expires_once_the_code_is_edited():
+    """The whole reason the verdict carries a hash: an OK verdict must never go
+    on vouching for code nobody checked."""
+    fields = {"code": "def f():\n    return 2", "verification": compose_verification(
+        status="ok", checked_code_sha=code_sha("def f():\n    return 1")
+    )}
+    view = verification_view(_note(), fields)
+    assert view["current"] is False
+    assert view["needs_attention"] is True
+
+
+def test_never_checked_reads_as_unverified():
+    view = verification_view(_note(), {"code": "x = 1"})
+    assert view["status"] == "unverified"
+    assert view["current"] is False
+
+
+def test_a_failing_verdict_needs_attention_even_while_current():
+    code = "x = 1"
+    fields = {"code": code, "verification": compose_verification(
+        status="missing", checked_code_sha=code_sha(code)
+    )}
+    assert verification_view(_note(), fields)["needs_attention"] is True
+
+
+# --- the data mirror ------------------------------------------------------
+
+
+def test_compose_data_mirrors_the_current_code_hash():
+    """`data.code_sha` is what makes "this verdict expired" an SQL-expressible
+    comparison rather than a post-filter that would break pagination totals."""
+    out = compose_data(name="n", code="def f():\n    return 1")
+    assert out["code_sha"] == code_sha("def f():\n    return 1")
+
+
+def test_compose_data_omits_the_hash_when_there_is_no_code():
+    assert "code_sha" not in compose_data(name="n")
+
+
+def test_the_code_itself_is_never_copied_into_data():
+    """data indexes AROUND the code; duplicating the blob would be pure weight."""
+    out = compose_data(name="n", code="secret_looking_blob")
+    assert "code" not in out
+    assert "secret_looking_blob" not in str(out)
+
+
+# --- the filter, Python dialect -------------------------------------------
+
+
+def _data(status, checked_sha, current_sha):
+    return {
+        "verification": {"status": status, "code_sha": checked_sha},
+        "code_sha": current_sha,
+    }
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("", True),            # no filter
+        ("ok", True),
+        ("attention", False),
+        ("drifted", False),
+        ("unverified", False),
+    ],
+)
+def test_python_dialect_on_a_current_ok_verdict(value, expected):
+    data = _data("ok", "aaa", "aaa")
+    assert knowledge_svc.verification_matches(data, value) is expected
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [("ok", False), ("attention", True), ("drifted", False), ("unverified", False)],
+)
+def test_python_dialect_on_an_expired_ok_verdict(value, expected):
+    """Expired-but-ok is the case that motivated `attention` existing at all: it
+    is not `drifted` (nothing was found wrong) and not `unverified` (a check did
+    happen), yet it plainly needs looking at."""
+    data = _data("ok", "aaa", "bbb")
+    assert knowledge_svc.verification_matches(data, value) is expected
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [("ok", False), ("attention", True), ("drifted", True), ("changed", True),
+     ("missing", False)],
+)
+def test_python_dialect_on_a_drifted_verdict(value, expected):
+    data = _data("changed", "aaa", "aaa")
+    assert knowledge_svc.verification_matches(data, value) is expected
+
+
+def test_python_dialect_on_a_row_with_no_data_at_all():
+    """Rows predating migration 0070 have no `data`. They are unverified, which
+    is true — nobody has checked them."""
+    assert knowledge_svc.verification_matches(None, "unverified") is True
+    assert knowledge_svc.verification_matches(None, "attention") is False
+    assert knowledge_svc.verification_matches(None, "ok") is False
+
+
+# --- the filter, SQL dialect ----------------------------------------------
+
+
+def test_no_filter_produces_no_clause():
+    assert knowledge_svc._verification_clause("") is None
+
+
+@pytest.mark.parametrize(
+    "value", ["ok", "drifted", "unverified", "attention", "missing", "moved", "changed"]
+)
+def test_every_filter_value_compiles_to_sql(value):
+    """The two dialects have to agree, and the SQL one has to be expressible at
+    all — the expired comparison relies on jsonpath reading two fields of the
+    same row, which is easy to get wrong silently."""
+    clause = knowledge_svc._verification_clause(value)
+    assert clause is not None
+    assert str(clause)
+
+
+def test_a_status_value_is_json_quoted_into_the_jsonpath():
+    """Statuses are validated on WRITE, but the filter takes caller input
+    directly — a quote must not be able to break out of the jsonpath and turn
+    a narrowing filter into a widening one."""
+    hostile = 'x" || @.status == "ok'
+    clause = knowledge_svc._verification_clause(hostile)
+    params = list(clause.compile().params.values())
+    jsonpath = next(p for p in params if isinstance(p, str) and "status" in p)
+    # The injected quote is escaped, so the whole hostile value stays ONE string
+    # literal rather than closing it and appending a second condition. Asserting
+    # on the exact json.dumps form rather than counting "@.status" — the hostile
+    # value legitimately contains that text inside the literal.
+    import json as _json
+
+    assert jsonpath == f"$.verification ? (@.status == {_json.dumps(hostile)})"
+    assert '\\"' in jsonpath
+
+
+# --- write path -----------------------------------------------------------
+
+
+async def test_recording_a_verdict_does_not_touch_the_body():
+    """A verdict is metadata about the snippet, not part of it. Writing it into
+    the body would put it into the embedding and change what the record recalls
+    against."""
+    from unittest.mock import AsyncMock, patch
+
+    note = SimpleNamespace(
+        id=5, user_id=1, note_type="snippet", deleted_at=None,
+        title="n — w", body="```python\nx = 1\n```", tags=["snippet"], data=None,
+    )
+    with (
+        patch.object(snippets_svc, "get_snippet", AsyncMock(return_value=note)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)),
+        patch.object(
+            snippets_svc.notes_svc, "update_note", AsyncMock(return_value=note)
+        ) as upd,
+    ):
+        await snippets_svc.record_verification(1, 5, status="moved", detail="renamed")
+
+    assert set(upd.call_args.kwargs) == {"data"}
+    stored = upd.call_args.kwargs["data"]["verification"]
+    assert stored["status"] == "moved"
+    assert stored["detail"] == "renamed"
+
+
+async def test_recording_a_verdict_requires_write_access():
+    """Being able to read a snippet someone shared must not let you mark it
+    broken — a verdict changes how the record is presented to them."""
+    from unittest.mock import AsyncMock, patch
+
+    note = SimpleNamespace(id=5, user_id=2, note_type="snippet", deleted_at=None,
+                           title="n — w", body="", tags=[], data=None)
+    with (
+        patch.object(snippets_svc, "get_snippet", AsyncMock(return_value=note)),
+        patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)),
+        patch.object(snippets_svc.notes_svc, "update_note", AsyncMock()) as upd,
+    ):
+        assert await snippets_svc.record_verification(1, 5, status="ok") is None
+    upd.assert_not_called()