diff --git a/src/fabledassistant/services/rulebooks.py b/src/fabledassistant/services/rulebooks.py
index 76f3293..72ae6bc 100644
--- a/src/fabledassistant/services/rulebooks.py
+++ b/src/fabledassistant/services/rulebooks.py
@@ -211,3 +211,135 @@ async def delete_topic(topic_id: int, user_id: int) -> None:
             return
         await session.delete(topic)
         await session.commit()
+
+
+# ── Rule CRUD ──────────────────────────────────────────────────────────
+
+from fabledassistant.models.rulebook import Rule
+
+
+async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
+    """Raise ValueError if topic doesn't exist or isn't in user's rulebook."""
+    result = await session.execute(
+        select(RulebookTopic)
+        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+        .where(
+            RulebookTopic.id == topic_id,
+            Rulebook.owner_user_id == user_id,
+        )
+    )
+    if result.scalar_one_or_none() is None:
+        raise ValueError(f"topic {topic_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,
+) -> Rule:
+    async with async_session() as session:
+        await _assert_topic_owned(session, topic_id, user_id)
+        rule = Rule(
+            topic_id=topic_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,
+    topic_id: int | None = None,
+    project_id: int | None = None,
+) -> list[Rule]:
+    """List rules filtered by any of the three IDs. All filters are ownership-scoped.
+
+    project_id resolves rules through project_rulebook_subscriptions.
+    """
+    from fabledassistant.models.rulebook import project_rulebook_subscriptions
+
+    async with async_session() as session:
+        stmt = (
+            select(Rule)
+            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
+            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+            .where(Rulebook.owner_user_id == user_id)
+        )
+        if topic_id:
+            stmt = stmt.where(Rule.topic_id == topic_id)
+        if rulebook_id:
+            stmt = stmt.where(RulebookTopic.rulebook_id == rulebook_id)
+        if project_id:
+            stmt = (
+                stmt.join(
+                    project_rulebook_subscriptions,
+                    project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
+                )
+                .where(project_rulebook_subscriptions.c.project_id == project_id)
+            )
+        stmt = stmt.order_by(
+            Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
+        )
+        result = await session.execute(stmt)
+        return list(result.scalars().all())
+
+
+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,
+            )
+        )
+        return result.scalar_one_or_none()
+
+
+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()
+        if rule is None:
+            return None
+        allowed = {"title", "statement", "why", "how_to_apply", "order_index"}
+        for key, value in fields.items():
+            if key in allowed and value is not None:
+                setattr(rule, key, value)
+        await session.commit()
+        await session.refresh(rule)
+        return 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()
+        if rule is None:
+            return
+        await session.delete(rule)
+        await session.commit()
diff --git a/tests/test_services_rulebooks.py b/tests/test_services_rulebooks.py
index cdcb9e3..21125e3 100644
--- a/tests/test_services_rulebooks.py
+++ b/tests/test_services_rulebooks.py
@@ -156,3 +156,68 @@ async def test_list_topics_returns_topics_for_owned_rulebook():
         results = await list_topics(rulebook_id=1, user_id=7)
     assert len(results) == 1
     assert results[0].title == "git-workflow"
+
+
+# ── Rule CRUD ───────────────────────────────────────────────────────────
+
+def _fake_rule(id=1, topic_id=10, title="dev is home",
+               statement="Work directly on dev", why="", how_to_apply=""):
+    r = MagicMock()
+    r.id = id
+    r.topic_id = topic_id
+    r.title = title
+    r.statement = statement
+    r.why = why
+    r.how_to_apply = how_to_apply
+    r.order_index = 0
+    r.created_at = datetime.now(timezone.utc)
+    r.updated_at = datetime.now(timezone.utc)
+    r.to_dict.return_value = {
+        "id": id, "topic_id": topic_id, "title": title,
+        "statement": statement, "why": why or "",
+        "how_to_apply": how_to_apply or "", "order_index": 0,
+    }
+    return r
+
+
+@pytest.mark.asyncio
+async def test_create_rule_requires_owned_topic():
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = None  # topic not found
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import create_rule
+        with pytest.raises(ValueError, match="topic .* not found"):
+            await create_rule(
+                topic_id=999, user_id=7, title="x", statement="y",
+            )
+
+
+@pytest.mark.asyncio
+async def test_list_rules_filters_by_topic_id():
+    """list_rules(topic_id=X) returns rules in that topic, ownership-scoped."""
+    rule = _fake_rule(id=1, topic_id=10)
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.all.return_value = [rule]
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import list_rules
+        results = await list_rules(user_id=7, topic_id=10)
+    assert len(results) == 1
+
+
+@pytest.mark.asyncio
+async def test_get_rule_returns_none_when_not_owner():
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = None
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import get_rule
+        result = await get_rule(rule_id=1, user_id=99)
+    assert result is None