avoid app context dependency in workflow socket auth

api/services/workflow_collaboration_service.py

@@ -6,7 +6,7 @@ from collections.abc import Mapping
 
 from sqlalchemy import select
 
-from extensions.ext_database import db
+from core.db.session_factory import session_factory
 from models.account import Account
 from models.model import App
 from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository, WorkflowSessionInfo
@@ -78,10 +78,9 @@ class WorkflowCollaborationService:
         return str(user_id), is_leader
 
     def _can_access_workflow(self, workflow_id: str, tenant_id: str) -> bool:
-        """Check that the collaboration room belongs to an active app in the caller's current tenant."""
-        app_id = db.session.scalar(
-            select(App.id).where(App.id == workflow_id, App.tenant_id == tenant_id).limit(1)
-        )
+        """Check room access without relying on Flask's app-context-bound scoped session."""
+        with session_factory.create_session() as session:
+            app_id = session.scalar(select(App.id).where(App.id == workflow_id, App.tenant_id == tenant_id).limit(1))
         return app_id is not None
 
     def disconnect_session(self, sid: str) -> None:

api/tests/unit_tests/services/test_workflow_collaboration_service.py

@@ -101,6 +101,25 @@ class TestWorkflowCollaborationService:
             {"user_id": "u-1", "username": "Jane", "avatar": "avatar.png", "tenant_id": "t-1"},
         )
 
+    def test_can_access_workflow_uses_session_factory(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        collaboration_service, _repository, _socketio = service
+        session = Mock()
+        session.scalar.return_value = "wf-1"
+        session_context = Mock()
+        session_context.__enter__ = Mock(return_value=session)
+        session_context.__exit__ = Mock(return_value=False)
+
+        with patch(
+            "services.workflow_collaboration_service.session_factory.create_session",
+            return_value=session_context,
+        ):
+            result = collaboration_service._can_access_workflow("wf-1", "tenant-1")
+
+        assert result is True
+        session.scalar.assert_called_once()
+
     def test_relay_collaboration_event_unauthorized(
         self, service: tuple[WorkflowCollaborationService, Mock, Mock]
     ) -> None: