diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
new file mode 100644
index 0000000..2181115
--- /dev/null
+++ b/.github/workflows/checks.yml
@@ -0,0 +1,24 @@
+name: Checks
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  check:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v7
+        with:
+          python-version: '3.11'
+      - run: python -m pip install -e ".[dev]"
+      - run: ruff check .
+      - run: python -m pytest
diff --git a/README.md b/README.md
index da06ee2..1d24daa 100644
--- a/README.md
+++ b/README.md
@@ -74,7 +74,9 @@ history and ordering can [bind directly to the channel](docs/slack.md#using-an-e
 
 The [Slack guide](docs/slack.md) covers setup, working indicators, user profiles,
 thread history, incoming attachments, rich replies, uploads, and mirrored inputs.
-Existing `from agentchat.channels import Slack` imports remain supported.
+Import each provider from its own package: `from agentchat.channels.slack import Slack`.
+Slack can also [use an application-owned event source](docs/slack.md#composing-event-sources)
+without an app token or Socket Mode connection.
 
 ## Development
 
@@ -87,3 +89,4 @@ python -m pytest
 Provider behavior tests live under `tests/channels/`, framework tests under
 `tests/integrations/`, and shared adapter contracts under `tests/contracts/`.
 Shared fakes and event builders live in `tests/support/`.
+CI runs lint and the full suite, including the channel-folder containment rule.
diff --git a/docs/adding-a-channel.md b/docs/adding-a-channel.md
index d2ca2ba..8b55237 100644
--- a/docs/adding-a-channel.md
+++ b/docs/adding-a-channel.md
@@ -5,6 +5,9 @@ A channel is an adapter for a messaging platform. Put its implementation in
 pass an instance to `AgentChat(channels=[...])`. No dispatcher changes or provider
 registration are required.
 
+CI enforces that `channels/` contains only `__init__.py`, `base.py`, and provider
+packages. Keep provider helpers inside their package; root-level shims fail the test.
+
 ## Required contract
 
 Implement the existing `Channel` protocol in `agentchat.channels.base`:
@@ -90,6 +93,26 @@ python -m pytest
 
 Finally add a runnable example and a provider setup guide. Keep the default
 package import free of the provider SDK. If adding dependencies or optional
-extras, test the installed package as well as the editable checkout. A new
-provider may add a convenience namespace export, but it should not change core
-runtime modules or another adapter.
+extras, test the installed package as well as the editable checkout. Provider exports belong in their own package, keeping `agentchat.channels`
+limited to shared contracts. Adding a provider should not change core runtime
+modules, the shared namespace or another adapter.
+
+
+## Compose transports when a provider needs them
+
+Keep the generic `Channel` contract focused on normalized messages. If one provider
+supports multiple ways to receive native events, inject its event source into the
+adapter and keep parsing/delivery shared. Slack demonstrates this with
+`SlackEventSource` in `channels/slack/transport.py` and the default `SocketMode`
+implementation in `channels/slack/socket_mode.py`.
+
+A source implements `run(receiver)`, `close()` and `is_connected()`. It owns
+transport resources and must stop intake and clean up when closed or cancelled.
+It supplies provider events to the adapter callback; it does not reimplement
+normalization, conversation state or agent execution. Authentication of an HTTP
+request and durable acknowledgement remain the host source's responsibility.
+
+The Slack source tests exercise the same adapter with an injected event queue,
+including duplicate messages, subscribed follow-ups and replies. This demonstrates
+composition without adding a provider registry or requiring every channel to
+adopt Slack's transport interface.
diff --git a/docs/architecture.md b/docs/architecture.md
index 7502da2..a7beecf 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -53,46 +53,91 @@ Applications own sender authorization, agent execution, durable work and deliver
 recovery. Direct channel binding is also supported when an application already
 owns deduplication, ordering and history; see the [Slack guide](slack.md#using-an-existing-agent-runner).
 
-## Slack adapter
+## Provider containment
+
+The shared channel namespace contains only adapter contracts. A provider owns
+its implementations, exports and optional dependency loading:
 
 ```text
-channels/slack/
-├── __init__.py    public Slack and supporting types
-├── adapter.py     lifecycle, subscriptions, history and API coordination
-├── events.py      pure event/attachment normalization
-├── rendering.py   pure text and rich-reply formatting
-├── files.py       uploads and bounded downloads
-└── mirroring.py   external-input rendering and sender/role recovery
+channels/
+├── __init__.py        shared contract exports only
+├── base.py            Channel and optional capability protocols
+└── slack/
+    ├── __init__.py    lazy public Slack exports
+    ├── adapter.py     shared Slack message behavior and Web API operations
+    ├── transport.py   SlackEventSource and its event receiver contract
+    ├── socket_mode.py Socket Mode connection, acknowledgements and shutdown
+    ├── events.py      pure event/attachment normalization
+    ├── rendering.py   pure text and rich-reply formatting
+    ├── files.py       uploads and bounded downloads
+    └── mirroring.py   external-input rendering and sender/role recovery
 ```
 
-Event decoding does not download files. File transport receives explicit request
-callbacks and authorization guards. Slack thread subscriptions stay with Slack;
-they are provider routing policy, not the application's conversation store.
-The adapter retains socket acknowledgement, bounded background tasks and cleanup.
-
-Keep new provider implementations in their own packages. Add internal files when
-their responsibilities justify it; a small provider can start with `adapter.py`.
-See [adding a channel](adding-a-channel.md) for the extension contract.
-
-## Public imports and compatibility
+`base.py` is shared because it defines what any messaging adapter must implement.
+Every Slack-specific module lives below `slack/`. There are no sibling Slack
+compatibility modules or provider aliases in the shared namespace.
 
-Both `from agentchat.channels import Slack` and
-`from agentchat.channels.slack import Slack` remain supported. Root-level message
-and context imports are unchanged; `agentchat.models.MessageContext` and
-`agentchat.models.Conversation` resolve to the same classes as `agentchat.context`.
+## Composition
 
-The `channels` and `integrations` namespaces load concrete implementations only
-when requested. Importing the core does not import Slack, aiohttp or OpenAI Agents.
-Installation defaults are unchanged: Slack dependencies remain installed with the
-base package, while OpenAI Agents still uses the `openai` extra.
+The generic adapter boundary remains `Channel`. Within Slack, replace event
+delivery independently of message behavior:
 
-The documented legacy helper modules forward to canonical implementations:
+```text
+SocketMode ────────────┐
+                       ├── SlackEventSource ── Slack.handle_event ── AgentChat
+application event feed ┘                           │
+                                                  ├── subscriptions/history
+                                                  └── replies/files/status
+```
 
-| Compatibility module | Canonical location |
+`Slack.run()` authenticates the bot and checks the configured workspace before
+starting its source. The source calls the same `handle_event` method for every
+payload, so normalization, subscription filtering, sender identity and replies
+remain shared. `Slack.close()` and `is_connected()` delegate source lifecycle.
+The default Socket Mode implementation owns socket SDK imports, acknowledgement,
+bounded event tasks and graceful shutdown.
+
+Supply exactly one of `app_token` (the Socket Mode convenience) or `event_source`
+(an application-owned source). A custom source requires no app token and does
+not import or instantiate Socket Mode. See the [composition guide](slack.md#composing-event-sources).
+
+| Replaceable concern | Injection point | Responsibility |
+| --- | --- | --- |
+| Incoming events | `Slack(event_source=...)` | Deliver authenticated Slack events; own input lifecycle |
+| Slack Web API calls | `Slack(web_client=...)` | Provide the existing `SlackWebClient` operations |
+| Followed threads | `Slack(thread_subscriptions=...)` | Persist or expire Slack thread subscriptions |
+| Conversation state | `AgentChat(state=...)` | Claim events, serialize conversations, retain history |
+| Agent execution | `@app.on_message` | Authorize sender and call the application's runner |
+
+Event decoding never downloads files. Standalone event and rendering helpers can
+be imported without provider SDKs. File transport helpers accept explicit request
+callbacks and authorization guards. Applications retain HTTP signature checking,
+durable queues, authorization and ambiguous-delivery recovery.
+
+Raw event envelopes, acknowledgement and connection behavior are provider-specific;
+they do not belong in the generic `Channel` contract. A new provider can use a
+similar composition seam when it has multiple input mechanisms, without inheriting
+Slack code or changing `AgentChat`.
+
+## Import migration
+
+This cleanup deliberately removes old Slack shortcut/helper imports. Update them
+as follows; no hidden aliases are retained:
+
+| Previous import | Current import |
 | --- | --- |
-| `channels.slack_files` | `channels.slack.events` for attachment normalization; `channels.slack.files` for downloads |
-| `channels.slack_media` | `channels.slack.rendering` for formatting; `channels.slack.files` for uploads |
-| `channels.slack_mirror` | `channels.slack.mirroring` |
-
-New code should import the canonical modules. Tests that patch implementation
-details should patch the defining module, not a compatibility re-export.
+| `from agentchat.channels import Slack, SlackUser` | `from agentchat.channels.slack import Slack, SlackUser` |
+| `agentchat.channels.slack_files.slack_attachments` | `agentchat.channels.slack.events.slack_attachments` |
+| `agentchat.channels.slack_files.download_slack_file` | `agentchat.channels.slack.files.download_slack_file` |
+| `agentchat.channels.slack_media.upload_slack_files` | `agentchat.channels.slack.files.upload_slack_files` |
+| `agentchat.channels.slack_media.rich_payload` | `agentchat.channels.slack.rendering.rich_payload` |
+| `agentchat.channels.slack_mirror.mirror_payload` | `agentchat.channels.slack.mirroring.mirror_payload` |
+
+Other rendering helpers (`text_blocks`), file helpers (`upload_bytes`,
+`validate_destination`, `SlackRequest`, `DeliveryGuard`) and `read_mirror` are
+available from the corresponding canonical modules above.
+
+Root-level message and context imports are unchanged. Importing the core does
+not import Slack, aiohttp or OpenAI Agents. Installation defaults are unchanged:
+Slack dependencies remain installed with the base package, while OpenAI Agents
+still uses the `openai` extra.
diff --git a/docs/slack.md b/docs/slack.md
index aeefb5f..003dc0d 100644
--- a/docs/slack.md
+++ b/docs/slack.md
@@ -404,12 +404,85 @@ With direct binding, the application owns deduplication and request ordering.
 
 `Slack.run()` checks the bot identity. Pass `workspace_id=` to reject a token or
 event from another workspace. `await slack.is_connected()` exposes socket readiness.
-Socket envelopes are acknowledged before processing in tracked background tasks.
+The default Socket Mode source acknowledges envelopes before processing in tracked background tasks.
 `max_pending_events` defaults to 64; excess envelopes are left unacknowledged for
 Slack's retry policy. `await slack.close()` closes the socket, then drains tasks
 for up to 20 seconds before cancelling unfinished work. This is bounded in-process
 delivery, not a durable queue; applications performing writes need their own journal.
 
+## Composing event sources
+
+`Slack` combines a replaceable incoming event source with shared Slack message
+behavior. Existing Socket Mode setup remains:
+
+```python
+slack = Slack(bot_token="xoxb-...", app_token="xapp-...")
+# Equivalent environment-based convenience:
+slack = Slack.from_env()
+```
+
+For a host-owned verified webhook queue or another Slack event feed, supply a
+`SlackEventSource` instead. No app token or Socket Mode connection is needed:
+
+```python
+import os
+
+from agentchat import AgentChat, MessageContext
+from agentchat.channels.slack import Slack, SlackEventSource
+
+
+async def serve(source: SlackEventSource) -> None:
+    slack = Slack(
+        bot_token=os.environ["SLACK_BOT_TOKEN"],
+        event_source=source,
+        ack_emoji=None,
+    )
+    app = AgentChat(channels=[slack])
+
+    @app.on_message
+    async def respond(context: MessageContext) -> str:
+        # The application authorizes the sender before running agent tools.
+        return context.message.text or "Received your attachment."
+
+    try:
+        await app.run()
+    finally:
+        await app.close()
+```
+
+Your source implements these operations structurally; no inheritance is required:
+
+- `async run(receiver)`: deliver already authenticated Slack Events API payloads
+  through `await receiver(payload)` for the source's lifetime.
+- `async close()`: stop intake and release resources, draining or cancelling
+  in-flight work according to the source's documented policy; repeated close calls
+  must be safe.
+- `async is_connected() -> bool`: report that source's readiness.
+
+The receiver takes the same mapping accepted by `slack.handle_event`, including
+`team_id` and `event`. Slack verifies bot identity and the configured workspace
+before starting the source. Subscriptions, parsing, files, history and replies
+are unchanged across sources, and AgentChat continues to own deduplication and
+conversation ordering. Outgoing Slack API calls are independently injectable
+through `web_client=`. `max_pending_events` configures the default Socket Mode
+source; a custom source owns its own concurrency and queue limits.
+
+Supply exactly one of `app_token` and `event_source`. A bare `Slack(bot_token=...)`
+without a source is rejected. For an existing host that binds directly to the
+channel, supply its source and let `Slack.run()` authenticate before delivering
+events through it.
+
+The source owns HTTP signature/timestamp verification and transport acknowledgements.
+`handle_event` accepts decoded events; it is not an HTTP signature verifier. Hosts
+using durable queues must define when events are acknowledged and how uncertain
+outcomes are recovered. The included Socket Mode source remains bounded,
+in-process delivery. It owns socket acknowledgements, connection state, pending
+event limits and shutdown independently of Slack's message behavior.
+
+See [the source protocol](../src/agentchat/channels/slack/transport.py) and
+[the injected-source tests](../tests/channels/slack/test_event_source.py) for a
+concrete queue-source composition that reuses the same adapter.
+
 ## Development
 
 ```bash
diff --git a/src/agentchat/channels/__init__.py b/src/agentchat/channels/__init__.py
index c2b9401..cb7c652 100644
--- a/src/agentchat/channels/__init__.py
+++ b/src/agentchat/channels/__init__.py
@@ -1,28 +1,23 @@
-from typing import TYPE_CHECKING
+"""Provider-neutral adapter contracts; import adapters from their own packages."""
 
-from agentchat.channels.base import Channel, ThreadHistoryChannel
+from agentchat.channels.base import (
+    Channel,
+    FileDownloadChannel,
+    FileUploadChannel,
+    MessageReceiver,
+    MirroringChannel,
+    RichReplyChannel,
+    StatusChannel,
+    ThreadHistoryChannel,
+)
 
-if TYPE_CHECKING:
-    from agentchat.channels.slack import Slack, SlackUser
-
-__all__ = ["Channel", "Slack", "SlackUser", "ThreadHistoryChannel"]
-
-
-def __getattr__(name: str) -> object:
-    if name in {"Slack", "SlackUser"}:
-        try:
-            from agentchat.channels import slack
-        except ModuleNotFoundError as exc:
-            if exc.name not in {"slack_sdk", "aiohttp"}:
-                raise
-            raise ModuleNotFoundError(
-                "Slack requires slack-sdk and aiohttp. "
-                "Install AgentChat with its dependencies: pip install agentchat",
-                name=exc.name,
-            ) from exc
-        return getattr(slack, name)
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-
-
-def __dir__() -> list[str]:
-    return sorted(set(globals()) | set(__all__))
+__all__ = [
+    "Channel",
+    "FileDownloadChannel",
+    "FileUploadChannel",
+    "MessageReceiver",
+    "MirroringChannel",
+    "RichReplyChannel",
+    "StatusChannel",
+    "ThreadHistoryChannel",
+]
diff --git a/src/agentchat/channels/slack/__init__.py b/src/agentchat/channels/slack/__init__.py
index 14f819d..4184e7b 100644
--- a/src/agentchat/channels/slack/__init__.py
+++ b/src/agentchat/channels/slack/__init__.py
@@ -1,17 +1,39 @@
-"""Slack channel and reusable Slack-specific helpers."""
+"""Slack adapter, replaceable event sources and reusable provider helpers."""
 
-from agentchat.channels.slack.adapter import (
-    MemoryThreadSubscriptions,
-    Slack,
-    SlackUser,
-    SlackWebClient,
-    ThreadSubscriptions,
-)
+from typing import TYPE_CHECKING
+
+from agentchat.channels.slack.transport import SlackEventReceiver, SlackEventSource
+
+if TYPE_CHECKING:
+    from agentchat.channels.slack.adapter import (
+        MemoryThreadSubscriptions,
+        Slack,
+        SlackUser,
+        SlackWebClient,
+        ThreadSubscriptions,
+    )
+    from agentchat.channels.slack.socket_mode import SocketMode
 
 __all__ = [
     "MemoryThreadSubscriptions",
     "Slack",
+    "SlackEventReceiver",
+    "SlackEventSource",
     "SlackUser",
     "SlackWebClient",
+    "SocketMode",
     "ThreadSubscriptions",
 ]
+
+
+def __getattr__(name: str) -> object:
+    if name == "SocketMode":
+        from agentchat.channels.slack.socket_mode import SocketMode
+
+        return SocketMode
+    if name in {"MemoryThreadSubscriptions", "Slack", "SlackUser", "SlackWebClient",
+                "ThreadSubscriptions"}:
+        from agentchat.channels.slack import adapter
+
+        return getattr(adapter, name)
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
diff --git a/src/agentchat/channels/slack/adapter.py b/src/agentchat/channels/slack/adapter.py
index 1deaa04..e4a89db 100644
--- a/src/agentchat/channels/slack/adapter.py
+++ b/src/agentchat/channels/slack/adapter.py
@@ -1,19 +1,14 @@
-"""Slack channel facade with Socket Mode lifecycle and native capabilities."""
+"""Slack channel behavior composed with a replaceable event source."""
 from __future__ import annotations
 
 import asyncio
-import logging
 import os
 from collections import OrderedDict, deque
-from collections.abc import Awaitable, Callable, Mapping, Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Protocol, cast
 
-from slack_sdk.socket_mode.aiohttp import SocketModeClient
-from slack_sdk.socket_mode.async_client import AsyncBaseSocketModeClient
-from slack_sdk.socket_mode.request import SocketModeRequest
-from slack_sdk.socket_mode.response import SocketModeResponse
 from slack_sdk.web.async_client import AsyncWebClient
 
 from agentchat.channels.base import MessageReceiver
@@ -25,6 +20,7 @@
 )
 from agentchat.channels.slack.mirroring import mirror_payload, read_mirror
 from agentchat.channels.slack.rendering import rich_payload, text_blocks
+from agentchat.channels.slack.transport import SlackEventSource
 from agentchat.models import (
     Attachment,
     DownloadedFile,
@@ -35,8 +31,6 @@
     UploadFile,
 )
 
-logger = logging.getLogger(__name__)
-
 
 @dataclass(frozen=True, slots=True)
 class SlackUser:
@@ -99,7 +93,8 @@ def __init__(
         self,
         *,
         bot_token: str,
-        app_token: str,
+        app_token: str | None = None,
+        event_source: SlackEventSource | None = None,
         web_client: SlackWebClient | None = None,
         ack_emoji: str | None = "eyes",
         workspace_id: str | None = None,
@@ -107,23 +102,33 @@ def __init__(
         thread_subscriptions: ThreadSubscriptions | None = None,
         max_pending_events: int = 64,
     ) -> None:
+        if (app_token is None) == (event_source is None):
+            raise ValueError("Provide exactly one of app_token or event_source")
         if max_pending_events < 1:
             raise ValueError("max_pending_events must be positive")
         self._bot_token = bot_token
-        self._app_token = app_token
         self._web_client = cast(
             SlackWebClient,
             web_client or AsyncWebClient(token=bot_token, retry_handlers=[]),
         )
-        self._socket_client: SocketModeClient | None = None
+        if event_source is None:
+            from agentchat.channels.slack.socket_mode import SocketMode
+
+            event_source = SocketMode(
+                app_token=cast(str, app_token),
+                web_client=cast(AsyncWebClient, self._web_client),
+                max_pending_events=max_pending_events,
+            )
+        self._event_source = event_source
         self._receiver: MessageReceiver | None = None
-        self._closed = asyncio.Event()
+        self._run_started = False
+        self._auth_task: asyncio.Task[Mapping[str, object]] | None = None
+        self._auth_cancelled_for_close = False
+        self._close_task: asyncio.Task[None] | None = None
         self._ack_emoji = ack_emoji
         self._workspace_id = workspace_id
         self._bot_user_id = bot_user_id
         self._threads = thread_subscriptions or MemoryThreadSubscriptions()
-        self._max_pending_events = max_pending_events
-        self._tasks: set[asyncio.Task[None]] = set()
 
     @classmethod
     def from_env(cls, *, ack_emoji: str | None = "eyes") -> Slack:
@@ -139,44 +144,45 @@ def bind(self, receiver: MessageReceiver) -> None:
     async def run(self) -> None:
         if self._receiver is None:
             raise RuntimeError("Slack channel is not bound to AgentChat")
-        identity = await self._web_client.auth_test()
-        workspace_id, bot_user_id = identity.get("team_id"), identity.get("user_id")
-        if not isinstance(workspace_id, str) or not isinstance(bot_user_id, str):
-            raise RuntimeError("Slack did not return a bot identity")
-        if self._workspace_id is not None and self._workspace_id != workspace_id:
-            raise RuntimeError("Slack token belongs to a different workspace")
-        self._workspace_id, self._bot_user_id = workspace_id, bot_user_id
-        self._socket_client = SocketModeClient(
-            app_token=self._app_token,
-            web_client=cast(AsyncWebClient, self._web_client),
-        )
-        listeners = cast(
-            list[
-                Callable[
-                    [AsyncBaseSocketModeClient, SocketModeRequest],
-                    Awaitable[None],
-                ]
-            ],
-            self._socket_client.socket_mode_request_listeners,
-        )
-        listeners.append(self._handle_socket_request)
-        connect = cast(Callable[[], Awaitable[None]], self._socket_client.connect)
-        await connect()
-        await self._closed.wait()
+        if self._run_started or self._close_task is not None:
+            raise RuntimeError("Slack channel has already been started or closed")
+        self._run_started = True
+        try:
+            self._auth_task = asyncio.create_task(self._web_client.auth_test())
+            try:
+                identity = await self._auth_task
+            except asyncio.CancelledError:
+                current_task = asyncio.current_task()
+                if not self._auth_cancelled_for_close or (
+                    current_task is not None and current_task.cancelling()
+                ):
+                    raise
+                return
+            workspace_id, bot_user_id = identity.get("team_id"), identity.get("user_id")
+            if not isinstance(workspace_id, str) or not isinstance(bot_user_id, str):
+                raise RuntimeError("Slack did not return a bot identity")
+            if self._workspace_id is not None and self._workspace_id != workspace_id:
+                raise RuntimeError("Slack token belongs to a different workspace")
+            self._workspace_id, self._bot_user_id = workspace_id, bot_user_id
+            if self._close_task is None:
+                await self._event_source.run(self.handle_event)
+        finally:
+            await self.close()
 
     async def close(self) -> None:
-        self._closed.set()
-        if self._socket_client is not None:
-            close = cast(Callable[[], Awaitable[None]], self._socket_client.close)
-            await close()
-        if self._tasks:
-            _, pending = await asyncio.wait(self._tasks, timeout=20)
-            for task in pending:
-                task.cancel()
-            await asyncio.gather(*pending, return_exceptions=True)
+        if self._close_task is None:
+            self._close_task = asyncio.create_task(self._close())
+        await asyncio.shield(self._close_task)
+
+    async def _close(self) -> None:
+        if self._auth_task is not None and not self._auth_task.done():
+            self._auth_cancelled_for_close = True
+            self._auth_task.cancel()
+            await asyncio.gather(self._auth_task, return_exceptions=True)
+        await self._event_source.close()
 
     async def is_connected(self) -> bool:
-        return self._socket_client is not None and await self._socket_client.is_connected()
+        return await self._event_source.is_connected()
 
     async def get_user(self, user_id: str) -> SlackUser:
         """Look up a profile on demand; email requires the users:read.email scope."""
@@ -442,20 +448,3 @@ async def mirror(self, source: Message, message: Message, *, origin: str = "web"
                 "origin": origin,
             }),
         )
-
-    async def _handle_socket_request(
-        self, client: AsyncBaseSocketModeClient, request: SocketModeRequest
-    ) -> None:
-        if request.type != "events_api":
-            return
-        if self._closed.is_set() or len(self._tasks) >= self._max_pending_events:
-            return
-        await client.send_socket_mode_response(SocketModeResponse(envelope_id=request.envelope_id))
-        task = asyncio.create_task(self.handle_event(request.payload))
-        self._tasks.add(task)
-        task.add_done_callback(self._event_done)
-
-    def _event_done(self, task: asyncio.Task[None]) -> None:
-        self._tasks.discard(task)
-        if not task.cancelled() and (error := task.exception()) is not None:
-            logger.warning("Slack handler failed (%s)", type(error).__name__)
diff --git a/src/agentchat/channels/slack/socket_mode.py b/src/agentchat/channels/slack/socket_mode.py
new file mode 100644
index 0000000..b58af55
--- /dev/null
+++ b/src/agentchat/channels/slack/socket_mode.py
@@ -0,0 +1,144 @@
+"""Slack Socket Mode ingress, acknowledgement and bounded event lifetime."""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+from typing import cast
+
+from slack_sdk.socket_mode.aiohttp import SocketModeClient
+from slack_sdk.socket_mode.async_client import AsyncBaseSocketModeClient
+from slack_sdk.socket_mode.request import SocketModeRequest
+from slack_sdk.socket_mode.response import SocketModeResponse
+from slack_sdk.web.async_client import AsyncWebClient
+
+from agentchat.channels.slack.transport import SlackEventReceiver
+
+logger = logging.getLogger(__name__)
+EVENT_DRAIN_TIMEOUT = 20
+
+
+class SocketMode:
+    """An event source that acknowledges Slack envelopes before handler work.
+
+    Pending acknowledgements and handlers share a capacity limit. Overflow is
+    left unacknowledged for Slack to retry; accepted work is drained on close.
+    Instances have a single run/close lifecycle.
+    """
+
+    def __init__(
+        self,
+        *,
+        app_token: str,
+        web_client: AsyncWebClient | None = None,
+        max_pending_events: int = 64,
+    ) -> None:
+        if not app_token:
+            raise ValueError("app_token is required for Socket Mode")
+        if max_pending_events < 1:
+            raise ValueError("max_pending_events must be positive")
+        self._app_token = app_token
+        self._web_client = web_client
+        self._max_pending_events = max_pending_events
+        self._socket_client: SocketModeClient | None = None
+        self._receiver: SlackEventReceiver | None = None
+        self._closed = asyncio.Event()
+        self._tasks: set[asyncio.Task[None]] = set()
+        self._connect_task: asyncio.Task[None] | None = None
+        self._close_task: asyncio.Task[None] | None = None
+        self._run_started = False
+
+    async def run(self, receiver: SlackEventReceiver) -> None:
+        if self._run_started or self._closed.is_set():
+            raise RuntimeError("Socket Mode has already been started or closed")
+        self._run_started = True
+        self._receiver = receiver
+        try:
+            self._socket_client = SocketModeClient(
+                app_token=self._app_token, web_client=self._web_client,
+            )
+            listeners = cast(
+                list[Callable[
+                    [AsyncBaseSocketModeClient, SocketModeRequest], Awaitable[None],
+                ]],
+                self._socket_client.socket_mode_request_listeners,
+            )
+            listeners.append(self._handle_socket_request)
+            connect = cast(Callable[[], Awaitable[None]], self._socket_client.connect)
+            self._connect_task = asyncio.create_task(connect())
+            try:
+                await self._connect_task
+            except asyncio.CancelledError:
+                if self._closed.is_set():
+                    return
+                raise
+            await self._closed.wait()
+        finally:
+            await self.close()
+
+    async def close(self) -> None:
+        if self._close_task is None:
+            self._closed.set()
+            self._close_task = asyncio.create_task(self._close())
+        await asyncio.shield(self._close_task)
+
+    async def _close(self) -> None:
+        if self._connect_task is not None and not self._connect_task.done():
+            self._connect_task.cancel()
+            await asyncio.gather(self._connect_task, return_exceptions=True)
+        try:
+            if self._socket_client is not None:
+                close = cast(Callable[[], Awaitable[None]], self._socket_client.close)
+                await close()
+        finally:
+            if self._tasks:
+                _, pending = await asyncio.wait(self._tasks, timeout=EVENT_DRAIN_TIMEOUT)
+                for task in pending:
+                    task.cancel()
+                await asyncio.gather(*pending, return_exceptions=True)
+
+    async def is_connected(self) -> bool:
+        return (
+            not self._closed.is_set()
+            and self._socket_client is not None
+            and await self._socket_client.is_connected()
+        )
+
+    async def _handle_socket_request(
+        self, client: AsyncBaseSocketModeClient, request: SocketModeRequest,
+    ) -> None:
+        if request.type != "events_api" or self._closed.is_set():
+            return
+        if len(self._tasks) >= self._max_pending_events:
+            return
+        acknowledged: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+        # SDK listener cancellation must not discard a successfully accepted event.
+        acknowledged.add_done_callback(
+            lambda result: result.exception() if not result.cancelled() else None,
+        )
+        task = asyncio.create_task(self._deliver_event(client, request, acknowledged))
+        self._tasks.add(task)
+        task.add_done_callback(self._event_done)
+        await asyncio.shield(acknowledged)
+
+    async def _deliver_event(
+        self,
+        client: AsyncBaseSocketModeClient,
+        request: SocketModeRequest,
+        acknowledged: asyncio.Future[None],
+    ) -> None:
+        try:
+            await client.send_socket_mode_response(
+                SocketModeResponse(envelope_id=request.envelope_id),
+            )
+        except BaseException as error:
+            acknowledged.set_exception(error)
+            raise
+        acknowledged.set_result(None)
+        assert self._receiver is not None
+        await self._receiver(request.payload)
+
+    def _event_done(self, task: asyncio.Task[None]) -> None:
+        self._tasks.discard(task)
+        if not task.cancelled() and (error := task.exception()) is not None:
+            logger.warning("Slack handler failed (%s)", type(error).__name__)
diff --git a/src/agentchat/channels/slack/transport.py b/src/agentchat/channels/slack/transport.py
new file mode 100644
index 0000000..935173f
--- /dev/null
+++ b/src/agentchat/channels/slack/transport.py
@@ -0,0 +1,27 @@
+"""Event-source contract for composing Slack with host-owned transports."""
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Protocol
+
+SlackEventReceiver = Callable[[Mapping[str, object]], Awaitable[None]]
+
+
+class SlackEventSource(Protocol):
+    """Deliver authenticated Events API payloads to the shared Slack adapter.
+
+    ``run`` stays active until shutdown and propagates startup/transport errors.
+    The source owns acknowledgement, backpressure and accepted background work;
+    ``close`` must stop intake and drain or cancel that work within a bound. It
+    must be safe to close before startup or more than once. Host HTTP sources
+    verify Slack signatures before delivery; the receiver does not authenticate
+    raw requests. A source instance is owned by one Slack adapter.
+    """
+
+    async def run(self, receiver: SlackEventReceiver) -> None: ...
+
+    async def close(self) -> None: ...
+
+    async def is_connected(self) -> bool:
+        """Whether the source is currently ready to deliver authenticated events."""
+        ...
diff --git a/src/agentchat/channels/slack_files.py b/src/agentchat/channels/slack_files.py
deleted file mode 100644
index b7a75ff..0000000
--- a/src/agentchat/channels/slack_files.py
+++ /dev/null
@@ -1,6 +0,0 @@
-"""Compatibility exports; use :mod:`agentchat.channels.slack.files` and `.events`."""
-
-from agentchat.channels.slack.events import FILE_ID, slack_attachments
-from agentchat.channels.slack.files import DeliveryGuard, SlackRequest, download_slack_file
-
-__all__ = ["FILE_ID", "DeliveryGuard", "SlackRequest", "download_slack_file", "slack_attachments"]
diff --git a/src/agentchat/channels/slack_media.py b/src/agentchat/channels/slack_media.py
deleted file mode 100644
index 32879f9..0000000
--- a/src/agentchat/channels/slack_media.py
+++ /dev/null
@@ -1,20 +0,0 @@
-"""Compatibility exports; use :mod:`agentchat.channels.slack.files` and `.rendering`."""
-
-from agentchat.channels.slack.files import (
-    DeliveryGuard,
-    SlackRequest,
-    upload_bytes,
-    upload_slack_files,
-    validate_destination,
-)
-from agentchat.channels.slack.rendering import rich_payload, text_blocks
-
-__all__ = [
-    "DeliveryGuard",
-    "SlackRequest",
-    "rich_payload",
-    "text_blocks",
-    "upload_bytes",
-    "upload_slack_files",
-    "validate_destination",
-]
diff --git a/src/agentchat/channels/slack_mirror.py b/src/agentchat/channels/slack_mirror.py
deleted file mode 100644
index fb70800..0000000
--- a/src/agentchat/channels/slack_mirror.py
+++ /dev/null
@@ -1,5 +0,0 @@
-"""Compatibility exports; use :mod:`agentchat.channels.slack.mirroring`."""
-
-from agentchat.channels.slack.mirroring import EVENT_TYPE, mirror_payload, read_mirror
-
-__all__ = ["EVENT_TYPE", "mirror_payload", "read_mirror"]
diff --git a/tests/channels/slack/test_adapter.py b/tests/channels/slack/test_adapter.py
index 27b1c3d..28bcb88 100644
--- a/tests/channels/slack/test_adapter.py
+++ b/tests/channels/slack/test_adapter.py
@@ -2,7 +2,6 @@
 
 import pytest
 from slack_sdk.errors import SlackApiError
-from slack_sdk.socket_mode.request import SocketModeRequest
 
 from agentchat import AgentChat
 from agentchat.channels.slack import Slack
@@ -326,49 +325,6 @@ async def respond(context):
     asyncio.run(scenario())
 
 
-def test_socket_acknowledges_next_event_while_handler_is_running():
-    async def scenario():
-        started, release = asyncio.Event(), asyncio.Event()
-        received = []
-        acknowledgements = []
-        slack = Slack(
-            bot_token="test",
-            app_token="test",
-            web_client=FakeSlackClient(),
-            bot_user_id="BOT",
-            max_pending_events=2,
-        )
-
-        async def receive(channel, message):
-            started.set()
-            await release.wait()
-            received.append(message.id)
-
-        class Socket:
-            async def send_socket_mode_response(self, response):
-                acknowledgements.append(response.envelope_id)
-
-        slack.bind(receive)
-        first = SocketModeRequest(type="events_api", envelope_id="one", payload=channel_event())
-        second = SocketModeRequest(
-            type="events_api", envelope_id="two", payload=channel_event("2.0")
-        )
-        await slack._handle_socket_request(Socket(), first)
-        await asyncio.wait_for(started.wait(), timeout=1)
-        await asyncio.wait_for(slack._handle_socket_request(Socket(), second), timeout=1)
-        assert acknowledgements == ["one", "two"]
-        assert received == []
-        overflow = SocketModeRequest(
-            type="events_api", envelope_id="three", payload=channel_event("3.0")
-        )
-        await slack._handle_socket_request(Socket(), overflow)
-        assert acknowledgements == ["one", "two"]
-        release.set()
-        await slack.close()
-        assert len(received) == 2
-
-    asyncio.run(scenario())
-
 
 def test_startup_verifies_workspace_and_exposes_connection_state(monkeypatch):
     async def scenario():
@@ -393,7 +349,7 @@ async def is_connected(self):
             async def close(self):
                 self.connected = False
 
-        monkeypatch.setattr("agentchat.channels.slack.adapter.SocketModeClient", Socket)
+        monkeypatch.setattr("agentchat.channels.slack.socket_mode.SocketModeClient", Socket)
         client = Client()
         slack = Slack(bot_token="test", app_token="test", web_client=client, workspace_id="team-1")
 
diff --git a/tests/channels/slack/test_event_source.py b/tests/channels/slack/test_event_source.py
new file mode 100644
index 0000000..91bc395
--- /dev/null
+++ b/tests/channels/slack/test_event_source.py
@@ -0,0 +1,329 @@
+"""The same Slack adapter can consume Socket Mode or host-owned event delivery."""
+
+import asyncio
+
+import pytest
+from slack_sdk.socket_mode.request import SocketModeRequest
+
+from agentchat import AgentChat
+from agentchat.channels.slack import Slack
+from tests.support.imports import run_isolated
+from tests.support.slack import FakeSlackClient, channel_event
+
+
+class QueueEventSource:
+    """A host queue implements the protocol without inheriting library classes."""
+
+    def __init__(self):
+        self.queue = asyncio.Queue()
+        self.started = asyncio.Event()
+        self.connected = False
+        self.close_calls = 0
+
+    async def run(self, receiver):
+        self.connected = True
+        self.started.set()
+        try:
+            while (payload := await self.queue.get()) is not None:
+                await receiver(payload)
+        finally:
+            self.connected = False
+
+    async def emit(self, payload):
+        await self.queue.put(payload)
+
+    async def close(self):
+        self.close_calls += 1
+        await self.queue.put(None)
+
+    async def is_connected(self):
+        return self.connected
+
+
+class AuthenticatedClient(FakeSlackClient):
+    async def auth_test(self):
+        return {"team_id": "team-1", "user_id": "BOT"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("delivery", ["queue", "socket"])
+async def test_sources_share_normalization_subscriptions_history_and_replies(delivery, monkeypatch):
+    client = AuthenticatedClient()
+    queue_source = QueueEventSource()
+    started = queue_source.started
+    source_options = {"event_source": queue_source}
+    emit = queue_source.emit
+
+    if delivery == "socket":
+        started = asyncio.Event()
+
+        class Socket:
+            def __init__(self, **kwargs):
+                self.socket_mode_request_listeners = []
+                self.connected = False
+
+            async def connect(self):
+                self.connected = True
+                started.set()
+
+            async def close(self):
+                self.connected = False
+
+            async def is_connected(self):
+                return self.connected
+
+            async def send_socket_mode_response(self, response):
+                pass
+
+        socket = Socket()
+        monkeypatch.setattr(
+            "agentchat.channels.slack.socket_mode.SocketModeClient", lambda **kwargs: socket,
+        )
+        source_options = {"app_token": "xapp-test"}
+
+        async def emit(payload):
+            request = SocketModeRequest(
+                type="events_api", envelope_id=payload["event_id"], payload=payload,
+            )
+            await socket.socket_mode_request_listeners[0](socket, request)
+
+    slack = Slack(bot_token="xoxb-test", web_client=client, **source_options)
+    processed = asyncio.Queue()
+    handle_event = slack.handle_event
+
+    async def track_event(payload):
+        await handle_event(payload)
+        await processed.put(payload["event_id"])
+
+    monkeypatch.setattr(slack, "handle_event", track_event)
+    app = AgentChat(channels=[slack])
+    received = []
+
+    @app.on_message
+    async def respond(context):
+        received.append(context.message)
+        await slack.subscribe(context.message)
+        history = await context.conversation.history()
+        return f"turn {len(history)}: {context.message.text}"
+
+    async def deliver(payload):
+        await emit(payload)
+        assert await asyncio.wait_for(processed.get(), timeout=1) == payload["event_id"]
+
+    assert not await slack.is_connected()
+    run = asyncio.create_task(app.run())
+    try:
+        await asyncio.wait_for(started.wait(), timeout=1)
+        assert await slack.is_connected()
+        followup = channel_event(
+            "2.0", "more detail", type="message", channel_type="channel", thread_ts="1.0",
+        )
+        await deliver(followup)
+        assert not client.posts
+        await deliver(channel_event(text="<@BOT> hello <@TARGET>"))
+        await deliver(channel_event(text="<@BOT> hello <@TARGET>"))
+        await deliver(followup)
+        await deliver(channel_event(
+            "3.0", "unrelated", type="message", channel_type="channel", thread_ts="other",
+        ))
+
+        assert [message.text for message in received] == ["hello <@TARGET>", "more detail"]
+        assert [message.addressed for message in received] == [True, False]
+        assert [post["text"] for post in client.posts] == [
+            "turn 1: hello <@TARGET>", "turn 3: more detail",
+        ]
+        assert [post["thread_ts"] for post in client.posts] == ["1.0", "1.0"]
+        assert {post["channel"] for post in client.posts} == {"channel-1"}
+        history = await app.state.history(received[0].conversation_id)
+        assert [(message.role, message.text) for message in history] == [
+            ("user", "hello <@TARGET>"), ("assistant", "turn 1: hello <@TARGET>"),
+            ("user", "more detail"), ("assistant", "turn 3: more detail"),
+        ]
+    finally:
+        await app.close()
+        await asyncio.wait_for(run, timeout=1)
+
+    assert not await slack.is_connected()
+    if delivery == "queue":
+        assert queue_source.close_calls == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("identity, workspace_id, error", [
+    ({"team_id": "team-1", "user_id": "BOT"}, "other", "different workspace"),
+    ({"team_id": "team-1"}, None, "bot identity"),
+    ({"team_id": 123, "user_id": "BOT"}, None, "bot identity"),
+])
+async def test_authentication_rejects_invalid_identity_before_starting_source(
+    identity, workspace_id, error,
+):
+    class Client(FakeSlackClient):
+        async def auth_test(self):
+            return identity
+
+    source = QueueEventSource()
+    slack = Slack(
+        bot_token="test", event_source=source, web_client=Client(), workspace_id=workspace_id,
+    )
+
+    async def receive(channel, message):
+        raise AssertionError("An unauthenticated source must not deliver messages")
+
+    slack.bind(receive)
+    with pytest.raises(RuntimeError, match=error):
+        await slack.run()
+    assert not source.started.is_set()
+    assert not await slack.is_connected()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("shutdown", ["close", "cancel_run"])
+async def test_stopping_during_authentication_cancels_startup_without_starting_source(shutdown):
+    authenticating, cancelled = asyncio.Event(), asyncio.Event()
+
+    class Client(FakeSlackClient):
+        async def auth_test(self):
+            authenticating.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                cancelled.set()
+
+    source = QueueEventSource()
+    slack = Slack(bot_token="test", event_source=source, web_client=Client())
+
+    async def receive(channel, message):
+        raise AssertionError("A closed adapter must not deliver messages")
+
+    slack.bind(receive)
+    run = asyncio.create_task(slack.run())
+    try:
+        await asyncio.wait_for(authenticating.wait(), timeout=1)
+        if shutdown == "close":
+            await asyncio.wait_for(slack.close(), timeout=1)
+        else:
+            run.cancel()
+        done, pending = await asyncio.wait({run}, timeout=1)
+        assert done == {run} and not pending
+        if shutdown == "close":
+            await run
+        else:
+            with pytest.raises(asyncio.CancelledError):
+                await run
+        assert cancelled.is_set()
+        assert not source.started.is_set()
+        assert source.close_calls == 1
+        await slack.close()
+        assert source.close_calls == 1
+    finally:
+        if not run.done():
+            run.cancel()
+        await asyncio.gather(run, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_authentication_failure_closes_source_and_preserves_native_error():
+    refusal = ConnectionError("Slack authentication unavailable")
+
+    class Client(FakeSlackClient):
+        async def auth_test(self):
+            raise refusal
+
+    source = QueueEventSource()
+    slack = Slack(bot_token="test", event_source=source, web_client=Client())
+
+    async def receive(channel, message):
+        raise AssertionError("An unauthenticated adapter must not deliver messages")
+
+    slack.bind(receive)
+    with pytest.raises(ConnectionError) as caught:
+        await slack.run()
+    assert caught.value is refusal
+    assert not source.started.is_set()
+    assert source.close_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_slack_run_cancellation_closes_the_host_source_and_pending_delivery():
+    source = QueueEventSource()
+    slack = Slack(bot_token="test", event_source=source, web_client=AuthenticatedClient())
+    started, cancelled = asyncio.Event(), asyncio.Event()
+
+    async def receive(channel, message):
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    slack.bind(receive)
+    run = asyncio.create_task(slack.run())
+    await asyncio.wait_for(source.started.wait(), timeout=1)
+    await source.emit(channel_event())
+    await asyncio.wait_for(started.wait(), timeout=1)
+    run.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(run, timeout=1)
+    assert cancelled.is_set()
+    assert not await slack.is_connected()
+    assert source.close_calls == 1
+    await slack.close()
+    assert source.close_calls == 1
+
+
+def test_slack_requires_exactly_one_event_source():
+    with pytest.raises(ValueError, match=r"app_token|event_source"):
+        Slack(bot_token="test", web_client=FakeSlackClient())
+    with pytest.raises(ValueError, match=r"app_token|event_source"):
+        Slack(
+            bot_token="test", app_token="test", event_source=QueueEventSource(),
+            web_client=FakeSlackClient(),
+        )
+
+
+def test_custom_source_does_not_import_socket_mode():
+    run_isolated("""
+        import importlib.abc
+        import sys
+
+        class BlockSocketMode(importlib.abc.MetaPathFinder):
+            def find_spec(self, fullname, path=None, target=None):
+                if fullname.startswith((
+                    "slack_sdk.socket_mode", "agentchat.channels.slack.socket_mode",
+                )):
+                    raise AssertionError("A custom source must not load Socket Mode")
+
+        sys.meta_path.insert(0, BlockSocketMode())
+        from agentchat.channels.slack import Slack
+
+        class Source:
+            async def run(self, receiver): pass
+            async def close(self): pass
+            async def is_connected(self): return False
+
+        Slack(bot_token="test", event_source=Source())
+        assert not any(name.startswith("slack_sdk.socket_mode") for name in sys.modules)
+    """)
+
+
+def test_slack_payload_helpers_and_event_source_contract_do_not_import_sdks():
+    run_isolated("""
+        import importlib.abc
+        import sys
+
+        blocked = {"slack_sdk", "aiohttp"}
+
+        class BlockSDKs(importlib.abc.MetaPathFinder):
+            def find_spec(self, fullname, path=None, target=None):
+                if fullname.split(".")[0] in blocked:
+                    raise AssertionError("Pure Slack modules must not load provider SDKs")
+
+        sys.meta_path.insert(0, BlockSDKs())
+        from agentchat.channels.slack import events, mirroring, rendering, transport
+
+        assert callable(events.to_message)
+        assert callable(mirroring.mirror_payload)
+        assert callable(rendering.text_blocks)
+        assert transport.SlackEventSource
+        assert not blocked.intersection(name.split(".")[0] for name in sys.modules)
+    """)
diff --git a/tests/channels/slack/test_transport.py b/tests/channels/slack/test_transport.py
new file mode 100644
index 0000000..a989502
--- /dev/null
+++ b/tests/channels/slack/test_transport.py
@@ -0,0 +1,258 @@
+"""Socket Mode owns acknowledgement, backpressure and connection cleanup."""
+
+import asyncio
+
+import pytest
+from slack_sdk.socket_mode.request import SocketModeRequest
+
+from agentchat.channels.slack.socket_mode import SocketMode
+from tests.support.slack import channel_event
+
+
+class FakeSocket:
+    def __init__(self):
+        self.socket_mode_request_listeners = []
+        self.started = asyncio.Event()
+        self.closed = asyncio.Event()
+        self.ack_started = asyncio.Event()
+        self.ack_release = None
+        self.ack_error = None
+        self.connect_error = None
+        self.connected = False
+        self.acknowledgements = []
+        self.close_calls = 0
+
+    async def connect(self):
+        if self.connect_error:
+            raise self.connect_error
+        self.connected = True
+        self.started.set()
+
+    async def is_connected(self):
+        return self.connected
+
+    async def close(self):
+        self.connected = False
+        self.close_calls += 1
+        self.closed.set()
+
+    async def send_socket_mode_response(self, response):
+        self.acknowledgements.append(response.envelope_id)
+        self.ack_started.set()
+        if self.ack_release:
+            await self.ack_release.wait()
+        if self.ack_error:
+            raise self.ack_error
+
+    async def emit(self, envelope_id, ts="1.0", request_type="events_api"):
+        request = SocketModeRequest(
+            type=request_type, envelope_id=envelope_id, payload=channel_event(ts),
+        )
+        await self.socket_mode_request_listeners[0](self, request)
+
+
+@pytest.fixture
+def socket(monkeypatch):
+    client = FakeSocket()
+    monkeypatch.setattr(
+        "agentchat.channels.slack.socket_mode.SocketModeClient", lambda **kwargs: client,
+    )
+    return client
+
+
+async def start_transport(socket, receive, *, max_pending_events=64):
+    source = SocketMode(app_token="xapp-test", max_pending_events=max_pending_events)
+    run = asyncio.create_task(source.run(receive))
+    await asyncio.wait_for(socket.started.wait(), timeout=1)
+    return source, run
+
+
+@pytest.mark.asyncio
+async def test_socket_acknowledges_without_waiting_for_handlers_and_drains_on_close(socket):
+    started, release = asyncio.Event(), asyncio.Event()
+    received = []
+
+    async def receive(payload):
+        started.set()
+        await release.wait()
+        received.append(payload["event_id"])
+
+    source, run = await start_transport(socket, receive, max_pending_events=2)
+    try:
+        await socket.emit("one")
+        await asyncio.wait_for(started.wait(), timeout=1)
+        await asyncio.wait_for(socket.emit("two", "2.0"), timeout=1)
+        assert socket.acknowledgements == ["one", "two"]
+        assert not received
+        await socket.emit("overflow", "3.0")
+        await socket.emit("unsupported", request_type="slash_commands")
+        assert socket.acknowledgements == ["one", "two"]
+
+        closing = asyncio.create_task(source.close())
+        await asyncio.wait_for(socket.closed.wait(), timeout=1)
+        await socket.emit("after-close", "4.0")
+        assert socket.acknowledgements == ["one", "two"]
+        release.set()
+        await asyncio.wait_for(closing, timeout=1)
+        assert received == ["event-1.0", "event-2.0"]
+        assert not await source.is_connected()
+    finally:
+        release.set()
+        await source.close()
+        await asyncio.wait_for(run, timeout=1)
+
+    assert socket.close_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_socket_reserves_capacity_before_awaiting_acknowledgement(socket):
+    socket.ack_release = asyncio.Event()
+    processed = asyncio.Event()
+
+    async def receive(payload):
+        processed.set()
+
+    source, run = await start_transport(socket, receive, max_pending_events=1)
+    first = asyncio.create_task(socket.emit("one"))
+    try:
+        await asyncio.wait_for(socket.ack_started.wait(), timeout=1)
+        await asyncio.wait_for(socket.emit("overflow", "2.0"), timeout=1)
+        assert socket.acknowledgements == ["one"]
+        assert not processed.is_set()
+        socket.ack_release.set()
+        await first
+        await asyncio.wait_for(processed.wait(), timeout=1)
+    finally:
+        socket.ack_release.set()
+        await first
+        await source.close()
+        await asyncio.wait_for(run, timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_failed_acknowledgement_releases_capacity_without_running_handler(socket):
+    processed = asyncio.Event()
+    received = []
+
+    async def receive(payload):
+        received.append(payload["event_id"])
+        processed.set()
+
+    source, run = await start_transport(socket, receive, max_pending_events=1)
+    try:
+        socket.ack_error = TimeoutError("acknowledgement failed")
+        with pytest.raises(TimeoutError, match="acknowledgement failed"):
+            await socket.emit("failed")
+        assert not received
+        socket.ack_error = None
+        await socket.emit("retry")
+        await asyncio.wait_for(processed.wait(), timeout=1)
+        assert received == ["event-1.0"]
+    finally:
+        await source.close()
+        await asyncio.wait_for(run, timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_listener_cancellation_does_not_discard_an_accepted_event(socket):
+    socket.ack_release = asyncio.Event()
+    processed = asyncio.Event()
+
+    async def receive(payload):
+        processed.set()
+
+    source, run = await start_transport(socket, receive)
+    listener = asyncio.create_task(socket.emit("one"))
+    try:
+        await asyncio.wait_for(socket.ack_started.wait(), timeout=1)
+        listener.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await listener
+        socket.ack_release.set()
+        await asyncio.wait_for(processed.wait(), timeout=1)
+        assert socket.acknowledgements == ["one"]
+    finally:
+        socket.ack_release.set()
+        await source.close()
+        await asyncio.wait_for(run, timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_connection_failure_closes_socket_and_preserves_the_error(socket):
+    refusal = ConnectionError("socket unavailable")
+    socket.connect_error = refusal
+    source = SocketMode(app_token="xapp-test")
+
+    async def receive(payload):
+        raise AssertionError("A failed connection cannot deliver events")
+
+    with pytest.raises(ConnectionError) as caught:
+        await source.run(receive)
+    assert caught.value is refusal
+    assert socket.close_calls == 1
+    assert not await source.is_connected()
+    await source.close()
+    assert socket.close_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_close_during_connection_startup_cancels_connection(socket, monkeypatch):
+    connecting, cancelled = asyncio.Event(), asyncio.Event()
+
+    async def connect():
+        connecting.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    monkeypatch.setattr(socket, "connect", connect)
+    source = SocketMode(app_token="xapp-test")
+
+    async def receive(payload):
+        raise AssertionError("A closed connection cannot deliver events")
+
+    run = asyncio.create_task(source.run(receive))
+    try:
+        await asyncio.wait_for(connecting.wait(), timeout=1)
+        await asyncio.wait_for(source.close(), timeout=1)
+        await asyncio.wait_for(run, timeout=1)
+        assert cancelled.is_set()
+        assert not await source.is_connected()
+        assert socket.close_calls == 1
+    finally:
+        await source.close()
+        await asyncio.wait_for(run, timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_run_cancellation_closes_connection_and_cancels_unfinished_handlers(
+    socket, monkeypatch,
+):
+    monkeypatch.setattr("agentchat.channels.slack.socket_mode.EVENT_DRAIN_TIMEOUT", 0)
+    started, cancelled = asyncio.Event(), asyncio.Event()
+
+    async def receive(payload):
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    source, run = await start_transport(socket, receive)
+    await socket.emit("one")
+    await asyncio.wait_for(started.wait(), timeout=1)
+    run.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(run, timeout=1)
+    assert cancelled.is_set()
+    assert socket.close_calls == 1
+    assert not await source.is_connected()
+    await source.close()
+    assert socket.close_calls == 1
+
+
+@pytest.mark.parametrize("capacity", [0, -1])
+def test_socket_rejects_nonpositive_pending_capacity(capacity):
+    with pytest.raises(ValueError, match="positive"):
+        SocketMode(app_token="xapp-test", max_pending_events=capacity)
diff --git a/tests/support/imports.py b/tests/support/imports.py
new file mode 100644
index 0000000..baabdf3
--- /dev/null
+++ b/tests/support/imports.py
@@ -0,0 +1,13 @@
+"""Run import checks without modules cached by other tests."""
+
+import subprocess
+import sys
+import textwrap
+
+
+def run_isolated(code: str) -> None:
+    result = subprocess.run(
+        [sys.executable, "-c", textwrap.dedent(code)],
+        capture_output=True, text=True, check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
diff --git a/tests/test_imports.py b/tests/test_imports.py
index 5d1dfd0..cab3ba2 100644
--- a/tests/test_imports.py
+++ b/tests/test_imports.py
@@ -1,16 +1,9 @@
-import subprocess
-import sys
 import textwrap
+from pathlib import Path
 
 import pytest
 
-
-def run_isolated(code: str) -> None:
-    result = subprocess.run(
-        [sys.executable, "-c", textwrap.dedent(code)],
-        capture_output=True, text=True, check=False,
-    )
-    assert result.returncode == 0, result.stdout + result.stderr
+from tests.support.imports import run_isolated
 
 
 def test_core_and_namespaces_import_without_loading_provider_dependencies():
@@ -39,7 +32,8 @@ def find_spec(self, fullname, path=None, target=None):
         assert agentchat.Conversation is Conversation
         assert agentchat.channels.Channel is Channel
         assert MemoryState().messages == {}
-        assert "Slack" in dir(agentchat.channels)
+        assert "Slack" not in dir(agentchat.channels)
+        assert "SlackUser" not in dir(agentchat.channels)
         assert "should_reply" in dir(agentchat.integrations)
         assert not blocked.intersection(name.split(".")[0] for name in sys.modules)
         assert "agentchat.channels.slack" not in sys.modules
@@ -69,36 +63,109 @@ def test_context_import_paths_keep_the_same_types(first_import):
     """))
 
 
-def test_provider_and_integration_exports_keep_the_same_objects():
-    from agentchat.channels import Slack, SlackUser
-    from agentchat.channels.slack import Slack as SlackAdapter
-    from agentchat.channels.slack import SlackUser as SlackProfile
+def test_slack_exports_live_in_the_slack_package():
+    from agentchat.channels.slack import Slack, SlackUser
+    from agentchat.channels.slack.adapter import Slack as SlackAdapter
+    from agentchat.channels.slack.adapter import SlackUser as SlackProfile
+
+    assert Slack is SlackAdapter
+    assert SlackUser is SlackProfile
+
+
+def test_integration_exports_keep_the_same_objects():
     from agentchat.integrations import should_reply, to_openai_input
     from agentchat.integrations.openai_agents import should_reply as openai_should_reply
     from agentchat.integrations.openai_agents import to_openai_input as openai_input
 
-    assert Slack is SlackAdapter
-    assert SlackUser is SlackProfile
     assert should_reply is openai_should_reply
     assert to_openai_input is openai_input
 
 
-def test_documented_slack_helper_imports_keep_the_same_callables():
-    from agentchat.channels import slack_files, slack_media, slack_mirror
-    from agentchat.channels.slack import events, files, mirroring, rendering
+def test_slack_helpers_are_importable_from_their_own_package():
+    from agentchat.channels.slack.events import slack_attachments
+    from agentchat.channels.slack.files import download_slack_file, upload_slack_files
+    from agentchat.channels.slack.mirroring import mirror_payload, read_mirror
+    from agentchat.channels.slack.rendering import rich_payload, text_blocks
+
+    for helper in (
+        slack_attachments, download_slack_file, upload_slack_files,
+        mirror_payload, read_mirror, rich_payload, text_blocks,
+    ):
+        assert callable(helper)
+        assert helper.__module__.startswith("agentchat.channels.slack.")
+
+
+def test_channels_root_contains_only_shared_adapter_contracts():
+    import agentchat.channels
+    from agentchat.channels import base
+
+    channel_directory = Path(agentchat.channels.__file__).parent
+    files = {path.name for path in channel_directory.iterdir() if path.is_file()}
+    allowed = {"__init__.py", "base.py"}
+    assert files == allowed, (
+        "Provider files belong in channels/<provider>/, not the shared channel root: "
+        f"{sorted(files - allowed)}"
+    )
+    for package in channel_directory.iterdir():
+        if package.is_dir() and package.name != "__pycache__":
+            assert (package / "__init__.py").is_file(), f"Expected adapter package: {package.name}"
+    assert all(
+        getattr(agentchat.channels, name) is getattr(base, name)
+        for name in agentchat.channels.__all__
+    )
+
+
+@pytest.mark.parametrize("legacy_module", ["slack_files", "slack_media", "slack_mirror"])
+def test_removed_slack_helper_paths_require_explicit_migration(legacy_module):
+    """Old paths deliberately break rather than leaving shims or dynamic aliases."""
+    run_isolated(f"""
+        import importlib
+
+        module_name = "agentchat.channels.{legacy_module}"
+        try:
+            importlib.import_module(module_name)
+        except ModuleNotFoundError as exc:
+            assert exc.name == module_name
+        else:
+            raise AssertionError("Slack helpers must live inside the Slack package")
+    """)
+
 
-    assert slack_files.slack_attachments is events.slack_attachments
-    assert slack_files.download_slack_file is files.download_slack_file
-    assert slack_media.upload_slack_files is files.upload_slack_files
-    assert slack_media.rich_payload is rendering.rich_payload
-    assert slack_media.text_blocks is rendering.text_blocks
-    assert slack_mirror.mirror_payload is mirroring.mirror_payload
-    assert slack_mirror.read_mirror is mirroring.read_mirror
+@pytest.mark.parametrize("legacy_export", ["Slack", "SlackUser"])
+def test_channels_root_no_longer_exports_provider_types(legacy_export):
+    run_isolated(f"""
+        try:
+            from agentchat.channels import {legacy_export}
+        except ImportError:
+            pass
+        else:
+            raise AssertionError("Import provider types from agentchat.channels.slack")
+    """)
+
+
+@pytest.mark.parametrize("dependency", ["slack_sdk", "aiohttp"])
+def test_canonical_slack_import_reports_missing_dependencies(dependency):
+    run_isolated(f"""
+        import importlib.abc
+        import sys
+
+        class MissingDependency(importlib.abc.MetaPathFinder):
+            def find_spec(self, fullname, path=None, target=None):
+                if fullname == {dependency!r}:
+                    raise ModuleNotFoundError("Dependency unavailable", name=fullname)
+
+        sys.meta_path.insert(0, MissingDependency())
+
+        try:
+            from agentchat.channels.slack import Slack
+        except ModuleNotFoundError as exc:
+            assert exc.name == {dependency!r}
+        else:
+            raise AssertionError("The requested provider dependency was not imported")
+    """)
 
 
 @pytest.mark.parametrize(("statement", "dependency", "installation"), [
-    ("from agentchat.channels import Slack", "slack_sdk", "pip install agentchat"),
-    ("from agentchat.channels import SlackUser", "aiohttp", "pip install agentchat"),
     ("from agentchat.integrations import should_reply", "agents", "agentchat[openai]"),
     ("from agentchat.integrations import to_openai_input", "agents", "agentchat[openai]"),
 ])
@@ -125,7 +192,7 @@ def find_spec(self, fullname, path=None, target=None):
 
 
 @pytest.mark.parametrize(("statement", "provider_module"), [
-    ("from agentchat.channels import Slack", "agentchat.channels.slack"),
+    ("from agentchat.channels.slack import Slack", "agentchat.channels.slack"),
     ("from agentchat.integrations import should_reply", "agentchat.integrations.openai_agents"),
 ])
 def test_provider_exports_preserve_unrelated_missing_dependencies(statement, provider_module):