import logging import os import shlex import tempfile from abc import ABC from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any, AsyncGenerator from uuid import UUID if TYPE_CHECKING: import httpx import base62 from openhands.app_server.app_conversation.app_conversation_models import ( AgentType, AppConversationStartTask, AppConversationStartTaskStatus, ) from openhands.app_server.app_conversation.app_conversation_service import ( AppConversationService, ) from openhands.app_server.app_conversation.skill_loader import ( build_org_config, build_sandbox_config, load_skills_from_agent_server, ) from openhands.app_server.integrations.service_types import ProviderType from openhands.app_server.sandbox.sandbox_models import SandboxInfo from openhands.app_server.user.user_context import UserContext from openhands.app_server.utils.auth import looks_like_jwt from openhands.app_server.utils.git import ensure_valid_git_branch_name from openhands.sdk import Agent, LLMSummarizingCondenser from openhands.sdk.context import AgentContext from openhands.sdk.llm import LLM from openhands.sdk.security import ( AlwaysConfirm, ConfirmationPolicyBase, ConfirmRisky, LLMSecurityAnalyzer, NeverConfirm, SecurityAnalyzerBase, ) from openhands.sdk.skills import Skill from openhands.sdk.workspace.remote.async_remote_workspace import AsyncRemoteWorkspace _logger = logging.getLogger(__name__) PRE_COMMIT_HOOK = '.git/hooks/pre-commit' PRE_COMMIT_LOCAL = '.git/hooks/pre-commit.local' def get_project_dir( working_dir: str, selected_repository: str | None = None, ) -> str: """Get the project root directory for a conversation. When a repository is selected, the project root is the cloned repo directory at {working_dir}/{repo_name}. This is the directory that contains the `.openhands/` configuration (setup.sh, pre-commit.sh, skills/, etc.). Without a repository, the project root is the working_dir itself. This must be used consistently for ALL features that depend on the project root: - workspace.working_dir (terminal CWD, file editor root, etc.) - .openhands/setup.sh execution - .openhands/pre-commit.sh (git hooks setup) - .openhands/skills/ (project skills) - PLAN.md path Args: working_dir: Base working directory path in the sandbox (e.g., '/workspace/project' from sandbox_spec) selected_repository: Repository name (e.g., 'OpenHands/software-agent-sdk') If provided, the repo name is appended to working_dir. Returns: The project root directory path. """ if selected_repository: repo_name = selected_repository.split('/')[-1] return f'{working_dir}/{repo_name}' return working_dir @dataclass class AppConversationServiceBase(AppConversationService, ABC): """App Conversation service which adds git specific functionality. Sets up repositories and installs hooks """ init_git_in_empty_workspace: bool user_context: UserContext async def load_and_merge_all_skills( self, sandbox: SandboxInfo, selected_repository: str | None, project_dir: str, agent_server_url: str, ) -> list[Skill]: """Load skills from all sources via the agent-server. This method calls the agent-server's /api/skills endpoint to load and merge skills from all sources. The agent-server handles: - Public skills (from OpenHands/skills GitHub repo) - User skills (from ~/.openhands/skills/) - Organization skills (from {org}/.openhands repo) - Project/repo skills (from repo .agents/skills/, .openhands/microagents/, and legacy .openhands/skills/) - Sandbox skills (from exposed URLs) Args: sandbox: SandboxInfo containing exposed URLs and agent-server URL selected_repository: Repository name or None project_dir: Project root directory (resolved via get_project_dir). agent_server_url: Agent-server URL (required) Returns: List of merged Skill objects from all sources, or empty list on failure """ try: _logger.debug('Loading skills for V1 conversation via agent-server') if not agent_server_url: _logger.warning('No agent-server URL available, cannot load skills') return [] # Build org config (authentication handled by app-server) org_config = await build_org_config(selected_repository, self.user_context) # Build sandbox config (exposed URLs) sandbox_config = build_sandbox_config(sandbox) # Single API call to agent-server for ALL skills all_skills = await load_skills_from_agent_server( agent_server_url=agent_server_url, session_api_key=sandbox.session_api_key, project_dir=project_dir, org_config=org_config, sandbox_config=sandbox_config, load_public=True, load_user=True, load_project=True, load_org=True, ) _logger.info( f'Loaded {len(all_skills)} total skills from agent-server: ' f'{[s.name for s in all_skills]}' ) return all_skills except Exception as e: _logger.warning(f'Failed to load skills: {e}', exc_info=True) # Return empty list on failure - skills will be loaded again later if needed return [] def _create_agent_with_skills(self, agent, skills: list[Skill]): """Create or update agent with skills in its context. Args: agent: The agent to update skills: List of Skill objects to add to agent context Returns: Updated agent with skills in context """ if agent.agent_context: # Merge with existing context (new skills override existing ones) existing_skills = agent.agent_context.skills all_skills = self._merge_skills([existing_skills, skills]) agent = agent.model_copy( update={ 'agent_context': agent.agent_context.model_copy( update={'skills': all_skills} ) } ) else: # Create new context agent_context = AgentContext(skills=skills) agent = agent.model_copy(update={'agent_context': agent_context}) return agent def _merge_skills(self, skill_lists: list[list[Skill]]) -> list[Skill]: """Merge multiple skill lists, avoiding duplicates by name. Later lists take precedence over earlier lists for duplicate names. Args: skill_lists: List of skill lists to merge Returns: Deduplicated list of skills with later lists overriding earlier ones """ skills_by_name: dict[str, Skill] = {} for skill_list in skill_lists: for skill in skill_list: skills_by_name[skill.name] = skill return list(skills_by_name.values()) async def _load_skills_and_update_agent( self, sandbox: SandboxInfo, agent: Agent, remote_workspace: AsyncRemoteWorkspace, selected_repository: str | None, project_dir: str, disabled_skills: list[str] | None = None, ): """Load all skills and update agent with them. Args: agent: The agent to update remote_workspace: AsyncRemoteWorkspace for loading repo skills selected_repository: Repository name or None (used for org config) project_dir: Project root directory (already resolved via get_project_dir). disabled_skills: Optional list of skill names to exclude Returns: Updated agent with skills loaded into context """ agent_server_url = remote_workspace.host all_skills = await self.load_and_merge_all_skills( sandbox, selected_repository, project_dir, agent_server_url, ) # Filter out disabled skills if disabled_skills: disabled_set = set(disabled_skills) all_skills = [s for s in all_skills if s.name not in disabled_set] # Update agent with skills agent = self._create_agent_with_skills(agent, all_skills) return agent async def run_setup_scripts( self, task: AppConversationStartTask, sandbox: SandboxInfo, workspace: AsyncRemoteWorkspace, agent_server_url: str, ) -> AsyncGenerator[AppConversationStartTask, None]: task.status = AppConversationStartTaskStatus.PREPARING_REPOSITORY yield task await self.clone_or_init_git_repo(task, workspace, sandbox) # Compute the project root — the cloned repo directory when a repo is # selected, or the sandbox working_dir otherwise. This must be used # for all .openhands/ features (setup.sh, pre-commit.sh, skills). project_dir = get_project_dir( workspace.working_dir, task.request.selected_repository ) task.status = AppConversationStartTaskStatus.RUNNING_SETUP_SCRIPT yield task await self.maybe_run_setup_script(workspace, project_dir) task.status = AppConversationStartTaskStatus.SETTING_UP_GIT_HOOKS yield task await self.maybe_setup_git_hooks(workspace, project_dir) task.status = AppConversationStartTaskStatus.SETTING_UP_SKILLS yield task await self.load_and_merge_all_skills( sandbox, task.request.selected_repository, project_dir, agent_server_url, ) async def _configure_git_user_settings( self, workspace: AsyncRemoteWorkspace, ) -> None: """Configure git global user settings from user preferences. Reads git_user_name and git_user_email from user settings and configures them as git global settings in the workspace. Args: workspace: The remote workspace to configure git settings in. """ try: user_info = await self.user_context.get_user_info() if user_info.git_user_name: cmd = f'git config --global user.name "{user_info.git_user_name}"' result = await workspace.execute_command(cmd, workspace.working_dir) if result.exit_code: _logger.warning(f'Git config user.name failed: {result.stderr}') else: _logger.info( f'Git configured with user.name={user_info.git_user_name}' ) if user_info.git_user_email: cmd = f'git config --global user.email "{user_info.git_user_email}"' result = await workspace.execute_command(cmd, workspace.working_dir) if result.exit_code: _logger.warning(f'Git config user.email failed: {result.stderr}') else: _logger.info( f'Git configured with user.email={user_info.git_user_email}' ) except Exception as e: _logger.warning(f'Failed to configure git user settings: {e}') async def clone_or_init_git_repo( self, task: AppConversationStartTask, workspace: AsyncRemoteWorkspace, sandbox: SandboxInfo | None = None, ): request = task.request # Create the projects directory if it does not exist yet parent = Path(workspace.working_dir).parent result = await workspace.execute_command( f'mkdir -p {workspace.working_dir}', parent ) if result.exit_code: _logger.warning(f'mkdir failed: {result.stderr}') # Configure git user settings from user preferences await self._configure_git_user_settings(workspace) if not request.selected_repository: if self.init_git_in_empty_workspace: _logger.debug('Initializing a new git repository in the workspace.') cmd = ( 'git init && git config --global ' f'--add safe.directory {workspace.working_dir}' ) result = await workspace.execute_command(cmd, workspace.working_dir) if result.exit_code: _logger.warning(f'Git init failed: {result.stderr}') else: _logger.info('Not initializing a new git repository.') return remote_repo_url: str = await self.user_context.get_authenticated_git_url( request.selected_repository ) if not remote_repo_url: raise ValueError('Missing either Git token or valid repository') dir_name = request.selected_repository.split('/')[-1] quoted_remote_repo_url = shlex.quote(remote_repo_url) quoted_dir_name = shlex.quote(dir_name) git_dir = Path(workspace.working_dir) / dir_name azure_devops_bearer_token = await self._get_azure_devops_bearer_token_for_git( request.git_provider, remote_repo_url, ) # Clone the repo - this is the slow part! if azure_devops_bearer_token: auth_header = shlex.quote( f'Authorization: Bearer {azure_devops_bearer_token}' ) clone_command = ( f'git -c http.extraheader={auth_header} clone ' f'{quoted_remote_repo_url} {quoted_dir_name}' ) else: clone_command = f'git clone {quoted_remote_repo_url} {quoted_dir_name}' result = await workspace.execute_command( clone_command, workspace.working_dir, 120 ) if result.exit_code: _logger.warning(f'Git clone failed: {result.stderr}') elif azure_devops_bearer_token: await self._configure_azure_devops_git_credential_helper( workspace, git_dir, request.selected_repository, sandbox, ) # Checkout the appropriate branch if request.selected_branch: ensure_valid_git_branch_name(request.selected_branch) checkout_command = f'git checkout {shlex.quote(request.selected_branch)}' else: # Generate a random branch name to avoid conflicts random_str = base62.encodebytes(os.urandom(16)) openhands_workspace_branch = f'openhands-workspace-{random_str}' checkout_command = ( f'git checkout -b {shlex.quote(openhands_workspace_branch)}' ) result = await workspace.execute_command(checkout_command, git_dir) if result.exit_code: _logger.warning(f'Git checkout failed: {result.stderr}') async def _get_azure_devops_bearer_token_for_git( self, git_provider: ProviderType | None, remote_repo_url: str, ) -> str | None: if ( git_provider != ProviderType.AZURE_DEVOPS and 'dev.azure.com' not in remote_repo_url ): return None try: token = await self.user_context.get_latest_token(ProviderType.AZURE_DEVOPS) except Exception as exc: _logger.warning(f'Failed to get Azure DevOps token for git: {exc}') return None if token and looks_like_jwt(token): return token return None async def _configure_azure_devops_git_credential_helper( self, workspace: AsyncRemoteWorkspace, git_dir: Path, selected_repository: str, sandbox: SandboxInfo | None, ) -> None: if sandbox is None: _logger.warning( 'Skipping Azure DevOps git credential helper setup: missing sandbox' ) return org = selected_repository.split('/')[0] helper_path = git_dir / '.git' / 'openhands-azure-devops-credential-helper' secret_path = ( f'/api/v1/sandboxes/{sandbox.id}/settings/secrets/azure_devops_token' ) web_url = getattr(self, 'web_url', None) if web_url is None: _logger.debug( 'Azure DevOps git credential helper has no configured web_url; ' 'it will rely on OH_WEBHOOKS_0_BASE_URL at runtime.' ) app_base_url = shlex.quote(web_url if isinstance(web_url, str) else '') helper_script = f"""#!/bin/sh if [ "$1" != "get" ]; then exit 0 fi session_api_key="${{OH_SESSION_API_KEYS_0:-${{SESSION_API_KEY:-}}}}" webhook_url="${{OH_WEBHOOKS_0_BASE_URL:-}}" app_base_url={app_base_url} if [ -n "$webhook_url" ]; then base_url="${{webhook_url%/api/v1/webhooks}}" else base_url="$app_base_url" fi if [ -z "$session_api_key" ] || [ -z "$base_url" ]; then exit 0 fi secret_url="$base_url{secret_path}" token="$(curl -fsS \\ -H "X-Session-API-Key: $session_api_key" \\ "$secret_url" 2>/dev/null)" || exit 0 if [ -z "$token" ]; then exit 0 fi printf '%s\\n' "username=oauth2" printf 'password=%s\\n' "$token" """ helper_path_quoted = shlex.quote(str(helper_path)) helper_value = shlex.quote(f'!{shlex.quote(str(helper_path))}') extraheader_key = shlex.quote(f'http.https://dev.azure.com/{org}/.extraheader') credential_username_key = shlex.quote( f'credential.https://dev.azure.com/{org}.username' ) credential_helper_key = shlex.quote( f'credential.https://dev.azure.com/{org}.helper' ) command = ( f'printf %s {shlex.quote(helper_script)} > {helper_path_quoted}' f' && chmod 700 {helper_path_quoted}' f' && (git config --local --unset-all {extraheader_key} || true)' f' && git config --local {credential_username_key} oauth2' f' && git config --local {credential_helper_key} {helper_value}' ' && git config --local credential.useHttpPath true' ) result = await workspace.execute_command(command, git_dir) if result.exit_code: _logger.warning( f'Azure DevOps git credential helper setup failed: {result.stderr}' ) async def maybe_run_setup_script( self, workspace: AsyncRemoteWorkspace, project_dir: str, ): """Run .openhands/setup.sh if it exists in the project root. Args: workspace: Remote workspace for command execution. project_dir: Project root directory (repo root when a repo is selected). """ setup_script = project_dir + '/.openhands/setup.sh' await workspace.execute_command( f'chmod +x {setup_script} && source {setup_script}', cwd=project_dir, timeout=600, ) async def maybe_setup_git_hooks( self, workspace: AsyncRemoteWorkspace, project_dir: str, ): """Set up git hooks if .openhands/pre-commit.sh exists in the project root. Args: workspace: Remote workspace for command execution. project_dir: Project root directory (repo root when a repo is selected). """ command = 'mkdir -p .git/hooks && chmod +x .openhands/pre-commit.sh' pre_commit_command_result = await workspace.execute_command( command, project_dir ) if pre_commit_command_result.exit_code: return # Check if there's an existing pre-commit hook with tempfile.TemporaryFile(mode='w+t') as temp_file: download_result = await workspace.file_download( PRE_COMMIT_HOOK, str(temp_file) ) if download_result.success: _logger.info('Preserving existing pre-commit hook') # an existing pre-commit hook exists if 'This hook was installed by OpenHands' not in temp_file.read(): # Move the existing hook to pre-commit.local command = ( f'mv {PRE_COMMIT_HOOK} {PRE_COMMIT_LOCAL} &&' f'chmod +x {PRE_COMMIT_LOCAL}' ) mv_chmod_result = await workspace.execute_command( command, project_dir ) if mv_chmod_result.exit_code != 0: _logger.error( f'Failed to preserve existing pre-commit hook: {mv_chmod_result.stderr}', ) return # write the pre-commit hook await workspace.file_upload( source_path=Path(__file__).parent / 'git' / 'pre-commit.sh', destination_path=PRE_COMMIT_HOOK, ) # Make the pre-commit hook executable chmod_result = await workspace.execute_command(f'chmod +x {PRE_COMMIT_HOOK}') if chmod_result.exit_code: _logger.error( f'Failed to make pre-commit hook executable: {chmod_result.stderr}' ) return _logger.info('Git pre-commit hook installed successfully') def _create_condenser( self, llm: LLM, agent_type: AgentType, condenser_max_size: int | None, ) -> LLMSummarizingCondenser: """Create a condenser based on user settings and agent type. Args: llm: The LLM instance to use for condensation agent_type: Type of agent (PLAN or DEFAULT) condenser_max_size: condenser_max_size setting Returns: Configured LLMSummarizingCondenser instance """ # LLMSummarizingCondenser SDK defaults: max_size=240, keep_first=2 condenser_kwargs: dict[str, Any] = { 'llm': llm.model_copy( update={ 'usage_id': ( 'condenser' if agent_type == AgentType.DEFAULT else 'planning_condenser' ) } ), } # Only override max_size if user has a custom value if condenser_max_size is not None: condenser_kwargs['max_size'] = condenser_max_size condenser = LLMSummarizingCondenser(**condenser_kwargs) return condenser def _create_security_analyzer_from_string( self, security_analyzer_str: str | None ) -> SecurityAnalyzerBase | None: """Convert security analyzer string from settings to SecurityAnalyzerBase instance. Args: security_analyzer_str: String value from settings. Valid values: - "llm" -> LLMSecurityAnalyzer - "none" or None -> None - Other values -> None (unsupported analyzers are ignored) Returns: SecurityAnalyzerBase instance or None """ if not security_analyzer_str or security_analyzer_str.lower() == 'none': return None if security_analyzer_str.lower() == 'llm': return LLMSecurityAnalyzer() # For unknown values, log a warning and return None _logger.warning( f'Unknown security analyzer value: {security_analyzer_str}. ' 'Supported values: "llm", "none". Defaulting to None.' ) return None def _select_confirmation_policy( self, confirmation_mode: bool, security_analyzer: str | None ) -> ConfirmationPolicyBase: """Choose confirmation policy using only mode flag and analyzer string.""" if not confirmation_mode: return NeverConfirm() analyzer_kind = (security_analyzer or '').lower() if analyzer_kind == 'llm': return ConfirmRisky() return AlwaysConfirm() async def _set_security_analyzer_from_settings( self, agent_server_url: str, session_api_key: str | None, conversation_id: UUID, security_analyzer_str: str | None, httpx_client: 'httpx.AsyncClient', ) -> None: """Set security analyzer on conversation using only the analyzer string. Args: agent_server_url: URL of the agent server session_api_key: Session API key for authentication conversation_id: ID of the conversation to update security_analyzer_str: String value from settings httpx_client: HTTP client for making API requests """ if session_api_key is None: return security_analyzer = self._create_security_analyzer_from_string( security_analyzer_str ) # Only make API call if we have a security analyzer to set # (None is the default, so we can skip the call if it's None) if security_analyzer is None: return try: # Prepare the request payload payload = {'security_analyzer': security_analyzer.model_dump()} # Call agent server API to set security analyzer response = await httpx_client.post( f'{agent_server_url}/api/conversations/{conversation_id}/security_analyzer', json=payload, headers={'X-Session-API-Key': session_api_key}, timeout=30.0, ) response.raise_for_status() _logger.info( f'Successfully set security analyzer for conversation {conversation_id}' ) except Exception as e: # Log error but don't fail conversation creation _logger.warning( f'Failed to set security analyzer for conversation {conversation_id}: {e}', exc_info=True, )