FabledGrimoire/tests/test_retrieval_scopes.py

"""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
from tests.helpers import fake_note, make_mock_session


pytestmark = pytest.mark.usefixtures("_no_supersession")


@pytest.fixture
def spy():
    """Patch semantic_search_notes everywhere it's imported and capture kwargs."""
    mock = AsyncMock(return_value=[])
    targets = [
        "grimoire.services.embeddings.semantic_search_notes",
        "grimoire.services.plugin_context.semantic_search_notes",
        "grimoire.mcp.tools.search.semantic_search_notes",
        "grimoire.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 grimoire.mcp._context import _user_id_ctx
    token = _user_id_ctx.set(7)
    try:
        with patch("grimoire.services.retrieval_telemetry.record_retrieval", MagicMock()):
            from grimoire.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 grimoire.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 grimoire.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 grimoire.services import knowledge

    captured: dict[str, object] = {}

    async def fake_semantic(**kwargs):
        captured["scope"] = kwargs.get("scope")
        return []

    from grimoire.models.note import Note
    with patch("grimoire.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 = make_mock_session()
        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 grimoire.services import plugin_context

    mine, theirs = fake_note(id=1, title="Mine", user_id=7), fake_note(id=2, title="Theirs", user_id=9)
    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 grimoire.services import plugin_context

    hits = [
        (0.92, fake_note(id=1, title="debounce — rate-limit a callback", note_type="snippet")),
        (0.91, fake_note(id=2, title="Release checklist", note_type="process")),
        (0.90, fake_note(id=3, title="Auth token expiry", is_task=True,
                         task_kind="issue", status="todo")),
        (0.89, fake_note(id=4, title="Ship the drafter", is_task=True,
                         status="done")),
        (0.88, fake_note(id=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".
    # A task also carries its status: #4 is DONE, and a line that says only
    # "[task]" lets a finished step be cited as an open one.
    assert "[issue (todo)]" in by_id[3]
    assert "[task (done)]" 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 "Grimoire 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.

    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 grimoire.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"
        )
        # 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"
                )