from types import MappingProxyType from github import Auth, Github, GithubIntegration from integrations.github.data_collector import GitHubDataCollector from integrations.github.github_view import ( GithubFactory, GithubFailingAction, GithubInlinePRComment, GithubIssue, GithubIssueComment, GithubPRComment, GithubViewType, ) from integrations.manager import Manager from integrations.models import ( Message, SourceType, ) from integrations.types import ResolverViewInterface from integrations.utils import ( CONVERSATION_URL, HOST_URL, OPENHANDS_RESOLVER_TEMPLATES_DIR, get_session_expired_message, get_user_not_found_message, ) from integrations.v1_utils import get_saas_user_auth from jinja2 import Environment, FileSystemLoader from pydantic import SecretStr from server.auth.auth_error import ExpiredError from server.auth.constants import GITHUB_APP_CLIENT_ID, GITHUB_APP_PRIVATE_KEY from server.auth.token_manager import TokenManager from openhands.app_server.integrations.provider import ProviderToken, ProviderType from openhands.app_server.integrations.service_types import AuthenticationError from openhands.app_server.secrets.secrets_models import Secrets from openhands.app_server.types import ( LLMAuthenticationError, MissingSettingsError, SessionExpiredError, ) from openhands.app_server.utils.logger import openhands_logger as logger class GithubManager(Manager[GithubViewType]): def __init__( self, token_manager: TokenManager, data_collector: GitHubDataCollector ): self.token_manager = token_manager self.data_collector = data_collector self.github_integration = GithubIntegration( auth=Auth.AppAuth(GITHUB_APP_CLIENT_ID, GITHUB_APP_PRIVATE_KEY) ) self.jinja_env = Environment( loader=FileSystemLoader(OPENHANDS_RESOLVER_TEMPLATES_DIR + 'github') ) def _confirm_incoming_source_type(self, message: Message): if message.source != SourceType.GITHUB: raise ValueError(f'Unexpected message source {message.source}') def _get_full_repo_name(self, repo_obj: dict) -> str: owner = repo_obj['owner']['login'] repo_name = repo_obj['name'] return f'{owner}/{repo_name}' def _get_installation_access_token(self, installation_id: int) -> str: token_data = self.github_integration.get_access_token(installation_id) return token_data.token def _add_reaction( self, github_view: ResolverViewInterface, reaction: str, installation_token: str ): """Add a reaction to the GitHub issue, PR, or comment. Args: github_view: The GitHub view object containing issue/PR/comment info reaction: The reaction to add (e.g. "eyes", "+1", "-1", "laugh", "confused", "heart", "hooray", "rocket") installation_token: GitHub installation access token for API access """ with Github(auth=Auth.Token(installation_token)) as github_client: repo = github_client.get_repo(github_view.full_repo_name) # Add reaction based on view type if isinstance(github_view, GithubInlinePRComment): pr = repo.get_pull(github_view.issue_number) inline_comment = pr.get_review_comment(github_view.comment_id) inline_comment.create_reaction(reaction) elif isinstance(github_view, (GithubIssueComment, GithubPRComment)): issue = repo.get_issue(github_view.issue_number) comment = issue.get_comment(github_view.comment_id) comment.create_reaction(reaction) else: issue = repo.get_issue(github_view.issue_number) issue.create_reaction(reaction) def _user_has_write_access_to_repo( self, installation_id: str, full_repo_name: str, username: str ) -> bool: """Check if the user is an owner, collaborator, or member of the repository.""" with self.github_integration.get_github_for_installation( installation_id, # type: ignore[arg-type] {}, ) as repos: repository = repos.get_repo(full_repo_name) # Check if the user is a collaborator try: collaborator = repository.get_collaborator_permission(username) if collaborator in ['admin', 'write']: return True except Exception: pass # If the above fails, check if the user is an owner or member org = repository.organization if org: user = org.get_members(username) return user is not None return False def _get_issue_number_from_payload(self, message: Message) -> int | None: """Extract issue/PR number from a GitHub webhook payload. Supports all event types that can trigger jobs: - Labeled issues: payload['issue']['number'] - Issue comments: payload['issue']['number'] - PR comments: payload['issue']['number'] (PRs are accessed via issue endpoint) - Inline PR comments: payload['pull_request']['number'] Args: message: The incoming GitHub webhook message Returns: The issue/PR number, or None if not found """ payload = message.message.get('payload', {}) # Labeled issues, issue comments, and PR comments all have 'issue' in payload if 'issue' in payload: return payload['issue']['number'] # Inline PR comments have 'pull_request' directly in payload if 'pull_request' in payload: return payload['pull_request']['number'] return None def _send_user_not_found_message(self, message: Message, username: str): """Send a message to the user informing them they need to create an OpenHands account. This method handles all supported trigger types: - Labeled issues (action='labeled' with openhands label) - Issue comments (comment containing @openhands) - PR comments (comment containing @openhands on a PR) - Inline PR review comments (comment containing @openhands) Args: message: The incoming GitHub webhook message username: The GitHub username to mention in the response """ payload = message.message.get('payload', {}) installation_id = message.message['installation'] repo_obj = payload['repository'] full_repo_name = self._get_full_repo_name(repo_obj) # Get installation token to post the comment installation_token = self._get_installation_access_token(installation_id) # Determine the issue/PR number based on the event type issue_number = self._get_issue_number_from_payload(message) if not issue_number: logger.warning( f'[GitHub] Could not determine issue/PR number to send user not found message for {username}. ' f'Payload keys: {list(payload.keys())}' ) return # Post the comment try: with Github(auth=Auth.Token(installation_token)) as github_client: repo = github_client.get_repo(full_repo_name) issue = repo.get_issue(number=issue_number) issue.create_comment(get_user_not_found_message(username)) except Exception as e: logger.error( f'[GitHub] Failed to send user not found message to {username} ' f'on {full_repo_name}#{issue_number}: {e}' ) async def is_job_requested(self, message: Message) -> bool: self._confirm_incoming_source_type(message) installation_id = message.message['installation'] payload = message.message.get('payload', {}) repo_obj = payload.get('repository') if not repo_obj: return False username = payload.get('sender', {}).get('login') repo_name = self._get_full_repo_name(repo_obj) # Suggestions contain `@openhands` macro; avoid kicking off jobs for system recommendations if GithubFactory.is_pr_comment( message ) and GithubFailingAction.unqiue_suggestions_header in payload.get( 'comment', {} ).get('body', ''): return False # Check event types before making expensive API calls (e.g., _user_has_write_access_to_repo) if not ( GithubFactory.is_labeled_issue(message) or GithubFactory.is_issue_comment(message) or GithubFactory.is_pr_comment(message) or GithubFactory.is_inline_pr_comment(message) ): return False logger.info(f'[GitHub] Checking permissions for {username} in {repo_name}') user_has_write_access = self._user_has_write_access_to_repo( installation_id, repo_name, username ) if ( GithubFactory.is_eligible_for_conversation_starter(message) and user_has_write_access ): await GithubFactory.trigger_conversation_starter(message) return user_has_write_access async def receive_message(self, message: Message): self._confirm_incoming_source_type(message) try: await self.data_collector.process_payload(message) except Exception: logger.warning( '[Github]: Error processing payload for gh interaction', exc_info=True ) if await self.is_job_requested(message): payload = message.message.get('payload', {}) user_id = payload['sender']['id'] username = payload['sender']['login'] keycloak_user_id = await self.token_manager.get_user_id_from_idp_user_id( user_id, ProviderType.GITHUB ) # Check if the user has an OpenHands account if not keycloak_user_id: logger.warning( f'[GitHub] User {username} (id={user_id}) not found in Keycloak. ' f'User must create an OpenHands account first.' ) self._send_user_not_found_message(message, username) return github_view = await GithubFactory.create_github_view_from_payload( message, keycloak_user_id ) logger.info( f'[GitHub] Creating job for {github_view.user_info.username} in {github_view.full_repo_name}#{github_view.issue_number}' ) # Get the installation token installation_token = self._get_installation_access_token( github_view.installation_id ) # Store the installation token await self.token_manager.store_org_token( github_view.installation_id, installation_token ) # Add eyes reaction to acknowledge we've read the request self._add_reaction(github_view, 'eyes', installation_token) await self.start_job(github_view) async def send_message(self, message: str, github_view: GithubViewType): """Send a message to GitHub. Args: message: The message content to send (plain text string) github_view: The GitHub view object containing issue/PR/comment info """ installation_token = await self.token_manager.load_org_token( github_view.installation_id ) if not installation_token: logger.warning('Missing installation token') return if isinstance(github_view, GithubInlinePRComment): with Github(auth=Auth.Token(installation_token)) as github_client: repo = github_client.get_repo(github_view.full_repo_name) pr = repo.get_pull(github_view.issue_number) pr.create_review_comment_reply( comment_id=github_view.comment_id, body=message ) elif isinstance( github_view, (GithubPRComment, GithubIssueComment, GithubIssue) ): with Github(auth=Auth.Token(installation_token)) as github_client: repo = github_client.get_repo(github_view.full_repo_name) issue = repo.get_issue(number=github_view.issue_number) issue.create_comment(message) else: # Catch any new types added to GithubViewType that aren't handled above logger.warning( # type: ignore[unreachable] f'Unsupported github_view type: {type(github_view).__name__}' ) return async def start_job(self, github_view: GithubViewType) -> None: """Kick off a job with openhands agent using V1 app conversation system. 1. Get user credential 2. Initialize new conversation with repo 3. Save interaction data """ try: msg_info: str = '' try: user_info = github_view.user_info logger.info( f'[GitHub] Starting job for user {user_info.username} (id={user_info.user_id})' ) # Create conversation user_token = await self.token_manager.get_idp_token_from_idp_user_id( str(user_info.user_id), ProviderType.GITHUB ) if not user_token: logger.warning( f'[GitHub] No token found for user {user_info.username} (id={user_info.user_id})' ) raise MissingSettingsError('Missing settings') logger.info( f'[GitHub] Creating new conversation for user {user_info.username}' ) secret_store = Secrets( provider_tokens=MappingProxyType( { ProviderType.GITHUB: ProviderToken( token=SecretStr(user_token), user_id=str(user_info.user_id), ) } ) ) conversation_id = await github_view.initialize_new_conversation() saas_user_auth = await get_saas_user_auth( github_view.user_info.keycloak_user_id, self.token_manager ) await github_view.create_new_conversation( self.jinja_env, secret_store.provider_tokens, conversation_id, saas_user_auth, ) conversation_id_hex = github_view.conversation_id logger.info( f'[GitHub] Created conversation {conversation_id_hex} for user {user_info.username}' ) # V1 callback processors are registered by the view during conversation creation # Send message with conversation link conversation_link = CONVERSATION_URL.format(conversation_id_hex) msg_info = f"I'm on it! {user_info.username} can [track my progress at all-hands.dev]({conversation_link})" except MissingSettingsError as e: logger.warning( f'[GitHub] Missing settings error for user {user_info.username}: {str(e)}' ) msg_info = f'@{user_info.username} please re-login into [OpenHands Cloud]({HOST_URL}) before starting a job.' except LLMAuthenticationError as e: logger.warning( f'[GitHub] LLM authentication error for user {user_info.username}: {str(e)}' ) msg_info = f'@{user_info.username} please set a valid LLM API key in [OpenHands Cloud]({HOST_URL}) before starting a job.' except (AuthenticationError, ExpiredError, SessionExpiredError) as e: logger.warning( f'[GitHub] Session expired for user {user_info.username}: {str(e)}' ) msg_info = get_session_expired_message(user_info.username) await self.send_message(msg_info, github_view) except Exception: logger.exception('[Github]: Error starting job') await self.send_message( 'Uh oh! There was an unexpected error starting the job :(', github_view ) try: await self.data_collector.save_data(github_view) except Exception: logger.warning('[Github]: Error saving interaction data', exc_info=True)