from typing import Any 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, StartingConvoException, ) from integrations.slack.slack_view import ( SlackFactory, SlackNewConversationFromRepoFormView, SlackNewConversationView, 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 from server.constants import SLACK_CLIENT_ID from slack_sdk.oauth import AuthorizeUrlGenerator from slack_sdk.web.async_client import AsyncWebClient from sqlalchemy import select from storage.database import a_session_maker from storage.redis import get_redis_client_async from storage.slack_user import SlackUser from openhands.app_server.integrations.provider import ProviderHandler from openhands.app_server.integrations.service_types import ( AuthenticationError, ProviderTimeoutError, Repository, ) from openhands.app_server.shared import server_config from openhands.app_server.types import ( LLMAuthenticationError, MissingSettingsError, SessionExpiredError, ) from openhands.app_server.user_auth.user_auth import UserAuth from openhands.app_server.utils.logger import openhands_logger as logger authorize_url_generator = AuthorizeUrlGenerator( client_id=SLACK_CLIENT_ID, scopes=['app_mentions:read', 'chat:write'], 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): self.token_manager = token_manager self.login_link = ( 'User has not yet authenticated: [Click here to Login to OpenHands]({}).' ) self.jinja_env = Environment( loader=FileSystemLoader(OPENHANDS_RESOLVER_TEMPLATES_DIR + 'slack') ) def _confirm_incoming_source_type(self, message: Message): if message.source != SourceType.SLACK: raise ValueError(f'Unexpected message source {message.source}') async def authenticate_user( self, slack_user_id: str ) -> tuple[SlackUser | None, UserAuth | None]: # We get the user and correlate them back to a user in OpenHands - if we can slack_user = None async with a_session_maker() as session: result = await session.execute( select(SlackUser).where(SlackUser.slack_user_id == slack_user_id) ) slack_user = result.scalar_one_or_none() # slack_view.slack_to_openhands_user = slack_user # attach user auth info to view saas_user_auth = None if slack_user: saas_user_auth = await get_saas_user_auth( slack_user.keycloak_user_id, self.token_manager ) # slack_view.saas_user_auth = await self._get_user_auth(slack_view.slack_to_openhands_user.keycloak_user_id) return slack_user, saas_user_auth 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. 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. 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 Raises: SlackError: If storage fails (REDIS_STORE_FAILED) """ key = f'{SLACK_USER_MSG_KEY_PREFIX}:{message_ts}:{thread_ts}' try: redis = get_redis_client_async() 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 = get_redis_client_async() 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 [] access_token = await user_auth.get_access_token() user_id = await user_auth.get_user_id() client = ProviderHandler( provider_tokens=provider_tokens, external_auth_token=access_token, external_auth_id=user_id, ) 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, message_ts: str, thread_ts: str | None ) -> list[dict[str, Any]]: """Generate a repo selection form with immediate "No Repository" button and search dropdown. This form provides two options side-by-side: 1. A "No Repository" button - immediately clickable without any loading 2. An external_select dropdown - for searching repositories dynamically This design ensures "No Repository" is always immediately available while still providing full dynamic search capability for repositories. 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', 'text': { 'type': 'plain_text', 'text': 'Choose a repository', 'emoji': True, }, }, { 'type': 'section', 'text': { 'type': 'mrkdwn', 'text': 'Select a repository or continue without one:', }, }, { 'type': 'actions', 'elements': [ { 'type': 'button', 'action_id': f'no_repository:{message_ts}:{thread_ts}', 'text': { 'type': 'plain_text', 'text': 'No Repository', 'emoji': True, }, 'value': '-', }, { 'type': 'external_select', 'action_id': f'repository_select:{message_ts}:{thread_ts}', 'placeholder': { 'type': 'plain_text', 'text': 'Search repositories...', }, 'min_query_length': 0, }, ], }, ] def _build_repo_options(self, repos: list[Repository]) -> list[dict[str, Any]]: """Build Slack options list from repositories. Returns up to 100 repositories formatted as Slack options (Slack has a 100 option limit for external_select). Note: "No Repository" is handled by a separate button in the form, so it's not included in the dropdown options. Args: repos: List of Repository objects Returns: List of Slack option objects """ return [ { 'text': { 'type': 'plain_text', 'text': repo.full_name[:75], # Slack has 75 char limit for text }, 'value': repo.full_name, } for repo in repos[:100] ] 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), ) def _parse_form_action(self, action: dict) -> tuple[str, str | None, str] | None: """Parse action payload and extract message_ts, thread_ts, and selected value. This handles the different payload structures for button clicks vs dropdown selections in the repository selection form. Args: action: The action object from the Slack payload Returns: Tuple of (message_ts, thread_ts, selected_value) if action is recognized, None if the action_id is unknown. """ action_id = action['action_id'] if action_id.startswith('no_repository:'): # Button click - value is in 'value' field attribs = action_id.split('no_repository:')[-1] selected_value = action.get('value', '-') elif action_id.startswith('repository_select:'): # Dropdown selection - value is in 'selected_option' attribs = action_id.split('repository_select:')[-1] selected_value = action['selected_option']['value'] else: return None message_ts, thread_ts = attribs.split(':') thread_ts = None if thread_ts == 'None' else thread_ts return message_ts, thread_ts, selected_value async def receive_form_interaction(self, slack_payload: dict): """Process a Slack form interaction (repository selection or button click). This handles the block_actions payload when a user interacts with the repository selection form. It can handle: - "No Repository" button click: proceeds with conversation without a repo - Repository selection from dropdown: proceeds with the selected repo Args: slack_payload: The raw Slack interaction payload """ # Extract fields from the Slack interaction payload action = slack_payload['actions'][0] slack_user_id = slack_payload['user']['id'] channel_id = slack_payload['container']['channel_id'] team_id = slack_payload['team']['id'] # Parse the action to extract message_ts, thread_ts, and selected value parsed = self._parse_form_action(action) if parsed is None: logger.warning( 'slack_unknown_action_id', extra={ 'action_id': action['action_id'], 'slack_user_id': slack_user_id, }, ) return message_ts, thread_ts, selected_value = parsed # Build partial payload for error handling payload = { 'team_id': team_id, 'channel_id': channel_id, 'slack_user_id': slack_user_id, 'message_ts': message_ts, 'thread_ts': thread_ts, } # Convert "-" (No Repository) to None selected_repository = None if selected_value == '-' else selected_value # 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'] ) 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(), ) return slack_view def _generate_login_link_with_state(self, message: Message) -> str: """Generate OAuth login link with message state encoded.""" from storage.encrypt_utils import get_jwt_service state = get_jwt_service().create_jws_token(message.message) 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( 'slack_error_no_view', extra={ 'error_code': error.code.value, **error.log_context, }, ) return # 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 ) # Send user-facing message await self.send_message(error.get_user_message(), view, ephemeral=True) def _get_slack_team_store(self): """Get the SlackTeamStore instance (lazy import to avoid circular deps).""" from storage.slack_team_store import SlackTeamStore return SlackTeamStore.get_instance() async def send_message( self, message: str | dict[str, Any], slack_view: SlackMessageView, ephemeral: bool = False, ): """Send a message to Slack. Args: message: The message content. Can be a string (for simple text) or a dict with 'text' and 'blocks' keys (for structured messages). slack_view: The Slack view object containing channel/thread info. Can be either SlackMessageView (for unauthenticated users) or SlackViewInterface (for authenticated users). ephemeral: If True, send as an ephemeral message visible only to the user. """ client = AsyncWebClient(token=slack_view.bot_access_token) if ephemeral and isinstance(message, str): await client.chat_postEphemeral( channel=slack_view.channel_id, markdown_text=message, user=slack_view.slack_user_id, thread_ts=slack_view.thread_ts, ) elif ephemeral and isinstance(message, dict): await client.chat_postEphemeral( channel=slack_view.channel_id, user=slack_view.slack_user_id, thread_ts=slack_view.thread_ts, text=message['text'], blocks=message['blocks'], ) else: await client.chat_postMessage( channel=slack_view.channel_id, markdown_text=message, 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: """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 """ # Check if view type allows immediate start if isinstance(slack_view, SlackUpdateExistingConversationView): return True if isinstance(slack_view, SlackNewConversationFromRepoFormView): return True # 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) return False async def start_job(self, slack_view: SlackViewInterface) -> None: """Start a Slack job using V1 app conversation system.""" try: msg_info = None user_info = slack_view.slack_to_openhands_user try: logger.info( f'[Slack] Starting job for user {user_info.slack_display_name} (id={user_info.slack_user_id})', extra={'keyloak_user_id': user_info.keycloak_user_id}, ) conversation_id = await slack_view.create_or_update_conversation( self.jinja_env ) logger.info( f'[Slack] Created conversation {conversation_id} for user {user_info.slack_display_name}' ) # V1 callback processors are registered by the view during conversation creation msg_info = slack_view.get_response_msg() except MissingSettingsError as e: logger.warning( f'[Slack] Missing settings error for user {user_info.slack_display_name}: {str(e)}' ) msg_info = f'{user_info.slack_display_name} please re-login into [OpenHands Cloud]({HOST_URL}) before starting a job.' except LLMAuthenticationError as e: logger.warning( f'[Slack] LLM authentication error for user {user_info.slack_display_name}: {str(e)}' ) msg_info = f'@{user_info.slack_display_name} please set a valid LLM API key in [OpenHands Cloud]({HOST_URL}) before starting a job.' except SessionExpiredError as e: logger.warning( f'[Slack] Session expired for user {user_info.slack_display_name}: {str(e)}' ) msg_info = get_session_expired_message(user_info.slack_display_name) except StartingConvoException as e: msg_info = str(e) await self.send_message(msg_info, slack_view) except Exception: logger.exception('[Slack]: Error starting job') await self.send_message( 'Uh oh! There was an unexpected error starting the job :(', slack_view )