Merge remote-tracking branch 'origin/feat/knowledgebase-summaryIndex' into feat/knowledgebase-summaryIndex

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

@@ -21,6 +21,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.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
@@ -28,7 +29,6 @@ from core.variables.variables import Variable
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool

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

@@ -9,12 +9,12 @@ from core.app.entities.app_invoke_entities import (
     InvokeFrom,
     RagPipelineGenerateEntity,
 )
+from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.app.workflow.node_factory import DifyNodeFactory
 from core.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from core.workflow.entities.graph_init_params import GraphInitParams
 from core.workflow.enums import WorkflowType
 from core.workflow.graph import Graph
-from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.graph_events import GraphEngineEvent, GraphRunFailedEvent
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository

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

@@ -7,10 +7,10 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool

api/core/app/apps/workflow_app_runner.py

@@ -157,7 +157,7 @@ class WorkflowBasedAppRunner:
         # Create initial runtime state with variable pool containing environment variables
         graph_runtime_state = GraphRuntimeState(
             variable_pool=VariablePool(
-                system_variables=SystemVariable.empty(),
+                system_variables=SystemVariable.default(),
                 user_inputs={},
                 environment_variables=workflow.environment_variables,
             ),
@@ -272,7 +272,9 @@ class WorkflowBasedAppRunner:
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
+        graph = Graph.init(
+            graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, skip_validation=True
+        )
 
         if not graph:
             raise ValueError("graph not found in workflow")

api/core/app/workflow/layers/__init__.py

@@ -0,0 +1,10 @@
+"""Workflow-level GraphEngine layers that depend on outer infrastructure."""
+
+from .observability import ObservabilityLayer
+from .persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+
+__all__ = [
+    "ObservabilityLayer",
+    "PersistenceWorkflowInfo",
+    "WorkflowPersistenceLayer",
+]

api/core/app/workflow/layers/persistence.py

@@ -45,7 +45,6 @@ from core.workflow.graph_events import (
 from core.workflow.node_events import NodeRunResult
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_entry import WorkflowEntry
 from libs.datetime_utils import naive_utc_now
 
 
@@ -316,6 +315,9 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
                 # workflow inputs stay reusable without binding future runs to this conversation.
                 continue
             inputs[f"sys.{field_name}"] = value
+        # Local import to avoid circular dependency during app bootstrapping.
+        from core.workflow.workflow_entry import WorkflowEntry
+
         handled = WorkflowEntry.handle_special_values(inputs)
         return handled or {}
 

api/core/workflow/graph/graph.py

@@ -288,6 +288,7 @@ class Graph:
         graph_config: Mapping[str, object],
         node_factory: NodeFactory,
         root_node_id: str | None = None,
+        skip_validation: bool = False,
     ) -> Graph:
         """
         Initialize graph
@@ -339,8 +340,9 @@ class Graph:
             root_node=root_node,
         )
 
-        # Validate the graph structure using built-in validators
-        get_graph_validator().validate(graph)
+        if not skip_validation:
+            # Validate the graph structure using built-in validators
+            get_graph_validator().validate(graph)
 
         return graph
 

api/core/workflow/graph_engine/layers/__init__.py

@@ -8,11 +8,9 @@ with middleware-like components that can observe events and interact with execut
 from .base import GraphEngineLayer
 from .debug_logging import DebugLoggingLayer
 from .execution_limits import ExecutionLimitsLayer
-from .observability import ObservabilityLayer
 
 __all__ = [
     "DebugLoggingLayer",
     "ExecutionLimitsLayer",
     "GraphEngineLayer",
-    "ObservabilityLayer",
 ]

api/core/workflow/runtime/variable_pool.py

@@ -44,7 +44,7 @@ class VariablePool(BaseModel):
     )
     system_variables: SystemVariable = Field(
         description="System variables",
-        default_factory=SystemVariable.empty,
+        default_factory=SystemVariable.default,
     )
     environment_variables: Sequence[Variable] = Field(
         description="Environment variables.",
@@ -271,4 +271,4 @@ class VariablePool(BaseModel):
     @classmethod
     def empty(cls) -> VariablePool:
         """Create an empty variable pool."""
-        return cls(system_variables=SystemVariable.empty())
+        return cls(system_variables=SystemVariable.default())

api/core/workflow/system_variable.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping, Sequence
 from types import MappingProxyType
 from typing import Any
+from uuid import uuid4
 
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
 
@@ -72,8 +73,8 @@ class SystemVariable(BaseModel):
         return data
 
     @classmethod
-    def empty(cls) -> SystemVariable:
-        return cls()
+    def default(cls) -> SystemVariable:
+        return cls(workflow_execution_id=str(uuid4()))
 
     def to_dict(self) -> dict[SystemVariableKey, Any]:
         # NOTE: This method is provided for compatibility with legacy code.

api/core/workflow/workflow_entry.py

@@ -7,6 +7,7 @@ from typing import Any
 from configs import dify_config
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.workflow.layers.observability import ObservabilityLayer
 from core.app.workflow.node_factory import DifyNodeFactory
 from core.file.models import File
 from core.workflow.constants import ENVIRONMENT_VARIABLE_NODE_ID
@@ -15,7 +16,7 @@ from core.workflow.errors import WorkflowNodeRunFailedError
 from core.workflow.graph import Graph
 from core.workflow.graph_engine import GraphEngine
 from core.workflow.graph_engine.command_channels import InMemoryChannel
-from core.workflow.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer, ObservabilityLayer
+from core.workflow.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer
 from core.workflow.graph_engine.protocols.command_channel import CommandChannel
 from core.workflow.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
 from core.workflow.nodes import NodeType
@@ -276,7 +277,7 @@ class WorkflowEntry:
 
         # init variable pool
         variable_pool = VariablePool(
-            system_variables=SystemVariable.empty(),
+            system_variables=SystemVariable.default(),
             user_inputs={},
             environment_variables=[],
         )