feat(inception): always-on exclusions reach every rule surface — list_always_on_rules(project_id), get_applicable_rules, rules_payload.excluded_always_on, session context, exclude/include tools (#2880, milestone 297 step 2)

A project that opted out of an always-on rulebook at inception must not see
it anywhere: list_always_on_rules(project_id=) and the subscription-derived
set skip it (an exclusion is total), get_applicable_rules / rules_payload
carry `excluded_always_on` as the seventh key so the departure is visible
wherever the rules are, and the SessionStart block built for a bound project
names it ("Excluded for this project by its inception decision …"). MCP:
list_always_on_rules takes project_id; exclude_always_on_rulebook /
include_always_on_rulebook mirror suppress/unsuppress (owner-only; the
rulebook must be always_on — subscribed rulebooks are left by unsubscribing).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/services/rulebooks.py

@@ -394,15 +394,57 @@ async def list_rules(
         return rulebook_rules + list(proj_result.scalars().all())
 
 
-async def list_always_on_rules(user_id: int, limit: int = 100) -> list[Rule]:
+def _excluded_rulebook_ids_q(project_id: int):
+    """Subquery: the always-on rulebooks this project opted out of at
+    inception (milestone 297) — used by every rule-resolution path so an
+    exclusion is total, not just cosmetic."""
+    from scribe.models.rulebook import project_rulebook_exclusions
+
+    return select(project_rulebook_exclusions.c.rulebook_id).where(
+        project_rulebook_exclusions.c.project_id == project_id
+    )
+
+
+async def excluded_always_on_rulebooks(user_id: int, project_id: int) -> list[dict]:
+    """[{id, title}] of the always-on rulebooks excluded for ``project_id``
+    (owner-scoped). Empty for an undecided or inherit-all project."""
+    from scribe.models.rulebook import project_rulebook_exclusions
+
+    if not project_id:
+        return []
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(Rulebook.id, Rulebook.title)
+                .join(project_rulebook_exclusions,
+                      project_rulebook_exclusions.c.rulebook_id == Rulebook.id)
+                .where(
+                    project_rulebook_exclusions.c.project_id == project_id,
+                    Rulebook.owner_user_id == user_id,
+                    Rulebook.deleted_at.is_(None),
+                )
+                .order_by(Rulebook.title)
+            )
+        ).all()
+    return [{"id": rid, "title": title} for rid, title in rows]
+
+
+async def list_always_on_rules(
+    user_id: int, limit: int = 100, project_id: int = 0,
+) -> list[Rule]:
     """Return all rules from rulebooks flagged always_on for the user.
 
     Called by the MCP tool of the same name at session start to load the
     standing rules that apply regardless of which project (if any) is in
     scope. Ordering matches list_rules so results are stable across calls.
+
+    ``project_id`` (milestone 297): inside a project that excluded specific
+    always-on rulebooks at inception, those rulebooks' rules are NOT
+    returned — the project decided not to inherit them. 0 = the user-wide
+    set, which is what a session sees before a project is in scope.
     """
     async with async_session() as session:
-        result = await session.execute(
+        q = (
             select(Rule)
             .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
             .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
@@ -413,10 +455,13 @@ async def list_always_on_rules(user_id: int, limit: int = 100) -> list[Rule]:
                 RulebookTopic.deleted_at.is_(None),
                 Rulebook.deleted_at.is_(None),
             )
-            .order_by(
+        )
+        if project_id:
+            q = q.where(Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)))
+        result = await session.execute(
+            q.order_by(
                 Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
-            )
-            .limit(limit)
+            ).limit(limit)
         )
         return list(result.scalars().all())
 
@@ -489,6 +534,7 @@ 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
 
 
 async def subscribe_project(
@@ -568,6 +614,51 @@ async def unsuppress_rule_for_project(
         await session.commit()
 
 
+async def exclude_always_on_rulebook_for_project(
+    project_id: int, rulebook_id: int, user_id: int,
+) -> None:
+    """Opt one project out of a whole ALWAYS-ON rulebook (milestone 297).
+    Owner-only on both sides; the rulebook must be always_on — a subscribed
+    rulebook is left by unsubscribing, not excluding. Idempotent."""
+    from scribe.models.rulebook import project_rulebook_exclusions
+
+    async with async_session() as session:
+        await _assert_project_owned(session, project_id, user_id)
+        await _assert_rulebook_owned(session, rulebook_id, user_id)
+        rb = await session.get(Rulebook, rulebook_id)
+        if rb is None or not rb.always_on:
+            raise ValueError(
+                f"rulebook {rulebook_id} is not always-on — it binds only by "
+                "subscription; unsubscribe_project_from_rulebook instead"
+            )
+        try:
+            await session.execute(
+                insert(project_rulebook_exclusions).values(
+                    project_id=project_id, rulebook_id=rulebook_id,
+                )
+            )
+            await session.commit()
+        except IntegrityError:
+            await session.rollback()  # already excluded — idempotent
+
+
+async def include_always_on_rulebook_for_project(
+    project_id: int, rulebook_id: int, user_id: int,
+) -> None:
+    """Undo exclude_always_on_rulebook_for_project. Idempotent."""
+    from scribe.models.rulebook import project_rulebook_exclusions
+
+    async with async_session() as session:
+        await _assert_project_owned(session, project_id, user_id)
+        await session.execute(
+            sql_delete(project_rulebook_exclusions).where(
+                project_rulebook_exclusions.c.project_id == project_id,
+                project_rulebook_exclusions.c.rulebook_id == rulebook_id,
+            )
+        )
+        await session.commit()
+
+
 async def suppress_topic_for_project(
     project_id: int, topic_id: int, user_id: int,
 ) -> None:
@@ -731,6 +822,9 @@ async def get_applicable_rules(
                 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.
+                Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)),
             )
             .order_by(
                 Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
@@ -778,6 +872,7 @@ async def get_applicable_rules(
         "suppressed_topics": suppressed_topics,
         "truncated": truncated,
         "subscribed_rulebooks": subscribed_rulebooks,
+        "excluded_always_on": await excluded_always_on_rulebooks(user_id, project_id),
     }
 
 
@@ -786,9 +881,12 @@ def rules_payload(applicable: dict) -> dict:
 
     Every surface that hands rules to an agent (enter_project, get_project,
     get_milestone, get_task for legacy plans, start_planning) carries the
-    same six keys under the same names — so a reader learns them once. One
+    same seven 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.
+    `excluded_always_on` (milestone 297) names the always-on rulebooks this
+    project decided NOT to inherit, so the departure is visible wherever the
+    rules are.
     """
     return {
         "applicable_rules": applicable["rules"],
@@ -797,4 +895,5 @@ def rules_payload(applicable: dict) -> dict:
         "project_rules": applicable.get("project_rules", []),
         "suppressed_rules": applicable.get("suppressed_rules", []),
         "suppressed_topics": applicable.get("suppressed_topics", []),
+        "excluded_always_on": applicable.get("excluded_always_on", []),
     }