diff --git a/CHANGELOG.md b/CHANGELOG.md
index c6f415b..f72e6fe 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,17 @@ Parity catch-up with upstream `4.26.0`. No upstream version change.
   who've posted in the thread. Seeds from `current_message.author` (if
   eligible), then iterates `all_messages()` and dedupes by `user_id`.
   Mirrors upstream TS `Thread.getParticipants()`. Issue #54.
+- **`Chat.on_options_load(...)` + `Chat.process_options_load(...)`**: port of
+  upstream `onOptionsLoad` / `processOptionsLoad` for handling
+  external-select option-load events. Specific action IDs run before
+  catch-all handlers; errors are logged and skipped so later handlers still
+  get a chance. New public types: `OptionsLoadEvent`, `OptionsLoadHandler`.
+- **Slack `block_suggestion` dispatch**: the Slack adapter now routes
+  `block_suggestion` interactive payloads through `process_options_load`
+  and serializes the result to a Slack options JSON response. The handler
+  is raced against a 2.5s budget (`OPTIONS_LOAD_TIMEOUT_MS`); on timeout
+  the response is empty options and the orphaned task still logs errors
+  via `asyncio.shield`. Issue #50.
 - **`IoRedisStateAdapter`**: `RedisStateAdapter` subclass defaulting to the
   `ioredis_` lock-token prefix used by upstream Vercel Chat's `ioredis`-backed
   state. Enables cross-runtime Redis sharing between TS and Python chat-sdk
diff --git a/src/chat_sdk/__init__.py b/src/chat_sdk/__init__.py
index d911a99..b3ca66e 100644
--- a/src/chat_sdk/__init__.py
+++ b/src/chat_sdk/__init__.py
@@ -49,7 +49,7 @@
     text_element,
 )
 from chat_sdk.channel import ChannelImpl
-from chat_sdk.chat import Chat
+from chat_sdk.chat import Chat, OptionsLoadHandler
 from chat_sdk.emoji import (
     EmojiResolver,
     convert_emoji_placeholders,
@@ -155,6 +155,7 @@
     ModalCloseEvent,
     ModalResponse,
     ModalSubmitEvent,
+    OptionsLoadEvent,
     PlanUpdateChunk,
     Postable,
     PostableAst,
@@ -356,6 +357,8 @@
     "ModalCloseEvent",
     "ModalResponse",
     "ModalSubmitEvent",
+    "OptionsLoadEvent",
+    "OptionsLoadHandler",
     "PlanUpdateChunk",
     "Postable",
     "PostableAst",
diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py
index 85d12e2..6494fde 100644
--- a/src/chat_sdk/adapters/slack/adapter.py
+++ b/src/chat_sdk/adapters/slack/adapter.py
@@ -52,7 +52,7 @@
 )
 from chat_sdk.emoji import convert_emoji_placeholders, emoji_to_slack, resolve_emoji_from_slack
 from chat_sdk.logger import ConsoleLogger, Logger
-from chat_sdk.modals import ModalElement
+from chat_sdk.modals import ModalElement, SelectOptionElement
 from chat_sdk.shared.adapter_utils import extract_card, extract_files
 from chat_sdk.shared.errors import AdapterRateLimitError, AuthenticationError, ValidationError
 from chat_sdk.types import (
@@ -82,6 +82,7 @@
     ModalCloseEvent,
     ModalResponse,
     ModalSubmitEvent,
+    OptionsLoadEvent,
     RawMessage,
     ReactionEvent,
     ScheduledMessage,
@@ -93,6 +94,10 @@
     WebhookOptions,
 )
 
+# Slack expects block_suggestion responses within 3s. Leave headroom for
+# network latency so the HTTP response lands before Slack gives up.
+OPTIONS_LOAD_TIMEOUT_MS = 2500
+
 # Strong-reference set for fire-and-forget tasks to prevent GC collection.
 _background_tasks: set[asyncio.Task[Any]] = set()
 
@@ -881,6 +886,8 @@ async def _handle_interactive_payload(self, body: str, options: WebhookOptions |
         if payload_type == "block_actions":
             self._handle_block_actions(payload, options)
             return {"body": "", "status": 200}
+        elif payload_type == "block_suggestion":
+            return await self._handle_block_suggestion(payload, options)
         elif payload_type == "view_submission":
             return await self._handle_view_submission(payload, options)
         elif payload_type == "view_closed":
@@ -1001,6 +1008,111 @@ def _handle_block_actions(self, payload: dict[str, Any], options: WebhookOptions
             )
             self._chat.process_action(action_event, options)
 
+    # ==================================================================
+    # Block suggestion (external-select options load)
+    # ==================================================================
+
+    async def _handle_block_suggestion(
+        self, payload: dict[str, Any], options: WebhookOptions | None = None
+    ) -> dict[str, Any]:
+        """Handle a Slack block_suggestion interactive payload.
+
+        Slack requires a response within 3s for block_suggestion and does not
+        support an async ack pattern — options must be in the response body.
+        Race the handler against a 2.5s budget and fall back to an empty 200
+        so the menu shows "No results" instead of hanging for the user.
+        """
+        if not self._chat:
+            self._logger.warn("Chat instance not initialized, ignoring block suggestion")
+            return self._options_load_response([])
+
+        user_ref = payload.get("user") or {}
+        user_id = user_ref.get("id", "")
+        username = user_ref.get("username")
+        name = user_ref.get("name")
+        # Upstream uses `||` truthy-fallthrough intentionally: an empty-string
+        # username falls through to name, then user_id. See upstream
+        # packages/adapter-slack/src/index.ts lines ~1258-1260.
+        user_name = username or name or user_id
+        full_name = name or username or user_id
+
+        action_id = payload.get("action_id", "")
+        val = payload.get("value")
+        event = OptionsLoadEvent(
+            action_id=action_id,
+            query=val if val is not None else "",
+            user=Author(
+                user_id=user_id,
+                user_name=user_name,
+                full_name=full_name,
+                is_bot=False,
+                is_me=False,
+            ),
+            adapter=self,
+            raw=payload,
+        )
+
+        # Use asyncio.shield so the orphaned task still runs (and logs errors)
+        # if we time out. `wait_for` cancels the awaitable on timeout; shielding
+        # prevents that cancellation from propagating into the handler task.
+        load_task = asyncio.ensure_future(self._chat.process_options_load(event, options))
+
+        try:
+            result = await asyncio.wait_for(asyncio.shield(load_task), timeout=OPTIONS_LOAD_TIMEOUT_MS / 1000.0)
+        except asyncio.TimeoutError:
+            self._logger.warn(
+                "Options load handler timed out",
+                {"action_id": action_id, "timeout_ms": OPTIONS_LOAD_TIMEOUT_MS},
+            )
+
+            def _late_error(t: asyncio.Task[Any]) -> None:
+                if t.cancelled():
+                    return
+                exc = t.exception()
+                if exc is not None:
+                    self._logger.error(
+                        "Options load handler error after timeout",
+                        {"action_id": action_id, "error": str(exc)},
+                    )
+
+            load_task.add_done_callback(_late_error)
+            _pin_task(load_task)
+            # Register with wait_until so serverless/webhook runtimes
+            # (e.g. Vercel) keep the task alive past the HTTP response;
+            # otherwise the late-error logging path above can be killed
+            # before it runs. wait_until is user/runtime-provided, so
+            # guard against it raising — we still want to return the
+            # empty-options HTTP 200 fallback.
+            if options and options.wait_until:
+                try:
+                    options.wait_until(load_task)
+                except Exception as err:
+                    self._logger.warn(
+                        "wait_until raised while registering timed-out options load task",
+                        {"action_id": action_id, "error": str(err)},
+                    )
+            return self._options_load_response([])
+
+        return self._options_load_response(result if result is not None else [])
+
+    def _options_load_response(self, options_list: list[SelectOptionElement]) -> dict[str, Any]:
+        """Serialize ``SelectOptionElement`` entries to a Slack JSON response."""
+        slack_options: list[dict[str, Any]] = []
+        for opt in options_list[:100]:
+            entry: dict[str, Any] = {
+                "text": {"type": "plain_text", "text": opt.get("label", "")},
+                "value": opt.get("value", ""),
+            }
+            desc = opt.get("description")
+            if desc:
+                entry["description"] = {"type": "plain_text", "text": desc}
+            slack_options.append(entry)
+        return {
+            "body": json.dumps({"options": slack_options}),
+            "status": 200,
+            "headers": {"Content-Type": "application/json"},
+        }
+
     # ==================================================================
     # View submission / close
     # ==================================================================
diff --git a/src/chat_sdk/chat.py b/src/chat_sdk/chat.py
index aaac6c6..d4a7b30 100644
--- a/src/chat_sdk/chat.py
+++ b/src/chat_sdk/chat.py
@@ -21,6 +21,7 @@
 from chat_sdk.channel import ChannelImpl, _ChannelImplConfigWithAdapter
 from chat_sdk.errors import ChatError, LockError
 from chat_sdk.logger import ConsoleLogger, Logger
+from chat_sdk.modals import SelectOptionElement
 from chat_sdk.thread import (
     ThreadImpl,
     _active_chat,
@@ -53,6 +54,7 @@
     ModalResponse,
     ModalSubmitEvent,
     OnLockConflict,
+    OptionsLoadEvent,
     QueueEntry,
     ReactionEvent,
     SlashCommandEvent,
@@ -82,6 +84,10 @@
 SubscribedMessageHandler = Callable[[Any, Message, Any], Awaitable[None] | None]
 ReactionHandler = Callable[[ReactionEvent], Any]
 ActionHandler = Callable[[ActionEvent], Any]
+OptionsLoadHandler = Callable[
+    [OptionsLoadEvent],
+    Awaitable[list[SelectOptionElement] | None] | list[SelectOptionElement] | None,
+]
 ModalSubmitHandler = Callable[[ModalSubmitEvent], Any]
 ModalCloseHandler = Callable[[ModalCloseEvent], Any]
 SlashCommandHandler = Callable[[SlashCommandEvent], Any]
@@ -121,6 +127,14 @@ def __init__(self, action_ids: list[str], handler: ActionHandler) -> None:
         self.handler = handler
 
 
+class _OptionsLoadPattern:
+    __slots__ = ("action_ids", "handler")
+
+    def __init__(self, action_ids: list[str], handler: OptionsLoadHandler) -> None:
+        self.action_ids = action_ids
+        self.handler = handler
+
+
 class _ModalSubmitPattern:
     __slots__ = ("callback_ids", "handler")
 
@@ -358,6 +372,7 @@ def __init__(self, config: ChatConfig | None = None, **kwargs: Any) -> None:
         self._subscribed_message_handlers: list[SubscribedMessageHandler] = []
         self._reaction_handlers: list[_ReactionPattern] = []
         self._action_handlers: list[_ActionPattern] = []
+        self._options_load_handlers: list[_OptionsLoadPattern] = []
         self._modal_submit_handlers: list[_ModalSubmitPattern] = []
         self._modal_close_handlers: list[_ModalClosePattern] = []
         self._slash_command_handlers: list[_SlashCommandPattern] = []
@@ -654,6 +669,47 @@ def decorator(h: ActionHandler) -> ActionHandler:
 
         return decorator
 
+    # -- Options load ---
+
+    def on_options_load(
+        self,
+        action_ids_or_handler: str | list[str] | OptionsLoadHandler | None = None,
+        handler: OptionsLoadHandler | None = None,
+    ) -> OptionsLoadHandler | Callable[[OptionsLoadHandler], OptionsLoadHandler]:
+        """Register a handler for loading dynamic options for external selects.
+
+        Specific action IDs run before catch-all handlers.
+
+        Overloaded:
+        - ``chat.on_options_load(handler)`` -- all selects
+        - ``chat.on_options_load("id", handler)``
+        - ``chat.on_options_load(["id1", "id2"], handler)``
+        - Decorator: ``@chat.on_options_load("id")``
+        """
+        if callable(action_ids_or_handler) and handler is None:
+            self._options_load_handlers.append(_OptionsLoadPattern([], action_ids_or_handler))
+            self._logger.debug("Registered options load handler for all action IDs")
+            return action_ids_or_handler
+
+        if isinstance(action_ids_or_handler, (str, list)) and handler is not None:
+            ids = [action_ids_or_handler] if isinstance(action_ids_or_handler, str) else action_ids_or_handler
+            self._options_load_handlers.append(_OptionsLoadPattern(ids, handler))
+            self._logger.debug("Registered options load handler", {"action_ids": ids})
+            return handler
+
+        # Decorator form
+        ids = (
+            [action_ids_or_handler]
+            if isinstance(action_ids_or_handler, str)
+            else (action_ids_or_handler if isinstance(action_ids_or_handler, list) else [])
+        )
+
+        def decorator(h: OptionsLoadHandler) -> OptionsLoadHandler:
+            self._options_load_handlers.append(_OptionsLoadPattern(ids, h))
+            return h
+
+        return decorator
+
     # -- Modal submit ---
 
     def on_modal_submit(
@@ -892,6 +948,37 @@ def process_action(
             if options and options.wait_until:
                 options.wait_until(task)
 
+    async def process_options_load(
+        self,
+        event: OptionsLoadEvent,
+        options: WebhookOptions | None = None,  # noqa: ARG002 (match upstream signature)
+    ) -> list[SelectOptionElement] | None:
+        """Process an options-load event (external-select suggestion lookup).
+
+        Runs specific-action-ID handlers before catch-all handlers and returns
+        the first handler result that isn't ``None`` — including an explicit
+        ``[]``, which short-circuits subsequent handlers (handler says "I
+        handled this action, show no options"). Errors are logged and skipped
+        so later handlers still get a chance. Mirrors upstream
+        ``processOptionsLoad`` (TS ``if (options) { return options; }``, where
+        ``[]`` is truthy and therefore short-circuits).
+        """
+        matching_handlers = [
+            pat for pat in self._options_load_handlers if pat.action_ids and event.action_id in pat.action_ids
+        ] + [pat for pat in self._options_load_handlers if not pat.action_ids]
+
+        for pat in matching_handlers:
+            try:
+                result = await self._invoke_handler(pat.handler, event)
+                if result is not None:
+                    return result
+            except Exception as exc:
+                self._logger.error(
+                    "Options load handler error",
+                    {"action_id": event.action_id, "error": str(exc)},
+                )
+        return None
+
     async def process_modal_submit(
         self,
         event: ModalSubmitEvent,
diff --git a/src/chat_sdk/types.py b/src/chat_sdk/types.py
index 0462b07..5c4c6f1 100644
--- a/src/chat_sdk/types.py
+++ b/src/chat_sdk/types.py
@@ -19,6 +19,7 @@
 from chat_sdk.cards import CardElement
 from chat_sdk.errors import ChatNotImplementedError
 from chat_sdk.logger import Logger, LogLevel
+from chat_sdk.modals import SelectOptionElement
 
 
 def _parse_iso(s: str) -> datetime:
@@ -987,6 +988,22 @@ async def open_modal(self, modal: Any) -> dict[str, str] | None:
         return None
 
 
+@dataclass
+class OptionsLoadEvent:
+    """Event emitted when an adapter needs dynamic options for an external select.
+
+    Port of upstream TS ``OptionsLoadEvent``. Slack dispatches a
+    ``block_suggestion`` payload to populate an external-select menu; the
+    handler returns the matching options for the current query text.
+    """
+
+    action_id: str
+    adapter: Adapter
+    query: str
+    user: Author
+    raw: Any = None
+
+
 @dataclass
 class ModalSubmitEvent:
     """Modal form submission event."""
@@ -1348,6 +1365,9 @@ def process_slash_command(self, event: Any, options: WebhookOptions | None = Non
     def process_modal_submit(
         self, event: Any, context_id: str | None = None, options: WebhookOptions | None = None
     ) -> Awaitable[ModalResponse | None]: ...
+    def process_options_load(
+        self, event: OptionsLoadEvent, options: WebhookOptions | None = None
+    ) -> Awaitable[list[SelectOptionElement] | None]: ...
     def process_modal_close(
         self, event: Any, context_id: str | None = None, options: WebhookOptions | None = None
     ) -> None: ...
diff --git a/tests/test_chat_faithful.py b/tests/test_chat_faithful.py
index 5d0f77f..d61e1e1 100644
--- a/tests/test_chat_faithful.py
+++ b/tests/test_chat_faithful.py
@@ -1242,6 +1242,148 @@ async def _handler(event):
         assert stored.get("thread") is None
 
 
+# ============================================================================
+# 12b. Options Load (4 tests)
+# ============================================================================
+
+
+class TestOptionsLoad:
+    # TS: "should call onOptionsLoad handler for a matching action ID"
+    async def test_should_call_onoptionsload_handler_for_a_matching_action_id(self):
+        from chat_sdk.types import OptionsLoadEvent
+
+        chat, adapter, state = await _init_chat()
+        captured: list[OptionsLoadEvent] = []
+
+        async def _handler(event: OptionsLoadEvent):
+            captured.append(event)
+            return [{"label": "Maria Garcia", "value": "person_123"}]
+
+        chat.on_options_load("person_select", _handler)
+
+        event = OptionsLoadEvent(
+            action_id="person_select",
+            query="mar",
+            user=_make_author(),
+            adapter=adapter,
+            raw={},
+        )
+        options = await chat.process_options_load(event)
+
+        assert len(captured) == 1
+        assert captured[0].action_id == "person_select"
+        assert captured[0].query == "mar"
+        assert options == [{"label": "Maria Garcia", "value": "person_123"}]
+
+    # TS: "should prefer specific handlers before catch-all handlers"
+    async def test_should_prefer_specific_handlers_before_catchall_handlers(self):
+        from chat_sdk.types import OptionsLoadEvent
+
+        chat, adapter, state = await _init_chat()
+        catchall_calls = 0
+        specific_calls = 0
+
+        async def _catchall(event: OptionsLoadEvent):
+            nonlocal catchall_calls
+            catchall_calls += 1
+            return [{"label": "Fallback", "value": "fallback"}]
+
+        async def _specific(event: OptionsLoadEvent):
+            nonlocal specific_calls
+            specific_calls += 1
+            return [{"label": "Specific", "value": "specific"}]
+
+        chat.on_options_load(_catchall)
+        chat.on_options_load("person_select", _specific)
+
+        event = OptionsLoadEvent(
+            action_id="person_select",
+            query="mar",
+            user=_make_author(),
+            adapter=adapter,
+            raw={},
+        )
+        options = await chat.process_options_load(event)
+
+        assert specific_calls == 1
+        assert catchall_calls == 0
+        assert options == [{"label": "Specific", "value": "specific"}]
+
+    # TS: "should fall back to catch-all handlers when no specific handler matches"
+    async def test_should_fall_back_to_catchall_handlers_when_no_specific_handler_matches(self):
+        from chat_sdk.types import OptionsLoadEvent
+
+        chat, adapter, state = await _init_chat()
+        catchall_calls = 0
+
+        async def _catchall(event: OptionsLoadEvent):
+            nonlocal catchall_calls
+            catchall_calls += 1
+            return [{"label": "Fallback", "value": "fallback"}]
+
+        chat.on_options_load(_catchall)
+
+        event = OptionsLoadEvent(
+            action_id="unknown_select",
+            query="test",
+            user=_make_author(),
+            adapter=adapter,
+            raw={},
+        )
+        options = await chat.process_options_load(event)
+
+        assert catchall_calls == 1
+        assert options == [{"label": "Fallback", "value": "fallback"}]
+
+    # TS: "should continue after handler errors"
+    async def test_should_continue_after_handler_errors(self):
+        from chat_sdk.types import OptionsLoadEvent
+
+        chat, adapter, state = await _init_chat()
+        logger = chat._logger
+        assert isinstance(logger, MockLogger)
+
+        failing_calls = 0
+        fallback_calls = 0
+
+        async def _failing(event: OptionsLoadEvent):
+            nonlocal failing_calls
+            failing_calls += 1
+            raise RuntimeError("boom")
+
+        async def _fallback(event: OptionsLoadEvent):
+            nonlocal fallback_calls
+            fallback_calls += 1
+            return [{"label": "Recovered", "value": "recovered"}]
+
+        chat.on_options_load("person_select", _failing)
+        chat.on_options_load(_fallback)
+
+        event = OptionsLoadEvent(
+            action_id="person_select",
+            query="mar",
+            user=_make_author(),
+            adapter=adapter,
+            raw={},
+        )
+        options = await chat.process_options_load(event)
+
+        assert failing_calls == 1
+        assert fallback_calls == 1
+        assert options == [{"label": "Recovered", "value": "recovered"}]
+
+        # Error was logged with the exact upstream string and actionId context.
+        error_calls = logger.error.calls
+        assert any(
+            call
+            and call[0] == "Options load handler error"
+            and len(call) > 1
+            and isinstance(call[1], dict)
+            and call[1].get("action_id") == "person_select"
+            for call in error_calls
+        )
+
+
 # ============================================================================
 # 13. openDM (tests 51-54)
 # ============================================================================
diff --git a/tests/test_options_load.py b/tests/test_options_load.py
new file mode 100644
index 0000000..edd0916
--- /dev/null
+++ b/tests/test_options_load.py
@@ -0,0 +1,329 @@
+"""Slack adapter tests for ``block_suggestion`` dispatch.
+
+Ports the two Slack adapter-level options-load tests from upstream
+``packages/adapter-slack/src/index.test.ts`` (lines 703, 750). The
+chat-orchestrator tests live alongside the other 1:1 ports in
+``test_chat_faithful.py``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import json
+import time
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from urllib.parse import quote_plus
+
+import pytest
+
+try:
+    from chat_sdk.adapters.slack.adapter import SlackAdapter
+    from chat_sdk.adapters.slack.types import SlackAdapterConfig
+
+    _SLACK_AVAILABLE = True
+except ImportError:
+    _SLACK_AVAILABLE = False
+
+pytestmark = pytest.mark.skipif(not _SLACK_AVAILABLE, reason="Slack adapter import failed")
+
+
+# ---------------------------------------------------------------------------
+# Helpers (mirrors pattern from tests/test_slack_webhook_extended.py)
+# ---------------------------------------------------------------------------
+
+
+def _slack_signature(body: str, secret: str, timestamp: int | None = None) -> tuple[str, str]:
+    ts = str(timestamp or int(time.time()))
+    sig_base = f"v0:{ts}:{body}"
+    sig = "v0=" + hmac.new(secret.encode(), sig_base.encode(), hashlib.sha256).hexdigest()
+    return ts, sig
+
+
+class _FakeRequest:
+    def __init__(self, body: str, headers: dict[str, str] | None = None):
+        self.body = body.encode("utf-8")
+        self.headers = headers or {}
+
+    async def text(self) -> str:
+        return self.body.decode("utf-8")
+
+
+def _make_signed_request(
+    body: str,
+    secret: str = "test-signing-secret",
+    content_type: str = "application/x-www-form-urlencoded",
+) -> _FakeRequest:
+    ts, sig = _slack_signature(body, secret)
+    return _FakeRequest(
+        body,
+        {
+            "x-slack-request-timestamp": ts,
+            "x-slack-signature": sig,
+            "content-type": content_type,
+        },
+    )
+
+
+def _make_mock_chat() -> MagicMock:
+    """Minimal ChatInstance mock for block_suggestion dispatch."""
+    chat = MagicMock()
+    chat.process_message = MagicMock()
+    chat.handle_incoming_message = AsyncMock()
+    chat.process_reaction = MagicMock()
+    chat.process_action = MagicMock()
+    chat.process_modal_submit = AsyncMock()
+    chat.process_modal_close = MagicMock()
+    chat.process_slash_command = MagicMock()
+    chat.process_options_load = AsyncMock()
+    chat.get_state = MagicMock(return_value=MagicMock())
+    chat.get_user_name = MagicMock(return_value="test-bot")
+    chat.get_logger = MagicMock(return_value=MagicMock())
+    return chat
+
+
+def _make_interactive_request(payload: dict[str, Any]) -> _FakeRequest:
+    body = f"payload={quote_plus(json.dumps(payload))}"
+    return _make_signed_request(body)
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestSlackBlockSuggestion:
+    """Port of ``adapter-slack/src/index.test.ts`` line 703 + 750."""
+
+    # TS: "handles block_suggestion payloads and returns options JSON"
+    @pytest.mark.asyncio
+    async def test_handles_block_suggestion_payloads_and_returns_options_json(self):
+        chat = _make_mock_chat()
+        chat.process_options_load = AsyncMock(return_value=[{"label": "Maria Garcia", "value": "person_123"}])
+
+        adapter = SlackAdapter(SlackAdapterConfig(signing_secret="test-signing-secret", bot_token="xoxb-test"))
+        await adapter.initialize(chat)
+
+        payload = {
+            "type": "block_suggestion",
+            "team": {"id": "T123"},
+            "user": {"id": "U123", "username": "testuser", "name": "Test User"},
+            "action_id": "person_select",
+            "block_id": "person_block",
+            "value": "mar",
+        }
+        req = _make_interactive_request(payload)
+
+        response = await adapter.handle_webhook(req)
+
+        assert response["status"] == 200
+        headers = response.get("headers") or {}
+        content_type = headers.get("Content-Type") or headers.get("content-type") or ""
+        assert "application/json" in content_type
+
+        # Verify the event handed to the chat carried the right fields.
+        chat.process_options_load.assert_awaited_once()
+        call_args, _ = chat.process_options_load.call_args
+        event = call_args[0]
+        assert event.action_id == "person_select"
+        assert event.query == "mar"
+        assert event.user.user_id == "U123"
+
+        parsed = json.loads(response["body"])
+        assert parsed == {
+            "options": [
+                {
+                    "text": {"type": "plain_text", "text": "Maria Garcia"},
+                    "value": "person_123",
+                }
+            ]
+        }
+
+    # TS: "returns empty options when block_suggestion handler exceeds 2.5s budget"
+    @pytest.mark.asyncio
+    async def test_returns_empty_options_when_block_suggestion_handler_exceeds_budget(self):
+        from chat_sdk.types import WebhookOptions
+
+        chat = _make_mock_chat()
+
+        slow_done = asyncio.Event()
+
+        async def _slow_handler(event: Any, options: Any = None):
+            # Handler that runs past the (patched 50ms) budget. The
+            # adapter should time out and return [] before this completes.
+            # Sleep is kept short so the orphaned task can be awaited in
+            # the finally block without lingering and flaking other tests.
+            try:
+                await asyncio.sleep(0.5)
+                return [{"label": "Too late", "value": "late"}]
+            finally:
+                slow_done.set()
+
+        chat.process_options_load = AsyncMock(side_effect=_slow_handler)
+
+        adapter = SlackAdapter(SlackAdapterConfig(signing_secret="test-signing-secret", bot_token="xoxb-test"))
+        await adapter.initialize(chat)
+
+        # Capture the shielded handler task via wait_until so we can
+        # clean it up in finally. Same pattern as
+        # test_timed_out_task_is_registered_with_wait_until.
+        registered: list[Any] = []
+
+        def _wait_until(awaitable: Any) -> None:
+            registered.append(awaitable)
+
+        options = WebhookOptions(wait_until=_wait_until)
+
+        # Patch the module-level timeout so the test runs quickly (real
+        # 2.5s budget is exercised in fidelity-level production code).
+        import chat_sdk.adapters.slack.adapter as slack_adapter_mod
+
+        original_timeout = slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS
+        slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = 50  # 50 ms for test speed
+        try:
+            payload = {
+                "type": "block_suggestion",
+                "team": {"id": "T123"},
+                "user": {"id": "U123", "username": "testuser", "name": "Test User"},
+                "action_id": "person_select",
+                "block_id": "person_block",
+                "value": "mar",
+            }
+            req = _make_interactive_request(payload)
+
+            response = await adapter.handle_webhook(req, options)
+
+            assert response["status"] == 200
+            parsed = json.loads(response["body"])
+            assert parsed == {"options": []}
+
+            # The orphaned handler task must still be running (shielded),
+            # not cancelled — that's how upstream logs late errors.
+            assert not slow_done.is_set()
+        finally:
+            slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = original_timeout
+            # Await the still-running slow task so it doesn't linger
+            # past this test and flake adjacent async tests.
+            if registered:
+                await asyncio.gather(*registered, return_exceptions=True)
+
+    # On timeout, the orphaned handler task must be registered with
+    # WebhookOptions.wait_until so serverless runtimes (e.g. Vercel) keep
+    # it alive until the late-error logging callback fires.
+    @pytest.mark.asyncio
+    async def test_timed_out_task_is_registered_with_wait_until(self):
+        from chat_sdk.types import WebhookOptions
+
+        chat = _make_mock_chat()
+
+        async def _slow_handler(event: Any, options: Any = None):
+            await asyncio.sleep(5.0)
+            return []
+
+        chat.process_options_load = AsyncMock(side_effect=_slow_handler)
+
+        adapter = SlackAdapter(SlackAdapterConfig(signing_secret="test-signing-secret", bot_token="xoxb-test"))
+        await adapter.initialize(chat)
+
+        registered: list[Any] = []
+
+        def _wait_until(awaitable: Any) -> None:
+            registered.append(awaitable)
+
+        options = WebhookOptions(wait_until=_wait_until)
+
+        import chat_sdk.adapters.slack.adapter as slack_adapter_mod
+
+        original_timeout = slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS
+        slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = 50
+        try:
+            payload = {
+                "type": "block_suggestion",
+                "team": {"id": "T123"},
+                "user": {"id": "U123", "username": "testuser", "name": "Test User"},
+                "action_id": "person_select",
+                "block_id": "person_block",
+                "value": "mar",
+            }
+            req = _make_interactive_request(payload)
+
+            response = await adapter.handle_webhook(req, options)
+
+            assert response["status"] == 200
+            parsed = json.loads(response["body"])
+            assert parsed == {"options": []}
+
+            # The timed-out handler task must be handed off to wait_until
+            # so serverless runtimes don't kill it prematurely.
+            assert len(registered) == 1
+            assert isinstance(registered[0], asyncio.Task)
+            assert not registered[0].done()
+        finally:
+            slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = original_timeout
+            # Cancel the still-running slow task so it doesn't leak.
+            if registered and not registered[0].done():
+                registered[0].cancel()
+
+    # If the caller-supplied wait_until raises (e.g. an adapter bug or a
+    # serverless runtime that rejects registration), the timeout branch
+    # must still return the empty-options HTTP 200 fallback rather than
+    # surfacing the exception to the webhook.
+    @pytest.mark.asyncio
+    async def test_timeout_falls_back_when_wait_until_raises(self):
+        from chat_sdk.types import WebhookOptions
+
+        chat = _make_mock_chat()
+
+        slow_task_ref: list[Any] = []
+
+        async def _slow_handler(event: Any, options: Any = None):
+            await asyncio.sleep(5.0)
+            return []
+
+        chat.process_options_load = AsyncMock(side_effect=_slow_handler)
+
+        adapter = SlackAdapter(SlackAdapterConfig(signing_secret="test-signing-secret", bot_token="xoxb-test"))
+        await adapter.initialize(chat)
+
+        def _raising_wait_until(awaitable: Any) -> None:
+            slow_task_ref.append(awaitable)
+            raise RuntimeError("runtime refuses to register background task")
+
+        options = WebhookOptions(wait_until=_raising_wait_until)
+
+        import chat_sdk.adapters.slack.adapter as slack_adapter_mod
+
+        original_timeout = slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS
+        slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = 50
+        try:
+            payload = {
+                "type": "block_suggestion",
+                "team": {"id": "T123"},
+                "user": {"id": "U123", "username": "testuser", "name": "Test User"},
+                "action_id": "person_select",
+                "block_id": "person_block",
+                "value": "mar",
+            }
+            req = _make_interactive_request(payload)
+
+            # Must not raise — the guard should swallow the wait_until error
+            # and still serve the timeout fallback response.
+            response = await adapter.handle_webhook(req, options)
+
+            assert response["status"] == 200
+            headers = response.get("headers") or {}
+            content_type = headers.get("Content-Type") or headers.get("content-type") or ""
+            assert "application/json" in content_type
+            parsed = json.loads(response["body"])
+            assert parsed == {"options": []}
+
+            # Sanity: wait_until was actually invoked (and raised) on the
+            # shielded task — that's what the guard is there to protect.
+            assert len(slow_task_ref) == 1
+            assert isinstance(slow_task_ref[0], asyncio.Task)
+        finally:
+            slack_adapter_mod.OPTIONS_LOAD_TIMEOUT_MS = original_timeout
+            if slow_task_ref and not slow_task_ref[0].done():
+                slow_task_ref[0].cancel()