Align failed spans with OTel error.type

Author: mdrideout

CHANGELOG.md

@@ -61,6 +61,7 @@ execution, state management, and lifecycle observation.
 - OpenTelemetry span attributes now use explicit identity names such as `junjo.executable_runtime_id`, `junjo.executable_structural_id`, and `junjo.enclosing_graph_structural_id` instead of the old generic `junjo.id` and `junjo.parent_id` keys.
 - OpenTelemetry and hook payloads now use `executable_definition_id` and `parent_executable_definition_id` instead of the older generic `definition_id` naming on those surfaces.
 - Workflow telemetry now records `junjo.workflow.execution_graph_snapshot` to make it explicit that the graph payload is an execution-scoped compiled snapshot containing both runtime and structural identities.
+- Failed workflow, subflow, node, concurrent, and hook-error spans now set the standard OpenTelemetry `error.type` attribute in addition to Junjo-specific error metadata.
 - `on_state_changed` hook payloads and state-change telemetry context now identify the active executable that performed the mutation, rather than mixing workflow metadata with node or subflow runtime identities.
 - Lifecycle observation examples and docs now show hook registration as a separate concern from workflow definition.
 - `_NestableWorkflow` remains documented in the generated API reference, but is no longer exported from the top-level `junjo` package for direct consumption.

docs/junjo_ai_studio.rst

@@ -326,6 +326,15 @@ Junjo-Specific Telemetry Attributes
 
 Junjo automatically adds these attributes to OpenTelemetry spans:
 
+When an executable span fails, Junjo also emits the standard OpenTelemetry
+error fields alongside the Junjo-specific attributes below:
+
+- ``error.type``: Exception class name for the failed operation
+- span status ``Error``
+- the standard ``exception`` span event with exception details
+
+Ordinary cancellations stay classified as cancellations rather than errors.
+
 Workflow Spans
 --------------
 

docs/opentelemetry.rst

@@ -165,6 +165,18 @@ Junjo's Custom Span Attributes
 
 Junjo adds workflow-specific attributes to all spans. These work with any OTLP exporter:
 
+Failed workflow, subflow, node, and concurrent-execution spans also follow the
+standard OpenTelemetry error contract in addition to the Junjo-specific fields
+below:
+
+- ``error.type`` is set to the exception class name on failed spans.
+- span status is set to ``Error``.
+- the standard ``exception`` span event is recorded via OpenTelemetry's
+  exception recording support.
+
+Cancelled spans do not set ``error.type`` and are not marked with ``Error``
+status unless they actually fail.
+
 Workflow/Subflow Span Attributes
 ---------------------------------
 

src/junjo/_lifecycle.py

@@ -11,6 +11,7 @@
 
 from . import hooks as hook_events
 from .telemetry.otel_schema import JUNJO_OTEL_MODULE_NAME, JunjoOtelSpanTypes
+from .telemetry.span_lifecycle import mark_span_failed
 
 if TYPE_CHECKING:
     from .hooks import Hooks
@@ -673,5 +674,5 @@ def _record_hook_error(
                     "junjo.parent_executable_structural_id",
                     event.parent_executable_structural_id,
                 )
-            span.set_status(trace.StatusCode.ERROR, str(exc))
+            mark_span_failed(span, exc)
             span.record_exception(exc)

src/junjo/node.py

@@ -8,7 +8,11 @@
 from ._lifecycle import ActiveExecutableIdentity, active_executable_identity, get_active_executable_identity
 from .store import StoreT
 from .telemetry.otel_schema import JUNJO_OTEL_MODULE_NAME, JunjoOtelSpanTypes
-from .telemetry.span_lifecycle import get_span_identifiers, mark_span_cancelled
+from .telemetry.span_lifecycle import (
+    get_span_identifiers,
+    mark_span_cancelled,
+    mark_span_failed,
+)
 from .util import generate_safe_id
 
 
@@ -259,7 +263,7 @@ async def execute(self, store: StoreT, parent_id: str) -> None:
 
             except Exception as exc:
                 print("Error executing node service", exc)
-                span.set_status(trace.StatusCode.ERROR, str(exc))
+                mark_span_failed(span, exc)
                 span.record_exception(exc)
                 failure = exc
                 if lifecycle_context is not None:

src/junjo/run_concurrent.py

@@ -9,7 +9,11 @@
 from .node import Node
 from .store import BaseStore
 from .telemetry.otel_schema import JUNJO_OTEL_MODULE_NAME, JunjoOtelSpanTypes
-from .telemetry.span_lifecycle import get_span_identifiers, mark_span_cancelled
+from .telemetry.span_lifecycle import (
+    get_span_identifiers,
+    mark_span_cancelled,
+    mark_span_failed,
+)
 from .util import generate_safe_id
 
 if TYPE_CHECKING:
@@ -286,7 +290,7 @@ async def execute(self, store: BaseStore, parent_id: str) -> None:  # noqa: C901
 
             except Exception as exc:
                 print(f"Error executing node service: {exc}")
-                span.set_status(trace.StatusCode.ERROR, str(exc))
+                mark_span_failed(span, exc)
                 span.record_exception(exc)
                 failure = exc
                 if lifecycle_context is not None:

src/junjo/telemetry/span_lifecycle.py

@@ -6,6 +6,13 @@
 from opentelemetry.trace import Span
 
 
+def mark_span_failed(span: Span, exc: Exception) -> None:
+    """Annotate a span as failed using standard OpenTelemetry error fields."""
+
+    span.set_attribute("error.type", get_error_type(exc))
+    span.set_status(trace.StatusCode.ERROR, str(exc))
+
+
 def mark_span_cancelled(span: Span, exc: asyncio.CancelledError) -> None:
     """Annotate a span as cancelled without treating cancellation as an error."""
 
@@ -14,6 +21,12 @@ def mark_span_cancelled(span: Span, exc: asyncio.CancelledError) -> None:
     span.set_attribute("junjo.cancelled_reason", reason)
 
 
+def get_error_type(exc: BaseException) -> str:
+    """Return the stable OpenTelemetry ``error.type`` value for an exception."""
+
+    return type(exc).__name__
+
+
 def get_span_identifiers(span: Span) -> tuple[str, str]:
     context = span.get_span_context()
     return (

src/junjo/workflow.py

@@ -21,7 +21,11 @@
 from .run_concurrent import RunConcurrent
 from .store import BaseStore, ParentStateT, ParentStoreT, StateT, StoreT
 from .telemetry.otel_schema import JUNJO_OTEL_MODULE_NAME, JunjoOtelSpanTypes
-from .telemetry.span_lifecycle import get_span_identifiers, mark_span_cancelled
+from .telemetry.span_lifecycle import (
+    get_span_identifiers,
+    mark_span_cancelled,
+    mark_span_failed,
+)
 from .util import generate_safe_id
 
 if TYPE_CHECKING:
@@ -370,7 +374,7 @@ async def execute(  # noqa: C901
 
             except Exception as exc:
                 print(f"Error executing workflow: {exc}")
-                span.set_status(trace.StatusCode.ERROR, str(exc))
+                mark_span_failed(span, exc)
                 span.record_exception(exc)
                 failure = exc
 

tests/test_hooks.py

@@ -9,6 +9,7 @@
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
     InMemorySpanExporter,
 )
+from opentelemetry.trace import StatusCode
 
 from junjo import BaseState, BaseStore, Graph, Hooks, Node, RunConcurrent, Subflow, Workflow
 
@@ -33,6 +34,11 @@ async def service(self, store: HookStore) -> None:
         return None
 
 
+class FailingNode(Node[HookStore]):
+    async def service(self, store: HookStore) -> None:
+        raise RuntimeError("boom")
+
+
 class ExampleSubflow(Subflow[HookState, HookStore, HookState, HookStore]):
     async def pre_run_actions(self, parent_store: HookStore, subflow_store: HookStore) -> None:
         parent_state = await parent_store.get_state()
@@ -482,8 +488,14 @@ def failing_hook(event) -> None:
 
     hook_error_span = hook_error_spans[0]
     assert hook_error_span.attributes["junjo.hook.event"] == "workflow_started"
+    assert hook_error_span.attributes["error.type"] == "RuntimeError"
     assert hook_error_span.attributes["junjo.hook.error.type"] == "RuntimeError"
     assert hook_error_span.attributes["junjo.hook.error.message"] == "bad hook"
+    hook_error_exception = next(
+        event for event in hook_error_span.events if event.name == "exception"
+    )
+    assert hook_error_exception.attributes["exception.type"] == "RuntimeError"
+    assert hook_error_exception.attributes["exception.message"] == "bad hook"
     assert hook_error_span.attributes["junjo.run_id"] == result.run_id
     assert (
         hook_error_span.attributes["junjo.executable_definition_id"]
@@ -495,6 +507,34 @@ def failing_hook(event) -> None:
         hook_error_span.attributes["junjo.executable_structural_id"]
         == hook_error_span.attributes["junjo.enclosing_graph_structural_id"]
     )
+    assert hook_error_span.status.status_code is StatusCode.ERROR
+
+
+@pytest.mark.asyncio
+async def test_failed_workflow_and_node_spans_emit_standard_error_type(
+    span_exporter: InMemorySpanExporter,
+) -> None:
+    failing_node = FailingNode()
+
+    workflow = Workflow[HookState, HookStore](
+        name="Failing Workflow",
+        graph_factory=lambda: Graph(source=failing_node, sinks=[failing_node], edges=[]),
+        store_factory=lambda: HookStore(initial_state=HookState()),
+    )
+
+    with pytest.raises(RuntimeError, match="boom"):
+        await workflow.execute()
+
+    spans = {span.name: span for span in span_exporter.get_finished_spans()}
+    workflow_span = spans["Failing Workflow"]
+    node_span = spans["FailingNode"]
+
+    for span in (workflow_span, node_span):
+        assert span.status.status_code is StatusCode.ERROR
+        assert span.attributes["error.type"] == "RuntimeError"
+        exception_event = next(event for event in span.events if event.name == "exception")
+        assert exception_event.attributes["exception.type"] == "RuntimeError"
+        assert exception_event.attributes["exception.message"] == "boom"
 
 
 def test_old_hook_manager_module_is_removed() -> None:

tests/test_run_concurrent_cancellation_telemetry.py

@@ -79,11 +79,19 @@ def create_run_concurrent_graph() -> Graph:
 
     spans = {span.name: span for span in span_exporter.get_finished_spans()}
 
+    workflow_span = spans["Telemetry Workflow"]
+    run_concurrent_span = spans["Concurrent Execution"]
     failing_span = spans["FailingNode"]
     sibling_span = spans["WaitingSiblingNode"]
 
-    assert failing_span.status.status_code is StatusCode.ERROR
-    assert any(event.name == "exception" for event in failing_span.events)
+    for span in (workflow_span, run_concurrent_span, failing_span):
+        assert span.status.status_code is StatusCode.ERROR
+        assert span.attributes["error.type"] == "RuntimeError"
+        exception_event = next(event for event in span.events if event.name == "exception")
+        assert exception_event.attributes["exception.type"] == "RuntimeError"
+        assert exception_event.attributes["exception.message"] == "boom"
 
     assert sibling_span.attributes["junjo.cancelled"] is True
     assert sibling_span.attributes["junjo.cancelled_reason"] == "sibling_failed"
+    assert "error.type" not in sibling_span.attributes
+    assert sibling_span.status.status_code is not StatusCode.ERROR