refactor: explicit DB session propagation across backend paths (#38559)

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

api/core/workflow/nodes/agent_v2/validators.py

@@ -147,7 +147,7 @@ class WorkflowAgentNodeValidator:
             )
         cls._validate_agent_soul_env(binding=binding, agent_soul=agent_soul)
         cls._validate_agent_soul_tools(binding=binding, agent_soul=agent_soul)
-        cls._validate_agent_soul_knowledge(binding=binding, agent_soul=agent_soul)
+        cls._validate_agent_soul_knowledge(session=session, binding=binding, agent_soul=agent_soul)
         node_job = WorkflowNodeJobConfig.model_validate(binding.node_job_config_dict)
         cls.validate_node_job(session=session, binding=binding, node_job=node_job, topology=topology)
 
@@ -370,11 +370,13 @@ class WorkflowAgentNodeValidator:
     def _validate_agent_soul_knowledge(
         cls,
         *,
+        session: Session,
         binding: WorkflowAgentNodeBinding,
         agent_soul: AgentSoulConfig,
     ) -> None:
         """Validate knowledge set dataset rows against the publishing tenant."""
         missing_ids = list_missing_tenant_knowledge_dataset_ids(
+            session=session,
             tenant_id=binding.tenant_id,
             agent_soul=agent_soul,
         )

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -2,6 +2,9 @@ import logging
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, override
 
+from sqlalchemy.orm import Session
+
+from core.db.session_factory import session_factory
 from core.rag.index_processor.index_processor import IndexProcessor
 from core.rag.index_processor.index_processor_base import SummaryIndexSettingDict
 from core.rag.summary_index.summary_index import SummaryIndex
@@ -83,9 +86,15 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
                 # Get indexing_technique and summary_index_setting from node_data (workflow graph config)
                 # or fallback to dataset if not available in node_data
 
-                outputs = self.index_processor.get_preview_output(
-                    chunks, dataset_id, document_id, node_data.chunk_structure, summary_index_setting
-                )
+                with session_factory.create_session() as session:
+                    outputs = self.index_processor.get_preview_output(
+                        chunks,
+                        dataset_id,
+                        document_id,
+                        node_data.chunk_structure,
+                        summary_index_setting,
+                        session=session,
+                    )
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=variables,
@@ -97,15 +106,17 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
             if not batch:
                 raise KnowledgeIndexNodeError("Batch is required.")
 
-            results = self._invoke_knowledge_index(
-                dataset_id=dataset_id,
-                document_id=document_id,
-                original_document_id=original_document_id_segment.value if original_document_id_segment else "",
-                is_preview=is_preview,
-                batch=batch.value,
-                chunks=chunks,
-                summary_index_setting=summary_index_setting,
-            )
+            with session_factory.create_session() as session:
+                results = self._invoke_knowledge_index(
+                    session=session,
+                    dataset_id=dataset_id,
+                    document_id=document_id,
+                    original_document_id=original_document_id_segment.value if original_document_id_segment else "",
+                    is_preview=is_preview,
+                    batch=batch.value,
+                    chunks=chunks,
+                    summary_index_setting=summary_index_setting,
+                )
             return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=variables, outputs=results)
 
         except KnowledgeIndexNodeError as e:
@@ -134,12 +145,16 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
         batch: Any,
         chunks: Mapping[str, Any],
         summary_index_setting: SummaryIndexSettingDict | None = None,
+        *,
+        session: Session,
     ):
         if not document_id:
             raise KnowledgeIndexNodeError("document_id is required.")
         rst = self.index_processor.index_and_clean(
-            dataset_id, document_id, original_document_id, chunks, batch, summary_index_setting
+            dataset_id, document_id, original_document_id, chunks, batch, summary_index_setting, session=session
         )
+        # Summary generation opens independent sessions and must see the indexed rows.
+        session.commit()
         self.summary_index_service.generate_and_vectorize_summary(
             dataset_id, document_id, is_preview, summary_index_setting
         )

api/core/workflow/nodes/knowledge_index/protocols.py

@@ -2,6 +2,7 @@ from collections.abc import Mapping
 from typing import Any, Protocol, TypedDict
 
 from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
 
 
 class IndexingResultDict(TypedDict):
@@ -44,6 +45,8 @@ class IndexProcessorProtocol(Protocol):
         chunks: Mapping[str, Any],
         batch: Any,
         summary_index_setting: dict[str, Any] | None = None,
+        *,
+        session: Session,
     ) -> IndexingResultDict: ...
 
     def get_preview_output(
@@ -53,6 +56,8 @@ class IndexProcessorProtocol(Protocol):
         document_id: str,
         chunk_structure: str,
         summary_index_setting: dict[str, Any] | None,
+        *,
+        session: Session,
     ) -> Preview: ...