diff --git a/src/fabledassistant/services/rulebooks.py b/src/fabledassistant/services/rulebooks.py
index 72ae6bc..db01acb 100644
--- a/src/fabledassistant/services/rulebooks.py
+++ b/src/fabledassistant/services/rulebooks.py
@@ -343,3 +343,116 @@ async def delete_rule(rule_id: int, user_id: int) -> None:
             return
         await session.delete(rule)
         await session.commit()
+
+
+# ── Subscriptions + get_applicable_rules ───────────────────────────────
+
+from sqlalchemy import insert, delete as sql_delete
+
+
+async def subscribe_project(
+    project_id: int, rulebook_id: int, user_id: int,
+) -> None:
+    """Add a subscription. Idempotent — duplicates raise; we swallow."""
+    from fabledassistant.models.rulebook import project_rulebook_subscriptions
+
+    async with async_session() as session:
+        await _assert_rulebook_owned(session, rulebook_id, user_id)
+        # ON CONFLICT DO NOTHING via try/except to keep dialect-agnostic.
+        try:
+            await session.execute(
+                insert(project_rulebook_subscriptions).values(
+                    project_id=project_id, rulebook_id=rulebook_id,
+                )
+            )
+            await session.commit()
+        except Exception:
+            await session.rollback()  # PK collision = already subscribed; fine.
+
+
+async def unsubscribe_project(
+    project_id: int, rulebook_id: int, user_id: int,
+) -> None:
+    from fabledassistant.models.rulebook import project_rulebook_subscriptions
+
+    async with async_session() as session:
+        await _assert_rulebook_owned(session, rulebook_id, user_id)
+        await session.execute(
+            sql_delete(project_rulebook_subscriptions).where(
+                project_rulebook_subscriptions.c.project_id == project_id,
+                project_rulebook_subscriptions.c.rulebook_id == rulebook_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.
+
+    Shape:
+        {
+          "rules": [{id, title, statement, topic_title, rulebook_title}, ...],
+          "truncated": bool,
+          "subscribed_rulebooks": [{id, title}, ...]
+        }
+    """
+    from fabledassistant.models.rulebook import project_rulebook_subscriptions
+
+    async with async_session() as session:
+        # Subscribed rulebooks for the project (ownership-scoped).
+        sub_q = (
+            select(Rulebook.id, Rulebook.title)
+            .join(
+                project_rulebook_subscriptions,
+                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
+            )
+            .where(
+                project_rulebook_subscriptions.c.project_id == project_id,
+                Rulebook.owner_user_id == user_id,
+            )
+            .order_by(Rulebook.title)
+        )
+        sub_rows = (await session.execute(sub_q)).all()
+        subscribed_rulebooks = [
+            {"id": rb_id, "title": rb_title} for rb_id, rb_title in sub_rows
+        ]
+
+        # Applicable rules (limit + 1 so we can detect truncation).
+        rules_q = (
+            select(
+                Rule.id, Rule.title, Rule.statement,
+                RulebookTopic.title.label("topic_title"),
+                Rulebook.title.label("rulebook_title"),
+            )
+            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
+            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+            .join(
+                project_rulebook_subscriptions,
+                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
+            )
+            .where(
+                project_rulebook_subscriptions.c.project_id == project_id,
+                Rulebook.owner_user_id == user_id,
+            )
+            .order_by(
+                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
+            )
+            .limit(limit + 1)
+        )
+        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,
+            }
+            for rid, rtitle, stmt, tt, rbt in rule_rows[:limit]
+        ]
+
+    return {
+        "rules": rules,
+        "truncated": truncated,
+        "subscribed_rulebooks": subscribed_rulebooks,
+    }
diff --git a/tests/test_services_rulebooks.py b/tests/test_services_rulebooks.py
index 21125e3..05cf2ee 100644
--- a/tests/test_services_rulebooks.py
+++ b/tests/test_services_rulebooks.py
@@ -221,3 +221,69 @@ async def test_get_rule_returns_none_when_not_owner():
         from fabledassistant.services.rulebooks import get_rule
         result = await get_rule(rule_id=1, user_id=99)
     assert result is None
+
+
+# ── Subscriptions + applicable_rules ────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_subscribe_project_requires_owned_rulebook():
+    """subscribe_project raises if user doesn't own the rulebook."""
+    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 subscribe_project
+        with pytest.raises(ValueError, match="not found"):
+            await subscribe_project(
+                project_id=1, rulebook_id=999, user_id=7,
+            )
+
+
+@pytest.mark.asyncio
+async def test_get_applicable_rules_returns_shape():
+    """get_applicable_rules returns {rules, truncated, subscribed_rulebooks}."""
+    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")
+        for i in range(50)
+    ]
+    mock_session.execute = AsyncMock(side_effect=[sub_result, 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, limit=50)
+
+    assert "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["truncated"] is False  # exactly 50, not over
+
+
+@pytest.mark.asyncio
+async def test_get_applicable_rules_truncates_when_over_limit():
+    """When limit+1 rows are returned, truncated=True and only `limit` returned."""
+    mock_session = _make_mock_session()
+    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)
+    ]
+    mock_session.execute = AsyncMock(side_effect=[sub_result, 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, limit=50)
+
+    assert result["truncated"] is True
+    assert len(result["rules"]) == 50