feat(snippets): reverse lookup — find snippets by repo/path/symbol

"What canonical helpers already live in this file?" was unanswerable:
location lived only in the body markdown. It is now a jsonpath containment
query over the `notes.data` mirror added by migration 0070.

- One predicate in two dialects in services/knowledge.py: SQL (`data @?`,
  applied in the browse arm and the keyword arm before count/pagination, so
  totals stay honest) and Python (`location_matches`, for the semantic arm
  which post-filters candidates it already holds). Both must change together.
- Parts are ANDed within a SINGLE locations entry — repo A in one entry and
  path B in another is not "recorded at A/B". `path` also matches as a
  directory prefix, via jsonpath `starts with` rather than `@>`, which the
  same GIN index serves.
- `repo`/`path`/`symbol` reach the service, the REST list and the MCP tool
  under one name with one default (rule #33); the MCP docstring teaches the
  place form, and so does the reusing-code skill (plugin.json bumped).
- UI: a Location disclosure beside the snippet search, with its own empty
  state — "nothing kept there, so what you're about to write is new."

Settles #2083's open question (pre-0070 NULL `data`) by backfilling after
all: `backfill_snippet_data` runs at startup, deriving the mirror from the
body with the same parser the read path trusts. 0070's caution was about
mangling a hand-edited body; this never touches the body. The alternative
was a permanent second body-regex arm, or a query that silently answers
"nothing here" for an old snippet and gets the helper written twice.

Refs #2083, milestone #232.

tests/test_integration_snippet_locations.py

@@ -0,0 +1,299 @@
+"""Real-Postgres integration test for the snippet location reverse lookup (#2083).
+
+Runs only in the CI integration lane (real Postgres, schema built by
+`alembic upgrade head`, which includes migration 0070's `notes.data` + GIN index).
+This exercises what the unit mocks cannot:
+
+  - the jsonpath `@?` containment behind `_location_clause` — including
+    `starts with` for a path prefix, and the fact that parts must match within a
+    SINGLE `locations` entry;
+  - that the SQL dialect and the Python dialect (`location_matches`, used by the
+    semantic arm) return the SAME rows. Those two must never drift, or a snippet
+    is findable by wording and invisible by location;
+  - `backfill_snippet_data`, which is what stops a pre-0070 snippet from being
+    silently reported as "nothing recorded here."
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import delete, select
+
+from scribe.models import async_session, engine
+from scribe.models.note import Note
+from scribe.models.user import User
+from scribe.services.knowledge import location_matches, location_parts
+from scribe.services.snippets import (
+    SNIPPET_NOTE_TYPE,
+    backfill_snippet_data,
+    compose_body,
+    compose_data,
+    list_snippets,
+    snippet_fields,
+)
+
+pytestmark = pytest.mark.integration
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _dispose_engine():
+    # Per-loop pool: dispose after each test (see test_integration_db_maintenance).
+    yield
+    await engine.dispose()
+
+
+def _loc(repo="", path="", symbol=""):
+    return {"repo": repo, "path": path, "symbol": symbol}
+
+
+@pytest_asyncio.fixture
+async def seeded():
+    """A user with four snippets covering the cases the filter has to separate.
+
+    Returns (user_id, {label: note_id}).
+    """
+    async with async_session() as s:
+        user = User(username="snippet_loc_itest")
+        s.add(user)
+        await s.flush()
+
+        def _snippet(name, locations, code="return 1"):
+            return Note(
+                user_id=user.id,
+                note_type=SNIPPET_NOTE_TYPE,
+                title=f"{name} — does a thing",
+                body=compose_body(code=code, language="python", locations=locations),
+                tags=["python", "snippet"],
+                data=compose_data(name=name, language="python", locations=locations),
+            )
+
+        nested = _snippet("nested", [_loc("Scribe", "frontend/src/lib/x.ts", "helper")])
+        sibling = _snippet("sibling", [_loc("Scribe", "frontend/srcmap.ts", "other")])
+        # Two locations, deliberately crossed: repo Scribe at src/a.py and repo
+        # Portal at src/b.py. repo=Scribe + path=src/b.py must NOT match it.
+        multi = _snippet(
+            "multi",
+            [_loc("Scribe", "src/a.py", "alpha"), _loc("Portal", "src/b.py", "beta")],
+        )
+        # No structured location at all — must never satisfy a location filter,
+        # and must not error the query either.
+        placeless = _snippet("placeless", [])
+        # A plain note (not a snippet) with no `data` — proves the jsonpath
+        # operator tolerates NULL rather than raising.
+        plain = Note(user_id=user.id, title="plain", body="no data here")
+
+        s.add_all([nested, sibling, multi, placeless, plain])
+        await s.commit()
+        ids = (
+            user.id,
+            {
+                "nested": nested.id,
+                "sibling": sibling.id,
+                "multi": multi.id,
+                "placeless": placeless.id,
+                "plain": plain.id,
+            },
+        )
+    yield ids
+    user_id = ids[0]
+    async with async_session() as s:
+        await s.execute(delete(Note).where(Note.user_id == user_id))
+        await s.execute(delete(User).where(User.id == user_id))
+        await s.commit()
+
+
+async def _ids_for(user_id: int, **parts) -> set[int]:
+    items, total = await list_snippets(user_id, **parts)
+    assert total == len(items), "total must match the filtered page, not the whole list"
+    return {i["id"] for i in items}
+
+
+@pytest.mark.asyncio
+async def test_repo_filter_finds_every_snippet_recorded_in_it(seeded):
+    user_id, ids = seeded
+    found = await _ids_for(user_id, repo="Scribe")
+    assert found == {ids["nested"], ids["sibling"], ids["multi"]}
+    assert ids["placeless"] not in found
+
+
+@pytest.mark.asyncio
+async def test_exact_path_finds_only_that_file(seeded):
+    user_id, ids = seeded
+    assert await _ids_for(user_id, path="frontend/src/lib/x.ts") == {ids["nested"]}
+
+
+@pytest.mark.asyncio
+async def test_path_prefix_reaches_into_subdirectories(seeded):
+    user_id, ids = seeded
+    # ...and stops at the directory boundary: srcmap.ts is a sibling, not a child.
+    assert await _ids_for(user_id, path="frontend/src") == {ids["nested"]}
+    assert await _ids_for(user_id, path="frontend/src/") == {ids["nested"]}
+    assert await _ids_for(user_id, path="frontend") == {ids["nested"], ids["sibling"]}
+
+
+@pytest.mark.asyncio
+async def test_parts_must_match_within_one_location(seeded):
+    user_id, ids = seeded
+    assert await _ids_for(user_id, repo="Scribe", path="src/a.py") == {ids["multi"]}
+    # Same snippet, but repo and path belong to two different entries.
+    assert await _ids_for(user_id, repo="Scribe", path="src/b.py") == set()
+    assert await _ids_for(user_id, repo="Portal", path="src/b.py") == {ids["multi"]}
+
+
+@pytest.mark.asyncio
+async def test_symbol_filter_and_an_unrecorded_location(seeded):
+    user_id, ids = seeded
+    assert await _ids_for(user_id, symbol="helper") == {ids["nested"]}
+    assert await _ids_for(user_id, repo="NoSuchRepo") == set()
+
+
+@pytest.mark.asyncio
+async def test_no_location_filter_still_lists_everything(seeded):
+    """The filter is opt-in — a plain browse must not lose the placeless snippet."""
+    found = await _ids_for(user_id=seeded[0])
+    ids = seeded[1]
+    assert {ids["nested"], ids["sibling"], ids["multi"], ids["placeless"]} <= found
+    assert ids["plain"] not in found  # not a snippet
+
+
+@pytest.mark.asyncio
+async def test_keyword_arm_applies_the_location_filter(seeded):
+    """Search AND place compose. The embedder is stubbed out so this isolates the
+    keyword (ILIKE) arm, which is the second place the SQL clause has to appear."""
+    user_id, ids = seeded
+    with patch(
+        "scribe.services.embeddings.semantic_search_notes",
+        AsyncMock(return_value=[]),
+    ):
+        assert await _ids_for(user_id, q="nested", repo="Scribe") == {ids["nested"]}
+        # Matches the query but not the place.
+        assert await _ids_for(user_id, q="nested", repo="Portal") == set()
+
+
+@pytest.mark.asyncio
+async def test_semantic_arm_applies_the_location_filter(seeded):
+    """The semantic arm holds its candidates in memory, so it filters in Python.
+    A query that matches no title or body leaves only that arm in play."""
+    user_id, ids = seeded
+    async with async_session() as s:
+        rows = list(
+            (
+                await s.execute(
+                    select(Note)
+                    .where(Note.user_id == user_id)
+                    .where(Note.note_type == SNIPPET_NOTE_TYPE)
+                )
+            )
+            .scalars()
+            .all()
+        )
+    candidates = [(0.9, n) for n in rows]
+
+    with patch(
+        "scribe.services.embeddings.semantic_search_notes",
+        AsyncMock(return_value=candidates),
+    ):
+        # No keyword can match this, so every hit below came from the stub.
+        assert await _ids_for(user_id, q="zzzznomatch") == {
+            ids["nested"], ids["sibling"], ids["multi"], ids["placeless"],
+        }
+        assert await _ids_for(user_id, q="zzzznomatch", path="frontend/src") == {
+            ids["nested"],
+        }
+        assert await _ids_for(user_id, q="zzzznomatch", repo="Scribe", path="src/b.py") == set()
+
+
+@pytest.mark.asyncio
+async def test_sql_and_python_dialects_agree_on_the_same_rows(seeded):
+    """The drift guard: whatever the SQL arm returns, the Python arm must too."""
+    user_id, _ids = seeded
+    async with async_session() as s:
+        rows = list(
+            (
+                await s.execute(
+                    select(Note)
+                    .where(Note.user_id == user_id)
+                    .where(Note.note_type == SNIPPET_NOTE_TYPE)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        by_id = {n.id: n.data for n in rows}
+
+    cases = [
+        {"repo": "Scribe"},
+        {"path": "frontend/src"},
+        {"path": "frontend/src/lib/x.ts"},
+        {"repo": "Scribe", "path": "src/a.py"},
+        {"repo": "Scribe", "path": "src/b.py"},
+        {"symbol": "beta"},
+        {"repo": "Scribe", "path": "frontend", "symbol": "helper"},
+    ]
+    for case in cases:
+        sql_ids = await _ids_for(user_id, **case)
+        parts = location_parts(**case)
+        python_ids = {nid for nid, data in by_id.items() if location_matches(data, parts)}
+        assert sql_ids == python_ids, f"dialects disagree on {case}"
+
+
+@pytest.mark.asyncio
+async def test_backfill_makes_a_pre_0070_snippet_findable_by_location(seeded):
+    """A snippet stored before migration 0070 has no `data`, so the location query
+    can't see it. Unfilled, it reads as "nothing recorded here" — the failure the
+    backfill exists to prevent."""
+    user_id, _ids = seeded
+    locations = [_loc("Legacy", "old/path/y.py", "legacy_helper")]
+    async with async_session() as s:
+        old = Note(
+            user_id=user_id,
+            note_type=SNIPPET_NOTE_TYPE,
+            title="legacy — recorded before the data column existed",
+            body=compose_body(code="return 2", language="python", locations=locations),
+            tags=["python", "snippet"],
+            data=None,
+        )
+        s.add(old)
+        await s.commit()
+        old_id = old.id
+
+    assert await _ids_for(user_id, repo="Legacy") == set()
+
+    filled = await backfill_snippet_data()
+    assert filled >= 1
+
+    assert await _ids_for(user_id, repo="Legacy") == {old_id}
+    assert await _ids_for(user_id, path="old/path") == {old_id}
+
+    # The mirror agrees with the body it was derived from, and the body is untouched.
+    async with async_session() as s:
+        row = (await s.execute(select(Note).where(Note.id == old_id))).scalar_one()
+        assert row.data["locations"] == locations
+        assert snippet_fields(row)["symbol"] == "legacy_helper"
+        assert "old/path/y.py" in row.body
+
+
+@pytest.mark.asyncio
+async def test_backfill_is_idempotent_and_settles_a_fieldless_snippet(seeded):
+    """Second run fills nothing, and a snippet with no structured fields lands at
+    `{}` rather than staying NULL and being rescanned on every boot."""
+    user_id, _ids = seeded
+    async with async_session() as s:
+        bare = Note(
+            user_id=user_id,
+            note_type=SNIPPET_NOTE_TYPE,
+            title="",
+            body="",
+            tags=[],
+            data=None,
+        )
+        s.add(bare)
+        await s.commit()
+        bare_id = bare.id
+
+    await backfill_snippet_data()
+    assert await backfill_snippet_data() == 0
+
+    async with async_session() as s:
+        row = (await s.execute(select(Note).where(Note.id == bare_id))).scalar_one()
+        assert row.data == {}

tests/test_routes_snippets.py

@@ -72,6 +72,19 @@ def test_project_scoping_reaches_every_caller():
     assert "project_id" in inspect.getsource(routes.list_snippets_route)
 
 
+def test_location_lookup_reaches_every_caller():
+    """The reverse lookup — "what already lives here?" — has to be askable from
+    both surfaces, or an agent and a human get different answers about the same
+    file (rule #33)."""
+    from scribe.mcp.tools import snippets as tools
+    from scribe.routes import snippets as routes
+    from scribe.services import snippets as svc
+    for key in ("repo", "path", "symbol"):
+        assert key in inspect.signature(svc.list_snippets).parameters
+        assert key in inspect.signature(tools.list_snippets).parameters
+        assert f'request.args.get("{key}"' in inspect.getsource(routes)
+
+
 def test_update_field_map_matches_service_kwargs():
     """Every field the PATCH route forwards must be a real update_snippet kwarg
     (rule #33 interface-contract parity)."""

tests/test_snippet_location_filter.py

@@ -0,0 +1,205 @@
+"""Tests for the snippet location reverse lookup (#2083).
+
+Covers the pure half of the predicate — `location_parts`, `location_matches`,
+`location_jsonpath` — plus the parameter plumbing across the three surfaces
+(service / REST route / MCP tool), which rule #33 requires to agree on names and
+defaults.
+
+The SQL half can only be proved against real Postgres — see
+tests/test_integration_snippet_locations.py, which also pins the two dialects to
+the same answers on the same rows.
+"""
+import inspect
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services.knowledge import (
+    LOCATION_KEYS,
+    location_jsonpath,
+    location_matches,
+    location_parts,
+)
+
+
+def _data(*locations: dict) -> dict:
+    return {"locations": list(locations)}
+
+
+# --- location_parts -----------------------------------------------------------
+
+def test_location_parts_keeps_only_what_was_asked_for():
+    assert location_parts() == {}
+    assert location_parts(repo="Scribe") == {"repo": "Scribe"}
+    assert location_parts(repo="  Scribe  ", path=" src/x.py ") == {
+        "repo": "Scribe", "path": "src/x.py",
+    }
+
+
+def test_location_parts_treats_whitespace_as_absent():
+    """An empty query-string arg must not filter the list down to nothing."""
+    assert location_parts(repo="   ", path="", symbol=None or "") == {}
+
+
+def test_location_parts_covers_every_location_key():
+    assert set(location_parts(repo="a", path="b", symbol="c")) == set(LOCATION_KEYS)
+
+
+# --- location_matches ---------------------------------------------------------
+
+def test_no_parts_matches_everything():
+    assert location_matches(None, {}) is True
+    assert location_matches(_data({"repo": "x", "path": "y", "symbol": "z"}), {}) is True
+
+
+def test_matches_exact_repo_path_and_symbol():
+    data = _data({"repo": "Scribe", "path": "src/scribe/x.py", "symbol": "helper"})
+    assert location_matches(data, {"repo": "Scribe"})
+    assert location_matches(data, {"path": "src/scribe/x.py"})
+    assert location_matches(data, {"symbol": "helper"})
+    assert location_matches(data, {"repo": "Scribe", "symbol": "helper"})
+
+
+def test_path_matches_as_a_directory_prefix():
+    data = _data({"repo": "Scribe", "path": "frontend/src/lib/x.ts", "symbol": ""})
+    assert location_matches(data, {"path": "frontend/src"})
+    assert location_matches(data, {"path": "frontend/src/"})
+    assert location_matches(data, {"path": "frontend/src/lib/x.ts"})
+
+
+def test_path_prefix_stops_at_a_directory_boundary():
+    """`frontend/src` must not match `frontend/srcmap.ts` — a sibling, not a child."""
+    data = _data({"repo": "Scribe", "path": "frontend/srcmap.ts", "symbol": ""})
+    assert not location_matches(data, {"path": "frontend/src"})
+
+
+def test_parts_must_all_match_the_same_location():
+    """Repo A in one entry and path B in another is not "recorded at A/B"."""
+    data = _data(
+        {"repo": "Scribe", "path": "src/a.py", "symbol": ""},
+        {"repo": "Portal", "path": "src/b.py", "symbol": ""},
+    )
+    assert location_matches(data, {"repo": "Scribe", "path": "src/a.py"})
+    assert not location_matches(data, {"repo": "Scribe", "path": "src/b.py"})
+
+
+def test_missing_or_empty_data_never_matches():
+    assert not location_matches(None, {"repo": "Scribe"})
+    assert not location_matches({}, {"repo": "Scribe"})
+    assert not location_matches({"locations": []}, {"repo": "Scribe"})
+    assert not location_matches({"name": "x"}, {"repo": "Scribe"})
+
+
+def test_blank_recorded_part_does_not_match_a_requested_one():
+    data = _data({"repo": "Scribe", "path": "", "symbol": ""})
+    assert not location_matches(data, {"repo": "Scribe", "symbol": "helper"})
+
+
+# --- location_jsonpath --------------------------------------------------------
+
+def test_jsonpath_filters_within_one_locations_entry():
+    expr = location_jsonpath({"repo": "Scribe", "symbol": "helper"})
+    assert expr.startswith("$.locations[*] ? (")
+    assert '@.repo == "Scribe"' in expr
+    assert '@.symbol == "helper"' in expr
+    assert " && " in expr
+
+
+def test_jsonpath_path_asks_for_exact_or_prefix():
+    expr = location_jsonpath({"path": "frontend/src"})
+    assert '@.path == "frontend/src"' in expr
+    assert '@.path starts with "frontend/src/"' in expr
+    assert " || " in expr
+
+
+def test_jsonpath_does_not_double_the_prefix_slash():
+    assert 'starts with "frontend/src/"' in location_jsonpath({"path": "frontend/src/"})
+
+
+def test_jsonpath_quotes_values_as_json_literals():
+    """A quote in a repo name must stay inside the literal, not end it."""
+    nasty = 'we"ird'
+    expr = location_jsonpath({"repo": nasty})
+    assert json.dumps(nasty) in expr
+    assert '\\"' in expr
+
+
+def test_jsonpath_emits_parts_in_a_fixed_key_order():
+    """Stable text keeps the query plan cacheable and the tests readable."""
+    a = location_jsonpath({"symbol": "s", "repo": "r", "path": "p"})
+    b = location_jsonpath({"repo": "r", "path": "p", "symbol": "s"})
+    assert a == b
+
+
+# --- the SQL dialect renders (shape only; behaviour is the integration test) ---
+
+def test_location_clause_renders_a_jsonpath_containment():
+    """Catches an API slip in the fast lane rather than waiting for Postgres —
+    the clause has to compile to `data @? :jsonpath` against the pg dialect."""
+    from sqlalchemy.dialects import postgresql
+
+    from scribe.services.knowledge import _location_clause
+
+    sql = str(_location_clause({"repo": "Scribe"}).compile(dialect=postgresql.dialect()))
+    assert "@?" in sql
+    assert "data" in sql
+
+
+# --- surface plumbing (rule #33) ---------------------------------------------
+
+def test_agent_and_service_surfaces_take_the_same_location_params():
+    from scribe.mcp.tools.snippets import list_snippets as mcp_list
+    from scribe.services.snippets import list_snippets as svc_list
+
+    for fn in (mcp_list, svc_list):
+        params = inspect.signature(fn).parameters
+        for key in LOCATION_KEYS:
+            assert key in params, f"{fn.__qualname__} is missing {key}"
+            assert params[key].default == "", f"{fn.__qualname__}.{key} default drifted"
+
+
+def test_route_forwards_the_location_args_by_name():
+    """The REST surface is the third caller; it must use the same arg names."""
+    from scribe.routes import snippets as routes
+    src = inspect.getsource(routes)
+    for key in LOCATION_KEYS:
+        assert f'request.args.get("{key}"' in src
+        assert f"{key}={key}" in src
+
+
+@pytest.mark.asyncio
+async def test_mcp_tool_passes_locations_through_to_the_service():
+    with patch("scribe.services.snippets.list_snippets",
+               new=AsyncMock(return_value=([], 0))) as m, \
+         patch("scribe.services.access.label_shared_items",
+               new=AsyncMock(return_value=[])):
+        from scribe.mcp.tools.snippets import list_snippets
+        await list_snippets(repo="Scribe", path="src/scribe", symbol="helper")
+    kwargs = m.await_args.kwargs
+    assert kwargs["repo"] == "Scribe"
+    assert kwargs["path"] == "src/scribe"
+    assert kwargs["symbol"] == "helper"
+
+
+@pytest.mark.asyncio
+async def test_service_sends_no_location_filter_when_none_was_asked_for():
+    """A plain list must not carry an empty filter — that would exclude every
+    row whose `data` has no locations at all."""
+    with patch("scribe.services.knowledge.query_knowledge",
+               new=AsyncMock(return_value=([], 0))) as m:
+        from scribe.services.snippets import list_snippets
+        await list_snippets(1)
+    assert m.await_args.kwargs["locations"] is None
+
+
+@pytest.mark.asyncio
+async def test_service_builds_the_filter_from_the_parts_given():
+    with patch("scribe.services.knowledge.query_knowledge",
+               new=AsyncMock(return_value=([], 0))) as m:
+        from scribe.services.snippets import list_snippets
+        await list_snippets(1, q="debounce", repo="Scribe", path=" src/ ")
+    kwargs = m.await_args.kwargs
+    assert kwargs["locations"] == {"repo": "Scribe", "path": "src/"}
+    # Search and place compose — the location filter must not drop the query.
+    assert kwargs["q"] == "debounce"