fix: tell a stopped run apart from one the service killed

asyncio delivers a user pressing stop and a server-imposed execution ceiling
as the same CancelledError, and the flow span reported both as cancelled with
OTel status UNSET. That kept every timeout out of error-rate alerting, while
the client was served a terminal error and the job row was written FAILED.
The span was the only place claiming nothing had gone wrong.

The two are now separate outcomes. A user stop stays cancelled with status
UNSET, because a withdrawn request is not a service fault. Anything else that
cancels a run reports aborted with span status ERROR, which is what the rest
of the system already says about it.

The discriminator already existed: the producers stamp a marker on the
exception args, and the job service reads it to choose CANCELLED over FAILED.
It was a bare string repeated at five sites, so the span could not have
trusted it. It is one constant now, and it lives in lfx.constants rather than
next to the graph exceptions: importing lfx.graph.exceptions from a langflow
service module runs lfx/graph/__init__, which imports Graph, and that made
langflow.main fail to import at all. lfx.constants imports nothing but
pathlib.

Author: ogabrielluiz

src/backend/base/langflow/services/background_execution/runner.py

@@ -22,6 +22,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any
 
+from lfx.constants import USER_CANCELLED_MESSAGE
 from lfx.log.logger import logger
 
 from langflow.services.background_execution.live_bus import LiveFrame
@@ -122,7 +123,7 @@ async def _wrapped() -> None:
                 paused = True
                 await logger.adebug(f"Background job {job_id} suspended for human input")
         except asyncio.CancelledError as exc:
-            user_tagged = bool(exc.args) and exc.args[0] == "LANGFLOW_USER_CANCELLED"
+            user_tagged = bool(exc.args) and exc.args[0] == USER_CANCELLED_MESSAGE
             if not user_tagged and not await self._stop_requested(job_id):
                 raise
             await logger.adebug(f"Background job {job_id} stopped")
@@ -319,7 +320,7 @@ async def _stop_requested(self, job_id: UUID) -> bool:
     @staticmethod
     def _user_cancelled() -> asyncio.CancelledError:
         exc = asyncio.CancelledError()
-        exc.args = ("LANGFLOW_USER_CANCELLED",)
+        exc.args = (USER_CANCELLED_MESSAGE,)
         return exc
 
     @staticmethod

src/backend/base/langflow/services/job_queue/service.py

@@ -8,6 +8,7 @@
 if TYPE_CHECKING:
     from collections.abc import Coroutine
 
+from lfx.constants import USER_CANCELLED_MESSAGE
 from lfx.log.logger import logger
 
 from langflow.events.event_manager import EventManager
@@ -456,7 +457,7 @@ async def cleanup_job(self, job_id: str) -> None:
                     # User-initiated cancellation so we explicitly called task.cancel() above
                     await logger.adebug(f"Task for job_id {job_id} was successfully cancelled.")
                     # Re-raise with user cancellation message code
-                    exc.args = ("LANGFLOW_USER_CANCELLED",)
+                    exc.args = (USER_CANCELLED_MESSAGE,)
                     raise
                 # System-initiated cancellation for other reasons
                 await logger.adebug(f"Task for job_id {job_id} was cancelled by system.")

src/backend/base/langflow/services/jobs/service.py

@@ -11,6 +11,7 @@
 from datetime import datetime, timezone
 from uuid import UUID, uuid4
 
+from lfx.constants import USER_CANCELLED_MESSAGE
 from lfx.graph.exceptions import GraphPausedException
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlmodel import col, func, select
@@ -857,7 +858,7 @@ async def execute_with_status(self, job_id: UUID, run_coro_func, *args, **kwargs
 
         except asyncio.CancelledError as exc:
             # Check the message code to determine if this was user-initiated or system-initiated
-            if exc.args and exc.args[0] == "LANGFLOW_USER_CANCELLED":
+            if exc.args and exc.args[0] == USER_CANCELLED_MESSAGE:
                 # User-initiated cancellation, update status to CANCELLED
                 await logger.awarning(f"Job {job_id} was cancelled by user")
                 await self.update_job_status(job_id, JobStatus.CANCELLED, finished_timestamp=True)

src/backend/base/langflow/services/task/service.py

@@ -5,6 +5,8 @@
 from typing import TYPE_CHECKING, Any
 from uuid import UUID, uuid4
 
+from lfx.constants import USER_CANCELLED_MESSAGE
+
 from langflow.exceptions.api import WorkflowResourceError, WorkflowServiceUnavailableError
 from langflow.services.base import Service
 from langflow.services.deps import get_queue_service
@@ -98,7 +100,7 @@ async def revoke_task(self, task_id: UUID | str) -> bool:
         try:
             await job_queue_service.cleanup_job(str(task_id))
         except asyncio.CancelledError as e:
-            if str(e) != "LANGFLOW_USER_CANCELLED":
+            if str(e) != USER_CANCELLED_MESSAGE:
                 raise
             return True
         return True

src/backend/tests/unit/services/telemetry/test_flow_execution_span.py

@@ -203,17 +203,20 @@ async def main():
 """
 )
 
-# CancelledError is a BaseException, so an `except Exception` handler never sees it. A user
-# pressing stop and the v2 execution ceiling both cancel the driver this span wraps.
+# CancelledError is a BaseException, so an `except Exception` handler never sees it. Both a user
+# pressing stop and a server-imposed ceiling arrive as this one type, and they mean opposite
+# things to an operator, so the span has to tell them apart.
 CANCELLED_PROBE = (
     PROVIDER_SETUP
     + """
+from lfx.constants import USER_CANCELLED_MESSAGE
+
 async def main():
     graph = build_graph()
     raised = False
     try:
         with graph.flow_execution_span():
-            raise asyncio.CancelledError()
+            raise asyncio.CancelledError(USER_CANCELLED_MESSAGE)
     except asyncio.CancelledError:
         raised = True
     report({"raised": raised})
@@ -222,6 +225,28 @@ async def main():
 """
 )
 
+# A wall-clock ceiling, exactly as asyncio.wait_for delivers it: an untagged CancelledError.
+ABORTED_PROBE = (
+    PROVIDER_SETUP
+    + """
+async def main():
+    graph = build_graph()
+
+    async def forever():
+        with graph.flow_execution_span():
+            await asyncio.sleep(10)
+
+    timed_out = False
+    try:
+        await asyncio.wait_for(forever(), timeout=0.05)
+    except (asyncio.TimeoutError, TimeoutError):
+        timed_out = True
+    report({"timed_out": timed_out})
+
+asyncio.run(main())
+"""
+)
+
 NO_OTEL_PROBE = """
 import asyncio, json, sys
 
@@ -368,7 +393,7 @@ def test_an_inner_binding_does_not_overwrite_the_surface_that_took_the_request()
     assert result["after"] is None
 
 
-def test_a_cancelled_flow_is_not_recorded_as_a_successful_one():
+def test_a_flow_a_user_stopped_is_not_recorded_as_a_successful_one():
     result = run_probe(CANCELLED_PROBE)
     assert result["raised"] is True
 
@@ -378,3 +403,15 @@ def test_a_cancelled_flow_is_not_recorded_as_a_successful_one():
     # A withdrawn request is not a service fault, so it must not land on the error rate.
     assert span["status"] == "UNSET"
     assert "error.type" not in span["attrs"]
+
+
+def test_a_flow_killed_by_a_timeout_is_recorded_as_an_error():
+    """The client is served an error and the job row says FAILED, so the span must agree."""
+    result = run_probe(ABORTED_PROBE)
+    assert result["timed_out"] is True
+
+    assert len(result["spans"]) == 1
+    span = result["spans"][0]
+    assert span["attrs"]["status"] == "aborted"
+    assert span["status"] == "ERROR"
+    assert span["attrs"]["error.type"] == "CancelledError"

src/lfx/src/lfx/constants.py

@@ -4,3 +4,10 @@
 
 # Base path for components - will be in lfx package when components are moved
 BASE_COMPONENTS_PATH = str(Path(__file__).parent / "components")
+
+# Marks a CancelledError as "a human withdrew this run" rather than "the service killed it".
+# asyncio gives one exception type for both, so the distinction rides on args and the producers
+# stamp it. It lives here rather than next to the graph exceptions because langflow's service
+# modules read it too, and importing lfx.graph.* from them early enough drags lfx/graph/__init__
+# (which imports Graph) into a circular import. This module imports nothing but pathlib.
+USER_CANCELLED_MESSAGE = "LANGFLOW_USER_CANCELLED"

src/lfx/src/lfx/graph/graph/base.py

@@ -18,6 +18,7 @@
 
 from ag_ui.core import RunFinishedEvent, RunStartedEvent
 
+from lfx.constants import USER_CANCELLED_MESSAGE
 from lfx.exceptions.component import ComponentBuildError
 from lfx.graph.edge.base import CycleEdge, Edge
 from lfx.graph.exceptions import GraphPausedException
@@ -876,18 +877,27 @@ def flow_execution_span(self, *, make_current: bool = True):
             # driven through Graph.process by the durable runner, which opens its own span.
             status = "paused"
             raise
-        except asyncio.CancelledError:
+        except asyncio.CancelledError as exc:
             # CancelledError is a BaseException, so the handler below does not see it and the
-            # span would otherwise report the run as "ok". Reached by a user pressing stop (the
-            # job service marks the job CANCELLED and re-raises) and by any asyncio.wait_for
-            # ceiling wrapped around a span-carrying run: the v2 build driver, a2a, and the
-            # agentic assistant all have one. The run did not finish, so it gets its own value.
+            # span would otherwise report the run as "ok". Two very different things arrive as
+            # the same exception type, and they must not share an outcome:
             #
-            # Span status stays UNSET, which is right for a stop button and arguable for a
-            # timeout: a server-imposed ceiling is closer to a fault, and an operator alerting
-            # on span error rate will not see it. Left as one value for now because the two are
-            # indistinguishable here; the job row (CANCELLED vs FAILED) still tells them apart.
-            status = "cancelled"
+            # A user pressing stop withdrew the request. Nothing is wrong with the service, so
+            # span status stays UNSET and this never reaches error-rate alerting.
+            #
+            # Anything else cancelling the run is the service failing to deliver it: an
+            # asyncio.wait_for execution ceiling (the v2 build driver, a2a and the agentic
+            # assistant each have one), a worker shutdown, a parent task being torn down. The
+            # client is served an error and the job row is written FAILED, so the span says
+            # ERROR too. Reporting these as merely "cancelled" hid every timeout from alerting.
+            #
+            # The producers stamp the user case on args; see USER_CANCELLED_MESSAGE.
+            if exc.args and exc.args[0] == USER_CANCELLED_MESSAGE:
+                status = "cancelled"
+            else:
+                status = "aborted"
+                span.set_status(otel_trace.Status(otel_trace.StatusCode.ERROR, "CancelledError"))
+                span.set_attribute("error.type", "CancelledError")
             raise
         except Exception as exc:
             status = "error"