FabledScribe/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


def _note(id=1, user_id=7, title="A note"):
    n = MagicMock()
    n.id = id
    n.user_id = user_id
    n.title = title
    n.body = "body"
    n.tags = []
    n.is_task = False
    n.note_type = "note"
    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