fix(api): enforce ownership on conversation delete

api/services/conversation_service.py

@@ -180,6 +180,14 @@ class ConversationService:
 
     @classmethod
     def delete(cls, app_model: App, conversation_id: str, user: Union[Account, EndUser] | None):
+        """
+        Delete a conversation only if it belongs to the given user and app context.
+
+        Raises:
+            ConversationNotExistsError: When the conversation is not visible to the current user.
+        """
+        conversation = cls.get_conversation(app_model, conversation_id, user)
+
         try:
             logger.info(
                 "Initiating conversation deletion for app_name %s, conversation_id: %s",
@@ -187,10 +195,10 @@ class ConversationService:
                 conversation_id,
             )
 
-            db.session.query(Conversation).where(Conversation.id == conversation_id).delete(synchronize_session=False)
+            db.session.delete(conversation)
             db.session.commit()
 
-            delete_conversation_related_data.delay(conversation_id)
+            delete_conversation_related_data.delay(conversation.id)
 
         except Exception as e:
             db.session.rollback()

api/tests/test_containers_integration_tests/services/test_conversation_service.py

@@ -1034,3 +1034,34 @@ class TestConversationServiceExport:
         # Step 2: Async cleanup task triggered
         # The Celery task will handle cleanup of messages, annotations, etc.
         mock_delete_task.delay.assert_called_once_with(conversation_id)
+
+    @patch("services.conversation_service.delete_conversation_related_data")
+    def test_delete_conversation_not_owned_by_account(self, mock_delete_task, db_session_with_containers):
+        """
+        Test deletion is denied when conversation belongs to a different account.
+        """
+        # Arrange
+        app_model, owner_account = ConversationServiceIntegrationTestDataFactory.create_app_and_account(
+            db_session_with_containers
+        )
+        _, other_account = ConversationServiceIntegrationTestDataFactory.create_app_and_account(
+            db_session_with_containers
+        )
+        conversation = ConversationServiceIntegrationTestDataFactory.create_conversation(
+            db_session_with_containers,
+            app_model,
+            owner_account,
+        )
+
+        # Act & Assert
+        with pytest.raises(ConversationNotExistsError):
+            ConversationService.delete(
+                app_model=app_model,
+                conversation_id=conversation.id,
+                user=other_account,
+            )
+
+        # Verify no deletion and no async cleanup trigger
+        not_deleted = db_session_with_containers.scalar(select(Conversation).where(Conversation.id == conversation.id))
+        assert not_deleted is not None
+        mock_delete_task.delay.assert_not_called()