"""Every retrieval path must declare the right visibility scope. `semantic_search_notes` serves three different kinds of act, and the correct scope differs for each. Getting one wrong is silent — the code still works, it just sees too much or too little — so each caller is pinned here: explicit search → "read" the caller asked; reach everything they may read auto-injection → "browse" nobody asked; never a one-to-one shared record near-duplicate → "own" a verdict that blocks a write can't hinge on someone else's notes The keyword and semantic halves of one hybrid search must also agree, or a shared record would be findable by wording and invisible by meaning. """ from unittest.mock import AsyncMock, MagicMock, patch import pytest @pytest.fixture(autouse=True) def _no_supersession(): """The auto-inject menu now asks which of its lines are superseded (#278). That is a real database call on a path these tests exercise without one. Stubbed to "nothing superseded" — the ordinary state — rather than hidden behind a try/except in the product, which would make the code lie about what it does. The label's own behaviour is covered in tests/test_supersession_ranking.py. """ with patch("scribe.services.plugin_context.superseded_ids", AsyncMock(return_value=set())): yield def _note(id=1, user_id=7, title="A note", note_type="note", is_task=False, task_kind="work"): n = MagicMock() n.id = id n.user_id = user_id n.title = title n.body = "body" n.tags = [] # Real values, not auto-attributes: the menu reads these to label each line, # and a MagicMock is truthy — every record would render as a task (note 2109). n.is_task = is_task n.task_kind = task_kind n.note_type = note_type return n @pytest.fixture def spy(): """Patch semantic_search_notes everywhere it's imported and capture kwargs.""" mock = AsyncMock(return_value=[]) targets = [ "scribe.services.embeddings.semantic_search_notes", "scribe.services.plugin_context.semantic_search_notes", "scribe.mcp.tools.search.semantic_search_notes", "scribe.routes.search.semantic_search_notes", ] patches = [patch(t, mock) for t in targets] for p in patches: p.start() yield mock for p in patches: p.stop() @pytest.mark.asyncio async def test_mcp_search_uses_read_scope(spy): from scribe.mcp._context import _user_id_ctx token = _user_id_ctx.set(7) try: with patch("scribe.services.retrieval_telemetry.record_retrieval", MagicMock()): from scribe.mcp.tools.search import search await search(q="debounce") finally: _user_id_ctx.reset(token) assert spy.await_args.kwargs["scope"] == "read" @pytest.mark.asyncio async def test_auto_inject_uses_browse_scope(spy): """The one retrieval nobody requested. A record shared one-to-one with the operator must never arrive this way.""" from scribe.services import plugin_context with patch.object(plugin_context, "get_autoinject_config", AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 5})), \ patch.object(plugin_context, "record_retrieval", MagicMock()): await plugin_context.build_autoinject_hint(7, "how do I debounce") assert spy.await_args.kwargs["scope"] == "browse" @pytest.mark.asyncio async def test_dedup_gate_uses_own_scope(spy): """An empty title skips the title signal and goes straight to the semantic one — no session needed. The gate blocks a create and tells the caller to update the match instead, so matching another user's record would refuse their write and point them at something they may not be able to edit.""" from scribe.services import dedup await dedup.find_duplicate_note( 7, "", body="x" * 400, project_id=None, is_task=False, ) assert spy.await_args.kwargs["scope"] == "own" @pytest.mark.asyncio async def test_hybrid_search_halves_agree_on_scope(): """The keyword half and the semantic half of one search must see equally.""" from scribe.services import knowledge captured: dict[str, object] = {} async def fake_semantic(**kwargs): captured["scope"] = kwargs.get("scope") return [] from scribe.models.note import Note with patch("scribe.services.embeddings.semantic_search_notes", AsyncMock(side_effect=fake_semantic)), \ patch.object(knowledge, "readable_notes_clause", MagicMock(return_value=(Note.user_id == 7))) as read_clause, \ patch.object(knowledge, "async_session") as sess: session = AsyncMock() session.__aenter__ = AsyncMock(return_value=session) session.__aexit__ = AsyncMock(return_value=False) result = MagicMock() result.scalars.return_value.all.return_value = [] session.execute = AsyncMock(return_value=result) sess.return_value = session await knowledge.query_knowledge( user_id=7, note_type=None, tags=[], sort="modified", q="debounce", limit=10, offset=0, ) # Keyword half took the read scope... read_clause.assert_called_once_with(7) # ...and so did the semantic half. assert captured["scope"] == "read" @pytest.mark.asyncio async def test_injected_menu_attributes_another_users_note(): """Auto-inject can surface a collaborator's note via a shared project. The menu line is the only provenance the agent sees, so it has to name them.""" from scribe.services import plugin_context mine, theirs = _note(1, user_id=7, title="Mine"), _note(2, user_id=9, title="Theirs") with patch.object(plugin_context, "semantic_search_notes", AsyncMock(return_value=[(0.9, theirs), (0.88, mine)])), \ patch.object(plugin_context, "get_autoinject_config", AsyncMock(return_value={"enabled": True, "threshold": 0.5, "top_k": 5})), \ patch.object(plugin_context, "record_retrieval", MagicMock()), \ patch.object(plugin_context, "owner_names_for", AsyncMock(return_value={9: "alex"})): out = await plugin_context.build_autoinject_hint(7, "anything") lines = out["context"].splitlines() theirs_line = next(ln for ln in lines if "#2" in ln) mine_line = next(ln for ln in lines if "#1" in ln) assert "shared by alex" in theirs_line assert "suggestion" in theirs_line # The operator's own note stays unadorned — absence of a marker is the signal. assert "shared by" not in mine_line @pytest.mark.asyncio async def test_injected_menu_labels_the_record_kind(): """Every injected line renders the same shape, so without a kind marker a recorded snippet is indistinguishable from a stray dev-log — exactly where prior art most needs to stand out.""" from scribe.services import plugin_context hits = [ (0.92, _note(1, title="debounce — rate-limit a callback", note_type="snippet")), (0.91, _note(2, title="Release checklist", note_type="process")), (0.90, _note(3, title="Auth token expiry", is_task=True, task_kind="issue")), (0.89, _note(4, title="Ship the drafter", is_task=True)), (0.88, _note(5, title="Why we dropped CalDAV")), ] with patch.object(plugin_context, "semantic_search_notes", AsyncMock(return_value=hits)), \ patch.object(plugin_context, "get_autoinject_config", AsyncMock(return_value={"enabled": True, "threshold": 0.5, "top_k": 5})), \ patch.object(plugin_context, "record_retrieval", MagicMock()): out = await plugin_context.build_autoinject_hint(7, "anything") lines = out["context"].splitlines() by_id = {n: next(ln for ln in lines if f"#{n} " in ln) for n in (1, 2, 3, 4, 5)} assert "[snippet]" in by_id[1] assert "[process]" in by_id[2] # Task-ness wins over note_type, and an issue says so rather than "task". assert "[issue]" in by_id[3] assert "[task]" in by_id[4] assert "[note]" in by_id[5] # Still title-first: the marker is metadata, not an excuse to carry bodies. assert "body" not in out["context"] # The header can't claim they're all notes when the markers say otherwise. assert "Scribe records" in lines[0] def test_list_shaped_services_use_the_shared_visibility_clause(): """Every service that LISTS notes must declare its reach through notes_visibility_clause / browsable_notes_clause — never a bare owner filter (#47, #2462). Source inspection, because this is the shape no behavioural test catches: an owner-only list returns correct-looking rows and simply omits the shared ones, which is how list_notes shipped owner-only for months while query_knowledge beside it was deliberately browse-scoped. A task in a shared project was invisible in list_tasks, enter_project and the web UI while the same project's notes appeared. """ import ast import inspect import textwrap from scribe.services import knowledge, notes for fn, label in ((notes.list_notes, "notes.list_notes"), (knowledge.query_knowledge, "knowledge.query_knowledge")): source = textwrap.dedent(inspect.getsource(fn)) assert ( "notes_visibility_clause" in source or "browsable_notes_clause" in source ), ( f"{label} does not reference a shared visibility clause — a bare " f"owner filter silently hides shared-project records (#2462)" ) # And it must not ALSO carry a bare owner filter on the main query, # which would quietly re-narrow whatever the clause granted. tree = ast.parse(source) for node in ast.walk(tree): if (isinstance(node, ast.Compare) and isinstance(node.left, ast.Attribute) and node.left.attr == "user_id" and isinstance(node.left.value, ast.Name) and node.left.value.id == "Note"): raise AssertionError( f"{label} compares Note.user_id directly — reach must come " f"from the shared clause, not a bare owner filter" )