diff --git a/enterprise/integrations/slack/slack_errors.py b/enterprise/integrations/slack/slack_errors.py
new file mode 100644
index 0000000000..1bf73d0f42
--- /dev/null
+++ b/enterprise/integrations/slack/slack_errors.py
@@ -0,0 +1,128 @@
+"""Centralized error handling for Slack integration.
+
+This module provides:
+- SlackErrorCode: Unique error codes for traceability
+- SlackError: Exception class for user-facing errors
+- get_user_message(): Function to get user-facing messages for error codes
+"""
+
+import logging
+from enum import Enum
+from typing import Any
+
+from integrations.utils import HOST_URL
+
+logger = logging.getLogger(__name__)
+
+
+class SlackErrorCode(Enum):
+    """Unique error codes for traceability in logs and user messages."""
+
+    SESSION_EXPIRED = 'SLACK_ERR_001'
+    REDIS_STORE_FAILED = 'SLACK_ERR_002'
+    REDIS_RETRIEVE_FAILED = 'SLACK_ERR_003'
+    USER_NOT_AUTHENTICATED = 'SLACK_ERR_004'
+    PROVIDER_TIMEOUT = 'SLACK_ERR_005'
+    PROVIDER_AUTH_FAILED = 'SLACK_ERR_006'
+    LLM_AUTH_FAILED = 'SLACK_ERR_007'
+    MISSING_SETTINGS = 'SLACK_ERR_008'
+    UNEXPECTED_ERROR = 'SLACK_ERR_999'
+
+
+class SlackError(Exception):
+    """Exception for errors that should be communicated to the Slack user.
+
+    This exception is caught by the centralized error handler in SlackManager,
+    which logs the error and sends an appropriate message to the user.
+
+    Usage:
+        raise SlackError(SlackErrorCode.USER_NOT_AUTHENTICATED,
+                        message_kwargs={'login_link': link})
+    """
+
+    def __init__(
+        self,
+        code: SlackErrorCode,
+        message_kwargs: dict[str, Any] | None = None,
+        log_context: dict[str, Any] | None = None,
+    ):
+        """Initialize a SlackError.
+
+        Args:
+            code: The error code identifying the type of error
+            message_kwargs: Kwargs for formatting the user message
+                           (e.g., {'login_link': '...'})
+            log_context: Additional context for structured logging
+        """
+        self.code = code
+        self.message_kwargs = message_kwargs or {}
+        self.log_context = log_context or {}
+        super().__init__(f'{code.value}: {code.name}')
+
+    def get_user_message(self) -> str:
+        """Get the user-facing message for this error."""
+        return get_user_message(self.code, **self.message_kwargs)
+
+
+# Centralized user-facing messages
+_USER_MESSAGES: dict[SlackErrorCode, str] = {
+    SlackErrorCode.SESSION_EXPIRED: (
+        '⏰ Your session has expired. '
+        'Please mention me again with your request to start a new conversation.'
+    ),
+    SlackErrorCode.REDIS_STORE_FAILED: (
+        '⚠️ Something went wrong on our end (ref: {code}). '
+        'Please try again in a few moments.'
+    ),
+    SlackErrorCode.REDIS_RETRIEVE_FAILED: (
+        '⚠️ Something went wrong on our end (ref: {code}). '
+        'Please try again in a few moments.'
+    ),
+    SlackErrorCode.USER_NOT_AUTHENTICATED: (
+        '🔐 Please link your Slack account to OpenHands: '
+        '[Click here to Login]({login_link})'
+    ),
+    SlackErrorCode.PROVIDER_TIMEOUT: (
+        '⏱️ The request timed out while connecting to your git provider. '
+        'Please try again.'
+    ),
+    SlackErrorCode.PROVIDER_AUTH_FAILED: (
+        '🔐 Authentication with your git provider failed. '
+        f'Please re-login at [OpenHands Cloud]({HOST_URL}) and try again.'
+    ),
+    SlackErrorCode.LLM_AUTH_FAILED: (
+        '@{username} please set a valid LLM API key in '
+        f'[OpenHands Cloud]({HOST_URL}) before starting a job.'
+    ),
+    SlackErrorCode.MISSING_SETTINGS: (
+        '{username} please re-login into '
+        f'[OpenHands Cloud]({HOST_URL}) before starting a job.'
+    ),
+    SlackErrorCode.UNEXPECTED_ERROR: (
+        'Uh oh! There was an unexpected error (ref: {code}). Please try again later.'
+    ),
+}
+
+
+def get_user_message(error_code: SlackErrorCode, **kwargs) -> str:
+    """Get a user-facing message for a given error code.
+
+    Args:
+        error_code: The error code to get a message for
+        **kwargs: Additional formatting arguments (e.g., username, login_link)
+
+    Returns:
+        Formatted user-facing message string
+    """
+    msg = _USER_MESSAGES.get(
+        error_code, _USER_MESSAGES[SlackErrorCode.UNEXPECTED_ERROR]
+    )
+    try:
+        return msg.format(code=error_code.value, **kwargs)
+    except KeyError as e:
+        logger.warning(
+            f'Missing format key {e} in error message',
+            extra={'error_code': error_code.value},
+        )
+        # Return a generic error message with the code for debugging
+        return f'An error occurred (ref: {error_code.value}). Please try again later.'
diff --git a/enterprise/integrations/slack/slack_manager.py b/enterprise/integrations/slack/slack_manager.py
index 0db105805f..9d7924bfdf 100644
--- a/enterprise/integrations/slack/slack_manager.py
+++ b/enterprise/integrations/slack/slack_manager.py
@@ -1,9 +1,9 @@
-import re
 from typing import Any
 
 import jwt
 from integrations.manager import Manager
 from integrations.models import Message, SourceType
+from integrations.slack.slack_errors import SlackError, SlackErrorCode
 from integrations.slack.slack_types import (
     SlackMessageView,
     SlackViewInterface,
@@ -13,13 +13,13 @@ from integrations.slack.slack_view import (
     SlackFactory,
     SlackNewConversationFromRepoFormView,
     SlackNewConversationView,
-    SlackUnkownUserView,
     SlackUpdateExistingConversationView,
 )
 from integrations.utils import (
     HOST_URL,
     OPENHANDS_RESOLVER_TEMPLATES_DIR,
     get_session_expired_message,
+    infer_repo_from_message,
 )
 from integrations.v1_utils import get_saas_user_auth
 from jinja2 import Environment, FileSystemLoader
@@ -33,8 +33,12 @@ from storage.slack_user import SlackUser
 
 from openhands.core.logger import openhands_logger as logger
 from openhands.integrations.provider import ProviderHandler
-from openhands.integrations.service_types import ProviderTimeoutError, Repository
-from openhands.server.shared import config, server_config
+from openhands.integrations.service_types import (
+    AuthenticationError,
+    ProviderTimeoutError,
+    Repository,
+)
+from openhands.server.shared import config, server_config, sio
 from openhands.server.types import (
     LLMAuthenticationError,
     MissingSettingsError,
@@ -48,6 +52,12 @@ authorize_url_generator = AuthorizeUrlGenerator(
     user_scopes=['search:read'],
 )
 
+# Key prefix for storing user messages in Redis during repo selection flow
+SLACK_USER_MSG_KEY_PREFIX = 'slack_user_msg'
+# Expiration time for stored user messages (5 minutes)
+# Arbitrary timeout based on typical user attention span; may be tuned based on feedback
+SLACK_USER_MSG_EXPIRATION = 300
+
 
 class SlackManager(Manager[SlackViewInterface]):
     def __init__(self, token_manager):
@@ -86,18 +96,126 @@ class SlackManager(Manager[SlackViewInterface]):
 
         return slack_user, saas_user_auth
 
-    def _infer_repo_from_message(self, user_msg: str) -> str | None:
-        # Regular expression to match patterns like "OpenHands/OpenHands" or "deploy repo"
-        pattern = r'([a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+)|([a-zA-Z0-9_-]+)(?=\s+repo)'
-        match = re.search(pattern, user_msg)
+    async def _store_user_msg_for_form(
+        self, message_ts: str, thread_ts: str | None, user_msg: str
+    ) -> None:
+        """Store user message in Redis for later retrieval when form is submitted.
 
-        if match:
-            repo = match.group(1) if match.group(1) else match.group(2)
-            return repo
+        This is needed because when a user selects a repo from the external_select
+        dropdown, Slack sends a separate interaction payload that doesn't include
+        the original user message.
 
-        return None
+        Args:
+            message_ts: The message timestamp (unique identifier)
+            thread_ts: The thread timestamp (if in a thread)
+            user_msg: The original user message to store
 
-    async def _get_repositories(self, user_auth: UserAuth) -> list[Repository]:
+        Raises:
+            SlackError: If storage fails (REDIS_STORE_FAILED)
+        """
+        key = f'{SLACK_USER_MSG_KEY_PREFIX}:{message_ts}:{thread_ts}'
+        try:
+            redis = sio.manager.redis
+            await redis.set(key, user_msg, ex=SLACK_USER_MSG_EXPIRATION)
+            logger.info(
+                'slack_stored_user_msg',
+                extra={
+                    'message_ts': message_ts,
+                    'thread_ts': thread_ts,
+                    'key': key,
+                },
+            )
+        except Exception as e:
+            logger.error(
+                'slack_store_user_msg_failed',
+                extra={
+                    'message_ts': message_ts,
+                    'thread_ts': thread_ts,
+                    'key': key,
+                    'error': str(e),
+                },
+            )
+            raise SlackError(
+                SlackErrorCode.REDIS_STORE_FAILED,
+                log_context={'message_ts': message_ts, 'thread_ts': thread_ts},
+            )
+
+    async def _retrieve_user_msg_for_form(
+        self, message_ts: str, thread_ts: str | None
+    ) -> str:
+        """Retrieve stored user message from Redis.
+
+        Args:
+            message_ts: The message timestamp
+            thread_ts: The thread timestamp (if in a thread)
+
+        Returns:
+            The stored user message
+
+        Raises:
+            SlackError: If retrieval fails (REDIS_RETRIEVE_FAILED) or message
+                        not found (SESSION_EXPIRED)
+        """
+        key = f'{SLACK_USER_MSG_KEY_PREFIX}:{message_ts}:{thread_ts}'
+        try:
+            redis = sio.manager.redis
+            user_msg = await redis.get(key)
+            if user_msg:
+                # Redis returns bytes, decode to string
+                if isinstance(user_msg, bytes):
+                    user_msg = user_msg.decode('utf-8')
+                logger.info(
+                    'slack_retrieved_user_msg',
+                    extra={
+                        'message_ts': message_ts,
+                        'thread_ts': thread_ts,
+                        'key': key,
+                    },
+                )
+                return user_msg
+            else:
+                logger.warning(
+                    'slack_user_msg_not_found',
+                    extra={
+                        'message_ts': message_ts,
+                        'thread_ts': thread_ts,
+                        'key': key,
+                    },
+                )
+                raise SlackError(
+                    SlackErrorCode.SESSION_EXPIRED,
+                    log_context={'message_ts': message_ts, 'thread_ts': thread_ts},
+                )
+        except SlackError:
+            raise
+        except Exception as e:
+            logger.error(
+                'slack_retrieve_user_msg_failed',
+                extra={
+                    'message_ts': message_ts,
+                    'thread_ts': thread_ts,
+                    'key': key,
+                    'error': str(e),
+                },
+            )
+            raise SlackError(
+                SlackErrorCode.REDIS_RETRIEVE_FAILED,
+                log_context={'message_ts': message_ts, 'thread_ts': thread_ts},
+            )
+
+    async def _search_repositories(
+        self, user_auth: UserAuth, query: str = '', per_page: int = 100
+    ) -> list[Repository]:
+        """Search repositories for a user with optional query filtering.
+
+        Args:
+            user_auth: The user's authentication context
+            query: Search query to filter repositories (empty string returns all)
+            per_page: Maximum number of results to return
+
+        Returns:
+            List of matching Repository objects
+        """
         provider_tokens = await user_auth.get_provider_tokens()
         if provider_tokens is None:
             return []
@@ -108,31 +226,33 @@ class SlackManager(Manager[SlackViewInterface]):
             external_auth_token=access_token,
             external_auth_id=user_id,
         )
-        repos: list[Repository] = await client.get_repositories(
-            'pushed', server_config.app_mode, None, None, None, None
+        repos: list[Repository] = await client.search_repositories(
+            selected_provider=None,
+            query=query,
+            per_page=per_page,
+            sort='pushed',
+            order='desc',
+            app_mode=server_config.app_mode,
         )
         return repos
 
     def _generate_repo_selection_form(
-        self, repo_list: list[Repository], message_ts: str, thread_ts: str | None
-    ):
-        options = [
-            {
-                'text': {'type': 'plain_text', 'text': 'No Repository'},
-                'value': '-',
-            }
-        ]
-        options.extend(
-            {
-                'text': {
-                    'type': 'plain_text',
-                    'text': repo.full_name,
-                },
-                'value': repo.full_name,
-            }
-            for repo in repo_list
-        )
+        self, message_ts: str, thread_ts: str | None
+    ) -> list[dict[str, Any]]:
+        """Generate a repo selection form using external_select for dynamic loading.
 
+        This uses Slack's external_select element which allows:
+        - Type-ahead search for repositories
+        - Dynamic loading of options from an external endpoint
+        - Support for users with many repositories (no 100 option limit)
+
+        Args:
+            message_ts: The message timestamp for tracking
+            thread_ts: The thread timestamp if in a thread
+
+        Returns:
+            List of Slack Block Kit blocks for the selection form
+        """
         return [
             {
                 'type': 'header',
@@ -142,78 +262,250 @@ class SlackManager(Manager[SlackViewInterface]):
                     'emoji': True,
                 },
             },
+            {
+                'type': 'section',
+                'text': {
+                    'type': 'mrkdwn',
+                    'text': 'Type to search your repositories:',
+                },
+            },
             {
                 'type': 'actions',
                 'elements': [
                     {
-                        'type': 'static_select',
+                        'type': 'external_select',
                         'action_id': f'repository_select:{message_ts}:{thread_ts}',
-                        'options': options,
+                        'placeholder': {
+                            'type': 'plain_text',
+                            'text': 'Search repositories...',
+                        },
+                        'min_query_length': 0,  # Load initial options immediately
                     }
                 ],
             },
         ]
 
-    def filter_potential_repos_by_user_msg(
-        self, user_msg: str, user_repos: list[Repository]
-    ) -> tuple[bool, list[Repository]]:
-        inferred_repo = self._infer_repo_from_message(user_msg)
-        if not inferred_repo:
-            return False, user_repos[0:99]
+    def _build_repo_options(self, repos: list[Repository]) -> list[dict[str, Any]]:
+        """Build Slack options list from repositories.
 
-        final_repos = []
-        for repo in user_repos:
-            if inferred_repo.lower() in repo.full_name.lower():
-                final_repos.append(repo)
+        Always includes a "No Repository" option at the top, followed by up to 99
+        repositories (Slack has a 100 option limit for external_select).
 
-        # no repos matched, return original list
-        if len(final_repos) == 0:
-            return False, user_repos[0:99]
+        Args:
+            repos: List of Repository objects
 
-        # Found exact match
-        elif len(final_repos) == 1:
-            return True, final_repos
+        Returns:
+            List of Slack option objects
+        """
+        options: list[dict[str, Any]] = [
+            {
+                'text': {'type': 'plain_text', 'text': 'No Repository'},
+                'value': '-',
+            }
+        ]
+        options.extend(
+            {
+                'text': {
+                    'type': 'plain_text',
+                    'text': repo.full_name[:75],  # Slack has 75 char limit for text
+                },
+                'value': repo.full_name,
+            }
+            for repo in repos[:99]  # Leave room for "No Repository" option
+        )
+        return options
 
-        # Found partial matches
-        return False, final_repos[0:99]
+    async def search_repos_for_slack(
+        self, user_auth: UserAuth, query: str, per_page: int = 20
+    ) -> list[dict[str, Any]]:
+        """Public API for repository search with formatted Slack options.
+
+        This method searches for repositories and formats the results as Slack
+        external_select options.
+
+        Args:
+            user_auth: The user's authentication context
+            query: Search query to filter repositories (empty string returns all)
+            per_page: Maximum number of results to return (default: 20)
+
+        Returns:
+            List of Slack option objects ready for external_select response
+        """
+        repos = await self._search_repositories(
+            user_auth, query=query, per_page=per_page
+        )
+        return self._build_repo_options(repos)
 
     async def receive_message(self, message: Message):
+        """Process an incoming Slack message.
+
+        This is the single entry point for all Slack message processing.
+        All SlackErrors raised during processing are caught and handled here,
+        sending appropriate error messages to the user.
+        """
         self._confirm_incoming_source_type(message)
 
+        try:
+            slack_view = await self._process_message(message)
+            if slack_view and await self.is_job_requested(message, slack_view):
+                await self.start_job(slack_view)
+
+        except SlackError as e:
+            await self.handle_slack_error(message.message, e)
+
+        except Exception as e:
+            logger.exception(
+                'slack_unexpected_error',
+                extra={'error': str(e), **message.message},
+            )
+            await self.handle_slack_error(
+                message.message,
+                SlackError(SlackErrorCode.UNEXPECTED_ERROR),
+            )
+
+    async def receive_form_interaction(self, slack_payload: dict):
+        """Process a Slack form interaction (repository selection).
+
+        This handles the block_actions payload when a user selects a repository
+        from the dropdown form. It retrieves the original user message from Redis
+        and delegates to receive_message for processing.
+
+        Args:
+            slack_payload: The raw Slack interaction payload
+        """
+        # Extract fields from the Slack interaction payload
+        selected_repository = slack_payload['actions'][0]['selected_option']['value']
+        if selected_repository == '-':
+            selected_repository = None
+
+        slack_user_id = slack_payload['user']['id']
+        channel_id = slack_payload['container']['channel_id']
+        team_id = slack_payload['team']['id']
+
+        # Get original message_ts and thread_ts from action_id
+        attribs = slack_payload['actions'][0]['action_id'].split('repository_select:')[
+            -1
+        ]
+        message_ts, thread_ts = attribs.split(':')
+        thread_ts = None if thread_ts == 'None' else thread_ts
+
+        # Build partial payload for error handling during Redis retrieval
+        payload = {
+            'team_id': team_id,
+            'channel_id': channel_id,
+            'slack_user_id': slack_user_id,
+            'message_ts': message_ts,
+            'thread_ts': thread_ts,
+        }
+
+        # Retrieve the original user message from Redis
+        try:
+            user_msg = await self._retrieve_user_msg_for_form(message_ts, thread_ts)
+        except SlackError as e:
+            await self.handle_slack_error(payload, e)
+            return
+        except Exception as e:
+            logger.exception(
+                'slack_unexpected_error',
+                extra={'error': str(e), **payload},
+            )
+            await self.handle_slack_error(
+                payload, SlackError(SlackErrorCode.UNEXPECTED_ERROR)
+            )
+            return
+
+        # Complete the payload and delegate to receive_message
+        payload['selected_repo'] = selected_repository
+        payload['user_msg'] = user_msg
+
+        message = Message(source=SourceType.SLACK, message=payload)
+        await self.receive_message(message)
+
+    async def _process_message(self, message: Message) -> SlackViewInterface | None:
+        """Process message and return view if authenticated, or raise SlackError.
+
+        Returns:
+            SlackViewInterface if user is authenticated and ready to proceed,
+            None if processing should stop (but no error).
+
+        Raises:
+            SlackError: If user is not authenticated or other recoverable error.
+        """
         slack_user, saas_user_auth = await self.authenticate_user(
             slack_user_id=message.message['slack_user_id']
         )
 
-        try:
-            slack_view = await SlackFactory.create_slack_view_from_payload(
-                message, slack_user, saas_user_auth
+        slack_view = await SlackFactory.create_slack_view_from_payload(
+            message, slack_user, saas_user_auth
+        )
+
+        # Check if this is an unauthenticated user (SlackMessageView but not SlackViewInterface)
+        if not isinstance(slack_view, SlackViewInterface):
+            login_link = self._generate_login_link_with_state(message)
+            raise SlackError(
+                SlackErrorCode.USER_NOT_AUTHENTICATED,
+                message_kwargs={'login_link': login_link},
+                log_context=slack_view.to_log_context(),
             )
-        except Exception as e:
+
+        return slack_view
+
+    def _generate_login_link_with_state(self, message: Message) -> str:
+        """Generate OAuth login link with message state encoded."""
+        jwt_secret = config.jwt_secret
+        if not jwt_secret:
+            raise ValueError('Must configure jwt_secret')
+        state = jwt.encode(
+            message.message, jwt_secret.get_secret_value(), algorithm='HS256'
+        )
+        return authorize_url_generator.generate(state)
+
+    async def handle_slack_error(self, payload: dict, error: SlackError) -> None:
+        """Handle a SlackError by logging and sending user message.
+
+        This is the centralized error handler for all SlackErrors, used by both
+        the manager and routes.
+
+        Args:
+            payload: The Slack payload dict containing channel/user info
+            error: The SlackError to handle
+        """
+        # Create a minimal view for sending the error message
+        view = await SlackMessageView.from_payload(
+            payload, self._get_slack_team_store()
+        )
+
+        if not view:
             logger.error(
-                f'[Slack]: Failed to create slack view: {e}',
-                exc_info=True,
-                stack_info=True,
+                'slack_error_no_view',
+                extra={
+                    'error_code': error.code.value,
+                    **error.log_context,
+                },
             )
             return
 
-        if isinstance(slack_view, SlackUnkownUserView):
-            jwt_secret = config.jwt_secret
-            if not jwt_secret:
-                raise ValueError('Must configure jwt_secret')
-            state = jwt.encode(
-                message.message, jwt_secret.get_secret_value(), algorithm='HS256'
-            )
-            link = authorize_url_generator.generate(state)
-            msg = self.login_link.format(link)
+        # Log the error
+        log_level = (
+            'exception' if error.code == SlackErrorCode.UNEXPECTED_ERROR else 'warning'
+        )
+        log_data = {
+            'error_code': error.code.value,
+            **view.to_log_context(),
+            **error.log_context,
+        }
+        getattr(logger, log_level)(
+            f'slack_error_{error.code.name.lower()}', extra=log_data
+        )
 
-            logger.info('slack_not_yet_authenticated')
-            await self.send_message(msg, slack_view, ephemeral=True)
-            return
+        # Send user-facing message
+        await self.send_message(error.get_user_message(), view, ephemeral=True)
 
-        if not await self.is_job_requested(message, slack_view):
-            return
+    def _get_slack_team_store(self):
+        """Get the SlackTeamStore instance (lazy import to avoid circular deps)."""
+        from storage.slack_team_store import SlackTeamStore
 
-        await self.start_job(slack_view)
+        return SlackTeamStore.get_instance()
 
     async def send_message(
         self,
@@ -254,76 +546,109 @@ class SlackManager(Manager[SlackViewInterface]):
                 thread_ts=slack_view.message_ts,
             )
 
+    async def _try_verify_inferred_repo(
+        self, slack_view: SlackNewConversationView
+    ) -> bool:
+        """Try to infer and verify a repository from the user's message.
+
+        Returns:
+            True if a valid repo was found and verified, False otherwise
+        """
+        user = slack_view.slack_to_openhands_user
+        inferred_repos = infer_repo_from_message(slack_view.user_msg)
+
+        if len(inferred_repos) != 1:
+            return False
+
+        inferred_repo = inferred_repos[0]
+        logger.info(
+            f'[Slack] Verifying inferred repo "{inferred_repo}" '
+            f'for user {user.slack_display_name} (id={slack_view.saas_user_auth.get_user_id()})'
+        )
+
+        try:
+            provider_tokens = await slack_view.saas_user_auth.get_provider_tokens()
+            if not provider_tokens:
+                return False
+
+            access_token = await slack_view.saas_user_auth.get_access_token()
+            user_id = await slack_view.saas_user_auth.get_user_id()
+            provider_handler = ProviderHandler(
+                provider_tokens=provider_tokens,
+                external_auth_token=access_token,
+                external_auth_id=user_id,
+            )
+            repo = await provider_handler.verify_repo_provider(inferred_repo)
+            slack_view.selected_repo = repo.full_name
+            return True
+        except (AuthenticationError, ProviderTimeoutError) as e:
+            logger.info(
+                f'[Slack] Could not verify repo "{inferred_repo}": {e}. '
+                f'Showing repository selector.'
+            )
+            return False
+
+    async def _show_repo_selection_form(
+        self, slack_view: SlackNewConversationView
+    ) -> None:
+        """Display the repository selection form to the user.
+
+        Raises:
+            SlackError: If storing the user message fails (REDIS_STORE_FAILED)
+        """
+        user = slack_view.slack_to_openhands_user
+        logger.info(
+            'render_repository_selector',
+            extra={
+                'slack_user_id': user.slack_user_id,
+                'keycloak_user_id': user.keycloak_user_id,
+                'message_ts': slack_view.message_ts,
+                'thread_ts': slack_view.thread_ts,
+            },
+        )
+
+        # Store the user message for later retrieval - raises SlackError on failure
+        await self._store_user_msg_for_form(
+            slack_view.message_ts, slack_view.thread_ts, slack_view.user_msg
+        )
+
+        repo_selection_msg = {
+            'text': 'Choose a Repository:',
+            'blocks': self._generate_repo_selection_form(
+                slack_view.message_ts, slack_view.thread_ts
+            ),
+        }
+        await self.send_message(repo_selection_msg, slack_view, ephemeral=True)
+
     async def is_job_requested(
         self, message: Message, slack_view: SlackViewInterface
     ) -> bool:
-        """A job is always request we only receive webhooks for events associated with the slack bot
-        This method really just checks
-            1. Is the user is authenticated
-            2. Do we have the necessary information to start a job (either by inferring the selected repo, otherwise asking the user)
+        """Determine if a job should be started based on the current context.
+
+        This method checks:
+            1. If the view type allows immediate job start
+            2. If a repo can be inferred and verified from the message
+            3. Otherwise shows the repo selection form
+
+        Args:
+            slack_view: Must be a SlackViewType (authenticated view that can start jobs)
+
+        Returns:
+            True if job should start, False if waiting for user input
         """
-        # Infer repo from user message is not needed; user selected repo from the form or is updating existing convo
+        # Check if view type allows immediate start
         if isinstance(slack_view, SlackUpdateExistingConversationView):
             return True
-        elif isinstance(slack_view, SlackNewConversationFromRepoFormView):
+        if isinstance(slack_view, SlackNewConversationFromRepoFormView):
             return True
-        elif isinstance(slack_view, SlackNewConversationView):
-            user = slack_view.slack_to_openhands_user
 
-            # Fetch repositories, handling timeout errors from the provider
-            logger.info(
-                f'[Slack] Fetching repositories for user {user.slack_display_name} (id={slack_view.saas_user_auth.get_user_id()})'
-            )
-            try:
-                user_repos: list[Repository] = await self._get_repositories(
-                    slack_view.saas_user_auth
-                )
-            except ProviderTimeoutError:
-                logger.warning(
-                    'repo_query_timeout',
-                    extra={
-                        'slack_user_id': user.slack_user_id,
-                        'keycloak_user_id': user.keycloak_user_id,
-                    },
-                )
-                timeout_msg = (
-                    'The repository selection timed out while fetching your repository list. '
-                    'Please re-send your message with a more specific repository name '
-                    '(e.g., "owner/repo-name") to help me find it faster.'
-                )
-                await self.send_message(timeout_msg, slack_view, ephemeral=True)
-                return False
-
-            match, repos = self.filter_potential_repos_by_user_msg(
-                slack_view.user_msg, user_repos
-            )
-
-            # User mentioned a matching repo is their message, start job without repo selection form
-            if match:
-                slack_view.selected_repo = repos[0].full_name
+        # For new conversations, try to infer/verify repo or show selection form
+        if isinstance(slack_view, SlackNewConversationView):
+            if await self._try_verify_inferred_repo(slack_view):
                 return True
+            await self._show_repo_selection_form(slack_view)
 
-            logger.info(
-                'render_repository_selector',
-                extra={
-                    'slack_user_id': user,
-                    'keycloak_user_id': user.keycloak_user_id,
-                    'message_ts': slack_view.message_ts,
-                    'thread_ts': slack_view.thread_ts,
-                },
-            )
-
-            repo_selection_msg = {
-                'text': 'Choose a Repository:',
-                'blocks': self._generate_repo_selection_form(
-                    repos, slack_view.message_ts, slack_view.thread_ts
-                ),
-            }
-            await self.send_message(repo_selection_msg, slack_view, ephemeral=True)
-
-            return False
-
-        return True
+        return False
 
     async def start_job(self, slack_view: SlackViewInterface) -> None:
         # Importing here prevents circular import
diff --git a/enterprise/integrations/slack/slack_types.py b/enterprise/integrations/slack/slack_types.py
index efa355da4f..b2d4faedf6 100644
--- a/enterprise/integrations/slack/slack_types.py
+++ b/enterprise/integrations/slack/slack_types.py
@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 
 from integrations.types import SummaryExtractionTracker
 from jinja2 import Environment
@@ -7,12 +8,13 @@ from storage.slack_user import SlackUser
 from openhands.server.user_auth.user_auth import UserAuth
 
 
-class SlackMessageView(ABC):
-    """Minimal interface for sending messages to Slack.
+@dataclass
+class SlackMessageView:
+    """Minimal view for sending messages to Slack.
 
-    This base class contains only the fields needed to send messages,
-    without requiring user authentication. Used by both authenticated
-    and unauthenticated Slack views.
+    This class contains only the fields needed to send messages,
+    without requiring user authentication. Can be used directly for
+    simple message operations or as a base class for more complex views.
     """
 
     bot_access_token: str
@@ -20,6 +22,77 @@ class SlackMessageView(ABC):
     channel_id: str
     message_ts: str
     thread_ts: str | None
+    team_id: str
+
+    def to_log_context(self) -> dict:
+        """Return dict suitable for structured logging."""
+        return {
+            'slack_channel_id': self.channel_id,
+            'slack_user_id': self.slack_user_id,
+            'slack_team_id': self.team_id,
+            'slack_thread_ts': self.thread_ts,
+            'slack_message_ts': self.message_ts,
+        }
+
+    @classmethod
+    async def from_payload(
+        cls,
+        payload: dict,
+        slack_team_store,
+    ) -> 'SlackMessageView | None':
+        """Create a view from a raw Slack payload.
+
+        This factory method handles the various payload formats from different
+        Slack interactions (events, form submissions, block suggestions).
+
+        Args:
+            payload: Raw Slack payload dictionary
+            slack_team_store: Store for retrieving bot tokens
+
+        Returns:
+            SlackMessageView if all required fields are available,
+            None if required fields are missing or bot token unavailable.
+        """
+        from openhands.core.logger import openhands_logger as logger
+
+        team_id = payload.get('team', {}).get('id') or payload.get('team_id')
+        channel_id = (
+            payload.get('container', {}).get('channel_id')
+            or payload.get('channel', {}).get('id')
+            or payload.get('channel_id')
+        )
+        user_id = payload.get('user', {}).get('id') or payload.get('slack_user_id')
+        message_ts = payload.get('message_ts', '')
+        thread_ts = payload.get('thread_ts')
+
+        if not team_id or not channel_id or not user_id:
+            logger.warning(
+                'slack_message_view_from_payload_missing_fields',
+                extra={
+                    'has_team_id': bool(team_id),
+                    'has_channel_id': bool(channel_id),
+                    'has_user_id': bool(user_id),
+                    'payload_keys': list(payload.keys()),
+                },
+            )
+            return None
+
+        bot_token = await slack_team_store.get_team_bot_token(team_id)
+        if not bot_token:
+            logger.warning(
+                'slack_message_view_from_payload_no_bot_token',
+                extra={'team_id': team_id},
+            )
+            return None
+
+        return cls(
+            bot_access_token=bot_token,
+            slack_user_id=user_id,
+            channel_id=channel_id,
+            message_ts=message_ts,
+            thread_ts=thread_ts,
+            team_id=team_id,
+        )
 
 
 class SlackViewInterface(SlackMessageView, SummaryExtractionTracker, ABC):
@@ -27,6 +100,9 @@ class SlackViewInterface(SlackMessageView, SummaryExtractionTracker, ABC):
 
     All fields are required (non-None) because this interface is only used
     for users who have linked their Slack account to OpenHands.
+
+    Inherits from SlackMessageView:
+        bot_access_token, slack_user_id, channel_id, message_ts, thread_ts, team_id
     """
 
     user_msg: str
@@ -36,7 +112,6 @@ class SlackViewInterface(SlackMessageView, SummaryExtractionTracker, ABC):
     should_extract: bool
     send_summary_instruction: bool
     conversation_id: str
-    team_id: str
     v1_enabled: bool
 
     @abstractmethod
@@ -55,4 +130,4 @@ class SlackViewInterface(SlackMessageView, SummaryExtractionTracker, ABC):
 
 
 class StartingConvoException(Exception):
-    """Raised when trying to send message to a conversation that's is still starting up"""
+    """Raised when trying to send message to a conversation that is still starting up."""
diff --git a/enterprise/integrations/slack/slack_view.py b/enterprise/integrations/slack/slack_view.py
index cd14137ebf..eb336404a3 100644
--- a/enterprise/integrations/slack/slack_view.py
+++ b/enterprise/integrations/slack/slack_view.py
@@ -63,22 +63,6 @@ async def is_v1_enabled_for_slack_resolver(user_id: str) -> bool:
     return await get_user_v1_enabled_setting(user_id) and ENABLE_V1_SLACK_RESOLVER
 
 
-@dataclass
-class SlackUnkownUserView(SlackMessageView):
-    """View for unauthenticated Slack users who haven't linked their account.
-
-    This view only contains the minimal fields needed to send a login link
-    message back to the user. It does not implement SlackViewInterface
-    because it cannot create conversations without user authentication.
-    """
-
-    bot_access_token: str
-    slack_user_id: str
-    channel_id: str
-    message_ts: str
-    thread_ts: str | None
-
-
 @dataclass
 class SlackNewConversationView(SlackViewInterface):
     bot_access_token: str
@@ -478,7 +462,7 @@ class SlackUpdateExistingConversationView(SlackNewConversationView):
             )
 
             # 6. Send the message to the agent server
-            url = f'{agent_server_url.rstrip("/")}/api/conversations/{UUID(self.conversation_id)}/events'
+            url = f"{agent_server_url.rstrip('/')}/api/conversations/{UUID(self.conversation_id)}/events"
 
             headers = {'X-Session-API-Key': running_sandbox.session_api_key}
             payload = send_message_request.model_dump()
@@ -576,13 +560,15 @@ class SlackFactory:
             raise Exception('Did not find slack team')
 
         # Determine if this is a known slack user by openhands
+        # Return SlackMessageView (not SlackViewInterface) for unauthenticated users
         if not slack_user or not saas_user_auth or not channel_id or not message_ts:
-            return SlackUnkownUserView(
+            return SlackMessageView(
                 bot_access_token=bot_access_token,
                 slack_user_id=slack_user_id,
                 channel_id=channel_id or '',
                 message_ts=message_ts or '',
                 thread_ts=thread_ts,
+                team_id=team_id,
             )
 
         # At this point, we've verified slack_user, saas_user_auth, channel_id, and message_ts are set
@@ -657,3 +643,11 @@ class SlackFactory:
                 team_id=team_id,
                 v1_enabled=False,
             )
+
+
+# Type alias for all authenticated Slack view types that can start conversations
+SlackViewType = (
+    SlackNewConversationView
+    | SlackNewConversationFromRepoFormView
+    | SlackUpdateExistingConversationView
+)
diff --git a/enterprise/server/routes/integration/slack.py b/enterprise/server/routes/integration/slack.py
index dc98552bc3..94d8054de1 100644
--- a/enterprise/server/routes/integration/slack.py
+++ b/enterprise/server/routes/integration/slack.py
@@ -11,6 +11,7 @@ from fastapi.responses import (
     RedirectResponse,
 )
 from integrations.models import Message, SourceType
+from integrations.slack.slack_errors import SlackError, SlackErrorCode
 from integrations.slack.slack_manager import SlackManager
 from integrations.utils import (
     HOST_URL,
@@ -37,7 +38,7 @@ from storage.slack_team_store import SlackTeamStore
 from storage.slack_user import SlackUser
 from storage.user_store import UserStore
 
-from openhands.integrations.service_types import ProviderType
+from openhands.integrations.service_types import ProviderTimeoutError, ProviderType
 from openhands.server.shared import config, sio
 
 signature_verifier = SignatureVerifier(signing_secret=SLACK_SIGNING_SECRET)
@@ -322,9 +323,129 @@ async def on_event(request: Request, background_tasks: BackgroundTasks):
     return JSONResponse({'success': True})
 
 
+@slack_router.post('/on-options-load')
+async def on_options_load(request: Request, background_tasks: BackgroundTasks):
+    """Handle external_select options loading (block_suggestion payload).
+
+    This endpoint is called by Slack when a user interacts with an external_select
+    element. It supports dynamic repository search with pagination.
+
+    The endpoint:
+    1. Authenticates the Slack user
+    2. Searches for repositories matching the user's query
+    3. Returns up to 100 options for the dropdown
+
+    Configuration: Set the Options Load URL in Slack App settings to:
+    https://your-domain/slack/on-options-load
+    """
+    if not SLACK_WEBHOOKS_ENABLED:
+        return JSONResponse({'options': []})
+
+    body = await request.body()
+    form = await request.form()
+    payload_str = form.get('payload')
+    if not payload_str:
+        logger.warning('slack_on_options_load: No payload in request')
+        return JSONResponse({'options': []})
+
+    payload = json.loads(payload_str)
+
+    logger.info('slack_on_options_load', extra={'payload': payload})
+
+    # Verify the signature
+    if not signature_verifier.is_valid(
+        body=body,
+        timestamp=request.headers.get('X-Slack-Request-Timestamp'),
+        signature=request.headers.get('X-Slack-Signature'),
+    ):
+        raise HTTPException(status_code=403, detail='invalid_request')
+
+    # Verify this is a block_suggestion payload
+    if payload.get('type') != 'block_suggestion':
+        logger.warning(
+            f"slack_on_options_load: Unexpected payload type: {payload.get('type')}"
+        )
+        return JSONResponse({'options': []})
+
+    slack_user_id = payload['user']['id']
+    search_value = payload.get('value', '')  # What user typed in the search box
+
+    # Authenticate user
+    slack_user, saas_user_auth = await slack_manager.authenticate_user(slack_user_id)
+
+    if not slack_user or not saas_user_auth:
+        # Send ephemeral message asking user to link their account
+        background_tasks.add_task(
+            slack_manager.handle_slack_error,
+            payload,
+            SlackError(
+                SlackErrorCode.USER_NOT_AUTHENTICATED,
+                message_kwargs={'login_link': _generate_login_link()},
+                log_context={'slack_user_id': slack_user_id},
+            ),
+        )
+        return JSONResponse({'options': []})
+
+    try:
+        # Search for repositories matching the query
+        # Limit to 20 repos for fast initial load. Users can search for repos
+        # not in this list using the type-ahead search functionality.
+        options = await slack_manager.search_repos_for_slack(
+            saas_user_auth, query=search_value, per_page=20
+        )
+
+        logger.info(
+            'slack_on_options_load_success',
+            extra={
+                'slack_user_id': slack_user_id,
+                'search_value': search_value,
+                'num_options': len(options),
+            },
+        )
+
+        return JSONResponse({'options': options})
+
+    except ProviderTimeoutError as e:
+        # Handle provider timeout with user notification
+        background_tasks.add_task(
+            slack_manager.handle_slack_error,
+            payload,
+            SlackError(
+                SlackErrorCode.PROVIDER_TIMEOUT,
+                log_context={'slack_user_id': slack_user_id, 'error': str(e)},
+            ),
+        )
+        return JSONResponse({'options': []})
+
+    except Exception as e:
+        logger.exception(
+            'slack_options_load_error',
+            extra={
+                'slack_user_id': slack_user_id,
+                'search_value': search_value,
+                'error': str(e),
+            },
+        )
+        # Notify user about the unexpected error with error code
+        background_tasks.add_task(
+            slack_manager.handle_slack_error,
+            payload,
+            SlackError(
+                SlackErrorCode.UNEXPECTED_ERROR,
+                log_context={'slack_user_id': slack_user_id, 'error': str(e)},
+            ),
+        )
+        return JSONResponse({'options': []})
+
+
 @slack_router.post('/on-form-interaction')
 async def on_form_interaction(request: Request, background_tasks: BackgroundTasks):
-    """We check the nonce to start a conversation"""
+    """Handle repository selection form submission.
+
+    When a user selects a repository from the external_select dropdown,
+    this endpoint passes the payload to the manager which retrieves the
+    original user message from Redis and starts the conversation.
+    """
     if not SLACK_WEBHOOKS_ENABLED:
         return JSONResponse({'success': 'slack_webhooks_disabled'})
 
@@ -334,7 +455,7 @@ async def on_form_interaction(request: Request, background_tasks: BackgroundTask
 
     logger.info('slack_on_form_interaction', extra={'payload': payload})
 
-    # First verify the signature
+    # Verify the signature
     if not signature_verifier.is_valid(
         body=body,
         timestamp=request.headers.get('X-Slack-Request-Timestamp'),
@@ -343,40 +464,16 @@ async def on_form_interaction(request: Request, background_tasks: BackgroundTask
         raise HTTPException(status_code=403, detail='invalid_request')
 
     assert payload['type'] == 'block_actions'
-    selected_repository = payload['actions'][0]['selected_option'][
-        'value'
-    ]  # Get the repository
-    if selected_repository == '-':
-        selected_repository = None
-    slack_user_id = payload['user']['id']
-    channel_id = payload['container']['channel_id']
-    team_id = payload['team']['id']
-    # Hack - get original message_ts from element name
-    attribs = payload['actions'][0]['action_id'].split('repository_select:')[-1]
-    message_ts, thread_ts = attribs.split(':')
-    thread_ts = None if thread_ts == 'None' else thread_ts
-    # Get the original message
-    # Get the text message
-    # Start the conversation
 
-    payload = {
-        'message_ts': message_ts,
-        'thread_ts': thread_ts,
-        'channel_id': channel_id,
-        'slack_user_id': slack_user_id,
-        'selected_repo': selected_repository,
-        'team_id': team_id,
-    }
-
-    message = Message(
-        source=SourceType.SLACK,
-        message=payload,
-    )
-
-    background_tasks.add_task(slack_manager.receive_message, message)
+    background_tasks.add_task(slack_manager.receive_form_interaction, payload)
     return JSONResponse({'success': True})
 
 
+def _generate_login_link(state: str = '') -> str:
+    """Generate the OAuth login link for Slack authentication."""
+    return authorize_url_generator.generate(state)
+
+
 def _html_response(title: str, description: str, status_code: int) -> HTMLResponse:
     content = (
         '<style>body{background:#0d0f11;color:#ecedee;font-family:sans-serif;display:flex;justify-content:center;align-items:center;}</style>'
diff --git a/enterprise/tests/unit/test_slack_integration.py b/enterprise/tests/unit/test_slack_integration.py
index 4797341a2e..6ca52cc5e4 100644
--- a/enterprise/tests/unit/test_slack_integration.py
+++ b/enterprise/tests/unit/test_slack_integration.py
@@ -1,11 +1,21 @@
+import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from integrations.slack.slack_manager import SlackManager
+from fastapi import BackgroundTasks
+from integrations.slack.slack_manager import (
+    SLACK_USER_MSG_EXPIRATION,
+    SLACK_USER_MSG_KEY_PREFIX,
+    SlackManager,
+)
 from integrations.slack.slack_view import SlackNewConversationView
 from storage.slack_user import SlackUser
 
-from openhands.integrations.service_types import ProviderTimeoutError
+from openhands.integrations.service_types import (
+    ProviderTimeoutError,
+    ProviderType,
+    Repository,
+)
 from openhands.server.user_auth.user_auth import UserAuth
 
 
@@ -30,7 +40,9 @@ def mock_slack_user():
 def mock_user_auth():
     """Create a mock UserAuth."""
     auth = MagicMock(spec=UserAuth)
-    auth.get_provider_tokens = AsyncMock(return_value={})
+    auth.get_provider_tokens = AsyncMock(return_value={'github': 'test-token'})
+    auth.get_access_token = AsyncMock(return_value='access-token')
+    auth.get_user_id = AsyncMock(return_value='user-123')
     auth.get_secrets = AsyncMock(return_value=MagicMock(custom_secrets={}))
     return auth
 
@@ -59,68 +71,84 @@ def slack_new_conversation_view(mock_slack_user, mock_user_auth):
 @pytest.mark.parametrize(
     'message,expected',
     [
-        ('OpenHands/Openhands', 'OpenHands/Openhands'),
-        ('deploy repo', 'deploy'),
-        ('use hello world', None),
+        ('OpenHands/Openhands', ['OpenHands/Openhands']),
+        (
+            'help me with repo',
+            [],
+        ),  # Updated: this pattern is not matched by infer_repo_from_message
+        ('use hello world', []),
     ],
 )
-def test_infer_repo_from_message(message, expected, slack_manager):
-    # Test the extracted function
-    result = slack_manager._infer_repo_from_message(message)
+def test_infer_repo_from_message(message, expected):
+    # Test the infer_repo_from_message function from utils
+    from integrations.utils import infer_repo_from_message
+
+    result = infer_repo_from_message(message)
     assert result == expected
 
 
-class TestRepoQueryTimeoutHandling:
-    """Test timeout handling when fetching repositories for Slack integration."""
+class TestRepoVerificationHandling:
+    """Test repo verification handling for Slack integration."""
 
+    @patch('integrations.slack.slack_manager.sio')
+    @patch('integrations.slack.slack_manager.ProviderHandler')
     @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
-    @patch.object(SlackManager, '_get_repositories', new_callable=AsyncMock)
-    async def test_timeout_sends_user_friendly_message(
+    async def test_timeout_during_verification_shows_selector(
         self,
-        mock_get_repositories,
         mock_send_message,
+        mock_provider_handler_class,
+        mock_sio,
         slack_manager,
         slack_new_conversation_view,
     ):
-        """Test that when repository fetching times out, a user-friendly message is sent."""
-        # Setup: _get_repositories raises ProviderTimeoutError
-        mock_get_repositories.side_effect = ProviderTimeoutError(
-            'github API request timed out: ConnectTimeout'
+        """Test that when repo verification times out, selector is shown."""
+        # Setup Redis mock
+        mock_redis = AsyncMock()
+        mock_sio.manager.redis = mock_redis
+
+        # Setup: Modify message to include exactly one repo reference to trigger verification
+        slack_new_conversation_view.user_msg = 'Help me with OpenHands/OpenHands repo'
+
+        # Setup: verify_repo_provider raises ProviderTimeoutError
+        mock_provider_handler = MagicMock()
+        mock_provider_handler.verify_repo_provider = AsyncMock(
+            side_effect=ProviderTimeoutError(
+                'github API request timed out: ConnectTimeout'
+            )
         )
+        mock_provider_handler_class.return_value = mock_provider_handler
 
         # Execute
         result = await slack_manager.is_job_requested(
             MagicMock(), slack_new_conversation_view
         )
 
-        # Verify: should return False (job not started)
+        # Verify: should return False (job not started, but selector is shown)
         assert result is False
 
-        # Verify: send_message was called with the timeout message
+        # Verify: send_message was called once (for repo selector)
         mock_send_message.assert_called_once()
         call_args = mock_send_message.call_args
+        selector_message = call_args[0][0]
+        assert isinstance(selector_message, dict)
+        assert selector_message.get('text') == 'Choose a Repository:'
 
-        # Check the message content
-        message = call_args[0][0]
-        assert 'timed out' in message
-        assert 'repository name' in message
-        assert 'owner/repo-name' in message
-
-        # Check it was sent as ephemeral
-        assert call_args[1]['ephemeral'] is True
-
+    @patch('integrations.slack.slack_manager.sio')
     @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
-    @patch.object(SlackManager, '_get_repositories', new_callable=AsyncMock)
-    async def test_successful_repo_fetch_does_not_send_timeout_message(
+    async def test_no_repo_mentioned_shows_external_selector(
         self,
-        mock_get_repositories,
         mock_send_message,
+        mock_sio,
         slack_manager,
         slack_new_conversation_view,
     ):
-        """Test that successful repo fetch shows repo selector, not timeout message."""
-        # Setup: _get_repositories returns empty list (no repos, but no timeout)
-        mock_get_repositories.return_value = []
+        """Test that when no repo is mentioned, external_select repo selector is shown."""
+        # Setup Redis mock
+        mock_redis = AsyncMock()
+        mock_sio.manager.redis = mock_redis
+
+        # Setup: user message without any repo mention
+        slack_new_conversation_view.user_msg = 'Hello, can you help me?'
 
         # Execute
         result = await slack_manager.is_job_requested(
@@ -134,9 +162,980 @@ class TestRepoQueryTimeoutHandling:
         mock_send_message.assert_called_once()
         call_args = mock_send_message.call_args
 
-        # Check the message is NOT the timeout message
+        # Should be the repo selection form with external_select
         message = call_args[0][0]
-        assert 'timed out' not in str(message)
-        # Should be the repo selection form
         assert isinstance(message, dict)
         assert message.get('text') == 'Choose a Repository:'
+        # Verify it's using external_select
+        blocks = message.get('blocks', [])
+        actions_block = next((b for b in blocks if b.get('type') == 'actions'), None)
+        assert actions_block is not None
+        elements = actions_block.get('elements', [])
+        assert len(elements) > 0
+        assert elements[0].get('type') == 'external_select'
+
+    @patch('integrations.slack.slack_manager.sio')
+    @patch('integrations.slack.slack_manager.ProviderHandler')
+    @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
+    async def test_verified_repo_starts_job(
+        self,
+        mock_send_message,
+        mock_provider_handler_class,
+        mock_sio,
+        slack_manager,
+        slack_new_conversation_view,
+    ):
+        """Test that when repo is successfully verified, job starts without selector."""
+
+        # Setup Redis mock
+        mock_redis = AsyncMock()
+        mock_sio.manager.redis = mock_redis
+
+        # Setup: Modify message to include exactly one repo reference
+        slack_new_conversation_view.user_msg = 'Help me with OpenHands/OpenHands repo'
+
+        # Setup: verify_repo_provider returns a valid repo
+        mock_repo = Repository(
+            id='123',
+            full_name='OpenHands/OpenHands',
+            git_provider=ProviderType.GITHUB,
+            is_public=True,
+        )
+        mock_provider_handler = MagicMock()
+        mock_provider_handler.verify_repo_provider = AsyncMock(return_value=mock_repo)
+        mock_provider_handler_class.return_value = mock_provider_handler
+
+        # Execute
+        result = await slack_manager.is_job_requested(
+            MagicMock(), slack_new_conversation_view
+        )
+
+        # Verify: should return True (job started)
+        assert result is True
+
+        # Verify: send_message was NOT called (no selector needed)
+        mock_send_message.assert_not_called()
+
+        # Verify: selected_repo was set
+        assert slack_new_conversation_view.selected_repo == 'OpenHands/OpenHands'
+
+
+class TestBuildRepoOptions:
+    """Test the _build_repo_options helper method.
+
+    Note: _build_repo_options always includes the "No Repository" option at the top.
+    This is by design for the external_select dropdown.
+    """
+
+    def test_build_options_with_repos(self, slack_manager):
+        """Test building options from a list of repositories."""
+
+        repos = [
+            Repository(
+                id='1',
+                full_name='owner/repo1',
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+            Repository(
+                id='2',
+                full_name='owner/repo2',
+                git_provider=ProviderType.GITHUB,
+                is_public=False,
+            ),
+        ]
+
+        options = slack_manager._build_repo_options(repos)
+
+        # Should have 3 options: "No Repository" + 2 repos
+        assert len(options) == 3
+        assert options[0]['value'] == '-'
+        assert options[0]['text']['text'] == 'No Repository'
+        assert options[1]['value'] == 'owner/repo1'
+        assert options[2]['value'] == 'owner/repo2'
+
+    def test_build_options_empty_repos(self, slack_manager):
+        """Test building options with empty repo list still includes No Repository."""
+        options = slack_manager._build_repo_options([])
+
+        # Should have 1 option: just "No Repository"
+        assert len(options) == 1
+        assert options[0]['value'] == '-'
+        assert options[0]['text']['text'] == 'No Repository'
+
+    def test_build_options_truncates_long_names(self, slack_manager):
+        """Test that repo names longer than 75 chars are truncated."""
+
+        long_name = 'a' * 100
+        repos = [
+            Repository(
+                id='1',
+                full_name=long_name,
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+        ]
+
+        options = slack_manager._build_repo_options(repos)
+
+        # First option is "No Repository", second is the repo
+        assert len(options) == 2
+        # Text should be truncated to 75 chars
+        assert len(options[1]['text']['text']) == 75
+        # But value should have full name
+        assert options[1]['value'] == long_name
+
+
+class TestSearchRepositories:
+    """Test the _search_repositories method with real repository filtering logic."""
+
+    @patch('integrations.slack.slack_manager.ProviderHandler')
+    async def test_search_repositories_returns_repos_from_provider(
+        self, mock_provider_handler_class, slack_manager, mock_user_auth
+    ):
+        """Test that _search_repositories returns repositories from the provider."""
+
+        # Setup: Create real Repository objects
+        expected_repos = [
+            Repository(
+                id='1',
+                full_name='owner/frontend-app',
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+            Repository(
+                id='2',
+                full_name='owner/backend-api',
+                git_provider=ProviderType.GITHUB,
+                is_public=False,
+            ),
+            Repository(
+                id='3',
+                full_name='owner/shared-lib',
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+        ]
+
+        # Setup: Mock provider handler to return real repos
+        mock_provider_handler = MagicMock()
+        mock_provider_handler.search_repositories = AsyncMock(
+            return_value=expected_repos
+        )
+        mock_provider_handler_class.return_value = mock_provider_handler
+
+        # Setup: Mock user_auth to return valid tokens
+        mock_user_auth.get_provider_tokens = AsyncMock(
+            return_value={'github': 'test-token'}
+        )
+        mock_user_auth.get_access_token = AsyncMock(return_value='access-token')
+        mock_user_auth.get_user_id = AsyncMock(return_value='user-123')
+
+        # Execute: Search with a query
+        result = await slack_manager._search_repositories(
+            mock_user_auth, query='frontend', per_page=20
+        )
+
+        # Verify: The correct parameters were passed to search_repositories
+        mock_provider_handler.search_repositories.assert_called_once()
+        call_kwargs = mock_provider_handler.search_repositories.call_args[1]
+        assert call_kwargs['query'] == 'frontend'
+        assert call_kwargs['per_page'] == 20
+        assert call_kwargs['sort'] == 'pushed'
+        assert call_kwargs['order'] == 'desc'
+
+        # Verify: All repos are returned
+        assert len(result) == 3
+        assert result[0].full_name == 'owner/frontend-app'
+        assert result[1].full_name == 'owner/backend-api'
+        assert result[2].full_name == 'owner/shared-lib'
+
+    @patch('integrations.slack.slack_manager.ProviderHandler')
+    async def test_search_repositories_returns_empty_when_no_tokens(
+        self, mock_provider_handler_class, slack_manager, mock_user_auth
+    ):
+        """Test that _search_repositories returns empty list when user has no provider tokens."""
+        # Setup: User has no provider tokens
+        mock_user_auth.get_provider_tokens = AsyncMock(return_value=None)
+
+        # Execute
+        result = await slack_manager._search_repositories(mock_user_auth, query='test')
+
+        # Verify: Returns empty list, doesn't call ProviderHandler
+        assert result == []
+        mock_provider_handler_class.assert_not_called()
+
+    @patch('integrations.slack.slack_manager.ProviderHandler')
+    async def test_search_and_build_options_integration(
+        self, mock_provider_handler_class, slack_manager, mock_user_auth
+    ):
+        """Test the full flow: search repositories and build options for Slack.
+
+        This exercises the full code path from search → filter → options building.
+        """
+
+        # Setup: Create a realistic repository list
+        repos = [
+            Repository(
+                id='1',
+                full_name='myorg/react-dashboard',
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+            Repository(
+                id='2',
+                full_name='myorg/python-api',
+                git_provider=ProviderType.GITHUB,
+                is_public=False,
+            ),
+            Repository(
+                id='3',
+                full_name='myorg/docs-site',
+                git_provider=ProviderType.GITHUB,
+                is_public=True,
+            ),
+        ]
+
+        mock_provider_handler = MagicMock()
+        mock_provider_handler.search_repositories = AsyncMock(return_value=repos)
+        mock_provider_handler_class.return_value = mock_provider_handler
+
+        mock_user_auth.get_provider_tokens = AsyncMock(
+            return_value={'github': 'test-token'}
+        )
+        mock_user_auth.get_access_token = AsyncMock(return_value='access-token')
+        mock_user_auth.get_user_id = AsyncMock(return_value='user-123')
+
+        # Execute: Search and build options (simulating what slack route does)
+        search_results = await slack_manager._search_repositories(
+            mock_user_auth, query='', per_page=100
+        )
+        options = slack_manager._build_repo_options(search_results)
+
+        # Verify: Options are correctly built from search results
+        assert len(options) == 4  # "No Repository" + 3 repos
+
+        # First option should be "No Repository"
+        assert options[0]['value'] == '-'
+        assert options[0]['text']['text'] == 'No Repository'
+
+        # Remaining options should be the repos in order
+        assert options[1]['value'] == 'myorg/react-dashboard'
+        assert options[1]['text']['text'] == 'myorg/react-dashboard'
+        assert options[2]['value'] == 'myorg/python-api'
+        assert options[3]['value'] == 'myorg/docs-site'
+
+    @patch('integrations.slack.slack_manager.ProviderHandler')
+    async def test_search_with_empty_results_builds_no_repo_only_option(
+        self, mock_provider_handler_class, slack_manager, mock_user_auth
+    ):
+        """Test that when search returns no results, only 'No Repository' option is shown."""
+        # Setup: No matching repos
+        mock_provider_handler = MagicMock()
+        mock_provider_handler.search_repositories = AsyncMock(return_value=[])
+        mock_provider_handler_class.return_value = mock_provider_handler
+
+        mock_user_auth.get_provider_tokens = AsyncMock(
+            return_value={'github': 'test-token'}
+        )
+        mock_user_auth.get_access_token = AsyncMock(return_value='access-token')
+        mock_user_auth.get_user_id = AsyncMock(return_value='user-123')
+
+        # Execute
+        search_results = await slack_manager._search_repositories(
+            mock_user_auth, query='nonexistent-repo', per_page=100
+        )
+        options = slack_manager._build_repo_options(search_results)
+
+        # Verify: Only "No Repository" option
+        assert len(options) == 1
+        assert options[0]['value'] == '-'
+        assert options[0]['text']['text'] == 'No Repository'
+
+
+class TestUserMsgStorage:
+    """Test the user message storage for repo selection form flow.
+
+    Note: _store_user_msg_for_form and _retrieve_user_msg_for_form are private methods
+    that raise SlackError on failure instead of returning True/False.
+    """
+
+    @pytest.mark.parametrize(
+        'message_ts,thread_ts,user_msg',
+        [
+            (
+                '1234567890.123456',
+                '1234567890.111111',
+                'Hello OpenHands, help me with my code',
+            ),
+            ('1234567890.123456', None, 'Hello OpenHands'),
+            ('9999999999.999999', '8888888888.888888', 'Another test message'),
+        ],
+        ids=['with_thread', 'without_thread', 'different_timestamps'],
+    )
+    @patch('integrations.slack.slack_manager.sio')
+    async def test_store_user_msg_for_form(
+        self, mock_sio, slack_manager, message_ts, thread_ts, user_msg
+    ):
+        """Test storing user message in Redis with various timestamp combinations."""
+        mock_redis = AsyncMock()
+        mock_sio.manager.redis = mock_redis
+
+        # Should not raise an exception on success
+        await slack_manager._store_user_msg_for_form(message_ts, thread_ts, user_msg)
+
+        expected_key = f'{SLACK_USER_MSG_KEY_PREFIX}:{message_ts}:{thread_ts}'
+        mock_redis.set.assert_called_once_with(
+            expected_key, user_msg, ex=SLACK_USER_MSG_EXPIRATION
+        )
+
+    @pytest.mark.parametrize(
+        'exception_type,exception_msg',
+        [
+            (ConnectionError, 'Connection refused'),
+            (TimeoutError, 'Redis operation timed out'),
+            (Exception, 'Redis internal error'),
+        ],
+        ids=['connection_error', 'timeout_error', 'generic_exception'],
+    )
+    @patch('integrations.slack.slack_manager.sio')
+    async def test_store_user_msg_for_form_redis_failure(
+        self, mock_sio, slack_manager, exception_type, exception_msg
+    ):
+        """Test that Redis failures during store raise SlackError."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        mock_redis = AsyncMock()
+        mock_redis.set.side_effect = exception_type(exception_msg)
+        mock_sio.manager.redis = mock_redis
+
+        message_ts = '1234567890.123456'
+        thread_ts = '1234567890.111111'
+        user_msg = 'Hello OpenHands'
+
+        # Should raise SlackError when Redis fails
+        with pytest.raises(SlackError) as exc_info:
+            await slack_manager._store_user_msg_for_form(
+                message_ts, thread_ts, user_msg
+            )
+
+        assert exc_info.value.code == SlackErrorCode.REDIS_STORE_FAILED
+
+    @pytest.mark.parametrize(
+        'redis_return_value,expected_result',
+        [
+            (
+                b'Hello OpenHands, help me with my code',
+                'Hello OpenHands, help me with my code',
+            ),
+            ('Hello OpenHands', 'Hello OpenHands'),  # String instead of bytes
+        ],
+        ids=['bytes_response', 'string_response'],
+    )
+    @patch('integrations.slack.slack_manager.sio')
+    async def test_retrieve_user_msg_for_form(
+        self, mock_sio, slack_manager, redis_return_value, expected_result
+    ):
+        """Test retrieving user message from Redis with various response types."""
+        mock_redis = AsyncMock()
+        mock_redis.get.return_value = redis_return_value
+        mock_sio.manager.redis = mock_redis
+
+        message_ts = '1234567890.123456'
+        thread_ts = '1234567890.111111'
+
+        result = await slack_manager._retrieve_user_msg_for_form(message_ts, thread_ts)
+
+        expected_key = f'{SLACK_USER_MSG_KEY_PREFIX}:{message_ts}:{thread_ts}'
+        mock_redis.get.assert_called_once_with(expected_key)
+        assert result == expected_result
+
+    @patch('integrations.slack.slack_manager.sio')
+    async def test_retrieve_user_msg_for_form_key_not_found(
+        self, mock_sio, slack_manager
+    ):
+        """Test that missing key raises SlackError with SESSION_EXPIRED."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        mock_redis = AsyncMock()
+        mock_redis.get.return_value = None
+        mock_sio.manager.redis = mock_redis
+
+        message_ts = '1234567890.123456'
+        thread_ts = '1234567890.111111'
+
+        # Should raise SlackError when key not found
+        with pytest.raises(SlackError) as exc_info:
+            await slack_manager._retrieve_user_msg_for_form(message_ts, thread_ts)
+
+        assert exc_info.value.code == SlackErrorCode.SESSION_EXPIRED
+
+    @pytest.mark.parametrize(
+        'exception_type,exception_msg',
+        [
+            (ConnectionError, 'Connection refused'),
+            (TimeoutError, 'Redis operation timed out'),
+        ],
+        ids=['connection_error', 'timeout_error'],
+    )
+    @patch('integrations.slack.slack_manager.sio')
+    async def test_retrieve_user_msg_for_form_redis_failure(
+        self, mock_sio, slack_manager, exception_type, exception_msg
+    ):
+        """Test that Redis failures during retrieve raise SlackError."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        mock_redis = AsyncMock()
+        mock_redis.get.side_effect = exception_type(exception_msg)
+        mock_sio.manager.redis = mock_redis
+
+        message_ts = '1234567890.123456'
+        thread_ts = '1234567890.111111'
+
+        # Should raise SlackError when Redis fails
+        with pytest.raises(SlackError) as exc_info:
+            await slack_manager._retrieve_user_msg_for_form(message_ts, thread_ts)
+
+        assert exc_info.value.code == SlackErrorCode.REDIS_RETRIEVE_FAILED
+
+
+class TestIsJobRequestedWithUserMsgStorage:
+    """Test that is_job_requested properly stores user message for form flow."""
+
+    @patch('integrations.slack.slack_manager.sio')
+    @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
+    async def test_stores_user_msg_when_showing_repo_selector(
+        self,
+        mock_send_message,
+        mock_sio,
+        slack_manager,
+        slack_new_conversation_view,
+    ):
+        """Test that user_msg is stored in Redis when repo selector is shown."""
+        mock_redis = AsyncMock()
+        mock_sio.manager.redis = mock_redis
+
+        # Setup: user message without any repo mention (no repo inferred)
+        slack_new_conversation_view.user_msg = 'Hello, can you help me?'
+
+        # Execute
+        result = await slack_manager.is_job_requested(
+            MagicMock(), slack_new_conversation_view
+        )
+
+        # Verify: should return False (no repo selected yet)
+        assert result is False
+
+        # Verify: Redis set was called to store the user message
+        expected_key = f'{SLACK_USER_MSG_KEY_PREFIX}:{slack_new_conversation_view.message_ts}:{slack_new_conversation_view.thread_ts}'
+        mock_redis.set.assert_called_once_with(
+            expected_key,
+            slack_new_conversation_view.user_msg,
+            ex=SLACK_USER_MSG_EXPIRATION,
+        )
+
+
+class TestOnOptionsLoadEndpoint:
+    """Test the /on-options-load endpoint for external_select repo search."""
+
+    @pytest.fixture
+    def mock_request(self):
+        """Create a mock Request object."""
+        request = MagicMock()
+        request.headers = {
+            'X-Slack-Request-Timestamp': '1234567890',
+            'X-Slack-Signature': 'v0=test_signature',
+        }
+        return request
+
+    @pytest.fixture
+    def valid_block_suggestion_payload(self):
+        """Create a valid block_suggestion payload from Slack."""
+        return {
+            'type': 'block_suggestion',
+            'user': {'id': 'U1234567890'},
+            'value': 'test-query',
+            'team': {'id': 'T1234567890'},
+            'container': {'channel_id': 'C1234567890'},
+        }
+
+    @pytest.fixture
+    def background_tasks(self):
+        """Create mock BackgroundTasks."""
+        return MagicMock(spec=BackgroundTasks)
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', False)
+    async def test_on_options_load_disabled_returns_empty_options(
+        self, mock_request, background_tasks
+    ):
+        """Test that when webhooks are disabled, empty options are returned."""
+        from server.routes.integration.slack import on_options_load
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': []}
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    async def test_on_options_load_no_payload_returns_empty_options(
+        self, mock_request, background_tasks
+    ):
+        """Test that when no payload is in request, empty options are returned."""
+        from server.routes.integration.slack import on_options_load
+
+        mock_request.body = AsyncMock(return_value=b'')
+        mock_form = MagicMock()
+        mock_form.get.return_value = None
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': []}
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    async def test_on_options_load_invalid_signature_raises_403(
+        self,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        valid_block_suggestion_payload,
+    ):
+        """Test that invalid Slack signature raises 403 HTTPException."""
+        from fastapi import HTTPException
+        from server.routes.integration.slack import on_options_load
+
+        payload_str = json.dumps(valid_block_suggestion_payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = False
+
+        with pytest.raises(HTTPException) as exc_info:
+            await on_options_load(mock_request, background_tasks)
+
+        assert exc_info.value.status_code == 403
+        assert exc_info.value.detail == 'invalid_request'
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    async def test_on_options_load_wrong_payload_type_returns_empty_options(
+        self, mock_signature_verifier, mock_request, background_tasks
+    ):
+        """Test that non-block_suggestion payload returns empty options."""
+        from server.routes.integration.slack import on_options_load
+
+        payload = {
+            'type': 'interactive_message',  # Wrong type
+            'user': {'id': 'U1234567890'},
+        }
+        payload_str = json.dumps(payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': []}
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_unauthenticated_user_returns_empty_options(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        valid_block_suggestion_payload,
+    ):
+        """Test that unauthenticated users get empty options and linking message is queued."""
+        from server.routes.integration.slack import on_options_load
+
+        payload_str = json.dumps(valid_block_suggestion_payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(return_value=(None, None))
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': []}
+
+        # Verify background task was queued for account linking message
+        background_tasks.add_task.assert_called_once()
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_successful_search_with_repos(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        valid_block_suggestion_payload,
+        mock_slack_user,
+        mock_user_auth,
+    ):
+        """Test successful repository search returns properly formatted options.
+
+        This test verifies the endpoint calls search_repos_for_slack with the
+        correct parameters. The actual formatting is tested in TestBuildRepoOptions.
+        """
+        from server.routes.integration.slack import on_options_load
+
+        payload_str = json.dumps(valid_block_suggestion_payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(
+            return_value=(mock_slack_user, mock_user_auth)
+        )
+
+        # Expected options from search_repos_for_slack
+        expected_options = [
+            {'text': {'type': 'plain_text', 'text': 'No Repository'}, 'value': '-'},
+            {
+                'text': {'type': 'plain_text', 'text': 'owner/repo1'},
+                'value': 'owner/repo1',
+            },
+            {
+                'text': {'type': 'plain_text', 'text': 'owner/repo2'},
+                'value': 'owner/repo2',
+            },
+        ]
+        mock_slack_manager.search_repos_for_slack = AsyncMock(
+            return_value=expected_options
+        )
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': expected_options}
+
+        # Verify search_repos_for_slack was called with correct parameters
+        mock_slack_manager.search_repos_for_slack.assert_called_once_with(
+            mock_user_auth, query='test-query', per_page=20
+        )
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_empty_query_search(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        mock_slack_user,
+        mock_user_auth,
+    ):
+        """Test search with empty query (min_query_length: 0 in external_select)."""
+        from server.routes.integration.slack import on_options_load
+
+        # Payload with empty value (no search text entered yet)
+        payload = {
+            'type': 'block_suggestion',
+            'user': {'id': 'U1234567890'},
+            'value': '',  # Empty search
+            'team': {'id': 'T1234567890'},
+            'container': {'channel_id': 'C1234567890'},
+        }
+        payload_str = json.dumps(payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(
+            return_value=(mock_slack_user, mock_user_auth)
+        )
+        mock_slack_manager.search_repos_for_slack = AsyncMock(
+            return_value=[
+                {'text': {'type': 'plain_text', 'text': 'No Repository'}, 'value': '-'}
+            ]
+        )
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+
+        # Verify search_repos_for_slack was called with empty query
+        mock_slack_manager.search_repos_for_slack.assert_called_once_with(
+            mock_user_auth, query='', per_page=20
+        )
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_search_exception_returns_empty_options(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        valid_block_suggestion_payload,
+        mock_slack_user,
+        mock_user_auth,
+    ):
+        """Test that when search raises an exception, empty options are returned gracefully."""
+        from server.routes.integration.slack import on_options_load
+
+        payload_str = json.dumps(valid_block_suggestion_payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(
+            return_value=(mock_slack_user, mock_user_auth)
+        )
+        # Simulate search error (e.g., provider timeout)
+        mock_slack_manager.search_repos_for_slack = AsyncMock(
+            side_effect=Exception('GitHub API timeout')
+        )
+        mock_slack_manager.handle_slack_error = AsyncMock()
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert body == {'options': []}
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_missing_value_field_defaults_to_empty(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        mock_slack_user,
+        mock_user_auth,
+    ):
+        """Test that missing 'value' field in payload defaults to empty string."""
+        from server.routes.integration.slack import on_options_load
+
+        # Payload without 'value' key
+        payload = {
+            'type': 'block_suggestion',
+            'user': {'id': 'U1234567890'},
+            # 'value' is missing
+            'team': {'id': 'T1234567890'},
+            'container': {'channel_id': 'C1234567890'},
+        }
+        payload_str = json.dumps(payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(
+            return_value=(mock_slack_user, mock_user_auth)
+        )
+        mock_slack_manager.search_repos_for_slack = AsyncMock(return_value=[])
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+
+        # Should default to empty string for search
+        mock_slack_manager.search_repos_for_slack.assert_called_once_with(
+            mock_user_auth, query='', per_page=20
+        )
+
+    @pytest.mark.asyncio
+    @patch('server.routes.integration.slack.SLACK_WEBHOOKS_ENABLED', True)
+    @patch('server.routes.integration.slack.signature_verifier')
+    @patch('server.routes.integration.slack.slack_manager')
+    async def test_on_options_load_truncates_long_repo_names(
+        self,
+        mock_slack_manager,
+        mock_signature_verifier,
+        mock_request,
+        background_tasks,
+        valid_block_suggestion_payload,
+        mock_slack_user,
+        mock_user_auth,
+    ):
+        """Test that options with long repo names are properly handled.
+
+        Note: The actual truncation logic is tested in TestBuildRepoOptions.
+        This test just verifies the endpoint correctly passes through the formatted options.
+        """
+        from server.routes.integration.slack import on_options_load
+
+        payload_str = json.dumps(valid_block_suggestion_payload)
+        mock_request.body = AsyncMock(return_value=payload_str.encode())
+        mock_form = MagicMock()
+        mock_form.get.return_value = payload_str
+        mock_request.form = AsyncMock(return_value=mock_form)
+
+        mock_signature_verifier.is_valid.return_value = True
+        mock_slack_manager.authenticate_user = AsyncMock(
+            return_value=(mock_slack_user, mock_user_auth)
+        )
+
+        # Mock the formatted options that would come from search_repos_for_slack
+        expected_options = [
+            {'text': {'type': 'plain_text', 'text': 'No Repository'}, 'value': '-'},
+            {
+                'text': {
+                    'type': 'plain_text',
+                    'text': 'verylongorganizationname/very-long-repository-name-tha',
+                },
+                'value': 'verylongorganizationname/very-long-repository-name-that-exceeds-normal-length',
+            },
+        ]
+        mock_slack_manager.search_repos_for_slack = AsyncMock(
+            return_value=expected_options
+        )
+
+        response = await on_options_load(mock_request, background_tasks)
+
+        assert response.status_code == 200
+        body = json.loads(response.body)
+        assert 'options' in body
+        assert len(body['options']) == 2
+
+
+class TestHandleSlackError:
+    """Test the handle_slack_error method on SlackManager.
+
+    Note: Error handling now goes through SlackManager.handle_slack_error method
+    instead of a standalone _send_slack_error function.
+    """
+
+    @pytest.mark.asyncio
+    @patch('integrations.slack.slack_manager.SlackMessageView.from_payload')
+    @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
+    async def test_handle_slack_error_success(
+        self, mock_send_message, mock_from_payload, slack_manager
+    ):
+        """Test successful sending of error message to Slack user."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        payload = {
+            'team': {'id': 'T1234567890'},
+            'container': {'channel_id': 'C1234567890'},
+            'user': {'id': 'U1234567890'},
+        }
+        error = SlackError(
+            SlackErrorCode.USER_NOT_AUTHENTICATED,
+            message_kwargs={'login_link': 'https://test.link'},
+            log_context={'slack_user_id': 'U1234567890'},
+        )
+
+        # Mock the view creation
+        mock_view = MagicMock()
+        mock_view.to_log_context.return_value = {}
+        mock_from_payload.return_value = mock_view
+
+        await slack_manager.handle_slack_error(payload, error)
+
+        mock_send_message.assert_called_once()
+        # Verify ephemeral=True is passed
+        call_args = mock_send_message.call_args
+        assert call_args.kwargs.get('ephemeral') is True
+
+    @pytest.mark.asyncio
+    @patch('integrations.slack.slack_manager.SlackMessageView.from_payload')
+    @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
+    async def test_handle_slack_error_no_view(
+        self, mock_send_message, mock_from_payload, slack_manager
+    ):
+        """Test handling when view creation fails (returns None)."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        payload = {
+            'team': {},  # Invalid - missing id
+            'container': {},
+            'user': {},
+        }
+        error = SlackError(
+            SlackErrorCode.SESSION_EXPIRED,
+            log_context={'test': 'context'},
+        )
+
+        # Mock view creation returning None
+        mock_from_payload.return_value = None
+
+        # Should handle gracefully without raising
+        await slack_manager.handle_slack_error(payload, error)
+
+        # send_message should not be called when view creation fails
+        mock_send_message.assert_not_called()
+
+    @pytest.mark.asyncio
+    @patch('integrations.slack.slack_manager.SlackMessageView.from_payload')
+    @patch.object(SlackManager, 'send_message', new_callable=AsyncMock)
+    async def test_handle_slack_error_various_error_codes(
+        self, mock_send_message, mock_from_payload, slack_manager
+    ):
+        """Test that different error codes produce appropriate messages."""
+        from integrations.slack.slack_errors import SlackError, SlackErrorCode
+
+        payload = {
+            'team': {'id': 'T1234567890'},
+            'container': {'channel_id': 'C1234567890'},
+            'user': {'id': 'U1234567890'},
+        }
+
+        # Mock the view creation
+        mock_view = MagicMock()
+        mock_view.to_log_context.return_value = {}
+        mock_from_payload.return_value = mock_view
+
+        # Test different error codes
+        error_codes = [
+            SlackErrorCode.SESSION_EXPIRED,
+            SlackErrorCode.PROVIDER_TIMEOUT,
+            SlackErrorCode.REDIS_STORE_FAILED,
+            SlackErrorCode.UNEXPECTED_ERROR,
+        ]
+
+        for code in error_codes:
+            mock_send_message.reset_mock()
+            error = SlackError(code, log_context={'test': 'context'})
+
+            await slack_manager.handle_slack_error(payload, error)
+
+            mock_send_message.assert_called_once()
+            call_args = mock_send_message.call_args
+            message = call_args.args[0]
+            # Verify message is not empty
+            assert message
+            assert isinstance(message, str)