feat(rules)!: retire rulebook subscriptions and per-project suppressions (#4052)

A rule's home is its scope now: a rule in a rulebook topic is global, a rule on
a project applies to that project, and retrieval reads that directly (#4074).
A subscription had stopped changing anything a session received; a suppression
muted rules from a subscription. Operator, 2026-09-15: "we have global and
project scoped rules, we don't need the subscriptions now."

What goes, whole (rule 22):
- Migration 0101 drops project_rulebook_subscriptions, project_rule_suppressions
  and project_topic_suppressions, and strips subscribe_rulebooks (and 394's
  leftover exclude_always_on_rulebooks) from stored inception choices.
- Service, MCP and REST: subscribe/unsubscribe and the four suppress/unsuppress
  operations. The Subscribers checklist, the subscribe chips, the skip buttons
  and the Suppressed section in the rules UI.
- Inception asks two questions (design system, seed Systems). create_project and
  decide_project_inception lose subscribe_rulebooks.
- Backup v15 stops exporting the three sections; older archives still restore,
  the keys simply unread. Trash no longer hard-deletes suppression rows.

What changes meaning:
- get_applicable_rules is a project's LISTING: its own rules, plus the global
  rules tagged to an area it works in. Untagged global rules apply everywhere
  and arrive by retrieval, so they are not listed. A co_surfaces partner on a
  different project is not dragged in.
- list_rules(project_id) lists that project's own rules.
- rules_payload drops subscribed_rulebooks and suppressed_*; the handshake's
  brief form is project_rules alone.
- using-scribe's "Where a new rule goes" and inception sections, tool
  docstrings and docs say global vs project. Plugin 2026.09.15.1620.

Milestone 414 step 2.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy

src/scribe/services/rulebooks.py

@@ -91,7 +91,7 @@ async def update_rulebook(
 
 
 async def delete_rulebook(rulebook_id: int, user_id: int) -> None:
-    """Delete a rulebook. Cascade-deletes topics, rules, subscriptions."""
+    """Delete a rulebook. Cascade-deletes its topics and rules."""
     async with async_session() as session:
         result = await session.execute(
             select(Rulebook).where(
@@ -265,30 +265,6 @@ async def _assert_project_owned(session, project_id: int, user_id: int) -> None:
         raise ValueError(f"project {project_id} not found")
 
 
-async def _assert_rulebook_rule_owned(session, rule_id: int, user_id: int) -> None:
-    """Raise ValueError if rule isn't a rulebook rule the user owns.
-
-    Project-scoped rules (Rule.project_id set, topic_id NULL) are NOT
-    suppressible — they belong to the project; delete them instead. This
-    helper deliberately excludes them.
-    """
-    from scribe.models.rulebook import Rule
-    result = await session.execute(
-        select(Rule)
-        .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
-        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
-        .where(
-            Rule.id == rule_id,
-            Rulebook.owner_user_id == user_id,
-            Rule.deleted_at.is_(None),
-            RulebookTopic.deleted_at.is_(None),
-            Rulebook.deleted_at.is_(None),
-        )
-    )
-    if result.scalar_one_or_none() is None:
-        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).
@@ -414,9 +390,7 @@ def _refresh_rule_embedding(rule: Rule) -> None:
         logger.exception("embedding refresh failed for rule %s", rule.id)
 
 
-async def co_surfaced_partners(
-    user_id: int, rule_ids: list[int], exclude_ids: set[int] | None = None,
-) -> list[Rule]:
+async def co_surfaced_partners(user_id: int, rule_ids: list[int]) -> list[Rule]:
     """Rules that must arrive WITH the given ones, because they fail together.
 
     This is the whole reason `co_surfaces` exists. Rule 144 was split off rule
@@ -425,18 +399,14 @@ async def co_surfaced_partners(
     what the entire shape is intended to be." Merging was the only fix
     available; this is the fix that should have been available.
 
-    Two limits, both deliberate:
-
-    - Only rules the caller OWNS. An edge is not a back door into someone
-      else's rulebook.
-    - `exclude_ids` is honoured, and callers pass the project's SUPPRESSIONS.
-      A project that explicitly muted a rule should not have it dragged back in
-      by an edge — the suppression is a decision, and the edge does not
-      outrank it.
+    Only rules the caller OWNS: an edge is not a back door into someone else's
+    rulebook. Whether a partner can reach a given PROJECT is the caller's
+    question (get_applicable_rules drops another project's rule), because this
+    answers "what fails with these", which has no project in it.
     """
     if not rule_ids:
         return []
-    known = set(rule_ids) | (exclude_ids or set())
+    known = set(rule_ids)
     async with async_session() as session:
         edges = (await session.execute(
             select(RuleRelation).where(
@@ -523,9 +493,10 @@ async def create_project_rule(
 ) -> Rule:
     """Create a rule scoped to a single project (no rulebook ceremony).
 
-    Project-scoped rules apply only to the named project; they don't
-    propagate via rulebook subscriptions. Topic_id is left NULL — the
-    CHECK constraint enforces exactly-one of (topic_id, project_id).
+    Project-scoped rules apply only to the named project: retrieval surfaces
+    them in that project's sessions and nowhere else (milestone 414), where a
+    rule in a rulebook topic is global. Topic_id is left NULL — the CHECK
+    constraint enforces exactly-one of (topic_id, project_id).
     """
     async with async_session() as session:
         await _assert_project_owned(session, project_id, user_id)
@@ -555,18 +526,40 @@ async def list_rules(
     topic_id: int | None = None,
     project_id: int | None = None,
 ) -> list[Rule]:
-    """List rules filtered by any of the three IDs. All filters are ownership-scoped.
+    """List rules by rulebook, topic or project. Ownership-scoped.
 
-    When project_id is set, the result includes both rulebook rules reached via
-    project_rulebook_subscriptions AND project-scoped rules (Rule.project_id).
-    When rulebook_id or topic_id is set, project-scoped rules are excluded by
-    construction (they have neither). With no filter, only rulebook rules are
-    returned — adding all of a user's project-scoped rules unprompted would
-    surprise existing callers.
+    A rule has one home (milestone 414): a rulebook topic, where it is global,
+    or a project. So the filters name homes rather than reach:
+
+    - `project_id` lists that project's OWN rules. Global rules apply to every
+      project, so listing them under each one would say nothing; list them by
+      rulebook, or unfiltered. `rulebook_id` / `topic_id` don't combine with it
+      — a project rule has neither.
+    - `rulebook_id` / `topic_id` list global rules in that rulebook or topic.
+    - No filter lists every global rule. A user's project rules are left out:
+      they belong to their projects, and mixing them into the rulebook listing
+      would surprise its callers.
+
+    Before milestone 414, `project_id` returned the rules of every rulebook the
+    project SUBSCRIBED to plus its own. Subscriptions are gone.
     """
-    from scribe.models.rulebook import project_rulebook_subscriptions
+    from scribe.models.project import Project
 
     async with async_session() as session:
+        if project_id:
+            result = await session.execute(
+                select(Rule)
+                .join(Project, Rule.project_id == Project.id)
+                .where(
+                    Project.user_id == user_id,
+                    Rule.project_id == project_id,
+                    Rule.deleted_at.is_(None),
+                    Project.deleted_at.is_(None),
+                )
+                .order_by(Rule.order_index, Rule.title)
+            )
+            return list(result.scalars().all())
+
         stmt = (
             select(Rule)
             .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
@@ -582,39 +575,11 @@ async def list_rules(
             stmt = stmt.where(Rule.topic_id == topic_id)
         if rulebook_id:
             stmt = stmt.where(RulebookTopic.rulebook_id == rulebook_id)
-        if project_id:
-            stmt = (
-                stmt.join(
-                    project_rulebook_subscriptions,
-                    project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
-                )
-                .where(project_rulebook_subscriptions.c.project_id == project_id)
-            )
         stmt = stmt.order_by(
             Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
         )
         result = await session.execute(stmt)
-        rulebook_rules = list(result.scalars().all())
-
-        if not project_id:
-            return rulebook_rules
-
-        # Project-scoped rules (topic_id IS NULL, project_id matches).
-        # Verifies ownership by joining Project on user_id.
-        from scribe.models.project import Project
-        proj_stmt = (
-            select(Rule)
-            .join(Project, Rule.project_id == Project.id)
-            .where(
-                Project.user_id == user_id,
-                Rule.project_id == project_id,
-                Rule.deleted_at.is_(None),
-                Project.deleted_at.is_(None),
-            )
-            .order_by(Rule.order_index, Rule.title)
-        )
-        proj_result = await session.execute(proj_stmt)
-        return rulebook_rules + list(proj_result.scalars().all())
+        return list(result.scalars().all())
 
 
 async def _fetch_owned_rule(session, rule_id: int, user_id: int) -> Optional[Rule]:
@@ -930,282 +895,42 @@ async def delete_rule(rule_id: int, user_id: int) -> None:
         await session.commit()
 
 
-# ── Subscriptions + get_applicable_rules ───────────────────────────────
-
-async def subscribe_project(
-    project_id: int, rulebook_id: int, user_id: int,
-) -> None:
-    """Add a subscription. Idempotent — duplicates raise; we swallow."""
-    from scribe.models.rulebook import project_rulebook_subscriptions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await _assert_rulebook_owned(session, rulebook_id, user_id)
-        # ON CONFLICT DO NOTHING via try/except to keep dialect-agnostic.
-        try:
-            await session.execute(
-                insert(project_rulebook_subscriptions).values(
-                    project_id=project_id, rulebook_id=rulebook_id,
-                )
-            )
-            await session.commit()
-        except Exception:
-            await session.rollback()  # PK collision = already subscribed; fine.
-
-
-async def unsubscribe_project(
-    project_id: int, rulebook_id: int, user_id: int,
-) -> None:
-    from scribe.models.rulebook import project_rulebook_subscriptions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await _assert_rulebook_owned(session, rulebook_id, user_id)
-        await session.execute(
-            sql_delete(project_rulebook_subscriptions).where(
-                project_rulebook_subscriptions.c.project_id == project_id,
-                project_rulebook_subscriptions.c.rulebook_id == rulebook_id,
-            )
-        )
-        await session.commit()
-
-
-# ── Suppressions — project-level mute of rulebook rules / topics ────────
-
-async def suppress_rule_for_project(
-    project_id: int, rule_id: int, user_id: int,
-) -> None:
-    """Mute one rulebook rule for one project. Idempotent."""
-    from scribe.models.rulebook import project_rule_suppressions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await _assert_rulebook_rule_owned(session, rule_id, user_id)
-        try:
-            await session.execute(
-                insert(project_rule_suppressions).values(
-                    project_id=project_id, rule_id=rule_id,
-                )
-            )
-            await session.commit()
-        except Exception:
-            await session.rollback()  # PK collision = already suppressed; fine.
-
-
-async def unsuppress_rule_for_project(
-    project_id: int, rule_id: int, user_id: int,
-) -> None:
-    """Unmute one rulebook rule for one project. Idempotent."""
-    from scribe.models.rulebook import project_rule_suppressions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await session.execute(
-            sql_delete(project_rule_suppressions).where(
-                project_rule_suppressions.c.project_id == project_id,
-                project_rule_suppressions.c.rule_id == rule_id,
-            )
-        )
-        await session.commit()
-
-
-async def suppress_topic_for_project(
-    project_id: int, topic_id: int, user_id: int,
-) -> None:
-    """Mute every rule under one topic for one project. Idempotent."""
-    from scribe.models.rulebook import project_topic_suppressions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await _assert_topic_owned(session, topic_id, user_id)
-        try:
-            await session.execute(
-                insert(project_topic_suppressions).values(
-                    project_id=project_id, topic_id=topic_id,
-                )
-            )
-            await session.commit()
-        except Exception:
-            await session.rollback()
-
-
-async def unsuppress_topic_for_project(
-    project_id: int, topic_id: int, user_id: int,
-) -> None:
-    """Unmute a topic for one project. Idempotent."""
-    from scribe.models.rulebook import project_topic_suppressions
-
-    async with async_session() as session:
-        await _assert_project_owned(session, project_id, user_id)
-        await session.execute(
-            sql_delete(project_topic_suppressions).where(
-                project_topic_suppressions.c.project_id == project_id,
-                project_topic_suppressions.c.topic_id == topic_id,
-            )
-        )
-        await session.commit()
-
-
-def _tagged_rule_ids():
-    """Rules carrying at least one canonical area tag (milestone 394).
-
-    The complement is what matters: a rule NOT in this set was never narrowed
-    by its author, so it is general to its rulebook and applies wherever that
-    rulebook is subscribed. Expressed as a subquery rather than a fetched list
-    so the area test stays inside the one statement `limit` is counted on.
-    """
-    return select(rule_systems.c.rule_id)
-
+# ── get_applicable_rules ────────────────────────────────────────────────
 
 async def get_applicable_rules(
     project_id: int, user_id: int, limit: int = 50,
 ) -> dict:
-    """Return rules applicable to a project — both via rulebook subscriptions
-    and project-scoped rules (Rule.project_id matches), with suppressed rules
-    and suppressed topics filtered out.
+    """The rules a project's LISTING shows: its own, and the global rules
+    deterministically bound to the areas it works in.
 
     Shape:
         {
-          "rules": [{id, title, statement,
-                     topic_id, topic_title,
-                     rulebook_id, rulebook_title}, ...],
-          "project_rules": [{id, title, statement}, ...],
-          "suppressed_rules": [{id, title,
-                                topic_id, topic_title,
-                                rulebook_id, rulebook_title}, ...],
-          "suppressed_topics": [{id, title,
-                                 rulebook_id, rulebook_title}, ...],
+          "rules": [{id, title, statement, topic_id, topic_title,
+                     rulebook_id, rulebook_title, ...}, ...],
+          "project_rules": [{id, title, statement, ...}, ...],
           "truncated": bool,
-          "subscribed_rulebooks": [{id, title}, ...]
         }
 
-    `rules` is the subscription-derived set with project-level suppressions
-    applied. `project_rules` is the project-scoped set (never suppressed —
-    delete instead). `suppressed_rules` / `suppressed_topics` carry the
-    titles + rulebook context callers need to display what was filtered
-    without round-tripping for names.
+    NOT WHAT A SESSION RECEIVES. Rules reach a session by retrieval, which
+    reads a rule's home (milestone 414): global rules everywhere, a project's
+    own rules in that project. This is the listing a planning read carries so
+    a reader can see which constraints are on the table, and it is narrower
+    than retrieval on purpose — "every global rule" is not a list anyone reads.
+
+    `rules` is the global rules TAGGED to a canonical area this project works
+    in (milestone 307, D7): a deterministic tag match, never a similarity
+    score. Before milestone 414 this was every rule in a SUBSCRIBED rulebook,
+    narrowed by area only where an author had tagged one. With subscriptions
+    gone there is no opt-in left to scope the untagged ones, and an untagged
+    global rule is general by construction — it arrives by retrieval when the
+    work makes it relevant, like every other rule.
+
+    `project_rules` is the project's own, never filtered by area: a rule
+    written ON a project is scoped to it already.
     """
-    from scribe.models.rulebook import (
-        project_rulebook_subscriptions,
-        project_rule_suppressions,
-        project_topic_suppressions,
-    )
+    from scribe.models.project import Project
 
     async with async_session() as session:
-        # Subscribed rulebooks for the project (ownership-scoped).
-        sub_q = (
-            select(Rulebook.id, Rulebook.title)
-            .join(
-                project_rulebook_subscriptions,
-                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
-            )
-            .where(
-                project_rulebook_subscriptions.c.project_id == project_id,
-                Rulebook.owner_user_id == user_id,
-                Rulebook.deleted_at.is_(None),
-            )
-            .order_by(Rulebook.title)
-        )
-        sub_rows = (await session.execute(sub_q)).all()
-        subscribed_rulebooks = [
-            {"id": rb_id, "title": rb_title} for rb_id, rb_title in sub_rows
-        ]
-
-        # Suppressed rules — joined to topic + rulebook so callers can render
-        # context without a follow-up lookup. Ownership-scoped via rulebook.
-        suppressed_rules_q = (
-            select(
-                Rule.id, Rule.title,
-                RulebookTopic.id.label("topic_id"),
-                RulebookTopic.title.label("topic_title"),
-                Rulebook.id.label("rulebook_id"),
-                Rulebook.title.label("rulebook_title"),
-            )
-            .join(project_rule_suppressions, project_rule_suppressions.c.rule_id == Rule.id)
-            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
-            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
-            .where(
-                project_rule_suppressions.c.project_id == project_id,
-                Rulebook.owner_user_id == user_id,
-            )
-            .order_by(Rulebook.title, RulebookTopic.title, Rule.title)
-        )
-        suppressed_rule_rows = (await session.execute(suppressed_rules_q)).all()
-        suppressed_rules = [
-            {"id": rid, "title": rt, "topic_id": ti, "topic_title": tt,
-             "rulebook_id": rbi, "rulebook_title": rbt}
-            for rid, rt, ti, tt, rbi, rbt in suppressed_rule_rows
-        ]
-        suppressed_rule_ids = [r["id"] for r in suppressed_rules]
-
-        # Suppressed topics — joined to rulebook for context.
-        suppressed_topics_q = (
-            select(
-                RulebookTopic.id, RulebookTopic.title,
-                Rulebook.id.label("rulebook_id"),
-                Rulebook.title.label("rulebook_title"),
-            )
-            .join(project_topic_suppressions, project_topic_suppressions.c.topic_id == RulebookTopic.id)
-            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
-            .where(
-                project_topic_suppressions.c.project_id == project_id,
-                Rulebook.owner_user_id == user_id,
-            )
-            .order_by(Rulebook.title, RulebookTopic.title)
-        )
-        suppressed_topic_rows = (await session.execute(suppressed_topics_q)).all()
-        suppressed_topics = [
-            {"id": tid, "title": tt, "rulebook_id": rbi, "rulebook_title": rbt}
-            for tid, tt, rbi, rbt in suppressed_topic_rows
-        ]
-        suppressed_topic_ids = [t["id"] for t in suppressed_topics]
-
-        # 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,
-                RulebookTopic.title.label("topic_title"),
-                Rulebook.id.label("rulebook_id"),
-                Rulebook.title.label("rulebook_title"),
-            )
-            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
-            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
-            .join(
-                project_rulebook_subscriptions,
-                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
-            )
-            .where(
-                project_rulebook_subscriptions.c.project_id == project_id,
-                Rulebook.owner_user_id == user_id,
-                Rule.deleted_at.is_(None),
-                RulebookTopic.deleted_at.is_(None),
-                Rulebook.deleted_at.is_(None),
-                # An inception exclusion is total (milestone 297): a rulebook the
-                # project opted out of contributes nothing, subscribed or not.
-            )
-            .order_by(
-                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
-            )
-            .limit(limit + 1)
-        )
-        if suppressed_rule_ids:
-            rules_q = rules_q.where(Rule.id.notin_(suppressed_rule_ids))
-        if suppressed_topic_ids:
-            rules_q = rules_q.where(Rule.topic_id.notin_(suppressed_topic_ids))
-        # AREA BINDING (milestone 307, narrowed by 394). A rule reaches this
-        # project when it is tagged to an area the project actually works in —
-        # a deterministic tag match, never a similarity score, so bindingness
-        # never depends on a ranking (D7).
-        #
-        # Applied in SQL rather than by filtering afterwards, so `limit` counts
-        # the rules that will actually be surfaced instead of counting rules
-        # that are about to be dropped.
         project_area_ids = (await session.execute(
             select(System.canonical_id).where(
                 System.project_id == project_id,
@@ -1214,36 +939,37 @@ async def get_applicable_rules(
                 System.status == "active",
             ).distinct()
         )).scalars().all()
-        reachable = select(rule_systems.c.rule_id).where(
-            rule_systems.c.canonical_id.in_(project_area_ids)
-        ) if project_area_ids else None
-        # SUBSCRIPTION IS THE SCOPE; AREAS NARROW ONLY WHERE AN AUTHOR ASKED.
-        #
-        # This read `always_on OR reachable` (milestone 307). The tier arm is
-        # gone, and the first attempt at 394 kept only the reachable arm — so
-        # a subscribed rulebook's untagged rules stopped arriving at all. That
-        # was wrong twice over: the query above is ALREADY scoped to rulebooks
-        # this project subscribed to, so the project opted in and was then
-        # handed a subset of what it asked for; and the milestone is explicit
-        # that subscription-derived rules are not what it removes. The
-        # integration suite caught it through a co_surfaces partner that never
-        # arrived because the rule it travels with had been filtered out.
-        #
-        # So: every rule in a subscribed rulebook applies, EXCEPT that a rule
-        # tagged to specific areas applies only to a project working in one of
-        # them. An untagged rule is general to its rulebook by construction —
-        # nobody narrowed it — while tagging is an author saying "this is
-        # about CI" and meaning it. That keeps D7's deterministic narrowing
-        # where it was asked for without inventing it where it was not.
-        if reachable is not None:
-            rules_q = rules_q.where(
-                or_(Rule.id.in_(reachable), Rule.id.notin_(_tagged_rule_ids())),
-            )
-        else:
-            # No canonical areas on this project: nothing can match by area,
-            # so only the untagged (general) rules apply.
-            rules_q = rules_q.where(Rule.id.notin_(_tagged_rule_ids()))
-        rule_rows = (await session.execute(rules_q)).all()
+
+        rule_rows = []
+        if project_area_ids:
+            # Selects the ENTITY, not a column list: rule_brief is the one
+            # place that decides which fields a surfaced rule carries. Filtered
+            # in SQL so `limit` counts the rules that will actually be shown.
+            rule_rows = (await session.execute(
+                select(
+                    Rule,
+                    RulebookTopic.title.label("topic_title"),
+                    Rulebook.id.label("rulebook_id"),
+                    Rulebook.title.label("rulebook_title"),
+                )
+                .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
+                .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+                .where(
+                    Rulebook.owner_user_id == user_id,
+                    Rule.deleted_at.is_(None),
+                    RulebookTopic.deleted_at.is_(None),
+                    Rulebook.deleted_at.is_(None),
+                    Rule.id.in_(
+                        select(rule_systems.c.rule_id).where(
+                            rule_systems.c.canonical_id.in_(project_area_ids)
+                        )
+                    ),
+                )
+                .order_by(
+                    Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
+                )
+                .limit(limit + 1)
+            )).all()
         truncated = len(rule_rows) > limit
         rules = [
             rule_brief(rule, topic_title=tt, rulebook_id=rbi, rulebook_title=rbt)
@@ -1251,8 +977,7 @@ async def get_applicable_rules(
         ]
 
         # Project-scoped rules — verifies ownership via Project.user_id.
-        from scribe.models.project import Project
-        proj_rules_q = (
+        proj_rule_rows = (await session.execute(
             select(Rule)
             .join(Project, Rule.project_id == Project.id)
             .where(
@@ -1262,32 +987,26 @@ async def get_applicable_rules(
                 Project.deleted_at.is_(None),
             )
             .order_by(Rule.order_index, Rule.title)
-        )
-        # A PROJECT'S OWN RULES ARE NOT FILTERED BY AREA, and the asymmetry
-        # with the family query above is the point. A family rule has to earn
-        # its way into this project; a rule written ON this project is scoped
-        # to it by construction, and filtering it again would drop rules whose
-        # only fault is that nobody tagged them to a System.
-        proj_rule_rows = (await session.execute(proj_rules_q)).all()
+        )).all()
         project_rules = [rule_brief(rule) for (rule,) in proj_rule_rows]
 
     # Edges travel with the rules they belong to (milestone 307).
     #
     # A co_surfaces partner that was not otherwise selected is ADDED, because a
     # rule that arrives without the half it fails with is the failure the edge
-    # was created to prevent. Suppressions are passed as exclusions so an
-    # explicit mute still wins over an edge.
+    # was created to prevent — but only a partner that could reach this
+    # project at all. An edge to another project's rule is not a way in.
     surfaced_ids = [r["id"] for r in rules] + [r["id"] for r in project_rules]
-    partners = await co_surfaced_partners(
-        user_id, surfaced_ids, exclude_ids=set(suppressed_rule_ids),
-    )
+    partners = await co_surfaced_partners(user_id, surfaced_ids)
     for partner in partners:
+        if partner.project_id not in (None, project_id):
+            continue
         rules.append(rule_brief(partner, via="co_surfaces"))
         surfaced_ids.append(partner.id)
 
     # Relations on every surfaced rule, so a reader can see that an override
     # exists rather than discovering the contradiction by acting on the wrong
-    # one. Areas too — they are why a conditional rule is here at all.
+    # one. Areas too — they are why a global rule is in this listing at all.
     edges = await list_rule_relations(surfaced_ids)
     areas = await list_rule_systems(surfaced_ids)
     for brief in (*rules, *project_rules):
@@ -1296,14 +1015,7 @@ async def get_applicable_rules(
         if areas.get(brief["id"]):
             brief["systems"] = areas[brief["id"]]
 
-    return {
-        "rules": rules,
-        "project_rules": project_rules,
-        "suppressed_rules": suppressed_rules,
-        "suppressed_topics": suppressed_topics,
-        "truncated": truncated,
-        "subscribed_rulebooks": subscribed_rulebooks,
-    }
+    return {"rules": rules, "project_rules": project_rules, "truncated": truncated}
 
 
 def rules_payload(
@@ -1313,11 +1025,11 @@ def rules_payload(
 
     Every surface that hands rules to an agent (enter_project, get_project,
     get_milestone, get_task for legacy plans, start_planning) carries the
-    same seven keys under the same names — so a reader learns them once. One
+    same keys under the same names — so a reader learns them once. One
     place renames `rules` → `applicable_rules` and `truncated` →
     `applicable_rules_truncated`; the tools merge this into their payloads.
 
-    IT ALSO RECORDS THE SURFACING, which is why it now takes a caller and a
+    IT ALSO RECORDS THE SURFACING, which is why it takes a caller and a
     source. Every one of those surfaces is a bulk delivery — the applicable set
     handed over whole, chosen by nobody — so this is the one place that has to
     emit for all of them. Doing it per-caller instead would be five sites to
@@ -1330,17 +1042,16 @@ def rules_payload(
     `RANKED_SOURCES` in `rule_usage` is what folds them back together.
 
     Emitting from here is safe in a way emitting from `get_applicable_rules`
-    would not be: this function is only ever called to BUILD A REPLY. The two
-    other callers of the rules machinery — the write-path etag arm
-    (`plugin_context`) — computed a marker and showed nobody
-    anything, and counting those would put rules in the denominator that no
-    agent ever saw.
+    would not be: this function is only ever called to BUILD A REPLY. The
+    other caller of the rules machinery — the write-path etag arm
+    (`plugin_context`) — computes a marker and shows nobody anything, and
+    counting it would put rules in the denominator that no agent ever saw.
 
     `brief` is the session handshake's form (#4045): the project's own rules
-    as id and title, and the subscribed rulebooks, nothing else. Rules reach a
-    session in full by retrieval, which ignores subscriptions, so the handshake
-    lists which constraints exist rather than restating them; get_rule reads
-    one. Only what is shown is recorded as surfaced.
+    as id and title, nothing else. Rules reach a session in full by
+    retrieval, so the handshake lists which of the project's constraints exist
+    rather than restating them; get_rule reads one. Only what is shown is
+    recorded as surfaced.
     """
     if brief:
         project_rules = [
@@ -1350,10 +1061,7 @@ def rules_payload(
         record_rule_surfaced(
             user_id=user_id, rule_ids=[r["id"] for r in project_rules], source=source,
         )
-        return {
-            "project_rules": project_rules,
-            "subscribed_rulebooks": applicable["subscribed_rulebooks"],
-        }
+        return {"project_rules": project_rules}
     record_rule_surfaced(
         user_id=user_id,
         rule_ids=(
@@ -1365,10 +1073,7 @@ def rules_payload(
     return {
         "applicable_rules": applicable["rules"],
         "applicable_rules_truncated": applicable["truncated"],
-        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
         "project_rules": applicable.get("project_rules", []),
-        "suppressed_rules": applicable.get("suppressed_rules", []),
-        "suppressed_topics": applicable.get("suppressed_topics", []),
     }