Files
FabledScribe/src/scribe/mcp/tools/rulebooks.py
T
bvandeusenandClaude Opus 5 c3ecdf0972
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 10s
CI & Build / TypeScript typecheck (push) Successful in 11s
CI & Build / integration (push) Successful in 32s
CI & Build / Python tests (push) Successful in 1m6s
CI & Build / Build & push image (push) Successful in 25s
feat(rules): the rule gate becomes a practice with a question, not a prohibition (#3557)
The first cut opened "NOT YOURS TO CALL UNPROMPTED", and that is the wrong
instrument. A caller reading a prohibition stops NOTICING rule-shaped things
rather than noticing them and asking — which trades a small failure for a
larger one. The wanted behaviour is more proposals, not fewer.

So both docstrings now describe the practice: propose readily, state the
four things, and close with a question the operator answers in one word —
approve it as written / let's talk about it / no. Named options where the
interface has them, three written-out options where it does not.

"Approve it as written" is what makes element 1 load-bearing: they approved
TEXT, so that text is stored verbatim. "Let's talk about it" is framed as
the expected answer rather than a setback. "No" routes the observation to
create_note, which records without binding.

The argument for asking is also better than consent. The operator's yes is
the one moment the rule is certainly in front of them: afterwards a
conditional rule is not read aloud at session start, and a project rule is
absent from an unfiltered list_rules(). The proposal IS the review.

Guard gains the answers-offered-back element and drops the wording that
forbade; its header records why the framing changed, so the prohibition does
not get reintroduced as a tidy-up.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ
2026-09-04 20:11:42 -04:00

997 lines
47 KiB
Python

"""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
from scribe.services.rule_usage import record_rule_pulled, record_rule_surfaced
# ── 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, "title": rb.title, "deleted_batch_id": batch,
"message": f'Rulebook {rulebook_id} ("{rb.title}") moved to trash. '
f"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, "title": topic.title, "deleted_batch_id": batch,
"message": f'Topic {topic_id} ("{topic.title}") moved to trash. '
f"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.
Returns the ALWAYS-ON tier only (milestone 307). A `conditional` rule is
still binding when it applies; it just is not resident — it reaches a
session through enter_project (when the project works in an area the rule
is tagged to) or through search(content_type="rule"). Nothing here is a
behaviour change until rules are actually re-tiered: `tier` defaults to
always_on, so an existing rulebook returns exactly what it always did.
Pair with get_project(id).applicable_rules when working on a specific
project to also load that project's subscription-derived rules.
A rule carrying `last_verified` asserts a FACT about something outside the
operator's control — a runner's shell, a tool's existence, a setting
somewhere. It is still binding; the field says how long ago anyone
confirmed it, and "never" means nobody has. Follow the rule, and if you
are already standing where the check could be made, make it: get_rule
gives you its `verify_with`. Most rules have no such field, which means
they are decisions and there is nothing to check.
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)
# AMBIENT source: the resident set, handed over whole. No ranker chose
# these, so they must not land in the pull-through numerator's denominator
# — but they must land SOMEWHERE, or the largest rule surface in the
# product stays the one surface its own scoreboard cannot see (#3473).
record_rule_surfaced(
user_id=uid,
rule_ids=[r.id for r in rules],
source="list_always_on_rules",
)
return {
"rules": [_rule_summary(r) for r in rules],
"total": len(rules),
# A marker for the set you are now holding. It is not for you to read:
# the write-path hook carries it back and is told if these rules have
# moved since. Deliberately NOT on rules_payload's applicable_rules —
# that is a DIFFERENT set (subscription-derived), and one key name
# over two sets is how a comparison starts reporting phantom changes.
"rules_etag": rulebooks_svc.rules_etag(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")
# THE pull that matters. The write-path rule arm's own message ends "Read
# it with get_rule(N)", so this is the exact action the hint asks for and
# the only evidence that one landed. Recorded after the access check, so a
# refused read is not counted as a pull.
record_rule_pulled(user_id=uid, rule_id=int(rule.id), source="mcp_get_rule")
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, verify_with: str = "", expires_when: str = "",
force: bool = False,
) -> dict:
"""Create a new rule in a rulebook (a SHARED rule — keep it general).
PROPOSE RULES READILY, AND WRITE ONE WHEN THE OPERATOR SAYS YES. Noticing
that something has hardened into a standing instruction is valuable work,
and a session that notices it and says nothing has thrown the observation
away. So raise it whenever you see one. The single step that belongs
between noticing and writing is the operator's yes: a rule binds every
future session, and they are the person it binds.
Their yes is also the only moment the rule is reliably IN FRONT of them.
After the write it may not be again for months — a conditional rule is not
read aloud at session start, and a project-scoped one does not appear in
an unfiltered list_rules() at all. So the proposal is the review.
When the operator asks for a rule in so many words, that IS the yes —
write it and move on. The loop below is for the rule you thought of.
A PROPOSAL CARRIES FOUR THINGS, and the fourth is the one that decides it:
1. WHAT it would require — the statement, in the words it would carry,
not a gloss of them. The operator is agreeing to text.
2. INTENT — what it changes about how work gets done, and what goes
wrong today without it. "Be careful about X" is not an intent; the
behaviour that would differ tomorrow is.
3. WHY NOW — the incident, observation or decision behind it. Pass that
record as arose_from_id, and say it in the conversation too: the
field is for the reader six months out, the sentence is for the
person deciding.
4. HOW IT WOULD BE ENFORCED — a test, a CI check, a hook, a schema
constraint, a duplicate gate, a review step... or nothing, in which
case say so plainly: "nothing — this is prose a session has to
remember." Answer this one honestly and it will sometimes dissolve
the rule, which is the point rather than a side effect. What a test
can assert should BE that test; a rule is what remains when nothing
mechanical can hold the thing. A rulebook grows by default and
shrinks only on purpose, so a question that prevents a rule is worth
more than any question that improves one's wording.
THEN CLOSE WITH A QUESTION THEY CAN ANSWER IN ONE WORD. Offer three
answers, and make the middle one the easy one:
* "Approve it AS WRITTEN" — you create it with the statement exactly as
shown. This is what makes element 1 load-bearing: they approved TEXT,
so that text is what gets stored, verbatim.
* "LET'S TALK ABOUT IT" — the wording, the scope, the tier, whether it
wants to be a rule at all. Most good rules arrive this way, so treat
this answer as the expected one rather than a setback.
* "NO" — let it go. If the observation is still worth keeping, it is a
note (create_note): recorded, findable, and binding on nobody.
Where the interface offers structured choices, ask it that way — a
question with named options is answered in a click, while the same
question inside a paragraph is answered by scrolling past. Where it does
not, write the three options out as three options. Either way ask once
and let the answer stand; re-raising a declined proposal argues a rule
into existence, which is the thing this whole loop exists to prevent.
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.
Write it general WITHOUT hedging for the exceptions. A project that needs
to strengthen, narrow or replace this rule writes its own and links it
with relate_rules(kind="overrides"), and one that adds local specifics
uses "elaborates" — so the general form does not have to anticipate every
project it will ever reach. A rulebook rule padded with "unless…" clauses
for two projects is two project rules that were never written.
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.
This field is also the rule's RETRIEVAL SURFACE — it and the
statement are what a search is matched against, so it should
carry the SYMPTOM, not just the situation: the words someone
would actually type while stuck. Measured (note 3078): a rule
whose trigger named only its situation did not surface at all
for the problem it solves; adding the symptom to the same field
brought it back as the top hit. Where a rule prevents a specific
failure, put that failure's vocabulary here — the error text,
the wrong behaviour, the dead end.
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.
verify_with: How to CHECK this rule is still true. Set it only when
the rule asserts a fact about something outside your control — a
runner's shell, a bot's config, whether a tool exists. Those go
false silently, with nobody present. Give a command, a path, a
URL or a query; something runnable beats prose, because prose
has to be re-interpreted by whoever finds it.
LEAVE IT EMPTY for a rule that is a DECISION — a preference, a
standard, a way of working. A decision has no truth value: it
changes when you change it, and you know that you did. An empty
verify_with is not a gap, it is the marker for "there is nothing
to go and check," and the whole signal is worthless the moment
it is filled in out of tidiness.
expires_when: The STATE under which this rule stops being true —
"when the runner can be given a bash shell", "when the dashboard
approval setting is turned off". Deliberately not a date: a
constraint expires when the ground under it moves, not on a
schedule. Pairs with verify_with; both empty is the normal case.
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,
verify_with=verify_with, expires_when=expires_when,
)
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, verify_with: str = "", expires_when: str = "",
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=...).
PROPOSE, THEN WRITE ON A YES — create_rule's opening carries the whole
loop: the four things a proposal states (what it would require, its
intent, why now, and how it would be enforced) and the one-word question
that closes it (approve as written / talk about it / no). All of it
applies here unchanged. Reach for that loop MORE readily on this surface,
not less: a project rule stays out of an unfiltered list_rules(), and a
conditional one stays out of session start too, so the operator's yes is
the one moment this rule is certain to have been seen by the person it
binds.
Check first whether a rule is the right shape at all — create_rule's
opening asks that question and it applies identically here. A visual
standard is a design system; a procedure is a process (create_process);
reusable code is a snippet (create_snippet). Each of those is structure a
tool can resolve, render and check, where a rule is only prose someone
has to remember and apply.
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, and the rule's retrieval surface: name the SYMPTOM,
the words someone would type while stuck. See create_rule for the
full argument. It informs the tier below rather than deciding it,
since a project rule's tier turns on area-scope, not on whether
the trigger can be named.
tier: "always_on" (default) or "conditional". The SAME two values as
create_rule, judged against a different cost — do not import that
tool's test wholesale. There, always_on means every session in
every project, so the bar is high: the trigger must be nameless
("whenever you are working"). Here the rule is already scoped to
one project by construction, so always_on costs only that
project's sessions and the bar is correspondingly lower. A
project rule that names something specific is still ordinarily
always_on — being specific is what project rules are FOR.
Reach for conditional when the rule is about one AREA of a large
project — a CI quirk, a migration gotcha, one subsystem's
convention — so it arrives with that area instead of resident in
every session. The failure to avoid is local: forty always-on
rules on one project reproduces, inside that project, exactly the
preload bloat that made every rule compete for the same budget.
system_ids: Ids from list_canonical_systems — the global AREAS this
rule is about. Worth setting even on a project rule: it is what
lets a conditional one surface when the project is working in
that area.
arose_from_id: The note or task that CAUSED this rule. Reach for it
harder here than on a rulebook rule — a project rule usually
comes from one traceable incident in this repo, where a family
rule is more often a standing preference with no single origin.
The link is what lets a later reader judge whether the incident
still describes the project.
why: Optional rationale — the reason the rule exists.
how_to_apply: Optional operationalization — when / where it kicks in.
verify_with: How to check this rule is still true — see create_rule.
Set it when the rule asserts a fact about someone else's software;
leave it empty when the rule is a decision. Project rules are the
likelier home for a real check: they name this project's files,
paths and quirks, which is exactly the kind of claim that rots.
expires_when: The state under which the rule stops being true — see
create_rule. A state, not a date.
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,
verify_with=verify_with, expires_when=expires_when,
)
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,
verify_with: str = "", expires_when: str = "",
clear_fields: list[str] | None = None,
) -> 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).
TO EMPTY A FIELD, NAME IT: clear_fields=["verify_with"]. Passing "" cannot
do it — "" means "leave this alone" here, which is what lets you update
two fields without wiping the other six. Clearable: why, how_to_apply,
when_to_apply, verify_with, expires_when, arose_from_id. Clearing and
setting the same field in one call clears it first, so the new value wins.
Editing `verify_with` DROPS the rule's verification stamp. The stamp
certifies a check, not a rule; once the check is reworded the old stamp
vouches for something that no longer exists, so the rule re-enters the
staleness sweep as never-verified.
Args:
verify_with: How to check the rule is still true — set it when the
rule asserts a fact about someone else's software, leave it empty
when the rule is a decision. See create_rule.
expires_when: The state under which the rule stops being true. A
state, not a date. See create_rule.
clear_fields: Names of fields to empty, as above.
"""
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 verify_with:
fields["verify_with"] = verify_with
if expires_when:
fields["expires_when"] = expires_when
if order_index >= 0:
fields["order_index"] = order_index
rule = await rulebooks_svc.update_rule(
rule_id, uid, clear=clear_fields or (), **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 rule_history(rule_id: int, version_id: int = 0) -> dict:
"""What a rule USED TO SAY, newest change first.
Read this before you argue with a rule, and before you rewrite one. A
rule that has been reworded may have been reworded for a reason you are
about to rediscover the hard way — and the wording it replaced is often
the fastest way to see what the current one is guarding against. The
rescoping of rule 79 is the case this exists for: the superseded
statement had to be hand-copied into a task log to survive the edit.
EACH ENTRY HOLDS THE TEXT THE EDIT REPLACED, not the text it introduced.
So "what did this say before the most recent change?" is the first entry,
and the text the change PRODUCED is the rule as it stands now — read that
with get_rule. Pair the two and you have the diff.
An empty history is ordinary and means the rule has never been reworded,
not that its history was lost. Nothing is written before milestone 323,
so a rule edited before then starts empty too.
Args:
rule_id: The rule whose history to read.
version_id: 0 (default) lists the history — when each change
happened, by whom, and the title as it then stood. Pass an id
from that list to read that snapshot IN FULL. The list omits
statement and why on purpose: a rule's statement runs to
thousands of characters, and a history carrying every field would
cost more to read than the answer is worth.
There is deliberately no restore. Putting an old wording back is a
decision, so it goes through update_rule — which snapshots what it
replaces, leaving the undo visible in the history like any other edit. A
one-click revert would erase the only record of why the rewrite happened.
"""
uid = current_user_id()
if version_id:
version = await rulebooks_svc.get_rule_version(rule_id, version_id, uid)
if version is None:
raise ValueError(
f"version {version_id} not found on rule {rule_id}"
)
return version.to_dict(include_text=True)
versions = await rulebooks_svc.list_rule_versions(rule_id, uid)
if versions is None:
raise ValueError(f"rule {rule_id} not found")
# Fail-open, like the deletes: a missing title must not turn a readable
# history into an error.
try:
rule = await rulebooks_svc.get_rule(rule_id, uid)
except Exception:
rule = None
return {
"rule_id": rule_id,
"title": rule.title if rule else "",
"versions": [v.to_dict(include_text=False) for v in versions],
"total": len(versions),
# Said in-band because an empty list is the ordinary case and reads
# like a missing feature otherwise.
"note": (
"Each entry holds the text the edit REPLACED. The current wording "
"is on the rule itself — get_rule(%d)." % rule_id
if versions else
"This rule has never been reworded."
),
}
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, "title": rule.title, "deleted_batch_id": batch,
"message": f'Rule {rule_id} ("{rule.title}") moved to trash. '
f"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}
# ── The staleness sweep (milestone 312) ────────────────────────────────
async def rules_due_for_verification(
older_than_days: int = 0, tier: str = "", never_only: bool = False,
) -> dict:
"""Which standing rules assert a FACT that nobody has confirmed lately.
A rulebook holds two kinds of thing. Most rules are DECISIONS — how the
operator wants to work. They have no truth value and cannot rot. A few
assert a fact about someone else's software: what a CI runner does, which
tools exist, what a setting is currently set to. Those go false silently,
with nobody present, and they keep being handed to every session as
binding instructions long after they stopped being true.
This lists the second kind, oldest verification first, never-checked at
the top. Each row carries the rule's `verify_with` in full — you are
about to go and run it — plus `expires_when`, and `days_since_verified`.
Reach for it when you are curating the rulebook, when a rule's advice
just contradicted what you observed, or periodically. Then, for each row:
run the check, and call mark_rule_verified with what you found.
Rules with no `verify_with` never appear here. That is correct: they are
decisions, and there is nothing to go and check. Do not "fix" their
absence by giving them checks — the list is only worth reading while
everything on it genuinely can go false.
Args:
older_than_days: only rules last verified longer ago than this.
Never-checked rules always qualify. 0 = no age filter.
tier: "always_on" or "conditional" to narrow. An always-on constraint
that has gone false is the expensive kind — it is preloaded into
every session, so a wrong one is wrong everywhere at once.
never_only: only rules nobody has ever verified.
NOT filterable by project, deliberately: a project reaches rules through
project scope, subscriptions, always-on rulebooks and exclusions, and a
filter that missed one of those paths would UNDER-report — which is the
exact failure this whole surface exists to prevent. Read the whole list.
"""
uid = current_user_id()
rules = await rulebooks_svc.rules_due_for_verification(
uid, older_than_days=older_than_days, tier=tier, never_only=never_only,
)
return {
"rules": [rulebooks_svc.verification_row(r) for r in rules],
"total": len(rules),
}
async def mark_rule_verified(rule_id: int, still_true: bool = True) -> dict:
"""Record that you ran a rule's check — and what it said.
Call this AFTER actually running the rule's `verify_with`, never on the
strength of the rule sounding plausible. A stamp nobody earned is worse
than no stamp: it moves the rule to the bottom of the sweep and buys it
another long silence.
`still_true=False` writes NOTHING. A rule whose check failed is not in a
special state to be recorded — it is WRONG, and the only honest next
moves are to correct it, retire it, or find out why. So it stays at the
top of the sweep until someone deals with it, and the response tells you
what the rule said would end it.
Args:
rule_id: the rule whose check you ran.
still_true: True if the check passed. False if the fact it asserts is
no longer true — say so, that is the outcome worth having.
"""
uid = current_user_id()
rule = await rulebooks_svc.mark_rule_verified(rule_id, uid, still_true)
if rule is None:
raise ValueError(
f"rule {rule_id} not found, or carries no verify_with "
f"(nothing to verify is not the same as verified)"
)
data = await rulebooks_svc.rule_detail(uid, rule)
if still_true:
data["verified"] = True
return data
data["verified"] = False
data["next"] = (
"This rule is no longer true and is still binding on every session "
"that loads it. Correct it with update_rule, retire it with "
"delete_rule, or open a task to work out what replaced it. Its "
"verified_at is deliberately untouched, so it stays at the top of "
"rules_due_for_verification until one of those happens."
)
return data
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,
rules_due_for_verification, mark_rule_verified,
rule_history,
):
mcp.tool(name=fn.__name__)(fn)