diff --git a/api/core/app/apps/workflow_app_runner.py b/api/core/app/apps/workflow_app_runner.py index 01dcaef10db..3d2857f130a 100644 --- a/api/core/app/apps/workflow_app_runner.py +++ b/api/core/app/apps/workflow_app_runner.py @@ -26,6 +26,7 @@ from core.app.entities.queue_entities import ( QueueNodeSucceededEvent, QueueReasoningChunkEvent, QueueRetrieverResourcesEvent, + QueueStopEvent, QueueTextChunkEvent, QueueWorkflowFailedEvent, QueueWorkflowPartialSuccessEvent, @@ -424,7 +425,12 @@ class WorkflowBasedAppRunner: QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count) ) case GraphRunAbortedEvent(): - self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0)) + self._publish_event( + QueueStopEvent( + stopped_by=QueueStopEvent.StopBy.USER_MANUAL, + reason=event.reason or "Workflow execution aborted", + ) + ) case GraphRunPausedEvent(): runtime_state = workflow_entry.graph_engine.graph_runtime_state paused_nodes = runtime_state.get_paused_nodes() diff --git a/api/core/app/entities/queue_entities.py b/api/core/app/entities/queue_entities.py index 6bbffbbb160..88a3cf6e550 100644 --- a/api/core/app/entities/queue_entities.py +++ b/api/core/app/entities/queue_entities.py @@ -500,11 +500,15 @@ class QueueStopEvent(AppQueueEvent): event: QueueEvent = QueueEvent.STOP stopped_by: StopBy + reason: str | None = None def get_stop_reason(self) -> str: """ To stop reason """ + if self.reason: + return self.reason + reason_mapping = { QueueStopEvent.StopBy.USER_MANUAL: "Stopped by user.", QueueStopEvent.StopBy.ANNOTATION_REPLY: "Stopped by annotation reply.", diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py index 0c297d50a01..fd643893f69 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py +++ b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py @@ -17,6 +17,7 @@ from core.app.entities.queue_entities import ( QueueNodeRetryEvent, QueueNodeSucceededEvent, QueueReasoningChunkEvent, + QueueStopEvent, QueueTextChunkEvent, QueueWorkflowPausedEvent, QueueWorkflowStartedEvent, @@ -27,6 +28,7 @@ from core.workflow.system_variables import default_system_variables from graphon.entities.pause_reason import HitlRequired from graphon.enums import BuiltinNodeTypes from graphon.graph_events import ( + GraphRunAbortedEvent, GraphRunPausedEvent, GraphRunStartedEvent, GraphRunSucceededEvent, @@ -373,6 +375,23 @@ class TestWorkflowBasedAppRunner: assert paused_event.paused_nodes == ["node-1"] assert emails + def test_handle_graph_aborted_publishes_stopped_terminal(self): + published: list[object] = [] + + class _QueueManager: + def publish(self, event, publish_from): + del publish_from + published.append(event) + + runner = WorkflowBasedAppRunner(queue_manager=_QueueManager(), app_id="app") + workflow_entry = SimpleNamespace() + + runner._handle_event(workflow_entry, GraphRunAbortedEvent(reason="User requested stop", outputs={})) + + event = published[-1] + assert isinstance(event, QueueStopEvent) + assert event.get_stop_reason() == "User requested stop" + def test_handle_node_events_publishes_queue_events(self): published: list[object] = [] diff --git a/api/tests/unit_tests/core/app/entities/test_queue_entities.py b/api/tests/unit_tests/core/app/entities/test_queue_entities.py index a930d7325d4..43d69023e70 100644 --- a/api/tests/unit_tests/core/app/entities/test_queue_entities.py +++ b/api/tests/unit_tests/core/app/entities/test_queue_entities.py @@ -6,6 +6,13 @@ class TestQueueEntities: event = QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL) assert event.get_stop_reason() == "Stopped by user." + def test_get_stop_reason_prefers_explicit_reason(self): + event = QueueStopEvent( + stopped_by=QueueStopEvent.StopBy.USER_MANUAL, + reason="Workflow execution timed out", + ) + assert event.get_stop_reason() == "Workflow execution timed out" + def test_get_stop_reason_for_unknown_stop_by(self): event = QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL) event.stopped_by = "unknown"