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
+        # SUBSCRIPTION IS THE SCOPE; AREAS NARROW ONLY WHERE AN AUTHOR ASKED.
-        # `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.
         #
-        # A project with no canonical-tagged Systems therefore gets no bulk
+        # This read `always_on OR reachable` (milestone 307). The tier arm is
-        # rules here, and that is the reading rather than a gap: rules still
+        # gone, and the first attempt at 394 kept only the reachable arm — so
-        # reach it by retrieval, when something it is doing makes one
+        # a subscribed rulebook's untagged rules stopped arriving at all. That
-        # relevant. Handing over every subscribed rule instead would make this
+        # was wrong twice over: the query above is ALREADY scoped to rulebooks
-        # payload BIGGER than the preload this milestone exists to remove.
+        # this project subscribed to, so the project opted in and was then
-        rules_q = (rules_q.where(Rule.id.in_(reachable)) if reachable is not None
+        # handed a subset of what it asked for; and the milestone is explicit
-                   else rules_q.where(sa_false()))
+        # 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] == []