fix(workflow): report aborted runs as stopped

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()

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.",

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] = []
 

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"