diff --git a/src/fabledassistant/services/rulebooks.py b/src/fabledassistant/services/rulebooks.py
index 885e2be..76f3293 100644
--- a/src/fabledassistant/services/rulebooks.py
+++ b/src/fabledassistant/services/rulebooks.py
@@ -108,3 +108,106 @@ async def find_rulebook_by_title(
             )
         )
         return result.scalar_one_or_none()
+
+
+# ── Topic CRUD ──────────────────────────────────────────────────────────
+
+from fabledassistant.models.rulebook import RulebookTopic
+
+
+async def _assert_rulebook_owned(session, rulebook_id: int, user_id: int) -> None:
+    """Raise ValueError if rulebook doesn't exist or isn't owned by user.
+    Centralizes ownership check used by all topic/rule operations.
+    """
+    result = await session.execute(
+        select(Rulebook).where(
+            Rulebook.id == rulebook_id,
+            Rulebook.owner_user_id == user_id,
+        )
+    )
+    if result.scalar_one_or_none() is None:
+        raise ValueError(f"rulebook {rulebook_id} not found")
+
+
+async def create_topic(
+    rulebook_id: int, user_id: int, title: str,
+    description: str = "", order_index: int = 0,
+) -> RulebookTopic:
+    async with async_session() as session:
+        await _assert_rulebook_owned(session, rulebook_id, user_id)
+        topic = RulebookTopic(
+            rulebook_id=rulebook_id,
+            title=title,
+            description=description or None,
+            order_index=order_index,
+        )
+        session.add(topic)
+        await session.commit()
+        await session.refresh(topic)
+        return topic
+
+
+async def list_topics(rulebook_id: int, user_id: int) -> list[RulebookTopic]:
+    async with async_session() as session:
+        await _assert_rulebook_owned(session, rulebook_id, user_id)
+        result = await session.execute(
+            select(RulebookTopic)
+            .where(RulebookTopic.rulebook_id == rulebook_id)
+            .order_by(RulebookTopic.order_index, RulebookTopic.title)
+        )
+        return list(result.scalars().all())
+
+
+async def get_topic(topic_id: int, user_id: int) -> Optional[RulebookTopic]:
+    """Get a topic, scoped via the rulebook owner."""
+    async with async_session() as session:
+        result = await session.execute(
+            select(RulebookTopic)
+            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+            .where(
+                RulebookTopic.id == topic_id,
+                Rulebook.owner_user_id == user_id,
+            )
+        )
+        return result.scalar_one_or_none()
+
+
+async def update_topic(
+    topic_id: int, user_id: int, **fields,
+) -> Optional[RulebookTopic]:
+    async with async_session() as session:
+        result = await session.execute(
+            select(RulebookTopic)
+            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+            .where(
+                RulebookTopic.id == topic_id,
+                Rulebook.owner_user_id == user_id,
+            )
+        )
+        topic = result.scalar_one_or_none()
+        if topic is None:
+            return None
+        allowed = {"title", "description", "order_index"}
+        for key, value in fields.items():
+            if key in allowed and value is not None:
+                setattr(topic, key, value)
+        await session.commit()
+        await session.refresh(topic)
+        return topic
+
+
+async def delete_topic(topic_id: int, user_id: int) -> None:
+    async with async_session() as session:
+        result = await session.execute(
+            select(RulebookTopic)
+            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+            .where(
+                RulebookTopic.id == topic_id,
+                Rulebook.owner_user_id == user_id,
+            )
+        )
+        topic = result.scalar_one_or_none()
+        if topic is None:
+            return
+        await session.delete(topic)
+        await session.commit()
diff --git a/tests/test_services_rulebooks.py b/tests/test_services_rulebooks.py
index 17ba7dd..cdcb9e3 100644
--- a/tests/test_services_rulebooks.py
+++ b/tests/test_services_rulebooks.py
@@ -101,3 +101,58 @@ async def test_delete_rulebook_calls_delete():
         from fabledassistant.services.rulebooks import delete_rulebook
         await delete_rulebook(rulebook_id=1, user_id=7)
     assert mock_session.delete.called
+
+
+# ── Topic CRUD ───────────────────────────────────────────────────────────
+
+def _fake_topic(id=1, rulebook_id=1, title="git-workflow", description="", order_index=0):
+    t = MagicMock()
+    t.id = id
+    t.rulebook_id = rulebook_id
+    t.title = title
+    t.description = description
+    t.order_index = order_index
+    t.created_at = datetime.now(timezone.utc)
+    t.updated_at = datetime.now(timezone.utc)
+    t.to_dict.return_value = {
+        "id": id, "rulebook_id": rulebook_id, "title": title,
+        "description": description or "", "order_index": order_index,
+    }
+    return t
+
+
+@pytest.mark.asyncio
+async def test_create_topic_requires_owned_rulebook():
+    """create_topic raises ValueError if the rulebook isn't owned by user."""
+    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 create_topic
+        with pytest.raises(ValueError, match="not found"):
+            await create_topic(
+                rulebook_id=999, user_id=7, title="git-workflow",
+            )
+
+
+@pytest.mark.asyncio
+async def test_list_topics_returns_topics_for_owned_rulebook():
+    rb = _fake_rulebook(id=1)
+    topic = _fake_topic(id=10, rulebook_id=1, title="git-workflow")
+
+    # Two execute calls: ownership check, then topic select.
+    mock_session = _make_mock_session()
+    rb_result = MagicMock()
+    rb_result.scalar_one_or_none.return_value = rb
+    topic_result = MagicMock()
+    topic_result.scalars.return_value.all.return_value = [topic]
+    mock_session.execute = AsyncMock(side_effect=[rb_result, topic_result])
+
+    with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.rulebooks import list_topics
+        results = await list_topics(rulebook_id=1, user_id=7)
+    assert len(results) == 1
+    assert results[0].title == "git-workflow"