refactor(workflow_cycle_manager): Refactor `_workflow_parallel_branch_finished_to_stream_response`

Signed-off-by: -LAN- <laipz8200@outlook.com>

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

@@ -403,15 +403,10 @@ class AdvancedChatAppGenerateTaskPipeline:
                 if not self._workflow_run_id:
                     raise ValueError("workflow run not initialized.")
 
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
                 parallel_finish_resp = (
                     self._workflow_cycle_manager._workflow_parallel_branch_finished_to_stream_response(
-                            session=session,
                         task_id=self._application_generate_entity.task_id,
-                            workflow_run=workflow_run,
+                        workflow_execution_id=self._workflow_run_id,
                         event=event,
                     )
                 )

api/core/workflow/workflow_app_generate_task_pipeline.py

@@ -356,15 +356,10 @@ class WorkflowAppGenerateTaskPipeline:
                 if not self._workflow_run_id:
                     raise ValueError("workflow run not initialized.")
 
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
                 parallel_finish_resp = (
                     self._workflow_cycle_manager._workflow_parallel_branch_finished_to_stream_response(
-                            session=session,
                         task_id=self._application_generate_entity.task_id,
-                            workflow_run=workflow_run,
+                        workflow_execution_id=self._workflow_run_id,
                         event=event,
                     )
                 )

api/core/workflow/workflow_cycle_manager.py

@@ -654,15 +654,13 @@ class WorkflowCycleManager:
     def _workflow_parallel_branch_finished_to_stream_response(
         self,
         *,
-        session: Session,
         task_id: str,
-        workflow_run: WorkflowRun,
+        workflow_execution_id: str,
         event: QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent,
     ) -> ParallelBranchFinishedStreamResponse:
-        _ = session
         return ParallelBranchFinishedStreamResponse(
             task_id=task_id,
-            workflow_run_id=workflow_run.id,
+            workflow_run_id=workflow_execution_id,
             data=ParallelBranchFinishedStreamResponse.Data(
                 parallel_id=event.parallel_id,
                 parallel_branch_id=event.parallel_start_node_id,