feat(rules): project-scoped rules (S3)

Rules can now belong to either a rulebook topic OR a single project,
enforced by a CHECK constraint (exactly-one of topic_id/project_id).
Adds the create_project_rule MCP tool + REST endpoint, surfaces
project-scoped rules in get_project/get_task/start_planning under a
new project_rules field, and adds a project Rules tab section with an
inline create form so the operator can author project rules from the
UI without rulebook ceremony.

- migration 0059: rules.project_id (FK projects ON DELETE CASCADE),
  topic_id now nullable, CHECK ck_rule_topic_xor_project, index on
  project_id
- model: Rule gains project_id; to_dict exposes it
- service: create_project_rule with project-ownership guard; list_rules
  with project_id filter UNIONs subscription-derived + project-scoped;
  get_applicable_rules adds a project_rules field; get_rule / update_rule
  / delete_rule fetch via a shared _fetch_owned_rule that handles both
  rulebook and project ownership paths
- trash: project delete cascades to project-scoped rules
- MCP: create_project_rule tool registered; _INSTRUCTIONS mentions both
  create_rule and create_project_rule paths
- REST: POST /api/projects/<id>/rules (statement required, title derived
  if omitted)
- frontend: Rule type gains nullable topic_id + project_id; createProjectRule
  client; ProjectRulesTab.vue gains a "Project rules" section with inline
  create form and per-rule expand/delete
- tests: register count → 18; create_project_rule unit tests (required
  fields, title derivation, explicit-title pass-through); applicable_rules
  shape tests now include project_rules; trash cascade test updated to
  expect 5 executions

S1+S2 (always_on flag + Scribe-first prompt) shipped in 658348f.
S4 (enter_project handshake) follows.

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

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")

frontend/src/api/rulebooks.ts

@@ -22,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;
@@ -36,7 +37,7 @@ export interface RuleHeader {
   id: number;
   title: string;
   statement: string;
-  topic_id: number;
+  topic_id: number | null;
 }
 
 export interface ApplicableRules {
@@ -47,6 +48,11 @@ export interface ApplicableRules {
     topic_title: string;
     rulebook_title: string;
   }[];
+  project_rules: {
+    id: number;
+    title: string;
+    statement: string;
+  }[];
   truncated: boolean;
   subscribed_rulebooks: { id: number; title: string }[];
 }
@@ -134,3 +140,10 @@ 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);
+}

frontend/src/components/rules/ProjectRulesTab.vue

@@ -3,7 +3,7 @@ import { ref, onMounted, watch } from "vue";
 import { useRouter } from "vue-router";
 import {
   getProjectApplicableRules, subscribeProject, unsubscribeProject,
-  listRulebooks, getRule,
+  listRulebooks, getRule, createProjectRule, deleteRule,
 } from "@/api/rulebooks";
 import type { ApplicableRules, Rulebook } from "@/api/rulebooks";
 
@@ -16,6 +16,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);
 }
@@ -73,6 +76,26 @@ 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();
+}
+
 onMounted(async () => {
   await load();
   await loadAllRulebooks();
@@ -111,6 +134,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">
@@ -208,4 +294,22 @@ 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;
+}
 </style>

src/fabledassistant/mcp/server.py

@@ -47,13 +47,14 @@ 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. 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 an existing
-rulebook+topic or a new one.
+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.
 
 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

@@ -55,6 +55,7 @@ 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", [])
     return data
 
 

src/fabledassistant/mcp/tools/rulebooks.py

@@ -247,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.
@@ -256,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(
@@ -266,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,
@@ -336,7 +368,8 @@ 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, list_always_on_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,
     ):
         mcp.tool(name=fn.__name__)(fn)

src/fabledassistant/mcp/tools/tasks.py

@@ -82,6 +82,7 @@ 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", [])
     return data
 
 

src/fabledassistant/models/rulebook.py

@@ -81,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)
@@ -102,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 "",

src/fabledassistant/routes/rulebooks.py

@@ -239,3 +239,27 @@ 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>/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,7 @@ 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", []),
         "project_goal": getattr(project, "goal", "") or "",
         "open_task_count": open_count,
     }

src/fabledassistant/services/rulebooks.py

@@ -242,6 +242,20 @@ 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 create_rule(
     topic_id: int, user_id: int, title: str, statement: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
@@ -262,6 +276,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 +310,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,7 +347,27 @@ 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 list_always_on_rules(user_id: int, limit: int = 100) -> list[Rule]:
@@ -332,35 +397,51 @@ async def list_always_on_rules(user_id: int, limit: int = 100) -> list[Rule]:
         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:
-        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),
-            )
-        )
-        return result.scalar_one_or_none()
+        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"}
@@ -374,16 +455,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)
@@ -434,14 +506,19 @@ async def unsubscribe_project(
 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).
 
     Shape:
         {
           "rules": [{id, title, statement, topic_title, rulebook_title}, ...],
+          "project_rules": [{id, title, statement}, ...],
           "truncated": bool,
           "subscribed_rulebooks": [{id, title}, ...]
         }
+
+    `rules` is the subscription-derived set (legacy shape preserved).
+    `project_rules` is the project-scoped set; empty list when none exist.
     """
     from fabledassistant.models.rulebook import project_rulebook_subscriptions
 
@@ -500,8 +577,28 @@ async def get_applicable_rules(
             for rid, rtitle, stmt, tt, 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,
         "truncated": truncated,
         "subscribed_rulebooks": subscribed_rulebooks,
     }

src/fabledassistant/services/trash.py

@@ -53,6 +53,8 @@ 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)
         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_rulebooks.py

@@ -179,12 +179,13 @@ def test_register_attaches_all_sixteen_tools():
             return decorator
 
     register(FakeMCP())
-    assert len(registered) == 17
+    assert len(registered) == 18
     # 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
 
 
 @pytest.mark.asyncio
@@ -235,3 +236,51 @@ async def test_update_rulebook_omits_always_on_when_none():
     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"

tests/test_routes_rulebooks.py

@@ -44,7 +44,8 @@ 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", "list_always_on_rules",
+        "create_rule", "create_project_rule",
+        "list_rules", "list_always_on_rules",
         "get_rule", "update_rule", "delete_rule",
         "subscribe_project", "unsubscribe_project", "get_applicable_rules",
     ):
@@ -52,6 +53,20 @@ def test_service_signatures_require_user_id():
         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_rulebook_model_carries_always_on():
     """Migration 0058 added rulebooks.always_on — verify the model declares it."""
     from fabledassistant.models.rulebook import Rulebook

tests/test_services_rulebooks.py

@@ -243,7 +243,7 @@ async def test_subscribe_project_requires_owned_rulebook():
 
 @pytest.mark.asyncio
 async def test_get_applicable_rules_returns_shape():
-    """get_applicable_rules returns {rules, truncated, subscribed_rulebooks}."""
+    """get_applicable_rules returns {rules, project_rules, truncated, subscribed_rulebooks}."""
     mock_session = _make_mock_session()
     sub_result = MagicMock()
     sub_result.all.return_value = [(1, "FabledSword family")]
@@ -252,7 +252,9 @@ async def test_get_applicable_rules_returns_shape():
         (i, f"Rule {i}", f"Statement {i}", "git-workflow", "FabledSword family")
         for i in range(50)
     ]
-    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result])
+    proj_rules_result = MagicMock()
+    proj_rules_result.all.return_value = []
+    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result, proj_rules_result])
 
     with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
         mock_cls.return_value = mock_session
@@ -260,10 +262,12 @@ 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 "truncated" in result
     assert "subscribed_rulebooks" in result
     assert result["subscribed_rulebooks"] == [{"id": 1, "title": "FabledSword family"}]
     assert len(result["rules"]) == 50
+    assert result["project_rules"] == []
     assert result["truncated"] is False  # exactly 50, not over
 
 
@@ -278,7 +282,9 @@ async def test_get_applicable_rules_truncates_when_over_limit():
     rules_result.all.return_value = [
         (i, f"r{i}", "stmt", "topic", "rb") for i in range(51)
     ]
-    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result])
+    proj_rules_result = MagicMock()
+    proj_rules_result.all.return_value = []
+    mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result, proj_rules_result])
 
     with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
         mock_cls.return_value = mock_session
@@ -287,3 +293,28 @@ 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()
+    sub_result = MagicMock()
+    sub_result.all.return_value = []
+    rules_result = MagicMock()
+    rules_result.all.return_value = []
+    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=[sub_result, rules_result, 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

tests/test_services_trash.py

@@ -48,18 +48,18 @@ 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_and_project_rules():
     session = _make_mock_session()
-    # exists-check + 3 cascade updates (notes, milestones, project)
+    # exists-check + 4 cascade updates (notes, milestones, project-scoped rules, project)
     session.execute = AsyncMock(side_effect=[
-        _exists_result(True), MagicMock(), MagicMock(), MagicMock(),
+        _exists_result(True), 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 == 5
 
 
 @pytest.mark.asyncio