feat(rules): a rule can say when it applies, which area it is about, and what it belongs with (#3029, milestone 307 step 3, schema)

A rule could not state its trigger, its area, or its siblings, so all three
were being written as prose instead: a System's charter restating rule text,
a `why` naming the note that caused it, and two halves of one shape merged
into a single row because either could surface without the other.

Migration 0088 adds the four fields those workarounds stood in for:

- `when_to_apply` — the trigger. Nullable in the DB and required at the
  service layer: existing rules have none and a migration cannot invent one.
- `tier` — always_on | conditional, defaulting to always_on. This migration
  therefore changes NOTHING about which rules bind; an install upgrades and
  every rule keeps arriving exactly as before. Getting that backwards is the
  one failure this milestone exists to prevent, so _valid_tier falls back to
  always_on rather than silently un-binding a rule with a typo'd tier.
- `arose_from_id` — the record that caused the rule, the edge notes and tasks
  already have. SET NULL: trashing the source does not repeal the rule.
- `rule_systems` / `rule_relations` — the canon tag and the typed edges
  (co_surfaces / overrides / elaborates), each earned from a workaround its
  absence forced.

rule_brief() replaces the THREE hand-written trim dicts that had already
diverged — two carried topic_id, one didn't, and none carried the timestamps
the model has held all along. That omission is why a rule written before the
capability it duplicates was indistinguishable at read time from one still
doing work. It now carries updated_at as a DATE: the question is "how old is
this", and a full stamp across the always-on set is ~2k characters for
precision nobody reads. The two callers select the ENTITY rather than a column
list, so rule_brief stays the single place deciding what a surfaced rule says.

Backup: both new tables carried, area tags by canonical SLUG (ids are
per-install). The rule-relation restore runs after ALL rules exist and after
the catalog, because an edge names two rules and a tag names a global row —
sections renumbered so the file reads in dependency order. A pre-0088 payload
restores with tier=always_on, i.e. binding exactly as when it was taken.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/scribe/services/rulebooks.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import logging
 from typing import Optional
 
-from sqlalchemy import select
+from sqlalchemy import delete as sql_delete, insert, select
 
 from scribe.models import async_session
 from scribe.models.rulebook import Rulebook
@@ -223,7 +223,7 @@ async def delete_topic(topic_id: int, user_id: int) -> None:
 
 # ── Rule CRUD ──────────────────────────────────────────────────────────
 
-from scribe.models.rulebook import Rule
+from scribe.models.rulebook import Rule, RuleRelation
 
 
 async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
@@ -280,9 +280,64 @@ async def _assert_rulebook_rule_owned(session, rule_id: int, user_id: int) -> No
         raise ValueError(f"rule {rule_id} not found or not a rulebook rule")
 
 
+# The vocabularies migration 0088's CHECK constraints enforce. Named here so
+# a caller can be corrected before the database refuses it (rule 36 keeps the
+# two in step; this keeps the error readable).
+TIERS = ("always_on", "conditional")
+RELATION_KINDS = ("co_surfaces", "overrides", "elaborates")
+
+
+def _valid_tier(tier: str) -> str:
+    """An unrecognised tier falls back to always_on — the SAFE direction.
+
+    Getting this wrong the other way would silently stop a rule binding, which
+    is the one failure this whole milestone exists to prevent. A rule that
+    preloads when it did not need to costs context; a rule that quietly stops
+    preloading costs the behaviour it was written for.
+    """
+    return tier if tier in TIERS else "always_on"
+
+
+def rule_brief(rule: Rule, **extra) -> dict:
+    """The shape a rule takes when it is SURFACED rather than opened.
+
+    One builder for every payload that hands rules to an agent, because there
+    were three copies of this dict and they had already diverged — two carried
+    `topic_id`, one didn't, and none carried the timestamps the model has held
+    all along. That omission is why a rule written before the capability it
+    duplicates was indistinguishable, at read time, from one still doing work
+    (the FabledCurator case, note 3026).
+
+    `updated_at` is a DATE, not a stamp: the question it answers is "how old
+    is this?", and a full ISO string across an always-on set is ~2k characters
+    of payload for a precision nobody reads.
+
+    `why` and `how_to_apply` are deliberately NOT here — they are the depth a
+    caller gets from get_rule, and putting them in every listing is the bloat
+    this milestone is about.
+    """
+    out = {
+        "id": rule.id,
+        "title": rule.title,
+        "statement": rule.statement,
+        "topic_id": rule.topic_id,
+        "tier": rule.tier,
+        "updated_at": rule.updated_at.date().isoformat() if rule.updated_at else None,
+    }
+    # Attached only when present (#2483: never a null key that reads as a
+    # capability the record doesn't have).
+    if rule.when_to_apply:
+        out["when_to_apply"] = rule.when_to_apply
+    if rule.arose_from_id:
+        out["arose_from_id"] = rule.arose_from_id
+    out.update({k: v for k, v in extra.items() if v is not None})
+    return out
+
+
 async def create_rule(
     topic_id: int, user_id: int, title: str, statement: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
+    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
 ) -> Rule:
     async with async_session() as session:
         await _assert_topic_owned(session, topic_id, user_id)
@@ -290,8 +345,11 @@ async def create_rule(
             topic_id=topic_id,
             title=title,
             statement=statement,
+            when_to_apply=when_to_apply or None,
+            tier=_valid_tier(tier),
             why=why or None,
             how_to_apply=how_to_apply or None,
+            arose_from_id=arose_from_id or None,
             order_index=order_index,
         )
         session.add(rule)
@@ -303,6 +361,7 @@ async def create_rule(
 async def create_project_rule(
     project_id: int, user_id: int, title: str, statement: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
+    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
 ) -> Rule:
     """Create a rule scoped to a single project (no rulebook ceremony).
 
@@ -316,8 +375,11 @@ async def create_project_rule(
             project_id=project_id,
             title=title,
             statement=statement,
+            when_to_apply=when_to_apply or None,
+            tier=_valid_tier(tier),
             why=why or None,
             how_to_apply=how_to_apply or None,
+            arose_from_id=arose_from_id or None,
             order_index=order_index,
         )
         session.add(rule)
@@ -513,15 +575,174 @@ async def update_rule(rule_id: int, user_id: int, **fields) -> Optional[Rule]:
         rule = await _fetch_owned_rule(session, rule_id, user_id)
         if rule is None:
             return None
-        allowed = {"title", "statement", "why", "how_to_apply", "order_index"}
+        allowed = {
+            "title", "statement", "why", "how_to_apply", "order_index",
+            "when_to_apply", "tier", "arose_from_id",
+        }
         for key, value in fields.items():
             if key in allowed and value is not None:
-                setattr(rule, key, value)
+                setattr(rule, key, _valid_tier(value) if key == "tier" else value)
         await session.commit()
         await session.refresh(rule)
         return rule
 
 
+# ── Canon tags + typed edges (milestone 307) ───────────────────────────
+
+async def set_rule_systems(
+    rule_id: int, user_id: int, canonical_ids: list[int],
+) -> list[int] | None:
+    """Replace which global AREAS a rule is about. None if not owned.
+
+    Set-semantics like set_record_systems: the list given IS the state after,
+    so an empty list clears the tags. Points at the canonical catalog, never a
+    project's System — a family rule tagged to one project's row would bind
+    itself to that project's vocabulary.
+    """
+    from scribe.models.canonical_system import CanonicalSystem
+    from scribe.models.rulebook import rule_systems as rule_systems_t
+
+    async with async_session() as session:
+        rule = await _fetch_owned_rule(session, rule_id, user_id)
+        if rule is None:
+            return None
+        wanted = set(canonical_ids or [])
+        if wanted:
+            live = set((await session.execute(
+                select(CanonicalSystem.id).where(
+                    CanonicalSystem.id.in_(wanted),
+                    CanonicalSystem.deleted_at.is_(None),
+                )
+            )).scalars().all())
+            # Silently dropping an unknown id would leave the caller believing
+            # a tag exists; keep only the live ones and report what stuck.
+            wanted &= live
+        await session.execute(
+            sql_delete(rule_systems_t).where(rule_systems_t.c.rule_id == rule_id)
+        )
+        for canonical_id in sorted(wanted):
+            await session.execute(
+                insert(rule_systems_t).values(rule_id=rule_id, canonical_id=canonical_id)
+            )
+        await session.commit()
+        return sorted(wanted)
+
+
+async def list_rule_systems(rule_ids: list[int]) -> dict[int, list[dict]]:
+    """The canon tags for a batch of rules, keyed by rule id.
+
+    Batched on purpose: the surfacing paths ask about a whole payload of rules
+    at once, and one query per rule would turn every session start into an
+    N+1.
+    """
+    from scribe.models.canonical_system import CanonicalSystem
+    from scribe.models.rulebook import rule_systems as rule_systems_t
+
+    if not rule_ids:
+        return {}
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(rule_systems_t.c.rule_id, CanonicalSystem.id, CanonicalSystem.name)
+            .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
+            .where(
+                rule_systems_t.c.rule_id.in_(rule_ids),
+                CanonicalSystem.deleted_at.is_(None),
+            )
+            .order_by(CanonicalSystem.order_index)
+        )).all()
+    out: dict[int, list[dict]] = {}
+    for rule_id, canonical_id, name in rows:
+        out.setdefault(rule_id, []).append({"id": canonical_id, "name": name})
+    return out
+
+
+async def add_rule_relation(
+    user_id: int, from_rule_id: int, to_rule_id: int, kind: str, note: str = "",
+) -> RuleRelation | None:
+    """Draw a typed edge between two rules. None if either isn't owned.
+
+    Both ends are ownership-checked: an edge is only meaningful if the drawer
+    can see both rules, and a one-sided edge would surface a rule the caller
+    has no business reading.
+
+    Idempotent — re-drawing an existing edge returns it rather than raising, so
+    a true-up pass can be re-run without cleaning up first.
+    """
+    if kind not in RELATION_KINDS:
+        raise ValueError(f"kind must be one of {RELATION_KINDS}, got {kind!r}")
+    if from_rule_id == to_rule_id:
+        raise ValueError("a rule cannot relate to itself")
+    async with async_session() as session:
+        for rid in (from_rule_id, to_rule_id):
+            if await _fetch_owned_rule(session, rid, user_id) is None:
+                return None
+        existing = await session.scalar(
+            select(RuleRelation).where(
+                RuleRelation.from_rule_id == from_rule_id,
+                RuleRelation.to_rule_id == to_rule_id,
+                RuleRelation.kind == kind,
+            )
+        )
+        if existing is not None:
+            return existing
+        relation = RuleRelation(
+            from_rule_id=from_rule_id, to_rule_id=to_rule_id,
+            kind=kind, note=note or None,
+        )
+        session.add(relation)
+        await session.commit()
+        await session.refresh(relation)
+        return relation
+
+
+async def remove_rule_relation(user_id: int, relation_id: int) -> bool:
+    async with async_session() as session:
+        relation = await session.get(RuleRelation, relation_id)
+        if relation is None:
+            return False
+        if await _fetch_owned_rule(session, relation.from_rule_id, user_id) is None:
+            return False
+        await session.delete(relation)
+        await session.commit()
+        return True
+
+
+async def list_rule_relations(rule_ids: list[int]) -> dict[int, list[dict]]:
+    """Edges touching a batch of rules, keyed by rule id.
+
+    `co_surfaces` is reported from BOTH ends off a single stored row — it means
+    "these fail together", which is not a claim with a direction. The other two
+    are directional and are reported as stored, with `direction` naming which
+    end this rule is: an override read from the wrong end would invert what it
+    says.
+    """
+    if not rule_ids:
+        return {}
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(RuleRelation).where(
+                (RuleRelation.from_rule_id.in_(rule_ids))
+                | (RuleRelation.to_rule_id.in_(rule_ids))
+            )
+        )).scalars().all()
+    out: dict[int, list[dict]] = {}
+    wanted = set(rule_ids)
+    for relation in rows:
+        if relation.from_rule_id in wanted:
+            out.setdefault(relation.from_rule_id, []).append({
+                "id": relation.id, "kind": relation.kind,
+                "rule_id": relation.to_rule_id,
+                "direction": "outgoing", "note": relation.note or "",
+            })
+        if relation.to_rule_id in wanted:
+            out.setdefault(relation.to_rule_id, []).append({
+                "id": relation.id, "kind": relation.kind,
+                "rule_id": relation.from_rule_id,
+                "direction": "incoming", "note": relation.note or "",
+            })
+    return out
+
+
 async def delete_rule(rule_id: int, user_id: int) -> None:
     async with async_session() as session:
         rule = await _fetch_owned_rule(session, rule_id, user_id)
@@ -533,7 +754,6 @@ async def delete_rule(rule_id: int, user_id: int) -> None:
 
 # ── Subscriptions + get_applicable_rules ───────────────────────────────
 
-from sqlalchemy import insert, delete as sql_delete
 from sqlalchemy.exc import IntegrityError
 
 
@@ -802,10 +1022,13 @@ async def get_applicable_rules(
         # Applicable rules (limit + 1 so we can detect truncation). Filter
         # in SQL so truncation reflects the post-suppression count, not the
         # raw subscription count.
+        # Selects the ENTITY, not a column list: rule_brief is the one place
+        # that decides which fields a surfaced rule carries, and a column list
+        # here would be a second such decision to keep in step. The row count
+        # is bounded by `limit`, so this is a listing, not a scan.
         rules_q = (
             select(
-                Rule.id, Rule.title, Rule.statement,
-                RulebookTopic.id.label("topic_id"),
+                Rule,
                 RulebookTopic.title.label("topic_title"),
                 Rulebook.id.label("rulebook_id"),
                 Rulebook.title.label("rulebook_title"),
@@ -838,18 +1061,14 @@ async def get_applicable_rules(
         rule_rows = (await session.execute(rules_q)).all()
         truncated = len(rule_rows) > limit
         rules = [
-            {
-                "id": rid, "title": rtitle, "statement": stmt,
-                "topic_id": ti, "topic_title": tt,
-                "rulebook_id": rbi, "rulebook_title": rbt,
-            }
-            for rid, rtitle, stmt, ti, tt, rbi, rbt in rule_rows[:limit]
+            rule_brief(rule, topic_title=tt, rulebook_id=rbi, rulebook_title=rbt)
+            for rule, tt, rbi, rbt in rule_rows[:limit]
         ]
 
         # Project-scoped rules — verifies ownership via Project.user_id.
         from scribe.models.project import Project
         proj_rules_q = (
-            select(Rule.id, Rule.title, Rule.statement)
+            select(Rule)
             .join(Project, Rule.project_id == Project.id)
             .where(
                 Project.user_id == user_id,
@@ -860,10 +1079,7 @@ async def get_applicable_rules(
             .order_by(Rule.order_index, Rule.title)
         )
         proj_rule_rows = (await session.execute(proj_rules_q)).all()
-        project_rules = [
-            {"id": rid, "title": rtitle, "statement": stmt}
-            for rid, rtitle, stmt in proj_rule_rows
-        ]
+        project_rules = [rule_brief(rule) for (rule,) in proj_rule_rows]
 
     return {
         "rules": rules,