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

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"