diff --git a/src/scribe/mcp/tools/rulebooks.py b/src/scribe/mcp/tools/rulebooks.py index a2af6ae..ccb9979 100644 --- a/src/scribe/mcp/tools/rulebooks.py +++ b/src/scribe/mcp/tools/rulebooks.py @@ -10,6 +10,7 @@ spec. from __future__ import annotations from scribe.mcp._context import current_user_id +from scribe.services import dedup as dedup_svc from scribe.services import rulebooks as rulebooks_svc from scribe.services import trash as trash_svc @@ -257,6 +258,7 @@ async def get_rule(rule_id: int) -> dict: async def create_rule( topic_id: int, title: str, statement: str, why: str = "", how_to_apply: str = "", order_index: int = 0, + force: bool = False, ) -> dict: """Create a new rule in a rulebook (a SHARED rule — keep it general). @@ -275,8 +277,15 @@ async def create_rule( why: Optional rationale — the reason the rule exists. how_to_apply: Optional operationalization — when / where it kicks in. order_index: Display order within the topic (default 0). + force: Bypass the near-duplicate gate. By default, a title-identical rule + already in this topic BLOCKS creation and returns its id so you update + it instead. Set true only for a genuinely distinct rule. """ uid = current_user_id() + if not force: + dup = await dedup_svc.find_duplicate_rule(title, topic_id=topic_id) + if dup is not None: + return dedup_svc.duplicate_response(dup, "rule") rule = await rulebooks_svc.create_rule( topic_id=topic_id, user_id=uid, title=title, statement=statement, @@ -288,6 +297,7 @@ async def create_rule( async def create_project_rule( project_id: int, statement: str, title: str = "", why: str = "", how_to_apply: str = "", order_index: int = 0, + force: bool = False, ) -> dict: """Create a rule scoped to a single project (no rulebook needed). @@ -307,9 +317,16 @@ async def create_project_rule( why: Optional rationale — the reason the rule exists. how_to_apply: Optional operationalization — when / where it kicks in. order_index: Display order within the project's rule list (default 0). + force: Bypass the near-duplicate gate. By default, a title-identical rule + already on this project BLOCKS creation and returns its id so you + update it instead. Set true only for a genuinely distinct rule. """ uid = current_user_id() derived_title = title.strip() or statement.strip().split(".")[0][:50] + if not force: + dup = await dedup_svc.find_duplicate_rule(derived_title, project_id=project_id) + if dup is not None: + return dedup_svc.duplicate_response(dup, "rule") rule = await rulebooks_svc.create_project_rule( project_id=project_id, user_id=uid, title=derived_title, statement=statement, diff --git a/src/scribe/services/dedup.py b/src/scribe/services/dedup.py index 20556fc..980efed 100644 --- a/src/scribe/services/dedup.py +++ b/src/scribe/services/dedup.py @@ -29,6 +29,7 @@ from sqlalchemy import func, select from scribe.models import async_session from scribe.models.note import Note +from scribe.models.rulebook import Rule from scribe.services import embeddings as embeddings_svc logger = logging.getLogger(__name__) @@ -131,3 +132,33 @@ async def find_duplicate_note( return DuplicateMatch(note.id, note.title, round(score, 3), "semantic") return None + + +async def find_duplicate_rule( + title: str, + topic_id: int | None = None, + project_id: int | None = None, +) -> DuplicateMatch | None: + """Title-based near-duplicate of a rule, scoped to the same topic (a rulebook + rule) or the same project (a project rule). Rules aren't a semantic-retrieval + surface, so a normalized-title match is the right (and only) signal. Fail-open + like find_duplicate_note.""" + norm = " ".join((title or "").split()).lower() + if not norm or (topic_id is None and project_id is None): + return None + try: + async with async_session() as session: + stmt = select(Rule).where( + Rule.deleted_at.is_(None), + func.lower(func.trim(Rule.title)) == norm, + ) + if topic_id is not None: + stmt = stmt.where(Rule.topic_id == topic_id) + else: + stmt = stmt.where(Rule.project_id == project_id) + existing = (await session.execute(stmt.limit(1))).scalars().first() + if existing is not None: + return DuplicateMatch(existing.id, existing.title, 1.0, "title") + except Exception: + logger.debug("dedup rule title check skipped — query failed", exc_info=True) + return None diff --git a/tests/test_mcp_tool_rulebooks.py b/tests/test_mcp_tool_rulebooks.py index 80088f8..610e383 100644 --- a/tests/test_mcp_tool_rulebooks.py +++ b/tests/test_mcp_tool_rulebooks.py @@ -94,6 +94,34 @@ async def test_create_rule_passes_required_fields(): assert kwargs["statement"] == "Work directly on dev" +@pytest.mark.asyncio +async def test_create_rule_blocked_by_duplicate_gate(): + from scribe.services.dedup import DuplicateMatch + dup = DuplicateMatch(id=47, title="dev is home", similarity=1.0, reason="title") + create_mock = AsyncMock() + with patch("scribe.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule", + AsyncMock(return_value=dup)), \ + patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", create_mock): + from scribe.mcp.tools.rulebooks import create_rule + out = await create_rule(topic_id=10, title="dev is home", statement="x") + assert out["duplicate"] is True + assert out["existing_id"] == 47 + assert "update_rule" in out["message"] + create_mock.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_rule_force_bypasses_duplicate_gate(): + find_mock = AsyncMock() + with patch("scribe.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule", find_mock), \ + patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", + AsyncMock(return_value=_fake_rule(id=5))): + from scribe.mcp.tools.rulebooks import create_rule + out = await create_rule(topic_id=10, title="dev is home", statement="x", force=True) + assert out["id"] == 5 + find_mock.assert_not_called() + + @pytest.mark.asyncio async def test_update_rule_only_sends_non_default_fields(): rule = _fake_rule() diff --git a/tests/test_services_dedup.py b/tests/test_services_dedup.py index 1f6383e..17490c1 100644 --- a/tests/test_services_dedup.py +++ b/tests/test_services_dedup.py @@ -3,7 +3,12 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from scribe.services.dedup import DuplicateMatch, duplicate_response, find_duplicate_note +from scribe.services.dedup import ( + DuplicateMatch, + duplicate_response, + find_duplicate_note, + find_duplicate_rule, +) def _session_returning(note): @@ -74,6 +79,29 @@ async def test_semantic_match_of_other_note_type_is_ignored(): assert dup is None # type mismatch must not block +@pytest.mark.asyncio +async def test_rule_title_match_in_topic(): + rule = _fake_note(id=47, title="Honor the multi-user sharing ACL") + with patch("scribe.services.dedup.async_session", + return_value=_session_returning(rule)): + dup = await find_duplicate_rule( + "honor the multi-user sharing acl", topic_id=7, + ) + assert dup is not None + assert dup.id == 47 + assert dup.reason == "title" + + +@pytest.mark.asyncio +async def test_rule_requires_a_scope(): + # No topic_id and no project_id → nothing to scope to → no match, no query. + sess = AsyncMock() + with patch("scribe.services.dedup.async_session", return_value=sess): + dup = await find_duplicate_rule("anything") + assert dup is None + sess.__aenter__.assert_not_called() + + def test_duplicate_response_shape(): dm = DuplicateMatch(id=5, title="Foo", similarity=1.0, reason="title") r = duplicate_response(dm, "task")