diff --git a/api/core/app/apps/workflow/generate_task_pipeline.py b/api/core/app/apps/workflow/generate_task_pipeline.py
index b3112f3e8..74af464c3 100644
--- a/api/core/app/apps/workflow/generate_task_pipeline.py
+++ b/api/core/app/apps/workflow/generate_task_pipeline.py
@@ -130,6 +130,10 @@ class WorkflowAppGenerateTaskPipeline:
             workflow_node_execution_repository=workflow_node_execution_repository,
         )
 
+        # 将用户信息传递给工作流周期管理器
+        self._workflow_cycle_manager._user_id = self._user_id
+        self._workflow_cycle_manager._created_by_role = self._created_by_role
+
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
             user=user,
@@ -669,10 +673,10 @@ class WorkflowAppGenerateTaskPipeline:
         if isinstance(
             event,
             (
-                QueueNodeFailedEvent,
-                QueueNodeInIterationFailedEvent,
-                QueueNodeInLoopFailedEvent,
-                QueueNodeExceptionEvent,
+                    QueueNodeFailedEvent,
+                    QueueNodeInIterationFailedEvent,
+                    QueueNodeInLoopFailedEvent,
+                    QueueNodeExceptionEvent,
             ),
         ):
             yield from self._handle_node_failed_events(
diff --git a/api/core/workflow/workflow_cycle_manager.py b/api/core/workflow/workflow_cycle_manager.py
index 503273f56..b4c6ff7c2 100644
--- a/api/core/workflow/workflow_cycle_manager.py
+++ b/api/core/workflow/workflow_cycle_manager.py
@@ -202,8 +202,10 @@ class WorkflowCycleManager:
         domain_execution_dict = jsonable_encoder(domain_execution)
 
         # 添加用户信息到字典中
-        domain_execution_dict['created_by'] = self._application_generate_entity.user_id
-        domain_execution_dict['created_by_role'] = 'account'  # 工作流通常由账户用户调用
+        domain_execution_dict['created_by'] = getattr(self, '_user_id', self._application_generate_entity.user_id)
+        domain_execution_dict['created_by_role'] = getattr(self, '_created_by_role', None)
+        if domain_execution_dict['created_by_role']:
+            domain_execution_dict['created_by_role'] = domain_execution_dict['created_by_role'].value
 
         update_account_money_when_workflow_node_execution_created_extend.delay(domain_execution_dict)
         # 二开部分End - 额度限制