bvandeusen
dd1fc2d506
Completes the Phase 5 follow-up: rules now get the same update-over-create gate. Title-based only (rules aren't a semantic-retrieval/RAG surface), scoped to the same topic (rulebook rule) or same project (project rule). force=true overrides; fail-open like the note/task gate. Deferred-item decisions (operator): REST/web gating SKIPPED (kept MCP-only — humans rarely double-create and a hard block needs UI affordance); orphan scope kept orphan↔orphan (no change). So this rule gate is the only remaining build. - services/dedup.py: find_duplicate_rule(title, topic_id|project_id). - create_rule + create_project_rule: force param + gate. - tests: rule title match, scope-required guard, tool gate (block + force). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
478 lines
18 KiB
Python
478 lines
18 KiB
Python
"""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.
|
|
|
|
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 (e.g. "FabledSword family").
|
|
- 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 (e.g. "FabledSword family").
|
|
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 ──────────────────────────────────────────────────────────
|
|
|
|
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": [
|
|
{
|
|
"id": r.id, "title": r.title, "statement": r.statement,
|
|
"topic_id": r.topic_id,
|
|
}
|
|
for r in rows
|
|
],
|
|
"total": len(rows),
|
|
}
|
|
|
|
|
|
async def list_always_on_rules() -> 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.
|
|
"""
|
|
uid = current_user_id()
|
|
rules = await rulebooks_svc.list_always_on_rules(uid)
|
|
return {
|
|
"rules": [
|
|
{
|
|
"id": r.id, "title": r.title, "statement": r.statement,
|
|
"topic_id": r.topic_id,
|
|
}
|
|
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."""
|
|
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()
|
|
|
|
|
|
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).
|
|
|
|
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.
|
|
|
|
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.
|
|
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,
|
|
why=why, how_to_apply=how_to_apply, order_index=order_index,
|
|
)
|
|
return rule.to_dict()
|
|
|
|
|
|
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).
|
|
|
|
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=...).
|
|
|
|
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.
|
|
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,
|
|
why=why, how_to_apply=how_to_apply, order_index=order_index,
|
|
)
|
|
return rule.to_dict()
|
|
|
|
|
|
async def update_rule(
|
|
rule_id: int, title: str = "", statement: str = "",
|
|
why: str = "", how_to_apply: str = "", order_index: int = -1,
|
|
) -> dict:
|
|
"""Update a rule. Empty strings / order_index=-1 leave fields unchanged."""
|
|
uid = current_user_id()
|
|
fields: dict = {}
|
|
if title:
|
|
fields["title"] = title
|
|
if statement:
|
|
fields["statement"] = statement
|
|
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 rule.to_dict()
|
|
|
|
|
|
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 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}
|
|
|
|
|
|
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,
|
|
subscribe_project_to_rulebook, unsubscribe_project_from_rulebook,
|
|
suppress_rule_for_project, unsuppress_rule_for_project,
|
|
suppress_topic_for_project, unsuppress_topic_for_project,
|
|
):
|
|
mcp.tool(name=fn.__name__)(fn)
|