"""MCP tools for the Scribe Rulebook system. 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 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 # ── Rulebook CRUD ─────────────────────────────────────────────────────── async def list_rulebooks() -> dict: """List all rulebooks owned by the current user. Returns id, title, description for each. """ uid = current_user_id() rows = await rulebooks_svc.list_rulebooks(uid) return {"rulebooks": [rb.to_dict() for rb in rows]} async def get_rulebook(rulebook_id: int) -> dict: """Fetch a rulebook by id with its full topic list.""" uid = current_user_id() rb = await rulebooks_svc.get_rulebook(rulebook_id, uid) if rb is None: raise ValueError(f"rulebook {rulebook_id} not found") topics = await rulebooks_svc.list_topics(rulebook_id, uid) data = rb.to_dict() data["topics"] = [t.to_dict() for t in topics] return data async def create_rulebook(title: str, description: str = "") -> dict: """Create a new rulebook (a shared, reusable module of general rules). Two ways a rulebook reaches projects, set by its always_on flag (toggle via update_rulebook): - always_on = true -> binds EVERY one of your projects automatically. Use for universal cross-project norms that apply across every project, not just one. - always_on = false -> binds only projects that subscribe (subscribe_project_to_rulebook). Use for a THEMED body of rules a category of projects shares (e.g. a design system that visual apps opt into). Either way a rulebook is SHARED, so its rules must stay general — agnostic to any single project. Project-specific rules go in create_project_rule. Args: title: Rulebook name. description: Optional short description of what this rulebook covers. """ uid = current_user_id() rb = await rulebooks_svc.create_rulebook( user_id=uid, title=title, description=description, ) return rb.to_dict() async def update_rulebook( rulebook_id: int, title: str = "", description: str = "", always_on: bool | None = None, ) -> dict: """Update an existing rulebook. Only non-empty fields are changed. Args: rulebook_id: Rulebook to update. title: New title. Empty string leaves unchanged. description: New description. Empty string leaves unchanged. always_on: When True, rules in this rulebook are loaded at session start by list_always_on_rules regardless of project context. Pass None to leave unchanged. """ uid = current_user_id() fields: dict = {} if title: fields["title"] = title if description: fields["description"] = description if always_on is not None: fields["always_on"] = always_on rb = await rulebooks_svc.update_rulebook(rulebook_id, uid, **fields) if rb is None: raise ValueError(f"rulebook {rulebook_id} not found") return rb.to_dict() async def delete_rulebook(rulebook_id: int, confirmed: bool = False) -> dict: """Permanently delete a rulebook (cascades to all its topics and rules). Pass confirmed=True to actually delete. Without confirmation, returns a preview describing what will be cascaded. """ uid = current_user_id() rb = await rulebooks_svc.get_rulebook(rulebook_id, uid) if rb is None: raise ValueError(f"rulebook {rulebook_id} not found") if not confirmed: topics = await rulebooks_svc.list_topics(rulebook_id, uid) rule_count = 0 for t in topics: rule_count += len(await rulebooks_svc.list_rules(uid, topic_id=t.id)) return { "warning": ( f"Rulebook {rulebook_id} ('{rb.title}') contains " f"{len(topics)} topics and {rule_count} rules; all will be " f"deleted. Pass confirmed=True to proceed." ), "confirmed_required": True, } batch = await trash_svc.delete(uid, "rulebook", rulebook_id) return {"deleted": rulebook_id, "deleted_batch_id": batch, "message": f"Moved to trash. Restore with restore('{batch}')."} # ── Topic CRUD ───────────────────────────────────────────────────────── async def list_topics(rulebook_id: int) -> dict: """List topics inside a rulebook.""" uid = current_user_id() rows = await rulebooks_svc.list_topics(rulebook_id, uid) return {"topics": [t.to_dict() for t in rows]} async def create_topic( rulebook_id: int, title: str, description: str = "", order_index: int = 0, ) -> dict: """Create a topic within a rulebook. Args: rulebook_id: Rulebook to add the topic to. title: Topic name (e.g. "git-workflow"). description: Optional description. order_index: Display order (0-based; default 0). """ uid = current_user_id() topic = await rulebooks_svc.create_topic( rulebook_id=rulebook_id, user_id=uid, title=title, description=description, order_index=order_index, ) return topic.to_dict() async def update_topic( topic_id: int, title: str = "", description: str = "", order_index: int = -1, ) -> dict: """Update a topic. Sentinels: title="" / description="" leave unchanged; order_index=-1 leaves unchanged. """ uid = current_user_id() fields: dict = {} if title: fields["title"] = title if description: fields["description"] = description if order_index >= 0: fields["order_index"] = order_index topic = await rulebooks_svc.update_topic(topic_id, uid, **fields) if topic is None: raise ValueError(f"topic {topic_id} not found") return topic.to_dict() async def delete_topic(topic_id: int, confirmed: bool = False) -> dict: """Delete a topic and all its rules. Requires confirmed=True.""" uid = current_user_id() topic = await rulebooks_svc.get_topic(topic_id, uid) if topic is None: raise ValueError(f"topic {topic_id} not found") if not confirmed: rules = await rulebooks_svc.list_rules(uid, topic_id=topic_id) return { "warning": ( f"Topic {topic_id} ('{topic.title}') contains {len(rules)} " f"rules; all will be deleted. Pass confirmed=True to proceed." ), "confirmed_required": True, } batch = await trash_svc.delete(uid, "topic", topic_id) return {"deleted": topic_id, "deleted_batch_id": batch, "message": f"Moved to trash. Restore with restore('{batch}')."} # ── Rule CRUD ────────────────────────────────────────────────────────── 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. 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( rulebook_id: int = 0, topic_id: int = 0, project_id: int = 0, ) -> dict: """List rules — filter by rulebook, topic, and/or project. Args: rulebook_id: 0 = no filter; positive = restrict to that rulebook. topic_id: 0 = no filter; positive = restrict to that topic. project_id: 0 = no filter; positive = restrict to rules applicable to that project (via its rulebook subscriptions). All filters are AND-combined; ownership-scoped. """ uid = current_user_id() rows = await rulebooks_svc.list_rules( user_id=uid, rulebook_id=rulebook_id or None, topic_id=topic_id or None, project_id=project_id or None, ) return {"rules": [_rule_summary(r) for r in rows], "total": len(rows)} async def list_always_on_rules(project_id: int = 0) -> dict: """Return all rules from rulebooks flagged always_on for the current user. Call this at session start. Treat the returned rules as binding for the session — they apply regardless of which project (if any) is in scope. Pair with get_project(id).applicable_rules when working on a specific project to also load that project's subscription-derived rules. Args: project_id: 0 (default) = the user-wide set. Inside a project, pass its id: an always-on rulebook the project EXCLUDED at inception (see enter_project's `excluded_always_on`) is left out — the project decided not to inherit it. """ uid = current_user_id() rules = await rulebooks_svc.list_always_on_rules(uid, project_id=project_id) return {"rules": [_rule_summary(r) for r in rules], "total": len(rules)} async def get_rule(rule_id: int) -> dict: """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 await rulebooks_svc.rule_detail(uid, rule) async def create_rule( topic_id: int, title: str, statement: str, when_to_apply: str = "", why: str = "", how_to_apply: str = "", order_index: int = 0, 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). A rulebook rule is shared by every project that gets the rulebook: an always_on rulebook binds ALL your projects; a subscribed rulebook binds the projects that opt in. So a rulebook rule must read as a general standard — never pin it to one project's files, paths, or quirks. For a rule that applies to a single project only, use create_project_rule instead (no rulebook+topic ceremony). If it's a standard a CATEGORY of projects shares, put it in a themed subscribed rulebook, not the always-on one. Before writing a rule at all, check whether another entity already models the thing. A rule is prose an agent must remember and apply; the others are structure a tool can resolve, render and check. Visual standards are a DESIGN SYSTEM (tokens inherit, resolve per mode, render to a stylesheet — none of that survives being prose). A repeatable procedure is a PROCESS. 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). 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, 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 await rulebooks_svc.rule_detail(uid, rule, system_ids) async def create_project_rule( project_id: int, statement: str, title: str = "", when_to_apply: str = "", why: str = "", how_to_apply: str = "", order_index: int = 0, 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). Use this for anything SPECIFIC to one project — its files, paths, layout, or quirks. This is the correct home for the project-specific detail that must NOT go into a shared rulebook (where it would leak to every other project that gets the rulebook). General standards belong in a rulebook instead (create_rule). It bypasses the Rulebook -> Topic -> Rule ceremony; 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). 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, 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 await rulebooks_svc.rule_detail(uid, rule, system_ids) async def update_rule( 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. 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: fields["how_to_apply"] = how_to_apply if order_index >= 0: fields["order_index"] = order_index rule = await rulebooks_svc.update_rule(rule_id, uid, **fields) if rule is None: raise ValueError(f"rule {rule_id} not found") return await rulebooks_svc.rule_detail(uid, rule, system_ids) async def delete_rule(rule_id: int, confirmed: bool = False) -> dict: """Move a rule to the trash (recoverable). Requires confirmed=True.""" 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") if not confirmed: return { "warning": ( f"Rule {rule_id} ('{rule.title}') will be moved to the trash " f"(recoverable via restore). Pass confirmed=True to proceed." ), "confirmed_required": True, } batch = await trash_svc.delete(uid, "rule", rule_id) return {"deleted": rule_id, "deleted_batch_id": batch, "message": f"Moved to trash. Restore with restore('{batch}')."} # ── Subscriptions ────────────────────────────────────────────────────── async def subscribe_project_to_rulebook( project_id: int, rulebook_id: int, ) -> dict: """Subscribe a project to a rulebook — its rules then bind that project. Subscription is the opt-in path for a non-always_on rulebook: a reusable, themed module of GENERAL rules shared across the projects that subscribe. Subscribe a project because it fits the rulebook's theme (e.g. a visual app -> the design-system rulebook), not to host rules about this one project — those belong in create_project_rule. """ uid = current_user_id() await rulebooks_svc.subscribe_project( project_id=project_id, rulebook_id=rulebook_id, user_id=uid, ) return {"project_id": project_id, "rulebook_id": rulebook_id, "subscribed": True} async def unsubscribe_project_from_rulebook( project_id: int, rulebook_id: int, ) -> dict: """Remove a project's subscription to a rulebook.""" uid = current_user_id() await rulebooks_svc.unsubscribe_project( project_id=project_id, rulebook_id=rulebook_id, user_id=uid, ) return {"project_id": project_id, "rulebook_id": rulebook_id, "subscribed": False} # ── Suppressions — project-level mute of rulebook rules / topics ──────── async def exclude_always_on_rulebook(project_id: int, rulebook_id: int) -> dict: """Opt a project OUT of a whole always-on rulebook (milestone 297). Always-on rulebooks bind every project implicitly; an inception decision can say "not this one, not here". The exclusion is total for that project — list_always_on_rules(project_id), enter_project/get_project rules and the session-start context all leave it out and name it under `excluded_always_on`. Owner-only; the rulebook must be always_on (a subscribed rulebook is left with unsubscribe_project_from_rulebook). Idempotent; include_always_on_rulebook reverses it. Normally reached via decide_project_inception, not by hand. """ uid = current_user_id() await rulebooks_svc.exclude_always_on_rulebook_for_project( project_id=project_id, rulebook_id=rulebook_id, user_id=uid, ) return {"project_id": project_id, "rulebook_id": rulebook_id, "excluded": True} async def include_always_on_rulebook(project_id: int, rulebook_id: int) -> dict: """Reverse exclude_always_on_rulebook: the always-on rulebook binds this project again. Idempotent.""" uid = current_user_id() await rulebooks_svc.include_always_on_rulebook_for_project( project_id=project_id, rulebook_id=rulebook_id, user_id=uid, ) return {"project_id": project_id, "rulebook_id": rulebook_id, "excluded": False} async def suppress_rule_for_project( project_id: int, rule_id: int, ) -> dict: """Mute a single rulebook rule for one project. The rule stays in its rulebook for other projects; only this project skips it. Idempotent. Use unsuppress_rule_for_project to re-enable. Project-scoped rules (create_project_rule) are NOT suppressible — delete them with delete_rule instead. """ uid = current_user_id() await rulebooks_svc.suppress_rule_for_project( project_id=project_id, rule_id=rule_id, user_id=uid, ) return {"project_id": project_id, "rule_id": rule_id, "suppressed": True} async def unsuppress_rule_for_project( project_id: int, rule_id: int, ) -> dict: """Re-enable a previously-suppressed rule for one project. Idempotent.""" uid = current_user_id() await rulebooks_svc.unsuppress_rule_for_project( project_id=project_id, rule_id=rule_id, user_id=uid, ) return {"project_id": project_id, "rule_id": rule_id, "suppressed": False} async def suppress_topic_for_project( project_id: int, topic_id: int, ) -> dict: """Mute every rule under a topic for one project. Equivalent to suppressing each rule in the topic individually, but auto-includes new rules added to the topic later. Idempotent. """ uid = current_user_id() await rulebooks_svc.suppress_topic_for_project( project_id=project_id, topic_id=topic_id, user_id=uid, ) return {"project_id": project_id, "topic_id": topic_id, "suppressed": True} async def unsuppress_topic_for_project( project_id: int, topic_id: int, ) -> dict: """Re-enable a previously-suppressed topic for one project. Idempotent.""" uid = current_user_id() await rulebooks_svc.unsuppress_topic_for_project( project_id=project_id, topic_id=topic_id, user_id=uid, ) 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, exclude_always_on_rulebook, include_always_on_rulebook, ): mcp.tool(name=fn.__name__)(fn)