feat(rules): tier 1 preloads, tier 2 arrives by area — and a split rule can no longer be read half-way (#3031, milestone 307 step 5)

The payoff step: a rule stops having to be resident to be honoured.

- list_always_on_rules returns the ALWAYS-ON tier only. It is the session-start
  call, made before any project is in scope, so there is no area vocabulary to
  match a conditional rule against yet.
- get_applicable_rules carries a conditional rule when the project works in an
  area the rule is tagged to — resolved through systems.canonical_id, so the
  project's own NAME for the area is irrelevant, which is the entire reason the
  catalog exists. The gate is applied IN SQL, so `limit` counts rules that will
  actually surface rather than rules about to be dropped.
- Bindingness is a deterministic TAG match, never a similarity score (D7). The
  vector channel stays a suggestion, in search.

co_surfaced_partners is the fix that rule 144 never had. It was split off rule
46 and folded back the same day because "either rule could surface without the
other and miss exposing a project to what the entire shape is intended to be" —
correct, and merging was the only remedy available. Now a partner ARRIVES with
its other half even when nothing else selected it, tagged `via: co_surfaces` so
the payload says why. Two limits, both deliberate: only rules the caller owns,
because an edge is not a back door into someone else's rulebook; and a
project's suppressions are passed as exclusions, because an explicit mute is a
decision and an edge does not outrank it.

COMPATIBILITY, asserted first in the integration test rather than reasoned
about: a rule with no tier, no areas and no edges binds exactly as it did
before any of this existed. `tier` defaults to always_on, so an install
upgrades and every rule it already had keeps arriving. Getting that backwards
would silently stop enforcing rules people rely on, which is worse than any
amount of payload bloat.

Four unit tests were coupled to the ORDER of a mocked session's execute()
calls, so a new query broke them. Rather than pad the sequence and deepen that
coupling, the three post-query lookups are stubbed by name — they have their
own coverage, and the real wiring is proven against Postgres.

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

src/scribe/services/rulebooks.py

@@ -13,6 +13,7 @@ from typing import Optional
 from sqlalchemy import delete as sql_delete, insert, select
 
 from scribe.models import async_session
+from scribe.models.system import System
 from scribe.models.rulebook import Rulebook
 
 logger = logging.getLogger(__name__)
@@ -223,7 +224,7 @@ async def delete_topic(topic_id: int, user_id: int) -> None:
 
 # ── Rule CRUD ──────────────────────────────────────────────────────────
 
-from scribe.models.rulebook import Rule, RuleRelation
+from scribe.models.rulebook import Rule, RuleRelation, rule_systems
 
 
 async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
@@ -358,6 +359,54 @@ 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]:
+    """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
+    46 and folded back into it the same day, on the correct observation that
+    "either rule could surface without the other and miss exposing a project to
+    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.
+    """
+    if not rule_ids:
+        return []
+    known = set(rule_ids) | (exclude_ids or set())
+    async with async_session() as session:
+        edges = (await session.execute(
+            select(RuleRelation).where(
+                RuleRelation.kind == "co_surfaces",
+                or_(
+                    RuleRelation.from_rule_id.in_(rule_ids),
+                    RuleRelation.to_rule_id.in_(rule_ids),
+                ),
+            )
+        )).scalars().all()
+        partners = {
+            (edge.to_rule_id if edge.from_rule_id in known else edge.from_rule_id)
+            for edge in edges
+        } - known
+        if not partners:
+            return []
+        # Ownership re-checked per partner rather than assumed from the edge.
+        out = []
+        for partner_id in sorted(partners):
+            rule = await _fetch_owned_rule(session, partner_id, user_id)
+            if rule is not None:
+                out.append(rule)
+        return out
+
+
 async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict:
     """The full record, with its areas and edges attached.
 
@@ -567,6 +616,17 @@ async def list_always_on_rules(
                 Rule.deleted_at.is_(None),
                 RulebookTopic.deleted_at.is_(None),
                 Rulebook.deleted_at.is_(None),
+                # TIER (milestone 307). This is the SESSION-START call, made
+                # before any project is in scope — there is no area vocabulary
+                # to match a conditional rule against yet, so only the
+                # unconditional tier belongs here. A conditional rule reaches a
+                # session through enter_project (by area) or search (by
+                # meaning), not by being resident.
+                #
+                # Behaviour is unchanged until rules are actually re-tiered:
+                # `tier` defaults to always_on, so every existing rule still
+                # arrives exactly as it did.
+                Rule.tier == "always_on",
             )
         )
         if project_id:
@@ -1110,6 +1170,30 @@ async def get_applicable_rules(
             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))
+        # TIER (milestone 307). always_on rules are resident, as every rule was
+        # before tiers existed. A conditional rule is REACHABLE, and reaches
+        # this project only when it is tagged to an area this 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,
+                System.canonical_id.is_not(None),
+                System.deleted_at.is_(None),
+                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
+        tier_clause = (Rule.tier == "always_on")
+        if reachable is not None:
+            tier_clause = or_(tier_clause, Rule.id.in_(reachable))
+        rules_q = rules_q.where(tier_clause)
         rule_rows = (await session.execute(rules_q)).all()
         truncated = len(rule_rows) > limit
         rules = [
@@ -1130,9 +1214,40 @@ async def get_applicable_rules(
             )
             .order_by(Rule.order_index, Rule.title)
         )
+        if reachable is not None:
+            proj_rules_q = proj_rules_q.where(
+                or_(Rule.tier == "always_on", Rule.id.in_(reachable))
+            )
+        else:
+            proj_rules_q = proj_rules_q.where(Rule.tier == "always_on")
         proj_rule_rows = (await session.execute(proj_rules_q)).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.
+    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),
+    )
+    for partner in partners:
+        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.
+    edges = await list_rule_relations(surfaced_ids)
+    areas = await list_rule_systems(surfaced_ids)
+    for brief in (*rules, *project_rules):
+        if edges.get(brief["id"]):
+            brief["relations"] = edges[brief["id"]]
+        if areas.get(brief["id"]):
+            brief["systems"] = areas[brief["id"]]
+
     return {
         "rules": rules,
         "project_rules": project_rules,