bvandeusenandClaude Opus 5.5
f4e9cd429b
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 14s
CI & Build / TypeScript typecheck (push) Successful in 54s
CI & Build / integration (push) Successful in 1m1s
CI & Build / Python tests (push) Successful in 1m40s
CI & Build / Build & push image (push) Canceled after 27s
The four embedding tables stamped chunker_version but not the model, and vector(384) is a width, not an identity: a same-width model swap would write a second geometry beside the first with no error. - embedding_model on note/rule/milestone/system embeddings (0109; existing rows stamped with the only model any install has ever run). - Every write stamps EMBEDDING_MODEL; every backfill's "current" test is is_current_stamp(), both halves of calibration_stamp(). - migrate_floor refuses while any row its surface searches is off the live model, before sampling: re-embed, then migrate. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
"""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 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]:
|
|
with patch("scribe.services.embeddings.get_embedding",
|
|
AsyncMock(return_value=QUERY_VEC)):
|
|
hits = await semantic_search_rules(user_id, "anything", limit=10,
|
|
threshold=0.5, **scope)
|
|
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()
|