"""Unit tests for the methods in LiveStatusAppConversationService.""" import io import json import os import zipfile from datetime import datetime from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch from uuid import uuid4 import pytest from pydantic import SecretStr from wren.agent_server.models import ( SendMessageRequest, StartConversationRequest, TextContent, ) from wren.app_server.app_conversation.app_conversation_models import ( AgentType, AppConversationInfo, AppConversationStartRequest, ConversationTrigger, ) from wren.app_server.app_conversation.app_conversation_service import ( ConversationExportAlreadyRunning, ConversationExportLockUnavailable, ConversationExportTooLarge, ) from wren.app_server.app_conversation.live_status_app_conversation_service import ( LiveStatusAppConversationService, ) from wren.app_server.integrations.provider import ProviderToken, ProviderType from wren.app_server.integrations.service_types import SuggestedTask, TaskType from wren.app_server.sandbox.sandbox_models import ( AGENT_SERVER, ExposedUrl, SandboxInfo, SandboxPage, SandboxStatus, ) from wren.app_server.sandbox.sandbox_spec_models import SandboxSpecInfo from wren.app_server.settings.llm_profiles import LLMProfiles from wren.app_server.settings.settings_models import ( SandboxGroupingStrategy, Settings, ) from wren.app_server.user.user_context import UserContext from wren.app_server.utils.redis_lock import RedisLockUnavailable from wren.sdk import Agent, AgentContext, Event from wren.sdk.llm import LLM from wren.sdk.secret import LookupSecret, StaticSecret from wren.sdk.settings import ConversationSettings, OpenHandsAgentSettings from wren.sdk.workspace.remote.async_remote_workspace import AsyncRemoteWorkspace def _async_iter(items): async def iter_items(): for item in items: yield item return iter_items() class _FakeExportLock: def __init__(self): self.name = 'fake-lock' self.released = False self.reacquired = False async def reacquire(self): self.reacquired = True async def release(self): self.released = True def _build_test_user_agent_settings(user: SimpleNamespace) -> OpenHandsAgentSettings: llm_vals: dict = {} model = getattr(user, 'llm_model', '') or '' llm_vals['model'] = model llm_api_key = getattr(user, 'llm_api_key', None) if llm_api_key: llm_vals['api_key'] = llm_api_key llm_base_url = getattr(user, 'llm_base_url', None) if llm_base_url: llm_vals['base_url'] = llm_base_url agent_vals: dict = {'llm': llm_vals} mcp_config = getattr(user, '_mcp_config', None) or getattr(user, 'mcp_config', None) if mcp_config: agent_vals['mcp_config'] = mcp_config.model_dump(mode='python') return Settings(agent_settings=agent_vals).agent_settings class _TestUserInfo(SimpleNamespace): @property def agent_settings(self) -> OpenHandsAgentSettings: override = getattr(self, '_agent_settings_override', None) if override is not None: return override return _build_test_user_agent_settings(self) @agent_settings.setter def agent_settings(self, value): object.__setattr__(self, '_agent_settings_override', value) @property def llm_profiles(self) -> LLMProfiles: # Real UserInfo always carries llm_profiles; default to empty unless a # test sets profiles. override = getattr(self, '_llm_profiles_override', None) if override is not None: return override return LLMProfiles(profiles={}) @llm_profiles.setter def llm_profiles(self, value): object.__setattr__(self, '_llm_profiles_override', value) @property def registered_marketplaces(self) -> list: # Real UserInfo (via Settings) always carries registered_marketplaces, # defaulting to an empty list. Mirror that so marketplace composition at # conversation start doesn't raise AttributeError on the fake. override = getattr(self, '_registered_marketplaces_override', None) if override is not None: return override return [] @registered_marketplaces.setter def registered_marketplaces(self, value): object.__setattr__(self, '_registered_marketplaces_override', value) @property def conversation_settings(self) -> ConversationSettings: kwargs: dict = { 'confirmation_mode': getattr(self, 'confirmation_mode', False), 'security_analyzer': getattr(self, 'security_analyzer', None), } max_iter = getattr(self, 'max_iterations', None) if max_iter is not None: kwargs['max_iterations'] = max_iter return ConversationSettings(**kwargs) def to_agent_settings(self) -> OpenHandsAgentSettings: return self.agent_settings # Env var used by wren SDK LLM to skip context-window validation (e.g. for gpt-4 in tests) _ALLOW_SHORT_CONTEXT_WINDOWS = 'ALLOW_SHORT_CONTEXT_WINDOWS' @pytest.fixture(autouse=True) def allow_short_context_windows(): """Allow small context windows so unit tests can create LLM with gpt-4 etc.""" old = os.environ.pop(_ALLOW_SHORT_CONTEXT_WINDOWS, None) os.environ[_ALLOW_SHORT_CONTEXT_WINDOWS] = 'true' try: yield finally: if old is not None: os.environ[_ALLOW_SHORT_CONTEXT_WINDOWS] = old else: os.environ.pop(_ALLOW_SHORT_CONTEXT_WINDOWS, None) class TestLiveStatusAppConversationService: """Test cases for the methods in LiveStatusAppConversationService.""" def setup_method(self): """Set up test fixtures.""" # Create mock dependencies self.mock_user_context = Mock(spec=UserContext) self.mock_user_auth = Mock() self.mock_user_context.user_auth = self.mock_user_auth self.mock_user_context.get_user_email = AsyncMock(return_value=None) self.mock_jwt_service = Mock() self.mock_sandbox_service = Mock() self.mock_sandbox_spec_service = Mock() self.mock_app_conversation_info_service = Mock() self.mock_app_conversation_start_task_service = Mock() self.mock_event_callback_service = Mock() self.mock_event_service = Mock() self.mock_httpx_client = Mock() self.mock_pending_message_service = Mock() # Create service instance self.service = LiveStatusAppConversationService( init_git_in_empty_workspace=True, user_context=self.mock_user_context, app_conversation_info_service=self.mock_app_conversation_info_service, app_conversation_start_task_service=self.mock_app_conversation_start_task_service, event_callback_service=self.mock_event_callback_service, event_service=self.mock_event_service, sandbox_service=self.mock_sandbox_service, sandbox_spec_service=self.mock_sandbox_spec_service, jwt_service=self.mock_jwt_service, pending_message_service=self.mock_pending_message_service, sandbox_startup_timeout=30, sandbox_startup_poll_frequency=1, max_num_conversations_per_sandbox=20, httpx_client=self.mock_httpx_client, web_url='https://test.example.com', wren_provider_base_url='https://provider.example.com', access_token_hard_timeout=None, app_mode='test', ) # Mock user info self.mock_user = _TestUserInfo( id='test_user_123', llm_model='gpt-4', llm_base_url='https://api.openai.com/v1', llm_api_key='test_api_key', sandbox_grouping_strategy=SandboxGroupingStrategy.ADD_TO_ANY, confirmation_mode=False, security_analyzer='llm', search_api_key=None, mcp_config=None, disabled_skills=[], git_full_clone=False, ) # Mock sandbox self.mock_sandbox = Mock(spec=SandboxInfo) self.mock_sandbox.id = uuid4() self.mock_sandbox.status = SandboxStatus.RUNNING # Stable conversation ID for tests that call _configure_llm_and_mcp directly self.conversation_id = uuid4() # Default mock for hooks loading - returns None (no hooks found) # Tests that specifically test hooks loading can override this mock self.service._load_hooks_from_workspace = AsyncMock(return_value=None) @pytest.mark.asyncio async def test_seed_sandbox_profiles_upserts_resolved_keys_and_prunes(self): """Pushes each profile to the sandbox with its key resolved (managed key injected, BYOR key kept), then deletes profiles that no longer exist on the app-server. """ user = SimpleNamespace( llm_profiles=LLMProfiles( profiles={ 'Managed': LLM(model='litellm_proxy/minimax-m2.7', usage_id='p'), 'BYOR': LLM( model='anthropic/claude-sonnet-4-6', api_key='byor-key', usage_id='p', ), # Org names aren't character-restricted; this one must be # skipped so it can't path-inject the request URL. '../evil': LLM(model='openai/gpt-4o', usage_id='p'), } ), agent_settings=SimpleNamespace( llm=SimpleNamespace(api_key=SecretStr('managed-key')) ), ) self.mock_user_context.get_user_info = AsyncMock(return_value=user) ok = Mock(raise_for_status=Mock()) listing = Mock(raise_for_status=Mock()) listing.json = Mock( return_value={ 'profiles': [{'name': 'Managed'}, {'name': 'BYOR'}, {'name': 'Gone'}] } ) self.mock_httpx_client.post = AsyncMock(return_value=ok) self.mock_httpx_client.get = AsyncMock(return_value=listing) self.mock_httpx_client.delete = AsyncMock(return_value=ok) await self.service._seed_sandbox_profiles('http://agent.test', 'sess-key') base = 'http://agent.test/api/profiles' pushed = { call.args[0]: call.kwargs['json']['llm'] for call in self.mock_httpx_client.post.call_args_list } # Managed profile (no stored key) falls back to the effective key; BYOR # keeps its own. assert pushed[f'{base}/Managed']['api_key'] == 'managed-key' assert pushed[f'{base}/BYOR']['api_key'] == 'byor-key' # The unsafe-named profile is skipped entirely (never POSTed). assert self.mock_httpx_client.post.await_count == 2 assert not any('evil' in url for url in pushed) # The profile deleted on the app-server is pruned from the sandbox. self.mock_httpx_client.delete.assert_awaited_once() assert self.mock_httpx_client.delete.await_args.args[0] == f'{base}/Gone' def test_apply_suggested_task_sets_prompt_and_trigger(self): """Test suggested task prompts populate initial message and trigger.""" suggested_task = SuggestedTask( git_provider=ProviderType.GITHUB, task_type=TaskType.UNRESOLVED_COMMENTS, repo='owner/repo', issue_number=42, title='Handle review comments', ) request = AppConversationStartRequest(suggested_task=suggested_task) self.service._apply_suggested_task(request) assert request.initial_message is not None assert ( request.initial_message.content[0].text == suggested_task.get_prompt_for_task() ) assert request.trigger == ConversationTrigger.SUGGESTED_TASK assert request.selected_repository == suggested_task.repo assert request.git_provider == suggested_task.git_provider def test_apply_suggested_task_raises_if_initial_message_present(self): suggested_task = SuggestedTask( repo='foo/bar', git_provider=ProviderType.GITHUB, title='Some title', task_type=TaskType.OPEN_ISSUE, issue_number=123, ) request = AppConversationStartRequest( suggested_task=suggested_task, initial_message=SendMessageRequest( role='user', content=[TextContent(text='User provided message')], ), ) with pytest.raises(ValueError, match='initial_message cannot be provided'): self.service._apply_suggested_task(request) def test_apply_suggested_task_raises_if_prompt_empty(self): suggested_task = SuggestedTask( repo='foo/bar', git_provider=ProviderType.GITHUB, title='Some title', task_type=TaskType.OPEN_ISSUE, issue_number=123, ) request = AppConversationStartRequest(suggested_task=suggested_task) with patch.object(SuggestedTask, 'get_prompt_for_task', return_value=''): with pytest.raises(ValueError, match='empty prompt'): self.service._apply_suggested_task(request) @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_no_provider_tokens(self): """Test _setup_secrets_for_git_providers with no provider tokens.""" # Arrange # get_secrets returns StaticSecret objects (plain strings are not valid) base_secrets = { 'existing': StaticSecret(value=SecretStr('secret'), description='') } self.mock_user_context.get_secrets.return_value = base_secrets self.mock_user_context.get_provider_tokens = AsyncMock(return_value=None) self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert — custom secrets are passed through as-is (no LookupSecret wrapping) assert 'existing' in result assert isinstance(result['existing'], StaticSecret) assert result['existing'].description == '' self.mock_user_context.get_secrets.assert_called_once() self.mock_user_context.get_provider_tokens.assert_called_once() @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_with_web_url(self): """Test _setup_secrets_for_git_providers with web URL (creates access token).""" # Arrange base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Mock provider tokens provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), ProviderType.GITLAB: ProviderToken(token=SecretStr('gitlab_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert assert 'GITHUB_TOKEN' in result assert 'GITLAB_TOKEN' in result assert isinstance(result['GITHUB_TOKEN'], LookupSecret) assert isinstance(result['GITLAB_TOKEN'], LookupSecret) assert ( result['GITHUB_TOKEN'].url == 'https://test.example.com/api/v1/webhooks/secrets' ) assert result['GITHUB_TOKEN'].headers['X-Access-Token'] == 'test_access_token' # Verify descriptions are included assert result['GITHUB_TOKEN'].description == 'GITHUB authentication token' assert result['GITLAB_TOKEN'].description == 'GITLAB authentication token' # Should be called twice, once for each provider assert self.mock_jwt_service.create_jws_token.call_count == 2 @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_with_saas_mode(self): """Test _setup_secrets_for_git_providers with SaaS mode uses LookupSecret with X-Access-Token.""" # Arrange self.service.app_mode = 'saas' base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Mock provider tokens provider_tokens = { ProviderType.GITLAB: ProviderToken(token=SecretStr('gitlab_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert assert 'GITLAB_TOKEN' in result lookup_secret = result['GITLAB_TOKEN'] assert isinstance(lookup_secret, LookupSecret) assert 'X-Access-Token' in lookup_secret.headers assert lookup_secret.headers['X-Access-Token'] == 'test_access_token' # Verify no cookie is included (authentication is via X-Access-Token only) assert 'Cookie' not in lookup_secret.headers # Verify description is included assert lookup_secret.description == 'GITLAB authentication token' @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_without_web_url(self): """Test _setup_secrets_for_git_providers without web URL (uses static token).""" # Arrange self.service.web_url = None base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_user_context.get_latest_token.return_value = 'static_token_value' # Mock provider tokens provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert assert 'GITHUB_TOKEN' in result assert isinstance(result['GITHUB_TOKEN'], StaticSecret) assert result['GITHUB_TOKEN'].value.get_secret_value() == 'static_token_value' # Verify description is included assert result['GITHUB_TOKEN'].description == 'GITHUB authentication token' self.mock_user_context.get_latest_token.assert_called_once_with( ProviderType.GITHUB ) @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_no_static_token(self): """Test _setup_secrets_for_git_providers when no static token is available.""" # Arrange self.service.web_url = None base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_user_context.get_latest_token.return_value = None # Mock provider tokens provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert assert 'GITHUB_TOKEN' not in result assert result == base_secrets @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_descriptions_included(self): """Test _setup_secrets_for_git_providers includes descriptions for all provider types.""" # Arrange base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Mock provider tokens for multiple providers provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), ProviderType.GITLAB: ProviderToken(token=SecretStr('gitlab_token')), ProviderType.BITBUCKET: ProviderToken(token=SecretStr('bitbucket_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert - verify all secrets have correct descriptions assert 'GITHUB_TOKEN' in result assert isinstance(result['GITHUB_TOKEN'], LookupSecret) assert result['GITHUB_TOKEN'].description == 'GITHUB authentication token' assert 'GITLAB_TOKEN' in result assert isinstance(result['GITLAB_TOKEN'], LookupSecret) assert result['GITLAB_TOKEN'].description == 'GITLAB authentication token' assert 'BITBUCKET_TOKEN' in result assert isinstance(result['BITBUCKET_TOKEN'], LookupSecret) assert result['BITBUCKET_TOKEN'].description == 'BITBUCKET authentication token' @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_static_secret_description(self): """Test _setup_secrets_for_git_providers includes description for StaticSecret.""" # Arrange self.service.web_url = None base_secrets = {} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_user_context.get_latest_token.return_value = 'static_token_value' # Mock provider tokens for multiple providers provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), ProviderType.GITLAB: ProviderToken(token=SecretStr('gitlab_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert - verify StaticSecret objects have descriptions assert 'GITHUB_TOKEN' in result assert isinstance(result['GITHUB_TOKEN'], StaticSecret) assert result['GITHUB_TOKEN'].description == 'GITHUB authentication token' assert 'GITLAB_TOKEN' in result assert isinstance(result['GITLAB_TOKEN'], StaticSecret) assert result['GITLAB_TOKEN'].description == 'GITLAB authentication token' @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_preserves_custom_secret_descriptions( self, ): """Test _setup_secrets_for_git_providers preserves descriptions from custom secrets.""" # Arrange # Mock custom secrets with descriptions custom_secret_with_desc = StaticSecret( value=SecretStr('custom_secret_value'), description='Custom API key for external service', ) custom_secret_no_desc = StaticSecret( value=SecretStr('another_secret_value'), description=None, ) base_secrets = { 'CUSTOM_API_KEY': custom_secret_with_desc, 'ANOTHER_SECRET': custom_secret_no_desc, } self.mock_user_context.get_secrets.return_value = base_secrets self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Mock provider tokens provider_tokens = { ProviderType.GITHUB: ProviderToken(token=SecretStr('github_token')), } self.mock_user_context.get_provider_tokens = AsyncMock( return_value=provider_tokens ) # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert — custom secrets are passed through as-is; descriptions preserved. assert 'CUSTOM_API_KEY' in result assert isinstance(result['CUSTOM_API_KEY'], StaticSecret) assert ( result['CUSTOM_API_KEY'].value.get_secret_value() == 'custom_secret_value' ) assert ( result['CUSTOM_API_KEY'].description == 'Custom API key for external service' ) assert 'ANOTHER_SECRET' in result assert isinstance(result['ANOTHER_SECRET'], StaticSecret) assert ( result['ANOTHER_SECRET'].value.get_secret_value() == 'another_secret_value' ) assert result['ANOTHER_SECRET'].description is None # Verify git provider token is also included assert 'GITHUB_TOKEN' in result assert result['GITHUB_TOKEN'].description == 'GITHUB authentication token' @pytest.mark.asyncio async def test_setup_secrets_for_git_providers_custom_secret_empty_description( self, ): """Test _setup_secrets_for_git_providers handles custom secrets with empty descriptions.""" # Arrange custom_secret_empty_desc = StaticSecret( value=SecretStr('secret_value'), description='', # Empty string description ) base_secrets = {'MY_SECRET': custom_secret_empty_desc} self.mock_user_context.get_secrets.return_value = base_secrets self.mock_user_context.get_provider_tokens = AsyncMock(return_value=None) self.mock_jwt_service.create_jws_token.return_value = 'test_access_token' # Act result = await self.service._setup_secrets_for_git_providers(self.mock_user) # Assert — custom secrets are passed through as-is; empty description preserved assert 'MY_SECRET' in result assert isinstance(result['MY_SECRET'], StaticSecret) assert result['MY_SECRET'].description == '' @pytest.mark.asyncio async def test_configure_llm_and_mcp_with_custom_model(self): """Test _configure_llm_and_mcp with custom LLM model.""" # Arrange custom_model = 'gpt-3.5-turbo' self.mock_user_context.get_mcp_api_key.return_value = 'mcp_api_key' # Act llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, custom_model, self.conversation_id ) # Assert assert isinstance(llm, LLM) assert llm.model == custom_model assert llm.base_url == self.mock_user.llm_base_url assert llm.api_key.get_secret_value() == self.mock_user.llm_api_key assert llm.usage_id == 'agent' assert 'mcpServers' in mcp_config assert 'default' in mcp_config['mcpServers'] assert ( mcp_config['mcpServers']['default']['url'] == 'https://test.example.com/mcp/mcp' ) assert mcp_config['mcpServers']['default']['headers'][ 'X-OpenHands-ServerConversation-ID' ] == str(self.conversation_id) assert ( mcp_config['mcpServers']['default']['headers']['X-Session-API-Key'] == 'mcp_api_key' ) @pytest.mark.asyncio async def test_configure_llm_and_mcp_uses_user_llm_settings(self): """User LLM fields should drive the configured LLM.""" self.mock_user.llm_model = 'sdk-model' self.mock_user.llm_base_url = 'https://sdk-llm.example.com' self.mock_user.llm_api_key = 'test-key' self.mock_user_context.get_mcp_api_key.return_value = None llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) assert llm.model == 'sdk-model' assert llm.base_url == 'https://sdk-llm.example.com' assert llm.stream is True @pytest.mark.asyncio async def test_configure_llm_preserves_reasoning_effort(self): """User-configured reasoning_effort must survive _configure_llm. Previously _configure_llm created a bare LLM() which reset every field to its Pydantic default (reasoning_effort='high'). Users who set reasoning_effort=None (to disable thinking parameters for models whose litellm version cannot handle them) had their override silently discarded. """ from pydantic import SecretStr as _SecretStr user_llm = LLM( model='anthropic/claude-opus-4-7', api_key=_SecretStr('test-key'), reasoning_effort=None, extended_thinking_budget=None, ) agent_settings = self.mock_user.agent_settings.model_copy( update={'llm': user_llm} ) self.mock_user.agent_settings = agent_settings self.mock_user_context.get_mcp_api_key.return_value = None llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) assert llm.model == 'anthropic/claude-opus-4-7' assert llm.reasoning_effort is None, ( '_configure_llm must preserve reasoning_effort=None ' 'instead of resetting it to the SDK default' ) assert llm.extended_thinking_budget is None @pytest.mark.asyncio async def test_configure_llm_and_mcp_wren_model_uses_user_base_url( self, ): """wren/* model uses user's base_url when set.""" # Arrange self.mock_user.llm_model = 'wren/special' self.mock_user.llm_base_url = 'https://user-llm.example.com' self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, self.mock_user.llm_model, self.conversation_id ) # Assert — user base_url takes precedence for wren/ models assert llm.base_url == 'https://user-llm.example.com' @pytest.mark.asyncio async def test_configure_llm_and_mcp_wren_model_falls_back_to_provider_url( self, ): """wren/* model falls back to provider base URL when user has no base_url.""" # Arrange self.mock_user.llm_model = 'wren/default' self.mock_user.llm_base_url = None self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, self.mock_user.llm_model, self.conversation_id ) # Assert — falls back to service-level wren_provider_base_url assert llm.base_url == 'https://provider.example.com' @pytest.mark.asyncio async def test_configure_llm_and_mcp_wren_model_no_base_urls(self): """wren/* model is kept public when no explicit base URL exists.""" # Arrange self.mock_user.llm_model = 'wren/default' self.mock_user.llm_base_url = None self.service.wren_provider_base_url = None self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, self.mock_user.llm_model, self.conversation_id ) # Assert assert llm.model == 'wren/default' assert llm.base_url is None @pytest.mark.asyncio async def test_configure_llm_and_mcp_litellm_proxy_model_keeps_empty_base_url( self, ): """litellm_proxy/* is not treated as an OpenHands provider model.""" self.mock_user.llm_base_url = None self.mock_user_context.get_mcp_api_key.return_value = None llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, 'litellm_proxy/minimax-2.5', self.conversation_id ) assert llm.base_url is None @pytest.mark.asyncio async def test_configure_llm_and_mcp_litellm_proxy_model_preserves_user_base_url( self, ): """litellm_proxy/* model preserves an explicitly configured base_url.""" # Arrange self.mock_user.llm_base_url = 'https://user-llm.example.com' self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, 'litellm_proxy/minimax-2.5', self.conversation_id ) # Assert assert llm.base_url == 'https://user-llm.example.com' @pytest.mark.asyncio async def test_configure_llm_and_mcp_non_wren_model_ignores_provider(self): """Non-wren model ignores provider base URL and uses user base URL.""" # Arrange self.mock_user.llm_model = 'gpt-4' self.mock_user.llm_base_url = 'https://user-llm.example.com' self.service.wren_provider_base_url = 'https://provider.example.com' self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, _ = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) # Assert assert llm.base_url == 'https://user-llm.example.com' @pytest.mark.asyncio async def test_configure_llm_and_mcp_with_user_default_model(self): """Test _configure_llm_and_mcp using user's default model.""" # Arrange self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) # Assert assert llm.model == self.mock_user.llm_model assert 'mcpServers' in mcp_config assert 'default' in mcp_config['mcpServers'] headers = mcp_config['mcpServers']['default']['headers'] assert headers['X-OpenHands-ServerConversation-ID'] == str(self.conversation_id) assert 'X-Session-API-Key' not in headers @pytest.mark.asyncio async def test_configure_llm_and_mcp_without_web_url(self): """Test _configure_llm_and_mcp without web URL (no MCP config).""" # Arrange self.service.web_url = None # Act llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) # Assert assert isinstance(llm, LLM) assert mcp_config == {} def test_compute_plan_path_default_uses_agents_tmp(self): """Test _compute_plan_path returns .agents_tmp/PLAN.md for default/GitHub.""" # Arrange working_dir = '/workspace/project' # Act path_none = self.service._compute_plan_path(working_dir, None) path_github = self.service._compute_plan_path(working_dir, ProviderType.GITHUB) # Assert assert path_none == '/workspace/project/.agents_tmp/PLAN.md' assert path_github == '/workspace/project/.agents_tmp/PLAN.md' def test_compute_plan_path_gitlab_uses_agents_tmp_config(self): """Test _compute_plan_path returns agents-tmp-config/PLAN.md for GitLab.""" # Arrange working_dir = '/workspace/project' # Act path = self.service._compute_plan_path(working_dir, ProviderType.GITLAB) # Assert assert path == '/workspace/project/agents-tmp-config/PLAN.md' def test_compute_plan_path_azure_uses_agents_tmp_config(self): """Test _compute_plan_path returns agents-tmp-config/PLAN.md for Azure.""" # Arrange working_dir = '/workspace/project' # Act path = self.service._compute_plan_path(working_dir, ProviderType.AZURE_DEVOPS) # Assert assert path == '/workspace/project/agents-tmp-config/PLAN.md' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_with_skills(self, _mock_tools): """Skills are loaded when a remote_workspace is provided.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) mock_agent = Mock(spec=Agent) mock_agent.llm = real_llm mock_agent.condenser = None self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) self.service._load_skills_and_update_agent = AsyncMock(return_value=mock_agent) remote_workspace = Mock(spec=AsyncRemoteWorkspace) conversation_id = uuid4() result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=conversation_id, initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=remote_workspace, selected_repository='test_repo', ) assert isinstance(result, StartConversationRequest) assert result.conversation_id == conversation_id self.service._load_skills_and_update_agent.assert_called_once() @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_without_remote_workspace(self, _mock_tools): """Skills loading is skipped when no remote_workspace is provided.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) conversation_id = uuid4() result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=conversation_id, initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=None, ) assert isinstance(result, StartConversationRequest) assert result.conversation_id == conversation_id @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_skills_loading_fails_gracefully(self, _mock_tools): """Conversation still starts when skills loading raises.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) self.service._load_skills_and_update_agent = AsyncMock( side_effect=Exception('Skills loading failed') ) remote_workspace = Mock(spec=AsyncRemoteWorkspace) conversation_id = uuid4() with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service._logger' ) as mock_logger: result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=conversation_id, initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=remote_workspace, selected_repository='test_repo', ) assert isinstance(result, StartConversationRequest) mock_logger.warning.assert_called_once() def test_apply_server_overrides_sets_condenser_usage_id(self): """Condenser LLM must get usage_id='condenser' even when it inherits 'agent'.""" from wren.sdk.context.condenser import LLMSummarizingCondenser llm = LLM(model='wren/gpt-4', api_key='k', usage_id='agent') condenser = LLMSummarizingCondenser(llm=llm) agent = Agent(llm=llm, tools=[], condenser=condenser) updated = self.service._apply_server_agent_overrides( agent, AgentType.DEFAULT, {}, uuid4(), 'user-1' ) assert updated.llm.usage_id == 'agent' assert updated.condenser.llm.usage_id == 'condenser' def test_apply_server_overrides_condenser_non_wren_model(self): """Condenser usage_id is set even for non-wren models (no metadata).""" from wren.sdk.context.condenser import LLMSummarizingCondenser llm = LLM(model='gpt-4', api_key='k', usage_id='agent') condenser = LLMSummarizingCondenser(llm=llm) agent = Agent(llm=llm, tools=[], condenser=condenser) updated = self.service._apply_server_agent_overrides( agent, AgentType.DEFAULT, {}, uuid4(), 'user-1' ) # Non-wren model: main LLM unchanged, but condenser still gets usage_id assert updated.condenser.llm.usage_id == 'condenser' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_start_conversation_request_for_user_integration( self, _mock_tools ): """Test the main _build_start_conversation_request_for_user method integration.""" self.mock_user_context.get_user_info.return_value = self.mock_user mock_secrets = {'GITHUB_TOKEN': StaticSecret(value=SecretStr('tok'))} real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) mock_mcp_config = {'default': {'url': 'test'}} test_conversation_id = uuid4() self.service._setup_secrets_for_git_providers = AsyncMock( return_value=mock_secrets ) self.service._configure_llm_and_mcp = AsyncMock( return_value=(real_llm, mock_mcp_config) ) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=test_conversation_id, initial_message=None, system_message_suffix='Test suffix', git_provider=ProviderType.GITHUB, working_dir='/test/dir', agent_type=AgentType.DEFAULT, llm_model='gpt-4', remote_workspace=None, selected_repository='test/repo', ) assert isinstance(result, StartConversationRequest) assert result.conversation_id == test_conversation_id assert result.agent.llm.model == 'gpt-4' # Secrets are injected via agent_context assert result.agent.agent_context.secrets == mock_secrets # System message suffix includes the original suffix and web host context assert 'Test suffix' in result.agent.agent_context.system_message_suffix assert '' in result.agent.agent_context.system_message_suffix assert ( 'https://test.example.com' in result.agent.agent_context.system_message_suffix ) # Workspace points to the repo subdirectory assert result.workspace.working_dir == '/test/dir/repo' self.service._setup_secrets_for_git_providers.assert_called_once_with( self.mock_user ) self.service._configure_llm_and_mcp.assert_called_once_with( self.mock_user, 'gpt-4', test_conversation_id ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_populates_observability_metadata(self, _mock_tools): """Repo / branch / provider land on the request's observability_metadata so the agent-server attaches them to the Laminar trace. With no remote_workspace the commit can't be resolved, so it is omitted.""" self.mock_user_context.get_user_info.return_value = self.mock_user self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock( return_value=(LLM(model='gpt-4', api_key=SecretStr('k')), {}) ) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=ProviderType.GITHUB, working_dir='/test/dir', remote_workspace=None, selected_repository='test/repo', selected_branch='feature-x', ) assert result.observability_metadata == { 'repo': 'test/repo', 'branch': 'feature-x', 'git_provider': 'github', } @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_observability_metadata_includes_commit( self, _mock_tools ): """The post-clone HEAD sha is resolved from the workspace and added.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('k')) mock_agent = Mock(spec=Agent) mock_agent.llm = real_llm mock_agent.condenser = None self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) self.service._load_skills_and_update_agent = AsyncMock(return_value=mock_agent) remote_workspace = Mock(spec=AsyncRemoteWorkspace) remote_workspace.execute_command = AsyncMock( return_value=SimpleNamespace(exit_code=0, stdout='abc123sha\n') ) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=ProviderType.GITHUB, working_dir='/test/dir', remote_workspace=remote_workspace, selected_repository='test/repo', selected_branch='feature-x', ) assert result.observability_metadata == { 'repo': 'test/repo', 'branch': 'feature-x', 'git_provider': 'github', 'commit': 'abc123sha', } remote_workspace.execute_command.assert_awaited_once_with( 'git rev-parse --verify --quiet HEAD', '/test/dir/repo', timeout=10.0 ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_commit_resolution_is_best_effort(self, _mock_tools): """A failed HEAD lookup leaves commit out without breaking the build.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('k')) mock_agent = Mock(spec=Agent) mock_agent.llm = real_llm mock_agent.condenser = None self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) self.service._load_skills_and_update_agent = AsyncMock(return_value=mock_agent) remote_workspace = Mock(spec=AsyncRemoteWorkspace) remote_workspace.execute_command = AsyncMock(side_effect=RuntimeError('boom')) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=remote_workspace, selected_repository='test/repo', ) assert result.observability_metadata == {'repo': 'test/repo'} @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_no_repo_omits_observability_metadata( self, _mock_tools ): """A repo-less conversation adds no repo metadata to the trace.""" self.mock_user_context.get_user_info.return_value = self.mock_user self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock( return_value=(LLM(model='gpt-4', api_key=SecretStr('k')), {}) ) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=None, selected_repository=None, ) assert result.observability_metadata == {} @pytest.mark.asyncio async def test_build_request_routes_acp_user_to_observability_metadata(self): """End-to-end through the *routing* entrypoint (not the ACP builder directly): an ACP user hits the ``isinstance(..., ACPAgentSettings)`` branch in ``_build_start_conversation_request_for_user``, which must forward git_provider/selected_branch/remote_workspace into ``_build_acp_start_conversation_request`` for this to populate. """ from wren.sdk.settings import ACPAgentSettings self.mock_user.agent_settings = ACPAgentSettings( acp_server='claude-code', llm=LLM(model='claude-sonnet-4-5', api_key=SecretStr('k')), ) self.mock_user_context.get_user_info.return_value = self.mock_user self.mock_user_context.get_secrets = AsyncMock(return_value={}) self.mock_user_context.get_provider_tokens = AsyncMock(return_value=None) remote_workspace = Mock(spec=AsyncRemoteWorkspace) remote_workspace.execute_command = AsyncMock( return_value=SimpleNamespace(exit_code=0, stdout='def456sha\n') ) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=ProviderType.GITHUB, working_dir='/test/dir', remote_workspace=remote_workspace, selected_repository='test/repo', selected_branch='feature-x', ) assert result.agent.agent_kind == 'acp' assert result.observability_metadata == { 'repo': 'test/repo', 'branch': 'feature-x', 'git_provider': 'github', 'commit': 'def456sha', } @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_appends_shallow_clone_context(self, _mock_tools): self.mock_user.git_full_clone = False self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix='Existing integration instructions.', git_provider=ProviderType.GITHUB, working_dir='/test/dir', selected_repository='test/repo', ) suffix = result.agent.agent_context.system_message_suffix assert 'Existing integration instructions.' in suffix assert '' in suffix assert 'git fetch --unshallow' in suffix @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_omits_shallow_clone_context_for_full_clone( self, _mock_tools ): self.mock_user.git_full_clone = True self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix='Existing integration instructions.', git_provider=ProviderType.GITHUB, working_dir='/test/dir', selected_repository='test/repo', ) suffix = result.agent.agent_context.system_message_suffix assert 'Existing integration instructions.' in suffix assert '' not in suffix @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_registered_agent_definitions' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.register_builtins_agents' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_passes_enable_sub_agents_true( self, mock_tools, mock_register_builtins, mock_get_agent_definitions ): """Built-in sub-agents are registered when the user setting is on.""" from wren.sdk.settings import OpenHandsAgentSettings from wren.sdk.subagent.schema import AgentDefinition agent_definition = AgentDefinition( name='general-purpose', description='General-purpose subagent', tools=['terminal'], ) mock_get_agent_definitions.return_value = [agent_definition] agent_settings = OpenHandsAgentSettings( llm={'model': 'gpt-4', 'api_key': 'test-key'}, enable_sub_agents=True, ) self.mock_user.agent_settings = agent_settings self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=None, ) mock_register_builtins.assert_called_once_with(enable_browser=True) mock_get_agent_definitions.assert_called_once_with() mock_tools.assert_called_once_with(enable_browser=True, enable_sub_agents=True) assert result.agent_definitions == [agent_definition] @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_registered_agent_definitions' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.register_builtins_agents' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_passes_enable_sub_agents_false( self, mock_tools, mock_register_builtins, mock_get_agent_definitions ): """Built-in sub-agents are registered but not forwarded when disabled.""" from wren.sdk.settings import OpenHandsAgentSettings agent_settings = OpenHandsAgentSettings( llm={'model': 'gpt-4', 'api_key': 'test-key'}, enable_sub_agents=False, ) self.mock_user.agent_settings = agent_settings self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', remote_workspace=None, ) mock_register_builtins.assert_called_once_with(enable_browser=True) mock_get_agent_definitions.assert_not_called() mock_tools.assert_called_once_with(enable_browser=True, enable_sub_agents=False) assert result.agent_definitions == [] @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_start_conversation_request_with_api_secrets(self, _mock_tools): """Test _build_start_conversation_request_for_user with API-provided secrets.""" self.mock_user_context.get_user_info.return_value = self.mock_user # Existing secrets from git providers existing_secrets = { 'GITHUB_TOKEN': StaticSecret(value=SecretStr('github_tok')), 'EXISTING_SECRET': StaticSecret(value=SecretStr('existing_value')), } real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) mock_mcp_config = {'default': {'url': 'test'}} test_conversation_id = uuid4() self.service._setup_secrets_for_git_providers = AsyncMock( return_value=existing_secrets ) self.service._configure_llm_and_mcp = AsyncMock( return_value=(real_llm, mock_mcp_config) ) # API-provided secrets - should be merged with existing secrets api_secrets = { 'MY_API_KEY': SecretStr('my_api_key_value'), 'ANOTHER_SECRET': SecretStr('another_value'), } result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=test_conversation_id, initial_message=None, system_message_suffix=None, git_provider=ProviderType.GITHUB, working_dir='/test/dir', agent_type=AgentType.DEFAULT, llm_model='gpt-4', remote_workspace=None, selected_repository='test/repo', api_secrets=api_secrets, ) assert isinstance(result, StartConversationRequest) # All secrets should be present (existing + API-provided) secrets = result.agent.agent_context.secrets assert 'GITHUB_TOKEN' in secrets assert 'EXISTING_SECRET' in secrets assert 'MY_API_KEY' in secrets assert 'ANOTHER_SECRET' in secrets # API-provided secrets should be StaticSecret instances assert isinstance(secrets['MY_API_KEY'], StaticSecret) assert secrets['MY_API_KEY'].value.get_secret_value() == 'my_api_key_value' assert secrets['ANOTHER_SECRET'].value.get_secret_value() == 'another_value' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_start_conversation_request_api_secrets_override_existing( self, _mock_tools ): """Test that API-provided secrets override existing secrets with the same name.""" self.mock_user_context.get_user_info.return_value = self.mock_user # Existing secrets existing_secrets = { 'SHARED_SECRET': StaticSecret(value=SecretStr('original_value')), } real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) mock_mcp_config = None test_conversation_id = uuid4() self.service._setup_secrets_for_git_providers = AsyncMock( return_value=existing_secrets ) self.service._configure_llm_and_mcp = AsyncMock( return_value=(real_llm, mock_mcp_config) ) # API-provided secret with same name should override api_secrets = { 'SHARED_SECRET': SecretStr('overridden_value'), } result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=test_conversation_id, initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', agent_type=AgentType.DEFAULT, llm_model='gpt-4', remote_workspace=None, selected_repository=None, api_secrets=api_secrets, ) # API-provided secret should override the existing one secrets = result.agent.agent_context.secrets assert secrets['SHARED_SECRET'].value.get_secret_value() == 'overridden_value' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_start_conversation_request_no_api_secrets(self, _mock_tools): """Test _build_start_conversation_request_for_user without API-provided secrets.""" self.mock_user_context.get_user_info.return_value = self.mock_user existing_secrets = { 'GITHUB_TOKEN': StaticSecret(value=SecretStr('tok')), } real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) mock_mcp_config = None test_conversation_id = uuid4() self.service._setup_secrets_for_git_providers = AsyncMock( return_value=existing_secrets ) self.service._configure_llm_and_mcp = AsyncMock( return_value=(real_llm, mock_mcp_config) ) # No API secrets provided (None) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=test_conversation_id, initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/test/dir', agent_type=AgentType.DEFAULT, llm_model='gpt-4', remote_workspace=None, selected_repository=None, api_secrets=None, ) # Only existing secrets should be present secrets = result.agent.agent_context.secrets assert secrets == existing_secrets @pytest.mark.asyncio async def test_find_running_sandbox_for_user_found(self): """Test _find_running_sandbox_for_user when a running sandbox is found.""" # Arrange user_id = 'test_user_123' self.mock_user_context.get_user_id.return_value = user_id # Create mock sandboxes running_sandbox = Mock(spec=SandboxInfo) running_sandbox.id = 'sandbox_1' running_sandbox.status = SandboxStatus.RUNNING running_sandbox.created_by_user_id = user_id other_user_sandbox = Mock(spec=SandboxInfo) other_user_sandbox.id = 'sandbox_2' other_user_sandbox.status = SandboxStatus.RUNNING other_user_sandbox.created_by_user_id = 'other_user' paused_sandbox = Mock(spec=SandboxInfo) paused_sandbox.id = 'sandbox_3' paused_sandbox.status = SandboxStatus.PAUSED paused_sandbox.created_by_user_id = user_id # Mock sandbox service search mock_page = Mock(spec=SandboxPage) mock_page.items = [other_user_sandbox, running_sandbox, paused_sandbox] mock_page.next_page_id = None self.mock_sandbox_service.search_sandboxes = AsyncMock(return_value=mock_page) # Act result = await self.service._find_running_sandbox_for_user() # Assert assert result == running_sandbox self.mock_user_context.get_user_id.assert_called_once() self.mock_sandbox_service.search_sandboxes.assert_called_once_with( page_id=None, limit=100 ) @pytest.mark.asyncio async def test_find_running_sandbox_for_user_not_found(self): """Test _find_running_sandbox_for_user when no running sandbox is found.""" # Arrange user_id = 'test_user_123' self.mock_user_context.get_user_id.return_value = user_id # Create mock sandboxes (none running for this user) other_user_sandbox = Mock(spec=SandboxInfo) other_user_sandbox.id = 'sandbox_1' other_user_sandbox.status = SandboxStatus.RUNNING other_user_sandbox.created_by_user_id = 'other_user' paused_sandbox = Mock(spec=SandboxInfo) paused_sandbox.id = 'sandbox_2' paused_sandbox.status = SandboxStatus.PAUSED paused_sandbox.created_by_user_id = user_id # Mock sandbox service search mock_page = Mock(spec=SandboxPage) mock_page.items = [other_user_sandbox, paused_sandbox] mock_page.next_page_id = None self.mock_sandbox_service.search_sandboxes = AsyncMock(return_value=mock_page) # Act result = await self.service._find_running_sandbox_for_user() # Assert assert result is None self.mock_user_context.get_user_id.assert_called_once() self.mock_sandbox_service.search_sandboxes.assert_called_once_with( page_id=None, limit=100 ) @pytest.mark.asyncio async def test_find_running_sandbox_for_user_exception_handling(self): """Test _find_running_sandbox_for_user handles exceptions gracefully.""" # Arrange self.mock_user_context.get_user_id.side_effect = Exception('User context error') # Act with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service._logger' ) as mock_logger: result = await self.service._find_running_sandbox_for_user() # Assert assert result is None mock_logger.warning.assert_called_once() assert ( 'Error finding running sandbox for user' in mock_logger.warning.call_args[0][0] ) async def test_export_conversation_success(self): """Test successful download of conversation trajectory.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.title = 'Test Conversation' mock_conversation_info.created_at = datetime(2024, 1, 1, 12, 0, 0) mock_conversation_info.updated_at = datetime(2024, 1, 1, 13, 0, 0) mock_conversation_info.selected_repository = 'test/repo' mock_conversation_info.git_provider = 'github' mock_conversation_info.selected_branch = 'main' mock_conversation_info.model_dump_json = Mock( return_value='{"id": "test", "title": "Test Conversation"}' ) self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) mock_event1 = Mock(spec=Event) mock_event1.id = uuid4() mock_event1.model_dump = Mock( return_value={'id': str(mock_event1.id), 'type': 'action'} ) mock_event2 = Mock(spec=Event) mock_event2.id = uuid4() mock_event2.model_dump = Mock( return_value={'id': str(mock_event2.id), 'type': 'observation'} ) self.mock_event_service.count_events = AsyncMock(return_value=2) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([mock_event1, mock_event2]) ) self.mock_event_service.search_events = AsyncMock() result = await self.service.export_conversation(conversation_id) assert result is not None assert isinstance(result, bytes) with zipfile.ZipFile(io.BytesIO(result), 'r') as zipf: file_list = zipf.namelist() assert 'meta.json' in file_list assert any( f.startswith('event_') and f.endswith('.json') for f in file_list ) with zipf.open('meta.json') as meta_file: meta_content = meta_file.read().decode('utf-8') assert '"id": "test"' in meta_content assert '"title": "Test Conversation"' in meta_content event_files = [f for f in file_list if f.startswith('event_')] assert len(event_files) == 2 with zipf.open(event_files[0]) as event_file: event_content = json.loads(event_file.read().decode('utf-8')) assert 'id' in event_content assert 'type' in event_content self.mock_app_conversation_info_service.get_app_conversation_info.assert_called_once_with( conversation_id ) self.mock_event_service.count_events.assert_awaited_once_with(conversation_id) self.mock_event_service.iter_events_for_export.assert_called_once_with( conversation_id ) self.mock_event_service.search_events.assert_not_called() mock_conversation_info.model_dump_json.assert_called_once_with(indent=2) async def test_export_conversation_writes_utf8_when_platform_default_cannot_encode( self, ): """Test trajectory export writes JSON as UTF-8 regardless of platform default.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.title = 'Café 🔐' mock_conversation_info.model_dump_json = Mock( return_value='{"id": "test", "title": "Café 🔐"}' ) self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) mock_event = Mock(spec=Event) mock_event.id = uuid4() mock_event.model_dump = Mock( return_value={ 'id': str(mock_event.id), 'type': 'observation', 'message': 'résumé 🔐', } ) self.mock_event_service.count_events = AsyncMock(return_value=1) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([mock_event]) ) result = await self.service.export_conversation(conversation_id) with zipfile.ZipFile(io.BytesIO(result), 'r') as zipf: with zipf.open('meta.json') as meta_file: meta_content = meta_file.read().decode('utf-8') assert '"title": "Café 🔐"' in meta_content event_files = [ name for name in zipf.namelist() if name.startswith('event_') ] assert len(event_files) == 1 with zipf.open(event_files[0]) as event_file: event_content = json.loads(event_file.read().decode('utf-8')) assert event_content['message'] == 'résumé 🔐' @pytest.mark.asyncio async def test_export_conversation_conversation_not_found(self): """Test download when conversation is not found.""" conversation_id = uuid4() self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=None ) self.mock_event_service.count_events = AsyncMock() self.mock_event_service.iter_events_for_export = Mock() with pytest.raises( ValueError, match=f'Conversation not found: {conversation_id}' ): await self.service.export_conversation(conversation_id) self.mock_app_conversation_info_service.get_app_conversation_info.assert_called_once_with( conversation_id ) self.mock_event_service.count_events.assert_not_called() self.mock_event_service.iter_events_for_export.assert_not_called() @pytest.mark.asyncio async def test_export_conversation_empty_events(self): """Test download with conversation that has no events.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.title = 'Empty Conversation' mock_conversation_info.model_dump_json = Mock( return_value='{"id": "test", "title": "Empty Conversation"}' ) self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) self.mock_event_service.count_events = AsyncMock(return_value=0) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([]) ) result = await self.service.export_conversation(conversation_id) assert result is not None assert isinstance(result, bytes) with zipfile.ZipFile(io.BytesIO(result), 'r') as zipf: file_list = zipf.namelist() assert 'meta.json' in file_list assert len([f for f in file_list if f.startswith('event_')]) == 0 self.mock_app_conversation_info_service.get_app_conversation_info.assert_called_once_with( conversation_id ) self.mock_event_service.count_events.assert_awaited_once_with(conversation_id) self.mock_event_service.iter_events_for_export.assert_called_once_with( conversation_id ) @pytest.mark.asyncio async def test_export_conversation_calls_iterator_with_conversation_id( self, ): """Test that export_conversation calls the export iterator with conversation_id.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) self.mock_event_service.count_events = AsyncMock(return_value=0) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([]) ) self.mock_event_service.search_events = AsyncMock() await self.service.export_conversation(conversation_id) self.mock_event_service.iter_events_for_export.assert_called_once_with( conversation_id ) self.mock_event_service.search_events.assert_not_called() @pytest.mark.asyncio async def test_export_conversation_large_iterator(self): """Test download with many events from one export iterator pass.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.title = 'Large Conversation' mock_conversation_info.model_dump_json = Mock( return_value='{"id": "test", "title": "Large Conversation"}' ) self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) total_events = 12 all_events = [] for i in range(total_events): mock_event = Mock(spec=Event) mock_event.id = uuid4() mock_event.model_dump = Mock( return_value={ 'id': str(mock_event.id), 'type': f'event_{i}', } ) all_events.append(mock_event) self.mock_event_service.count_events = AsyncMock(return_value=total_events) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter(all_events) ) self.mock_event_service.search_events = AsyncMock() result = await self.service.export_conversation(conversation_id) assert result is not None assert isinstance(result, bytes) with zipfile.ZipFile(io.BytesIO(result), 'r') as zipf: file_list = zipf.namelist() assert 'meta.json' in file_list event_files = [f for f in file_list if f.startswith('event_')] assert len(event_files) == total_events self.mock_event_service.iter_events_for_export.assert_called_once_with( conversation_id ) self.mock_event_service.search_events.assert_not_called() @pytest.mark.asyncio async def test_open_conversation_export_streams_with_lock(self): """Test streaming export acquires and releases the conversation export lock.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') mock_event = Mock(spec=Event) mock_event.id = uuid4() mock_event.model_dump = Mock(return_value={'id': str(mock_event.id)}) self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) self.mock_event_service.count_events = AsyncMock(return_value=1) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([mock_event]) ) fake_lock = _FakeExportLock() with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.try_acquire_redis_lock', new=AsyncMock(return_value=fake_lock), ) as mock_acquire_lock: stream = await self.service.open_conversation_export(conversation_id) result = b''.join([chunk async for chunk in stream]) assert result assert fake_lock.released is True mock_acquire_lock.assert_awaited_once_with( f'app_conversation_export:{conversation_id.hex}', self.service.export_lock_ttl_seconds, ) @pytest.mark.asyncio async def test_open_conversation_export_rejects_duplicate_export(self): """Test streaming export rejects duplicate export attempts.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.try_acquire_redis_lock', new=AsyncMock(return_value=None), ): with pytest.raises(ConversationExportAlreadyRunning): await self.service.open_conversation_export(conversation_id) self.mock_event_service.count_events.assert_not_called() @pytest.mark.asyncio async def test_open_conversation_export_fails_closed_when_lock_unavailable(self): """Test streaming export fails closed when Redis lock cannot be checked.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') self.service.export_lock_required = True self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.try_acquire_redis_lock', new=AsyncMock(side_effect=RedisLockUnavailable()), ): with pytest.raises(ConversationExportLockUnavailable): await self.service.open_conversation_export(conversation_id) self.mock_event_service.count_events.assert_not_called() @pytest.mark.asyncio async def test_open_conversation_export_proceeds_when_lock_not_required(self): """Test non-SaaS export proceeds without Redis when locking is best effort.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') mock_event = Mock(spec=Event) mock_event.id = uuid4() mock_event.model_dump = Mock(return_value={'id': str(mock_event.id)}) self.service.export_lock_required = False self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) self.mock_event_service.count_events = AsyncMock(return_value=1) self.mock_event_service.iter_events_for_export = Mock( return_value=_async_iter([mock_event]) ) with patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.try_acquire_redis_lock', new=AsyncMock(side_effect=RedisLockUnavailable()), ): stream = await self.service.open_conversation_export(conversation_id) result = b''.join([chunk async for chunk in stream]) assert result self.mock_event_service.count_events.assert_awaited_once_with(conversation_id) @pytest.mark.asyncio async def test_export_conversation_rejects_too_many_events(self): """Test export rejects conversations over the configured event limit.""" conversation_id = uuid4() mock_conversation_info = Mock(spec=AppConversationInfo) mock_conversation_info.id = conversation_id mock_conversation_info.model_dump_json = Mock(return_value='{}') self.service.export_max_events = 1 self.mock_app_conversation_info_service.get_app_conversation_info = AsyncMock( return_value=mock_conversation_info ) self.mock_event_service.count_events = AsyncMock(return_value=2) self.mock_event_service.iter_events_for_export = Mock() with pytest.raises(ConversationExportTooLarge): await self.service.export_conversation(conversation_id) self.mock_event_service.iter_events_for_export.assert_not_called() @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.AsyncRemoteWorkspace' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.ConversationInfo' ) async def test_start_app_conversation_default_title_uses_first_five_characters( self, mock_conversation_info_class, mock_remote_workspace_class ): """Test that v1 conversations use first 5 characters of conversation ID for default title.""" # Arrange conversation_id = uuid4() conversation_id_hex = conversation_id.hex expected_title = f'Conversation {conversation_id_hex[:5]}' # Mock user context self.mock_user_context.get_user_id = AsyncMock(return_value='test_user_123') self.mock_user_context.get_user_info = AsyncMock(return_value=self.mock_user) # Mock sandbox and sandbox spec mock_sandbox_spec = Mock(spec=SandboxSpecInfo) mock_sandbox_spec.working_dir = '/test/workspace' self.mock_sandbox.sandbox_spec_id = str(uuid4()) self.mock_sandbox.id = str(uuid4()) # Ensure sandbox.id is a string self.mock_sandbox.session_api_key = 'test_session_key' exposed_url = ExposedUrl( name=AGENT_SERVER, url='http://agent-server:8000', port=60000 ) self.mock_sandbox.exposed_urls = [exposed_url] self.mock_sandbox_service.get_sandbox = AsyncMock( return_value=self.mock_sandbox ) self.mock_sandbox_spec_service.get_sandbox_spec = AsyncMock( return_value=mock_sandbox_spec ) # Mock remote workspace mock_remote_workspace = Mock() mock_remote_workspace_class.return_value = mock_remote_workspace # Mock the wait for sandbox and setup scripts async def mock_wait_for_sandbox(task): task.sandbox_id = self.mock_sandbox.id yield task async def mock_run_setup_scripts( task, sandbox, workspace, agent_server_url, conversation_id ): yield task self.service._wait_for_sandbox_start = mock_wait_for_sandbox self.service.run_setup_scripts = mock_run_setup_scripts # Mock build start conversation request mock_agent = Mock(spec=Agent) mock_agent.llm = Mock(spec=LLM) mock_agent.llm.model = 'gpt-4' mock_start_request = Mock(spec=StartConversationRequest) mock_start_request.agent = mock_agent mock_start_request.model_dump.return_value = {'test': 'data'} self.service._build_start_conversation_request_for_user = AsyncMock( return_value=mock_start_request ) # Mock ConversationInfo returned from agent server mock_conversation_info = Mock() mock_conversation_info.id = conversation_id mock_conversation_info_class.model_validate.return_value = ( mock_conversation_info ) # Mock HTTP response from agent server mock_response = Mock() mock_response.json.return_value = {'id': str(conversation_id)} mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Mock event callback service self.mock_event_callback_service.save_event_callback = AsyncMock() # Create request request = AppConversationStartRequest() # Act async for task in self.service._start_app_conversation(request): # Consume all tasks to reach the point where title is set pass # Assert # Verify that save_app_conversation_info was called with the correct title format self.mock_app_conversation_info_service.save_app_conversation_info.assert_called_once() call_args = ( self.mock_app_conversation_info_service.save_app_conversation_info.call_args ) saved_info = call_args[0][0] # First positional argument assert saved_info.title == expected_title, ( f'Expected title to be "{expected_title}" (first 5 chars), ' f'but got "{saved_info.title}"' ) assert saved_info.id == conversation_id @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.AsyncRemoteWorkspace' ) @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.ConversationInfo' ) @pytest.mark.asyncio async def test_start_app_conversation_stores_acp_model_as_llm_model( self, mock_conversation_info_class, mock_remote_workspace_class ): """ACP conversations persist acp_model as llm_model in AppConversationInfo.""" from wren.sdk.settings import ACPAgentSettings conversation_id = uuid4() self.mock_user_context.get_user_id = AsyncMock(return_value='test_user_123') acp_user = _TestUserInfo( id='test_user_123', llm_model='gpt-4', sandbox_grouping_strategy=SandboxGroupingStrategy.NO_GROUPING, ) acp_user.agent_settings = ACPAgentSettings( acp_server='codex', acp_model='gpt-5.5/high' ) self.mock_user_context.get_user_info = AsyncMock(return_value=acp_user) mock_sandbox_spec = Mock(spec=SandboxSpecInfo) mock_sandbox_spec.working_dir = '/test/workspace' self.mock_sandbox.sandbox_spec_id = str(uuid4()) self.mock_sandbox.id = str(uuid4()) self.mock_sandbox.session_api_key = 'test_session_key' exposed_url = ExposedUrl( name=AGENT_SERVER, url='http://agent-server:8000', port=60000 ) self.mock_sandbox.exposed_urls = [exposed_url] self.mock_sandbox_service.get_sandbox = AsyncMock( return_value=self.mock_sandbox ) self.mock_sandbox_spec_service.get_sandbox_spec = AsyncMock( return_value=mock_sandbox_spec ) mock_remote_workspace_class.return_value = Mock() async def mock_wait_for_sandbox(task): task.sandbox_id = self.mock_sandbox.id yield task async def mock_run_setup_scripts( task, sandbox, workspace, agent_server_url, conversation_id ): yield task self.service._wait_for_sandbox_start = mock_wait_for_sandbox self.service.run_setup_scripts = mock_run_setup_scripts self.service._seed_sandbox_profiles = AsyncMock() # ACP start request: agent_kind='acp', acp_model='gpt-5.5/high' mock_acp_agent = Mock() mock_acp_agent.agent_kind = 'acp' mock_acp_agent.acp_model = 'gpt-5.5/high' mock_start_request = Mock(spec=StartConversationRequest) mock_start_request.agent = mock_acp_agent mock_start_request.model_dump.return_value = {} self.service._build_start_conversation_request_for_user = AsyncMock( return_value=mock_start_request ) mock_conversation_info = Mock() mock_conversation_info.id = conversation_id mock_conversation_info_class.model_validate.return_value = ( mock_conversation_info ) mock_response = Mock() mock_response.json.return_value = {'id': str(conversation_id)} mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) self.mock_event_callback_service.save_event_callback = AsyncMock() self.mock_app_conversation_info_service.save_app_conversation_info = AsyncMock() async for _ in self.service._start_app_conversation( AppConversationStartRequest() ): pass self.mock_app_conversation_info_service.save_app_conversation_info.assert_called_once() saved_info = self.mock_app_conversation_info_service.save_app_conversation_info.call_args[ 0 ][0] assert saved_info.llm_model == 'gpt-5.5/high' assert saved_info.agent_kind == 'acp' assert saved_info.tags.get('acpserver') == 'codex' @pytest.mark.asyncio async def test_configure_llm_and_mcp_with_custom_remote_servers(self): """Test _configure_llm_and_mcp merges custom remote servers.""" from fastmcp.mcp_config import MCPConfig, RemoteMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'linear': RemoteMCPServer( url='https://linear.app/sse', transport='sse', auth='linear_key' ), 'notion': RemoteMCPServer( url='https://notion.com/sse', transport='sse' ), } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) assert isinstance(llm, LLM) assert 'mcpServers' in mcp_config mcp_servers = mcp_config['mcpServers'] assert 'default' in mcp_servers assert 'linear' in mcp_servers assert 'notion' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_with_custom_http_servers(self): """Test _configure_llm_and_mcp merges custom HTTP servers with timeout.""" from fastmcp.mcp_config import MCPConfig, RemoteMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'custom-http': RemoteMCPServer( url='https://example.com/mcp', transport='http', auth='test_key', timeout=120, ) } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) assert isinstance(llm, LLM) mcp_servers = mcp_config['mcpServers'] assert 'custom-http' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_with_custom_stdio_servers(self): """Test _configure_llm_and_mcp merges custom STDIO servers with explicit names.""" from fastmcp.mcp_config import MCPConfig, StdioMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'my-custom-server': StdioMCPServer( command='npx', args=['-y', 'my-package'], env={'API_KEY': 'secret'}, ) } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) assert isinstance(llm, LLM) mcp_servers = mcp_config['mcpServers'] assert 'my-custom-server' in mcp_servers server_config = mcp_servers['my-custom-server'] assert server_config['command'] == 'npx' assert server_config['args'] == ['-y', 'my-package'] assert server_config['env'] == {'API_KEY': 'secret'} @pytest.mark.asyncio async def test_configure_llm_and_mcp_merges_system_and_custom_servers(self): """Test _configure_llm_and_mcp merges both system and custom MCP servers.""" from fastmcp.mcp_config import ( MCPConfig, RemoteMCPServer, StdioMCPServer, ) self.mock_user.mcp_config = MCPConfig( mcpServers={ 'custom-sse': RemoteMCPServer( url='https://custom.com/sse', transport='sse' ), 'custom-stdio': StdioMCPServer(command='node', args=['app.js']), } ) self.mock_user_context.get_mcp_api_key.return_value = 'mcp_api_key' llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] # System provides default MCP server (Tavily is proxied through it if configured) assert 'default' in mcp_servers # Custom servers are merged assert 'custom-sse' in mcp_servers assert 'custom-stdio' in mcp_servers assert len(mcp_servers) == 3 @pytest.mark.asyncio async def test_configure_llm_and_mcp_custom_config_error_handling(self): """Test _configure_llm_and_mcp handles invalid custom MCP config gracefully.""" # Arrange invalid_mcp_config = Mock() invalid_mcp_config.model_dump.return_value = 'not-a-dict' self.mock_user._agent_settings_override = SimpleNamespace( mcp_config=invalid_mcp_config ) self.service._configure_llm = Mock( return_value=LLM.model_validate({'model': 'gpt-4', 'usage_id': 'agent'}) ) self.mock_user_context.get_mcp_api_key.return_value = None # Act llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) # Assert - should still return valid config with system servers only assert isinstance(llm, LLM) mcp_servers = mcp_config['mcpServers'] assert 'default' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_sdk_format_with_mcpservers_wrapper(self): """Test _configure_llm_and_mcp returns SDK-required format with mcpServers key.""" # Arrange self.mock_user_context.get_mcp_api_key.return_value = 'mcp_key' # Act llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) # Assert - SDK expects {'mcpServers': {...}} format assert 'mcpServers' in mcp_config assert isinstance(mcp_config['mcpServers'], dict) # Verify structure matches SDK expectations for server_name, server_config in mcp_config['mcpServers'].items(): assert isinstance(server_name, str) assert isinstance(server_config, dict) @pytest.mark.asyncio async def test_configure_llm_and_mcp_empty_custom_config(self): """Test _configure_llm_and_mcp handles empty custom MCP config.""" from fastmcp.mcp_config import MCPConfig self.mock_user.mcp_config = MCPConfig(mcpServers={}) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'default' in mcp_servers assert len(mcp_servers) == 1 @pytest.mark.asyncio async def test_configure_llm_and_mcp_remote_server_without_auth(self): """Test _configure_llm_and_mcp handles remote servers without auth.""" from fastmcp.mcp_config import MCPConfig, RemoteMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'public': RemoteMCPServer(url='https://public.com/sse', transport='sse') } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'public' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_http_server_default_timeout(self): """Test _configure_llm_and_mcp handles HTTP servers with default timeout.""" from fastmcp.mcp_config import MCPConfig, RemoteMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'http-server': RemoteMCPServer( url='https://example.com/mcp', transport='http' ) } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'http-server' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_stdio_server_without_env(self): """Test _configure_llm_and_mcp handles STDIO servers without environment variables.""" from fastmcp.mcp_config import MCPConfig, StdioMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'simple-server': StdioMCPServer(command='node', args=['app.js']) } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'simple-server' in mcp_servers server_config = mcp_servers['simple-server'] assert server_config['command'] == 'node' assert server_config['args'] == ['app.js'] @pytest.mark.asyncio async def test_configure_llm_and_mcp_multiple_servers_same_type(self): """Test _configure_llm_and_mcp handles multiple custom servers of the same type.""" from fastmcp.mcp_config import MCPConfig, RemoteMCPServer self.mock_user.mcp_config = MCPConfig( mcpServers={ 'server1': RemoteMCPServer( url='https://server1.com/sse', transport='sse' ), 'server2': RemoteMCPServer( url='https://server2.com/sse', transport='sse' ), 'server3': RemoteMCPServer( url='https://server3.com/sse', transport='sse' ), } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'server1' in mcp_servers assert 'server2' in mcp_servers assert 'server3' in mcp_servers @pytest.mark.asyncio async def test_configure_llm_and_mcp_mixed_server_types(self): """Test _configure_llm_and_mcp handles all server types together.""" from fastmcp.mcp_config import ( MCPConfig, RemoteMCPServer, StdioMCPServer, ) self.mock_user.mcp_config = MCPConfig( mcpServers={ 'sse-server': RemoteMCPServer( url='https://sse.example.com/sse', transport='sse', auth='sse_key', ), 'http-server': RemoteMCPServer( url='https://shttp.example.com/mcp', transport='http', timeout=90, ), 'stdio-server': StdioMCPServer( command='npx', args=['mcp-server'], env={'TOKEN': 'value'}, ), } ) self.mock_user_context.get_mcp_api_key.return_value = None llm, mcp_config = await self.service._configure_llm_and_mcp( self.mock_user, None, self.conversation_id ) mcp_servers = mcp_config['mcpServers'] assert 'sse-server' in mcp_servers assert 'http-server' in mcp_servers assert 'stdio-server' in mcp_servers stdio_server = mcp_servers['stdio-server'] assert stdio_server['command'] == 'npx' assert stdio_server['env'] == {'TOKEN': 'value'} # ------------------------------------------------------------------ # # Regression tests: workspace.working_dir == project_dir # # ------------------------------------------------------------------ # def test_get_project_dir_with_repo(self): """get_project_dir appends repo name to working_dir.""" from wren.app_server.app_conversation.app_conversation_service_base import ( get_project_dir, ) assert ( get_project_dir('/workspace/project', 'OpenHands/software-agent-sdk') == '/workspace/project/software-agent-sdk' ) assert get_project_dir('/w', 'org/repo-name') == '/w/repo-name' def test_get_project_dir_without_repo(self): """get_project_dir returns working_dir unchanged when no repo selected.""" from wren.app_server.app_conversation.app_conversation_service_base import ( get_project_dir, ) assert get_project_dir('/workspace/project', None) == '/workspace/project' assert get_project_dir('/workspace/project', '') == '/workspace/project' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_workspace_uses_project_dir(self, _mock_tools): """workspace.working_dir in StartConversationRequest must equal project_dir. This is the root cause of the V1 hook-stop bug: if workspace.working_dir points to the sandbox mount root (/workspace/project) instead of the cloned repo (/workspace/project/), the agent's CWD is wrong and .wren/hooks/on_stop.sh is not found. """ self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace/project', selected_repository='OpenHands/software-agent-sdk', ) assert ( result.workspace.working_dir == '/workspace/project/software-agent-sdk' ), 'workspace.working_dir must point to the repo root, not the sandbox mount' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_no_repo_workspace_unchanged(self, _mock_tools): """Without selected_repository, workspace.working_dir == sandbox working_dir.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace/project', selected_repository=None, ) assert result.workspace.working_dir == '/workspace/project' @pytest.mark.asyncio async def test_search_app_conversations_with_sandbox_id_filter(self): """Test that search_app_conversations passes sandbox_id__eq to the info service. This verifies that the sandbox_id filter is correctly propagated through the service layer to the underlying info service. """ from wren.app_server.app_conversation.app_conversation_models import ( AppConversationInfoPage, ) # Create test data with different sandbox IDs sandbox_id_alpha = 'sandbox-alpha-123' sandbox_id_beta = 'sandbox-beta-456' conv_alpha = AppConversationInfo( id=uuid4(), created_by_user_id=None, sandbox_id=sandbox_id_alpha, title='Alpha Conversation', ) conv_beta = AppConversationInfo( id=uuid4(), created_by_user_id=None, sandbox_id=sandbox_id_beta, title='Beta Conversation', ) # Mock the info service to return filtered results based on sandbox_id__eq async def mock_search(sandbox_id__eq=None, **kwargs): if sandbox_id__eq == sandbox_id_alpha: return AppConversationInfoPage(items=[conv_alpha]) elif sandbox_id__eq == sandbox_id_beta: return AppConversationInfoPage(items=[conv_beta]) else: return AppConversationInfoPage(items=[conv_alpha, conv_beta]) self.mock_app_conversation_info_service.search_app_conversation_info = ( AsyncMock(side_effect=mock_search) ) # Mock sandbox service to return running status for sandbox lookups self.mock_sandbox_service.batch_get_sandboxes = AsyncMock(return_value=[]) # Test filtering by sandbox_id_alpha result = await self.service.search_app_conversations( sandbox_id__eq=sandbox_id_alpha ) # Verify the info service was called with the correct sandbox_id__eq self.mock_app_conversation_info_service.search_app_conversation_info.assert_called() call_kwargs = self.mock_app_conversation_info_service.search_app_conversation_info.call_args[ 1 ] assert call_kwargs.get('sandbox_id__eq') == sandbox_id_alpha # Verify only alpha conversation is returned assert len(result.items) == 1 assert result.items[0].sandbox_id == sandbox_id_alpha @pytest.mark.asyncio async def test_count_app_conversations_with_sandbox_id_filter(self): """Test that count_app_conversations passes sandbox_id__eq to the info service. This verifies that the sandbox_id filter is correctly propagated through the service layer to the underlying info service for count operations. """ sandbox_id = 'sandbox-count-test-789' # Mock the info service to return count based on sandbox_id__eq async def mock_count(sandbox_id__eq=None, **kwargs): if sandbox_id__eq == sandbox_id: return 3 # 3 conversations match this sandbox else: return 10 # 10 total conversations self.mock_app_conversation_info_service.count_app_conversation_info = AsyncMock( side_effect=mock_count ) # Test counting with sandbox_id filter result = await self.service.count_app_conversations(sandbox_id__eq=sandbox_id) # Verify the info service was called with the correct sandbox_id__eq self.mock_app_conversation_info_service.count_app_conversation_info.assert_called_once() call_kwargs = self.mock_app_conversation_info_service.count_app_conversation_info.call_args[ 1 ] assert call_kwargs.get('sandbox_id__eq') == sandbox_id # Verify filtered count is returned assert result == 3 @pytest.mark.asyncio async def test_search_app_conversations_sandbox_id_filter_returns_empty(self): """Test that search with non-matching sandbox_id returns empty results.""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationInfoPage, ) # Mock the info service to return empty for non-matching sandbox self.mock_app_conversation_info_service.search_app_conversation_info = ( AsyncMock(return_value=AppConversationInfoPage(items=[])) ) self.mock_sandbox_service.batch_get_sandboxes = AsyncMock(return_value=[]) # Test filtering by non-existent sandbox_id result = await self.service.search_app_conversations( sandbox_id__eq='non-existent-sandbox' ) # Verify empty results assert len(result.items) == 0 class TestPluginHandling: """Test cases for plugin-related functionality in LiveStatusAppConversationService.""" def setup_method(self): """Set up test fixtures.""" # Create mock dependencies self.mock_user_context = Mock(spec=UserContext) self.mock_user_auth = Mock() self.mock_user_context.user_auth = self.mock_user_auth self.mock_user_context.get_user_email = AsyncMock(return_value=None) self.mock_jwt_service = Mock() self.mock_sandbox_service = Mock() self.mock_sandbox_spec_service = Mock() self.mock_app_conversation_info_service = Mock() self.mock_app_conversation_start_task_service = Mock() self.mock_event_callback_service = Mock() self.mock_event_service = Mock() self.mock_httpx_client = Mock() self.mock_pending_message_service = Mock() # Create service instance self.service = LiveStatusAppConversationService( init_git_in_empty_workspace=True, user_context=self.mock_user_context, app_conversation_info_service=self.mock_app_conversation_info_service, app_conversation_start_task_service=self.mock_app_conversation_start_task_service, event_callback_service=self.mock_event_callback_service, event_service=self.mock_event_service, sandbox_service=self.mock_sandbox_service, sandbox_spec_service=self.mock_sandbox_spec_service, jwt_service=self.mock_jwt_service, pending_message_service=self.mock_pending_message_service, sandbox_startup_timeout=30, sandbox_startup_poll_frequency=1, max_num_conversations_per_sandbox=20, httpx_client=self.mock_httpx_client, web_url='https://test.example.com', wren_provider_base_url='https://provider.example.com', access_token_hard_timeout=None, app_mode='test', ) # Mock user info self.mock_user = _TestUserInfo( id='test_user_123', llm_model='gpt-4', llm_base_url='https://api.openai.com/v1', llm_api_key='test_api_key', confirmation_mode=False, search_api_key=None, mcp_config=None, security_analyzer=None, ) # Mock sandbox self.mock_sandbox = Mock(spec=SandboxInfo) self.mock_sandbox.id = uuid4() self.mock_sandbox.status = SandboxStatus.RUNNING def test_construct_initial_message_with_plugin_params_no_plugins(self): """Test _construct_initial_message_with_plugin_params with no plugins returns original message.""" from wren.agent_server.models import SendMessageRequest, TextContent # Test with None initial message and None plugins result = self.service._construct_initial_message_with_plugin_params(None, None) assert result is None # Test with None initial message and empty plugins list result = self.service._construct_initial_message_with_plugin_params(None, []) assert result is None # Test with initial message but None plugins initial_msg = SendMessageRequest(content=[TextContent(text='Hello world')]) result = self.service._construct_initial_message_with_plugin_params( initial_msg, None ) assert result is initial_msg def test_construct_initial_message_with_plugin_params_no_params(self): """Test _construct_initial_message_with_plugin_params with plugins but no parameters.""" from wren.agent_server.models import SendMessageRequest, TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) # Plugin with no parameters plugins = [PluginSpec(source='github:owner/repo')] # Test with None initial message result = self.service._construct_initial_message_with_plugin_params( None, plugins ) assert result is None # Test with initial message initial_msg = SendMessageRequest(content=[TextContent(text='Hello world')]) result = self.service._construct_initial_message_with_plugin_params( initial_msg, plugins ) assert result is initial_msg def test_construct_initial_message_with_plugin_params_creates_new_message(self): """Test _construct_initial_message_with_plugin_params creates message when no initial message.""" from wren.agent_server.models import TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugins = [ PluginSpec( source='github:owner/repo', parameters={'api_key': 'test123', 'debug': True}, ) ] result = self.service._construct_initial_message_with_plugin_params( None, plugins ) assert result is not None assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert 'Plugin Configuration Parameters:' in result.content[0].text assert '- api_key: test123' in result.content[0].text assert '- debug: True' in result.content[0].text assert result.run is True def test_construct_initial_message_with_plugin_params_appends_to_message(self): """Test _construct_initial_message_with_plugin_params appends to existing message.""" from wren.agent_server.models import SendMessageRequest, TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) initial_msg = SendMessageRequest( content=[TextContent(text='Please analyze this codebase')], run=False, ) plugins = [ PluginSpec( source='github:owner/repo', ref='v1.0.0', parameters={'target_dir': '/src', 'verbose': True}, ) ] result = self.service._construct_initial_message_with_plugin_params( initial_msg, plugins ) assert result is not None assert len(result.content) == 1 text = result.content[0].text assert text.startswith('Please analyze this codebase') assert 'Plugin Configuration Parameters:' in text assert '- target_dir: /src' in text assert '- verbose: True' in text assert result.run is False def test_construct_initial_message_with_plugin_params_preserves_role(self): """Test _construct_initial_message_with_plugin_params preserves message role.""" from wren.agent_server.models import SendMessageRequest, TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) initial_msg = SendMessageRequest( role='system', content=[TextContent(text='System message')], ) plugins = [PluginSpec(source='github:owner/repo', parameters={'key': 'value'})] result = self.service._construct_initial_message_with_plugin_params( initial_msg, plugins ) assert result is not None assert result.role == 'system' def test_construct_initial_message_with_plugin_params_empty_content(self): """Test _construct_initial_message_with_plugin_params handles empty content list.""" from wren.agent_server.models import SendMessageRequest, TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) initial_msg = SendMessageRequest(content=[]) plugins = [PluginSpec(source='github:owner/repo', parameters={'key': 'value'})] result = self.service._construct_initial_message_with_plugin_params( initial_msg, plugins ) assert result is not None assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert 'Plugin Configuration Parameters:' in result.content[0].text def test_construct_initial_message_with_multiple_plugins(self): """Test _construct_initial_message_with_plugin_params handles multiple plugins.""" from wren.agent_server.models import TextContent from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugins = [ PluginSpec( source='github:owner/plugin1', parameters={'key1': 'value1'}, ), PluginSpec( source='github:owner/plugin2', parameters={'key2': 'value2'}, ), ] result = self.service._construct_initial_message_with_plugin_params( None, plugins ) assert result is not None assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) text = result.content[0].text assert 'Plugin Configuration Parameters:' in text # Multiple plugins should show grouped by plugin name assert 'plugin1' in text assert 'plugin2' in text assert 'key1: value1' in text assert 'key2: value2' in text @pytest.mark.asyncio @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) async def test_build_request_with_plugins(self, _mock_tools): """Plugins are converted to PluginSource and included in the request.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) plugins = [ PluginSpec( source='github:owner/my-plugin', ref='v1.0.0', parameters={'api_key': 'test123'}, ) ] result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace', plugins=plugins, ) assert isinstance(result, StartConversationRequest) assert result.plugins is not None assert len(result.plugins) == 1 assert result.plugins[0].source == 'github:owner/my-plugin' assert result.plugins[0].ref == 'v1.0.0' # Plugin params are folded into the initial message assert result.initial_message is not None assert ( 'Plugin Configuration Parameters:' in result.initial_message.content[0].text ) assert '- api_key: test123' in result.initial_message.content[0].text @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_without_plugins(self, _mock_tools): """Without plugins, result.plugins is None.""" self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace', ) assert isinstance(result, StartConversationRequest) assert result.plugins is None @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_plugin_with_repo_path(self, _mock_tools): """repo_path is propagated through to PluginSource.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) plugins = [ PluginSpec( source='github:owner/marketplace-repo', ref='main', repo_path='plugins/city-weather', ) ] result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace', plugins=plugins, ) assert result.plugins is not None assert len(result.plugins) == 1 assert result.plugins[0].source == 'github:owner/marketplace-repo' assert result.plugins[0].ref == 'main' assert result.plugins[0].repo_path == 'plugins/city-weather' @patch( 'wren.app_server.app_conversation.live_status_app_conversation_service.get_default_tools', return_value=[], ) @pytest.mark.asyncio async def test_build_request_multiple_plugins(self, _mock_tools): """Multiple plugins are all converted correctly.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) self.mock_user_context.get_user_info.return_value = self.mock_user real_llm = LLM(model='gpt-4', api_key=SecretStr('test-key')) self.service._setup_secrets_for_git_providers = AsyncMock(return_value={}) self.service._configure_llm_and_mcp = AsyncMock(return_value=(real_llm, {})) plugins = [ PluginSpec(source='github:owner/security-plugin', ref='v2.0.0'), PluginSpec( source='github:owner/monorepo', repo_path='plugins/logging', ), PluginSpec(source='/local/path/to/plugin'), ] result = await self.service._build_start_conversation_request_for_user( sandbox=self.mock_sandbox, conversation_id=uuid4(), initial_message=None, system_message_suffix=None, git_provider=None, working_dir='/workspace', plugins=plugins, ) assert result.plugins is not None assert len(result.plugins) == 3 assert result.plugins[0].source == 'github:owner/security-plugin' assert result.plugins[0].ref == 'v2.0.0' assert result.plugins[1].source == 'github:owner/monorepo' assert result.plugins[1].repo_path == 'plugins/logging' assert result.plugins[2].source == '/local/path/to/plugin' class TestPluginSpecModel: """Test cases for the PluginSpec model.""" def test_plugin_spec_with_all_fields(self): """Test PluginSpec with all fields provided.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec( source='github:owner/repo', ref='v1.0.0', repo_path='plugins/my-plugin', parameters={'key1': 'value1', 'key2': 123, 'key3': True}, ) assert plugin.source == 'github:owner/repo' assert plugin.ref == 'v1.0.0' assert plugin.repo_path == 'plugins/my-plugin' assert plugin.parameters == {'key1': 'value1', 'key2': 123, 'key3': True} def test_plugin_spec_with_only_source(self): """Test PluginSpec with only source provided.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='https://github.com/owner/repo.git') assert plugin.source == 'https://github.com/owner/repo.git' assert plugin.ref is None assert plugin.repo_path is None assert plugin.parameters is None def test_plugin_spec_serialization(self): """Test PluginSpec serialization to JSON.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec( source='github:owner/repo', ref='main', repo_path='plugins/my-plugin', parameters={'debug': True}, ) json_data = plugin.model_dump() assert json_data == { 'source': 'github:owner/repo', 'ref': 'main', 'repo_path': 'plugins/my-plugin', 'parameters': {'debug': True}, } def test_plugin_spec_deserialization(self): """Test PluginSpec deserialization from dict.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) data = { 'source': 'github:owner/repo', 'ref': 'v2.0.0', 'repo_path': 'plugins/weather', 'parameters': {'timeout': 30}, } plugin = PluginSpec.model_validate(data) assert plugin.source == 'github:owner/repo' assert plugin.ref == 'v2.0.0' assert plugin.repo_path == 'plugins/weather' assert plugin.parameters == {'timeout': 30} def test_plugin_spec_display_name_github_format(self): """Test display_name extracts repo name from github:owner/repo format.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='github:owner/my-plugin') assert plugin.display_name == 'my-plugin' def test_plugin_spec_display_name_git_url(self): """Test display_name extracts repo name from git URL.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='https://github.com/owner/repo.git') assert plugin.display_name == 'repo.git' def test_plugin_spec_display_name_local_path(self): """Test display_name extracts directory name from local path.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='/local/path/to/plugin') assert plugin.display_name == 'plugin' def test_plugin_spec_display_name_no_slash(self): """Test display_name returns source as-is when no slash present.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='local-plugin') assert plugin.display_name == 'local-plugin' def test_plugin_spec_format_params_as_text(self): """Test format_params_as_text formats parameters as text.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec( source='github:owner/repo', parameters={'key1': 'value1', 'key2': 123}, ) result = plugin.format_params_as_text() assert result == '- key1: value1\n- key2: 123' def test_plugin_spec_format_params_as_text_with_indent(self): """Test format_params_as_text with custom indent.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec( source='github:owner/repo', parameters={'debug': True}, ) result = plugin.format_params_as_text(indent=' ') assert result == ' - debug: True' def test_plugin_spec_format_params_as_text_no_params(self): """Test format_params_as_text returns None when no parameters.""" from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) plugin = PluginSpec(source='github:owner/repo') assert plugin.format_params_as_text() is None def test_plugin_spec_inherits_repo_path_validation(self): """Test PluginSpec inherits validation from SDK's PluginSource.""" import pytest from wren.app_server.app_conversation.app_conversation_models import ( PluginSpec, ) # Should reject absolute paths with pytest.raises(ValueError, match='must be relative'): PluginSpec(source='github:owner/repo', repo_path='/absolute/path') # Should reject parent traversal with pytest.raises(ValueError, match="cannot contain '..'"): PluginSpec(source='github:owner/repo', repo_path='../parent/path') class TestAppConversationStartRequestWithPlugins: """Test cases for AppConversationStartRequest with plugins field.""" def test_start_request_with_plugins(self): """Test AppConversationStartRequest with plugins field.""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationStartRequest, PluginSpec, ) plugins = [ PluginSpec( source='github:owner/my-plugin', ref='v1.0.0', parameters={'api_key': 'test'}, ) ] request = AppConversationStartRequest( title='Test conversation', plugins=plugins, ) assert request.plugins is not None assert len(request.plugins) == 1 assert request.plugins[0].source == 'github:owner/my-plugin' assert request.plugins[0].ref == 'v1.0.0' assert request.plugins[0].parameters == {'api_key': 'test'} def test_start_request_without_plugins(self): """Test AppConversationStartRequest without plugins field (backwards compatible).""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationStartRequest, ) request = AppConversationStartRequest( title='Test conversation', ) assert request.plugins is None def test_start_request_serialization_with_plugins(self): """Test AppConversationStartRequest serialization includes plugins.""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationStartRequest, PluginSpec, ) plugins = [PluginSpec(source='github:owner/repo')] request = AppConversationStartRequest(plugins=plugins) json_data = request.model_dump() assert 'plugins' in json_data assert len(json_data['plugins']) == 1 assert json_data['plugins'][0]['source'] == 'github:owner/repo' def test_start_request_deserialization_with_plugins(self): """Test AppConversationStartRequest deserialization from JSON with plugins.""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationStartRequest, ) data = { 'title': 'Test', 'plugins': [ { 'source': 'github:owner/plugin', 'ref': 'main', 'parameters': {'key': 'value'}, }, ], } request = AppConversationStartRequest.model_validate(data) assert request.plugins is not None assert len(request.plugins) == 1 assert request.plugins[0].source == 'github:owner/plugin' assert request.plugins[0].ref == 'main' assert request.plugins[0].parameters == {'key': 'value'} def test_start_request_with_multiple_plugins(self): """Test AppConversationStartRequest with multiple plugins.""" from wren.app_server.app_conversation.app_conversation_models import ( AppConversationStartRequest, PluginSpec, ) plugins = [ PluginSpec(source='github:owner/plugin1', ref='v1.0.0'), PluginSpec(source='github:owner/plugin2', repo_path='plugins/sub'), PluginSpec(source='/local/path'), ] request = AppConversationStartRequest( title='Test conversation', plugins=plugins, ) assert request.plugins is not None assert len(request.plugins) == 3 assert request.plugins[0].source == 'github:owner/plugin1' assert request.plugins[1].repo_path == 'plugins/sub' assert request.plugins[2].source == '/local/path' class TestLoadHooksFromWorkspace: """Test cases for _load_hooks_from_workspace method.""" def setup_method(self): """Set up test fixtures.""" # Create mock dependencies self.mock_user_context = Mock(spec=UserContext) self.mock_jwt_service = Mock() self.mock_sandbox_service = Mock() self.mock_sandbox_spec_service = Mock() self.mock_app_conversation_info_service = Mock() self.mock_app_conversation_start_task_service = Mock() self.mock_event_callback_service = Mock() self.mock_event_service = Mock() self.mock_httpx_client = AsyncMock() self.mock_pending_message_service = Mock() # Create service instance self.service = LiveStatusAppConversationService( init_git_in_empty_workspace=True, user_context=self.mock_user_context, app_conversation_info_service=self.mock_app_conversation_info_service, app_conversation_start_task_service=self.mock_app_conversation_start_task_service, event_callback_service=self.mock_event_callback_service, event_service=self.mock_event_service, sandbox_service=self.mock_sandbox_service, sandbox_spec_service=self.mock_sandbox_spec_service, jwt_service=self.mock_jwt_service, pending_message_service=self.mock_pending_message_service, sandbox_startup_timeout=30, sandbox_startup_poll_frequency=1, max_num_conversations_per_sandbox=20, httpx_client=self.mock_httpx_client, web_url='https://test.example.com', wren_provider_base_url='https://provider.example.com', access_token_hard_timeout=None, app_mode='test', ) @pytest.mark.asyncio async def test_load_hooks_from_workspace_success(self): """Test loading hooks from workspace when hooks.json exists.""" # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {'X-Session-API-Key': 'test-key'} hooks_response = { 'hook_config': { 'stop': [ { 'matcher': '*', 'hooks': [{'type': 'command', 'command': 'echo "stop hook"'}], } ] } } mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = hooks_response mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Act result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace' ) # Assert assert result is not None assert not result.is_empty() self.mock_httpx_client.post.assert_called_once_with( 'http://agent-server:8000/api/hooks', json={'project_dir': '/workspace'}, headers={ 'Content-Type': 'application/json', 'X-Session-API-Key': 'test-key', }, timeout=30.0, ) @pytest.mark.asyncio async def test_load_hooks_from_workspace_file_not_found(self): """Test loading hooks when hooks.json does not exist.""" # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {} # Agent server returns hook_config: None when file not found mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = {'hook_config': None} mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Act result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace' ) # Assert assert result is None @pytest.mark.asyncio async def test_load_hooks_from_workspace_empty_hooks(self): """Test loading hooks when hooks.json is empty or has no hooks.""" # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {} # Agent server returns empty hook_config mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = {'hook_config': {}} mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Act result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace' ) # Assert assert result is None @pytest.mark.asyncio async def test_load_hooks_from_workspace_http_error(self): """Test loading hooks when HTTP request fails.""" # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {} self.mock_httpx_client.post = AsyncMock( side_effect=Exception('Connection error') ) # Act result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace' ) # Assert assert result is None def test_get_project_dir_for_hooks_with_selected_repository(self): """Test get_project_dir_for_hooks with a selected repository.""" from wren.app_server.app_conversation.hook_loader import ( get_project_dir_for_hooks, ) result = get_project_dir_for_hooks( '/workspace/project', 'OpenHands/software-agent-sdk', ) assert result == '/workspace/project/software-agent-sdk' def test_get_project_dir_for_hooks_without_selected_repository(self): """Test get_project_dir_for_hooks without a selected repository.""" from wren.app_server.app_conversation.hook_loader import ( get_project_dir_for_hooks, ) result = get_project_dir_for_hooks('/workspace/project', None) assert result == '/workspace/project' def test_get_project_dir_for_hooks_with_empty_string(self): """Test get_project_dir_for_hooks with empty string repository.""" from wren.app_server.app_conversation.hook_loader import ( get_project_dir_for_hooks, ) # Empty string should be treated as no repository result = get_project_dir_for_hooks('/workspace/project', '') assert result == '/workspace/project' @pytest.mark.asyncio async def test_load_hooks_from_workspace_with_project_dir(self): """Test loading hooks with a pre-resolved project_dir. The caller is responsible for computing the project_dir (which already includes the repo name when a repo is selected). _load_hooks_from_workspace should use the project_dir as-is. """ # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {'X-Session-API-Key': 'test-key'} hooks_response = { 'hook_config': { 'stop': [ { 'matcher': '*', 'hooks': [{'type': 'command', 'command': 'echo "stop hook"'}], } ] } } mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = hooks_response mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Act - project_dir already includes repo name result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace/project/software-agent-sdk', ) # Assert assert result is not None assert not result.is_empty() # The project_dir should be passed as-is without doubling self.mock_httpx_client.post.assert_called_once_with( 'http://agent-server:8000/api/hooks', json={'project_dir': '/workspace/project/software-agent-sdk'}, headers={ 'Content-Type': 'application/json', 'X-Session-API-Key': 'test-key', }, timeout=30.0, ) @pytest.mark.asyncio async def test_load_hooks_from_workspace_base_dir(self): """Test loading hooks with a base workspace directory (no repo selected).""" # Arrange mock_remote_workspace = Mock(spec=AsyncRemoteWorkspace) mock_remote_workspace.host = 'http://agent-server:8000' mock_remote_workspace._headers = {'X-Session-API-Key': 'test-key'} hooks_response = { 'hook_config': { 'stop': [ { 'matcher': '*', 'hooks': [{'type': 'command', 'command': 'echo "stop hook"'}], } ] } } mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = hooks_response mock_response.raise_for_status = Mock() self.mock_httpx_client.post = AsyncMock(return_value=mock_response) # Act - no repo selected, project_dir is base working_dir result = await self.service._load_hooks_from_workspace( mock_remote_workspace, '/workspace/project', ) # Assert assert result is not None self.mock_httpx_client.post.assert_called_once_with( 'http://agent-server:8000/api/hooks', json={'project_dir': '/workspace/project'}, headers={ 'Content-Type': 'application/json', 'X-Session-API-Key': 'test-key', }, timeout=30.0, ) class TestAgentKindConversationUrl: """Regression tests for conversation_url / live-status route dispatch. Both LLM and ACP conversations are served by the unified ``/api/conversations`` endpoint (the SDK's ``AgentBase`` discriminated union accepts both ``Agent`` and ``ACPAgent`` payloads on that route). Getting this wrong would make ACP conversations look stuck on "Loading" because the frontend polls the wrong route and 404s. """ @pytest.mark.parametrize('agent_kind', ['wren', 'acp']) def test_build_conversation_url_uses_unified_path(self, agent_kind): from uuid import UUID from wren.app_server.app_conversation.app_conversation_models import ( AppConversationInfo, ) from wren.app_server.sandbox.sandbox_models import ( AGENT_SERVER, ExposedUrl, SandboxInfo, SandboxStatus, ) # Instantiate a stripped service (no deps needed for _build_conversation). service = LiveStatusAppConversationService.__new__( LiveStatusAppConversationService ) info = AppConversationInfo( id=UUID('11111111-1111-1111-1111-111111111111'), created_by_user_id=None, sandbox_id='sandbox-a', agent_kind=agent_kind, ) sandbox = SandboxInfo( id='sandbox-a', created_by_user_id=None, sandbox_spec_id='spec', status=SandboxStatus.RUNNING, session_api_key='sk', exposed_urls=[ ExposedUrl(name=AGENT_SERVER, url='http://localhost:8000', port=8000), ], ) result = service._build_conversation(info, sandbox, None) assert result is not None assert result.conversation_url == ( 'http://localhost:8000/api/conversations/11111111111111111111111111111111' ) class TestBuildAcpStartConversationRequestSecrets: """Tests for secret injection in ``_build_acp_start_conversation_request``. Secrets from the Secrets panel and git provider tokens flow through ``request.secrets`` as the sole canonical channel. ``llm.api_key`` is never read for ACP (the CLI subprocess handles its own LLM calls). """ @pytest.fixture def service(self): mock_user_context = Mock(spec=UserContext) return LiveStatusAppConversationService( init_git_in_empty_workspace=True, user_context=mock_user_context, app_conversation_info_service=Mock(), app_conversation_start_task_service=Mock(), event_callback_service=Mock(), event_service=Mock(), sandbox_service=Mock(), sandbox_spec_service=Mock(), jwt_service=Mock(), pending_message_service=Mock(), sandbox_startup_timeout=30, sandbox_startup_poll_frequency=1, max_num_conversations_per_sandbox=20, httpx_client=Mock(), web_url=None, wren_provider_base_url=None, access_token_hard_timeout=None, app_mode='test', ) def _make_acp_user( self, acp_server='claude-code', context_secrets=None, api_key=None ): try: from wren.sdk.settings import ( ACPAgentSettings, # type: ignore[attr-defined] ) except ImportError: pytest.skip('ACPAgentSettings not available in this SDK build') user = _TestUserInfo( id='user1', llm_model='', llm_base_url=None, llm_api_key=None, sandbox_grouping_strategy=SandboxGroupingStrategy.ADD_TO_ANY, confirmation_mode=False, security_analyzer=None, search_api_key=None, mcp_config=None, disabled_skills=[], ) agent_context = ( AgentContext(secrets=context_secrets) if context_secrets else None ) user.agent_settings = ACPAgentSettings( acp_server=acp_server, # type: ignore[arg-type] llm=LLM( model='claude-sonnet-4-5', api_key=SecretStr(api_key) if api_key else None, ), agent_context=agent_context, ) return user def _call_build( self, service, user, tmp_path, *, secrets=None, git_provider=None, selected_repository=None, selected_branch=None, remote_workspace=None, ): """Wire user_context and call _build_acp_start_conversation_request.""" service.user_context.get_user_info = AsyncMock(return_value=user) service.user_context.get_user_email = AsyncMock(return_value=None) service.user_context.get_secrets = AsyncMock(return_value=secrets or {}) service.user_context.get_provider_tokens = AsyncMock(return_value=None) sandbox = Mock(spec=SandboxInfo) return service._build_acp_start_conversation_request( sandbox=sandbox, conversation_id=uuid4(), initial_message=None, working_dir=str(tmp_path), git_provider=git_provider, selected_repository=selected_repository, selected_branch=selected_branch, remote_workspace=remote_workspace, plugins=None, ) @pytest.mark.asyncio async def test_secrets_passed_via_request_secrets(self, service, tmp_path): """Secrets are forwarded via request.secrets as SecretSource objects.""" github_secret = StaticSecret(value=SecretStr('ghp_test123')) api_secret = StaticSecret(value=SecretStr('secret-value')) user = self._make_acp_user() request = await self._call_build( service, user, tmp_path, secrets={'GITHUB_TOKEN': github_secret, 'MY_API_KEY': api_secret}, ) assert request.secrets.get('GITHUB_TOKEN') is github_secret assert request.secrets.get('MY_API_KEY') is api_secret # Isolation is enabled regardless of whether secrets are present. assert request.agent.acp_isolate_data_dir is True @pytest.mark.asyncio async def test_lookup_secret_forwarded_as_source(self, service, tmp_path): """LookupSecrets from user_context are forwarded as-is.""" lookup = LookupSecret(url='https://example.com/token', headers={}) user = self._make_acp_user() request = await self._call_build( service, user, tmp_path, secrets={'GITHUB_TOKEN': lookup}, ) assert request.secrets.get('GITHUB_TOKEN') is lookup @pytest.mark.asyncio async def test_explicit_context_secret_preserved(self, service, tmp_path): """Explicit agent_context.secrets entries survive when secrets also present.""" user = self._make_acp_user(context_secrets={'MY_TOKEN': 'explicit-override'}) other_secret = StaticSecret(value=SecretStr('other-value')) request = await self._call_build( service, user, tmp_path, secrets={'OTHER': other_secret}, ) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert agent_ctx_secrets.get('MY_TOKEN') == 'explicit-override' assert request.secrets.get('OTHER') is other_secret @pytest.mark.asyncio async def test_llm_api_key_not_forwarded_to_request_secrets( self, service, tmp_path ): """llm.api_key must not bleed into request.secrets for ACP. ACP does not use the SDK's LLM layer — the CLI subprocess handles its own model calls. Credentials must be supplied via the Secrets panel (request.secrets channel), not via llm.api_key. """ user = self._make_acp_user(acp_server='claude-code', api_key='sk-ui-key') request = await self._call_build(service, user, tmp_path) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert 'ANTHROPIC_API_KEY' not in agent_ctx_secrets assert 'ANTHROPIC_API_KEY' not in request.secrets @pytest.mark.asyncio async def test_no_secrets_request_secrets_empty(self, service, tmp_path): """When there are no panel secrets, request.secrets is empty.""" user = self._make_acp_user() request = await self._call_build(service, user, tmp_path) assert request.secrets == {} @pytest.mark.asyncio async def test_isolate_data_dir_enabled(self, service, tmp_path): """The cloud start path builds the ACP agent with data-dir isolation on so the CLI session subtree lives on /workspace and the SDK self-resumes across pause/resume (#1274).""" user = self._make_acp_user() request = await self._call_build(service, user, tmp_path) assert request.agent.acp_isolate_data_dir is True @pytest.mark.asyncio async def test_context_secret_explicit_override(self, service, tmp_path): """Explicit agent_context.secrets is independent of request.secrets — both preserved.""" user = self._make_acp_user( acp_server='claude-code', context_secrets={'ANTHROPIC_API_KEY': 'sk-explicit-override'}, ) request = await self._call_build(service, user, tmp_path) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert agent_ctx_secrets.get('ANTHROPIC_API_KEY') == 'sk-explicit-override' # No panel secrets → request.secrets is empty (agent_context.secrets is a separate channel). assert 'ANTHROPIC_API_KEY' not in request.secrets @pytest.mark.asyncio async def test_secrets_forwarded_via_request_secrets(self, service, tmp_path): """Panel secrets flow through request.secrets; not pre-resolved into agent_context.""" gh_secret = StaticSecret(value=SecretStr('ghp_test123')) user = self._make_acp_user() request = await self._call_build( service, user, tmp_path, secrets={'GH_TOKEN': gh_secret}, ) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert 'GH_TOKEN' not in agent_ctx_secrets assert request.secrets.get('GH_TOKEN') is gh_secret @pytest.mark.asyncio async def test_panel_secret_is_sole_credentials_channel(self, service, tmp_path): """Panel secret is the only credentials channel; llm.api_key is ignored. When a panel secret and llm.api_key both target the same env var, only the panel secret appears in request.secrets — llm.api_key is not read. """ user = self._make_acp_user(acp_server='claude-code', api_key='sk-ui-key') panel_secret = StaticSecret(value=SecretStr('sk-from-secrets-panel')) request = await self._call_build( service, user, tmp_path, secrets={'ANTHROPIC_API_KEY': panel_secret}, ) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert 'ANTHROPIC_API_KEY' not in agent_ctx_secrets assert request.secrets.get('ANTHROPIC_API_KEY') is panel_secret @pytest.mark.asyncio async def test_explicit_context_secret_and_panel_secret_coexist( self, service, tmp_path ): """agent_context.secrets and request.secrets are independent channels. An explicit agent_context.secrets entry is available to the subprocess, while the panel secret still flows through request.secrets unchanged. """ panel_secret = StaticSecret(value=SecretStr('panel-token')) user = self._make_acp_user(context_secrets={'GH_TOKEN': 'explicit-token'}) request = await self._call_build( service, user, tmp_path, secrets={'GH_TOKEN': panel_secret}, ) agent_ctx_secrets = ( request.agent.agent_context.secrets if request.agent.agent_context else {} ) or {} assert agent_ctx_secrets.get('GH_TOKEN') == 'explicit-token' assert request.secrets.get('GH_TOKEN') is panel_secret @pytest.mark.asyncio async def test_observability_metadata_populated(self, service, tmp_path): """Repo / branch / provider land on the request's observability_metadata for ACP conversations too, matching the OpenHands-agent path.""" user = self._make_acp_user() request = await self._call_build( service, user, tmp_path, git_provider=ProviderType.GITHUB, selected_repository='test/repo', selected_branch='feature-x', ) assert request.observability_metadata == { 'repo': 'test/repo', 'branch': 'feature-x', 'git_provider': 'github', } @pytest.mark.asyncio async def test_observability_metadata_includes_commit(self, service, tmp_path): """The post-clone HEAD sha is resolved from the workspace and added, mirroring the OpenHands-agent path's commit resolution.""" user = self._make_acp_user() remote_workspace = Mock(spec=AsyncRemoteWorkspace) remote_workspace.execute_command = AsyncMock( return_value=SimpleNamespace(exit_code=0, stdout='abc123sha\n') ) request = await self._call_build( service, user, tmp_path, git_provider=ProviderType.GITHUB, selected_repository='test/repo', selected_branch='feature-x', remote_workspace=remote_workspace, ) assert request.observability_metadata == { 'repo': 'test/repo', 'branch': 'feature-x', 'git_provider': 'github', 'commit': 'abc123sha', } @pytest.mark.asyncio async def test_no_repo_omits_observability_metadata(self, service, tmp_path): """A repo-less ACP conversation adds no repo metadata to the trace.""" user = self._make_acp_user() request = await self._call_build(service, user, tmp_path) assert request.observability_metadata == {}