feat: add dedicated StopEvent subclasses for workflow termination scenarios (#264)

* feat: add dedicated StopEvent subclasses for workflow termination scenarios

Add WorkflowTerminationEvent base class and specific subclasses to provide
detailed information when workflows end abnormally:

- WorkflowTimedOutEvent: Published when workflow exceeds timeout, includes
  timeout duration and list of active steps
- WorkflowCancelledEvent: Published when workflow is cancelled by user
- WorkflowFailedEvent: Published when a step fails permanently, includes
  step name, exception type, and exception message

These events replace the empty StopEvent() instances that were previously
published, allowing consumers to understand why a workflow ended and take
appropriate action.

Closes: issue about providing exception information in StopEvent

* refactor: simplify termination events to inherit directly from StopEvent

* chore: add changeset

* feat: add traceback and fully qualified exception type to WorkflowFailedEvent

* feat: add retry info (attempts, elapsed_seconds) to WorkflowFailedEvent

Also fixes a bug where step_function.py used time.monotonic() instead of
time.time(), causing incompatible timestamps with first_attempt_at.

* fix lints/locks

* docs: add workflow termination events to streaming docs

---------

Co-authored-by: Claude <noreply@anthropic.com>

.changeset/workflow-termination-events.md

@@ -0,0 +1,5 @@
+---
+"llama-index-workflows": minor
+---
+
+Add dedicated StopEvent subclasses for workflow termination (timeout, cancellation, failure)

docs/api_docs/docs/api_reference/events.md

@@ -6,3 +6,6 @@
         - HumanResponseEvent
         - StartEvent
         - StopEvent
+        - WorkflowTimedOutEvent
+        - WorkflowCancelledEvent
+        - WorkflowFailedEvent

docs/src/content/docs/llamaagents/workflows/streaming.mdx

@@ -99,3 +99,27 @@ if __name__ == "__main__":
 ```
 
 `run` runs the workflow in the background, while `stream_events` will provide any event that gets written to the stream. It stops when the stream delivers a `StopEvent`, after which you can get the final result of the workflow as you normally would.
+
+## Handling workflow termination
+
+When a workflow ends abnormally (timeout, cancellation, or step failure), a specific `StopEvent` subclass is published to the stream before the exception is raised:
+
+- **`WorkflowTimedOutEvent`** - Published when the workflow exceeds its timeout. Contains `timeout` (seconds) and `active_steps` (list of step names that were running).
+- **`WorkflowCancelledEvent`** - Published when the workflow is cancelled by the user.
+- **`WorkflowFailedEvent`** - Published when a step fails permanently after exhausting retries. Contains `step_name`, `exception_type`, `exception_message`, `traceback`, `attempts`, and `elapsed_seconds`.
+
+```python
+from workflows.events import (
+    WorkflowTimedOutEvent,
+    WorkflowCancelledEvent,
+    WorkflowFailedEvent,
+)
+
+async for ev in handler.stream_events():
+    if isinstance(ev, WorkflowTimedOutEvent):
+        print(f"Workflow timed out after {ev.timeout}s")
+    elif isinstance(ev, WorkflowCancelledEvent):
+        print("Workflow was cancelled")
+    elif isinstance(ev, WorkflowFailedEvent):
+        print(f"Step '{ev.step_name}' failed after {ev.attempts} attempts: {ev.exception_message}")
+```

packages/llama-index-workflows/src/workflows/events.py

@@ -209,6 +209,79 @@ class StopEvent(Event):
         return str(self._result)
 
 
+class WorkflowTimedOutEvent(StopEvent):
+    """Published when a workflow exceeds its configured timeout.
+
+    This event is published to the event stream when a workflow times out,
+    allowing consumers to understand why the workflow ended before the
+    WorkflowTimeoutError exception is raised.
+
+    Attributes:
+        timeout: The timeout duration in seconds that was exceeded.
+        active_steps: List of step names that were still active when the timeout occurred.
+
+    Examples:
+        ```python
+        async for event in handler.stream_events():
+            if isinstance(event, WorkflowTimedOutEvent):
+                print(f"Workflow timed out after {event.timeout}s")
+                print(f"Active steps: {event.active_steps}")
+        ```
+    """
+
+    timeout: float
+    active_steps: list[str]
+
+
+class WorkflowCancelledEvent(StopEvent):
+    """Published when a workflow is cancelled by the user.
+
+    This event is published to the event stream when a workflow is cancelled
+    via the handler or programmatically, allowing consumers to understand why
+    the workflow ended before the WorkflowCancelledByUser exception is raised.
+
+    Examples:
+        ```python
+        async for event in handler.stream_events():
+            if isinstance(event, WorkflowCancelledEvent):
+                print("Workflow was cancelled by user")
+        ```
+    """
+
+
+class WorkflowFailedEvent(StopEvent):
+    """Published when a workflow step fails permanently.
+
+    This event is published to the event stream when a step fails and all
+    retries are exhausted, allowing consumers to understand why the workflow
+    ended before the exception is raised.
+
+    Attributes:
+        step_name: The name of the step that failed.
+        exception_type: The fully qualified type name of the exception that caused the failure.
+        exception_message: The string representation of the exception message.
+        traceback: The formatted stack trace of the exception.
+        attempts: The total number of attempts made before giving up.
+        elapsed_seconds: Time in seconds from first attempt to final failure.
+
+    Examples:
+        ```python
+        async for event in handler.stream_events():
+            if isinstance(event, WorkflowFailedEvent):
+                print(f"Step '{event.step_name}' failed after {event.attempts} attempts")
+                print(f"Total time: {event.elapsed_seconds:.2f}s")
+                print(event.traceback)
+        ```
+    """
+
+    step_name: str
+    exception_type: str
+    exception_message: str
+    traceback: str
+    attempts: int
+    elapsed_seconds: float
+
+
 class InputRequiredEvent(Event):
     """Emitted when human input is required to proceed.
 

packages/llama-index-workflows/src/workflows/runtime/control_loop.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import time
+import traceback
 from dataclasses import replace
 from typing import TYPE_CHECKING
 
@@ -22,7 +23,10 @@ from workflows.events import (
     StepState,
     StepStateChanged,
     StopEvent,
+    WorkflowCancelledEvent,
+    WorkflowFailedEvent,
     WorkflowIdleEvent,
+    WorkflowTimedOutEvent,
 )
 from workflows.runtime.types.commands import (
     CommandCompleteRun,
@@ -454,14 +458,34 @@ def _process_step_result_tick(
                     )
                 )
             else:
-                # used as a sentinel to end the stream. Perhaps reconsider this and have an alternate failure stop event
+                # Publish a WorkflowFailedEvent to inform stream consumers about the failure
                 state.is_running = False
-                commands.append(CommandPublishEvent(event=StopEvent()))
-                commands.append(
-                    CommandFailWorkflow(
-                        step_name=tick.step_name, exception=result.exception
+                exception = result.exception
+                exc_type = type(exception)
+                exc_module = exc_type.__module__
+                exc_qualname = f"{exc_module}.{exc_type.__qualname__}"
+                exc_traceback = "".join(
+                    traceback.format_exception(
+                        exc_type, exception, exception.__traceback__
                     )
                 )
+                total_attempts = this_execution.attempts + 1
+                elapsed = result.failed_at - this_execution.first_attempt_at
+                commands.append(
+                    CommandPublishEvent(
+                        event=WorkflowFailedEvent(
+                            step_name=tick.step_name,
+                            exception_type=exc_qualname,
+                            exception_message=str(exception),
+                            traceback=exc_traceback,
+                            attempts=total_attempts,
+                            elapsed_seconds=elapsed,
+                        )
+                    )
+                )
+                commands.append(
+                    CommandFailWorkflow(step_name=tick.step_name, exception=exception)
+                )
         elif isinstance(result, AddCollectedEvent):
             # The current state of collected events.
             collected_events = state.workers[
@@ -681,11 +705,9 @@ def _process_cancel_run_tick(
     tick: TickCancelRun, init: BrokerState
 ) -> tuple[BrokerState, list[WorkflowCommand]]:
     state = init.deepcopy()
-    # retain running state, for resumption.
-    # TODO - when/if we persist stream events, this StopEvent should be reconsidered, as there should only ever be one stop event.
-    # Perhaps on resumption, if the workflow is running, then any existing stop events of a "cancellation" type should be omitted from the stream.
+    # Retain running state for resumption.
     return state, [
-        CommandPublishEvent(event=StopEvent()),
+        CommandPublishEvent(event=WorkflowCancelledEvent()),
         CommandHalt(exception=WorkflowCancelledByUser()),
     ]
 
@@ -713,7 +735,12 @@ def _process_timeout_tick(
         else "No steps active"
     )
     return state, [
-        CommandPublishEvent(event=StopEvent()),
+        CommandPublishEvent(
+            event=WorkflowTimedOutEvent(
+                timeout=tick.timeout,
+                active_steps=active_steps,
+            )
+        ),
         CommandHalt(
             exception=WorkflowTimeoutError(
                 f"Operation timed out after {tick.timeout} seconds. {steps_info}"

packages/llama-index-workflows/src/workflows/runtime/types/step_function.py

@@ -125,7 +125,7 @@ def as_step_worker_function(func: Callable[P, Awaitable[R]]) -> StepWorkerFuncti
                 returns.return_values.append(e.add)
             except Exception as e:
                 returns.return_values.append(
-                    StepWorkerFailed(exception=e, failed_at=time.monotonic())
+                    StepWorkerFailed(exception=e, failed_at=time.time())
                 )
             return returns.return_values
         finally:

packages/llama-index-workflows/tests/runtime/test_control_loop.py

@@ -28,7 +28,10 @@ from workflows.events import (
     StartEvent,
     StepStateChanged,
     StopEvent,
+    WorkflowCancelledEvent,
+    WorkflowFailedEvent,
     WorkflowIdleEvent,
+    WorkflowTimedOutEvent,
 )
 from workflows.retry_policy import ConstantDelayRetryPolicy, RetryPolicy
 from workflows.runtime.control_loop import control_loop
@@ -267,10 +270,10 @@ async def test_control_loop_with_external_event(test_plugin: MockRuntimePlugin)
 @pytest.mark.asyncio
 async def test_control_loop_timeout(test_plugin: MockRuntimePlugin) -> None:
     """
-    Test that workflow timeout raises WorkflowTimeoutError and publishes StopEvent.
+    Test that workflow timeout raises WorkflowTimeoutError and publishes WorkflowTimedOutEvent.
 
-    When a workflow times out, an empty StopEvent should be published to the stream
-    to signal stream closure before the exception is raised.
+    When a workflow times out, a WorkflowTimedOutEvent should be published to the stream
+    to inform consumers about the timeout before the exception is raised.
     """
 
     class SlowWorkflow(Workflow):
@@ -296,11 +299,15 @@ async def test_control_loop_timeout(test_plugin: MockRuntimePlugin) -> None:
     with pytest.raises(WorkflowTimeoutError):
         await asyncio.wait_for(task, timeout=1.0)
 
-    # Verify an empty StopEvent was published to the stream
+    # Verify a WorkflowTimedOutEvent was published to the stream
     assert stop_event is not None, (
-        "Timeout should publish empty StopEvent to stream before raising exception"
+        "Timeout should publish WorkflowTimedOutEvent to stream before raising exception"
     )
-    assert stop_event.result is None, "Timeout StopEvent should have None result"
+    assert isinstance(stop_event, WorkflowTimedOutEvent), (
+        f"Expected WorkflowTimedOutEvent, got {type(stop_event).__name__}"
+    )
+    assert stop_event.timeout == 0.01, "Timeout event should contain the timeout value"
+    assert stop_event.active_steps == ["slow"], "Timeout event should list active steps"
 
 
 @pytest.mark.asyncio
@@ -337,9 +344,9 @@ async def test_control_loop_step_failure_publishes_stop_event(
 ) -> None:
     """
     Test that when a step fails permanently (retries exhausted),
-    an empty StopEvent is published to the stream before raising the exception.
+    a WorkflowFailedEvent is published to the stream before raising the exception.
 
-    This allows external consumers to know the workflow stream has ended.
+    This allows external consumers to know why the workflow stream has ended.
     """
 
     class FailingWorkflow(Workflow):
@@ -363,11 +370,27 @@ async def test_control_loop_step_failure_publishes_stop_event(
     with pytest.raises(ValueError, match="intentional failure"):
         await asyncio.wait_for(task, timeout=1.0)
 
-    # Verify that an empty StopEvent was published before the exception
+    # Verify that a WorkflowFailedEvent was published before the exception
     assert stop_event is not None, (
-        "Empty StopEvent should be published to stream when step fails permanently"
+        "WorkflowFailedEvent should be published to stream when step fails permanently"
     )
-    assert stop_event.result is None, "Failure StopEvent should have None result"
+    assert isinstance(stop_event, WorkflowFailedEvent), (
+        f"Expected WorkflowFailedEvent, got {type(stop_event).__name__}"
+    )
+    assert stop_event.step_name == "always_fails", (
+        "Failed event should contain the step name"
+    )
+    assert stop_event.exception_type == "builtins.ValueError", (
+        "Failed event should contain the fully qualified exception type"
+    )
+    assert stop_event.exception_message == "intentional failure", (
+        "Failed event should contain the exception message"
+    )
+    assert "ValueError: intentional failure" in stop_event.traceback, (
+        "Failed event should contain the traceback"
+    )
+    assert stop_event.attempts == 1, "Failed event should contain the attempt count"
+    assert stop_event.elapsed_seconds >= 0, "Failed event should contain elapsed time"
 
 
 @pytest.mark.asyncio
@@ -615,10 +638,10 @@ async def test_control_loop_concurrency_queueing(
 @pytest.mark.asyncio
 async def test_control_loop_user_cancellation(test_plugin: MockRuntimePlugin) -> None:
     """
-    Test that user cancellation raises WorkflowCancelledByUser and publishes StopEvent.
+    Test that user cancellation raises WorkflowCancelledByUser and publishes WorkflowCancelledEvent.
 
-    When a workflow is cancelled, an empty StopEvent should be published to the stream
-    to signal stream closure before the exception is raised.
+    When a workflow is cancelled, a WorkflowCancelledEvent should be published to the stream
+    to inform consumers about the cancellation before the exception is raised.
     """
 
     class CancelWorkflow(Workflow):
@@ -648,11 +671,13 @@ async def test_control_loop_user_cancellation(test_plugin: MockRuntimePlugin) ->
     with pytest.raises(WorkflowCancelledByUser):
         await asyncio.wait_for(task, timeout=1.0)
 
-    # Verify an empty StopEvent was published to the stream
+    # Verify a WorkflowCancelledEvent was published to the stream
     assert stop_event is not None, (
-        "Cancellation should publish empty StopEvent to stream before raising exception"
+        "Cancellation should publish WorkflowCancelledEvent to stream before raising exception"
+    )
+    assert isinstance(stop_event, WorkflowCancelledEvent), (
+        f"Expected WorkflowCancelledEvent, got {type(stop_event).__name__}"
     )
-    assert stop_event.result is None, "Cancellation StopEvent should have None result"
 
 
 @pytest.mark.asyncio

packages/llama-index-workflows/tests/test_event.py

@@ -6,7 +6,13 @@ from typing import Any, cast
 import pytest
 from pydantic import PrivateAttr
 from workflows.context import JsonSerializer
-from workflows.events import Event, StopEvent
+from workflows.events import (
+    Event,
+    StopEvent,
+    WorkflowCancelledEvent,
+    WorkflowFailedEvent,
+    WorkflowTimedOutEvent,
+)
 
 
 class _TestEvent(Event):
@@ -162,3 +168,164 @@ def test_custom_stop_event_repr_no_result() -> None:
     ev = CustomStopEvent(foo="foo", bar=42)
     rep = repr(ev)
     assert rep == "CustomStopEvent(foo='foo', bar=42)"
+
+
+# Tests for workflow termination event subclasses
+
+
+def test_workflow_termination_events_are_stop_events() -> None:
+    """Verify workflow termination events are subclasses of StopEvent."""
+    assert issubclass(WorkflowTimedOutEvent, StopEvent)
+    assert issubclass(WorkflowCancelledEvent, StopEvent)
+    assert issubclass(WorkflowFailedEvent, StopEvent)
+
+
+def test_workflow_timed_out_event() -> None:
+    """Test WorkflowTimedOutEvent creation and attributes."""
+    ev = WorkflowTimedOutEvent(timeout=30.0, active_steps=["step1", "step2"])
+    assert ev.timeout == 30.0
+    assert ev.active_steps == ["step1", "step2"]
+    assert isinstance(ev, StopEvent)
+
+
+def test_workflow_timed_out_event_empty_active_steps() -> None:
+    """Test WorkflowTimedOutEvent with no active steps."""
+    ev = WorkflowTimedOutEvent(timeout=5.0, active_steps=[])
+    assert ev.timeout == 5.0
+    assert ev.active_steps == []
+
+
+def test_workflow_timed_out_event_serialization() -> None:
+    """Test WorkflowTimedOutEvent serialization and deserialization."""
+    ev = WorkflowTimedOutEvent(timeout=30.0, active_steps=["step1", "step2"])
+    data_dict = ev.model_dump()
+    assert data_dict == {"timeout": 30.0, "active_steps": ["step1", "step2"]}
+
+    serializer = JsonSerializer()
+    serialized_ev = serializer.serialize(ev)
+    deserialized_ev = serializer.deserialize(serialized_ev)
+
+    assert type(deserialized_ev).__name__ == type(ev).__name__
+    deserialized_ev = cast(WorkflowTimedOutEvent, deserialized_ev)
+    assert ev.timeout == deserialized_ev.timeout
+    assert ev.active_steps == deserialized_ev.active_steps
+
+
+def test_workflow_timed_out_event_repr() -> None:
+    """Test WorkflowTimedOutEvent string representation."""
+    ev = WorkflowTimedOutEvent(timeout=10.0, active_steps=["my_step"])
+    rep = repr(ev)
+    assert "WorkflowTimedOutEvent" in rep
+    assert "timeout=10.0" in rep
+    assert "active_steps=['my_step']" in rep
+
+
+def test_workflow_cancelled_event() -> None:
+    """Test WorkflowCancelledEvent creation."""
+    ev = WorkflowCancelledEvent()
+    assert isinstance(ev, StopEvent)
+
+
+def test_workflow_cancelled_event_serialization() -> None:
+    """Test WorkflowCancelledEvent serialization and deserialization."""
+    ev = WorkflowCancelledEvent()
+    data_dict = ev.model_dump()
+    assert data_dict == {}
+
+    serializer = JsonSerializer()
+    serialized_ev = serializer.serialize(ev)
+    deserialized_ev = serializer.deserialize(serialized_ev)
+
+    assert type(deserialized_ev).__name__ == type(ev).__name__
+
+
+def test_workflow_cancelled_event_repr() -> None:
+    """Test WorkflowCancelledEvent string representation."""
+    ev = WorkflowCancelledEvent()
+    rep = repr(ev)
+    assert rep == "WorkflowCancelledEvent()"
+
+
+def test_workflow_failed_event() -> None:
+    """Test WorkflowFailedEvent creation and attributes."""
+    ev = WorkflowFailedEvent(
+        step_name="my_step",
+        exception_type="builtins.ValueError",
+        exception_message="Something went wrong",
+        traceback="Traceback (most recent call last):\n  ...\nValueError: Something went wrong\n",
+        attempts=3,
+        elapsed_seconds=1.5,
+    )
+    assert ev.step_name == "my_step"
+    assert ev.exception_type == "builtins.ValueError"
+    assert ev.exception_message == "Something went wrong"
+    assert "ValueError" in ev.traceback
+    assert ev.attempts == 3
+    assert ev.elapsed_seconds == 1.5
+    assert isinstance(ev, StopEvent)
+
+
+def test_workflow_failed_event_serialization() -> None:
+    """Test WorkflowFailedEvent serialization and deserialization."""
+    ev = WorkflowFailedEvent(
+        step_name="failing_step",
+        exception_type="builtins.RuntimeError",
+        exception_message="Test failure",
+        traceback="Traceback...\nRuntimeError: Test failure\n",
+        attempts=2,
+        elapsed_seconds=0.5,
+    )
+    data_dict = ev.model_dump()
+    assert data_dict == {
+        "step_name": "failing_step",
+        "exception_type": "builtins.RuntimeError",
+        "exception_message": "Test failure",
+        "traceback": "Traceback...\nRuntimeError: Test failure\n",
+        "attempts": 2,
+        "elapsed_seconds": 0.5,
+    }
+
+    serializer = JsonSerializer()
+    serialized_ev = serializer.serialize(ev)
+    deserialized_ev = serializer.deserialize(serialized_ev)
+
+    assert type(deserialized_ev).__name__ == type(ev).__name__
+    deserialized_ev = cast(WorkflowFailedEvent, deserialized_ev)
+    assert ev.step_name == deserialized_ev.step_name
+    assert ev.exception_type == deserialized_ev.exception_type
+    assert ev.exception_message == deserialized_ev.exception_message
+    assert ev.traceback == deserialized_ev.traceback
+    assert ev.attempts == deserialized_ev.attempts
+    assert ev.elapsed_seconds == deserialized_ev.elapsed_seconds
+
+
+def test_workflow_failed_event_repr() -> None:
+    """Test WorkflowFailedEvent string representation."""
+    ev = WorkflowFailedEvent(
+        step_name="my_step",
+        exception_type="builtins.ValueError",
+        exception_message="error msg",
+        traceback="...",
+        attempts=1,
+        elapsed_seconds=0.1,
+    )
+    rep = repr(ev)
+    assert "WorkflowFailedEvent" in rep
+    assert "step_name='my_step'" in rep
+    assert "exception_type='builtins.ValueError'" in rep
+    assert "exception_message='error msg'" in rep
+
+
+def test_workflow_failed_event_with_nested_exception_type() -> None:
+    """Test WorkflowFailedEvent with a qualified exception type name."""
+    ev = WorkflowFailedEvent(
+        step_name="api_step",
+        exception_type="http.client.HTTPException",
+        exception_message="Connection refused",
+        traceback="Traceback...",
+        attempts=5,
+        elapsed_seconds=10.0,
+    )
+    assert ev.exception_type == "http.client.HTTPException"
+    assert ev.attempts == 5
+    assert ev.elapsed_seconds == 10.0

uv.lock

@@ -1644,7 +1644,7 @@ wheels = [
 
 [[package]]
 name = "llama-index-utils-workflow"
-version = "0.5.2"
+version = "0.6.0"
 source = { editable = "packages/llama-index-utils-workflow" }
 dependencies = [
     { name = "llama-index-core" },