From ffb7a0fe38e9cf5435cec334018fd3904d8e09fb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 26 Aug 2026 14:17:41 -0400 Subject: [PATCH] feat(rules): both doors carry the trigger, the tier, the areas and the edges (#3029, milestone 307 step 3, surfaces) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCP and REST both gain when_to_apply / tier / system_ids / arose_from_id on create and update, plus relate_rules / unrelate_rules for the typed edges, and get_rule now returns a rule's areas and relations alongside it. rule_detail() is a SERVICE function, not one per door. It started as a copy in each — identical, and the prior-art hook flagged it immediately, which is the same lesson rules_payload (#2858) already recorded: a second copy drifts. Both doors call the one seam, so create, update and get cannot disagree about what a rule looks like coming back. The authoring guidance lands in create_rule's docstring rather than in a rule, per rule 119 as the operator described it: this is behaviour every instance should inherit, not one operator's preference. It states the test — ONE RULE = ONE THING YOU COULD VIOLATE. Rules that FAIL TOGETHER get linked with relate_rules(kind="co_surfaces"), never merged into one row. — and names why merging loses: a merged rule cannot be cited, surfaced or suppressed a clause at a time, and it grows without limit because adding to it is always cheaper than adding a rule. create_project_rule says the same about "overrides", which is what FabledCurator's 85/86 should have been instead of near-copies that drift from their parent. The tier arg carries the test itself: can you name the trigger WITHOUT naming a system, an artifact type or a moment? If the honest answer is "whenever you are working", it is always_on. Tests: the applicable-rules cases fabricated raw tuples matching the old column lists, so they move to the entity shape via fake_rule; new cases pin rule_brief (a DATE not a stamp, the depth left to get_rule, no null keys) and that an unknown tier falls back to BINDING. fake_rule gains when_to_apply / tier / arose_from_id for the note-2109 reason the helper exists: unnamed, they would be truthy MagicMocks. The tool tests stub the new rule_detail seam — they are about argument forwarding and have no database. The module header's "Sixteen tools" had been wrong for two milestones; the registration count test is what actually catches that, so the header now says so instead of carrying a number. Co-Authored-By: Claude Opus 5 (1M context) --- src/scribe/mcp/tools/rulebooks.py | 151 ++++++++++++++++++++++++++---- src/scribe/routes/rulebooks.py | 59 ++++++++++-- src/scribe/services/rulebooks.py | 25 +++++ tests/helpers.py | 6 +- tests/test_mcp_tool_rulebooks.py | 75 +++++++++++++-- tests/test_routes_rulebooks.py | 5 +- tests/test_services_rulebooks.py | 64 ++++++++++++- 7 files changed, 346 insertions(+), 39 deletions(-) diff --git a/src/scribe/mcp/tools/rulebooks.py b/src/scribe/mcp/tools/rulebooks.py index b496648..6fc4f00 100644 --- a/src/scribe/mcp/tools/rulebooks.py +++ b/src/scribe/mcp/tools/rulebooks.py @@ -1,7 +1,12 @@ """MCP tools for the Scribe Rulebook system. -Sixteen tools: rulebook/topic/rule CRUD + subscription management. Thin -wrappers over services/rulebooks.py — ownership is enforced in the service. +Rulebook / topic / rule CRUD, subscription management, and the rule-to-rule +edges. Thin wrappers over services/rulebooks.py — ownership is enforced in the +service, and the record shape comes from rule_brief / rule_detail there rather +than being rebuilt here. + +(The header used to say "Sixteen tools" and had been wrong for two milestones; +the count lives in the registration test, which fails when it drifts.) Destructive ops (delete_*) require confirmed=True; otherwise return a preview-style warning. Mirrors the pattern in delete_event and the design @@ -195,8 +200,12 @@ async def delete_topic(topic_id: int, confirmed: bool = False) -> dict: def _rule_summary(r) -> dict: """The list-row shape for a rule: what an agent needs to APPLY it. The - full record (why, how_to_apply, timestamps) is get_rule's job.""" - return {"id": r.id, "title": r.title, "statement": r.statement, "topic_id": r.topic_id} + full record (why, how_to_apply, timestamps) is get_rule's job. + + One line, because the shape itself lives in the service — this was one of + three hand-written copies that had already drifted apart (note 3026). + """ + return rulebooks_svc.rule_brief(r) async def list_rules( @@ -242,18 +251,25 @@ async def list_always_on_rules(project_id: int = 0) -> dict: async def get_rule(rule_id: int) -> dict: - """Fetch a rule by id — full statement + why + how_to_apply.""" + """Fetch a rule by id — full statement + why + how_to_apply. + + Also carries what a listing leaves out: the global `systems` this rule is + about, and its `relations`. Read the relations before acting on the rule — + a rule with a `co_surfaces` edge is half of a shape, and an `overrides` + edge means one of the pair is not in force here. + """ uid = current_user_id() rule = await rulebooks_svc.get_rule(rule_id, uid) if rule is None: raise ValueError(f"rule {rule_id} not found") - return rule.to_dict() + return await rulebooks_svc.rule_detail(uid, rule) async def create_rule( - topic_id: int, title: str, statement: str, + topic_id: int, title: str, statement: str, when_to_apply: str = "", why: str = "", how_to_apply: str = "", order_index: int = 0, - force: bool = False, + tier: str = "always_on", system_ids: list[int] | None = None, + arose_from_id: int = 0, force: bool = False, ) -> dict: """Create a new rule in a rulebook (a SHARED rule — keep it general). @@ -273,10 +289,36 @@ async def create_rule( Reusable code is a SNIPPET. Reach for a rule only when the thing genuinely is a standing instruction about how to work and nothing else can hold it. + ONE RULE = ONE THING YOU COULD VIOLATE. If a clause can be broken on its + own, and fixing that breakage doesn't require the neighbouring clauses, it + is a separate rule. Rules that FAIL TOGETHER get linked with relate_rules + (kind="co_surfaces"), never merged into one row: a merged rule cannot be + cited, surfaced or suppressed a clause at a time, and it grows without + limit because adding to it is always cheaper than adding a rule. + Args: topic_id: The topic to attach the rule to. title: A short imperative title (e.g. "dev is home"). statement: The actionable instruction (required). 1-2 sentences. + when_to_apply: WHEN this rule fires — the trigger, not the + instruction. State the moment or the material: "before any git + push", "when adding a value to a CHECK-gated column", "when a + release is being cut". Write it even though the parameter is + optional: it decides the tier below, it is how the rule is found + when it matters, and a rule nobody can place is a rule nobody + applies. + tier: "always_on" (default) or "conditional". + The test: can you name the trigger WITHOUT naming a system, an + artifact type or a moment? If the honest answer is "whenever you + are working", it is always_on. If you had to name something, it is + conditional — and conditional costs nothing when it is irrelevant, + which is what lets it be as long as it needs to be. + system_ids: Ids from list_canonical_systems — the global AREAS this + rule is about. This is what lets a rule reach a project that is + working in that area, so a CI rule surfaces on a CI change. + arose_from_id: The note or task that CAUSED this rule (an incident, a + decision). Prefer this over naming the record inside `why`, which + cannot be followed and does not survive a rewording. 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). @@ -291,16 +333,18 @@ async def create_rule( return dedup_svc.duplicate_response(dup, "rule") rule = await rulebooks_svc.create_rule( topic_id=topic_id, user_id=uid, - title=title, statement=statement, + title=title, statement=statement, when_to_apply=when_to_apply, + tier=tier, arose_from_id=arose_from_id, why=why, how_to_apply=how_to_apply, order_index=order_index, ) - return rule.to_dict() + return await rulebooks_svc.rule_detail(uid, rule, system_ids) async def create_project_rule( - project_id: int, statement: str, title: str = "", + project_id: int, statement: str, title: str = "", when_to_apply: str = "", why: str = "", how_to_apply: str = "", order_index: int = 0, - force: bool = False, + tier: str = "always_on", system_ids: list[int] | None = None, + arose_from_id: int = 0, force: bool = False, ) -> dict: """Create a rule scoped to a single project (no rulebook needed). @@ -312,11 +356,24 @@ async def create_project_rule( the rule is returned in get_project's applicable_rules (under project_rules) and in list_rules(project_id=...). + ONE RULE = ONE THING YOU COULD VIOLATE — see create_rule. A rule that + STRICTENS or REPLACES an inherited one is not a fresh rule: write it, then + relate_rules(kind="overrides") to the rule it supersedes, so the pair stays + connected instead of drifting into a contradiction nobody notices. A rule + that merely adds local detail to an inherited one uses "elaborates". + Args: project_id: The project to attach the rule to. statement: The actionable instruction (required). 1-2 sentences. title: Short imperative title. If empty, derived from the first ~50 characters of statement. + when_to_apply: WHEN this rule fires — the trigger, not the + instruction. See create_rule; it decides the tier and it is how + the rule is found at the moment it matters. + tier: "always_on" (default) or "conditional" — see create_rule. + system_ids: Ids from list_canonical_systems — the global AREAS this + rule is about. + arose_from_id: The note or task that CAUSED this 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). @@ -332,23 +389,36 @@ async def create_project_rule( 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, + title=derived_title, statement=statement, when_to_apply=when_to_apply, + tier=tier, arose_from_id=arose_from_id, why=why, how_to_apply=how_to_apply, order_index=order_index, ) - return rule.to_dict() + return await rulebooks_svc.rule_detail(uid, rule, system_ids) async def update_rule( - rule_id: int, title: str = "", statement: str = "", + rule_id: int, title: str = "", statement: str = "", when_to_apply: str = "", why: str = "", how_to_apply: str = "", order_index: int = -1, + tier: str = "", system_ids: list[int] | None = None, arose_from_id: int = 0, ) -> dict: - """Update a rule. Empty strings / order_index=-1 leave fields unchanged.""" + """Update a rule. Empty strings / order_index=-1 leave fields unchanged. + + Adding `when_to_apply` and a `tier` to an existing rule is the ordinary way + a rule stops being preloaded into every session and starts arriving when it + is relevant. `system_ids` REPLACES the rule's areas (pass [] to clear). + """ uid = current_user_id() fields: dict = {} if title: fields["title"] = title if statement: fields["statement"] = statement + if when_to_apply: + fields["when_to_apply"] = when_to_apply + if tier: + fields["tier"] = tier + if arose_from_id: + fields["arose_from_id"] = arose_from_id if why: fields["why"] = why if how_to_apply: @@ -358,7 +428,7 @@ async def update_rule( rule = await rulebooks_svc.update_rule(rule_id, uid, **fields) if rule is None: raise ValueError(f"rule {rule_id} not found") - return rule.to_dict() + return await rulebooks_svc.rule_detail(uid, rule, system_ids) async def delete_rule(rule_id: int, confirmed: bool = False) -> dict: @@ -496,12 +566,59 @@ async def unsuppress_topic_for_project( return {"project_id": project_id, "topic_id": topic_id, "suppressed": False} + + +async def relate_rules( + from_rule_id: int, to_rule_id: int, kind: str, note: str = "", +) -> dict: + """Draw a typed edge between two rules. Both must be yours. + + Reach for this INSTEAD of merging or duplicating: + + - kind="co_surfaces" — these two fail together, so they must arrive + together. Use it when you are tempted to fold one rule into another + because "either could surface without the other": that instinct is + right and merging is the wrong fix, because a merged rule cannot be + cited, suppressed or surfaced a clause at a time. Symmetric — draw it + once, it reads from both ends. + - kind="overrides" — this rule supersedes that one for its scope. Use it + when a project rule is stricter than, or replaces, an inherited one, + instead of writing a near-copy that will drift from its parent. + - kind="elaborates" — this rule adds local specifics to that one, and + should arrive with it rather than instead of it. + + Idempotent: re-drawing an existing edge returns it. + + Args: + note: WHY the edge holds. Worth writing for the same reason a rule + carries `why` — a later reader deciding whether it still applies + needs the reasoning, not just the fact. + """ + uid = current_user_id() + relation = await rulebooks_svc.add_rule_relation( + uid, from_rule_id, to_rule_id, kind, note, + ) + if relation is None: + raise ValueError( + f"rule {from_rule_id} or {to_rule_id} not found (both must be yours)" + ) + return relation.to_dict() + + +async def unrelate_rules(relation_id: int) -> dict: + """Remove one edge between rules (from relate_rules / get_rule.relations).""" + uid = current_user_id() + if not await rulebooks_svc.remove_rule_relation(uid, relation_id): + raise ValueError(f"relation {relation_id} not found") + return {"deleted": relation_id} + def register(mcp) -> None: for fn in ( list_rulebooks, get_rulebook, create_rulebook, update_rulebook, delete_rulebook, list_topics, create_topic, update_topic, delete_topic, list_rules, list_always_on_rules, get_rule, create_rule, create_project_rule, update_rule, delete_rule, + relate_rules, unrelate_rules, subscribe_project_to_rulebook, unsubscribe_project_from_rulebook, suppress_rule_for_project, unsuppress_rule_for_project, suppress_topic_for_project, unsuppress_topic_for_project, diff --git a/src/scribe/routes/rulebooks.py b/src/scribe/routes/rulebooks.py index 7d84987..08d0d9e 100644 --- a/src/scribe/routes/rulebooks.py +++ b/src/scribe/routes/rulebooks.py @@ -162,33 +162,73 @@ async def create_rule(topic_id: int): why=data.get("why", ""), how_to_apply=data.get("how_to_apply", ""), order_index=data.get("order_index", 0), + when_to_apply=data.get("when_to_apply", ""), + tier=data.get("tier", "always_on"), + arose_from_id=data.get("arose_from_id", 0) or 0, ) except ValueError as exc: return jsonify({"error": str(exc)}), 404 - return jsonify(rule.to_dict()), 201 + return jsonify(await rulebooks_svc.rule_detail( + get_current_user_id(), rule, data.get("system_ids"), + )), 201 @rulebooks_bp.get("/rules/") @login_required async def get_rule(rule_id: int): - rule = await rulebooks_svc.get_rule(rule_id, get_current_user_id()) + uid = get_current_user_id() + rule = await rulebooks_svc.get_rule(rule_id, uid) if rule is None: return jsonify({"error": "rule not found"}), 404 - return jsonify(rule.to_dict()) + return jsonify(await rulebooks_svc.rule_detail(uid, rule)) @rulebooks_bp.patch("/rules/") @login_required async def update_rule(rule_id: int): data = await request.get_json() or {} + uid = get_current_user_id() fields = { k: v for k, v in data.items() - if k in ("title", "statement", "why", "how_to_apply", "order_index") + if k in ("title", "statement", "why", "how_to_apply", "order_index", + "when_to_apply", "tier", "arose_from_id") } - rule = await rulebooks_svc.update_rule(rule_id, get_current_user_id(), **fields) + rule = await rulebooks_svc.update_rule(rule_id, uid, **fields) if rule is None: return jsonify({"error": "rule not found"}), 404 - return jsonify(rule.to_dict()) + return jsonify(await rulebooks_svc.rule_detail(uid, rule, data.get("system_ids"))) + + +@rulebooks_bp.post("/rules//relations") +@login_required +async def relate_rules(rule_id: int): + """Draw a typed edge FROM this rule to another. + + Body: {"to_rule_id": N, "kind": "co_surfaces"|"overrides"|"elaborates", + "note": "..."}. Idempotent — re-drawing an edge returns the existing one. + """ + data = await request.get_json() or {} + to_rule_id = data.get("to_rule_id") + if not isinstance(to_rule_id, int): + return jsonify({"error": "to_rule_id is required"}), 400 + try: + relation = await rulebooks_svc.add_rule_relation( + get_current_user_id(), rule_id, to_rule_id, + data.get("kind", ""), data.get("note", ""), + ) + except ValueError as exc: + return jsonify({"error": str(exc)}), 400 + if relation is None: + return jsonify({"error": "rule not found"}), 404 + return jsonify(relation.to_dict()), 201 + + +@rulebooks_bp.delete("/rule-relations/") +@login_required +async def unrelate_rules(relation_id: int): + if not await rulebooks_svc.remove_rule_relation(get_current_user_id(), relation_id): + return jsonify({"error": "relation not found"}), 404 + return "", 204 @rulebooks_bp.delete("/rules/") @@ -332,7 +372,12 @@ async def create_project_rule(project_id: int): why=data.get("why", ""), how_to_apply=data.get("how_to_apply", ""), order_index=data.get("order_index", 0), + when_to_apply=data.get("when_to_apply", ""), + tier=data.get("tier", "always_on"), + arose_from_id=data.get("arose_from_id", 0) or 0, ) except ValueError as exc: return jsonify({"error": str(exc)}), 404 - return jsonify(rule.to_dict()), 201 + return jsonify(await rulebooks_svc.rule_detail( + get_current_user_id(), rule, data.get("system_ids"), + )), 201 diff --git a/src/scribe/services/rulebooks.py b/src/scribe/services/rulebooks.py index 1d7de8c..acdcca7 100644 --- a/src/scribe/services/rulebooks.py +++ b/src/scribe/services/rulebooks.py @@ -334,6 +334,31 @@ def rule_brief(rule: Rule, **extra) -> dict: return out +async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict: + """The full record, with its areas and edges attached. + + ONE seam for both doors and every write path, so create, update and get + cannot disagree about what a rule looks like coming back — the same + reasoning as attach_relations for notes (#2859), and the same reasoning + rule_brief exists for one level down. + + `system_ids=None` means "leave the tags alone"; a list (including []) + REPLACES them. + """ + if system_ids is not None: + await set_rule_systems(rule.id, user_id, system_ids) + data = rule.to_dict() + systems = (await list_rule_systems([rule.id])).get(rule.id, []) + relations = (await list_rule_relations([rule.id])).get(rule.id, []) + # Attached only when present (#2483): an empty key reads as a capability + # the record has and isn't using, which is a different claim. + if systems: + data["systems"] = systems + if relations: + data["relations"] = relations + return data + + async def create_rule( topic_id: int, user_id: int, title: str, statement: str, why: str = "", how_to_apply: str = "", order_index: int = 0, diff --git a/tests/helpers.py b/tests/helpers.py index 56bffcf..e263e61 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -153,8 +153,12 @@ def fake_topic(**attrs) -> MagicMock: def fake_rule(**attrs) -> MagicMock: return _with_defaults({ - "id": 1, "topic_id": 10, "title": "dev is home", + "id": 1, "topic_id": 10, "project_id": None, "title": "dev is home", "statement": "Work directly on dev", "why": "", "how_to_apply": "", + # Named for the note-2109 reason the whole helper exists: unnamed, + # `when_to_apply` and `arose_from_id` would be truthy MagicMocks and + # rule_brief would attach both keys on every stand-in. + "when_to_apply": None, "tier": "always_on", "arose_from_id": None, "order_index": 0, "created_at": _now(), "updated_at": _now(), }, attrs) diff --git a/tests/test_mcp_tool_rulebooks.py b/tests/test_mcp_tool_rulebooks.py index 526a2aa..33fbd58 100644 --- a/tests/test_mcp_tool_rulebooks.py +++ b/tests/test_mcp_tool_rulebooks.py @@ -1,5 +1,5 @@ """Tests for MCP rulebook tools — patches the service layer.""" -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from tests.helpers import FakeMCP, fake_rule, fake_rulebook, fake_topic @@ -48,11 +48,25 @@ async def test_get_rulebook_raises_when_not_found(): await get_rulebook(rulebook_id=999) +def _plain_detail(): + """Stub the rule_detail seam these tool tests are not about. + + create/update/get_rule now return through services.rulebooks.rule_detail, + which reads the rule's areas and edges from the database. These are unit + tests with no database, and what they assert is that the TOOL forwards the + right arguments — so the seam is stubbed to the plain record, the same way + they already stub the create/update calls themselves. + """ + async def _detail(_uid, rule, _system_ids=None): + return rule.to_dict() + return patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", _detail) + + @pytest.mark.asyncio async def test_create_rule_passes_required_fields(): rule = fake_rule(id=100, title="r", statement="s", topic_id=10) mock = AsyncMock(return_value=rule) - with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", mock): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", mock), _plain_detail(): from scribe.mcp.tools.rulebooks import create_rule await create_rule( topic_id=10, title="dev is home", statement="Work directly on dev", @@ -84,7 +98,8 @@ 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, title="r", statement="s", topic_id=10))): + AsyncMock(return_value=fake_rule(id=5, title="r", statement="s", topic_id=10))), \ + _plain_detail(): 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 @@ -95,7 +110,7 @@ async def test_create_rule_force_bypasses_duplicate_gate(): async def test_update_rule_only_sends_non_default_fields(): rule = fake_rule(id=100, title="r", statement="s", topic_id=10) mock = AsyncMock(return_value=rule) - with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock), _plain_detail(): from scribe.mcp.tools.rulebooks import update_rule await update_rule(rule_id=1, statement="new statement") args, kwargs = mock.call_args @@ -169,7 +184,7 @@ def test_register_attaches_all_sixteen_tools(): mcp = FakeMCP() register(mcp) - assert len(mcp.names) == 24 # +exclude/include_always_on_rulebook (milestone 297) + assert len(mcp.names) == 26 # +relate_rules/unrelate_rules (milestone 307) # spot-check a few names assert "list_rulebooks" in mcp.names assert "create_rule" in mcp.names @@ -239,7 +254,7 @@ async def test_update_rulebook_omits_always_on_when_none(): async def test_create_project_rule_passes_required_fields(): rule = fake_rule(id=100, title="r", statement="s", topic_id=10) mock = AsyncMock(return_value=rule) - with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail(): from scribe.mcp.tools.rulebooks import create_project_rule await create_project_rule( project_id=42, @@ -257,7 +272,7 @@ async def test_create_project_rule_passes_required_fields(): async def test_create_project_rule_derives_title_from_statement(): rule = fake_rule(id=100, title="r", statement="s", topic_id=10) mock = AsyncMock(return_value=rule) - with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail(): from scribe.mcp.tools.rulebooks import create_project_rule await create_project_rule( project_id=42, @@ -272,7 +287,7 @@ async def test_create_project_rule_derives_title_from_statement(): async def test_create_project_rule_uses_explicit_title_when_given(): rule = fake_rule(id=100, title="r", statement="s", topic_id=10) mock = AsyncMock(return_value=rule) - with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail(): from scribe.mcp.tools.rulebooks import create_project_rule await create_project_rule( project_id=42, @@ -325,3 +340,47 @@ async def test_unsuppress_topic_for_project_passes_through(): kwargs = mock.call_args.kwargs assert kwargs == {"project_id": 3, "topic_id": 22, "user_id": 7} assert out == {"project_id": 3, "topic_id": 22, "suppressed": False} + + +# ── Typed edges between rules (milestone 307) ─────────────────────────── + +@pytest.mark.asyncio +async def test_relate_rules_forwards_the_kind_and_the_why(): + """The edge exists so a shape stops being merged into one row. The `note` + travels with it for the same reason a rule carries `why`: whoever later + decides whether the edge still holds needs the reasoning.""" + relation = MagicMock() + relation.to_dict.return_value = {"id": 9, "kind": "co_surfaces"} + mock = AsyncMock(return_value=relation) + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.add_rule_relation", mock): + from scribe.mcp.tools.rulebooks import relate_rules + out = await relate_rules( + from_rule_id=46, to_rule_id=144, kind="co_surfaces", + note="a stale channel tag and an unparseable version both read as " + "no update available", + ) + assert out["id"] == 9 + args = mock.call_args.args + assert args[0] == 7 and args[1] == 46 and args[2] == 144 + assert args[3] == "co_surfaces" + assert "no update available" in args[4] + + +@pytest.mark.asyncio +async def test_relate_rules_raises_when_either_end_is_not_yours(): + """The service returns None when it cannot see both rules — a one-sided + edge would surface a rule the caller has no business reading.""" + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.add_rule_relation", + AsyncMock(return_value=None)): + from scribe.mcp.tools.rulebooks import relate_rules + with pytest.raises(ValueError, match="not found"): + await relate_rules(from_rule_id=1, to_rule_id=2, kind="overrides") + + +@pytest.mark.asyncio +async def test_unrelate_rules_raises_when_the_edge_is_gone(): + with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.remove_rule_relation", + AsyncMock(return_value=False)): + from scribe.mcp.tools.rulebooks import unrelate_rules + with pytest.raises(ValueError, match="not found"): + await unrelate_rules(relation_id=99) diff --git a/tests/test_routes_rulebooks.py b/tests/test_routes_rulebooks.py index 9ee5d9e..01641f6 100644 --- a/tests/test_routes_rulebooks.py +++ b/tests/test_routes_rulebooks.py @@ -44,7 +44,8 @@ def test_service_signatures_require_user_id(): "create_rulebook", "list_rulebooks", "get_rulebook", "update_rulebook", "delete_rulebook", "find_rulebook_by_title", "create_topic", "list_topics", "get_topic", "update_topic", "delete_topic", - "create_rule", "create_project_rule", + "create_rule", "create_project_rule", "rule_detail", + "set_rule_systems", "add_rule_relation", "remove_rule_relation", "list_rules", "list_always_on_rules", "get_rule", "update_rule", "delete_rule", "subscribe_project", "unsubscribe_project", "get_applicable_rules", @@ -117,5 +118,7 @@ def test_rule_and_subscription_handlers_callable(): for name in ( "list_rules", "create_rule", "get_rule", "update_rule", "delete_rule", "subscribe_project", "unsubscribe_project", "get_project_rules", + # The typed edges — both doors carry them (rule 33). + "relate_rules", "unrelate_rules", ): assert callable(getattr(rb_routes, name)) diff --git a/tests/test_services_rulebooks.py b/tests/test_services_rulebooks.py index 2be99cb..6d52944 100644 --- a/tests/test_services_rulebooks.py +++ b/tests/test_services_rulebooks.py @@ -2,6 +2,7 @@ Mirrors the pattern in tests/test_events_service.py. """ +from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -206,8 +207,10 @@ async def test_get_applicable_rules_returns_shape(): sub_result.all.return_value = [(1, "FabledSword family")] rules_result = MagicMock() rules_result.all.return_value = [ - # (rule_id, title, statement, topic_id, topic_title, rulebook_id, rulebook_title) - (i, f"Rule {i}", f"Statement {i}", 2, "git-workflow", 1, "FabledSword family") + # The query selects the ENTITY plus three labels, so rule_brief stays + # the one place deciding what a surfaced rule carries (note 3026). + (fake_rule(id=i, title=f"Rule {i}", statement=f"Statement {i}", topic_id=2), + "git-workflow", 1, "FabledSword family") for i in range(50) ] # Execute order: sub_q, suppressed_rules_q, suppressed_topics_q, rules_q, proj_rules_q @@ -244,7 +247,7 @@ async def test_get_applicable_rules_truncates_when_over_limit(): sub_result.all.return_value = [] rules_result = MagicMock() rules_result.all.return_value = [ - (i, f"r{i}", "stmt", 2, "topic", 1, "rb") for i in range(51) + (fake_rule(id=i, title=f"r{i}"), "topic", 1, "rb") for i in range(51) ] mock_session.execute = AsyncMock(side_effect=[ sub_result, _empty(), _empty(), rules_result, _empty(), @@ -265,8 +268,10 @@ async def test_get_applicable_rules_includes_project_scoped_rules(): mock_session = make_mock_session() proj_rules_result = MagicMock() proj_rules_result.all.return_value = [ - (100, "Use alembic", "Always run migrations via alembic, never raw SQL."), - (101, "PR-bound", "Land schema changes in their own PR."), + (fake_rule(id=100, topic_id=None, project_id=3, title="Use alembic", + statement="Always run migrations via alembic, never raw SQL."),), + (fake_rule(id=101, topic_id=None, project_id=3, title="PR-bound", + statement="Land schema changes in their own PR."),), ] mock_session.execute = AsyncMock(side_effect=[ _empty(), _empty(), _empty(), _empty(), proj_rules_result, @@ -311,3 +316,52 @@ async def test_get_applicable_rules_surfaces_suppressed_with_context(): assert result["suppressed_rules"][0]["rulebook_title"] == "FabledSword family" assert len(result["suppressed_topics"]) == 1 assert result["suppressed_topics"][0]["title"] == "design-system" + + +# ── rule_brief + tier (milestone 307) ─────────────────────────────────── + +def test_rule_brief_carries_age_but_not_the_deep_fields(): + """The shape a SURFACED rule takes, and the reason it exists. + + There were three hand-written copies of this dict and they had already + diverged — none carried the timestamps the model has always held, which is + why a rule written before the capability it duplicates was + indistinguishable at read time from one still doing work (note 3026). + """ + from scribe.services.rulebooks import rule_brief + + out = rule_brief(fake_rule( + when_to_apply="before any git push", + updated_at=datetime(2026, 6, 1, 14, 30, tzinfo=timezone.utc), + )) + assert out["when_to_apply"] == "before any git push" + assert out["tier"] == "always_on" + # A DATE, not a stamp: the question is "how old is this", and a full ISO + # string across the always-on set is ~2k characters of payload. + assert out["updated_at"] == "2026-06-01" + # The depth stays with get_rule — putting it in every listing is the bloat + # this milestone is about. + assert "why" not in out and "how_to_apply" not in out + + +def test_rule_brief_omits_keys_a_rule_has_no_value_for(): + """#2483: a null key reads as a capability the record has and isn't using, + which is a different claim from not having one.""" + from scribe.services.rulebooks import rule_brief + + out = rule_brief(fake_rule()) + assert "when_to_apply" not in out + assert "arose_from_id" not in out + + +def test_an_unknown_tier_falls_back_to_binding(): + """The asymmetry that decides the direction: a rule that preloads when it + needn't costs context; a rule that quietly stops preloading costs the + behaviour it was written for. So a typo binds.""" + from scribe.services.rulebooks import _valid_tier + + assert _valid_tier("conditional") == "conditional" + assert _valid_tier("always_on") == "always_on" + assert _valid_tier("Conditional") == "always_on" + assert _valid_tier("") == "always_on" + assert _valid_tier("occasionally") == "always_on"