refactor(models): pass session into WorkflowNodeExecutionModel accessors

api/controllers/console/app/workflow.py

@@ -77,7 +77,10 @@
 from fields.base import ResponseModel
 from fields.conversation_variable_fields import WorkflowConversationVariableResponse
 from fields.member_fields import SimpleAccount
-from fields.workflow_run_fields import WorkflowRunNodeExecutionResponse
+from fields.workflow_run_fields import (
+    WorkflowRunNodeExecutionResponse,
+    node_execution_response_source,
+)
 from graphon.enums import NodeType
 from graphon.file import File
 from graphon.file import helpers as file_helpers
@@ -1260,7 +1263,7 @@ def post(self, payload: DraftWorkflowNodeRunPayload, current_user: Account, app_
         )
 
         return WorkflowRunNodeExecutionResponse.model_validate(
-            workflow_node_execution, from_attributes=True
+            node_execution_response_source(workflow_node_execution, session=db.session()), from_attributes=True
         ).model_dump(mode="json")
 
 
@@ -1680,7 +1683,9 @@ def get(self, app_model: App, node_id: str):
         )
         if node_exec is None:
             raise NotFound("last run not found")
-        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            node_execution_response_source(node_exec, session=db.session()), from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/run")

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -53,6 +53,7 @@
     WorkflowRunNodeExecutionListResponse,
     WorkflowRunNodeExecutionResponse,
     WorkflowRunPaginationResponse,
+    node_execution_response_source,
 )
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
@@ -506,7 +507,7 @@ def post(self, req_data: NodeRunRequiredPayload, current_user: Account, pipeline
             raise ValueError("Workflow node execution not found")
 
         return WorkflowRunNodeExecutionResponse.model_validate(
-            workflow_node_execution, from_attributes=True
+            node_execution_response_source(workflow_node_execution, session=db.session()), from_attributes=True
         ).model_dump(mode="json")
 
 
@@ -1023,7 +1024,9 @@ def get(self, pipeline: Pipeline, node_id: str):
         )
         if node_exec is None:
             raise NotFound("last run not found")
-        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            node_execution_response_source(node_exec, session=db.session()), from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/transform/datasets/<uuid:dataset_id>")
@@ -1085,7 +1088,7 @@ def post(self, req_data: DatasourceVariablesPayload, current_user: Account, pipe
             current_user=current_user,
         )
         return WorkflowRunNodeExecutionResponse.model_validate(
-            workflow_node_execution, from_attributes=True
+            node_execution_response_source(workflow_node_execution, session=db.session()), from_attributes=True
         ).model_dump(mode="json")
 
 

api/controllers/console/snippets/snippet_workflow.py

@@ -51,6 +51,7 @@
     WorkflowRunNodeExecutionListResponse,
     WorkflowRunNodeExecutionResponse,
     WorkflowRunPaginationResponse,
+    node_execution_response_source,
 )
 from graphon.graph_engine.manager import GraphEngineManager
 from libs import helper
@@ -635,7 +636,7 @@ def post(
         )
 
         return WorkflowRunNodeExecutionResponse.model_validate(
-            workflow_node_execution, from_attributes=True
+            node_execution_response_source(workflow_node_execution, session=db.session()), from_attributes=True
         ).model_dump(mode="json")
 
 
@@ -672,7 +673,9 @@ def get(self, snippet: CustomizedSnippet, node_id: str):
         if node_exec is None:
             raise NotFound("Node last run not found")
 
-        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            node_execution_response_source(node_exec, session=db.session()), from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/iteration/nodes/<string:node_id>/run")

api/fields/workflow_run_fields.py

@@ -7,11 +7,13 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
 
 from flask_restx import Namespace, fields
 from pydantic import AliasChoices, Field, field_validator
+from sqlalchemy.orm import Session
 
 from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
@@ -187,5 +189,28 @@ def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
         return to_timestamp(value)
 
 
+@dataclass(frozen=True)
+class WorkflowNodeExecutionResponseSource:
+    """Expose session-backed node-execution accessors during response validation."""
+
+    node_execution: Any
+    session: Session
+
+    @property
+    def created_by_account(self) -> Any:
+        return self.node_execution.created_by_account(self.session)
+
+    @property
+    def created_by_end_user(self) -> Any:
+        return self.node_execution.created_by_end_user(self.session)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self.node_execution, name)  # guard-ignore: no-new-getattr -- delegates model fields
+
+
+def node_execution_response_source(node_execution: Any, *, session: Session) -> WorkflowNodeExecutionResponseSource:
+    return WorkflowNodeExecutionResponseSource(node_execution=node_execution, session=session)
+
+
 class WorkflowRunNodeExecutionListResponse(ResponseModel):
     data: list[WorkflowRunNodeExecutionResponse]

api/models/workflow.py

@@ -1100,22 +1100,20 @@ def preload_offload_data_and_files(
             )
         )
 
-    @property
-    def created_by_account(self):
+    def created_by_account(self, session: orm.Session) -> Account | None:
         created_by_role = CreatorUserRole(self.created_by_role)
         if created_by_role == CreatorUserRole.ACCOUNT:
             stmt = select(Account).where(Account.id == self.created_by)
-            return db.session.scalar(stmt)
+            return session.scalar(stmt)
         return None
 
-    @property
-    def created_by_end_user(self):
+    def created_by_end_user(self, session: orm.Session):
         from .model import EndUser
 
         created_by_role = CreatorUserRole(self.created_by_role)
         if created_by_role == CreatorUserRole.END_USER:
             stmt = select(EndUser).where(EndUser.id == self.created_by)
-            return db.session.scalar(stmt)
+            return session.scalar(stmt)
         return None
 
     @property

api/services/rag_pipeline/rag_pipeline.py

@@ -1266,7 +1266,9 @@ def get_rag_pipeline_workflow_run_node_executions(
             app_id=pipeline.id,
             workflow_run_id=run_id,
         )
-        return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo)
+        return assemble_workflow_node_execution_traces(
+            node_executions, self._node_execution_service_repo, session=self._session
+        )
 
     @staticmethod
     def publish_customized_pipeline_template(

api/services/snippet_service.py

@@ -989,7 +989,10 @@ def get_snippet_workflow_run_node_executions(
             workflow_run_id=workflow_run.id,
         )
 
-        return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo)
+        with self._session_scope() as session:
+            return assemble_workflow_node_execution_traces(
+                node_executions, self._node_execution_service_repo, session=session
+            )
 
     # --- Node Execution Operations ---
 

api/services/workflow_node_execution_trace_service.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 from pydantic import BaseModel, ConfigDict, ValidationError
+from sqlalchemy.orm import Session
 
 from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
 from libs.helper import to_timestamp
@@ -58,23 +59,27 @@ class WorkflowNodeExecutionTrace(BaseModel):
 def assemble_workflow_node_execution_traces(
     executions: Sequence[WorkflowNodeExecutionModel],
     repository: DifyAPIWorkflowNodeExecutionRepository,
+    *,
+    session: Session,
 ) -> list[WorkflowNodeExecutionTrace]:
     """Expand valid persisted retry attempts before each terminal execution."""
     traces: list[WorkflowNodeExecutionTrace] = []
     for execution in executions:
-        traces.extend(_expand_execution(execution, repository))
+        traces.extend(_expand_execution(execution, repository, session=session))
     return traces
 
 
 def _expand_execution(
     execution: WorkflowNodeExecutionModel,
     repository: DifyAPIWorkflowNodeExecutionRepository,
+    *,
+    session: Session,
 ) -> list[WorkflowNodeExecutionTrace]:
     full_process_data = _load_full_process_data(execution, repository)
     retry_attempts = _parse_retry_attempts(execution, full_process_data)
     terminal_metadata = execution.execution_metadata_dict
-    traces = [_retry_trace(execution, attempt, terminal_metadata) for attempt in retry_attempts]
-    traces.append(_terminal_trace(execution))
+    traces = [_retry_trace(execution, attempt, terminal_metadata, session=session) for attempt in retry_attempts]
+    traces.append(_terminal_trace(execution, session=session))
     return traces
 
 
@@ -133,6 +138,8 @@ def _retry_trace(
     execution: WorkflowNodeExecutionModel,
     attempt: WorkflowNodeRetryAttempt,
     terminal_metadata: Mapping[str, Any],
+    *,
+    session: Session,
 ) -> WorkflowNodeExecutionTrace:
     truncator = VariableTruncator.default()
     inputs, inputs_truncated = truncator.truncate_variable_mapping(attempt.inputs)
@@ -163,8 +170,8 @@ def _retry_trace(
         created_at=attempt.created_at,
         created_by_role=_enum_value(execution.created_by_role),
         created_by=execution.created_by,
-        created_by_account=execution.created_by_account,
-        created_by_end_user=execution.created_by_end_user,
+        created_by_account=execution.created_by_account(session),
+        created_by_end_user=execution.created_by_end_user(session),
         finished_at=attempt.finished_at,
         inputs_truncated=inputs_truncated,
         outputs_truncated=outputs_truncated,
@@ -173,7 +180,7 @@ def _retry_trace(
     )
 
 
-def _terminal_trace(execution: WorkflowNodeExecutionModel) -> WorkflowNodeExecutionTrace:
+def _terminal_trace(execution: WorkflowNodeExecutionModel, *, session: Session) -> WorkflowNodeExecutionTrace:
     process_data = execution.process_data_dict
     if process_data is not None and RETRY_HISTORY_PROCESS_DATA_KEY in process_data:
         process_data = dict(process_data)
@@ -202,8 +209,8 @@ def _terminal_trace(execution: WorkflowNodeExecutionModel) -> WorkflowNodeExecut
         created_at=to_timestamp(execution.created_at),
         created_by_role=_enum_value(execution.created_by_role),
         created_by=execution.created_by,
-        created_by_account=execution.created_by_account,
-        created_by_end_user=execution.created_by_end_user,
+        created_by_account=execution.created_by_account(session),
+        created_by_end_user=execution.created_by_end_user(session),
         finished_at=to_timestamp(execution.finished_at),
         inputs_truncated=execution.inputs_truncated,
         outputs_truncated=execution.outputs_truncated,

api/services/workflow_run_service.py

@@ -5,6 +5,7 @@
 
 import contexts
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
+from extensions.ext_database import db
 from graphon.enums import WorkflowExecutionStatus
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from machinery.context import RequestContext
@@ -202,7 +203,7 @@ def get_workflow_run_node_executions(
             app_id=app_id,
             workflow_run_id=run_id,
         )
-        return assemble_workflow_node_execution_traces(node_executions, self._node_executions)
+        return assemble_workflow_node_execution_traces(node_executions, self._node_executions, session=db.session())
 
     def get_pause_details(
         self,

api/tests/test_containers_integration_tests/models/test_workflow_node_execution_model.py

@@ -109,7 +109,7 @@ def test_created_by_account_returns_account_when_role_is_account(self, db_sessio
             created_by=account.id,
         )
 
-        result = execution.created_by_account
+        result = execution.created_by_account(session=db_session_with_containers)
 
         assert result is not None
         assert result.id == account.id
@@ -126,7 +126,7 @@ def test_created_by_account_returns_none_when_role_is_end_user(self, db_session_
             created_by=account.id,
         )
 
-        result = execution.created_by_account
+        result = execution.created_by_account(session=db_session_with_containers)
 
         assert result is None
 
@@ -146,7 +146,7 @@ def test_created_by_end_user_returns_end_user_when_role_is_end_user(
             created_by=end_user.id,
         )
 
-        result = execution.created_by_end_user
+        result = execution.created_by_end_user(session=db_session_with_containers)
 
         assert result is not None
         assert result.id == end_user.id
@@ -165,6 +165,6 @@ def test_created_by_end_user_returns_none_when_role_is_account(self, db_session_
             created_by=end_user.id,
         )
 
-        result = execution.created_by_end_user
+        result = execution.created_by_end_user(session=db_session_with_containers)
 
         assert result is None

api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py

@@ -15,6 +15,7 @@
 from controllers.common.errors import NotFoundError
 from controllers.console.app import workflow_run as workflow_run_module
 from extensions.ext_database import db
+from fields.workflow_run_fields import node_execution_response_source
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus
 from machinery.context import RequestContext
 from models import Account, App, AppMode
@@ -348,7 +349,9 @@ def test_workflow_run_node_executions_return_frontend_trace_contract(
     _account(sqlite_session)
     execution = _workflow_run_node_execution(sqlite_session)
     workflow_runs = Mock()
-    workflow_runs.get_workflow_run_node_executions.return_value = [execution]
+    workflow_runs.get_workflow_run_node_executions.return_value = [
+        node_execution_response_source(execution, session=sqlite_session)
+    ]
     _mock_application_services(monkeypatch, workflow_runs)
     monkeypatch.setattr(db, "session", sqlite_session)
     request_context = _request_context()

api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow_apis.py

@@ -45,6 +45,7 @@
     WorkflowListQuery,
     WorkflowUpdatePayload,
 )
+from fields.workflow_run_fields import node_execution_response_source
 from graphon.enums import WorkflowNodeExecutionStatus
 from libs.datetime_utils import naive_utc_now
 from models.account import Account, TenantAccountRole
@@ -802,8 +803,12 @@ def test_get_node_executions_passes_current_user(self, app: Flask) -> None:
         run_id = uuid4()
         node_exec = make_node_execution(workflow_run_id=str(run_id))
 
+        session_stub = MagicMock()
+        session_stub.scalar.return_value = None
         service = MagicMock()
-        service.get_rag_pipeline_workflow_run_node_executions.return_value = [node_exec]
+        service.get_rag_pipeline_workflow_run_node_executions.return_value = [
+            node_execution_response_source(node_exec, session=session_stub)
+        ]
 
         with (
             app.test_request_context("/"),