fix: workflow_finish_to_stream_response assert exception with celery … (#24674)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

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

@@ -144,6 +144,7 @@ class AdvancedChatAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._task_state = WorkflowTaskState()

api/core/app/apps/common/workflow_response_converter.py

@@ -3,7 +3,6 @@ from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime
 from typing import Any, Optional, Union, cast
 
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
@@ -53,9 +52,7 @@ from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import (
     Account,
-    CreatorUserRole,
     EndUser,
-    WorkflowRun,
 )
 
 
@@ -64,8 +61,10 @@ class WorkflowResponseConverter:
         self,
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
+        user: Union[Account, EndUser],
     ) -> None:
         self._application_generate_entity = application_generate_entity
+        self._user = user
 
     def workflow_start_to_stream_response(
         self,
@@ -92,27 +91,21 @@ class WorkflowResponseConverter:
         workflow_execution: WorkflowExecution,
     ) -> WorkflowFinishStreamResponse:
         created_by = None
-        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
+
-        assert workflow_run is not None
+        user = self._user
-        if workflow_run.created_by_role == CreatorUserRole.ACCOUNT:
+        if isinstance(user, Account):
-            stmt = select(Account).where(Account.id == workflow_run.created_by)
+            created_by = {
-            account = session.scalar(stmt)
+                "id": user.id,
-            if account:
+                "name": user.name,
-                created_by = {
+                "email": user.email,
-                    "id": account.id,
+            }
-                    "name": account.name,
+        elif isinstance(user, EndUser):
-                    "email": account.email,
+            created_by = {
-                }
+                "id": user.id,
-        elif workflow_run.created_by_role == CreatorUserRole.END_USER:
+                "user": user.session_id,
-            stmt = select(EndUser).where(EndUser.id == workflow_run.created_by)
+            }
-            end_user = session.scalar(stmt)
-            if end_user:
-                created_by = {
-                    "id": end_user.id,
-                    "user": end_user.session_id,
-                }
         else:
-            raise NotImplementedError(f"unknown created_by_role: {workflow_run.created_by_role}")
+            raise NotImplementedError(f"User type not supported: {type(user)}")
 
         # Handle the case where finished_at is None by using current time as default
         finished_at_timestamp = (

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

@@ -131,6 +131,7 @@ class WorkflowAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._application_generate_entity = application_generate_entity