chore: not use request.scoped session (#37421)

Co-authored-by: WH-2099 <wh2099@pm.me>

api/core/app/apps/advanced_chat/app_runner.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -22,7 +22,7 @@ from core.app.entities.queue_entities import (
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
 from core.app.layers.conversation_variable_persist_layer import ConversationVariablePersistenceLayer
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
-from core.db.session_factory import session_factory
+from core.db.session_factory import create_session, session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
@@ -107,7 +107,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             workflow_execution_id=self.application_generate_entity.workflow_run_id,
         )
 
-        with Session(db.engine, expire_on_commit=False) as session:
+        with create_session() as session:
             app_record = session.scalar(select(App).where(App.id == app_config.app_id))
 
         if not app_record:
@@ -204,6 +204,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 trace_session_id=self.application_generate_entity.extras.get("trace_session_id"),
             )
 
+        # Release the Flask scoped session before workflow execution so a checked-out DB connection
+        # is not held for the lifetime of the graph run.
         db.session.close()
 
         # RUN WORKFLOW
@@ -368,7 +370,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         :return: List of conversation variables ready for use
         """
-        with sessionmaker(bind=db.engine).begin() as session:
+        with create_session() as session, session.begin():
             existing_variables = self._load_existing_conversation_variables(session)
 
             if not existing_variables:

api/core/app/apps/agent_chat/app_runner.py

@@ -12,10 +12,10 @@ from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.base_app_runner import AppRunner
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
+from core.db.session_factory import create_session
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
-from extensions.ext_database import db
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
 from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
@@ -47,7 +47,10 @@ class AgentChatAppRunner(AppRunner):
         app_config = application_generate_entity.app_config
         app_config = cast(AgentChatAppConfig, app_config)
         app_stmt = select(App).where(App.id == app_config.app_id)
-        app_record = db.session.scalar(app_stmt)
+        with create_session() as session:
+            app_record = session.scalar(app_stmt)
+            if app_record:
+                session.expunge(app_record)
         if not app_record:
             raise ValueError("App not found")
 
@@ -185,14 +188,18 @@ class AgentChatAppRunner(AppRunner):
         if {ModelFeature.MULTI_TOOL_CALL, ModelFeature.TOOL_CALL}.intersection(model_schema.features or []):
             agent_entity.strategy = AgentEntity.Strategy.FUNCTION_CALLING
         conversation_stmt = select(Conversation).where(Conversation.id == conversation.id)
-        conversation_result = db.session.scalar(conversation_stmt)
-        if conversation_result is None:
-            raise ValueError("Conversation not found")
         msg_stmt = select(Message).where(Message.id == message.id)
-        message_result = db.session.scalar(msg_stmt)
+        with create_session() as session:
+            conversation_result = session.scalar(conversation_stmt)
+            if conversation_result is None:
+                raise ValueError("Conversation not found")
+
+            message_result = session.scalar(msg_stmt)
+            if message_result is not None:
+                session.expunge(message_result)
+            session.expunge(conversation_result)
         if message_result is None:
             raise ValueError("Message not found")
-        db.session.close()
 
         runner_cls: type[FunctionCallAgentRunner] | type[CotChatAgentRunner] | type[CotCompletionAgentRunner]
         # start agent runner

api/core/app/apps/chat/app_runner.py

@@ -11,6 +11,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
+from core.db.session_factory import create_session
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
@@ -46,7 +47,10 @@ class ChatAppRunner(AppRunner):
         app_config = application_generate_entity.app_config
         app_config = cast(ChatAppConfig, app_config)
         stmt = select(App).where(App.id == app_config.app_id)
-        app_record = db.session.scalar(stmt)
+        with create_session() as session:
+            app_record = session.scalar(stmt)
+            if app_record:
+                session.expunge(app_record)
         if not app_record:
             raise ValueError("App not found")
 
@@ -216,6 +220,8 @@ class ChatAppRunner(AppRunner):
             model=application_generate_entity.model_conf.model,
         )
 
+        # Release the Flask scoped session before LLM streaming so a checked-out DB connection
+        # is not held for the lifetime of the provider response.
         db.session.close()
 
         invoke_result = model_instance.invoke_llm(

api/core/app/apps/completion/app_runner.py

@@ -10,6 +10,7 @@ from core.app.entities.app_invoke_entities import (
     CompletionAppGenerateEntity,
 )
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
+from core.db.session_factory import create_session
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
@@ -39,7 +40,10 @@ class CompletionAppRunner(AppRunner):
         app_config = application_generate_entity.app_config
         app_config = cast(CompletionAppConfig, app_config)
         stmt = select(App).where(App.id == app_config.app_id)
-        app_record = db.session.scalar(stmt)
+        with create_session() as session:
+            app_record = session.scalar(stmt)
+            if app_record:
+                session.expunge(app_record)
         if not app_record:
             raise ValueError("App not found")
 
@@ -174,6 +178,8 @@ class CompletionAppRunner(AppRunner):
             model=application_generate_entity.model_conf.model,
         )
 
+        # Release the Flask scoped session before LLM streaming so a checked-out DB connection
+        # is not held for the lifetime of the provider response.
         db.session.close()
 
         invoke_result = model_instance.invoke_llm(

api/core/app/apps/pipeline/pipeline_runner.py

@@ -3,6 +3,7 @@ import time
 from typing import cast
 
 from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.pipeline.pipeline_config_manager import PipelineConfig
@@ -14,12 +15,12 @@ from core.app.entities.app_invoke_entities import (
     build_dify_run_context,
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+from core.db.session_factory import create_session
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import DifyGraphInitContext, DifyNodeFactory, get_default_root_node_id
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from extensions.ext_database import db
 from graphon.enums import WorkflowType
 from graphon.graph import Graph
 from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
@@ -83,22 +84,24 @@ class PipelineRunner(WorkflowBasedAppRunner):
         user_from = self._resolve_user_from(invoke_from)
 
         user_id = None
-        if invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
-            end_user = db.session.get(EndUser, self.application_generate_entity.user_id)
-            if end_user:
-                user_id = end_user.session_id
-        else:
-            user_id = self.application_generate_entity.user_id
+        with create_session() as session:
+            if invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
+                end_user = session.get(EndUser, self.application_generate_entity.user_id)
+                if end_user:
+                    user_id = end_user.session_id
+            else:
+                user_id = self.application_generate_entity.user_id
 
-        pipeline = db.session.get(Pipeline, app_config.app_id)
-        if not pipeline:
-            raise ValueError("Pipeline not found")
+            pipeline = session.get(Pipeline, app_config.app_id)
+            if not pipeline:
+                raise ValueError("Pipeline not found")
 
-        workflow = self.get_workflow(pipeline=pipeline, workflow_id=app_config.workflow_id)
-        if not workflow:
-            raise ValueError("Workflow not initialized")
+            workflow = self.get_workflow(session=session, pipeline=pipeline, workflow_id=app_config.workflow_id)
+            if not workflow:
+                raise ValueError("Workflow not initialized")
 
-        db.session.close()
+            session.expunge(pipeline)
+            session.expunge(workflow)
 
         # if only single iteration run is requested
         if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
@@ -208,12 +211,12 @@ class PipelineRunner(WorkflowBasedAppRunner):
             )
             self._handle_event(workflow_entry, event)
 
-    def get_workflow(self, pipeline: Pipeline, workflow_id: str) -> Workflow | None:
+    def get_workflow(self, session: Session, pipeline: Pipeline, workflow_id: str) -> Workflow | None:
         """
         Get workflow
         """
         # fetch workflow by workflow_id
-        workflow = db.session.scalar(
+        workflow = session.scalar(
             select(Workflow)
             .where(Workflow.tenant_id == pipeline.tenant_id, Workflow.app_id == pipeline.id, Workflow.id == workflow_id)
             .limit(1)
@@ -298,11 +301,11 @@ class PipelineRunner(WorkflowBasedAppRunner):
         """
         if isinstance(event, GraphRunFailedEvent):
             if document_id and dataset_id:
-                document = db.session.scalar(
-                    select(Document).where(Document.id == document_id, Document.dataset_id == dataset_id).limit(1)
-                )
-                if document:
-                    document.indexing_status = "error"
-                    document.error = event.error or "Unknown error"
-                    db.session.add(document)
-                    db.session.commit()
+                with create_session() as session, session.begin():
+                    document = session.scalar(
+                        select(Document).where(Document.id == document_id, Document.dataset_id == dataset_id).limit(1)
+                    )
+                    if document:
+                        document.indexing_status = "error"
+                        document.error = event.error or "Unknown error"
+                        session.add(document)

api/core/plugin/backwards_invocation/app.py

@@ -3,7 +3,6 @@ from collections.abc import Generator, Mapping
 from typing import Any, cast
 
 from sqlalchemy import select
-from sqlalchemy.orm import Session
 
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
@@ -13,10 +12,19 @@ from core.app.apps.completion.app_generator import CompletionAppGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig
+from core.db.session_factory import create_session
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
 from extensions.ext_database import db
-from models import Account
-from models.model import App, AppMode, EndUser
+from models import Account, TenantAccountJoin
+from models.model import (
+    App,
+    AppMode,
+    AppModelConfig,
+    AppModelConfigDict,
+    EndUser,
+    load_annotation_reply_config,
+)
+from models.workflow import Workflow
 from services.end_user_service import EndUserService
 
 
@@ -30,18 +38,18 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
 
         """Retrieve app parameters."""
         if app.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app.workflow
+            workflow = cls._get_workflow(app)
             if workflow is None:
                 raise ValueError("unexpected app type")
 
             features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
-            app_model_config = app.app_model_config
-            if app_model_config is None:
+            app_model_config_dict = cls._get_app_model_config_dict(app)
+            if app_model_config_dict is None:
                 raise ValueError("unexpected app type")
 
-            features_dict = cast(dict[str, Any], app_model_config.to_dict())
+            features_dict = cast(dict[str, Any], app_model_config_dict)
 
             user_input_form = features_dict.get("user_input_form", [])
 
@@ -68,7 +76,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         if not user_id:
             user = EndUserService.get_or_create_end_user(app)
         else:
-            user = cls._get_user(user_id)
+            user = cls._get_user(user_id, app)
 
         conversation_id = conversation_id or ""
 
@@ -79,7 +87,10 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
 
                 return cls.invoke_chat_app(app, user, conversation_id, query, stream, inputs, files)
             case AppMode.WORKFLOW:
-                return cls.invoke_workflow_app(app, user, stream, inputs, files)
+                workflow = cls._get_workflow(app)
+                if not workflow:
+                    raise ValueError("unexpected app type")
+                return cls.invoke_workflow_app(app, workflow, user, stream, inputs, files)
             case AppMode.COMPLETION:
                 return cls.invoke_completion_app(app, user, stream, inputs, files)
             case _:
@@ -101,7 +112,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         """
         match app.mode:
             case AppMode.ADVANCED_CHAT:
-                workflow = app.workflow
+                workflow = cls._get_workflow(app)
                 if not workflow:
                     raise ValueError("unexpected app type")
 
@@ -158,6 +169,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
     def invoke_workflow_app(
         cls,
         app: App,
+        workflow: Workflow,
         user: EndUser | Account,
         stream: bool,
         inputs: Mapping,
@@ -166,10 +178,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         """
         invoke workflow app
         """
-        workflow = app.workflow
-        if not workflow:
-            raise ValueError("unexpected app type")
-
         pause_config = PauseStateLayerConfig(
             session_factory=db.engine,
             state_owner_user_id=workflow.created_by,
@@ -207,16 +215,26 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         )
 
     @classmethod
-    def _get_user(cls, user_id: str) -> EndUser | Account:
+    def _get_user(cls, user_id: str, app: App) -> EndUser | Account:
         """
         get the user by user id
         """
-        with Session(db.engine, expire_on_commit=False) as session:
-            stmt = select(EndUser).where(EndUser.id == user_id)
+        with create_session() as session:
+            stmt = select(EndUser).where(
+                EndUser.id == user_id,
+                EndUser.tenant_id == app.tenant_id,
+                EndUser.app_id == app.id,
+            )
             user = session.scalar(stmt)
             if not user:
-                stmt = select(Account).where(Account.id == user_id)
+                stmt = select(Account).where(
+                    Account.id == user_id,
+                    Account.id == TenantAccountJoin.account_id,
+                    TenantAccountJoin.tenant_id == app.tenant_id,
+                )
                 user = session.scalar(stmt)
+            if user:
+                session.expunge(user)
 
         if not user:
             raise ValueError("user not found")
@@ -229,7 +247,10 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         get app
         """
         try:
-            app = db.session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id).limit(1))
+            with create_session() as session:
+                app = session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id).limit(1))
+                if app:
+                    session.expunge(app)
         except Exception:
             raise ValueError("app not found")
 
@@ -237,3 +258,41 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             raise ValueError("app not found")
 
         return app
+
+    @classmethod
+    def _get_workflow(cls, app: App) -> Workflow | None:
+        """
+        get workflow without relying on App.workflow's request-scoped session property
+        """
+        if not app.workflow_id:
+            return None
+
+        with create_session() as session:
+            workflow = session.scalar(
+                select(Workflow)
+                .where(Workflow.id == app.workflow_id, Workflow.tenant_id == app.tenant_id, Workflow.app_id == app.id)
+                .limit(1)
+            )
+            if workflow:
+                session.expunge(workflow)
+            return workflow
+
+    @classmethod
+    def _get_app_model_config_dict(cls, app: App) -> AppModelConfigDict | None:
+        """
+        get app model config features without relying on request-scoped session-backed model properties
+        """
+        if not app.app_model_config_id:
+            return None
+
+        with create_session() as session:
+            app_model_config = session.scalar(
+                select(AppModelConfig)
+                .where(AppModelConfig.id == app.app_model_config_id, AppModelConfig.app_id == app.id)
+                .limit(1)
+            )
+            if app_model_config is None:
+                return None
+
+            annotation_reply = load_annotation_reply_config(session, app_model_config.app_id)
+            return app_model_config.to_dict(annotation_reply=annotation_reply)