fix(394): subscription is the scope — areas narrow only where an author asked

I got this wrong in 0e10f6b and the integration suite caught it.

get_applicable_rules filtered `always_on OR area-reachable`. Removing the
tier, I kept only the reachable arm and argued that a project with no
canonical-tagged Systems should get no bulk rules and reach them by
retrieval instead.

Two things wrong with that. The query is ALREADY scoped to rulebooks the
project SUBSCRIBED to, so the project had opted in and was then handed a
subset of what it asked for — subscription is not bulk delivery, it is the
opt-in. And milestone 394 is explicit that subscription-derived rules are not
the always-on tier and are not what it removes; I narrowed something the
milestone said to leave alone.

The failure that surfaced it is a good one: a co_surfaces partner never
arrived, because the rule it travels with had been filtered out before the
edge could drag it in. A behaviour two steps from the change.

What ships instead is narrower than "drop the clause" and wider than what I
had: 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 was never narrowed by anyone, so it is general to its rulebook
by construction; a tagged one is an author saying "this is about CI" and
meaning it. D7's deterministic narrowing is kept where it was asked for and
not invented where it was not.

Also: three tests covering the SessionStart preload, the always-on tool and
the two marker paths — all surfaces that no longer exist — and the backup's
declared-section list, which still named the section its table took with it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ

src/scribe/services/rulebooks.py

@@ -11,7 +11,7 @@ import logging
 from collections.abc import Iterable
 from typing import Optional
 
-from sqlalchemy import and_, delete as sql_delete, false as sa_false, insert, or_, select
+from sqlalchemy import and_, delete as sql_delete, insert, or_, select
 
 from scribe.models import async_session
 from scribe.models.system import System
@@ -1046,6 +1046,17 @@ async def unsuppress_topic_for_project(
         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)
+
+
 async def get_applicable_rules(
     project_id: int, user_id: int, limit: int = 50,
 ) -> dict:
@@ -1206,21 +1217,32 @@ async def get_applicable_rules(
         reachable = select(rule_systems.c.rule_id).where(
             rule_systems.c.canonical_id.in_(project_area_ids)
         ) if project_area_ids else None
-        # AREA REACHABILITY IS NOW THE WHOLE TEST (milestone 394). This read
-        # `always_on OR reachable`, so a subscribed rulebook's resident rules
-        # arrived here whatever the project did. The tier is gone, and
-        # dropping its arm rather than the whole clause is the deliberate
-        # half: what survives is the DETERMINISTIC one — a rule binds this
-        # project because it is tagged to an area the project actually works
-        # in (D7), never because a similarity score cleared a bar.
+        # SUBSCRIPTION IS THE SCOPE; AREAS NARROW ONLY WHERE AN AUTHOR ASKED.
         #
-        # A project with no canonical-tagged Systems therefore gets no bulk
-        # rules here, and that is the reading rather than a gap: rules still
-        # reach it by retrieval, when something it is doing makes one
-        # relevant. Handing over every subscribed rule instead would make this
-        # payload BIGGER than the preload this milestone exists to remove.
-        rules_q = (rules_q.where(Rule.id.in_(reachable)) if reachable is not None
-                   else rules_q.where(sa_false()))
+        # 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()
         truncated = len(rule_rows) > limit
         rules = [

tests/test_rule_usage_wiring.py

@@ -309,57 +309,8 @@ def test_the_bulk_loaders_are_not_counted_as_pulls():
 # lookalike call sites which show nobody anything do NOT.
 
 
-@pytest.mark.asyncio
-async def test_the_session_start_preload_records_what_it_delivered():
-    """The block every session opens with. Chosen by nobody, paid for every
-    turn — and until it emitted, invisible to the scoreboard that judges every
-    other surface."""
-    from scribe.services import plugin_context as pc
-
-    rec = MagicMock()
-    rules = [fake_rule(id=1, title="`dev` is home"),
-             fake_rule(id=2, title="`main` — never without explicit request")]
-    with ExitStack() as stack:
-        stack.enter_context(
-            patch.object(pc.rulebooks_svc, "list_always_on_rules",
-                         AsyncMock(return_value=rules))
-        )
-        stack.enter_context(
-            patch.object(pc.rulebooks_svc, "excluded_always_on_rulebooks",
-                         AsyncMock(return_value=[]))
-        )
-        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
-        stack.enter_context(
-            patch.object(pc, "_topic_titles", AsyncMock(return_value={}))
-        )
-        await pc.build_session_context(1, project_id=0)
-
-    assert rec.call_count == 1, "the preload recorded nothing"
-    kw = rec.call_args.kwargs
-    assert kw["rule_ids"] == [1, 2]
-    assert kw["source"] == "session_start"
 
 
-@pytest.mark.asyncio
-async def test_the_always_on_tool_records_what_it_handed_over():
-    from scribe.mcp.tools import rulebooks as tools
-
-    rec = MagicMock()
-    rules = [fake_rule(id=3, title="No GitHub — Fabled-Git only")]
-    with ExitStack() as stack:
-        stack.enter_context(
-            patch.object(tools.rulebooks_svc, "list_always_on_rules",
-                         AsyncMock(return_value=rules))
-        )
-        stack.enter_context(
-            patch.object(tools.rulebooks_svc, "rules_etag",
-                         MagicMock(return_value="etag"))
-        )
-        stack.enter_context(patch.object(tools, "record_rule_surfaced", rec))
-        await tools.list_always_on_rules()
-
-    assert rec.call_args.kwargs["rule_ids"] == [3]
-    assert rec.call_args.kwargs["source"] == "list_always_on_rules"
 
 
 def test_rules_payload_records_both_the_family_and_project_halves():
@@ -406,26 +357,6 @@ def test_every_rules_payload_caller_names_itself():
     }, f"a rules_payload caller is missing or misnamed: {sorted(seen)}"
 
 
-def test_the_marker_paths_stay_silent():
-    """The two call sites that read the rules and show NOBODY anything.
-
-    `rules_etag_for` and the write-path staleness arm both call
-    `list_always_on_rules` to build or compare a marker. Emitting there would
-    put rules in the denominator that no agent ever saw — the exact inflation
-    `record_rule_surfaced`'s docstring forbids, arriving from the one direction
-    nothing else guards.
-    """
-    svc_src = Path("src/scribe/services/rulebooks.py").read_text()
-    etag_fn = svc_src.split("async def rules_etag_for")[1].split("\ndef ")[0]
-    assert "record_rule_surfaced" not in etag_fn, (
-        "rules_etag_for emits a surfacing — it builds a marker, it shows nothing"
-    )
-
-    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
-    staleness = pc_src.split("if rules_etag:")[1].split("# The guard sits BELOW")[0]
-    assert "record_rule_surfaced" not in staleness, (
-        "the staleness arm emits a surfacing — it compares a marker, it shows nothing"
-    )
 
 
 # ── The PRE-TOOL arm: rules keyed on the action (#3476) ────────────────

tests/test_services_backup.py

@@ -251,7 +251,7 @@ def test_the_column_guard_covers_every_table_with_a_row_helper():
     # caught on its own first run.
     join_tables = {
         "project_rulebook_subscriptions", "project_rule_suppressions",
-        "project_topic_suppressions", "project_rulebook_exclusions",
+        "project_topic_suppressions",
         "rule_systems",
     }
     covered = set(_column_guard_targets()) | join_tables
@@ -355,7 +355,7 @@ async def test_export_full_backup_contains_every_declared_section():
                 "systems", "record_systems", "design_systems",
                 "design_tokens", "note_usage_events", "repo_bindings",
                 "note_supersessions", "code_shapes", "code_shape_events",
-                "code_shape_uses", "rulebook_exclusions"):
+                "code_shape_uses"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []