Merge pull request 'Project rule + topic suppressions' (#54) from dev into main

Lets a project mute individual rules or whole topics from rulebooks it
subscribes to, without unsubscribing the rulebook. Two new association
tables (migration 0060), 4 MCP tools (suppress/unsuppress × rule/topic),
4 REST endpoints, and an inline "× skip" affordance plus collapsed
"Suppressed (N)" section in the project's Rules tab.

get_applicable_rules now emits suppressed_rules and suppressed_topics
(detail objects with rulebook/topic context, not just IDs) so the UI
can render the suppressed list without a follow-up lookup. The main
rules projection grew topic_id and rulebook_id columns for the per-row
suppress affordance.

Project deletion cascades the suppression rows via hard DELETE — they
are pure associations with no soft-delete column, and restoring a
deleted project should start fresh, not inherit stale mutes.

Project-scoped rules (Rule.project_id) are deliberately not suppressible
— delete them with delete_rule instead.

Implements plan-task #187.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

alembic/versions/0058_rulebook_always_on.py

@@ -0,0 +1,39 @@
+"""rulebook always_on flag
+
+Revision ID: 0058
+Revises: 0057
+Create Date: 2026-06-01
+
+Adds a boolean `always_on` to the `rulebooks` table. Rules from rulebooks
+flagged always_on are loaded at session start by the new
+`list_always_on_rules` MCP tool — they apply regardless of which project
+(if any) is in scope. Seeds the FabledSword family rulebook to always_on
+because that's the cross-project standards rulebook by design.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0058"
+down_revision = "0057"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "rulebooks",
+        sa.Column(
+            "always_on",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+    )
+    op.execute(
+        "UPDATE rulebooks SET always_on = TRUE WHERE title = 'FabledSword family'"
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("rulebooks", "always_on")

alembic/versions/0059_rule_project_scope.py

@@ -0,0 +1,49 @@
+"""project-scoped rules
+
+Revision ID: 0059
+Revises: 0058
+Create Date: 2026-06-01
+
+Rules can now belong to either a rulebook topic (cross-project standard) or
+a single project (project-scoped). Adds `rules.project_id`, makes `topic_id`
+nullable, and adds a CHECK constraint enforcing exactly-one. The previous
+unique constraint on (topic_id, title) still applies because PostgreSQL
+treats NULL as distinct — two project-scoped rules with the same title and
+NULL topic_id remain unique.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0059"
+down_revision = "0058"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "rules",
+        sa.Column(
+            "project_id",
+            sa.BigInteger(),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"),
+            nullable=True,
+        ),
+    )
+    op.alter_column("rules", "topic_id", nullable=True)
+    op.create_index("ix_rules_project_id", "rules", ["project_id"])
+    op.create_check_constraint(
+        "ck_rule_topic_xor_project",
+        "rules",
+        "(topic_id IS NULL) <> (project_id IS NULL)",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("ck_rule_topic_xor_project", "rules", type_="check")
+    op.drop_index("ix_rules_project_id", table_name="rules")
+    # Any rule with NULL topic_id will block re-tightening. Operator must
+    # migrate or delete project-scoped rules before downgrading.
+    op.alter_column("rules", "topic_id", nullable=False)
+    op.drop_column("rules", "project_id")

alembic/versions/0060_project_rule_suppressions.py

@@ -0,0 +1,73 @@
+"""project rule + topic suppressions
+
+Revision ID: 0060
+Revises: 0059
+Create Date: 2026-06-01
+
+Lets a project mute specific rules or whole topics from rulebooks it
+subscribes to, without unsubscribing the rulebook. Two pure many-to-many
+association tables; FKs CASCADE so removing a project / rule / topic
+cleans the suppression rows automatically.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0060"
+down_revision = "0059"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "project_rule_suppressions",
+        sa.Column(
+            "project_id",
+            sa.BigInteger(),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column(
+            "rule_id",
+            sa.BigInteger(),
+            sa.ForeignKey("rules.id", ondelete="CASCADE"),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_table(
+        "project_topic_suppressions",
+        sa.Column(
+            "project_id",
+            sa.BigInteger(),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column(
+            "topic_id",
+            sa.BigInteger(),
+            sa.ForeignKey("rulebook_topics.id", ondelete="CASCADE"),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("project_topic_suppressions")
+    op.drop_table("project_rule_suppressions")

frontend/src/api/rulebooks.ts

@@ -5,6 +5,7 @@ export interface Rulebook {
   owner_user_id: number;
   title: string;
   description: string;
+  always_on: boolean;
   created_at: string | null;
   updated_at: string | null;
 }
@@ -21,7 +22,8 @@ export interface RulebookTopic {
 
 export interface Rule {
   id: number;
-  topic_id: number;
+  topic_id: number | null;
+  project_id: number | null;
   title: string;
   statement: string;
   why: string;
@@ -35,7 +37,7 @@ export interface RuleHeader {
   id: number;
   title: string;
   statement: string;
-  topic_id: number;
+  topic_id: number | null;
 }
 
 export interface ApplicableRules {
@@ -43,7 +45,28 @@ export interface ApplicableRules {
     id: number;
     title: string;
     statement: string;
+    topic_id: number;
     topic_title: string;
+    rulebook_id: number;
+    rulebook_title: string;
+  }[];
+  project_rules: {
+    id: number;
+    title: string;
+    statement: string;
+  }[];
+  suppressed_rules: {
+    id: number;
+    title: string;
+    topic_id: number;
+    topic_title: string;
+    rulebook_id: number;
+    rulebook_title: string;
+  }[];
+  suppressed_topics: {
+    id: number;
+    title: string;
+    rulebook_id: number;
     rulebook_title: string;
   }[];
   truncated: boolean;
@@ -65,7 +88,7 @@ export async function createRulebook(data: { title: string; description?: string
   return apiPost("/api/rulebooks", data);
 }
 
-export async function updateRulebook(id: number, data: Partial<{ title: string; description: string }>): Promise<Rulebook> {
+export async function updateRulebook(id: number, data: Partial<{ title: string; description: string; always_on: boolean }>): Promise<Rulebook> {
   return apiPatch(`/api/rulebooks/${id}`, data);
 }
 
@@ -133,3 +156,28 @@ export async function unsubscribeProject(projectId: number, rulebookId: number):
 export async function getProjectApplicableRules(projectId: number): Promise<ApplicableRules> {
   return apiGet(`/api/projects/${projectId}/rules`);
 }
+
+export async function createProjectRule(
+  projectId: number,
+  data: { statement: string; title?: string; why?: string; how_to_apply?: string },
+): Promise<Rule> {
+  return apiPost(`/api/projects/${projectId}/rules`, data);
+}
+
+// ── Suppressions ───────────────────────────────────────────────────
+
+export async function suppressRuleForProject(projectId: number, ruleId: number): Promise<void> {
+  await apiPost(`/api/projects/${projectId}/suppressions/rules/${ruleId}`, {});
+}
+
+export async function unsuppressRuleForProject(projectId: number, ruleId: number): Promise<void> {
+  return apiDelete(`/api/projects/${projectId}/suppressions/rules/${ruleId}`);
+}
+
+export async function suppressTopicForProject(projectId: number, topicId: number): Promise<void> {
+  await apiPost(`/api/projects/${projectId}/suppressions/topics/${topicId}`, {});
+}
+
+export async function unsuppressTopicForProject(projectId: number, topicId: number): Promise<void> {
+  return apiDelete(`/api/projects/${projectId}/suppressions/topics/${topicId}`);
+}

frontend/src/components/rules/ProjectRulesTab.vue

@@ -3,7 +3,9 @@ import { ref, onMounted, watch } from "vue";
 import { useRouter } from "vue-router";
 import {
   getProjectApplicableRules, subscribeProject, unsubscribeProject,
-  listRulebooks, getRule,
+  listRulebooks, getRule, createProjectRule, deleteRule,
+  suppressRuleForProject, unsuppressRuleForProject,
+  suppressTopicForProject, unsuppressTopicForProject,
 } from "@/api/rulebooks";
 import type { ApplicableRules, Rulebook } from "@/api/rulebooks";
 
@@ -16,6 +18,9 @@ const expandedRuleIds = ref<Set<number>>(new Set());
 
 const ruleDetails = ref<Record<number, { why: string; how_to_apply: string }>>({});
 
+const showProjectRuleForm = ref(false);
+const newProjectRule = ref({ title: "", statement: "", why: "", how_to_apply: "" });
+
 async function load() {
   applicable.value = await getProjectApplicableRules(props.projectId);
 }
@@ -59,18 +64,74 @@ function openInRulesView(rulebookId: number, ruleId?: number) {
   router.push({ path: "/rules", query });
 }
 
-function groupByRulebookAndTopic(rules: ApplicableRules["rules"]) {
-  const grouped: Record<string, Record<string, ApplicableRules["rules"]>> = {};
+interface TopicGroup {
+  topic_id: number;
+  topic_title: string;
+  rules: ApplicableRules["rules"];
+}
+interface RulebookGroup {
+  rulebook_id: number;
+  rulebook_title: string;
+  topics: TopicGroup[];
+}
+function groupByRulebookAndTopic(rules: ApplicableRules["rules"]): RulebookGroup[] {
+  const byRulebook = new Map<number, RulebookGroup>();
   for (const r of rules) {
-    if (!grouped[r.rulebook_title]) grouped[r.rulebook_title] = {};
-    if (!grouped[r.rulebook_title][r.topic_title]) grouped[r.rulebook_title][r.topic_title] = [];
-    grouped[r.rulebook_title][r.topic_title].push(r);
+    let rb = byRulebook.get(r.rulebook_id);
+    if (!rb) {
+      rb = { rulebook_id: r.rulebook_id, rulebook_title: r.rulebook_title, topics: [] };
+      byRulebook.set(r.rulebook_id, rb);
+    }
+    let topic = rb.topics.find((t) => t.topic_id === r.topic_id);
+    if (!topic) {
+      topic = { topic_id: r.topic_id, topic_title: r.topic_title, rules: [] };
+      rb.topics.push(topic);
+    }
+    topic.rules.push(r);
   }
-  return grouped;
+  return Array.from(byRulebook.values());
 }
 
-function rulebookIdForTitle(title: string): number | undefined {
-  return applicable.value?.subscribed_rulebooks.find((rb) => rb.title === title)?.id;
+async function submitProjectRule() {
+  const statement = newProjectRule.value.statement.trim();
+  if (!statement) return;
+  await createProjectRule(props.projectId, {
+    statement,
+    title: newProjectRule.value.title.trim() || undefined,
+    why: newProjectRule.value.why.trim() || undefined,
+    how_to_apply: newProjectRule.value.how_to_apply.trim() || undefined,
+  });
+  newProjectRule.value = { title: "", statement: "", why: "", how_to_apply: "" };
+  showProjectRuleForm.value = false;
+  await load();
+}
+
+async function removeProjectRule(ruleId: number) {
+  if (!confirm("Delete this project rule? It will move to the trash.")) return;
+  await deleteRule(ruleId);
+  await load();
+}
+
+const showSuppressed = ref(false);
+
+async function suppressRule(ruleId: number) {
+  await suppressRuleForProject(props.projectId, ruleId);
+  await load();
+}
+
+async function unsuppressRule(ruleId: number) {
+  await unsuppressRuleForProject(props.projectId, ruleId);
+  await load();
+}
+
+async function suppressTopic(topicId: number) {
+  await suppressTopicForProject(props.projectId, topicId);
+  await load();
+}
+
+async function unsuppressTopic(topicId: number) {
+  await unsuppressTopicForProject(props.projectId, topicId);
+  await load();
 }
 
 onMounted(async () => {
@@ -111,6 +172,69 @@ watch(() => props.projectId, load);
       </div>
     </section>
 
+    <section class="project-rules">
+      <div class="section-head">
+        <h3>Project rules</h3>
+        <button
+          v-if="!showProjectRuleForm"
+          class="add"
+          @click="showProjectRuleForm = true"
+        >
+          + New project rule
+        </button>
+      </div>
+      <form v-if="showProjectRuleForm" class="new-rule-form" @submit.prevent="submitProjectRule">
+        <input
+          v-model="newProjectRule.title"
+          placeholder="Title (optional — derived from statement if blank)"
+        />
+        <textarea
+          v-model="newProjectRule.statement"
+          required
+          autofocus
+          placeholder="Statement (required) — the actionable instruction, 1-2 sentences"
+          rows="2"
+        ></textarea>
+        <textarea
+          v-model="newProjectRule.why"
+          placeholder="Why (optional) — the rationale"
+          rows="2"
+        ></textarea>
+        <textarea
+          v-model="newProjectRule.how_to_apply"
+          placeholder="How to apply (optional) — when / where it kicks in"
+          rows="2"
+        ></textarea>
+        <div class="form-buttons">
+          <button type="submit">Create</button>
+          <button type="button" @click="showProjectRuleForm = false">Cancel</button>
+        </div>
+      </form>
+      <ul v-if="applicable.project_rules && applicable.project_rules.length > 0" class="rule-list">
+        <li v-for="r in applicable.project_rules" :key="r.id" class="rule">
+          <div class="rule-head" @click="toggleRuleExpand(r.id)">
+            <span class="rule-title">{{ r.title }}</span>
+            <span class="rule-statement">{{ r.statement }}</span>
+          </div>
+          <div v-if="expandedRuleIds.has(r.id) && ruleDetails[r.id]" class="rule-detail">
+            <div v-if="ruleDetails[r.id].why">
+              <strong>Why:</strong> {{ ruleDetails[r.id].why }}
+            </div>
+            <div v-if="ruleDetails[r.id].how_to_apply">
+              <strong>How to apply:</strong> {{ ruleDetails[r.id].how_to_apply }}
+            </div>
+            <button class="delete-link" @click="removeProjectRule(r.id)">Delete</button>
+          </div>
+        </li>
+      </ul>
+      <p
+        v-else-if="!showProjectRuleForm"
+        class="empty"
+      >
+        No project-only rules yet.
+      </p>
+    </section>
+
     <section class="applicable">
       <h3>Applicable rules</h3>
       <p v-if="applicable.rules.length === 0" class="empty">
@@ -118,18 +242,32 @@ watch(() => props.projectId, load);
         <a @click="router.push('/rules')">Rulebooks</a>.
       </p>
       <div
-        v-for="(topics, rbTitle) in groupByRulebookAndTopic(applicable.rules)"
-        :key="rbTitle"
+        v-for="rb in groupByRulebookAndTopic(applicable.rules)"
+        :key="rb.rulebook_id"
         class="rb-group"
       >
-        <h4>{{ rbTitle }}</h4>
-        <div v-for="(rules, topicTitle) in topics" :key="topicTitle" class="topic-group">
-          <h5>{{ topicTitle }}</h5>
+        <h4>{{ rb.rulebook_title }}</h4>
+        <div v-for="topic in rb.topics" :key="topic.topic_id" class="topic-group">
+          <h5>
+            <span>{{ topic.topic_title }}</span>
+            <button
+              class="skip-btn"
+              :title="`Skip the entire ${topic.topic_title} topic for this project`"
+              @click="suppressTopic(topic.topic_id)"
+            >× skip topic</button>
+          </h5>
           <ul>
-            <li v-for="r in rules" :key="r.id" class="rule">
-              <div class="rule-head" @click="toggleRuleExpand(r.id)">
-                <span class="rule-title">{{ r.title }}</span>
-                <span class="rule-statement">{{ r.statement }}</span>
+            <li v-for="r in topic.rules" :key="r.id" class="rule">
+              <div class="rule-head">
+                <div class="rule-head-text" @click="toggleRuleExpand(r.id)">
+                  <span class="rule-title">{{ r.title }}</span>
+                  <span class="rule-statement">{{ r.statement }}</span>
+                </div>
+                <button
+                  class="skip-btn"
+                  title="Skip this rule for this project"
+                  @click.stop="suppressRule(r.id)"
+                >× skip</button>
               </div>
               <div v-if="expandedRuleIds.has(r.id) && ruleDetails[r.id]" class="rule-detail">
                 <div v-if="ruleDetails[r.id].why">
@@ -140,7 +278,7 @@ watch(() => props.projectId, load);
                 </div>
                 <button
                   class="edit-link"
-                  @click="rulebookIdForTitle(String(rbTitle)) && openInRulesView(rulebookIdForTitle(String(rbTitle))!, r.id)"
+                  @click="openInRulesView(r.rulebook_id, r.id)"
                 >
                   Edit in Rulebook →
                 </button>
@@ -153,6 +291,32 @@ watch(() => props.projectId, load);
         Truncated at 50 rules — there are more applicable rules.
       </p>
     </section>
+
+    <section
+      v-if="applicable.suppressed_rules.length + applicable.suppressed_topics.length > 0"
+      class="suppressed"
+    >
+      <button class="suppressed-toggle" @click="showSuppressed = !showSuppressed">
+        <span>Suppressed ({{ applicable.suppressed_rules.length + applicable.suppressed_topics.length }})</span>
+        <span class="caret">{{ showSuppressed ? "▾" : "▸" }}</span>
+      </button>
+      <div v-if="showSuppressed" class="suppressed-body">
+        <ul v-if="applicable.suppressed_topics.length > 0" class="suppressed-list">
+          <li v-for="t in applicable.suppressed_topics" :key="`topic-${t.id}`">
+            <span class="suppressed-kind">topic</span>
+            <span class="suppressed-path">{{ t.rulebook_title }} → {{ t.title }}</span>
+            <button class="reenable-btn" @click="unsuppressTopic(t.id)">↻ re-enable</button>
+          </li>
+        </ul>
+        <ul v-if="applicable.suppressed_rules.length > 0" class="suppressed-list">
+          <li v-for="r in applicable.suppressed_rules" :key="`rule-${r.id}`">
+            <span class="suppressed-kind">rule</span>
+            <span class="suppressed-path">{{ r.rulebook_title }} → {{ r.topic_title }} → {{ r.title }}</span>
+            <button class="reenable-btn" @click="unsuppressRule(r.id)">↻ re-enable</button>
+          </li>
+        </ul>
+      </div>
+    </section>
   </div>
 </template>
 
@@ -208,4 +372,67 @@ ul { list-style: none; padding: 0; margin: 0; }
 }
 .empty, .truncated { opacity: 0.7; font-style: italic; }
 .empty a { cursor: pointer; text-decoration: underline; }
+.project-rules { margin-top: 1.5rem; }
+.section-head { display: flex; justify-content: space-between; align-items: center; }
+.new-rule-form {
+  display: flex; flex-direction: column; gap: 0.5rem;
+  padding: 0.75rem; margin: 0.5rem 0;
+  background: var(--color-bg, #111113);
+  border: 1px solid var(--color-border, #2a2a2e); border-radius: 6px;
+}
+.new-rule-form input, .new-rule-form textarea {
+  background: var(--color-surface, #18181b); color: inherit;
+  border: 1px solid var(--color-border, #2a2a2e); border-radius: 6px;
+  padding: 0.5rem; font: inherit; resize: vertical;
+}
+.rule-list { margin-top: 0.5rem; }
+.delete-link {
+  background: none; border: none; cursor: pointer;
+  color: var(--color-destructive, #b85a4a); padding: 0.5rem 0 0 0;
+}
+/* Per-rule / per-topic suppress affordance — quiet by default, reveal on hover */
+.topic-group h5 {
+  display: flex; justify-content: space-between; align-items: center; gap: 0.5rem;
+}
+.rule-head {
+  display: flex; justify-content: space-between; align-items: flex-start; gap: 0.5rem;
+}
+.rule-head-text { flex: 1; cursor: pointer; }
+.skip-btn {
+  background: none; border: none; cursor: pointer;
+  color: var(--color-muted, #888); font-size: 0.75rem;
+  padding: 0.1rem 0.4rem; opacity: 0; transition: opacity 0.15s;
+  white-space: nowrap;
+}
+.topic-group h5:hover .skip-btn,
+.rule:hover .skip-btn,
+.skip-btn:focus { opacity: 1; }
+.skip-btn:hover { color: var(--color-destructive, #b85a4a); }
+/* Suppressed section */
+.suppressed { margin-top: 1.5rem; }
+.suppressed-toggle {
+  display: flex; align-items: center; gap: 0.4rem;
+  background: none; border: none; cursor: pointer;
+  font-size: 0.85rem; opacity: 0.7; padding: 0.25rem 0; color: inherit;
+}
+.suppressed-toggle:hover { opacity: 1; }
+.suppressed-toggle .caret { font-size: 0.7em; }
+.suppressed-body { margin-top: 0.5rem; }
+.suppressed-list { padding-left: 0; }
+.suppressed-list li {
+  display: flex; align-items: center; gap: 0.5rem;
+  padding: 0.25rem 0; opacity: 0.75;
+}
+.suppressed-kind {
+  font-size: 0.7em; text-transform: uppercase; letter-spacing: 0.05em;
+  padding: 0.1rem 0.4rem; border-radius: 3px;
+  background: var(--color-bg, #111113);
+  border: 1px solid var(--color-border, #2a2a2e);
+}
+.suppressed-path { flex: 1; }
+.reenable-btn {
+  background: none; border: none; cursor: pointer;
+  color: var(--color-primary, #6366f1); font-size: 0.85em;
+}
+.reenable-btn:hover { text-decoration: underline; }
 </style>

frontend/src/components/rules/RulebookDetailPane.vue

@@ -1,5 +1,5 @@
 <script setup lang="ts">
-import { ref, onMounted, watch } from "vue";
+import { ref, computed, onMounted, watch } from "vue";
 import { useRulebooksStore } from "@/stores/rulebooks";
 import { apiGet } from "@/api/client";
 import {
@@ -18,6 +18,10 @@ const store = useRulebooksStore();
 const isCreating = ref(false);
 const newTitle = ref("");
 
+const currentRulebook = computed(() =>
+  store.rulebooks.find((rb) => rb.id === props.rulebookId),
+);
+
 interface ProjectLite { id: number; title: string }
 const projects = ref<ProjectLite[]>([]);
 // Map<project_id, Set<rulebook_id>>
@@ -68,7 +72,17 @@ watch(() => props.rulebookId, () => {/* re-render of isSubscribed from existing
 
 <template>
   <section class="pane">
-    <header><h2>Topics</h2></header>
+    <header>
+      <h2>Topics</h2>
+      <label v-if="currentRulebook" class="always-on-toggle" title="When on, rules from this rulebook load at session start regardless of project context">
+        <input
+          type="checkbox"
+          :checked="currentRulebook.always_on"
+          @change="store.toggleAlwaysOn(currentRulebook.id)"
+        />
+        <span>Always on</span>
+      </label>
+    </header>
     <ul>
       <li
         v-for="t in topics"
@@ -109,7 +123,14 @@ watch(() => props.rulebookId, () => {/* re-render of isSubscribed from existing
 
 <style scoped>
 .pane { background: var(--color-surface, #18181b); padding: 1rem; overflow-y: auto; }
+header { display: flex; align-items: center; justify-content: space-between; gap: 1rem; }
 header h2 { font-family: Fraunces, serif; font-style: italic; margin: 0 0 0.5rem 0; }
+.always-on-toggle {
+  display: flex; align-items: center; gap: 0.4rem;
+  font-size: 0.85rem; opacity: 0.85; cursor: pointer;
+  user-select: none;
+}
+.always-on-toggle input { cursor: pointer; }
 ul { list-style: none; padding: 0; margin: 1rem 0; }
 li { padding: 0.5rem; cursor: pointer; border-radius: 6px; }
 li.active { background: var(--color-primary-bg, rgba(99,102,241,0.15)); }

frontend/src/components/rules/RulebookListPane.vue

@@ -31,6 +31,7 @@ async function submitNew() {
         @click="emit('select', rb.id)"
       >
         <span class="title">{{ rb.title }}</span>
+        <span v-if="rb.always_on" class="always-on-badge" title="Loaded at session start">always on</span>
       </li>
     </ul>
     <div class="new-rulebook">
@@ -50,9 +51,19 @@ async function submitNew() {
 .pane { background: var(--color-surface, #18181b); padding: 1rem; overflow-y: auto; }
 header h2 { font-family: Fraunces, serif; font-style: italic; margin: 0 0 0.5rem 0; }
 ul { list-style: none; padding: 0; margin: 1rem 0; }
-li { padding: 0.5rem; cursor: pointer; border-radius: 6px; }
+li { padding: 0.5rem; cursor: pointer; border-radius: 6px; display: flex; align-items: center; gap: 0.5rem; }
 li.active { background: var(--color-primary-bg, rgba(99,102,241,0.15)); }
 li:hover { background: var(--color-hover, rgba(255,255,255,0.05)); }
+.always-on-badge {
+  font-size: 0.7rem;
+  text-transform: uppercase;
+  letter-spacing: 0.05em;
+  padding: 0.1rem 0.4rem;
+  border-radius: 3px;
+  background: var(--color-accent, rgba(91,74,138,0.25));
+  color: var(--color-accent-fg, inherit);
+  margin-left: auto;
+}
 .new-rulebook { margin-top: 1rem; }
 .new-rulebook input {
   width: 100%; margin-bottom: 0.5rem;

frontend/src/stores/rulebooks.ts

@@ -54,13 +54,19 @@ export const useRulebooksStore = defineStore("rulebooks", () => {
     return rb;
   }
 
-  async function updateRulebook(id: number, data: Partial<Pick<Rulebook, "title" | "description">>) {
+  async function updateRulebook(id: number, data: Partial<Pick<Rulebook, "title" | "description" | "always_on">>) {
     const rb = await api.updateRulebook(id, data);
     const idx = rulebooks.value.findIndex((r) => r.id === id);
     if (idx >= 0) rulebooks.value[idx] = rb;
     return rb;
   }
 
+  async function toggleAlwaysOn(id: number) {
+    const current = rulebooks.value.find((r) => r.id === id);
+    if (!current) return;
+    return updateRulebook(id, { always_on: !current.always_on });
+  }
+
   async function deleteRulebook(id: number) {
     await api.deleteRulebook(id);
     rulebooks.value = rulebooks.value.filter((r) => r.id !== id);
@@ -121,7 +127,7 @@ export const useRulebooksStore = defineStore("rulebooks", () => {
   return {
     rulebooks, topicsByRulebook, rulesByTopic, currentRule, loading,
     fetchRulebooks, fetchTopics, fetchRules, fetchRule,
-    createRulebook, updateRulebook, deleteRulebook,
+    createRulebook, updateRulebook, toggleAlwaysOn, deleteRulebook,
     createTopic, updateTopic, deleteTopic,
     createRule, updateRule, deleteRule,
   };

src/fabledassistant/mcp/server.py

@@ -39,14 +39,29 @@ Mechanics:
   "not set".
 
 Scribe maintains a Rulebook system (Rulebook -> Topic -> Rule). Rules carry
-an actionable statement plus optional Why and How-to-apply context. When you
-start work on a project, get_project(id) returns applicable_rules (the rules
-from rulebooks the project subscribes to) and subscribed_rulebooks. Consult
-these before making decisions about workflow, conventions, or scope. Full
-text (Why / How-to-apply) is available via get_rule(id). You may create new
-rules via create_rule when you notice a pattern worth codifying — coordinate
-with the operator on whether it belongs in an existing rulebook+topic or a
-new one.
+an actionable statement plus optional Why and How-to-apply context. At the
+start of any session that touches Scribe, call list_always_on_rules() to
+load the standing rules — treat them as binding. When you also have a project
+in scope, get_project(id) returns applicable_rules (rules from rulebooks the
+project subscribes to) and subscribed_rulebooks; consult those too. Full text
+(Why / How-to-apply) is available via get_rule(id).
+
+Engineering and workflow rules live in Scribe. When you notice a pattern
+worth codifying, call create_rule (cross-project, lands in a rulebook+topic)
+or create_project_rule (one project only, no rulebook ceremony). Do NOT add
+new engineering rules to CLAUDE.md or to ~/.claude/.../memory/feedback_*.md
+— those stores are reserved for facts about the user (preferences, role,
+communication style) and codebase onboarding pointers, respectively. Before
+creating a rule, call list_always_on_rules and list_rules(project_id=...) to
+avoid duplicates. Coordinate with the operator on whether a new rule belongs
+in a project, an existing rulebook+topic, or a new rulebook.
+
+When you are working on a specific project, call enter_project(project_id)
+ONCE at session start (or whenever the active project changes). It returns the
+project, its applicable_rules + project_rules + subscribed_rulebooks, milestone
+summary, open tasks, and recent notes — everything you need to know the lay of
+the land before mutating. Don't call get_project + get_applicable_rules + a
+search separately when enter_project already composes them.
 
 Plans are tasks with kind=plan, and Scribe is the canonical home for them.
 When you begin non-trivial work, call start_planning(project_id, title) FIRST —

src/fabledassistant/mcp/tools/projects.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 
 from fabledassistant.mcp._context import current_user_id
 from fabledassistant.services import milestones as milestones_svc
+from fabledassistant.services import notes as notes_svc
 from fabledassistant.services import projects as projects_svc
 from fabledassistant.services import rulebooks as rulebooks_svc
 from fabledassistant.services import trash as trash_svc
@@ -34,6 +35,73 @@ async def list_projects() -> dict:
     return {"projects": [p.to_dict() for p in rows]}
 
 
+async def enter_project(project_id: int) -> dict:
+    """Session-start handshake: load full context for working on a project.
+
+    Call this FIRST whenever you're about to do project-scoped work
+    (start_planning, create_task, update_*, anything that takes a project_id).
+    One round-trip returns the project, its applicable rules (both rulebook-
+    subscribed and project-scoped), milestone progress, open tasks, and
+    recently-updated notes — everything you need to know the lay of the land
+    before mutating.
+
+    No persistent server state: this is a read snapshot. Re-call if the
+    session goes idle long enough that the data feels stale.
+
+    Args:
+        project_id: The project to enter.
+
+    Returns a dict with keys: project, milestone_summary, applicable_rules,
+    project_rules, subscribed_rulebooks, applicable_rules_truncated,
+    open_tasks, recent_notes.
+    """
+    uid = current_user_id()
+    project = await projects_svc.get_project(uid, project_id)
+    if project is None:
+        raise ValueError(f"project {project_id} not found")
+
+    applicable = await rulebooks_svc.get_applicable_rules(
+        project_id=project_id, user_id=uid,
+    )
+    milestone_summary = await milestones_svc.get_project_milestone_summary(
+        uid, project_id,
+    )
+    open_tasks, _ = await notes_svc.list_notes(
+        uid, is_task=True, project_id=project_id,
+        status=["todo", "in_progress"], sort="updated_at", limit=10,
+    )
+    recent_notes, _ = await notes_svc.list_notes(
+        uid, is_task=False, project_id=project_id,
+        sort="updated_at", limit=5,
+    )
+
+    return {
+        "project": project.to_dict(),
+        "milestone_summary": milestone_summary,
+        "applicable_rules": applicable["rules"],
+        "project_rules": applicable.get("project_rules", []),
+        "suppressed_rules": applicable.get("suppressed_rules", []),
+        "suppressed_topics": applicable.get("suppressed_topics", []),
+        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
+        "applicable_rules_truncated": applicable["truncated"],
+        "open_tasks": [
+            {
+                "id": t.id, "title": t.title, "status": t.status,
+                "priority": t.priority, "task_kind": t.task_kind,
+                "milestone_id": t.milestone_id,
+            }
+            for t in open_tasks
+        ],
+        "recent_notes": [
+            {
+                "id": n.id, "title": n.title,
+                "updated_at": n.updated_at.isoformat() if n.updated_at else None,
+            }
+            for n in recent_notes
+        ],
+    }
+
+
 async def get_project(project_id: int) -> dict:
     """Fetch a Scribe project by ID.
 
@@ -55,6 +123,9 @@ async def get_project(project_id: int) -> dict:
     data["applicable_rules"] = applicable["rules"]
     data["applicable_rules_truncated"] = applicable["truncated"]
     data["subscribed_rulebooks"] = applicable["subscribed_rulebooks"]
+    data["project_rules"] = applicable.get("project_rules", [])
+    data["suppressed_rules"] = applicable.get("suppressed_rules", [])
+    data["suppressed_topics"] = applicable.get("suppressed_topics", [])
     return data
 
 
@@ -136,6 +207,7 @@ async def delete_project(project_id: int) -> dict:
 def register(mcp) -> None:
     for fn in (
         list_projects,
+        enter_project,
         get_project,
         create_project,
         update_project,

src/fabledassistant/mcp/tools/rulebooks.py

@@ -54,14 +54,26 @@ async def create_rulebook(title: str, description: str = "") -> dict:
 
 async def update_rulebook(
     rulebook_id: int, title: str = "", description: str = "",
+    always_on: bool | None = None,
 ) -> dict:
-    """Update an existing rulebook. Only non-empty fields are changed."""
+    """Update an existing rulebook. Only non-empty fields are changed.
+
+    Args:
+        rulebook_id: Rulebook to update.
+        title: New title. Empty string leaves unchanged.
+        description: New description. Empty string leaves unchanged.
+        always_on: When True, rules in this rulebook are loaded at session
+            start by list_always_on_rules regardless of project context.
+            Pass None to leave unchanged.
+    """
     uid = current_user_id()
     fields: dict = {}
     if title:
         fields["title"] = title
     if description:
         fields["description"] = description
+    if always_on is not None:
+        fields["always_on"] = always_on
     rb = await rulebooks_svc.update_rulebook(rulebook_id, uid, **fields)
     if rb is None:
         raise ValueError(f"rulebook {rulebook_id} not found")
@@ -200,6 +212,28 @@ async def list_rules(
     }
 
 
+async def list_always_on_rules() -> dict:
+    """Return all rules from rulebooks flagged always_on for the current user.
+
+    Call this at session start. Treat the returned rules as binding for the
+    session — they apply regardless of which project (if any) is in scope.
+    Pair with get_project(id).applicable_rules when working on a specific
+    project to also load that project's subscription-derived rules.
+    """
+    uid = current_user_id()
+    rules = await rulebooks_svc.list_always_on_rules(uid)
+    return {
+        "rules": [
+            {
+                "id": r.id, "title": r.title, "statement": r.statement,
+                "topic_id": r.topic_id,
+            }
+            for r in rules
+        ],
+        "total": len(rules),
+    }
+
+
 async def get_rule(rule_id: int) -> dict:
     """Fetch a rule by id — full statement + why + how_to_apply."""
     uid = current_user_id()
@@ -213,7 +247,7 @@ async def create_rule(
     topic_id: int, title: str, statement: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
 ) -> dict:
-    """Create a new rule under a topic.
+    """Create a new rule under a topic (cross-project rulebook rule).
 
     Args:
         topic_id: The topic to attach the rule to.
@@ -222,6 +256,9 @@ async def create_rule(
         why: Optional rationale — the reason the rule exists.
         how_to_apply: Optional operationalization — when / where it kicks in.
         order_index: Display order within the topic (default 0).
+
+    For a rule that applies to a single project only, use create_project_rule
+    instead — no rulebook+topic ceremony required.
     """
     uid = current_user_id()
     rule = await rulebooks_svc.create_rule(
@@ -232,6 +269,35 @@ async def create_rule(
     return rule.to_dict()
 
 
+async def create_project_rule(
+    project_id: int, statement: str, title: str = "",
+    why: str = "", how_to_apply: str = "", order_index: int = 0,
+) -> dict:
+    """Create a rule scoped to a single project (no rulebook needed).
+
+    Use this when a rule only applies to one project — it bypasses the
+    Rulebook -> Topic -> Rule ceremony. The rule is returned in get_project's
+    applicable_rules (under project_rules) and in list_rules(project_id=...).
+
+    Args:
+        project_id: The project to attach the rule to.
+        statement: The actionable instruction (required). 1-2 sentences.
+        title: Short imperative title. If empty, derived from the first ~50
+            characters of statement.
+        why: Optional rationale — the reason the rule exists.
+        how_to_apply: Optional operationalization — when / where it kicks in.
+        order_index: Display order within the project's rule list (default 0).
+    """
+    uid = current_user_id()
+    derived_title = title.strip() or statement.strip().split(".")[0][:50]
+    rule = await rulebooks_svc.create_project_rule(
+        project_id=project_id, user_id=uid,
+        title=derived_title, statement=statement,
+        why=why, how_to_apply=how_to_apply, order_index=order_index,
+    )
+    return rule.to_dict()
+
+
 async def update_rule(
     rule_id: int, title: str = "", statement: str = "",
     why: str = "", how_to_apply: str = "", order_index: int = -1,
@@ -298,11 +364,70 @@ async def unsubscribe_project_from_rulebook(
     return {"project_id": project_id, "rulebook_id": rulebook_id, "subscribed": False}
 
 
+# ── Suppressions — project-level mute of rulebook rules / topics ────────
+
+async def suppress_rule_for_project(
+    project_id: int, rule_id: int,
+) -> dict:
+    """Mute a single rulebook rule for one project.
+
+    The rule stays in its rulebook for other projects; only this project
+    skips it. Idempotent. Use unsuppress_rule_for_project to re-enable.
+    Project-scoped rules (create_project_rule) are NOT suppressible — delete
+    them with delete_rule instead.
+    """
+    uid = current_user_id()
+    await rulebooks_svc.suppress_rule_for_project(
+        project_id=project_id, rule_id=rule_id, user_id=uid,
+    )
+    return {"project_id": project_id, "rule_id": rule_id, "suppressed": True}
+
+
+async def unsuppress_rule_for_project(
+    project_id: int, rule_id: int,
+) -> dict:
+    """Re-enable a previously-suppressed rule for one project. Idempotent."""
+    uid = current_user_id()
+    await rulebooks_svc.unsuppress_rule_for_project(
+        project_id=project_id, rule_id=rule_id, user_id=uid,
+    )
+    return {"project_id": project_id, "rule_id": rule_id, "suppressed": False}
+
+
+async def suppress_topic_for_project(
+    project_id: int, topic_id: int,
+) -> dict:
+    """Mute every rule under a topic for one project.
+
+    Equivalent to suppressing each rule in the topic individually, but
+    auto-includes new rules added to the topic later. Idempotent.
+    """
+    uid = current_user_id()
+    await rulebooks_svc.suppress_topic_for_project(
+        project_id=project_id, topic_id=topic_id, user_id=uid,
+    )
+    return {"project_id": project_id, "topic_id": topic_id, "suppressed": True}
+
+
+async def unsuppress_topic_for_project(
+    project_id: int, topic_id: int,
+) -> dict:
+    """Re-enable a previously-suppressed topic for one project. Idempotent."""
+    uid = current_user_id()
+    await rulebooks_svc.unsuppress_topic_for_project(
+        project_id=project_id, topic_id=topic_id, user_id=uid,
+    )
+    return {"project_id": project_id, "topic_id": topic_id, "suppressed": False}
+
+
 def register(mcp) -> None:
     for fn in (
         list_rulebooks, get_rulebook, create_rulebook, update_rulebook, delete_rulebook,
         list_topics, create_topic, update_topic, delete_topic,
-        list_rules, get_rule, create_rule, update_rule, delete_rule,
+        list_rules, list_always_on_rules, get_rule,
+        create_rule, create_project_rule, update_rule, delete_rule,
         subscribe_project_to_rulebook, unsubscribe_project_from_rulebook,
+        suppress_rule_for_project, unsuppress_rule_for_project,
+        suppress_topic_for_project, unsuppress_topic_for_project,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/fabledassistant/mcp/tools/tasks.py

@@ -82,6 +82,9 @@ async def get_task(task_id: int) -> dict:
         data["applicable_rules"] = applicable["rules"]
         data["subscribed_rulebooks"] = applicable["subscribed_rulebooks"]
         data["applicable_rules_truncated"] = applicable["truncated"]
+        data["project_rules"] = applicable.get("project_rules", [])
+        data["suppressed_rules"] = applicable.get("suppressed_rules", [])
+        data["suppressed_topics"] = applicable.get("suppressed_topics", [])
     return data
 
 

src/fabledassistant/models/rulebook.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timezone
 
-from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, Integer, Table, Text, UniqueConstraint
+from sqlalchemy import BigInteger, Boolean, Column, DateTime, ForeignKey, Integer, Table, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from fabledassistant.models import Base
@@ -16,6 +16,9 @@ class Rulebook(Base, SoftDeleteMixin):
     )
     title: Mapped[str] = mapped_column(Text)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    always_on: Mapped[bool] = mapped_column(
+        Boolean, default=False, nullable=False, server_default="false"
+    )
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
     )
@@ -31,6 +34,7 @@ class Rulebook(Base, SoftDeleteMixin):
             "owner_user_id": self.owner_user_id,
             "title": self.title,
             "description": self.description or "",
+            "always_on": self.always_on,
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None,
         }
@@ -77,8 +81,18 @@ class Rule(Base, SoftDeleteMixin):
     )
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
-    topic_id: Mapped[int] = mapped_column(
-        BigInteger, ForeignKey("rulebook_topics.id", ondelete="CASCADE")
+    # Exactly one of topic_id / project_id is set — enforced by CHECK
+    # constraint ck_rule_topic_xor_project (migration 0059).
+    topic_id: Mapped[int | None] = mapped_column(
+        BigInteger,
+        ForeignKey("rulebook_topics.id", ondelete="CASCADE"),
+        nullable=True,
+    )
+    project_id: Mapped[int | None] = mapped_column(
+        BigInteger,
+        ForeignKey("projects.id", ondelete="CASCADE"),
+        nullable=True,
+        index=True,
     )
     title: Mapped[str] = mapped_column(Text)
     statement: Mapped[str] = mapped_column(Text)
@@ -98,6 +112,7 @@ class Rule(Base, SoftDeleteMixin):
         return {
             "id": self.id,
             "topic_id": self.topic_id,
+            "project_id": self.project_id,
             "title": self.title,
             "statement": self.statement,
             "why": self.why or "",
@@ -116,3 +131,22 @@ project_rulebook_subscriptions = Table(
     Column("rulebook_id", BigInteger, ForeignKey("rulebooks.id", ondelete="CASCADE"), primary_key=True),
     Column("created_at", DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)),
 )
+
+# Suppressions — let a project mute individual rules or whole topics from
+# rulebooks it subscribes to, without unsubscribing the rulebook itself.
+# FKs CASCADE so the row vanishes when its parent is removed.
+project_rule_suppressions = Table(
+    "project_rule_suppressions",
+    Base.metadata,
+    Column("project_id", BigInteger, ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True),
+    Column("rule_id", BigInteger, ForeignKey("rules.id", ondelete="CASCADE"), primary_key=True),
+    Column("created_at", DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)),
+)
+
+project_topic_suppressions = Table(
+    "project_topic_suppressions",
+    Base.metadata,
+    Column("project_id", BigInteger, ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True),
+    Column("topic_id", BigInteger, ForeignKey("rulebook_topics.id", ondelete="CASCADE"), primary_key=True),
+    Column("created_at", DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)),
+)

src/fabledassistant/routes/rulebooks.py

@@ -55,7 +55,7 @@ async def get_rulebook(rulebook_id: int):
 @login_required
 async def update_rulebook(rulebook_id: int):
     data = await request.get_json() or {}
-    fields = {k: v for k, v in data.items() if k in ("title", "description")}
+    fields = {k: v for k, v in data.items() if k in ("title", "description", "always_on")}
     rb = await rulebooks_svc.update_rulebook(rulebook_id, _uid(), **fields)
     if rb is None:
         return jsonify({"error": "rulebook not found"}), 404
@@ -239,3 +239,75 @@ async def get_project_rules(project_id: int):
         project_id=project_id, user_id=_uid(),
     )
     return jsonify(result)
+
+
+@rulebooks_bp.post("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
+@login_required
+async def suppress_project_rule(project_id: int, rule_id: int):
+    try:
+        await rulebooks_svc.suppress_rule_for_project(
+            project_id=project_id, rule_id=rule_id, user_id=_uid(),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 404
+    return "", 204
+
+
+@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
+@login_required
+async def unsuppress_project_rule(project_id: int, rule_id: int):
+    try:
+        await rulebooks_svc.unsuppress_rule_for_project(
+            project_id=project_id, rule_id=rule_id, user_id=_uid(),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 404
+    return "", 204
+
+
+@rulebooks_bp.post("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
+@login_required
+async def suppress_project_topic(project_id: int, topic_id: int):
+    try:
+        await rulebooks_svc.suppress_topic_for_project(
+            project_id=project_id, topic_id=topic_id, user_id=_uid(),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 404
+    return "", 204
+
+
+@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
+@login_required
+async def unsuppress_project_topic(project_id: int, topic_id: int):
+    try:
+        await rulebooks_svc.unsuppress_topic_for_project(
+            project_id=project_id, topic_id=topic_id, user_id=_uid(),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 404
+    return "", 204
+
+
+@rulebooks_bp.post("/projects/<int:project_id>/rules")
+@login_required
+async def create_project_rule(project_id: int):
+    """Create a rule scoped to a single project. Frontend fast path."""
+    data = await request.get_json() or {}
+    statement = (data.get("statement") or "").strip()
+    if not statement:
+        return jsonify({"error": "statement is required"}), 400
+    title = (data.get("title") or "").strip() or statement.split(".")[0][:50]
+    try:
+        rule = await rulebooks_svc.create_project_rule(
+            project_id=project_id,
+            user_id=_uid(),
+            title=title,
+            statement=statement,
+            why=data.get("why", ""),
+            how_to_apply=data.get("how_to_apply", ""),
+            order_index=data.get("order_index", 0),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 404
+    return jsonify(rule.to_dict()), 201

src/fabledassistant/services/planning.py

@@ -58,6 +58,9 @@ async def start_planning(user_id: int, project_id: int, title: str) -> dict:
         "applicable_rules": applicable["rules"],
         "subscribed_rulebooks": applicable["subscribed_rulebooks"],
         "applicable_rules_truncated": applicable["truncated"],
+        "project_rules": applicable.get("project_rules", []),
+        "suppressed_rules": applicable.get("suppressed_rules", []),
+        "suppressed_topics": applicable.get("suppressed_topics", []),
         "project_goal": getattr(project, "goal", "") or "",
         "open_task_count": open_count,
     }

src/fabledassistant/services/rulebooks.py

@@ -72,7 +72,7 @@ async def update_rulebook(
         rb = result.scalar_one_or_none()
         if rb is None:
             return None
-        allowed = {"title", "description"}
+        allowed = {"title", "description", "always_on"}
         for key, value in fields.items():
             if key in allowed and value is not None:
                 setattr(rb, key, value)
@@ -242,6 +242,44 @@ async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
         raise ValueError(f"topic {topic_id} not found")
 
 
+async def _assert_project_owned(session, project_id: int, user_id: int) -> None:
+    """Raise ValueError if project doesn't exist or isn't owned by user."""
+    from fabledassistant.models.project import Project
+    result = await session.execute(
+        select(Project).where(
+            Project.id == project_id,
+            Project.user_id == user_id,
+            Project.deleted_at.is_(None),
+        )
+    )
+    if result.scalar_one_or_none() is 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 fabledassistant.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")
+
+
 async def create_rule(
     topic_id: int, user_id: int, title: str, statement: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
@@ -262,6 +300,32 @@ async def create_rule(
         return rule
 
 
+async def create_project_rule(
+    project_id: int, user_id: int, title: str, statement: str,
+    why: str = "", how_to_apply: str = "", order_index: int = 0,
+) -> 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).
+    """
+    async with async_session() as session:
+        await _assert_project_owned(session, project_id, user_id)
+        rule = Rule(
+            project_id=project_id,
+            title=title,
+            statement=statement,
+            why=why or None,
+            how_to_apply=how_to_apply or None,
+            order_index=order_index,
+        )
+        session.add(rule)
+        await session.commit()
+        await session.refresh(rule)
+        return rule
+
+
 async def list_rules(
     user_id: int,
     rulebook_id: int | None = None,
@@ -270,7 +334,12 @@ async def list_rules(
 ) -> list[Rule]:
     """List rules filtered by any of the three IDs. All filters are ownership-scoped.
 
-    project_id resolves rules through project_rulebook_subscriptions.
+    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.
     """
     from fabledassistant.models.rulebook import project_rulebook_subscriptions
 
@@ -302,38 +371,101 @@ async def list_rules(
             Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
         )
         result = await session.execute(stmt)
-        return list(result.scalars().all())
+        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 fabledassistant.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())
 
 
-async def get_rule(rule_id: int, user_id: int) -> Optional[Rule]:
+async def list_always_on_rules(user_id: int, limit: int = 100) -> 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.
+    """
     async with async_session() as session:
         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,
+                Rulebook.always_on.is_(True),
                 Rule.deleted_at.is_(None),
                 RulebookTopic.deleted_at.is_(None),
                 Rulebook.deleted_at.is_(None),
             )
+            .order_by(
+                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
+            )
+            .limit(limit)
         )
-        return result.scalar_one_or_none()
+        return list(result.scalars().all())
+
+
+async def _fetch_owned_rule(session, rule_id: int, user_id: int) -> Optional[Rule]:
+    """Fetch a rule by id, scoped to user owning either its rulebook
+    (via topic) or its project (via project_id). Honors soft-delete.
+    Returns None when not found or not owned.
+    """
+    from fabledassistant.models.project import Project
+
+    # Path A — rulebook rule.
+    rulebook_rule = (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),
+        )
+    )).scalar_one_or_none()
+    if rulebook_rule is not None:
+        return rulebook_rule
+
+    # Path B — project-scoped rule.
+    project_rule = (await session.execute(
+        select(Rule)
+        .join(Project, Rule.project_id == Project.id)
+        .where(
+            Rule.id == rule_id,
+            Project.user_id == user_id,
+            Rule.deleted_at.is_(None),
+            Project.deleted_at.is_(None),
+        )
+    )).scalar_one_or_none()
+    return project_rule
+
+
+async def get_rule(rule_id: int, user_id: int) -> Optional[Rule]:
+    async with async_session() as session:
+        return await _fetch_owned_rule(session, rule_id, user_id)
 
 
 async def update_rule(rule_id: int, user_id: int, **fields) -> Optional[Rule]:
     async with async_session() as session:
-        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 = result.scalar_one_or_none()
+        rule = await _fetch_owned_rule(session, rule_id, user_id)
         if rule is None:
             return None
         allowed = {"title", "statement", "why", "how_to_apply", "order_index"}
@@ -347,16 +479,7 @@ async def update_rule(rule_id: int, user_id: int, **fields) -> Optional[Rule]:
 
 async def delete_rule(rule_id: int, user_id: int) -> None:
     async with async_session() as session:
-        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 = result.scalar_one_or_none()
+        rule = await _fetch_owned_rule(session, rule_id, user_id)
         if rule is None:
             return
         await session.delete(rule)
@@ -404,19 +527,115 @@ async def unsubscribe_project(
         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 fabledassistant.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 fabledassistant.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 fabledassistant.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 fabledassistant.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()
+
+
 async def get_applicable_rules(
     project_id: int, user_id: int, limit: int = 50,
 ) -> dict:
-    """Return rules applicable to a project via its subscriptions.
+    """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.
 
     Shape:
         {
-          "rules": [{id, title, statement, topic_title, rulebook_title}, ...],
+          "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}, ...],
           "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.
     """
-    from fabledassistant.models.rulebook import project_rulebook_subscriptions
+    from fabledassistant.models.rulebook import (
+        project_rulebook_subscriptions,
+        project_rule_suppressions,
+        project_topic_suppressions,
+    )
 
     async with async_session() as session:
         # Subscribed rulebooks for the project (ownership-scoped).
@@ -438,11 +657,64 @@ async def get_applicable_rules(
             {"id": rb_id, "title": rb_title} for rb_id, rb_title in sub_rows
         ]
 
-        # Applicable rules (limit + 1 so we can detect truncation).
+        # 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.
         rules_q = (
             select(
                 Rule.id, Rule.title, Rule.statement,
+                RulebookTopic.id.label("topic_id"),
                 RulebookTopic.title.label("topic_title"),
+                Rulebook.id.label("rulebook_id"),
                 Rulebook.title.label("rulebook_title"),
             )
             .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
@@ -463,18 +735,45 @@ async def get_applicable_rules(
             )
             .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))
         rule_rows = (await session.execute(rules_q)).all()
         truncated = len(rule_rows) > limit
         rules = [
             {
                 "id": rid, "title": rtitle, "statement": stmt,
-                "topic_title": tt, "rulebook_title": rbt,
+                "topic_id": ti, "topic_title": tt,
+                "rulebook_id": rbi, "rulebook_title": rbt,
             }
-            for rid, rtitle, stmt, tt, rbt in rule_rows[:limit]
+            for rid, rtitle, stmt, ti, tt, rbi, rbt in rule_rows[:limit]
+        ]
+
+        # Project-scoped rules — verifies ownership via Project.user_id.
+        from fabledassistant.models.project import Project
+        proj_rules_q = (
+            select(Rule.id, Rule.title, Rule.statement)
+            .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_rule_rows = (await session.execute(proj_rules_q)).all()
+        project_rules = [
+            {"id": rid, "title": rtitle, "statement": stmt}
+            for rid, rtitle, stmt in proj_rule_rows
         ]
 
     return {
         "rules": rules,
+        "project_rules": project_rules,
+        "suppressed_rules": suppressed_rules,
+        "suppressed_topics": suppressed_topics,
         "truncated": truncated,
         "subscribed_rulebooks": subscribed_rulebooks,
     }

src/fabledassistant/services/trash.py

@@ -53,6 +53,25 @@ async def _cascade(session, user_id: int, etype: str, eid: int, batch: str, now)
     if etype == "project":
         await _set(session, Note, [Note.user_id == user_id, Note.project_id == eid], batch, now)
         await _set(session, Milestone, [Milestone.user_id == user_id, Milestone.project_id == eid], batch, now)
+        # Project-scoped rules cascade with the project they're attached to.
+        await _set(session, Rule, [Rule.project_id == eid], batch, now)
+        # Suppressions are pure associations (no deleted_at) — hard-delete
+        # them here so restoring the project doesn't bring stale mutes back.
+        # FK CASCADE would handle a full DELETE on the project row, but the
+        # soft-delete path keeps the project row alive; this guarantees the
+        # rows are gone whether or not the project ever gets purged.
+        from sqlalchemy import delete as _sql_delete
+        from fabledassistant.models.rulebook import (
+            project_rule_suppressions, project_topic_suppressions,
+        )
+        await session.execute(
+            _sql_delete(project_rule_suppressions)
+            .where(project_rule_suppressions.c.project_id == eid)
+        )
+        await session.execute(
+            _sql_delete(project_topic_suppressions)
+            .where(project_topic_suppressions.c.project_id == eid)
+        )
         await _set(session, Project, [Project.user_id == user_id, Project.id == eid], batch, now)
     elif etype == "milestone":
         await _set(session, Note, [Note.user_id == user_id, Note.milestone_id == eid], batch, now)

tests/test_mcp_tool_projects.py

@@ -6,7 +6,7 @@ import pytest
 from fabledassistant.mcp._context import _user_id_ctx
 from fabledassistant.mcp.tools.projects import (
     list_projects, get_project, create_project,
-    update_project,
+    update_project, enter_project,
 )
 
 
@@ -130,3 +130,75 @@ async def test_update_project_raises_when_not_found():
     ):
         with pytest.raises(ValueError, match="project 999 not found"):
             await update_project(project_id=999, title="x")
+
+
+@pytest.mark.asyncio
+async def test_enter_project_composes_full_context():
+    """enter_project pulls project + rules + milestone summary + open tasks +
+    recent notes in one composed call."""
+    p = _fake_project(id=5, title="P")
+    applicable_payload = {
+        "rules": [{"id": 1, "title": "r1", "statement": "s",
+                   "topic_title": "t", "rulebook_title": "rb"}],
+        "project_rules": [{"id": 99, "title": "pr1", "statement": "ps"}],
+        "truncated": False,
+        "subscribed_rulebooks": [{"id": 2, "title": "rb"}],
+    }
+    milestone_summary = [{"id": 10, "title": "MS", "task_count": 3}]
+
+    task1 = MagicMock()
+    task1.id = 100; task1.title = "T1"; task1.status = "in_progress"
+    task1.priority = "high"; task1.task_kind = "work"; task1.milestone_id = 10
+    note1 = MagicMock()
+    note1.id = 200; note1.title = "N1"
+    note1.updated_at = None  # avoids datetime mocking
+
+    with patch(
+        "fabledassistant.mcp.tools.projects.projects_svc.get_project",
+        AsyncMock(return_value=p),
+    ), patch(
+        "fabledassistant.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
+        AsyncMock(return_value=applicable_payload),
+    ), patch(
+        "fabledassistant.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
+        AsyncMock(return_value=milestone_summary),
+    ), patch(
+        "fabledassistant.mcp.tools.projects.notes_svc.list_notes",
+        AsyncMock(side_effect=[([task1], 1), ([note1], 1)]),
+    ):
+        out = await enter_project(project_id=5)
+
+    assert out["project"]["id"] == 5
+    assert out["milestone_summary"] == milestone_summary
+    assert out["applicable_rules"][0]["title"] == "r1"
+    assert out["project_rules"][0]["id"] == 99
+    assert out["subscribed_rulebooks"] == [{"id": 2, "title": "rb"}]
+    assert out["open_tasks"][0]["id"] == 100
+    assert out["open_tasks"][0]["status"] == "in_progress"
+    assert out["recent_notes"][0]["id"] == 200
+
+
+@pytest.mark.asyncio
+async def test_enter_project_raises_when_project_not_found():
+    with patch(
+        "fabledassistant.mcp.tools.projects.projects_svc.get_project",
+        AsyncMock(return_value=None),
+    ):
+        with pytest.raises(ValueError, match="project 999 not found"):
+            await enter_project(project_id=999)
+
+
+def test_enter_project_registered_in_register():
+    """register(mcp) registers enter_project alongside the existing tools."""
+    from fabledassistant.mcp.tools.projects import register
+    registered: list[str] = []
+
+    class FakeMCP:
+        def tool(self, name=None):
+            def decorator(fn):
+                registered.append(name)
+                return fn
+            return decorator
+
+    register(FakeMCP())
+    assert "enter_project" in registered

tests/test_mcp_tool_rulebooks.py

@@ -179,8 +179,156 @@ def test_register_attaches_all_sixteen_tools():
             return decorator
 
     register(FakeMCP())
-    assert len(registered) == 16
+    assert len(registered) == 22
     # spot-check a few names
     assert "list_rulebooks" in registered
     assert "create_rule" in registered
     assert "subscribe_project_to_rulebook" in registered
+    assert "list_always_on_rules" in registered
+    assert "create_project_rule" in registered
+    assert "suppress_rule_for_project" in registered
+    assert "unsuppress_rule_for_project" in registered
+    assert "suppress_topic_for_project" in registered
+    assert "unsuppress_topic_for_project" in registered
+
+
+@pytest.mark.asyncio
+async def test_list_always_on_rules_returns_empty_when_no_always_on_rulebooks():
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.list_always_on_rules",
+        AsyncMock(return_value=[]),
+    ):
+        from fabledassistant.mcp.tools.rulebooks import list_always_on_rules
+        out = await list_always_on_rules()
+    assert out == {"rules": [], "total": 0}
+
+
+@pytest.mark.asyncio
+async def test_list_always_on_rules_projects_each_rule():
+    rules = [_fake_rule(id=100), _fake_rule(id=101)]
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.list_always_on_rules",
+        AsyncMock(return_value=rules),
+    ):
+        from fabledassistant.mcp.tools.rulebooks import list_always_on_rules
+        out = await list_always_on_rules()
+    assert out["total"] == 2
+    assert {r["id"] for r in out["rules"]} == {100, 101}
+    assert all("topic_id" in r for r in out["rules"])
+
+
+@pytest.mark.asyncio
+async def test_update_rulebook_forwards_always_on_when_set():
+    rb = _fake_rulebook(id=1, title="t")
+    mock = AsyncMock(return_value=rb)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.update_rulebook", mock):
+        from fabledassistant.mcp.tools.rulebooks import update_rulebook
+        await update_rulebook(rulebook_id=1, always_on=True)
+    kwargs = mock.call_args.kwargs
+    assert kwargs.get("always_on") is True
+    assert "title" not in kwargs
+    assert "description" not in kwargs
+
+
+@pytest.mark.asyncio
+async def test_update_rulebook_omits_always_on_when_none():
+    rb = _fake_rulebook(id=1, title="t")
+    mock = AsyncMock(return_value=rb)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.update_rulebook", mock):
+        from fabledassistant.mcp.tools.rulebooks import update_rulebook
+        await update_rulebook(rulebook_id=1, title="new title")
+    kwargs = mock.call_args.kwargs
+    assert "always_on" not in kwargs
+    assert kwargs["title"] == "new title"
+
+
+@pytest.mark.asyncio
+async def test_create_project_rule_passes_required_fields():
+    rule = _fake_rule()
+    mock = AsyncMock(return_value=rule)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
+        from fabledassistant.mcp.tools.rulebooks import create_project_rule
+        await create_project_rule(
+            project_id=42,
+            statement="Always run migrations through alembic, not raw SQL.",
+            why="audit trail",
+        )
+    kwargs = mock.call_args.kwargs
+    assert kwargs["user_id"] == 7
+    assert kwargs["project_id"] == 42
+    assert kwargs["statement"].startswith("Always run migrations")
+    assert kwargs["why"] == "audit trail"
+
+
+@pytest.mark.asyncio
+async def test_create_project_rule_derives_title_from_statement():
+    rule = _fake_rule()
+    mock = AsyncMock(return_value=rule)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
+        from fabledassistant.mcp.tools.rulebooks import create_project_rule
+        await create_project_rule(
+            project_id=42,
+            statement="Avoid auto-generated docstrings. Reviewers find them noise.",
+        )
+    kwargs = mock.call_args.kwargs
+    # Title should be derived from the first sentence, capped at 50 chars
+    assert kwargs["title"] == "Avoid auto-generated docstrings"
+
+
+@pytest.mark.asyncio
+async def test_create_project_rule_uses_explicit_title_when_given():
+    rule = _fake_rule()
+    mock = AsyncMock(return_value=rule)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
+        from fabledassistant.mcp.tools.rulebooks import create_project_rule
+        await create_project_rule(
+            project_id=42,
+            statement="anything",
+            title="no auto-docstrings",
+        )
+    kwargs = mock.call_args.kwargs
+    assert kwargs["title"] == "no auto-docstrings"
+
+
+@pytest.mark.asyncio
+async def test_suppress_rule_for_project_passes_through():
+    mock = AsyncMock(return_value=None)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.suppress_rule_for_project", mock):
+        from fabledassistant.mcp.tools.rulebooks import suppress_rule_for_project
+        out = await suppress_rule_for_project(project_id=3, rule_id=17)
+    kwargs = mock.call_args.kwargs
+    assert kwargs == {"project_id": 3, "rule_id": 17, "user_id": 7}
+    assert out == {"project_id": 3, "rule_id": 17, "suppressed": True}
+
+
+@pytest.mark.asyncio
+async def test_unsuppress_rule_for_project_passes_through():
+    mock = AsyncMock(return_value=None)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.unsuppress_rule_for_project", mock):
+        from fabledassistant.mcp.tools.rulebooks import unsuppress_rule_for_project
+        out = await unsuppress_rule_for_project(project_id=3, rule_id=17)
+    kwargs = mock.call_args.kwargs
+    assert kwargs == {"project_id": 3, "rule_id": 17, "user_id": 7}
+    assert out == {"project_id": 3, "rule_id": 17, "suppressed": False}
+
+
+@pytest.mark.asyncio
+async def test_suppress_topic_for_project_passes_through():
+    mock = AsyncMock(return_value=None)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.suppress_topic_for_project", mock):
+        from fabledassistant.mcp.tools.rulebooks import suppress_topic_for_project
+        out = await suppress_topic_for_project(project_id=3, topic_id=22)
+    kwargs = mock.call_args.kwargs
+    assert kwargs == {"project_id": 3, "topic_id": 22, "user_id": 7}
+    assert out == {"project_id": 3, "topic_id": 22, "suppressed": True}
+
+
+@pytest.mark.asyncio
+async def test_unsuppress_topic_for_project_passes_through():
+    mock = AsyncMock(return_value=None)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.unsuppress_topic_for_project", mock):
+        from fabledassistant.mcp.tools.rulebooks import unsuppress_topic_for_project
+        out = await unsuppress_topic_for_project(project_id=3, topic_id=22)
+    kwargs = mock.call_args.kwargs
+    assert kwargs == {"project_id": 3, "topic_id": 22, "user_id": 7}
+    assert out == {"project_id": 3, "topic_id": 22, "suppressed": False}

tests/test_routes_rulebooks.py

@@ -44,13 +44,74 @@ def test_service_signatures_require_user_id():
         "create_rulebook", "list_rulebooks", "get_rulebook",
         "update_rulebook", "delete_rulebook", "find_rulebook_by_title",
         "create_topic", "list_topics", "get_topic", "update_topic", "delete_topic",
-        "create_rule", "list_rules", "get_rule", "update_rule", "delete_rule",
+        "create_rule", "create_project_rule",
+        "list_rules", "list_always_on_rules",
+        "get_rule", "update_rule", "delete_rule",
         "subscribe_project", "unsubscribe_project", "get_applicable_rules",
+        "suppress_rule_for_project", "unsuppress_rule_for_project",
+        "suppress_topic_for_project", "unsuppress_topic_for_project",
     ):
         sig = inspect.signature(getattr(svc, fn_name))
         assert "user_id" in sig.parameters, f"{fn_name} missing user_id param"
 
 
+def test_rule_model_carries_project_id_and_topic_id_nullable():
+    """Migration 0059 made topic_id nullable and added project_id."""
+    from fabledassistant.models.rulebook import Rule
+    assert "project_id" in Rule.__table__.columns
+    assert Rule.__table__.columns["topic_id"].nullable is True
+    assert Rule.__table__.columns["project_id"].nullable is True
+
+
+def test_create_project_rule_route_exists():
+    """POST /api/projects/<id>/rules — the frontend fast-path endpoint."""
+    from fabledassistant.routes import rulebooks as rb_routes
+    assert callable(getattr(rb_routes, "create_project_rule"))
+
+
+def test_suppression_route_handlers_exist():
+    """The 4 suppression endpoint handlers are registered as Python callables."""
+    from fabledassistant.routes import rulebooks as rb_routes
+    for name in (
+        "suppress_project_rule", "unsuppress_project_rule",
+        "suppress_project_topic", "unsuppress_project_topic",
+    ):
+        assert callable(getattr(rb_routes, name)), f"missing route handler: {name}"
+
+
+def test_suppression_association_tables_declared():
+    """Migration 0060 created two new association tables; the models module
+    must declare matching Table() objects so the rest of the service layer
+    can reference them via .c.<column>."""
+    from fabledassistant.models import rulebook as rb_models
+    for tbl_name in ("project_rule_suppressions", "project_topic_suppressions"):
+        tbl = getattr(rb_models, tbl_name, None)
+        assert tbl is not None, f"models.rulebook missing {tbl_name}"
+        cols = {c.name for c in tbl.columns}
+        assert "project_id" in cols
+        assert "rule_id" in cols or "topic_id" in cols
+
+
+def test_rulebook_model_carries_always_on():
+    """Migration 0058 added rulebooks.always_on — verify the model declares it."""
+    from fabledassistant.models.rulebook import Rulebook
+    assert "always_on" in Rulebook.__table__.columns
+    col = Rulebook.__table__.columns["always_on"]
+    assert col.nullable is False
+
+
+def test_update_rulebook_route_accepts_always_on():
+    """PATCH /api/rulebooks/<id> must pass always_on through to the service.
+
+    The handler filters body keys against a whitelist; that whitelist needs to
+    include always_on or toggling from the UI silently drops the field.
+    """
+    import inspect as _inspect
+    from fabledassistant.routes import rulebooks as rb_routes
+    src = _inspect.getsource(rb_routes.update_rulebook)
+    assert "always_on" in src, "update_rulebook handler missing always_on in field whitelist"
+
+
 def test_rule_and_subscription_handlers_callable():
     from fabledassistant.routes import rulebooks as rb_routes
     for name in (

tests/test_services_rulebooks.py

@@ -241,18 +241,31 @@ async def test_subscribe_project_requires_owned_rulebook():
             )
 
 
+def _empty():
+    """A MagicMock result whose .all() returns [] (or .scalars().all() returns [])."""
+    r = MagicMock()
+    r.all.return_value = []
+    r.scalars.return_value.all.return_value = []
+    return r
+
+
 @pytest.mark.asyncio
 async def test_get_applicable_rules_returns_shape():
-    """get_applicable_rules returns {rules, truncated, subscribed_rulebooks}."""
+    """get_applicable_rules returns the full projection — including the
+    new suppression fields and rulebook/topic IDs on each rule."""
     mock_session = _make_mock_session()
     sub_result = MagicMock()
     sub_result.all.return_value = [(1, "FabledSword family")]
     rules_result = MagicMock()
     rules_result.all.return_value = [
-        (i, f"Rule {i}", f"Statement {i}", "git-workflow", "FabledSword family")
+        # (rule_id, title, statement, topic_id, topic_title, rulebook_id, rulebook_title)
+        (i, f"Rule {i}", f"Statement {i}", 2, "git-workflow", 1, "FabledSword family")
         for i in range(50)
     ]
-    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result])
+    # Execute order: sub_q, suppressed_rules_q, suppressed_topics_q, rules_q, proj_rules_q
+    mock_session.execute = AsyncMock(side_effect=[
+        sub_result, _empty(), _empty(), rules_result, _empty(),
+    ])
 
     with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
         mock_cls.return_value = mock_session
@@ -260,11 +273,19 @@ async def test_get_applicable_rules_returns_shape():
         result = await get_applicable_rules(project_id=3, user_id=7, limit=50)
 
     assert "rules" in result
+    assert "project_rules" in result
+    assert "suppressed_rules" in result
+    assert "suppressed_topics" in result
     assert "truncated" in result
     assert "subscribed_rulebooks" in result
     assert result["subscribed_rulebooks"] == [{"id": 1, "title": "FabledSword family"}]
     assert len(result["rules"]) == 50
-    assert result["truncated"] is False  # exactly 50, not over
+    assert result["rules"][0]["topic_id"] == 2
+    assert result["rules"][0]["rulebook_id"] == 1
+    assert result["project_rules"] == []
+    assert result["suppressed_rules"] == []
+    assert result["suppressed_topics"] == []
+    assert result["truncated"] is False
 
 
 @pytest.mark.asyncio
@@ -274,11 +295,12 @@ async def test_get_applicable_rules_truncates_when_over_limit():
     sub_result = MagicMock()
     sub_result.all.return_value = []
     rules_result = MagicMock()
-    # 51 rows means truncation detected
     rules_result.all.return_value = [
-        (i, f"r{i}", "stmt", "topic", "rb") for i in range(51)
+        (i, f"r{i}", "stmt", 2, "topic", 1, "rb") for i in range(51)
     ]
-    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result])
+    mock_session.execute = AsyncMock(side_effect=[
+        sub_result, _empty(), _empty(), rules_result, _empty(),
+    ])
 
     with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
         mock_cls.return_value = mock_session
@@ -287,3 +309,57 @@ async def test_get_applicable_rules_truncates_when_over_limit():
 
     assert result["truncated"] is True
     assert len(result["rules"]) == 50
+
+
+@pytest.mark.asyncio
+async def test_get_applicable_rules_includes_project_scoped_rules():
+    """Project-scoped rules surface in the project_rules field."""
+    mock_session = _make_mock_session()
+    proj_rules_result = MagicMock()
+    proj_rules_result.all.return_value = [
+        (100, "Use alembic", "Always run migrations via alembic, never raw SQL."),
+        (101, "PR-bound", "Land schema changes in their own PR."),
+    ]
+    mock_session.execute = AsyncMock(side_effect=[
+        _empty(), _empty(), _empty(), _empty(), proj_rules_result,
+    ])
+
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import get_applicable_rules
+        result = await get_applicable_rules(project_id=3, user_id=7)
+
+    assert len(result["project_rules"]) == 2
+    assert result["project_rules"][0]["title"] == "Use alembic"
+    assert result["project_rules"][1]["id"] == 101
+
+
+@pytest.mark.asyncio
+async def test_get_applicable_rules_surfaces_suppressed_with_context():
+    """Suppressed rules and topics come back with full title + rulebook context
+    so the UI can render them without an extra round-trip."""
+    mock_session = _make_mock_session()
+    suppressed_rules_result = MagicMock()
+    suppressed_rules_result.all.return_value = [
+        # (rule_id, title, topic_id, topic_title, rulebook_id, rulebook_title)
+        (17, "Old rule", 5, "old-topic", 1, "FabledSword family"),
+    ]
+    suppressed_topics_result = MagicMock()
+    suppressed_topics_result.all.return_value = [
+        # (topic_id, topic_title, rulebook_id, rulebook_title)
+        (22, "design-system", 1, "FabledSword family"),
+    ]
+    mock_session.execute = AsyncMock(side_effect=[
+        _empty(), suppressed_rules_result, suppressed_topics_result, _empty(), _empty(),
+    ])
+
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import get_applicable_rules
+        result = await get_applicable_rules(project_id=3, user_id=7)
+
+    assert len(result["suppressed_rules"]) == 1
+    assert result["suppressed_rules"][0]["id"] == 17
+    assert result["suppressed_rules"][0]["rulebook_title"] == "FabledSword family"
+    assert len(result["suppressed_topics"]) == 1
+    assert result["suppressed_topics"][0]["title"] == "design-system"

tests/test_services_trash.py

@@ -48,18 +48,24 @@ async def test_delete_returns_none_when_not_found():
 
 
 @pytest.mark.asyncio
-async def test_delete_project_cascades_three_tables():
+async def test_delete_project_cascades_to_notes_milestones_project_rules_and_suppressions():
     session = _make_mock_session()
-    # exists-check + 3 cascade updates (notes, milestones, project)
+    # exists-check + 6 cascade ops:
+    #   notes (soft) → milestones (soft) → project-scoped rules (soft) →
+    #   project_rule_suppressions (hard DELETE) → project_topic_suppressions (hard DELETE) →
+    #   project (soft)
     session.execute = AsyncMock(side_effect=[
-        _exists_result(True), MagicMock(), MagicMock(), MagicMock(),
+        _exists_result(True),
+        MagicMock(), MagicMock(), MagicMock(),
+        MagicMock(), MagicMock(),
+        MagicMock(),
     ])
     with patch("fabledassistant.services.trash.async_session") as cls:
         cls.return_value = session
         from fabledassistant.services.trash import delete
         batch = await delete(user_id=1, entity_type="project", entity_id=3)
     assert isinstance(batch, str)
-    assert session.execute.await_count == 4
+    assert session.execute.await_count == 7
 
 
 @pytest.mark.asyncio