test(workflow): pass explicit tenant context

Add regression coverage that workflow persistence Celery tasks route to workflow_storage and that both Docker and local worker startup defaults consume the queue. Document the Docker worker override for running a dedicated workflow storage worker.

api/.env.example

@@ -574,12 +574,6 @@ GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
 # Seconds of idle time before scaling down workers (default: 5.0)
 GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0
 
-# Workflow storage configuration
-# Options: rdbms, hybrid
-# rdbms: Use only the relational database (default)
-# hybrid: Save new data to object storage, read from both object storage and RDBMS
-WORKFLOW_NODE_EXECUTION_STORAGE=rdbms
-
 # Repository configuration
 # Core workflow execution repository implementation
 CORE_WORKFLOW_EXECUTION_REPOSITORY=core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository

api/configs/feature/__init__.py

@@ -941,11 +941,6 @@ class WorkflowNodeExecutionConfig(BaseSettings):
         default=100,
     )
 
-    WORKFLOW_NODE_EXECUTION_STORAGE: str = Field(
-        default="rdbms",
-        description="Storage backend for WorkflowNodeExecution. Options: 'rdbms', 'hybrid'",
-    )
-
 
 class RepositoryConfig(BaseSettings):
     """
@@ -953,18 +948,12 @@ class RepositoryConfig(BaseSettings):
     """
 
     CORE_WORKFLOW_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowExecution. Options: "
-        "'core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository' (default), "
-        "'core.repositories.celery_workflow_execution_repository.CeleryWorkflowExecutionRepository'",
+        description="Repository implementation for WorkflowExecution. Specify as a module path.",
         default="core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository",
     )
 
     CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowNodeExecution. Options: "
-        "'core.repositories.sqlalchemy_workflow_node_execution_repository."
-        "SQLAlchemyWorkflowNodeExecutionRepository' (default), "
-        "'core.repositories.celery_workflow_node_execution_repository."
-        "CeleryWorkflowNodeExecutionRepository'",
+        description="Repository implementation for WorkflowNodeExecution. Specify as a module path.",
         default="core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository",
     )
 

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

@@ -263,6 +263,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,
+            invoke_from=invoke_from,
             trace_manager=self.application_generate_entity.trace_manager,
         )
 

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

@@ -228,6 +228,7 @@ class PipelineRunner(WorkflowBasedAppRunner):
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,
+            invoke_from=invoke_from,
             trace_manager=self.application_generate_entity.trace_manager,
         )
 

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

@@ -193,6 +193,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,
+            invoke_from=invoke_from,
             trace_manager=self.application_generate_entity.trace_manager,
         )
 

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

@@ -4,17 +4,19 @@ This layer mirrors the former ``WorkflowCycleManager`` responsibilities by
 listening to ``GraphEngineEvent`` instances directly and persisting workflow
 and node execution state via the injected repositories.
 
-The design keeps domain persistence concerns inside the engine thread, while
-allowing presentation layers to remain read-only observers of repository
-state.
+The layer owns domain-to-persistence event handling, while the injected
+repositories choose the write strategy. Debug executions use synchronous
+writes so developer tools can read DB state immediately; non-debug app
+executions use a Celery-backed write path to keep DB writes out of the engine
+thread.
 """
 
 from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Union, override
+from typing import Any, Protocol, override, runtime_checkable
 
-from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity
 from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
 from core.helper.trace_id_helper import ParentTraceContext
 from core.ops.entities.trace_entity import TraceTaskName
@@ -66,6 +68,16 @@ class PersistenceWorkflowInfo:
     graph_data: Mapping[str, Any]
 
 
+@runtime_checkable
+class _AsyncPersistenceConfigurable(Protocol):
+    def set_async_persistence(self, enabled: bool) -> None: ...
+
+
+def _configure_async_persistence(repository: object, enabled: bool) -> None:
+    if isinstance(repository, _AsyncPersistenceConfigurable):
+        repository.set_async_persistence(enabled)
+
+
 @dataclass(slots=True)
 class _NodeRuntimeSnapshot:
     """Lightweight cache to keep node metadata across event phases."""
@@ -84,10 +96,11 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     def __init__(
         self,
         *,
-        application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
+        application_generate_entity: AdvancedChatAppGenerateEntity | WorkflowAppGenerateEntity,
         workflow_info: PersistenceWorkflowInfo,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        invoke_from: InvokeFrom | None = None,
         trace_manager: TraceQueueManager | None = None,
     ) -> None:
         super().__init__()
@@ -95,6 +108,10 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         self._workflow_info = workflow_info
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
+        effective_invoke_from = invoke_from if invoke_from is not None else application_generate_entity.invoke_from
+        use_async_persistence = effective_invoke_from != InvokeFrom.DEBUGGER
+        _configure_async_persistence(self._workflow_execution_repository, use_async_persistence)
+        _configure_async_persistence(self._workflow_node_execution_repository, use_async_persistence)
         self._trace_manager = trace_manager
 
         self._workflow_execution: WorkflowExecution | None = None

api/core/repositories/__init__.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-from .celery_workflow_execution_repository import CeleryWorkflowExecutionRepository
-from .celery_workflow_node_execution_repository import CeleryWorkflowNodeExecutionRepository
 from .factory import (
     DifyCoreRepositoryFactory,
     OrderConfig,
@@ -15,8 +13,6 @@ from .sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutio
 from .sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
 
 __all__ = [
-    "CeleryWorkflowExecutionRepository",
-    "CeleryWorkflowNodeExecutionRepository",
     "DifyCoreRepositoryFactory",
     "OrderConfig",
     "RepositoryImportError",

api/core/repositories/celery_workflow_execution_repository.py

@@ -1,127 +0,0 @@
-"""
-Celery-based implementation of the WorkflowExecutionRepository.
-
-This implementation uses Celery tasks for asynchronous storage operations,
-providing improved performance by offloading database operations to background workers.
-"""
-
-import logging
-from typing import override
-
-from sqlalchemy.engine import Engine
-from sqlalchemy.orm import sessionmaker
-
-from core.repositories.factory import WorkflowExecutionRepository
-from graphon.entities import WorkflowExecution
-from models import Account, CreatorUserRole, EndUser
-from models.enums import WorkflowRunTriggeredFrom
-from tasks.workflow_execution_tasks import (
-    save_workflow_execution_task,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class CeleryWorkflowExecutionRepository(WorkflowExecutionRepository):
-    """
-    Celery-based implementation of the WorkflowExecutionRepository interface.
-
-    This implementation provides asynchronous storage capabilities by using Celery tasks
-    to handle database operations in background workers. This improves performance by
-    reducing the blocking time for workflow execution storage operations.
-
-    Key features:
-    - Asynchronous save operations using Celery tasks
-    - Support for multi-tenancy through tenant/app filtering
-    - Automatic retry and error handling through Celery
-    """
-
-    _session_factory: sessionmaker
-    _tenant_id: str
-    _app_id: str | None
-    _triggered_from: WorkflowRunTriggeredFrom | None
-    _creator_user_id: str
-    _creator_user_role: CreatorUserRole
-
-    def __init__(
-        self,
-        session_factory: sessionmaker | Engine,
-        tenant_id: str,
-        user: Account | EndUser,
-        app_id: str | None,
-        triggered_from: WorkflowRunTriggeredFrom | None,
-    ):
-        """
-        Initialize the repository with Celery task configuration and context information.
-
-        Args:
-            session_factory: SQLAlchemy sessionmaker or engine for fallback operations
-            tenant_id: Tenant that owns the workflow execution
-            user: Account or EndUser used for creator attribution
-            app_id: App ID for filtering by application (can be None)
-            triggered_from: Source of the execution trigger (DEBUGGING or APP_RUN)
-        """
-        # Store session factory for fallback operations
-        match session_factory:
-            case Engine():
-                self._session_factory = sessionmaker(bind=session_factory, expire_on_commit=False)
-            case sessionmaker():
-                self._session_factory = session_factory
-            case _:
-                raise ValueError(
-                    f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
-                )
-
-        if not tenant_id:
-            raise ValueError("tenant_id is required")
-        self._tenant_id = tenant_id
-
-        # Store app context
-        self._app_id = app_id
-
-        # Extract user context
-        self._triggered_from = triggered_from
-        self._creator_user_id = user.id
-
-        # Determine user role based on user type
-        self._creator_user_role = CreatorUserRole.ACCOUNT if isinstance(user, Account) else CreatorUserRole.END_USER
-
-        logger.info(
-            "Initialized CeleryWorkflowExecutionRepository for tenant %s, app %s, triggered_from %s",
-            self._tenant_id,
-            self._app_id,
-            self._triggered_from,
-        )
-
-    @override
-    def save(self, execution: WorkflowExecution):
-        """
-        Save or update a WorkflowExecution instance asynchronously using Celery.
-
-        This method queues the save operation as a Celery task and returns immediately,
-        providing improved performance for high-throughput scenarios.
-
-        Args:
-            execution: The WorkflowExecution instance to save or update
-        """
-        try:
-            # Serialize execution for Celery task
-            execution_data = execution.model_dump()
-
-            # Queue the save operation as a Celery task (fire and forget)
-            save_workflow_execution_task.delay(  # type: ignore
-                execution_data=execution_data,
-                tenant_id=self._tenant_id,
-                app_id=self._app_id or "",
-                triggered_from=self._triggered_from.value if self._triggered_from else "",
-                creator_user_id=self._creator_user_id,
-                creator_user_role=self._creator_user_role.value,
-            )
-
-            logger.debug("Queued async save for workflow execution: %s", execution.id_)
-
-        except Exception:
-            logger.exception("Failed to queue save operation for execution %s", execution.id_)
-            # In case of Celery failure, we could implement a fallback to synchronous save
-            # For now, we'll re-raise the exception
-            raise

api/core/repositories/celery_workflow_node_execution_repository.py

@@ -1,199 +0,0 @@
-"""
-Celery-based implementation of the WorkflowNodeExecutionRepository.
-
-This implementation uses Celery tasks for asynchronous storage operations,
-providing improved performance by offloading database operations to background workers.
-"""
-
-import logging
-from collections.abc import Sequence
-from typing import override
-
-from sqlalchemy.engine import Engine
-from sqlalchemy.orm import sessionmaker
-
-from core.repositories.factory import (
-    OrderConfig,
-    WorkflowNodeExecutionRepository,
-)
-from graphon.entities import WorkflowNodeExecution
-from models import Account, CreatorUserRole, EndUser
-from models.workflow import WorkflowNodeExecutionTriggeredFrom
-from tasks.workflow_node_execution_tasks import (
-    save_workflow_node_execution_task,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
-    """
-    Celery-based implementation of the WorkflowNodeExecutionRepository interface.
-
-    This implementation provides asynchronous storage capabilities by using Celery tasks
-    to handle database operations in background workers. This improves performance by
-    reducing the blocking time for workflow node execution storage operations.
-
-    Key features:
-    - Asynchronous save operations using Celery tasks
-    - In-memory cache for immediate reads
-    - Support for multi-tenancy through tenant/app filtering
-    - Automatic retry and error handling through Celery
-    """
-
-    _session_factory: sessionmaker
-    _tenant_id: str
-    _app_id: str | None
-    _triggered_from: WorkflowNodeExecutionTriggeredFrom | None
-    _creator_user_id: str
-    _creator_user_role: CreatorUserRole
-    _execution_cache: dict[str, WorkflowNodeExecution]
-    _workflow_execution_mapping: dict[str, list[str]]
-
-    def __init__(
-        self,
-        session_factory: sessionmaker | Engine,
-        tenant_id: str,
-        user: Account | EndUser,
-        app_id: str | None,
-        triggered_from: WorkflowNodeExecutionTriggeredFrom | None,
-    ):
-        """
-        Initialize the repository with Celery task configuration and context information.
-
-        Args:
-            session_factory: SQLAlchemy sessionmaker or engine for fallback operations
-            tenant_id: Tenant that owns the workflow node execution
-            user: Account or EndUser used for creator attribution
-            app_id: App ID for filtering by application (can be None)
-            triggered_from: Source of the execution trigger (SINGLE_STEP or WORKFLOW_RUN)
-        """
-        # Store session factory for fallback operations
-        match session_factory:
-            case Engine():
-                self._session_factory = sessionmaker(bind=session_factory, expire_on_commit=False)
-            case sessionmaker():
-                self._session_factory = session_factory
-            case _:
-                raise ValueError(
-                    f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
-                )
-
-        if not tenant_id:
-            raise ValueError("tenant_id is required")
-        self._tenant_id = tenant_id
-
-        # Store app context
-        self._app_id = app_id
-
-        # Extract user context
-        self._triggered_from = triggered_from
-        self._creator_user_id = user.id
-
-        # Determine user role based on user type
-        self._creator_user_role = CreatorUserRole.ACCOUNT if isinstance(user, Account) else CreatorUserRole.END_USER
-
-        # In-memory cache for workflow node executions
-        self._execution_cache = {}
-
-        # Cache for mapping workflow_execution_ids to execution IDs for efficient retrieval
-        self._workflow_execution_mapping = {}
-
-        logger.info(
-            "Initialized CeleryWorkflowNodeExecutionRepository for tenant %s, app %s, triggered_from %s",
-            self._tenant_id,
-            self._app_id,
-            self._triggered_from,
-        )
-
-    @override
-    def save(self, execution: WorkflowNodeExecution):
-        """
-        Save or update a WorkflowNodeExecution instance to cache and asynchronously to database.
-
-        This method stores the execution in cache immediately for fast reads and queues
-        the save operation as a Celery task without tracking the task status.
-
-        Args:
-            execution: The WorkflowNodeExecution instance to save or update
-        """
-        try:
-            # Store in cache immediately for fast reads
-            self._execution_cache[execution.id] = execution
-
-            # Update workflow execution mapping for efficient retrieval
-            if execution.workflow_execution_id:
-                if execution.workflow_execution_id not in self._workflow_execution_mapping:
-                    self._workflow_execution_mapping[execution.workflow_execution_id] = []
-                if execution.id not in self._workflow_execution_mapping[execution.workflow_execution_id]:
-                    self._workflow_execution_mapping[execution.workflow_execution_id].append(execution.id)
-
-            # Serialize execution for Celery task
-            execution_data = execution.model_dump()
-
-            # Queue the save operation as a Celery task (fire and forget)
-            save_workflow_node_execution_task.delay(
-                execution_data=execution_data,
-                tenant_id=self._tenant_id,
-                app_id=self._app_id or "",
-                triggered_from=self._triggered_from.value if self._triggered_from else "",
-                creator_user_id=self._creator_user_id,
-                creator_user_role=self._creator_user_role.value,
-            )
-
-            logger.debug("Cached and queued async save for workflow node execution: %s", execution.id)
-
-        except Exception:
-            logger.exception("Failed to cache or queue save operation for node execution %s", execution.id)
-            # In case of Celery failure, we could implement a fallback to synchronous save
-            # For now, we'll re-raise the exception
-            raise
-
-    @override
-    def get_by_workflow_execution(
-        self,
-        workflow_execution_id: str,
-        order_config: OrderConfig | None = None,
-    ) -> Sequence[WorkflowNodeExecution]:
-        """
-        Retrieve all workflow node executions for a workflow execution from cache.
-
-        Args:
-            workflow_execution_id: The workflow execution identifier
-            order_config: Optional configuration for ordering results
-
-        Returns:
-            A sequence of WorkflowNodeExecution instances
-        """
-        try:
-            # Get execution IDs for this workflow execution from cache
-            execution_ids = self._workflow_execution_mapping.get(workflow_execution_id, [])
-
-            # Retrieve executions from cache
-            result = []
-            for execution_id in execution_ids:
-                if execution_id in self._execution_cache:
-                    result.append(self._execution_cache[execution_id])
-
-            # Apply ordering if specified
-            if order_config and result:
-                # Sort based on the order configuration
-                reverse = order_config.order_direction == "desc"
-
-                # Sort by multiple fields if specified
-                for field_name in reversed(order_config.order_by):
-                    result.sort(key=lambda x: getattr(x, field_name, 0), reverse=reverse)
-
-            logger.debug(
-                "Retrieved %d workflow node executions for execution %s from cache",
-                len(result),
-                workflow_execution_id,
-            )
-            return result
-
-        except Exception:
-            logger.exception(
-                "Failed to get workflow node executions for execution %s from cache",
-                workflow_execution_id,
-            )
-            return []

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -20,6 +20,7 @@ from models import (
     WorkflowRun,
 )
 from models.enums import WorkflowRunTriggeredFrom
+from tasks.workflow_execution_tasks import save_workflow_execution_task
 
 logger = logging.getLogger(__name__)
 
@@ -36,6 +37,8 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
     performance by reducing database queries.
     """
 
+    _use_async_persistence: bool
+
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
@@ -82,6 +85,16 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
         # Initialize in-memory cache for workflow executions
         # Key: execution_id, Value: WorkflowRun (DB model)
         self._execution_cache: dict[str, WorkflowRun] = {}
+        self._use_async_persistence = False
+
+    def set_async_persistence(self, enabled: bool) -> None:
+        """
+        Configure whether save operations should be queued through Celery.
+
+        Debug executions keep this disabled so the debugger can immediately read persisted
+        workflow state. Non-debug app executions enable it from the persistence layer.
+        """
+        self._use_async_persistence = enabled
 
     def _to_domain_model(self, db_model: WorkflowRun) -> WorkflowExecution:
         """
@@ -192,6 +205,10 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
         Args:
             execution: The WorkflowExecution domain entity to persist
         """
+        if self._use_async_persistence:
+            self._queue_async_save(execution)
+            return
+
         # Convert domain model to database model using tenant context and other attributes
         db_model = self._to_db_model(execution)
 
@@ -211,3 +228,20 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
 
             # Update the in-memory cache for faster subsequent lookups
             self._execution_cache[db_model.id] = db_model
+
+    def _queue_async_save(self, execution: WorkflowExecution) -> None:
+        if not self._triggered_from:
+            raise ValueError("triggered_from is required in repository constructor")
+        if not self._creator_user_id:
+            raise ValueError("created_by is required in repository constructor")
+        if not self._creator_user_role:
+            raise ValueError("created_by_role is required in repository constructor")
+
+        save_workflow_execution_task.delay(
+            execution_data=execution.model_dump(),
+            tenant_id=self._tenant_id,
+            app_id=self._app_id or "",
+            triggered_from=self._triggered_from.value,
+            creator_user_id=self._creator_user_id,
+            creator_user_role=self._creator_user_role.value,
+        )

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -36,6 +36,10 @@ from models.model import UploadFile
 from models.workflow import WorkflowNodeExecutionOffload
 from services.file_service import FileService
 from services.variable_truncator import VariableTruncator
+from tasks.workflow_node_execution_tasks import (
+    save_workflow_node_execution_data_task,
+    save_workflow_node_execution_task,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -43,7 +47,6 @@ logger = logging.getLogger(__name__)
 @dataclasses.dataclass(frozen=True)
 class _InputsOutputsTruncationResult:
     truncated_value: Mapping[str, Any]
-    file: UploadFile
     offload: WorkflowNodeExecutionOffload
 
 
@@ -59,6 +62,8 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
     performance by reducing database queries.
     """
 
+    _use_async_persistence: bool
+
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
@@ -108,6 +113,16 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
 
         # Initialize FileService for handling offloaded data
         self._file_service = FileService(session_factory)
+        self._use_async_persistence = False
+
+    def set_async_persistence(self, enabled: bool) -> None:
+        """
+        Configure whether save operations should be queued through Celery.
+
+        Debug executions keep this disabled so node data is readable immediately. Non-debug
+        app executions enable it from the workflow persistence layer.
+        """
+        self._use_async_persistence = enabled
 
     def _create_truncator(self) -> VariableTruncator:
         return VariableTruncator(
@@ -310,7 +325,6 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
         )
         return _InputsOutputsTruncationResult(
             truncated_value=truncated_values,
-            file=upload_file,
             offload=offload,
         )
 
@@ -338,6 +352,10 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
         # Only the final call contains the complete inputs and outputs payloads, so earlier invocations
         # must tolerate missing data without attempting to offload variables.
 
+        if self._use_async_persistence:
+            self._queue_async_save(execution)
+            return
+
         # Convert domain model to database model using tenant context and other attributes
         db_model = self._to_db_model(execution)
 
@@ -372,7 +390,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             logger.exception("Failed to save workflow node execution after all retries")
             raise
 
-    def _persist_to_database(self, db_model: WorkflowNodeExecutionModel):
+    def _persist_to_database(self, db_model: WorkflowNodeExecutionModel) -> None:
         """
         Persist the database model to the database.
 
@@ -386,6 +404,8 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             existing = session.get(WorkflowNodeExecutionModel, db_model.id)
 
             if existing:
+                if existing.tenant_id != self._tenant_id:
+                    raise ValueError("Unauthorized access to workflow node execution")
                 # Update existing record by copying all non-private attributes
                 for key, value in db_model.__dict__.items():
                     if not key.startswith("_"):
@@ -396,25 +416,34 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
 
             session.commit()
 
-            # Update the in-memory cache for faster subsequent lookups
-            # Only cache if we have a node_execution_id to use as the cache key
-            if db_model.node_execution_id:
-                self._node_execution_cache[db_model.node_execution_id] = db_model
-
     @override
-    def save_execution_data(self, execution: WorkflowNodeExecution):
-        domain_model = execution
-        with self._session_factory(expire_on_commit=False) as session:
-            query = WorkflowNodeExecutionModel.preload_offload_data(select(WorkflowNodeExecutionModel)).where(
-                WorkflowNodeExecutionModel.id == domain_model.id
-            )
-            db_model: WorkflowNodeExecutionModel | None = session.execute(query).scalars().first()
+    def save_execution_data(self, execution: WorkflowNodeExecution) -> None:
+        """Persist node payload fields without letting an older task overwrite newer state.
 
-        if db_model is not None:
-            offload_data = db_model.offload_data
-        else:
-            db_model = self._to_db_model(domain_model)
-            offload_data = db_model.offload_data
+        Storage uploads happen before the short row-locked transaction. The transaction updates
+        only payload/offload fields, plus terminal metadata needed to make task retries monotonic.
+        """
+        if self._use_async_persistence:
+            self._queue_async_save_execution_data(execution)
+            return
+
+        domain_model = execution
+        with self._session_factory() as session:
+            query = select(WorkflowNodeExecutionModel).where(
+                WorkflowNodeExecutionModel.id == domain_model.id,
+                WorkflowNodeExecutionModel.tenant_id == self._tenant_id,
+            )
+            existing_model = session.scalar(query)
+            if existing_model is not None and existing_model.finished_at is not None:
+                if domain_model.finished_at is None or domain_model.finished_at < existing_model.finished_at:
+                    return
+
+        inputs: str | None = None
+        outputs: str | None = None
+        process_data: str | None = None
+        inputs_offload: WorkflowNodeExecutionOffload | None = None
+        outputs_offload: WorkflowNodeExecutionOffload | None = None
+        process_data_offload: WorkflowNodeExecutionOffload | None = None
 
         if domain_model.inputs is not None:
             result = self._truncate_and_upload(
@@ -423,11 +452,11 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                 ExecutionOffLoadType.INPUTS,
             )
             if result is not None:
-                db_model.inputs = self._json_encode(result.truncated_value)
+                inputs = self._json_encode(result.truncated_value)
                 domain_model.set_truncated_inputs(result.truncated_value)
-                offload_data = _replace_or_append_offload(offload_data, result.offload)
+                inputs_offload = result.offload
             else:
-                db_model.inputs = self._json_encode(domain_model.inputs)
+                inputs = self._json_encode(domain_model.inputs)
 
         if domain_model.outputs is not None:
             result = self._truncate_and_upload(
@@ -436,11 +465,11 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                 ExecutionOffLoadType.OUTPUTS,
             )
             if result is not None:
-                db_model.outputs = self._json_encode(result.truncated_value)
+                outputs = self._json_encode(result.truncated_value)
                 domain_model.set_truncated_outputs(result.truncated_value)
-                offload_data = _replace_or_append_offload(offload_data, result.offload)
+                outputs_offload = result.offload
             else:
-                db_model.outputs = self._json_encode(domain_model.outputs)
+                outputs = self._json_encode(domain_model.outputs)
 
         if domain_model.process_data is not None:
             result = self._truncate_and_upload(
@@ -449,17 +478,100 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                 ExecutionOffLoadType.PROCESS_DATA,
             )
             if result is not None:
-                db_model.process_data = self._json_encode(result.truncated_value)
+                process_data = self._json_encode(result.truncated_value)
                 domain_model.set_truncated_process_data(result.truncated_value)
-                offload_data = _replace_or_append_offload(offload_data, result.offload)
+                process_data_offload = result.offload
             else:
-                db_model.process_data = self._json_encode(domain_model.process_data)
+                process_data = self._json_encode(domain_model.process_data)
 
-        db_model.offload_data = offload_data
         with self._session_factory() as session, session.begin():
-            session.merge(db_model)
+            query = WorkflowNodeExecutionModel.preload_offload_data(
+                select(WorkflowNodeExecutionModel)
+                .where(
+                    WorkflowNodeExecutionModel.id == domain_model.id,
+                    WorkflowNodeExecutionModel.tenant_id == self._tenant_id,
+                )
+                .with_for_update()
+            )
+            db_model = session.execute(query).scalars().first()
+
+            if db_model is not None and db_model.finished_at is not None:
+                if domain_model.finished_at is None or domain_model.finished_at < db_model.finished_at:
+                    return
+
+            if db_model is None:
+                db_model = self._to_db_model(domain_model)
+                session.add(db_model)
+
+            if domain_model.inputs is not None:
+                db_model.inputs = inputs
+                db_model.offload_data = _replace_offload(
+                    db_model.offload_data,
+                    ExecutionOffLoadType.INPUTS,
+                    inputs_offload,
+                )
+            if domain_model.outputs is not None:
+                db_model.outputs = outputs
+                db_model.offload_data = _replace_offload(
+                    db_model.offload_data,
+                    ExecutionOffLoadType.OUTPUTS,
+                    outputs_offload,
+                )
+            if domain_model.process_data is not None:
+                db_model.process_data = process_data
+                db_model.offload_data = _replace_offload(
+                    db_model.offload_data,
+                    ExecutionOffLoadType.PROCESS_DATA,
+                    process_data_offload,
+                )
+
+            # The terminal data task may run before its metadata-only sibling. Persisting
+            # the terminal markers here prevents a delayed retry snapshot from winning later.
+            if domain_model.finished_at is not None:
+                db_model.status = domain_model.status
+                db_model.error = domain_model.error
+                db_model.elapsed_time = domain_model.elapsed_time
+                db_model.execution_metadata = (
+                    json.dumps(jsonable_encoder(domain_model.metadata)) if domain_model.metadata else None
+                )
+                db_model.finished_at = domain_model.finished_at
+
             session.flush()
 
+    def _queue_async_save(self, execution: WorkflowNodeExecution) -> None:
+        if not self._triggered_from:
+            raise ValueError("triggered_from is required in repository constructor")
+        if not self._creator_user_id:
+            raise ValueError("created_by is required in repository constructor")
+        if not self._creator_user_role:
+            raise ValueError("created_by_role is required in repository constructor")
+
+        save_workflow_node_execution_task.delay(
+            execution_data=execution.model_dump(exclude={"inputs", "process_data", "outputs"}),
+            tenant_id=self._tenant_id,
+            app_id=self._app_id or "",
+            triggered_from=self._triggered_from.value,
+            creator_user_id=self._creator_user_id,
+            creator_user_role=self._creator_user_role.value,
+        )
+
+    def _queue_async_save_execution_data(self, execution: WorkflowNodeExecution) -> None:
+        if not self._triggered_from:
+            raise ValueError("triggered_from is required in repository constructor")
+        if not self._creator_user_id:
+            raise ValueError("created_by is required in repository constructor")
+        if not self._creator_user_role:
+            raise ValueError("created_by_role is required in repository constructor")
+
+        save_workflow_node_execution_data_task.delay(
+            execution_data=execution.model_dump(),
+            tenant_id=self._tenant_id,
+            app_id=self._app_id or "",
+            triggered_from=self._triggered_from.value,
+            creator_user_id=self._creator_user_id,
+            creator_user_role=self._creator_user_role.value,
+        )
+
     def get_db_models_by_workflow_run(
         self,
         workflow_run_id: str,
@@ -569,19 +681,12 @@ def _filter_by_offload_type(offload_type: ExecutionOffLoadType) -> Callable[[Wor
     return f
 
 
-def _replace_or_append_offload(
-    seq: list[WorkflowNodeExecutionOffload], elem: WorkflowNodeExecutionOffload
+def _replace_offload(
+    seq: list[WorkflowNodeExecutionOffload],
+    type_: ExecutionOffLoadType,
+    elem: WorkflowNodeExecutionOffload | None,
 ) -> list[WorkflowNodeExecutionOffload]:
-    """Replace all elements in `seq` that satisfy the equality condition defined by `eq_func` with `elem`.
-
-    Args:
-        seq: The sequence of elements to process.
-        elem: The new element to insert.
-        eq_func: A function that determines equality between elements.
-
-    Returns:
-        A new sequence with the specified elements replaced or appended.
-    """
-    ls = [i for i in seq if i.type_ != elem.type_]
-    ls.append(elem)
-    return ls
+    result = [item for item in seq if item.type_ != type_]
+    if elem is not None:
+        result.append(elem)
+    return result

api/tasks/workflow_execution_tasks.py

@@ -43,7 +43,7 @@ def save_workflow_execution_task(
         creator_user_role: Role of the user who created the execution
 
     Returns:
-        True if successful, False otherwise
+        True when the snapshot is persisted or safely ignored as stale.
     """
     try:
         with session_factory.create_session() as session:
@@ -51,9 +51,21 @@ def save_workflow_execution_task(
             execution = WorkflowExecution.model_validate(execution_data)
 
             # Check if workflow run already exists
-            existing_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == execution.id_))
+            existing_run = session.scalar(
+                select(WorkflowRun)
+                .where(
+                    WorkflowRun.id == execution.id_,
+                    WorkflowRun.tenant_id == tenant_id,
+                )
+                .with_for_update()
+            )
 
             if existing_run:
+                if existing_run.finished_at is not None and (
+                    execution.finished_at is None or execution.finished_at < existing_run.finished_at
+                ):
+                    logger.debug("Ignored stale workflow execution: %s", execution.id_)
+                    return True
                 # Update existing workflow run
                 _update_workflow_run_from_execution(existing_run, execution)
                 logger.debug("Updated existing workflow run: %s", execution.id_)
@@ -110,9 +122,10 @@ def _create_workflow_run_from_execution(
         else "{}"
     )
     workflow_run.error = execution.error_message
-    workflow_run.elapsed_time = execution.elapsed_time
+    workflow_run.elapsed_time = _calculate_elapsed_time(execution)
     workflow_run.total_tokens = execution.total_tokens
     workflow_run.total_steps = execution.total_steps
+    workflow_run.exceptions_count = execution.exceptions_count
     workflow_run.created_by_role = creator_user_role
     workflow_run.created_by = creator_user_id
     workflow_run.created_at = execution.started_at
@@ -131,7 +144,14 @@ def _update_workflow_run_from_execution(workflow_run: WorkflowRun, execution: Wo
         json.dumps(json_converter.to_json_encodable(execution.outputs)) if execution.outputs else "{}"
     )
     workflow_run.error = execution.error_message
-    workflow_run.elapsed_time = execution.elapsed_time
+    workflow_run.elapsed_time = _calculate_elapsed_time(execution)
     workflow_run.total_tokens = execution.total_tokens
     workflow_run.total_steps = execution.total_steps
+    workflow_run.exceptions_count = execution.exceptions_count
     workflow_run.finished_at = execution.finished_at
+
+
+def _calculate_elapsed_time(execution: WorkflowExecution) -> float:
+    if execution.finished_at is None:
+        return 0.0
+    return max((execution.finished_at - execution.started_at).total_seconds(), 0.0)

api/tasks/workflow_node_execution_tasks.py

@@ -7,7 +7,7 @@ improving performance by offloading storage operations to background workers.
 
 import json
 import logging
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from celery import shared_task
 from sqlalchemy import select
@@ -16,10 +16,15 @@ from core.db.session_factory import session_factory
 from graphon.entities.workflow_node_execution import (
     WorkflowNodeExecution,
 )
-from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
-from models import CreatorUserRole, WorkflowNodeExecutionModel
+from graphon.model_runtime.utils.encoders import jsonable_encoder
+from models import Account, CreatorUserRole, EndUser, WorkflowNodeExecutionModel
 from models.workflow import WorkflowNodeExecutionTriggeredFrom
 
+if TYPE_CHECKING:
+    from core.repositories.sqlalchemy_workflow_node_execution_repository import (
+        SQLAlchemyWorkflowNodeExecutionRepository,
+    )
+
 logger = logging.getLogger(__name__)
 
 
@@ -34,7 +39,7 @@ def save_workflow_node_execution_task(
     creator_user_role: str,
 ) -> bool:
     """
-    Asynchronously save or update a workflow node execution to the database.
+    Asynchronously save or update workflow node execution metadata to the database.
 
     Args:
         execution_data: Serialized WorkflowNodeExecution data
@@ -45,24 +50,30 @@ def save_workflow_node_execution_task(
         creator_user_role: Role of the user who created the execution
 
     Returns:
-        True if successful, False otherwise
+        True when the snapshot is persisted or safely ignored as stale.
     """
     try:
         with session_factory.create_session() as session:
-            # Deserialize execution data
             execution = WorkflowNodeExecution.model_validate(execution_data)
 
-            # Check if node execution already exists
             existing_execution = session.scalar(
-                select(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == execution.id)
+                select(WorkflowNodeExecutionModel)
+                .where(
+                    WorkflowNodeExecutionModel.id == execution.id,
+                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
+                )
+                .with_for_update()
             )
 
             if existing_execution:
-                # Update existing node execution
-                _update_node_execution_from_domain(existing_execution, execution)
-                logger.debug("Updated existing workflow node execution: %s", execution.id)
+                if existing_execution.finished_at is not None and (
+                    execution.finished_at is None or execution.finished_at < existing_execution.finished_at
+                ):
+                    logger.debug("Ignored stale workflow node execution metadata: %s", execution.id)
+                    return True
+                _update_node_execution_metadata(existing_execution, execution)
+                logger.debug("Updated existing workflow node execution metadata: %s", execution.id)
             else:
-                # Create new node execution
                 node_execution = _create_node_execution_from_domain(
                     execution=execution,
                     tenant_id=tenant_id,
@@ -79,10 +90,80 @@ def save_workflow_node_execution_task(
 
     except Exception as e:
         logger.exception("Failed to save workflow node execution %s", execution_data.get("id", "unknown"))
-        # Retry the task with exponential backoff
         raise self.retry(exc=e, countdown=60 * (2**self.request.retries))
 
 
+@shared_task(queue="workflow_storage", bind=True, max_retries=3, default_retry_delay=60)
+def save_workflow_node_execution_data_task(
+    self,
+    execution_data: dict[str, Any],
+    tenant_id: str,
+    app_id: str,
+    triggered_from: str,
+    creator_user_id: str,
+    creator_user_role: str,
+) -> bool:
+    """
+    Asynchronously save full workflow node execution data to the database.
+
+    This path preserves the SQLAlchemy repository's truncation and offload behavior while
+    moving the blocking database and storage work out of the workflow engine thread.
+    """
+    try:
+        execution = WorkflowNodeExecution.model_validate(execution_data)
+        repository = _create_sqlalchemy_repository(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            triggered_from=triggered_from,
+            creator_user_id=creator_user_id,
+            creator_user_role=creator_user_role,
+        )
+        repository.save_execution_data(execution)
+        return True
+    except Exception as e:
+        logger.exception("Failed to save workflow node execution data %s", execution_data.get("id", "unknown"))
+        raise self.retry(exc=e, countdown=60 * (2**self.request.retries))
+
+
+def _create_sqlalchemy_repository(
+    *,
+    tenant_id: str,
+    app_id: str,
+    triggered_from: str,
+    creator_user_id: str,
+    creator_user_role: str,
+) -> "SQLAlchemyWorkflowNodeExecutionRepository":
+    from core.repositories.sqlalchemy_workflow_node_execution_repository import (
+        SQLAlchemyWorkflowNodeExecutionRepository,
+    )
+
+    session_maker = session_factory.get_session_maker()
+    role = CreatorUserRole(creator_user_role)
+    user: Account | EndUser | None
+    with session_maker() as session:
+        if role == CreatorUserRole.ACCOUNT:
+            user = session.get(Account, creator_user_id)
+            if user is not None:
+                user.set_tenant_id(tenant_id)
+        else:
+            user = session.scalar(
+                select(EndUser).where(
+                    EndUser.id == creator_user_id,
+                    EndUser.tenant_id == tenant_id,
+                )
+            )
+
+    if user is None:
+        raise ValueError(f"Creator user {creator_user_id} not found for workflow node execution persistence")
+
+    return SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=session_maker,
+        user=user,
+        app_id=app_id or None,
+        triggered_from=WorkflowNodeExecutionTriggeredFrom(triggered_from),
+    )
+
+
 def _create_node_execution_from_domain(
     execution: WorkflowNodeExecution,
     tenant_id: str,
@@ -108,23 +189,11 @@ def _create_node_execution_from_domain(
     node_execution.title = execution.title
     node_execution.node_execution_id = execution.node_execution_id
 
-    # Serialize complex data as JSON
-    json_converter = WorkflowRuntimeTypeConverter()
-    node_execution.inputs = json.dumps(json_converter.to_json_encodable(execution.inputs)) if execution.inputs else "{}"
-    node_execution.process_data = (
-        json.dumps(json_converter.to_json_encodable(execution.process_data)) if execution.process_data else "{}"
-    )
-    node_execution.outputs = (
-        json.dumps(json_converter.to_json_encodable(execution.outputs)) if execution.outputs else "{}"
-    )
-    # Convert metadata enum keys to strings for JSON serialization
-    if execution.metadata:
-        metadata_for_json = {
-            key.value if hasattr(key, "value") else str(key): value for key, value in execution.metadata.items()
-        }
-        node_execution.execution_metadata = json.dumps(json_converter.to_json_encodable(metadata_for_json))
-    else:
-        node_execution.execution_metadata = "{}"
+    node_execution.inputs = "{}"
+    node_execution.process_data = "{}"
+    node_execution.outputs = "{}"
+
+    node_execution.execution_metadata = json.dumps(jsonable_encoder(execution.metadata)) if execution.metadata else "{}"
 
     node_execution.status = execution.status
     node_execution.error = execution.error
@@ -137,29 +206,12 @@ def _create_node_execution_from_domain(
     return node_execution
 
 
-def _update_node_execution_from_domain(node_execution: WorkflowNodeExecutionModel, execution: WorkflowNodeExecution):
+def _update_node_execution_metadata(node_execution: WorkflowNodeExecutionModel, execution: WorkflowNodeExecution):
     """
-    Update a WorkflowNodeExecutionModel database model from a WorkflowNodeExecution domain entity.
+    Update WorkflowNodeExecutionModel metadata without changing persisted data payload fields.
     """
-    # Update serialized data
-    json_converter = WorkflowRuntimeTypeConverter()
-    node_execution.inputs = json.dumps(json_converter.to_json_encodable(execution.inputs)) if execution.inputs else "{}"
-    node_execution.process_data = (
-        json.dumps(json_converter.to_json_encodable(execution.process_data)) if execution.process_data else "{}"
-    )
-    node_execution.outputs = (
-        json.dumps(json_converter.to_json_encodable(execution.outputs)) if execution.outputs else "{}"
-    )
-    # Convert metadata enum keys to strings for JSON serialization
-    if execution.metadata:
-        metadata_for_json = {
-            key.value if hasattr(key, "value") else str(key): value for key, value in execution.metadata.items()
-        }
-        node_execution.execution_metadata = json.dumps(json_converter.to_json_encodable(metadata_for_json))
-    else:
-        node_execution.execution_metadata = "{}"
+    node_execution.execution_metadata = json.dumps(jsonable_encoder(execution.metadata)) if execution.metadata else "{}"
 
-    # Update other fields
     node_execution.status = execution.status
     node_execution.error = execution.error
     node_execution.elapsed_time = execution.elapsed_time

api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_node_execution_repository.py

@@ -169,6 +169,44 @@ class TestSave:
             assert saved.status == WorkflowNodeExecutionStatus.SUCCEEDED
             assert saved.elapsed_time == 2.5
 
+    def test_save_execution_data_ignores_stale_snapshot(self, db_session_with_containers: Session) -> None:
+        account = _create_account_with_tenant(db_session_with_containers)
+        repo = _make_repo(db_session_with_containers, account, str(uuid4()))
+        created_at = datetime(2026, 1, 1, 12, 0, 0)
+        finished_at = datetime(2026, 1, 1, 12, 0, 1)
+        execution = WorkflowNodeExecution(
+            id=str(uuid4()),
+            workflow_id=str(uuid4()),
+            node_execution_id=str(uuid4()),
+            workflow_execution_id=str(uuid4()),
+            index=1,
+            node_id="node-1",
+            node_type=BuiltinNodeTypes.START,
+            title="Test Node",
+            inputs={"value": "final"},
+            outputs={"result": "final"},
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            created_at=created_at,
+            finished_at=finished_at,
+        )
+        repo.save(execution)
+        repo.save_execution_data(execution)
+
+        stale_execution = execution.model_copy(deep=True)
+        stale_execution.status = WorkflowNodeExecutionStatus.RETRY
+        stale_execution.inputs = {"value": "stale"}
+        stale_execution.finished_at = None
+        repo.save_execution_data(stale_execution)
+
+        engine = db_session_with_containers.get_bind()
+        assert isinstance(engine, Engine)
+        with sessionmaker(bind=engine, expire_on_commit=False)() as verify_session:
+            saved = verify_session.get(WorkflowNodeExecutionModel, execution.id)
+            assert saved is not None
+            assert saved.status == WorkflowNodeExecutionStatus.SUCCEEDED
+            assert saved.inputs_dict == {"value": "final"}
+            assert saved.finished_at == finished_at
+
 
 class TestGetByWorkflowExecution:
     def test_returns_executions_ordered(self, db_session_with_containers: Session) -> None:

api/tests/unit_tests/core/app/workflow/test_persistence_layer.py

@@ -5,8 +5,11 @@ from types import SimpleNamespace
 
 import pytest
 
-from core.app.entities.app_invoke_entities import WorkflowAppGenerateEntity
-from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.app.workflow.layers.persistence import (
+    PersistenceWorkflowInfo,
+    WorkflowPersistenceLayer,
+)
 from core.ops.ops_trace_manager import TraceTask, TraceTaskName
 from core.workflow.system_variables import SystemVariableKey, build_system_variables
 from graphon.entities import WorkflowNodeExecution
@@ -40,6 +43,7 @@ class _RepoRecorder:
     def __init__(self) -> None:
         self.saved: list[object] = []
         self.saved_exec_data: list[object] = []
+        self.async_enabled: bool | None = None
 
     def save(self, entity):
         self.saved.append(entity)
@@ -47,6 +51,9 @@ class _RepoRecorder:
     def save_execution_data(self, entity):
         self.saved_exec_data.append(entity)
 
+    def set_async_persistence(self, enabled: bool) -> None:
+        self.async_enabled = enabled
+
 
 def _naive_utc_now() -> datetime:
     return datetime.now(UTC).replace(tzinfo=None)
@@ -56,6 +63,7 @@ def _make_layer(
     system_variables: list | None = None,
     *,
     extras: dict | None = None,
+    invoke_from: InvokeFrom = InvokeFrom.DEBUGGER,
     trace_manager: object | None = None,
 ):
     system_variables = system_variables or build_system_variables(
@@ -75,7 +83,7 @@ def _make_layer(
         files=[],
         user_id="user",
         stream=False,
-        invoke_from=None,
+        invoke_from=invoke_from,
         trace_manager=None,
         workflow_execution_id="run-id",
         extras=extras or {},
@@ -97,6 +105,7 @@ def _make_layer(
         workflow_info=workflow_info,
         workflow_execution_repository=workflow_execution_repo,
         workflow_node_execution_repository=workflow_node_execution_repo,
+        invoke_from=invoke_from,
         trace_manager=trace_manager,
     )
     layer.initialize(read_only_state, command_channel=None)
@@ -105,6 +114,18 @@ def _make_layer(
 
 
 class TestWorkflowPersistenceLayer:
+    def test_configures_repositories_for_debug_synchronous_persistence(self):
+        _, exec_repo, node_repo, _ = _make_layer(invoke_from=InvokeFrom.DEBUGGER)
+
+        assert exec_repo.async_enabled is False
+        assert node_repo.async_enabled is False
+
+    def test_configures_repositories_for_non_debug_async_persistence(self):
+        _, exec_repo, node_repo, _ = _make_layer(invoke_from=InvokeFrom.WEB_APP)
+
+        assert exec_repo.async_enabled is True
+        assert node_repo.async_enabled is True
+
     def test_on_graph_start_resets_state(self):
         layer, _, _, _ = _make_layer()
         layer._workflow_execution = object()

api/tests/unit_tests/core/repositories/test_celery_workflow_execution_repository.py

@@ -1,275 +0,0 @@
-"""
-Unit tests for CeleryWorkflowExecutionRepository.
-
-These tests verify the Celery-based asynchronous storage functionality
-for workflow execution data.
-"""
-
-from unittest.mock import Mock, patch
-from uuid import uuid4
-
-import pytest
-
-from core.repositories.celery_workflow_execution_repository import CeleryWorkflowExecutionRepository
-from graphon.entities import WorkflowExecution
-from graphon.enums import WorkflowType
-from libs.datetime_utils import naive_utc_now
-from models import Account, EndUser
-from models.enums import WorkflowRunTriggeredFrom
-
-RESOURCE_TENANT_ID = "resource-tenant-id"
-
-
-@pytest.fixture
-def mock_session_factory():
-    """Mock SQLAlchemy session factory."""
-    from sqlalchemy import create_engine
-    from sqlalchemy.orm import sessionmaker
-
-    # Create a real sessionmaker with in-memory SQLite for testing
-    engine = create_engine("sqlite:///:memory:")
-    return sessionmaker(bind=engine)
-
-
-@pytest.fixture
-def mock_account():
-    """Mock Account user."""
-    account = Mock(spec=Account)
-    account.id = str(uuid4())
-    account.current_tenant_id = str(uuid4())
-    return account
-
-
-@pytest.fixture
-def mock_end_user():
-    """Mock EndUser."""
-    user = Mock(spec=EndUser)
-    user.id = str(uuid4())
-    user.tenant_id = str(uuid4())
-    return user
-
-
-@pytest.fixture
-def sample_workflow_execution():
-    """Sample WorkflowExecution for testing."""
-    return WorkflowExecution.new(
-        id_=str(uuid4()),
-        workflow_id=str(uuid4()),
-        workflow_type=WorkflowType.WORKFLOW,
-        workflow_version="1.0",
-        graph={"nodes": [], "edges": []},
-        inputs={"input1": "value1"},
-        started_at=naive_utc_now(),
-    )
-
-
-class TestCeleryWorkflowExecutionRepository:
-    """Test cases for CeleryWorkflowExecutionRepository."""
-
-    def test_init_with_sessionmaker(self, mock_session_factory, mock_account):
-        """Test repository initialization with sessionmaker."""
-        app_id = "test-app-id"
-        triggered_from = WorkflowRunTriggeredFrom.APP_RUN
-
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id=app_id,
-            triggered_from=triggered_from,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-        assert repo._app_id == app_id
-        assert repo._triggered_from == triggered_from
-        assert repo._creator_user_id == mock_account.id
-        assert repo._creator_user_role is not None
-
-    def test_init_basic_functionality(self, mock_session_factory, mock_account):
-        """Test repository initialization basic functionality."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
-        )
-
-        # Verify basic initialization
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-        assert repo._app_id == "test-app"
-        assert repo._triggered_from == WorkflowRunTriggeredFrom.DEBUGGING
-
-    def test_init_with_end_user(self, mock_session_factory, mock_end_user):
-        """Test repository initialization with EndUser."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_end_user,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-
-    def test_init_without_tenant_id_raises_error(self, mock_session_factory):
-        """Test that initialization fails without tenant_id."""
-        # Create a mock Account with no tenant_id
-        user = Mock(spec=Account)
-        user.current_tenant_id = None
-        user.id = str(uuid4())
-
-        with pytest.raises(ValueError, match="tenant_id is required"):
-            CeleryWorkflowExecutionRepository(
-                session_factory=mock_session_factory,
-                tenant_id="",
-                user=user,
-                app_id="test-app",
-                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-            )
-
-    def test_init_uses_resource_tenant_when_account_has_no_current_tenant(self, mock_session_factory):
-        user = Mock(spec=Account)
-        user.current_tenant_id = None
-        user.id = str(uuid4())
-
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=user,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-        assert repo._creator_user_id == user.id
-
-    @patch("core.repositories.celery_workflow_execution_repository.save_workflow_execution_task")
-    def test_save_queues_celery_task(self, mock_task, mock_session_factory, mock_account, sample_workflow_execution):
-        """Test that save operation queues a Celery task without tracking."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        repo.save(sample_workflow_execution)
-
-        # Verify Celery task was queued with correct parameters
-        mock_task.delay.assert_called_once()
-        call_args = mock_task.delay.call_args[1]
-
-        assert call_args["execution_data"] == sample_workflow_execution.model_dump()
-        assert call_args["tenant_id"] == RESOURCE_TENANT_ID
-        assert call_args["app_id"] == "test-app"
-        assert call_args["triggered_from"] == WorkflowRunTriggeredFrom.APP_RUN
-        assert call_args["creator_user_id"] == mock_account.id
-
-        # Verify no task tracking occurs (no _pending_saves attribute)
-        assert not hasattr(repo, "_pending_saves")
-
-    @patch("core.repositories.celery_workflow_execution_repository.save_workflow_execution_task")
-    def test_save_handles_celery_failure(
-        self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
-    ):
-        """Test that save operation handles Celery task failures."""
-        mock_task.delay.side_effect = Exception("Celery is down")
-
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        with pytest.raises(Exception, match="Celery is down"):
-            repo.save(sample_workflow_execution)
-
-    @patch("core.repositories.celery_workflow_execution_repository.save_workflow_execution_task")
-    def test_save_operation_fire_and_forget(
-        self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
-    ):
-        """Test that save operation works in fire-and-forget mode."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        # Test that save doesn't block or maintain state
-        repo.save(sample_workflow_execution)
-
-        # Verify no pending saves are tracked (no _pending_saves attribute)
-        assert not hasattr(repo, "_pending_saves")
-
-    @patch("core.repositories.celery_workflow_execution_repository.save_workflow_execution_task")
-    def test_multiple_save_operations(self, mock_task, mock_session_factory, mock_account):
-        """Test multiple save operations work correctly."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        # Create multiple executions
-        exec1 = WorkflowExecution.new(
-            id_=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_type=WorkflowType.WORKFLOW,
-            workflow_version="1.0",
-            graph={"nodes": [], "edges": []},
-            inputs={"input1": "value1"},
-            started_at=naive_utc_now(),
-        )
-        exec2 = WorkflowExecution.new(
-            id_=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_type=WorkflowType.WORKFLOW,
-            workflow_version="1.0",
-            graph={"nodes": [], "edges": []},
-            inputs={"input2": "value2"},
-            started_at=naive_utc_now(),
-        )
-
-        # Save both executions
-        repo.save(exec1)
-        repo.save(exec2)
-
-        # Should work without issues and not maintain state (no _pending_saves attribute)
-        assert not hasattr(repo, "_pending_saves")
-
-    @patch("core.repositories.celery_workflow_execution_repository.save_workflow_execution_task")
-    def test_save_with_different_user_types(self, mock_task, mock_session_factory, mock_end_user):
-        """Test save operation with different user types."""
-        repo = CeleryWorkflowExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=mock_end_user.tenant_id,
-            user=mock_end_user,
-            app_id="test-app",
-            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
-        )
-
-        execution = WorkflowExecution.new(
-            id_=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_type=WorkflowType.WORKFLOW,
-            workflow_version="1.0",
-            graph={"nodes": [], "edges": []},
-            inputs={"input1": "value1"},
-            started_at=naive_utc_now(),
-        )
-
-        repo.save(execution)
-
-        # Verify task was called with EndUser context
-        mock_task.delay.assert_called_once()
-        call_args = mock_task.delay.call_args[1]
-        assert call_args["tenant_id"] == mock_end_user.tenant_id
-        assert call_args["creator_user_id"] == mock_end_user.id

api/tests/unit_tests/core/repositories/test_celery_workflow_node_execution_repository.py

@@ -1,378 +0,0 @@
-"""
-Unit tests for CeleryWorkflowNodeExecutionRepository.
-
-These tests verify the Celery-based asynchronous storage functionality
-for workflow node execution data.
-"""
-
-from unittest.mock import Mock, patch
-from uuid import uuid4
-
-import pytest
-
-from core.repositories.celery_workflow_node_execution_repository import CeleryWorkflowNodeExecutionRepository
-from core.repositories.factory import OrderConfig
-from graphon.entities.workflow_node_execution import (
-    WorkflowNodeExecution,
-    WorkflowNodeExecutionStatus,
-)
-from graphon.enums import BuiltinNodeTypes
-from libs.datetime_utils import naive_utc_now
-from models import Account, EndUser
-from models.workflow import WorkflowNodeExecutionTriggeredFrom
-
-RESOURCE_TENANT_ID = "resource-tenant-id"
-
-
-@pytest.fixture
-def mock_session_factory():
-    """Mock SQLAlchemy session factory."""
-    from sqlalchemy import create_engine
-    from sqlalchemy.orm import sessionmaker
-
-    # Create a real sessionmaker with in-memory SQLite for testing
-    engine = create_engine("sqlite:///:memory:")
-    return sessionmaker(bind=engine)
-
-
-@pytest.fixture
-def mock_account():
-    """Mock Account user."""
-    account = Mock(spec=Account)
-    account.id = str(uuid4())
-    account.current_tenant_id = str(uuid4())
-    return account
-
-
-@pytest.fixture
-def mock_end_user():
-    """Mock EndUser."""
-    user = Mock(spec=EndUser)
-    user.id = str(uuid4())
-    user.tenant_id = str(uuid4())
-    return user
-
-
-@pytest.fixture
-def sample_workflow_node_execution():
-    """Sample WorkflowNodeExecution for testing."""
-    return WorkflowNodeExecution(
-        id=str(uuid4()),
-        node_execution_id=str(uuid4()),
-        workflow_id=str(uuid4()),
-        workflow_execution_id=str(uuid4()),
-        index=1,
-        node_id="test_node",
-        node_type=BuiltinNodeTypes.START,
-        title="Test Node",
-        inputs={"input1": "value1"},
-        status=WorkflowNodeExecutionStatus.RUNNING,
-        created_at=naive_utc_now(),
-    )
-
-
-class TestCeleryWorkflowNodeExecutionRepository:
-    """Test cases for CeleryWorkflowNodeExecutionRepository."""
-
-    def test_init_with_sessionmaker(self, mock_session_factory, mock_account):
-        """Test repository initialization with sessionmaker."""
-        app_id = "test-app-id"
-        triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
-
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id=app_id,
-            triggered_from=triggered_from,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-        assert repo._app_id == app_id
-        assert repo._triggered_from == triggered_from
-        assert repo._creator_user_id == mock_account.id
-        assert repo._creator_user_role is not None
-
-    def test_init_with_cache_initialized(self, mock_session_factory, mock_account):
-        """Test repository initialization with cache properly initialized."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
-        )
-
-        assert repo._execution_cache == {}
-        assert repo._workflow_execution_mapping == {}
-
-    def test_init_with_end_user(self, mock_session_factory, mock_end_user):
-        """Test repository initialization with EndUser."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_end_user,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-
-    def test_init_without_tenant_id_raises_error(self, mock_session_factory):
-        """Test that initialization fails without tenant_id."""
-        # Create a mock Account with no tenant_id
-        user = Mock(spec=Account)
-        user.current_tenant_id = None
-        user.id = str(uuid4())
-
-        with pytest.raises(ValueError, match="tenant_id is required"):
-            CeleryWorkflowNodeExecutionRepository(
-                session_factory=mock_session_factory,
-                tenant_id="",
-                user=user,
-                app_id="test-app",
-                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-            )
-
-    def test_init_uses_resource_tenant_when_account_has_no_current_tenant(self, mock_session_factory):
-        user = Mock(spec=Account)
-        user.current_tenant_id = None
-        user.id = str(uuid4())
-
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=user,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        assert repo._tenant_id == RESOURCE_TENANT_ID
-        assert repo._creator_user_id == user.id
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_save_caches_and_queues_celery_task(
-        self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution
-    ):
-        """Test that save operation caches execution and queues a Celery task."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        repo.save(sample_workflow_node_execution)
-
-        # Verify Celery task was queued with correct parameters
-        mock_task.delay.assert_called_once()
-        call_args = mock_task.delay.call_args[1]
-
-        assert call_args["execution_data"] == sample_workflow_node_execution.model_dump()
-        assert call_args["tenant_id"] == RESOURCE_TENANT_ID
-        assert call_args["app_id"] == "test-app"
-        assert call_args["triggered_from"] == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
-        assert call_args["creator_user_id"] == mock_account.id
-
-        # Verify execution is cached
-        assert sample_workflow_node_execution.id in repo._execution_cache
-        assert repo._execution_cache[sample_workflow_node_execution.id] == sample_workflow_node_execution
-
-        # Verify workflow execution mapping is updated
-        assert sample_workflow_node_execution.workflow_execution_id in repo._workflow_execution_mapping
-        assert (
-            sample_workflow_node_execution.id
-            in repo._workflow_execution_mapping[sample_workflow_node_execution.workflow_execution_id]
-        )
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_save_handles_celery_failure(
-        self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution
-    ):
-        """Test that save operation handles Celery task failures."""
-        mock_task.delay.side_effect = Exception("Celery is down")
-
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        with pytest.raises(Exception, match="Celery is down"):
-            repo.save(sample_workflow_node_execution)
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_get_by_workflow_execution_from_cache(
-        self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution
-    ):
-        """Test that get_by_workflow_execution retrieves executions from cache."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        # Save execution to cache first
-        repo.save(sample_workflow_node_execution)
-
-        workflow_execution_id = sample_workflow_node_execution.workflow_execution_id
-        order_config = OrderConfig(order_by=["index"], order_direction="asc")
-
-        result = repo.get_by_workflow_execution(workflow_execution_id, order_config)
-
-        # Verify results were retrieved from cache
-        assert len(result) == 1
-        assert result[0].id == sample_workflow_node_execution.id
-        assert result[0] is sample_workflow_node_execution
-
-    def test_get_by_workflow_execution_without_order_config(self, mock_session_factory, mock_account):
-        """Test get_by_workflow_execution without order configuration."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        result = repo.get_by_workflow_execution("workflow-run-id")
-
-        # Should return empty list since nothing in cache
-        assert len(result) == 0
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_cache_operations(self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution):
-        """Test cache operations work correctly."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        # Test saving to cache
-        repo.save(sample_workflow_node_execution)
-
-        # Verify cache contains the execution
-        assert sample_workflow_node_execution.id in repo._execution_cache
-
-        # Test retrieving from cache
-        result = repo.get_by_workflow_execution(sample_workflow_node_execution.workflow_execution_id)
-        assert len(result) == 1
-        assert result[0].id == sample_workflow_node_execution.id
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_multiple_executions_same_workflow(self, mock_task, mock_session_factory, mock_account):
-        """Test multiple executions for the same workflow."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=RESOURCE_TENANT_ID,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        # Create multiple executions for the same workflow
-        workflow_execution_id = str(uuid4())
-        exec1 = WorkflowNodeExecution(
-            id=str(uuid4()),
-            node_execution_id=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_execution_id=workflow_execution_id,
-            index=1,
-            node_id="node1",
-            node_type=BuiltinNodeTypes.START,
-            title="Node 1",
-            inputs={"input1": "value1"},
-            status=WorkflowNodeExecutionStatus.RUNNING,
-            created_at=naive_utc_now(),
-        )
-        exec2 = WorkflowNodeExecution(
-            id=str(uuid4()),
-            node_execution_id=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_execution_id=workflow_execution_id,
-            index=2,
-            node_id="node2",
-            node_type=BuiltinNodeTypes.LLM,
-            title="Node 2",
-            inputs={"input2": "value2"},
-            status=WorkflowNodeExecutionStatus.RUNNING,
-            created_at=naive_utc_now(),
-        )
-
-        # Save both executions
-        repo.save(exec1)
-        repo.save(exec2)
-
-        # Verify both are cached and mapped
-        assert len(repo._execution_cache) == 2
-        assert len(repo._workflow_execution_mapping[workflow_execution_id]) == 2
-
-        # Test retrieval
-        result = repo.get_by_workflow_execution(workflow_execution_id)
-        assert len(result) == 2
-
-    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
-    def test_ordering_functionality(self, mock_task, mock_session_factory, mock_account):
-        """Test ordering functionality works correctly."""
-        repo = CeleryWorkflowNodeExecutionRepository(
-            session_factory=mock_session_factory,
-            tenant_id=mock_account.current_tenant_id,
-            user=mock_account,
-            app_id="test-app",
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        # Create executions with different indices
-        workflow_execution_id = str(uuid4())
-        exec1 = WorkflowNodeExecution(
-            id=str(uuid4()),
-            node_execution_id=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_execution_id=workflow_execution_id,
-            index=2,
-            node_id="node2",
-            node_type=BuiltinNodeTypes.START,
-            title="Node 2",
-            inputs={},
-            status=WorkflowNodeExecutionStatus.RUNNING,
-            created_at=naive_utc_now(),
-        )
-        exec2 = WorkflowNodeExecution(
-            id=str(uuid4()),
-            node_execution_id=str(uuid4()),
-            workflow_id=str(uuid4()),
-            workflow_execution_id=workflow_execution_id,
-            index=1,
-            node_id="node1",
-            node_type=BuiltinNodeTypes.LLM,
-            title="Node 1",
-            inputs={},
-            status=WorkflowNodeExecutionStatus.RUNNING,
-            created_at=naive_utc_now(),
-        )
-
-        # Save in random order
-        repo.save(exec1)
-        repo.save(exec2)
-
-        # Test ascending order
-        order_config = OrderConfig(order_by=["index"], order_direction="asc")
-        result = repo.get_by_workflow_execution(workflow_execution_id, order_config)
-        assert len(result) == 2
-        assert result[0].index == 1
-        assert result[1].index == 2
-
-        # Test descending order
-        order_config = OrderConfig(order_by=["index"], order_direction="desc")
-        result = repo.get_by_workflow_execution(workflow_execution_id, order_config)
-        assert len(result) == 2
-        assert result[0].index == 2
-        assert result[1].index == 1

api/tests/unit_tests/core/repositories/test_sqlalchemy_workflow_execution_repository.py

@@ -1,5 +1,5 @@
 from datetime import UTC, datetime
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -287,6 +287,81 @@ class TestSQLAlchemyWorkflowExecutionRepository:
         cached_model = repo._execution_cache[sample_workflow_execution.id_]
         assert cached_model.id == sample_workflow_execution.id_
 
+    def test_save_rejects_existing_run_from_other_tenant(
+        self, mock_session_factory, mock_account, sample_workflow_execution
+    ):
+        repo = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=mock_session_factory,
+            tenant_id=RESOURCE_TENANT_ID,
+            user=mock_account,
+            app_id="test_app",
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        )
+        existing_run = WorkflowRun()
+        existing_run.id = sample_workflow_execution.id_
+        existing_run.tenant_id = "other-tenant"
+        session = mock_session_factory.return_value.__enter__.return_value
+        session.get.return_value = existing_run
+
+        with pytest.raises(ValueError, match="Unauthorized access to workflow run"):
+            repo.save(sample_workflow_execution)
+
+        session.merge.assert_not_called()
+        session.commit.assert_not_called()
+
+    @patch("core.repositories.sqlalchemy_workflow_execution_repository.save_workflow_execution_task")
+    def test_save_queues_celery_task_when_async_persistence_enabled(
+        self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
+    ):
+        repo = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=mock_session_factory,
+            tenant_id=RESOURCE_TENANT_ID,
+            user=mock_account,
+            app_id="test_app",
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        )
+        repo.set_async_persistence(True)
+
+        repo.save(sample_workflow_execution)
+
+        mock_task.delay.assert_called_once()
+        call_args = mock_task.delay.call_args.kwargs
+        assert call_args["execution_data"] == sample_workflow_execution.model_dump()
+        assert call_args["tenant_id"] == RESOURCE_TENANT_ID
+        assert call_args["app_id"] == "test_app"
+        assert call_args["triggered_from"] == WorkflowRunTriggeredFrom.APP_RUN
+        assert call_args["creator_user_id"] == mock_account.id
+        session = mock_session_factory.return_value.__enter__.return_value
+        session.merge.assert_not_called()
+
+    @patch("core.repositories.sqlalchemy_workflow_execution_repository.save_workflow_execution_task")
+    def test_queue_async_save_requires_context(
+        self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
+    ):
+        repo = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=mock_session_factory,
+            tenant_id=RESOURCE_TENANT_ID,
+            user=mock_account,
+            app_id="test_app",
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        )
+
+        repo._triggered_from = None
+        with pytest.raises(ValueError, match="triggered_from is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        repo._triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+        repo._creator_user_id = None
+        with pytest.raises(ValueError, match="created_by is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        repo._creator_user_id = "user-id"
+        repo._creator_user_role = None
+        with pytest.raises(ValueError, match="created_by_role is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        mock_task.delay.assert_not_called()
+
     def test_save_uses_execution_started_at_when_record_does_not_exist(
         self, mock_session_factory, mock_account, sample_workflow_execution
     ):

api/tests/unit_tests/core/repositories/test_sqlalchemy_workflow_node_execution_repository.py

@@ -6,7 +6,7 @@ from collections.abc import Mapping
 from datetime import UTC, datetime
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock, Mock
+from unittest.mock import MagicMock, Mock, patch
 
 import psycopg2.errors
 import pytest
@@ -21,7 +21,7 @@ from core.repositories.sqlalchemy_workflow_node_execution_repository import (
     _deterministic_json_dump,
     _filter_by_offload_type,
     _find_first,
-    _replace_or_append_offload,
+    _replace_offload,
 )
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import (
@@ -60,6 +60,7 @@ def _execution(
     outputs: Mapping[str, Any] | None = None,
     process_data: Mapping[str, Any] | None = None,
     metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None,
+    finished_at: datetime | None = None,
 ) -> WorkflowNodeExecution:
     return WorkflowNodeExecution(
         id=execution_id,
@@ -79,7 +80,7 @@ def _execution(
         elapsed_time=1.0,
         metadata=metadata,
         created_at=datetime.now(UTC),
-        finished_at=None,
+        finished_at=finished_at,
     )
 
 
@@ -212,7 +213,7 @@ def test_create_truncator_uses_config(monkeypatch: pytest.MonkeyPatch) -> None:
     assert created["max_size_bytes"] == dify_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE
 
 
-def test_helpers_find_first_and_replace_or_append_and_filter() -> None:
+def test_helpers_find_first_replace_and_filter() -> None:
     assert _deterministic_json_dump({"b": 1, "a": 2}) == '{"a": 2, "b": 1}'
     assert _find_first([], lambda _: True) is None
     assert _find_first([1, 2, 3], lambda x: x > 1) == 2
@@ -221,9 +222,11 @@ def test_helpers_find_first_and_replace_or_append_and_filter() -> None:
     off2 = WorkflowNodeExecutionOffload(type_=ExecutionOffLoadType.OUTPUTS)
     assert _find_first([off1, off2], _filter_by_offload_type(ExecutionOffLoadType.OUTPUTS)) is off2
 
-    replaced = _replace_or_append_offload([off1, off2], WorkflowNodeExecutionOffload(type_=ExecutionOffLoadType.INPUTS))
+    replacement = WorkflowNodeExecutionOffload(type_=ExecutionOffLoadType.INPUTS)
+    replaced = _replace_offload([off1, off2], ExecutionOffLoadType.INPUTS, replacement)
     assert len(replaced) == 2
     assert [o.type_ for o in replaced] == [ExecutionOffLoadType.OUTPUTS, ExecutionOffLoadType.INPUTS]
+    assert _replace_offload(replaced, ExecutionOffLoadType.INPUTS, None) == [off2]
 
 
 def test_to_db_model_requires_constructor_context(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -330,19 +333,39 @@ def test_persist_to_database_updates_existing_and_inserts_new(monkeypatch: pytes
     db_model.foo = "bar"  # type: ignore[attr-defined]
     db_model.__dict__["_private"] = "x"
 
-    existing = SimpleNamespace()
+    existing = SimpleNamespace(tenant_id="tenant")
     session.get.return_value = existing
     repo._persist_to_database(db_model)
     assert existing.foo == "bar"
     session.add.assert_not_called()
-    assert repo._node_execution_cache["node1"] is db_model
 
     session.reset_mock()
     session.get.return_value = None
-    repo._node_execution_cache.clear()
     repo._persist_to_database(db_model)
     session.add.assert_called_once_with(db_model)
-    assert repo._node_execution_cache["node1"] is db_model
+
+
+def test_persist_to_database_rejects_existing_execution_from_other_tenant(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    session = MagicMock()
+    session.get.return_value = SimpleNamespace(tenant_id="other-tenant")
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=_session_factory(session),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id=None,
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    db_model = WorkflowNodeExecutionModel()
+    db_model.id = "id1"
+
+    with pytest.raises(ValueError, match="Unauthorized access"):
+        repo._persist_to_database(db_model)
+
+    session.commit.assert_not_called()
 
 
 def test_truncate_and_upload_returns_none_when_no_values_or_not_truncated(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -518,14 +541,16 @@ def test_save_execution_data_handles_existing_db_model_and_truncation(monkeypatc
         lambda *_: SimpleNamespace(upload_file=Mock()),
     )
     session = MagicMock()
-    session.execute.return_value.scalars.return_value.first.return_value = SimpleNamespace(
+    existing = SimpleNamespace(
         id="id",
         offload_data=[WorkflowNodeExecutionOffload(type_=ExecutionOffLoadType.INPUTS)],
         inputs=None,
         outputs=None,
         process_data=None,
+        finished_at=None,
     )
-    session.merge = Mock()
+    session.scalar.return_value = existing
+    session.execute.return_value.scalars.return_value.first.return_value = existing
     session.flush = Mock()
     session.begin.return_value.__enter__ = Mock(return_value=session)
     session.begin.return_value.__exit__ = Mock(return_value=None)
@@ -551,11 +576,10 @@ def test_save_execution_data_handles_existing_db_model_and_truncation(monkeypatc
 
     repo.save_execution_data(execution)
     # Inputs should be truncated, outputs/process_data encoded directly
-    db_model = session.merge.call_args.args[0]
-    assert json.loads(db_model.inputs) == {"trunc": True}
-    assert json.loads(db_model.outputs) == {"b": 2}
-    assert json.loads(db_model.process_data) == {"c": 3}
-    assert any(off.type_ == ExecutionOffLoadType.INPUTS for off in db_model.offload_data)
+    assert json.loads(existing.inputs) == {"trunc": True}
+    assert json.loads(existing.outputs) == {"b": 2}
+    assert json.loads(existing.process_data) == {"c": 3}
+    assert any(off.type_ == ExecutionOffLoadType.INPUTS for off in existing.offload_data)
     assert execution.get_truncated_inputs() == {"trunc": True}
 
 
@@ -570,10 +594,11 @@ def test_save_execution_data_truncates_outputs_and_process_data(monkeypatch: pyt
         inputs=None,
         outputs=None,
         process_data=None,
+        finished_at=None,
     )
     session = MagicMock()
+    session.scalar.return_value = existing
     session.execute.return_value.scalars.return_value.first.return_value = existing
-    session.merge = Mock()
     session.flush = Mock()
     session.begin.return_value.__enter__ = Mock(return_value=session)
     session.begin.return_value.__exit__ = Mock(return_value=None)
@@ -605,9 +630,8 @@ def test_save_execution_data_truncates_outputs_and_process_data(monkeypatch: pyt
     monkeypatch.setattr(repo, "_json_encode", lambda values: json.dumps(values, sort_keys=True))
 
     repo.save_execution_data(execution)
-    db_model = session.merge.call_args.args[0]
-    assert json.loads(db_model.outputs) == {"b": "trunc"}
-    assert json.loads(db_model.process_data) == {"c": "trunc"}
+    assert json.loads(existing.outputs) == {"b": "trunc"}
+    assert json.loads(existing.process_data) == {"c": "trunc"}
     assert execution.get_truncated_outputs() == {"b": "trunc"}
     assert execution.get_truncated_process_data() == {"c": "trunc"}
 
@@ -618,8 +642,8 @@ def test_save_execution_data_handles_missing_db_model(monkeypatch: pytest.Monkey
         lambda *_: SimpleNamespace(upload_file=Mock()),
     )
     session = MagicMock()
+    session.scalar.return_value = None
     session.execute.return_value.scalars.return_value.first.return_value = None
-    session.merge = Mock()
     session.flush = Mock()
     session.begin.return_value.__enter__ = Mock(return_value=session)
     session.begin.return_value.__exit__ = Mock(return_value=None)
@@ -639,8 +663,200 @@ def test_save_execution_data_handles_missing_db_model(monkeypatch: pytest.Monkey
     monkeypatch.setattr(repo, "_json_encode", lambda values: json.dumps(values))
 
     repo.save_execution_data(execution)
-    merged = session.merge.call_args.args[0]
-    assert merged.inputs == '{"a": 1}'
+    session.add.assert_called_once_with(fake_db_model)
+    assert fake_db_model.inputs == '{"a": 1}'
+
+
+def test_save_execution_data_updates_terminal_metadata_without_merging_whole_row(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    existing = SimpleNamespace(
+        offload_data=[],
+        inputs=None,
+        outputs=None,
+        process_data=None,
+        status=WorkflowNodeExecutionStatus.RUNNING,
+        error=None,
+        elapsed_time=0.0,
+        execution_metadata=None,
+        finished_at=None,
+    )
+    session = MagicMock()
+    session.scalar.return_value = None
+    session.execute.return_value.scalars.return_value.first.return_value = existing
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=_session_factory(session),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    monkeypatch.setattr(repo, "_truncate_and_upload", lambda *_args, **_kwargs: None)
+    finished_at = datetime.now(UTC)
+    execution = _execution(
+        inputs={"final": True},
+        metadata={WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: 2},
+        finished_at=finished_at,
+    )
+
+    repo.save_execution_data(execution)
+
+    assert existing.status == WorkflowNodeExecutionStatus.SUCCEEDED
+    assert existing.finished_at == finished_at
+    assert json.loads(existing.inputs) == {"final": True}
+    assert json.loads(existing.execution_metadata) == {"total_tokens": 2}
+    session.merge.assert_not_called()
+
+
+def test_save_execution_data_ignores_stale_nonterminal_snapshot(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    existing = SimpleNamespace(finished_at=datetime(2026, 1, 1), inputs='{"current": true}')
+    session = MagicMock()
+    session.scalar.return_value = existing
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=_session_factory(session),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    truncate = Mock()
+    monkeypatch.setattr(repo, "_truncate_and_upload", truncate)
+
+    repo.save_execution_data(_execution(inputs={"stale": True}))
+
+    truncate.assert_not_called()
+    session.execute.assert_not_called()
+    assert existing.inputs == '{"current": true}'
+
+
+@patch("core.repositories.sqlalchemy_workflow_node_execution_repository.save_workflow_node_execution_task")
+def test_save_queues_celery_task_when_async_persistence_enabled(mock_task, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    repo.set_async_persistence(True)
+    execution = _execution(inputs={"a": 1})
+
+    repo.save(execution)
+
+    mock_task.delay.assert_called_once()
+    call_args = mock_task.delay.call_args.kwargs
+    assert call_args["execution_data"] == execution.model_dump(exclude={"inputs", "process_data", "outputs"})
+    assert call_args["tenant_id"] == "tenant"
+    assert call_args["app_id"] == "app"
+    assert call_args["triggered_from"] == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    assert call_args["creator_user_id"] == "user"
+    assert set(call_args) == {
+        "execution_data",
+        "tenant_id",
+        "app_id",
+        "triggered_from",
+        "creator_user_id",
+        "creator_user_role",
+    }
+
+
+@patch("core.repositories.sqlalchemy_workflow_node_execution_repository.save_workflow_node_execution_data_task")
+def test_save_execution_data_queues_celery_task_when_async_persistence_enabled(
+    mock_task, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    repo.set_async_persistence(True)
+    execution = _execution(inputs={"a": 1})
+
+    repo.save_execution_data(execution)
+
+    mock_task.delay.assert_called_once()
+    call_args = mock_task.delay.call_args.kwargs
+    assert call_args["execution_data"] == execution.model_dump()
+    assert call_args["tenant_id"] == "tenant"
+    assert call_args["app_id"] == "app"
+    assert call_args["triggered_from"] == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    assert call_args["creator_user_id"] == "user"
+
+
+def test_queue_async_save_requires_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    execution = _execution()
+
+    repo._triggered_from = None
+    with pytest.raises(ValueError, match="triggered_from is required"):
+        repo._queue_async_save(execution)
+
+    repo._triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    repo._creator_user_id = None
+    with pytest.raises(ValueError, match="created_by is required"):
+        repo._queue_async_save(execution)
+
+    repo._creator_user_id = "user"
+    repo._creator_user_role = None
+    with pytest.raises(ValueError, match="created_by_role is required"):
+        repo._queue_async_save(execution)
+
+
+def test_queue_async_save_execution_data_requires_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        tenant_id=RESOURCE_TENANT_ID,
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    execution = _execution()
+
+    repo._triggered_from = None
+    with pytest.raises(ValueError, match="triggered_from is required"):
+        repo._queue_async_save_execution_data(execution)
+
+    repo._triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    repo._creator_user_id = None
+    with pytest.raises(ValueError, match="created_by is required"):
+        repo._queue_async_save_execution_data(execution)
+
+    repo._creator_user_id = "user"
+    repo._creator_user_role = None
+    with pytest.raises(ValueError, match="created_by_role is required"):
+        repo._queue_async_save_execution_data(execution)
 
 
 def test_save_retries_duplicate_and_logs_non_duplicate(

api/tests/unit_tests/core/repositories/test_workflow_node_execution_conflict_handling.py

@@ -99,6 +99,7 @@ class TestWorkflowNodeExecutionConflictHandling:
 
         # Mock existing record
         mock_existing = MagicMock()
+        mock_existing.tenant_id = "test-tenant-id"
         mock_session.get.return_value = mock_existing
         mock_session.commit.return_value = None
 

api/tests/unit_tests/tasks/test_workflow_execution_tasks.py

@@ -0,0 +1,115 @@
+from datetime import datetime
+from unittest.mock import Mock, patch
+
+from graphon.entities import WorkflowExecution
+from graphon.enums import WorkflowExecutionStatus, WorkflowType
+from models import CreatorUserRole, WorkflowRun
+from models.enums import WorkflowRunTriggeredFrom
+from tasks.workflow_execution_tasks import (
+    _calculate_elapsed_time,
+    _create_workflow_run_from_execution,
+    _update_workflow_run_from_execution,
+    save_workflow_execution_task,
+)
+
+
+def _execution(
+    *,
+    exceptions_count: int = 2,
+    finished_at: datetime | None = None,
+    status: WorkflowExecutionStatus = WorkflowExecutionStatus.SUCCEEDED,
+) -> WorkflowExecution:
+    started_at = datetime(2026, 1, 1, 12, 0, 0)
+    return WorkflowExecution(
+        id_="workflow-run-id",
+        workflow_id="workflow-id",
+        workflow_type=WorkflowType.WORKFLOW,
+        workflow_version="1.0",
+        graph={"nodes": [], "edges": []},
+        inputs={"input": "value"},
+        outputs={"output": "value"},
+        status=status,
+        error_message="",
+        total_tokens=100,
+        total_steps=5,
+        exceptions_count=exceptions_count,
+        started_at=started_at,
+        finished_at=finished_at,
+    )
+
+
+def test_create_workflow_run_calculates_elapsed_time_and_exceptions_count() -> None:
+    execution = _execution(finished_at=datetime(2026, 1, 1, 12, 0, 12), exceptions_count=3)
+
+    workflow_run = _create_workflow_run_from_execution(
+        execution=execution,
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT,
+    )
+
+    assert workflow_run.elapsed_time == 12.0
+    assert workflow_run.exceptions_count == 3
+
+
+def test_update_workflow_run_calculates_elapsed_time_and_exceptions_count() -> None:
+    workflow_run = WorkflowRun()
+    execution = _execution(finished_at=datetime(2026, 1, 1, 12, 0, 8), exceptions_count=4)
+
+    _update_workflow_run_from_execution(workflow_run, execution)
+
+    assert workflow_run.elapsed_time == 8.0
+    assert workflow_run.exceptions_count == 4
+
+
+def test_calculate_elapsed_time_is_zero_until_finished() -> None:
+    assert _calculate_elapsed_time(_execution(finished_at=None)) == 0.0
+
+
+def test_calculate_elapsed_time_clamps_negative_duration_to_zero() -> None:
+    execution = _execution(finished_at=datetime(2026, 1, 1, 11, 59, 59))
+
+    assert _calculate_elapsed_time(execution) == 0.0
+
+
+@patch("tasks.workflow_execution_tasks.session_factory.create_session")
+def test_save_workflow_execution_task_ignores_stale_nonterminal_snapshot(mock_create_session: Mock) -> None:
+    existing_run = WorkflowRun()
+    existing_run.status = WorkflowExecutionStatus.SUCCEEDED
+    existing_run.finished_at = datetime(2026, 1, 1, 12, 0, 10)
+    session = _TaskSession(existing_run)
+    mock_create_session.return_value = session
+    execution = _execution(finished_at=None, status=WorkflowExecutionStatus.RUNNING)
+
+    result = save_workflow_execution_task.run(
+        execution_data=execution.model_dump(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+
+    assert result is True
+    assert session.committed is False
+    assert existing_run.status == WorkflowExecutionStatus.SUCCEEDED
+
+
+class _TaskSession:
+    def __init__(self, existing_run: WorkflowRun) -> None:
+        self._existing_run = existing_run
+        self.committed = False
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        return None
+
+    def scalar(self, _stmt):
+        return self._existing_run
+
+    def commit(self) -> None:
+        self.committed = True

api/tests/unit_tests/tasks/test_workflow_node_execution_tasks.py

@@ -1,488 +1,354 @@
-# """
-# Unit tests for workflow node execution Celery tasks.
-
-# These tests verify the asynchronous storage functionality for workflow node execution data,
-# including truncation and offloading logic.
-# """
-
-# import json
-# from unittest.mock import MagicMock, Mock, patch
-# from uuid import uuid4
-
-# import pytest
-
-# from graphon.entities.workflow_node_execution import (
-#     WorkflowNodeExecution,
-#     WorkflowNodeExecutionStatus,
-# )
-# from graphon.enums import BuiltinNodeTypes
-# from libs.datetime_utils import naive_utc_now
-# from models import WorkflowNodeExecutionModel
-# from models.enums import ExecutionOffLoadType
-# from models.model import UploadFile
-# from models.workflow import WorkflowNodeExecutionOffload, WorkflowNodeExecutionTriggeredFrom
-# from tasks.workflow_node_execution_tasks import (
-#     _create_truncator,
-#     _json_encode,
-#     _replace_or_append_offload,
-#     _truncate_and_upload_async,
-#     save_workflow_node_execution_data_task,
-#     save_workflow_node_execution_task,
-# )
-
-
-# @pytest.fixture
-# def sample_execution_data():
-#     """Sample execution data for testing."""
-#     execution = WorkflowNodeExecution(
-#         id=str(uuid4()),
-#         node_execution_id=str(uuid4()),
-#         workflow_id=str(uuid4()),
-#         workflow_execution_id=str(uuid4()),
-#         index=1,
-#         node_id="test_node",
-#         node_type=BuiltinNodeTypes.LLM,
-#         title="Test Node",
-#         inputs={"input_key": "input_value"},
-#         outputs={"output_key": "output_value"},
-#         process_data={"process_key": "process_value"},
-#         status=WorkflowNodeExecutionStatus.RUNNING,
-#         created_at=naive_utc_now(),
-#     )
-#     return execution.model_dump()
-
-
-# @pytest.fixture
-# def mock_db_model():
-#     """Mock database model for testing."""
-#     db_model = Mock(spec=WorkflowNodeExecutionModel)
-#     db_model.id = "test-execution-id"
-#     db_model.offload_data = []
-#     return db_model
-
-
-# @pytest.fixture
-# def mock_file_service():
-#     """Mock file service for testing."""
-#     file_service = Mock()
-#     mock_upload_file = Mock(spec=UploadFile)
-#     mock_upload_file.id = "mock-file-id"
-#     file_service.upload_file.return_value = mock_upload_file
-#     return file_service
-
-
-# class TestSaveWorkflowNodeExecutionDataTask:
-#     """Test cases for save_workflow_node_execution_data_task."""
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     @patch("tasks.workflow_node_execution_tasks.select")
-#     def test_save_execution_data_task_success(
-#         self, mock_select, mock_sessionmaker, sample_execution_data, mock_db_model
-#     ):
-#         """Test successful execution of save_workflow_node_execution_data_task."""
-#         # Setup mocks
-#         mock_session = MagicMock()
-#         mock_sessionmaker.return_value.return_value.__enter__.return_value = mock_session
-#         mock_session.execute.return_value.scalars.return_value.first.return_value = mock_db_model
-
-#         # Execute task
-#         result = save_workflow_node_execution_data_task(
-#             execution_data=sample_execution_data,
-#             tenant_id="test-tenant-id",
-#             app_id="test-app-id",
-#             user_data={"user_id": "test-user-id", "user_type": "account"},
-#         )
-
-#         # Verify success
-#         assert result is True
-#         mock_session.merge.assert_called_once_with(mock_db_model)
-#         mock_session.commit.assert_called_once()
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     @patch("tasks.workflow_node_execution_tasks.select")
-#     def test_save_execution_data_task_execution_not_found(self, mock_select, mock_sessionmaker,
-# sample_execution_data):
-#         """Test task when execution is not found in database."""
-#         # Setup mocks
-#         mock_session = MagicMock()
-#         mock_sessionmaker.return_value.return_value.__enter__.return_value = mock_session
-#         mock_session.execute.return_value.scalars.return_value.first.return_value = None
-
-#         # Execute task
-#         result = save_workflow_node_execution_data_task(
-#             execution_data=sample_execution_data,
-#             tenant_id="test-tenant-id",
-#             app_id="test-app-id",
-#             user_data={"user_id": "test-user-id", "user_type": "account"},
-#         )
-
-#         # Verify failure
-#         assert result is False
-#         mock_session.merge.assert_not_called()
-#         mock_session.commit.assert_not_called()
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     @patch("tasks.workflow_node_execution_tasks.select")
-#     def test_save_execution_data_task_with_truncation(self, mock_select, mock_sessionmaker, mock_db_model):
-#         """Test task with data that requires truncation."""
-#         # Create execution with large data
-#         large_data = {"large_field": "x" * 10000}
-#         execution = WorkflowNodeExecution(
-#             id=str(uuid4()),
-#             node_execution_id=str(uuid4()),
-#             workflow_id=str(uuid4()),
-#             workflow_execution_id=str(uuid4()),
-#             index=1,
-#             node_id="test_node",
-#             node_type=BuiltinNodeTypes.LLM,
-#             title="Test Node",
-#             inputs=large_data,
-#             outputs=large_data,
-#             process_data=large_data,
-#             status=WorkflowNodeExecutionStatus.RUNNING,
-#             created_at=naive_utc_now(),
-#         )
-#         execution_data = execution.model_dump()
-
-#         # Setup mocks
-#         mock_session = MagicMock()
-#         mock_sessionmaker.return_value.return_value.__enter__.return_value = mock_session
-#         mock_session.execute.return_value.scalars.return_value.first.return_value = mock_db_model
-
-#         # Create mock upload file
-#         mock_upload_file = Mock(spec=UploadFile)
-#         mock_upload_file.id = "mock-file-id"
-
-#         # Execute task
-#         with patch("tasks.workflow_node_execution_tasks._truncate_and_upload_async") as mock_truncate:
-#             # Mock truncation results
-#             mock_truncate.return_value = {
-#                 "truncated_value": {"large_field": "[TRUNCATED]"},
-#                 "file": mock_upload_file,
-#                 "offload": WorkflowNodeExecutionOffload(
-#                     id=str(uuid4()),
-#                     tenant_id="test-tenant-id",
-#                     app_id="test-app-id",
-#                     node_execution_id=execution.id,
-#                     type_=ExecutionOffLoadType.INPUTS,
-#                     file_id=mock_upload_file.id,
-#                 ),
-#             }
-
-#             result = save_workflow_node_execution_data_task(
-#                 execution_data=execution_data,
-#                 tenant_id="test-tenant-id",
-#                 app_id="test-app-id",
-#                 user_data={"user_id": "test-user-id", "user_type": "account"},
-#             )
-
-#             # Verify success and truncation was called
-#             assert result is True
-#             assert mock_truncate.call_count == 3  # inputs, outputs, process_data
-#             mock_session.merge.assert_called_once_with(mock_db_model)
-#             mock_session.commit.assert_called_once()
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     def test_save_execution_data_task_retry_on_exception(self, mock_sessionmaker, sample_execution_data):
-#         """Test task retry mechanism on exception."""
-#         # Setup mock to raise exception
-#         mock_sessionmaker.side_effect = Exception("Database error")
-
-#         # Create a mock task instance with proper retry behavior
-#         with patch.object(save_workflow_node_execution_data_task, "retry") as mock_retry:
-#             mock_retry.side_effect = Exception("Retry called")
-
-#             # Execute task and expect retry
-#             with pytest.raises(Exception, match="Retry called"):
-#                 save_workflow_node_execution_data_task(
-#                     execution_data=sample_execution_data,
-#                     tenant_id="test-tenant-id",
-#                     app_id="test-app-id",
-#                     user_data={"user_id": "test-user-id", "user_type": "account"},
-#                 )
-
-#             # Verify retry was called
-#             mock_retry.assert_called_once()
-
-
-# class TestTruncateAndUploadAsync:
-#     """Test cases for _truncate_and_upload_async function."""
-
-#     def test_truncate_and_upload_with_none_values(self, mock_file_service):
-#         """Test _truncate_and_upload_async with None values."""
-#         # The function handles None values internally, so we test with empty dict instead
-#         result = _truncate_and_upload_async(
-#             values={},
-#             execution_id="test-id",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             user_data={"user_id": "test-user", "user_type": "account"},
-#             file_service=mock_file_service,
-#         )
-
-#         # Empty dict should not require truncation
-#         assert result is None
-#         mock_file_service.upload_file.assert_not_called()
-
-#     @patch("tasks.workflow_node_execution_tasks._create_truncator")
-#     def test_truncate_and_upload_no_truncation_needed(self, mock_create_truncator, mock_file_service):
-#         """Test _truncate_and_upload_async when no truncation is needed."""
-#         # Mock truncator to return no truncation
-#         mock_truncator = Mock()
-#         mock_truncator.truncate_variable_mapping.return_value = ({"small": "data"}, False)
-#         mock_create_truncator.return_value = mock_truncator
-
-#         small_values = {"small": "data"}
-#         result = _truncate_and_upload_async(
-#             values=small_values,
-#             execution_id="test-id",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             user_data={"user_id": "test-user", "user_type": "account"},
-#             file_service=mock_file_service,
-#         )
-
-#         assert result is None
-#         mock_file_service.upload_file.assert_not_called()
-
-#     @patch("tasks.workflow_node_execution_tasks._create_truncator")
-#     @patch("models.Account")
-#     @patch("models.Tenant")
-#     def test_truncate_and_upload_with_account_user(
-#         self, mock_tenant_class, mock_account_class, mock_create_truncator, mock_file_service
-#     ):
-#         """Test _truncate_and_upload_async with account user."""
-#         # Mock truncator to return truncation needed
-#         mock_truncator = Mock()
-#         mock_truncator.truncate_variable_mapping.return_value = ({"truncated": "data"}, True)
-#         mock_create_truncator.return_value = mock_truncator
-
-#         # Mock user and tenant creation
-#         mock_account = Mock()
-#         mock_account.id = "test-user"
-#         mock_account_class.return_value = mock_account
-
-#         mock_tenant = Mock()
-#         mock_tenant.id = "test-tenant"
-#         mock_tenant_class.return_value = mock_tenant
-
-#         large_values = {"large": "x" * 10000}
-#         result = _truncate_and_upload_async(
-#             values=large_values,
-#             execution_id="test-id",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             user_data={"user_id": "test-user", "user_type": "account"},
-#             file_service=mock_file_service,
-#         )
-
-#         # Verify result structure
-#         assert result is not None
-#         assert "truncated_value" in result
-#         assert "file" in result
-#         assert "offload" in result
-#         assert result["truncated_value"] == {"truncated": "data"}
-
-#         # Verify file upload was called
-#         mock_file_service.upload_file.assert_called_once()
-#         upload_call = mock_file_service.upload_file.call_args
-#         assert upload_call[1]["filename"] == "node_execution_test-id_inputs.json"
-#         assert upload_call[1]["mimetype"] == "application/json"
-#         assert upload_call[1]["user"] == mock_account
-
-#     @patch("tasks.workflow_node_execution_tasks._create_truncator")
-#     @patch("models.EndUser")
-#     def test_truncate_and_upload_with_end_user(self, mock_end_user_class, mock_create_truncator, mock_file_service):
-#         """Test _truncate_and_upload_async with end user."""
-#         # Mock truncator to return truncation needed
-#         mock_truncator = Mock()
-#         mock_truncator.truncate_variable_mapping.return_value = ({"truncated": "data"}, True)
-#         mock_create_truncator.return_value = mock_truncator
-
-#         # Mock end user creation
-#         mock_end_user = Mock()
-#         mock_end_user.id = "test-user"
-#         mock_end_user.tenant_id = "test-tenant"
-#         mock_end_user_class.return_value = mock_end_user
-
-#         large_values = {"large": "x" * 10000}
-#         result = _truncate_and_upload_async(
-#             values=large_values,
-#             execution_id="test-id",
-#             type_=ExecutionOffLoadType.OUTPUTS,
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             user_data={"user_id": "test-user", "user_type": "end_user"},
-#             file_service=mock_file_service,
-#         )
-
-#         # Verify result structure
-#         assert result is not None
-#         assert result["truncated_value"] == {"truncated": "data"}
-
-#         # Verify file upload was called with end user
-#         mock_file_service.upload_file.assert_called_once()
-#         upload_call = mock_file_service.upload_file.call_args
-#         assert upload_call[1]["filename"] == "node_execution_test-id_outputs.json"
-#         assert upload_call[1]["user"] == mock_end_user
-
-
-# class TestHelperFunctions:
-#     """Test cases for helper functions."""
-
-#     @patch("tasks.workflow_node_execution_tasks.dify_config")
-#     def test_create_truncator(self, mock_config):
-#         """Test _create_truncator function."""
-#         mock_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE = 1000
-#         mock_config.WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH = 100
-#         mock_config.WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH = 500
-
-#         truncator = _create_truncator()
-
-#         # Verify truncator was created with correct config
-#         assert truncator is not None
-
-#     def test_json_encode(self):
-#         """Test _json_encode function."""
-#         test_data = {"key": "value", "number": 42}
-#         result = _json_encode(test_data)
-
-#         assert isinstance(result, str)
-#         decoded = json.loads(result)
-#         assert decoded == test_data
-
-#     def test_replace_or_append_offload_replace_existing(self):
-#         """Test _replace_or_append_offload replaces existing offload of same type."""
-#         existing_offload = WorkflowNodeExecutionOffload(
-#             id=str(uuid4()),
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             node_execution_id="test-execution",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             file_id="old-file-id",
-#         )
-
-#         new_offload = WorkflowNodeExecutionOffload(
-#             id=str(uuid4()),
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             node_execution_id="test-execution",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             file_id="new-file-id",
-#         )
-
-#         result = _replace_or_append_offload([existing_offload], new_offload)
-
-#         assert len(result) == 1
-#         assert result[0].file_id == "new-file-id"
-
-#     def test_replace_or_append_offload_append_new_type(self):
-#         """Test _replace_or_append_offload appends new offload of different type."""
-#         existing_offload = WorkflowNodeExecutionOffload(
-#             id=str(uuid4()),
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             node_execution_id="test-execution",
-#             type_=ExecutionOffLoadType.INPUTS,
-#             file_id="inputs-file-id",
-#         )
-
-#         new_offload = WorkflowNodeExecutionOffload(
-#             id=str(uuid4()),
-#             tenant_id="test-tenant",
-#             app_id="test-app",
-#             node_execution_id="test-execution",
-#             type_=ExecutionOffLoadType.OUTPUTS,
-#             file_id="outputs-file-id",
-#         )
-
-#         result = _replace_or_append_offload([existing_offload], new_offload)
-
-#         assert len(result) == 2
-#         file_ids = [offload.file_id for offload in result]
-#         assert "inputs-file-id" in file_ids
-#         assert "outputs-file-id" in file_ids
-
-
-# class TestSaveWorkflowNodeExecutionTask:
-#     """Test cases for save_workflow_node_execution_task."""
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     @patch("tasks.workflow_node_execution_tasks.select")
-#     def test_save_workflow_node_execution_task_create_new(self, mock_select, mock_sessionmaker,
-# sample_execution_data):
-#         """Test creating a new workflow node execution."""
-#         # Setup mocks
-#         mock_session = MagicMock()
-#         mock_sessionmaker.return_value.return_value.__enter__.return_value = mock_session
-#         mock_session.scalar.return_value = None  # No existing execution
-
-#         # Execute task
-#         result = save_workflow_node_execution_task(
-#             execution_data=sample_execution_data,
-#             tenant_id="test-tenant-id",
-#             app_id="test-app-id",
-#             triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
-#             creator_user_id="test-user-id",
-#             creator_user_role="account",
-#         )
-
-#         # Verify success
-#         assert result is True
-#         mock_session.add.assert_called_once()
-#         mock_session.commit.assert_called_once()
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     @patch("tasks.workflow_node_execution_tasks.select")
-#     def test_save_workflow_node_execution_task_update_existing(
-#         self, mock_select, mock_sessionmaker, sample_execution_data
-#     ):
-#         """Test updating an existing workflow node execution."""
-#         # Setup mocks
-#         mock_session = MagicMock()
-#         mock_sessionmaker.return_value.return_value.__enter__.return_value = mock_session
-
-#         existing_execution = Mock(spec=WorkflowNodeExecutionModel)
-#         mock_session.scalar.return_value = existing_execution
-
-#         # Execute task
-#         result = save_workflow_node_execution_task(
-#             execution_data=sample_execution_data,
-#             tenant_id="test-tenant-id",
-#             app_id="test-app-id",
-#             triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
-#             creator_user_id="test-user-id",
-#             creator_user_role="account",
-#         )
-
-#         # Verify success
-#         assert result is True
-#         mock_session.add.assert_not_called()  # Should not add new, just update existing
-#         mock_session.commit.assert_called_once()
-
-#     @patch("tasks.workflow_node_execution_tasks.sessionmaker")
-#     def test_save_workflow_node_execution_task_retry_on_exception(self, mock_sessionmaker, sample_execution_data):
-#         """Test task retry mechanism on exception."""
-#         # Setup mock to raise exception
-#         mock_sessionmaker.side_effect = Exception("Database error")
-
-#         # Create a mock task instance with proper retry behavior
-#         with patch.object(save_workflow_node_execution_task, "retry") as mock_retry:
-#             mock_retry.side_effect = Exception("Retry called")
-
-#             # Execute task and expect retry
-#             with pytest.raises(Exception, match="Retry called"):
-#                 save_workflow_node_execution_task(
-#                     execution_data=sample_execution_data,
-#                     tenant_id="test-tenant-id",
-#                     app_id="test-app-id",
-#                     triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
-#                     creator_user_id="test-user-id",
-#                     creator_user_role="account",
-#                 )
-
-#             # Verify retry was called
-#             mock_retry.assert_called_once()
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from unittest.mock import Mock, patch
+
+import pytest
+
+from graphon.entities.workflow_node_execution import WorkflowNodeExecution
+from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from models import Account, EndUser
+from models.enums import CreatorUserRole
+from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom
+from tasks.workflow_node_execution_tasks import (
+    _create_node_execution_from_domain,
+    _create_sqlalchemy_repository,
+    _update_node_execution_metadata,
+    save_workflow_node_execution_data_task,
+    save_workflow_node_execution_task,
+)
+
+
+def _execution(
+    *,
+    metadata: Mapping[WorkflowNodeExecutionMetadataKey, object] | None = None,
+) -> WorkflowNodeExecution:
+    if metadata is None:
+        metadata = {WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: 10}
+
+    return WorkflowNodeExecution(
+        id="exec-id",
+        node_execution_id="node-exec-id",
+        workflow_id="workflow-id",
+        workflow_execution_id="run-id",
+        index=1,
+        node_id="node-id",
+        node_type=BuiltinNodeTypes.LLM,
+        title="LLM",
+        inputs={"input": "value"},
+        process_data={"process": "value"},
+        outputs={"output": "value"},
+        status=WorkflowNodeExecutionStatus.SUCCEEDED,
+        metadata=metadata,
+        created_at=datetime.now(UTC).replace(tzinfo=None),
+        finished_at=datetime.now(UTC).replace(tzinfo=None),
+    )
+
+
+def test_create_node_execution_persists_metadata_without_data_payloads() -> None:
+    db_model = _create_node_execution_from_domain(
+        execution=_execution(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT,
+    )
+
+    assert db_model.inputs == "{}"
+    assert db_model.process_data == "{}"
+    assert db_model.outputs == "{}"
+    assert db_model.execution_metadata == '{"total_tokens": 10}'
+
+
+def test_create_node_execution_defaults_empty_metadata() -> None:
+    db_model = _create_node_execution_from_domain(
+        execution=_execution(metadata={}),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT,
+    )
+
+    assert db_model.execution_metadata == "{}"
+
+
+def test_update_node_execution_metadata_preserves_data_payloads() -> None:
+    db_model = WorkflowNodeExecutionModel()
+    db_model.inputs = '{"old_input": true}'
+    db_model.process_data = '{"old_process": true}'
+    db_model.outputs = '{"old_output": true}'
+
+    _update_node_execution_metadata(db_model, _execution())
+
+    assert db_model.inputs == '{"old_input": true}'
+    assert db_model.process_data == '{"old_process": true}'
+    assert db_model.outputs == '{"old_output": true}'
+    assert db_model.status == WorkflowNodeExecutionStatus.SUCCEEDED
+
+
+def test_update_node_execution_metadata_defaults_empty_metadata() -> None:
+    db_model = WorkflowNodeExecutionModel()
+
+    _update_node_execution_metadata(db_model, _execution(metadata={}))
+
+    assert db_model.execution_metadata == "{}"
+
+
+@patch("tasks.workflow_node_execution_tasks._create_sqlalchemy_repository")
+def test_save_workflow_node_execution_data_task_uses_sqlalchemy_repository(mock_create_repository: Mock) -> None:
+    repository = Mock()
+    mock_create_repository.return_value = repository
+    execution = _execution()
+
+    result = save_workflow_node_execution_data_task.run(
+        execution_data=execution.model_dump(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+
+    assert result is True
+    mock_create_repository.assert_called_once_with(
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+    repository.save.assert_not_called()
+    saved_data_execution = repository.save_execution_data.call_args.args[0]
+    assert saved_data_execution.model_dump() == execution.model_dump()
+
+
+@patch("tasks.workflow_node_execution_tasks._create_sqlalchemy_repository")
+def test_save_workflow_node_execution_data_task_retries_on_failure(mock_create_repository: Mock) -> None:
+    mock_create_repository.side_effect = RuntimeError("db unavailable")
+    execution = _execution()
+
+    with (
+        patch.object(
+            save_workflow_node_execution_data_task,
+            "retry",
+            side_effect=RuntimeError("retry requested"),
+        ) as retry,
+        pytest.raises(RuntimeError, match="retry requested"),
+    ):
+        save_workflow_node_execution_data_task.run(
+            execution_data=execution.model_dump(),
+            tenant_id="tenant-id",
+            app_id="app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+            creator_user_id="user-id",
+            creator_user_role=CreatorUserRole.ACCOUNT.value,
+        )
+
+    retry.assert_called_once()
+    assert isinstance(retry.call_args.kwargs["exc"], RuntimeError)
+    assert retry.call_args.kwargs["countdown"] == 60
+
+
+@patch("tasks.workflow_node_execution_tasks.session_factory.create_session")
+def test_save_workflow_node_execution_task_creates_metadata_record(mock_create_session: Mock) -> None:
+    session = _TaskSession(existing_execution=None)
+    mock_create_session.return_value = session
+
+    result = save_workflow_node_execution_task.run(
+        execution_data=_execution().model_dump(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+
+    assert result is True
+    assert session.committed is True
+    assert isinstance(session.added_execution, WorkflowNodeExecutionModel)
+    assert session.added_execution.inputs == "{}"
+    assert session.added_execution.process_data == "{}"
+    assert session.added_execution.outputs == "{}"
+
+
+@patch("tasks.workflow_node_execution_tasks.session_factory.create_session")
+def test_save_workflow_node_execution_task_updates_metadata_without_payloads(mock_create_session: Mock) -> None:
+    existing_execution = WorkflowNodeExecutionModel()
+    existing_execution.inputs = '{"old_input": true}'
+    existing_execution.process_data = '{"old_process": true}'
+    existing_execution.outputs = '{"old_output": true}'
+    session = _TaskSession(existing_execution=existing_execution)
+    mock_create_session.return_value = session
+
+    result = save_workflow_node_execution_task.run(
+        execution_data=_execution().model_dump(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+
+    assert result is True
+    assert session.committed is True
+    assert session.added_execution is None
+    assert existing_execution.inputs == '{"old_input": true}'
+    assert existing_execution.process_data == '{"old_process": true}'
+    assert existing_execution.outputs == '{"old_output": true}'
+    assert existing_execution.status == WorkflowNodeExecutionStatus.SUCCEEDED
+
+
+@patch("tasks.workflow_node_execution_tasks.session_factory.create_session")
+def test_save_workflow_node_execution_task_ignores_stale_nonterminal_snapshot(mock_create_session: Mock) -> None:
+    existing_execution = WorkflowNodeExecutionModel()
+    existing_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
+    existing_execution.finished_at = datetime(2026, 1, 1)
+    session = _TaskSession(existing_execution=existing_execution)
+    mock_create_session.return_value = session
+    execution = _execution()
+    execution.status = WorkflowNodeExecutionStatus.RUNNING
+    execution.finished_at = None
+
+    result = save_workflow_node_execution_task.run(
+        execution_data=execution.model_dump(),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT.value,
+    )
+
+    assert result is True
+    assert session.committed is False
+    assert existing_execution.status == WorkflowNodeExecutionStatus.SUCCEEDED
+
+
+def test_create_sqlalchemy_repository_builds_account_context(monkeypatch) -> None:
+    account = Mock()
+    session = _Session({Account: account})
+
+    def session_maker():
+        return session
+
+    monkeypatch.setattr(
+        "tasks.workflow_node_execution_tasks.session_factory.get_session_maker",
+        lambda: session_maker,
+    )
+
+    with patch(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository"
+    ) as repository_class:
+        repository = _create_sqlalchemy_repository(
+            tenant_id="tenant-id",
+            app_id="",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+            creator_user_id="user-id",
+            creator_user_role=CreatorUserRole.ACCOUNT.value,
+        )
+
+    assert repository == repository_class.return_value
+    account.set_tenant_id.assert_called_once_with("tenant-id")
+    repository_class.assert_called_once_with(
+        session_factory=session_maker,
+        user=account,
+        app_id=None,
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+
+
+def test_create_sqlalchemy_repository_builds_end_user_context(monkeypatch) -> None:
+    end_user = Mock()
+    session = _Session({EndUser: end_user})
+
+    def session_maker():
+        return session
+
+    monkeypatch.setattr(
+        "tasks.workflow_node_execution_tasks.session_factory.get_session_maker",
+        lambda: session_maker,
+    )
+
+    with patch(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository"
+    ) as repository_class:
+        repository = _create_sqlalchemy_repository(
+            tenant_id="tenant-id",
+            app_id="app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+            creator_user_id="user-id",
+            creator_user_role=CreatorUserRole.END_USER.value,
+        )
+
+    assert repository == repository_class.return_value
+    repository_class.assert_called_once_with(
+        session_factory=session_maker,
+        user=end_user,
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+
+
+def test_create_sqlalchemy_repository_raises_for_missing_creator(monkeypatch) -> None:
+    session = _Session({})
+
+    def session_maker():
+        return session
+
+    monkeypatch.setattr(
+        "tasks.workflow_node_execution_tasks.session_factory.get_session_maker",
+        lambda: session_maker,
+    )
+
+    with (
+        patch(
+            "core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository"
+        ),
+        pytest.raises(ValueError, match="Creator user missing-user not found"),
+    ):
+        _create_sqlalchemy_repository(
+            tenant_id="tenant-id",
+            app_id="app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+            creator_user_id="missing-user",
+            creator_user_role=CreatorUserRole.ACCOUNT.value,
+        )
+
+
+class _Session:
+    def __init__(self, users: dict[type, object]) -> None:
+        self._users = users
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        return None
+
+    def get(self, model, _id: str):
+        return self._users.get(model)
+
+    def scalar(self, _stmt):
+        return self._users.get(EndUser)
+
+
+class _TaskSession:
+    def __init__(self, existing_execution: WorkflowNodeExecutionModel | None) -> None:
+        self._existing_execution = existing_execution
+        self.added_execution: WorkflowNodeExecutionModel | None = None
+        self.committed = False
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        return None
+
+    def scalar(self, _stmt):
+        return self._existing_execution
+
+    def add(self, execution: WorkflowNodeExecutionModel) -> None:
+        self.added_execution = execution
+
+    def commit(self) -> None:
+        self.committed = True

docker/envs/core-services/shared.env.example

@@ -360,7 +360,6 @@ BAIDU_VECTOR_DB_AUTO_BUILD_ROW_COUNT_INCREMENT_RATIO=0.05
 BAIDU_VECTOR_DB_REBUILD_INDEX_TIMEOUT_IN_SECONDS=300
 HUAWEI_CLOUD_HOSTS=https://127.0.0.1:9200
 HUAWEI_CLOUD_USER=admin
-WORKFLOW_NODE_EXECUTION_STORAGE=rdbms
 CORE_WORKFLOW_EXECUTION_REPOSITORY=core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository
 CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY=core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository
 API_WORKFLOW_RUN_REPOSITORY=repositories.sqlalchemy_api_workflow_run_repository.DifyAPISQLAlchemyWorkflowRunRepository