From 35b60a752872840dd4b2d6871357163ba0e4b958 Mon Sep 17 00:00:00 2001 From: enyst Date: Wed, 15 Jul 2026 10:39:37 +0000 Subject: [PATCH] Simplify lazy conversation catalog invariants Co-authored-by: openhands --- .../agent_server/conversation_service.py | 146 +++++---- .../agent_server/test_conversation_service.py | 279 ++++++++++++------ 2 files changed, 273 insertions(+), 152 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 37f4edaede..a396d79c8f 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -16,7 +16,9 @@ from openhands.agent_server.config import Config, WebhookSpec from openhands.agent_server.conversation_lease import ( DEFAULT_LEASE_TTL_SECONDS, + ConversationLease, ConversationLeaseHeldError, + ConversationOwnershipLostError, ) from openhands.agent_server.event_service import ( LEASE_RENEW_INTERVAL_SECONDS, @@ -501,6 +503,7 @@ class ConversationService: owner_instance_id: str = field(default_factory=lambda: uuid4().hex) max_concurrent_runs: int = 10 lease_ttl_seconds: float = DEFAULT_LEASE_TTL_SECONDS + # Every live runtime has a catalog record; unloaded records have no runtime. _event_services: dict[UUID, EventService] | None = field(default=None, init=False) _conversation_records: dict[UUID, _ConversationRecord] = field( default_factory=dict, init=False @@ -594,25 +597,6 @@ async def _execution_status( record.execution_status = state.execution_status return record.execution_status - async def _reconcile_active_records(self) -> None: - """Fill catalog entries for services injected outside normal startup. - - Normal service lifecycle paths maintain the catalog themselves. This - small reconciliation keeps direct embedders and existing test fixtures - that populate ``_event_services`` compatible. - """ - event_services = self._event_services - if event_services is None: - raise ValueError("inactive_service") - for conversation_id, event_service in event_services.items(): - if conversation_id in self._conversation_records: - continue - state = await event_service.get_state() - self._conversation_records[conversation_id] = _ConversationRecord( - stored=event_service.stored, - execution_status=state.execution_status, - ) - def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: if stored.tool_module_qualnames: for tool_name, module_qualname in stored.tool_module_qualnames.items(): @@ -679,16 +663,7 @@ async def get_conversation(self, conversation_id: UUID) -> ConversationInfo | No raise ValueError("inactive_service") record = self._conversation_records.get(conversation_id) if record is None: - event_service = self._event_services.get(conversation_id) - if event_service is None: - return None - state = await event_service.get_state() - record = _ConversationRecord( - stored=event_service.stored, - execution_status=state.execution_status, - ) - self._conversation_records[conversation_id] = record - return _compose_conversation_info(event_service.stored, state) + return None return await self._conversation_info(conversation_id, record) async def get_acp_conversation( @@ -741,7 +716,6 @@ async def _search_conversations( ) -> tuple[list[ConversationInfo], str | None]: if self._event_services is None: raise ValueError("inactive_service") - await self._reconcile_active_records() records = list(self._conversation_records.items()) if sort_order in ( @@ -798,7 +772,6 @@ async def _count_conversations( """Count conversations matching the given filters.""" if self._event_services is None: raise ValueError("inactive_service") - await self._reconcile_active_records() if execution_status is None: return len(self._conversation_records) @@ -887,10 +860,6 @@ async def _start_conversation( existing_event_service = self._event_services.get(conversation_id) if existing_event_service is not None and existing_event_service.is_open(): state = await existing_event_service.get_state() - self._conversation_records[conversation_id] = _ConversationRecord( - stored=existing_event_service.stored, - execution_status=state.execution_status, - ) return ( _compose_conversation_info(existing_event_service.stored, state), False, @@ -1081,45 +1050,96 @@ async def delete_conversation(self, conversation_id: UUID) -> bool: event_services = self._event_services if event_services is None: raise ValueError("inactive_service") - event_service = await self._get_or_load_event_service_locked( - conversation_id - ) - if event_service is None: + record = self._conversation_records.get(conversation_id) + if record is None: return False - event_services.pop(conversation_id, None) - self._conversation_records.pop(conversation_id, None) - # Notify conversation webhooks about the stopped conversation before closing - try: - state = await event_service.get_state() - conversation_info = _compose_webhook_conversation_info( - event_service.stored, state + event_service = event_services.get(conversation_id) + if event_service is not None and not event_service.is_open(): + event_services.pop(conversation_id, None) + event_service = None + conversation_dir = ( + event_service.conversation_dir + if event_service is not None + else self.conversations_dir / conversation_id.hex + ) + deletion_lease: ConversationLease | None = None + lease_generation: int | None = None + if event_service is None and self.lease_ttl_seconds > 0: + deletion_lease = ConversationLease( + conversation_dir=conversation_dir, + owner_instance_id=self.owner_instance_id, + ttl_seconds=self.lease_ttl_seconds, ) - conversation_info.execution_status = ( - ConversationExecutionStatus.DELETING + try: + claim = await asyncio.to_thread(deletion_lease.claim) + except ConversationLeaseHeldError as exc: + logger.debug( + "Skipping deletion of conversation %s owned by %s until %s", + conversation_id, + exc.owner_instance_id, + exc.expires_at, + ) + return False + lease_generation = claim.generation + + try: + state = ( + await event_service.get_state() + if event_service is not None + else await asyncio.to_thread( + self._load_persisted_state_sync, conversation_id + ) ) - await self._notify_conversation_webhooks(conversation_info) + if state is not None: + stored = ( + event_service.stored + if event_service is not None + else record.stored + ) + conversation_info = _compose_webhook_conversation_info( + stored, state + ) + conversation_info.execution_status = ( + ConversationExecutionStatus.DELETING + ) + await self._notify_conversation_webhooks(conversation_info) except Exception as e: logger.warning( f"Failed to notify webhooks for conversation {conversation_id}: {e}" ) - # Close the event service - try: - await event_service.close() - except Exception as e: - logger.warning( - f"Failed to close event service for conversation " - f"{conversation_id}: {e}" - ) + if deletion_lease is not None and lease_generation is not None: + try: + await asyncio.to_thread(deletion_lease.renew, lease_generation) + except ConversationOwnershipLostError: + logger.warning( + "Lost ownership before deleting conversation %s", + conversation_id, + ) + return False + + event_services.pop(conversation_id, None) + self._conversation_records.pop(conversation_id, None) + if event_service is not None: + try: + await event_service.close() + except Exception as e: + logger.warning( + f"Failed to close event service for conversation " + f"{conversation_id}: {e}" + ) - # Safely remove only the conversation directory (workspace is preserved). - # This operation may fail due to permission issues, but we don't want that - # to prevent the conversation from being marked as deleted. - safe_rmtree( - event_service.conversation_dir, + removed = safe_rmtree( + conversation_dir, f"conversation directory for {conversation_id}", ) + if ( + not removed + and deletion_lease is not None + and lease_generation is not None + ): + await asyncio.to_thread(deletion_lease.release, lease_generation) logger.info(f"Successfully deleted conversation {conversation_id}") return True @@ -1483,12 +1503,12 @@ async def _start_event_service(self, stored: StoredConversation) -> EventService await event_service.close() raise - event_services[stored.id] = event_service state = await event_service.get_state() self._conversation_records[stored.id] = _ConversationRecord( stored=event_service.stored, execution_status=state.execution_status, ) + event_services[stored.id] = event_service return event_service diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 4803076ffa..6eb3e2a6c1 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -20,6 +20,7 @@ from openhands.agent_server.conversation_service import ( AutoTitleSubscriber, ConversationService, + _ConversationRecord, _get_worktree_start_point, ) from openhands.agent_server.event_service import EventService @@ -76,6 +77,20 @@ def sample_stored_conversation(): ) +def _add_event_service( + conversation_service: ConversationService, + event_service: EventService, + execution_status: ConversationExecutionStatus = ConversationExecutionStatus.IDLE, +) -> None: + assert conversation_service._event_services is not None + conversation_id = event_service.stored.id + conversation_service._conversation_records[conversation_id] = _ConversationRecord( + stored=event_service.stored, + execution_status=execution_status, + ) + conversation_service._event_services[conversation_id] = event_service + + def _create_running_terminal_action(tool_call_id: str = "call_1") -> ActionEvent: tool_call = MessageToolCall.from_chat_tool_call( ChatCompletionMessageToolCall( @@ -585,6 +600,131 @@ async def test_startup_and_search_do_not_hydrate_idle_conversation(tmp_path): assert restarted._event_services == {} +@pytest.mark.asyncio +async def test_delete_unloaded_conversation_without_hydration(persisted_conversation): + conversations_dir, conversation_id = persisted_conversation + + async with ConversationService(conversations_dir=conversations_dir) as service: + with ( + patch.object( + service, + "_start_event_service", + side_effect=AssertionError("delete should not hydrate event history"), + ) as start_event_service, + patch.object( + service, "_notify_conversation_webhooks", new=AsyncMock() + ) as notify, + ): + assert service._event_services == {} + assert await service.delete_conversation(conversation_id) is True + + start_event_service.assert_not_awaited() + assert notify.await_args is not None + assert notify.await_args.args[0].execution_status == ( + ConversationExecutionStatus.DELETING + ) + assert conversation_id not in service._conversation_records + assert service._event_services == {} + assert not (conversations_dir / conversation_id.hex).exists() + + +@pytest.mark.asyncio +async def test_delete_closed_runtime_uses_unloaded_path(persisted_conversation): + conversations_dir, conversation_id = persisted_conversation + + async with ConversationService(conversations_dir=conversations_dir) as service: + record = service._conversation_records[conversation_id] + closed_runtime = AsyncMock(spec=EventService) + closed_runtime.is_open.return_value = False + closed_runtime.stored = record.stored + assert service._event_services is not None + service._event_services[conversation_id] = closed_runtime + + assert await service.delete_conversation(conversation_id) is True + + closed_runtime.get_state.assert_not_awaited() + closed_runtime.close.assert_not_awaited() + assert service._event_services == {} + assert conversation_id not in service._conversation_records + assert not (conversations_dir / conversation_id.hex).exists() + + +@pytest.mark.asyncio +async def test_delete_unloaded_conversation_with_leasing_disabled( + persisted_conversation, +): + conversations_dir, conversation_id = persisted_conversation + service = ConversationService( + conversations_dir=conversations_dir, + lease_ttl_seconds=0, + ) + + with patch( + "openhands.agent_server.conversation_service.ConversationLease", + side_effect=AssertionError("leasing is disabled"), + ): + async with service: + assert await service.delete_conversation(conversation_id) is True + + assert not (conversations_dir / conversation_id.hex).exists() + + +@pytest.mark.asyncio +async def test_delete_unloaded_conversation_respects_live_lease( + persisted_conversation, +): + conversations_dir, conversation_id = persisted_conversation + + async with ConversationService(conversations_dir=conversations_dir) as owner: + assert await owner.get_event_service(conversation_id) is not None + async with ConversationService(conversations_dir=conversations_dir) as observer: + with patch.object( + observer, + "_start_event_service", + side_effect=AssertionError("delete should not hydrate event history"), + ) as start_event_service: + assert await observer.delete_conversation(conversation_id) is False + + start_event_service.assert_not_awaited() + assert conversation_id in observer._conversation_records + assert observer._event_services == {} + assert (conversations_dir / conversation_id.hex).exists() + + +@pytest.mark.asyncio +async def test_waiting_hydration_cannot_restore_unloaded_deleted_conversation( + persisted_conversation, +): + conversations_dir, conversation_id = persisted_conversation + + async with ConversationService(conversations_dir=conversations_dir) as service: + with patch.object( + service, + "_start_event_service", + side_effect=AssertionError("deleted conversation should not hydrate"), + ) as start_event_service: + await service._lifecycle_lock.acquire() + try: + delete_task = asyncio.create_task( + service.delete_conversation(conversation_id) + ) + await asyncio.sleep(0) + getter_task = asyncio.create_task( + service.get_event_service(conversation_id) + ) + await asyncio.sleep(0) + finally: + service._lifecycle_lock.release() + + deleted, event_service = await asyncio.gather(delete_task, getter_task) + + assert deleted is True + assert event_service is None + start_event_service.assert_not_awaited() + assert service._event_services == {} + assert conversation_id not in service._conversation_records + + @pytest.mark.asyncio async def test_concurrent_access_hydrates_conversation_once(tmp_path): conversations_dir = tmp_path / "conversations" @@ -811,6 +951,25 @@ async def test_search_conversations_empty_result(self, conversation_service): assert result.items == [] assert result.next_page_id is None + @pytest.mark.asyncio + async def test_catalog_queries_ignore_runtime_without_record( + self, conversation_service, sample_stored_conversation + ): + mock_service = AsyncMock(spec=EventService) + mock_service.stored = sample_stored_conversation + assert conversation_service._event_services is not None + conversation_service._event_services[sample_stored_conversation.id] = ( + mock_service + ) + + assert ( + await conversation_service.get_conversation(sample_stored_conversation.id) + is None + ) + assert (await conversation_service.search_conversations()).items == [] + assert await conversation_service.count_conversations() == 0 + assert conversation_service._conversation_records == {} + @pytest.mark.asyncio async def test_search_conversations_basic( self, conversation_service, sample_stored_conversation @@ -829,7 +988,7 @@ async def test_search_conversations_basic( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) result = await conversation_service.search_conversations() @@ -872,7 +1031,7 @@ async def test_search_conversations_with_critic_redacts_api_key( execution_status=ConversationExecutionStatus.IDLE, confirmation_policy=stored_conv.confirmation_policy, ) - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) result = await conversation_service.search_conversations() @@ -921,7 +1080,7 @@ async def test_search_conversations_status_filter(self, conversation_service): ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) conversations.append((stored_conv.id, status)) # Test filtering by IDLE status @@ -975,7 +1134,7 @@ async def test_search_conversations_sorting(self, conversation_service): ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) conversations.append(stored_conv) # Test CREATED_AT (ascending) @@ -1050,7 +1209,7 @@ async def test_search_conversations_pagination(self, conversation_service): ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) conversation_ids.append(stored_conv.id) # Test first page with limit 2 @@ -1120,7 +1279,7 @@ async def test_search_conversations_combined_filter_and_sort( ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) # Filter by IDLE status and sort by CREATED_AT_DESC result = await conversation_service.search_conversations( @@ -1148,9 +1307,7 @@ async def test_search_conversations_invalid_page_id( ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[sample_stored_conversation.id] = ( - mock_service - ) + _add_event_service(conversation_service, mock_service) # Use a non-existent page_id invalid_page_id = uuid4().hex @@ -1197,8 +1354,7 @@ async def test_count_conversations_basic( ) mock_service.get_state.return_value = mock_state - conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) result = await conversation_service.count_conversations() assert result == 1 @@ -1237,7 +1393,7 @@ async def test_count_conversations_status_filter(self, conversation_service): ) mock_service.get_state.return_value = mock_state - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) # Test counting all conversations result = await conversation_service.count_conversations() @@ -1296,7 +1452,7 @@ async def test_count_conversations_includes_regular_and_acp( execution_status=ConversationExecutionStatus.IDLE, confirmation_policy=stored_conv.confirmation_policy, ) - conversation_service._event_services[stored_conv.id] = mock_service + _add_event_service(conversation_service, mock_service) assert await conversation_service.count_conversations() == 2 @@ -1763,66 +1919,6 @@ async def test_start_conversation_with_duplicate_id(self, conversation_service): assert result.id == custom_id assert not is_new - @pytest.mark.asyncio - async def test_start_conversation_reuse_checks_is_open(self, conversation_service): - """Test that conversation reuse checks if event service is open.""" - custom_id = uuid4() - - # Create a mock event service that exists but is not open - mock_event_service = AsyncMock(spec=EventService) - mock_event_service.is_open.return_value = False - mock_event_service.stored = StoredConversation( - id=custom_id, - agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), - workspace=LocalWorkspace(working_dir="workspace/project"), - confirmation_policy=NeverConfirm(), - initial_message=None, - metrics=None, - created_at=datetime(2025, 1, 1, 12, 0, 0, tzinfo=UTC), - updated_at=datetime(2025, 1, 1, 12, 30, 0, tzinfo=UTC), - ) - conversation_service._event_services[custom_id] = mock_event_service - - with tempfile.TemporaryDirectory() as temp_dir: - request = StartConversationRequest( - agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), - workspace=LocalWorkspace(working_dir=temp_dir), - confirmation_policy=NeverConfirm(), - conversation_id=custom_id, - ) - - # Mock the _start_event_service method to avoid actual startup - with patch.object( - conversation_service, "_start_event_service" - ) as mock_start: - mock_new_service = AsyncMock(spec=EventService) - mock_new_service.stored = StoredConversation( - id=custom_id, - agent=request.agent, - workspace=request.workspace, - confirmation_policy=request.confirmation_policy, - initial_message=request.initial_message, - metrics=None, - created_at=datetime(2025, 1, 1, 12, 0, 0, tzinfo=UTC), - updated_at=datetime(2025, 1, 1, 12, 30, 0, tzinfo=UTC), - ) - mock_state = ConversationState( - id=custom_id, - agent=request.agent, - workspace=request.workspace, - execution_status=ConversationExecutionStatus.IDLE, - confirmation_policy=request.confirmation_policy, - ) - mock_new_service.get_state.return_value = mock_state - mock_start.return_value = mock_new_service - - result, is_new = await conversation_service.start_conversation(request) - - # Should create a new conversation since existing one is not open - assert result.id == custom_id - assert is_new - mock_start.assert_called_once() - @pytest.mark.asyncio async def test_start_conversation_reuse_when_open(self, conversation_service): """Test that conversation is reused when event service is open.""" @@ -1849,7 +1945,7 @@ async def test_start_conversation_reuse_when_open(self, conversation_service): confirmation_policy=mock_event_service.stored.confirmation_policy, ) mock_event_service.get_state.return_value = mock_state - conversation_service._event_services[custom_id] = mock_event_service + _add_event_service(conversation_service, mock_event_service) with tempfile.TemporaryDirectory() as temp_dir: request = StartConversationRequest( @@ -1895,7 +1991,7 @@ async def test_start_conversation_returns_existing_acp_conversation( execution_status=ConversationExecutionStatus.IDLE, confirmation_policy=stored.confirmation_policy, ) - conversation_service._event_services[custom_id] = mock_event_service + _add_event_service(conversation_service, mock_event_service) with tempfile.TemporaryDirectory() as temp_dir: request = StartConversationRequest( @@ -1978,6 +2074,7 @@ async def test_start_event_service_success_stores_service( ) as mock_event_service_class: mock_event_service = AsyncMock() mock_event_service.start = AsyncMock() # Successful startup + mock_event_service.stored = stored mock_event_service_class.return_value = mock_event_service # Start event service should succeed @@ -1992,6 +2089,10 @@ async def test_start_event_service_success_stores_service( conversation_service._event_services[stored.id] == mock_event_service ) + assert ( + conversation_service._conversation_records[stored.id].stored + == stored + ) assert result == mock_event_service @@ -2016,7 +2117,7 @@ async def test_update_conversation_success( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Update the title new_title = "My Updated Conversation Title" @@ -2047,7 +2148,7 @@ async def test_update_conversation_strips_whitespace( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Update with title that has whitespace new_title = " Whitespace Test " @@ -2082,7 +2183,7 @@ async def test_update_conversation_tags_uses_state_lock( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) request = UpdateConversationRequest(tags={"env": "prod"}) result = await conversation_service.update_conversation( @@ -2112,7 +2213,7 @@ async def test_update_conversation_tags_wait_does_not_block_event_loop( mock_service.get_state.return_value = state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) lock_acquired = threading.Event() release_lock = threading.Event() @@ -2199,7 +2300,7 @@ async def test_update_conversation_notifies_webhooks( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock webhook notification with patch.object( @@ -2246,7 +2347,7 @@ async def test_update_acp_conversation_notifies_webhooks_with_acp_shape( mock_service.get_state.return_value = mock_state conversation_id = stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) with patch.object( conversation_service, "_notify_conversation_webhooks", new=AsyncMock() @@ -2278,7 +2379,7 @@ async def test_update_conversation_persists_changes( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Initial title should be None assert mock_service.stored.title is None @@ -2310,7 +2411,7 @@ async def test_update_conversation_multiple_times( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # First update request1 = UpdateConversationRequest(title="First Title") @@ -2361,7 +2462,7 @@ async def test_update_conversation_sets_updated_at( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) original_updated_at = mock_service.stored.updated_at @@ -2416,7 +2517,7 @@ async def test_delete_conversation_success(self, conversation_service): mock_service.get_state.return_value = mock_state # Add to service - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock the directory removal to avoid actual filesystem operations with patch( @@ -2462,7 +2563,7 @@ async def test_delete_conversation_notifies_webhooks_with_deleting_status( mock_service.get_state.return_value = mock_state conversation_id = sample_stored_conversation.id - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock webhook notification with patch.object( @@ -2521,7 +2622,7 @@ async def test_delete_conversation_webhook_failure(self, conversation_service): mock_service.get_state.side_effect = Exception("Webhook notification failed") # Add to service - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock the directory removal with patch( @@ -2572,7 +2673,7 @@ async def test_delete_conversation_close_failure(self, conversation_service): mock_service.close.side_effect = Exception("Close failed") # Add to service - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock the directory removal with patch( @@ -2619,7 +2720,7 @@ async def test_delete_conversation_directory_removal_failure( mock_service.get_state.return_value = mock_state # Add to service - conversation_service._event_services[conversation_id] = mock_service + _add_event_service(conversation_service, mock_service) # Mock directory removal to fail (simulating permission errors) with patch(