From 87bbcc4e3ba21433f7793dc40f40311e72d58493 Mon Sep 17 00:00:00 2001 From: "yunlu.wen" Date: Tue, 21 Apr 2026 17:18:11 +0800 Subject: [PATCH] improve workflow stop event handling --- api/core/app/apps/workflow_app_runner.py | 2 +- api/services/app_task_service.py | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/api/core/app/apps/workflow_app_runner.py b/api/core/app/apps/workflow_app_runner.py index 047b54c86c..c7469f6d6f 100644 --- a/api/core/app/apps/workflow_app_runner.py +++ b/api/core/app/apps/workflow_app_runner.py @@ -410,7 +410,7 @@ class WorkflowBasedAppRunner: elif isinstance(event, GraphRunFailedEvent): self._publish_event(QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)) elif isinstance(event, GraphRunAbortedEvent): - self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0)) + self._publish_event(QueueWorkflowPartialSuccessEvent(outputs={}, exceptions_count=0)) elif isinstance(event, GraphRunPausedEvent): runtime_state = workflow_entry.graph_engine.graph_runtime_state paused_nodes = runtime_state.get_paused_nodes() diff --git a/api/services/app_task_service.py b/api/services/app_task_service.py index 6e9d6b1c73..0696b0975e 100644 --- a/api/services/app_task_service.py +++ b/api/services/app_task_service.py @@ -37,10 +37,13 @@ class AppTaskService: Returns: None """ - # Legacy mechanism: Set stop flag in Redis - AppQueueManager.set_stop_flag(task_id, invoke_from, user_id) - # New mechanism: Send stop command via GraphEngine for workflow-based apps # This ensures proper workflow status recording in the persistence layer if app_mode in (AppMode.ADVANCED_CHAT, AppMode.WORKFLOW): + # Let the event handler process the Graphon abort event instead of + # stopping the queue listener immediately. Otherwise, events may be + # lost and the workflow run can remain stuck in the running state. GraphEngineManager(redis_client).send_stop_command(task_id) + else: + # Legacy mechanism: Set stop flag in Redis + AppQueueManager.set_stop_flag(task_id, invoke_from, user_id)