FabledScribe/tests/test_integration_rule_scope.py

"""Real-Postgres tests for WHERE a rule reaches (milestone 414, step 1).

A rule lives in a rulebook topic (global) or on one project. Retrieval used to
ignore that and search every rule the user owned, so each project's rules were
injected into every other project's sessions. What a mock cannot show is the
join doing the scoping: these seed real rules with hand-made vectors and stub
only the embedder, so every rule is an equally good match and the home alone
decides what comes back.
"""
import traceback
import uuid
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio

from scribe.models import async_session
from scribe.models.embedding import EMBEDDING_DIM, RuleEmbedding
from scribe.models.project import Project
from scribe.models.share import ProjectShare
from scribe.services import rulebooks as rulebooks_svc
from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL, semantic_search_rules
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

QUERY_VEC = [1.0] + [0.0] * (EMBEDDING_DIM - 1)


@pytest_asyncio.fixture
async def homes():
    """One global rule, one rule on project A, one on project B, and a
    collaborator A is shared with. Every rule embeds identically to the query."""
    # Fresh users per test: every rule matches the query equally, so a rule
    # left by another test would be indistinguishable from a scoping leak.
    tag = uuid.uuid4().hex[:8]
    async with async_session() as s:
        owner = await ensure_user(s, f"rule_scope_owner_{tag}")
        collaborator = await ensure_user(s, f"rule_scope_collaborator_{tag}")
        a = Project(user_id=owner.id, title="Scope A")
        b = Project(user_id=owner.id, title="Scope B")
        s.add_all([a, b])
        await s.flush()
        s.add(ProjectShare(project_id=a.id, shared_with_user_id=collaborator.id,
                           permission="viewer", invited_by=owner.id))
        ids = {"owner": owner.id, "collaborator": collaborator.id, "a": a.id, "b": b.id}
        await s.commit()

    with patch("scribe.services.rulebooks._refresh_rule_embedding", MagicMock()):
        book = await rulebooks_svc.create_rulebook(ids["owner"], "Scope house style")
        topic = await rulebooks_svc.create_topic(book.id, ids["owner"], "everywhere")
        glob = await rulebooks_svc.create_rule(
            topic.id, ids["owner"], "Global scope rule", "Applies in every project.",
            when_to_apply="always",
        )
        on_a = await rulebooks_svc.create_project_rule(
            ids["a"], ids["owner"], "Project A rule", "Applies to A only.",
            when_to_apply="working on A",
        )
        on_b = await rulebooks_svc.create_project_rule(
            ids["b"], ids["owner"], "Project B rule", "Applies to B only.",
            when_to_apply="working on B",
        )

    async with async_session() as s:
        for rule in (glob, on_a, on_b):
            s.add(RuleEmbedding(
                rule_id=rule.id, chunk_index=0, embedding=QUERY_VEC,
                chunk_text=rule.title, chunker_version=CHUNKER_VERSION,
                embedding_model=EMBEDDING_MODEL,
            ))
        await s.commit()
    ids.update(glob=glob.id, on_a=on_a.id, on_b=on_b.id)
    return ids


async def _found(user_id: int, **scope) -> set[int]:
    """The rule ids a search finds — and a failure, not an empty set, when the
    search never ran. `semantic_search_rules` fails open (an exception becomes
    []), so without `report["searched"]` an error and a scoping miss read the
    same (#4958)."""
    report: dict = {}
    raised: list[str] = []
    with patch("scribe.services.embeddings.get_embedding",
               AsyncMock(return_value=QUERY_VEC)), \
            patch("scribe.services.embeddings.logger") as log:
        # Called inside the except block, so the traceback is still current.
        log.warning.side_effect = lambda *_a, **_k: raised.append(traceback.format_exc())
        hits = await semantic_search_rules(user_id, "anything", limit=10,
                                           threshold=0.5, report=report, **scope)
    assert report.get("searched"), "the rule search did not run:\n" + "\n".join(raised)
    return {rule.id for _score, rule in hits}


async def test_retrieval_reaches_a_rule_only_from_its_home(homes):
    owner = homes["owner"]
    glob, on_a, on_b = homes["glob"], homes["on_a"], homes["on_b"]

    # A session bound to A: the global rule and A's own, never B's.
    assert await _found(owner, project_id=homes["a"]) == {glob, on_a}
    assert await _found(owner, project_id=homes["b"]) == {glob, on_b}

    # No bound project, and the default: global only. A caller that forgets
    # to pass a scope surfaces less, not another project's rules.
    assert await _found(owner) == {glob}

    # The explicit whole-rulebook question still reaches everything.
    assert await _found(owner, everywhere=True) == {glob, on_a, on_b}


async def test_a_shared_project_brings_its_rules_to_a_collaborator(homes):
    """Readability goes through access.can_read_project (rule 78): a viewer on
    A gets A's rules. Not the owner's global rules — rulebooks are the
    owner's — and not B's, which is not shared."""
    collaborator = homes["collaborator"]
    assert await _found(collaborator, project_id=homes["a"]) == {homes["on_a"]}
    assert await _found(collaborator, project_id=homes["b"]) == set()