diff --git a/AGENTS.md b/AGENTS.md
index d0b6b78..509c733 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -95,12 +95,74 @@ async def resource(self, interaction, action: str):
 ### Group Cogs
 For complex features with multiple distinct sub-functions, use `commands.GroupCog`.
 
-## 4. Error Handling
+## 4. Internal DM Service
+
+Direct messaging is an internal service, **not** a user-facing cog. Do not add `/dm`-style command surfaces for bulk messaging.
+
+### Location
+Use:
+
+*   `capy_discord.services.dm`
+*   `capy_discord.services.policies`
+
+### Safety Model
+*   DM sends are **deny-all by default** via `policies.DENY_ALL`.
+*   Developers must opt into explicit allowlists with helpers like `policies.allow_users(...)`, `policies.allow_roles(...)`, or `policies.allow_targets(...)`.
+*   The service rejects `@everyone`, rejects targets outside the allowed policy, and enforces `max_recipients`.
+
+### Usage Pattern
+Developers should think in terms of:
+
+1.  The exact user or role to DM.
+2.  The predefined policy that permits that target.
+
+Prefer explicit entrypoints over generic audience bags:
+
+```python
+from capy_discord.services import dm, policies
+
+EVENT_POLICY = policies.allow_roles(EVENT_ROLE_ID, max_recipients=20)
+
+draft = await dm.compose_to_role(
+    guild,
+    EVENT_ROLE_ID,
+    "Reminder: event starts at 7 PM.",
+    policy=EVENT_POLICY,
+)
+result = await dm.send(guild, draft)
+```
+
+For self-test or single-user flows, use `dm.compose_to_user(...)` with `policies.allow_users(...)`.
+
+### Cog Usage
+If you need an operator-facing entrypoint for DM functionality, keep it narrow and task-specific rather than exposing a generic DM surface.
+
+Use a small cog command only for explicit, safe flows such as self-test notifications:
+
+```python
+from capy_discord.services import dm, policies
+
+policy = policies.allow_users(interaction.user.id, max_recipients=1)
+
+draft = await dm.compose_to_user(
+    guild,
+    interaction.user.id,
+    message,
+    policy=policy,
+)
+self.log.info("Notify preview\n%s", dm.render_preview(draft))
+
+result = await dm.send(guild, draft)
+```
+
+This pattern is implemented in `capy_discord/exts/tools/notify.py`. Do not add broad `/dm` or bulk-message commands; use explicit commands tied to a specific feature or operational workflow.
+
+## 5. Error Handling
 We use a global `on_tree_error` handler in `bot.py`.
 *   Exceptions are logged with the specific module name.
 *   Do not wrap every command in `try/except` blocks unless handling specific business logic errors.
 
-## 5. Logging
+## 6. Logging
 All logs follow a standardized format for consistency across the console and log files.
 
 *   **Format**: `[{asctime}] [{levelname:<8}] {name}: {message}`
@@ -113,12 +175,12 @@ self.log = logging.getLogger(__name__)
 self.log.info("Starting feature X")
 ```
 
-## 6. Time and Timezones
+## 7. Time and Timezones
 **Always use `zoneinfo.ZoneInfo`**.
 *   **Storage**: `UTC`.
 *   **Usage**: `datetime.now(ZoneInfo("UTC"))`.
 
-## 7. Development Workflow
+## 8. Development Workflow
 
 ### Linear & Branching
 *   **Issue Tracking**: Every task must have a Linear issue.
@@ -146,7 +208,7 @@ Format: `<type>(<scope>): <subject>`
 2.  **Reviewers**: Must include `Shamik` and `Jason`.
 3.  **Checks**: All CI checks (Lint, Test, Build) must pass.
 
-## 8. Cog Standards
+## 9. Cog Standards
 
 ### Initialization
 All Cogs **MUST** accept the `bot` instance in `__init__`. The use of the global `capy_discord.instance` is **deprecated** and should not be used in new code.
diff --git a/capy_discord/exts/guild/guild.py b/capy_discord/exts/guild/guild.py
index 151de36..1541b56 100644
--- a/capy_discord/exts/guild/guild.py
+++ b/capy_discord/exts/guild/guild.py
@@ -89,7 +89,10 @@ async def _open_channels(self, interaction: discord.Interaction, settings: Guild
 
     async def _open_roles(self, interaction: discord.Interaction, settings: GuildSettings) -> None:
         """Launch the role settings modal pre-filled with current values."""
-        initial = {"admin": settings.admin_role, "member": settings.member_role}
+        initial = {
+            "admin": settings.admin_role,
+            "member": settings.member_role,
+        }
         modal = ModelModal(
             model_cls=RoleSettingsForm, callback=self._handle_roles, title="Role Settings", initial_data=initial
         )
diff --git a/capy_discord/exts/tools/notify.py b/capy_discord/exts/tools/notify.py
new file mode 100644
index 0000000..2f98259
--- /dev/null
+++ b/capy_discord/exts/tools/notify.py
@@ -0,0 +1,47 @@
+"""Safe notification command for testing the internal DM module."""
+
+import logging
+
+import discord
+from discord import app_commands
+from discord.ext import commands
+
+from capy_discord.errors import UserFriendlyError
+from capy_discord.services import dm, policies
+
+
+class Notify(commands.Cog):
+    """Cog for sending a self-targeted test DM."""
+
+    def __init__(self, bot: commands.Bot) -> None:
+        """Initialize the Notify cog."""
+        self.bot = bot
+        self.log = logging.getLogger(__name__)
+        self.log.info("Notify cog initialized")
+
+    @app_commands.command(name="notify", description="Send yourself a test DM")
+    @app_commands.describe(message="Message content to send to your own DMs")
+    @app_commands.guild_only()
+    async def notify(self, interaction: discord.Interaction, message: str) -> None:
+        """Send a test DM to the invoking user."""
+        guild = interaction.guild
+        if guild is None:
+            await interaction.response.send_message("This must be used in a server.", ephemeral=True)
+            return
+
+        policy = policies.allow_users(interaction.user.id, max_recipients=1)
+
+        draft = await dm.compose_to_user(guild, interaction.user.id, message, policy=policy)
+        self.log.info("Notify preview\n%s", dm.render_preview(draft))
+
+        result = await dm.send(guild, draft)
+        if result.sent_count != 1:
+            msg = "Failed to send test DM to the invoking user."
+            raise UserFriendlyError(msg, "I couldn't DM you. Check your Discord privacy settings and try again.")
+
+        await interaction.response.send_message("Sent you a DM.", ephemeral=True)
+
+
+async def setup(bot: commands.Bot) -> None:
+    """Set up the Notify cog."""
+    await bot.add_cog(Notify(bot))
diff --git a/capy_discord/guild/__init__.py b/capy_discord/guild/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/capy_discord/guild/_schemas.py b/capy_discord/guild/_schemas.py
deleted file mode 100644
index e0550c5..0000000
--- a/capy_discord/guild/_schemas.py
+++ /dev/null
@@ -1,49 +0,0 @@
-"""Pydantic models for guild settings used by ModelModal."""
-
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-
-class ChannelSettingsForm(BaseModel):
-    """Form for configuring guild channel destinations."""
-
-    reports: str = Field(default="", title="Reports Channel", description="Channel ID for bug reports")
-    announcements: str = Field(default="", title="Announcements Channel", description="Channel ID for announcements")
-    feedback: str = Field(default="", title="Feedback Channel", description="Channel ID for feedback routing")
-
-
-class RoleSettingsForm(BaseModel):
-    """Form for configuring guild role scopes."""
-
-    admin: str = Field(default="", title="Admin Role", description="Role ID for administrator access")
-    member: str = Field(default="", title="Member Role", description="Role ID for general member access")
-
-
-class AnnouncementChannelForm(BaseModel):
-    """Form for setting the announcement channel."""
-
-    channel: str = Field(default="", title="Announcement Channel", description="Channel ID for global pings")
-
-
-class FeedbackChannelForm(BaseModel):
-    """Form for setting the feedback channel."""
-
-    channel: str = Field(default="", title="Feedback Channel", description="Channel ID for feedback routing")
-
-
-class WelcomeMessageForm(BaseModel):
-    """Form for customizing the onboarding welcome message."""
-
-    message: str = Field(default="", title="Welcome Message", description="Custom welcome message for your guild")
-
-
-class GuildSettings(BaseModel):
-    """Persisted guild settings (not a form — internal state)."""
-
-    reports_channel: int | None = None
-    announcements_channel: int | None = None
-    feedback_channel: int | None = None
-    admin_role: str | None = None
-    member_roles: list[str] = Field(default_factory=list)  # Store multiple member role IDs as strings
-    onboarding_welcome: str | None = None
diff --git a/capy_discord/guild/guild.py b/capy_discord/guild/guild.py
deleted file mode 100644
index 32f52bd..0000000
--- a/capy_discord/guild/guild.py
+++ /dev/null
@@ -1,241 +0,0 @@
-import logging
-from typing import Literal
-
-import discord
-from discord import app_commands
-from discord.ext import commands
-
-from capy_discord.ui.embeds import error_embed
-
-from ._schemas import (
-    GuildSettings,
-)
-
-
-class GuildCog(commands.Cog):
-    """Guild settings management for the capy_discord framework."""
-
-    def __init__(self, bot: commands.Bot) -> None:
-        """Initialize the GuildCog and attach an in-memory settings store to the bot."""
-        self.bot = bot
-        self.log = logging.getLogger(__name__)
-        # In-memory store keyed by guild_id, attached to the bot instance
-        # so it persists across cog reloads.
-        store: dict[int, GuildSettings] | None = getattr(bot, "guild_settings_store", None)
-        if store is None:
-            store = {}
-            setattr(bot, "guild_settings_store", store)  # noqa: B010
-        self._store = store
-
-    def _ensure_settings(self, guild_id: int) -> GuildSettings:
-        """Return existing settings for a guild or create defaults."""
-        if guild_id not in self._store:
-            self._store[guild_id] = GuildSettings()
-        return self._store[guild_id]
-
-    guild = app_commands.Group(name="guild", description="Manage guild settings (single-line)")
-
-    @guild.command(name="channels", description="Set channel IDs in one line")
-    @app_commands.guild_only()
-    @app_commands.describe(
-        reports="Reports channel",
-        announcements="Announcements channel",
-        feedback="Feedback channel",
-    )
-    async def guild_channels(
-        self,
-        interaction: discord.Interaction,
-        reports: discord.TextChannel | None = None,
-        announcements: discord.TextChannel | None = None,
-        feedback: discord.TextChannel | None = None,
-    ) -> None:
-        """Update channels for reporting, announcement, and feedback purposes."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-        settings = self._ensure_settings(interaction.guild.id)
-        if reports is not None:
-            settings.reports_channel = reports.id
-        if announcements is not None:
-            settings.announcements_channel = announcements.id
-        if feedback is not None:
-            settings.feedback_channel = feedback.id
-        await interaction.response.send_message("✅ Channel settings saved.", ephemeral=True)
-
-    @guild.command(name="channels-clear", description="Clear saved channel IDs")
-    @app_commands.guild_only()
-    @app_commands.describe(target="Which channel setting to clear")
-    @app_commands.choices(
-        target=[
-            app_commands.Choice(name="reports", value="reports"),
-            app_commands.Choice(name="announcements", value="announcements"),
-            app_commands.Choice(name="feedback", value="feedback"),
-            app_commands.Choice(name="all", value="all"),
-        ]
-    )
-    async def guild_channels_clear(
-        self,
-        interaction: discord.Interaction,
-        target: Literal["reports", "announcements", "feedback", "all"],
-    ) -> None:
-        """Clear one or all saved channel settings."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-
-        settings = self._ensure_settings(interaction.guild.id)
-        if target in {"reports", "all"}:
-            settings.reports_channel = None
-        if target in {"announcements", "all"}:
-            settings.announcements_channel = None
-        if target in {"feedback", "all"}:
-            settings.feedback_channel = None
-        await interaction.response.send_message(f"✅ Cleared channel setting(s): {target}.", ephemeral=True)
-
-    @guild.command(name="roles", description="Set roles in one line")
-    @app_commands.guild_only()
-    @app_commands.describe(admin="Admin role", member="Member role")
-    async def guild_roles(
-        self,
-        interaction: discord.Interaction,
-        admin: discord.Role | None = None,
-        member: discord.Role | None = None,
-    ) -> None:
-        """Give users roles."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-        settings = self._ensure_settings(interaction.guild.id)
-        if admin is not None:
-            settings.admin_role = str(admin.id)
-        if member is not None:
-            member_id = str(member.id)
-            if member_id not in settings.member_roles:
-                settings.member_roles.append(member_id)
-        await interaction.response.send_message("✅ Role settings saved.", ephemeral=True)
-
-    @guild.command(name="roles-clear", description="Clear saved role settings")
-    @app_commands.guild_only()
-    @app_commands.describe(target="Which role setting to clear")
-    @app_commands.choices(
-        target=[
-            app_commands.Choice(name="admin", value="admin"),
-            app_commands.Choice(name="member_roles", value="member_roles"),
-            app_commands.Choice(name="all", value="all"),
-        ]
-    )
-    async def guild_roles_clear(
-        self,
-        interaction: discord.Interaction,
-        target: Literal["admin", "member_roles", "all"],
-    ) -> None:
-        """Clear one or all saved role settings."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-
-        settings = self._ensure_settings(interaction.guild.id)
-        if target in {"admin", "all"}:
-            settings.admin_role = None
-        if target in {"member_roles", "all"}:
-            settings.member_roles.clear()
-        await interaction.response.send_message(f"✅ Cleared role setting(s): {target}.", ephemeral=True)
-
-    @guild.command(name="onboarding", description="Set the onboarding welcome message")
-    @app_commands.guild_only()
-    @app_commands.describe(message="Welcome message shown during onboarding. Use {user} to reference interacting user.")
-    async def guild_onboarding(self, interaction: discord.Interaction, message: str | None = None) -> None:
-        """Customize onboarding message."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-        settings = self._ensure_settings(interaction.guild.id)
-
-        settings.onboarding_welcome = message or None
-
-        if not settings.onboarding_welcome:
-            await interaction.response.send_message(
-                "✅ Welcome message cleared. (No onboarding message will be sent.)",
-                ephemeral=True,
-            )
-            return
-
-        # A simple "test run" preview:
-        # - let you use {user} in the template
-        preview = settings.onboarding_welcome.replace("{user}", interaction.user.mention)
-
-        # Send an ephemeral preview so it doesn't spam the server
-        await interaction.response.send_message(
-            "✅ Welcome message updated. Here's a test preview (ephemeral):",
-            ephemeral=True,
-        )
-        await interaction.followup.send(preview, ephemeral=True, allowed_mentions=discord.AllowedMentions(users=True))
-
-    @guild.command(name="summary", description="Return a summary of current guild settings")
-    @app_commands.guild_only()
-    async def guild_summary(self, interaction: discord.Interaction) -> None:
-        """Return current guild settings."""
-        if interaction.guild is None:
-            await interaction.response.send_message(
-                embed=error_embed(description="This command can only be used in a server (not in DMs)."),
-                ephemeral=True,
-            )
-            return
-
-        guild = interaction.guild
-        settings = self._ensure_settings(guild.id)
-
-        def channel_mention(channel_id: int | None) -> str:
-            if not channel_id:
-                return "Not set"
-            ch = guild.get_channel(channel_id)
-            return ch.mention if ch else f"<#{channel_id}> (not found)"
-
-        def role_mention(role_id: int | str | None) -> str:
-            if not role_id:
-                return "Not set"
-            normalized_role_id = int(role_id) if isinstance(role_id, str) else role_id
-            role = guild.get_role(normalized_role_id)
-            return role.mention if role else f"<@&{normalized_role_id}> (not found)"
-
-        announcements = channel_mention(getattr(settings, "announcements_channel", None))
-        reports = channel_mention(getattr(settings, "reports_channel", None))
-        feedback = channel_mention(getattr(settings, "feedback_channel", None))
-
-        admin_role = role_mention(getattr(settings, "admin_role", None))
-        member_roles: list[str] = getattr(settings, "member_roles", [])
-        member_role = ", ".join(role_mention(role_id) for role_id in member_roles) if member_roles else "Not set"
-
-        onboarding = settings.onboarding_welcome or "Not set"
-
-        summary = (
-            "**Current Guild Settings**\n"
-            f"Announcements Channel: {announcements}\n"
-            f"Reports Channel: {reports}\n"
-            f"Feedback Channel: {feedback}\n"
-            f"Admin Role: {admin_role}\n"
-            f"Member Roles: {member_role}\n"
-            f"Onboarding Welcome: {onboarding}"
-        )
-
-        await interaction.response.send_message(summary, ephemeral=True)
-
-
-async def setup(bot: commands.Bot) -> None:
-    """Set up the Guild cog."""
-    await bot.add_cog(GuildCog(bot))
diff --git a/capy_discord/resources/__init__.py b/capy_discord/resources/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/capy_discord/services/__init__.py b/capy_discord/services/__init__.py
new file mode 100644
index 0000000..27b1521
--- /dev/null
+++ b/capy_discord/services/__init__.py
@@ -0,0 +1,5 @@
+"""Internal service-layer modules."""
+
+from . import dm, policies
+
+__all__ = ("dm", "policies")
diff --git a/capy_discord/services/dm.py b/capy_discord/services/dm.py
new file mode 100644
index 0000000..a021479
--- /dev/null
+++ b/capy_discord/services/dm.py
@@ -0,0 +1,385 @@
+"""Internal-safe direct message helpers."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import discord
+
+DEFAULT_MAX_RECIPIENTS = 25
+MAX_MESSAGE_LENGTH = 2000
+MAX_PREVIEW_NAMES = 10
+
+
+class DmSafetyError(ValueError):
+    """Raised when a DM operation violates safety constraints."""
+
+
+@dataclass(frozen=True, slots=True)
+class Policy:
+    """Allowlist and cap used to validate a DM request."""
+
+    allowed_user_ids: frozenset[int] = frozenset()
+    allowed_role_ids: frozenset[int] = frozenset()
+    max_recipients: int = DEFAULT_MAX_RECIPIENTS
+
+    def __post_init__(self) -> None:
+        """Validate policy bounds."""
+        if self.max_recipients < 1:
+            msg = "DM policy max_recipients must be at least 1."
+            raise DmSafetyError(msg)
+
+
+@dataclass(slots=True)
+class MessagePayload:
+    """Normalized message content for DM sending."""
+
+    content: str
+
+
+@dataclass(slots=True)
+class AudiencePreview:
+    """Resolved recipient set and preview metadata."""
+
+    recipients: list[discord.Member]
+    skipped_ids: list[int] = field(default_factory=list)
+    source_user_ids: tuple[int, ...] = ()
+    source_role_ids: tuple[int, ...] = ()
+
+    @property
+    def recipient_count(self) -> int:
+        """Return the number of unique resolved recipients."""
+        return len(self.recipients)
+
+
+@dataclass(slots=True)
+class Draft:
+    """Validated DM draft ready for preview or sending."""
+
+    guild_id: int
+    preview: AudiencePreview
+    payload: MessagePayload
+    policy: Policy
+    created_at: datetime = field(default_factory=lambda: datetime.now(ZoneInfo("UTC")))
+
+
+@dataclass(slots=True)
+class SendResult:
+    """Result of a DM batch send."""
+
+    sent_count: int = 0
+    failed_ids: list[int] = field(default_factory=list)
+
+
+class DirectMessenger:
+    """Compose and send direct messages through explicit audience policies."""
+
+    def __init__(self) -> None:
+        """Initialize the DM service logger."""
+        self.log = logging.getLogger(__name__)
+
+    async def compose(
+        self,
+        guild: discord.Guild,
+        content: str,
+        *,
+        user_ids: tuple[int, ...] = (),
+        role_ids: tuple[int, ...] = (),
+        policy: Policy | None = None,
+    ) -> Draft:
+        """Validate the requested audience and return a DM draft."""
+        return await self._compose(
+            guild,
+            content,
+            user_ids=user_ids,
+            role_ids=role_ids,
+            policy=self._resolve_policy(policy),
+        )
+
+    async def compose_to_user(
+        self,
+        guild: discord.Guild,
+        user_id: int,
+        content: str,
+        *,
+        policy: Policy | None = None,
+    ) -> Draft:
+        """Compose a DM draft for a single user."""
+        return await self.compose(guild, content, user_ids=(user_id,), policy=policy)
+
+    async def compose_to_users(
+        self,
+        guild: discord.Guild,
+        user_ids: tuple[int, ...],
+        content: str,
+        *,
+        policy: Policy | None = None,
+    ) -> Draft:
+        """Compose a DM draft for explicit users."""
+        return await self.compose(guild, content, user_ids=user_ids, policy=policy)
+
+    async def compose_to_role(
+        self,
+        guild: discord.Guild,
+        role_id: int,
+        content: str,
+        *,
+        policy: Policy | None = None,
+    ) -> Draft:
+        """Compose a DM draft for a single role."""
+        return await self.compose(guild, content, role_ids=(role_id,), policy=policy)
+
+    async def compose_to_roles(
+        self,
+        guild: discord.Guild,
+        role_ids: tuple[int, ...],
+        content: str,
+        *,
+        policy: Policy | None = None,
+    ) -> Draft:
+        """Compose a DM draft for explicit roles."""
+        return await self.compose(guild, content, role_ids=role_ids, policy=policy)
+
+    async def send(self, guild: discord.Guild, draft: Draft) -> SendResult:
+        """Send a validated DM draft."""
+        if draft.guild_id != guild.id:
+            msg = "DM draft guild does not match the provided guild."
+            raise DmSafetyError(msg)
+
+        self._validate_send_policy(draft.policy, draft.preview)
+        result = SendResult()
+
+        for recipient in draft.preview.recipients:
+            try:
+                await recipient.send(
+                    draft.payload.content,
+                    allowed_mentions=discord.AllowedMentions.none(),
+                )
+                result.sent_count += 1
+            except (discord.Forbidden, discord.HTTPException):
+                result.failed_ids.append(recipient.id)
+
+        self.log.info(
+            "DM batch complete guild=%s recipients=%s sent=%s failed=%s",
+            guild.id,
+            draft.preview.recipient_count,
+            result.sent_count,
+            len(result.failed_ids),
+        )
+        return result
+
+    def render_preview(self, draft: Draft) -> str:
+        """Render a compact preview for logging or operator review."""
+        mentions = [recipient.mention for recipient in draft.preview.recipients[:MAX_PREVIEW_NAMES]]
+        preview_mentions = ", ".join(mentions) if mentions else "None"
+        if draft.preview.recipient_count > MAX_PREVIEW_NAMES:
+            preview_mentions = f"{preview_mentions}, ..."
+
+        return (
+            f"DM draft for guild={draft.guild_id}\n"
+            f"Recipients: {draft.preview.recipient_count}\n"
+            f"Skipped IDs: {len(draft.preview.skipped_ids)}\n"
+            f"Source user IDs: {len(draft.preview.source_user_ids)}\n"
+            f"Source role IDs: {len(draft.preview.source_role_ids)}\n"
+            f"Recipients preview: {preview_mentions}\n\n"
+            f"Message:\n{draft.payload.content}"
+        )
+
+    async def _compose(
+        self,
+        guild: discord.Guild,
+        content: str,
+        *,
+        user_ids: tuple[int, ...],
+        role_ids: tuple[int, ...],
+        policy: Policy,
+    ) -> Draft:
+        normalized_content = self._normalize_content(content)
+        self._validate_requested_audience(user_ids, role_ids, policy, guild.default_role.id)
+        preview = await self._resolve_audience(guild, user_ids=user_ids, role_ids=role_ids)
+        self._validate_send_policy(policy, preview)
+
+        draft = Draft(
+            guild_id=guild.id,
+            preview=preview,
+            payload=MessagePayload(content=normalized_content),
+            policy=policy,
+        )
+        self.log.info(
+            "DM draft composed guild=%s users=%s roles=%s recipients=%s",
+            guild.id,
+            len(preview.source_user_ids),
+            len(preview.source_role_ids),
+            preview.recipient_count,
+        )
+        return draft
+
+    def _resolve_policy(self, policy: Policy | None) -> Policy:
+        if policy is not None:
+            return policy
+
+        return Policy()
+
+    def _normalize_content(self, content: str) -> str:
+        normalized = content.strip()
+        if not normalized:
+            msg = "DM content must not be empty."
+            raise DmSafetyError(msg)
+        if len(normalized) > MAX_MESSAGE_LENGTH:
+            msg = f"DM content cannot exceed {MAX_MESSAGE_LENGTH} characters."
+            raise DmSafetyError(msg)
+        return normalized
+
+    def _validate_requested_audience(
+        self,
+        user_ids: tuple[int, ...],
+        role_ids: tuple[int, ...],
+        policy: Policy,
+        default_role_id: int,
+    ) -> None:
+        if not user_ids and not role_ids:
+            msg = "DM request must include at least one explicit user ID or role ID."
+            raise DmSafetyError(msg)
+
+        if default_role_id in role_ids or default_role_id in policy.allowed_role_ids:
+            msg = "The @everyone role cannot be used in DM policies or requests."
+            raise DmSafetyError(msg)
+
+        disallowed_users = set(user_ids) - set(policy.allowed_user_ids)
+        if disallowed_users:
+            msg = f"DM request includes user IDs outside the allowed policy: {sorted(disallowed_users)}"
+            raise DmSafetyError(msg)
+
+        disallowed_roles = set(role_ids) - set(policy.allowed_role_ids)
+        if disallowed_roles:
+            msg = f"DM request includes role IDs outside the allowed policy: {sorted(disallowed_roles)}"
+            raise DmSafetyError(msg)
+
+    async def _resolve_audience(
+        self,
+        guild: discord.Guild,
+        *,
+        user_ids: tuple[int, ...],
+        role_ids: tuple[int, ...],
+    ) -> AudiencePreview:
+        recipients_by_id: dict[int, discord.Member] = {}
+        skipped_ids: list[int] = []
+
+        for user_id in user_ids:
+            member = await self._resolve_member(guild, user_id)
+            if member is None:
+                skipped_ids.append(user_id)
+                continue
+            recipients_by_id[member.id] = member
+
+        for role_id in role_ids:
+            role = guild.get_role(role_id)
+            if role is None:
+                skipped_ids.append(role_id)
+                continue
+            if role == guild.default_role:
+                msg = "The @everyone role cannot be used for DMs."
+                raise DmSafetyError(msg)
+            for member in role.members:
+                recipients_by_id[member.id] = member
+
+        if not recipients_by_id:
+            msg = "No recipients were resolved. Use explicit users or non-default roles."
+            raise DmSafetyError(msg)
+
+        return AudiencePreview(
+            recipients=list(recipients_by_id.values()),
+            skipped_ids=skipped_ids,
+            source_user_ids=user_ids,
+            source_role_ids=role_ids,
+        )
+
+    def _validate_send_policy(self, policy: Policy, preview: AudiencePreview) -> None:
+        if preview.recipient_count > policy.max_recipients:
+            msg = (
+                f"Resolved audience has {preview.recipient_count} recipients, "
+                f"which exceeds the cap of {policy.max_recipients}."
+            )
+            raise DmSafetyError(msg)
+
+    async def _resolve_member(self, guild: discord.Guild, user_id: int) -> discord.Member | None:
+        member = guild.get_member(user_id)
+        if member is not None:
+            return member
+
+        try:
+            return await guild.fetch_member(user_id)
+        except (discord.NotFound, discord.Forbidden, discord.HTTPException):
+            return None
+
+
+_MESSENGER = DirectMessenger()
+
+
+async def compose(
+    guild: discord.Guild,
+    content: str,
+    *,
+    user_ids: tuple[int, ...] = (),
+    role_ids: tuple[int, ...] = (),
+    policy: Policy | None = None,
+) -> Draft:
+    """Compose a DM draft through the shared messenger."""
+    return await _MESSENGER.compose(guild, content, user_ids=user_ids, role_ids=role_ids, policy=policy)
+
+
+async def compose_to_user(
+    guild: discord.Guild,
+    user_id: int,
+    content: str,
+    *,
+    policy: Policy | None = None,
+) -> Draft:
+    """Compose a DM draft for a single user."""
+    return await _MESSENGER.compose_to_user(guild, user_id, content, policy=policy)
+
+
+async def compose_to_users(
+    guild: discord.Guild,
+    user_ids: tuple[int, ...],
+    content: str,
+    *,
+    policy: Policy | None = None,
+) -> Draft:
+    """Compose a DM draft for explicit users."""
+    return await _MESSENGER.compose_to_users(guild, user_ids, content, policy=policy)
+
+
+async def compose_to_role(
+    guild: discord.Guild,
+    role_id: int,
+    content: str,
+    *,
+    policy: Policy | None = None,
+) -> Draft:
+    """Compose a DM draft for a single role."""
+    return await _MESSENGER.compose_to_role(guild, role_id, content, policy=policy)
+
+
+async def compose_to_roles(
+    guild: discord.Guild,
+    role_ids: tuple[int, ...],
+    content: str,
+    *,
+    policy: Policy | None = None,
+) -> Draft:
+    """Compose a DM draft for explicit roles."""
+    return await _MESSENGER.compose_to_roles(guild, role_ids, content, policy=policy)
+
+
+async def send(guild: discord.Guild, draft: Draft) -> SendResult:
+    """Send a previously composed draft through the shared messenger."""
+    return await _MESSENGER.send(guild, draft)
+
+
+def render_preview(draft: Draft) -> str:
+    """Render a compact preview for a draft."""
+    return _MESSENGER.render_preview(draft)
diff --git a/capy_discord/services/policies.py b/capy_discord/services/policies.py
new file mode 100644
index 0000000..6017b75
--- /dev/null
+++ b/capy_discord/services/policies.py
@@ -0,0 +1,37 @@
+"""Safe policy helpers for direct messaging."""
+
+from __future__ import annotations
+
+from capy_discord.services.dm import DEFAULT_MAX_RECIPIENTS, Policy
+
+DENY_ALL = Policy()
+
+
+def allow_users(*user_ids: int, max_recipients: int = DEFAULT_MAX_RECIPIENTS) -> Policy:
+    """Build a policy that only permits the provided user IDs."""
+    return Policy(
+        allowed_user_ids=frozenset(user_ids),
+        max_recipients=max_recipients,
+    )
+
+
+def allow_roles(*role_ids: int, max_recipients: int = DEFAULT_MAX_RECIPIENTS) -> Policy:
+    """Build a policy that only permits the provided role IDs."""
+    return Policy(
+        allowed_role_ids=frozenset(role_ids),
+        max_recipients=max_recipients,
+    )
+
+
+def allow_targets(
+    *,
+    user_ids: frozenset[int] = frozenset(),
+    role_ids: frozenset[int] = frozenset(),
+    max_recipients: int = DEFAULT_MAX_RECIPIENTS,
+) -> Policy:
+    """Build a policy that permits the provided user and role IDs."""
+    return Policy(
+        allowed_user_ids=user_ids,
+        allowed_role_ids=role_ids,
+        max_recipients=max_recipients,
+    )
diff --git a/tests/capy_discord/services/__init__.py b/tests/capy_discord/services/__init__.py
new file mode 100644
index 0000000..02460ba
--- /dev/null
+++ b/tests/capy_discord/services/__init__.py
@@ -0,0 +1 @@
+"""Tests for service-layer modules."""
diff --git a/tests/capy_discord/services/test_dm.py b/tests/capy_discord/services/test_dm.py
new file mode 100644
index 0000000..6527b30
--- /dev/null
+++ b/tests/capy_discord/services/test_dm.py
@@ -0,0 +1,125 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import discord
+import pytest
+
+from capy_discord.services import dm, policies
+
+
+def make_member(member_id: int) -> MagicMock:
+    member = MagicMock(spec=discord.Member)
+    member.id = member_id
+    member.mention = f"<@{member_id}>"
+    member.send = AsyncMock()
+    return member
+
+
+@pytest.mark.asyncio
+async def test_compose_defaults_to_deny_all_policy():
+    guild = MagicMock(spec=discord.Guild)
+    guild.default_role.id = 999
+
+    with pytest.raises(dm.DmSafetyError, match="outside the allowed policy"):
+        await dm.compose_to_user(
+            guild,
+            42,
+            "Hello",
+        )
+
+
+@pytest.mark.asyncio
+async def test_compose_rejects_everyone_role():
+    guild = MagicMock(spec=discord.Guild)
+    guild.default_role.id = 1
+
+    with pytest.raises(dm.DmSafetyError, match="@everyone"):
+        await dm.compose_to_role(
+            guild,
+            1,
+            "Hello",
+            policy=policies.allow_roles(1),
+        )
+
+
+@pytest.mark.asyncio
+async def test_compose_to_user_rejects_target_outside_policy():
+    guild = MagicMock(spec=discord.Guild)
+    guild.default_role.id = 999
+
+    with pytest.raises(dm.DmSafetyError, match="outside the allowed policy"):
+        await dm.compose_to_user(
+            guild,
+            42,
+            "Hello",
+            policy=policies.allow_users(7),
+        )
+
+
+@pytest.mark.asyncio
+async def test_compose_deduplicates_users_from_roles_and_explicit_ids():
+    member = make_member(42)
+    role = MagicMock(spec=discord.Role)
+    role.members = [member]
+
+    guild = MagicMock(spec=discord.Guild)
+    guild.id = 123
+    guild.default_role.id = 999
+    guild.get_member.return_value = member
+    guild.get_role.return_value = role
+
+    draft = await dm.compose(
+        guild,
+        "Hello",
+        user_ids=(42,),
+        role_ids=(7,),
+        policy=policies.allow_targets(
+            user_ids=frozenset({42}),
+            role_ids=frozenset({7}),
+            max_recipients=1,
+        ),
+    )
+
+    assert draft.preview.recipient_count == 1
+    assert draft.preview.recipients == [member]
+    assert draft.preview.skipped_ids == []
+
+
+@pytest.mark.asyncio
+async def test_compose_rejects_audience_above_cap():
+    guild = MagicMock(spec=discord.Guild)
+    guild.id = 123
+    guild.default_role.id = 999
+    guild.get_member.side_effect = [make_member(1), make_member(2)]
+
+    with pytest.raises(dm.DmSafetyError, match="exceeds the cap"):
+        await dm.compose_to_users(
+            guild,
+            (1, 2),
+            "Hello",
+            policy=policies.allow_users(1, 2, max_recipients=1),
+        )
+
+
+@pytest.mark.asyncio
+async def test_send_tracks_failures():
+    ok_member = make_member(1)
+    blocked_member = make_member(2)
+    blocked_member.send.side_effect = discord.Forbidden(
+        response=SimpleNamespace(status=403, reason="forbidden"),
+        message="forbidden",
+    )
+
+    guild = MagicMock(spec=discord.Guild)
+    guild.id = 123
+    draft = dm.Draft(
+        guild_id=123,
+        preview=dm.AudiencePreview(recipients=[ok_member, blocked_member]),
+        payload=dm.MessagePayload(content="Hello"),
+        policy=policies.allow_users(1, 2, max_recipients=2),
+    )
+
+    result = await dm.send(guild, draft)
+
+    assert result.sent_count == 1
+    assert result.failed_ids == [2]