Record hook failures on active spans

- replace standalone junjo.hook_error spans with junjo.hook_error events on the surrounding workflow, subflow, node, or run-concurrent span
- dispatch terminal hooks before span close so terminal hook failures stay attached to the real execution span
- update hook telemetry tests to prove the new timing and event model
- keep public docs focused on user-facing hook behavior while tracking the telemetry change in the changelog

Author: mdrideout

CHANGELOG.md

@@ -61,9 +61,10 @@ 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.
+- Failed workflow, subflow, node, and concurrent spans now set the standard OpenTelemetry `error.type` attribute in addition to Junjo-specific error metadata.
 - `JunjoOtelExporter` now exposes `shutdown()`, and the docs/examples now teach provider shutdown as the normal OpenTelemetry lifecycle while keeping `flush()` as a targeted manual drain tool.
 - `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.
+- Hook callback failures are now recorded as `junjo.hook_error` events on the surrounding workflow, subflow, node, or concurrent span, and terminal hooks now dispatch before span close so those events stay attached to the real execution span.
 - 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.
 - Public docstrings and examples were updated to reflect the current execution, hooks, and result APIs.

src/junjo/_lifecycle.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import inspect
+import traceback
 from collections.abc import Mapping
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -10,8 +11,7 @@
 from opentelemetry import trace
 
 from . import hooks as hook_events
-from .telemetry.otel_schema import JUNJO_OTEL_MODULE_NAME, JunjoOtelSpanTypes
-from .telemetry.span_lifecycle import mark_span_failed
+from .telemetry.otel_schema import JunjoOtelSpanTypes
 
 if TYPE_CHECKING:
     from .hooks import Hooks
@@ -94,7 +94,7 @@ async def dispatch(self, prepared: PreparedHookEvent | None) -> None:
                 if inspect.isawaitable(result):
                     await result
             except Exception as exc:
-                self._record_hook_error(prepared.event_name, callback, exc, prepared.event)
+                self._record_hook_error(prepared.event_name, callback, exc)
 
     async def workflow_started(
         self,
@@ -621,58 +621,22 @@ def _record_hook_error(
         event_name: str,
         callback,
         exc: Exception,
-        event: hook_events.LifecycleEvent,
     ) -> None:
-        tracer = trace.get_tracer(JUNJO_OTEL_MODULE_NAME)
+        span = trace.get_current_span()
+        if not span.is_recording():
+            return
+
         callback_name = getattr(callback, "__qualname__", callback.__class__.__qualname__)
         callback_module = getattr(callback, "__module__", callback.__class__.__module__)
-
-        with tracer.start_as_current_span("junjo.hook_error") as span:
-            span.set_attribute("junjo.hook.event", event_name)
-            span.set_attribute(
-                "junjo.hook.callback",
-                f"{callback_module}.{callback_name}",
-            )
-            span.set_attribute("junjo.hook.error.type", type(exc).__name__)
-            span.set_attribute("junjo.hook.error.message", str(exc))
-            span.set_attribute("junjo.trace_id", event.trace_id)
-            span.set_attribute("junjo.span_id", event.span_id)
-            span.set_attribute("junjo.run_id", event.run_id)
-            span.set_attribute(
-                "junjo.executable_definition_id",
-                event.executable_definition_id,
-            )
-            parent_executable_definition_id = getattr(
-                event,
-                "parent_executable_definition_id",
-                None,
-            )
-            if parent_executable_definition_id is not None:
-                span.set_attribute(
-                    "junjo.parent_executable_definition_id",
-                    parent_executable_definition_id,
-                )
-            span.set_attribute(
-                "junjo.executable_runtime_id",
-                event.executable_runtime_id,
-            )
-            span.set_attribute(
-                "junjo.executable_structural_id",
-                event.executable_structural_id,
-            )
-            span.set_attribute(
-                "junjo.enclosing_graph_structural_id",
-                event.enclosing_graph_structural_id,
-            )
-            if event.parent_executable_runtime_id is not None:
-                span.set_attribute(
-                    "junjo.parent_executable_runtime_id",
-                    event.parent_executable_runtime_id,
-                )
-            if event.parent_executable_structural_id is not None:
-                span.set_attribute(
-                    "junjo.parent_executable_structural_id",
-                    event.parent_executable_structural_id,
-                )
-            mark_span_failed(span, exc)
-            span.record_exception(exc)
+        error_attributes: dict[str, str] = {
+            "junjo.hook.event": event_name,
+            "junjo.hook.callback": f"{callback_module}.{callback_name}",
+            "junjo.hook.error.type": type(exc).__name__,
+            "junjo.hook.error.message": str(exc),
+            "exception.type": type(exc).__name__,
+            "exception.message": str(exc),
+            "exception.stacktrace": "".join(
+                traceback.format_exception(type(exc), exc, exc.__traceback__)
+            ),
+        }
+        span.add_event("junjo.hook_error", error_attributes)

src/junjo/hooks.py

@@ -210,6 +210,9 @@ class Hooks:
     Registry for optional Junjo lifecycle callbacks.
 
     Hooks are observers. They do not create spans or control workflow execution.
+    If a hook callback raises, Junjo keeps execution isolated and continues
+    dispatching the remaining callbacks for that hook.
+
     To use them, register one or more callbacks and pass the registry to a
     workflow or subflow.
 

src/junjo/node.py

@@ -295,8 +295,8 @@ async def execute(self, store: StoreT, parent_id: str) -> None:
                         ),
                     )
 
-        if lifecycle_context is not None:
-            await lifecycle_context.dispatcher.dispatch(prepared_terminal_event)
+            if lifecycle_context is not None:
+                await lifecycle_context.dispatcher.dispatch(prepared_terminal_event)
 
         if cancellation is not None:
             raise cancellation

src/junjo/run_concurrent.py

@@ -322,8 +322,8 @@ async def execute(self, store: BaseStore, parent_id: str) -> None:  # noqa: C901
                         ),
                     )
 
-        if lifecycle_context is not None:
-            await lifecycle_context.dispatcher.dispatch(prepared_terminal_event)
+            if lifecycle_context is not None:
+                await lifecycle_context.dispatcher.dispatch(prepared_terminal_event)
 
         if cancellation is not None:
             raise cancellation

src/junjo/workflow.py

@@ -473,7 +473,7 @@ async def execute(  # noqa: C901
                         ),
                     )
 
-        await ctx.dispatcher.dispatch(prepared_terminal_event)
+                await ctx.dispatcher.dispatch(prepared_terminal_event)
 
         if cancellation is not None:
             raise cancellation

tests/test_hooks.py

@@ -220,24 +220,31 @@ def create_graph() -> Graph:
 
 
 @pytest.mark.asyncio
-async def test_terminal_hooks_run_after_span_close(
+async def test_terminal_hooks_run_before_span_close(
     span_exporter: InMemorySpanExporter,
 ) -> None:
     hooks = Hooks()
-    seen_closed: list[bool] = []
+    span_open: list[bool] = []
+    current_span_matches_event: list[bool] = []
 
     def on_completed(event) -> None:
         finished_span_ids = {
             format(span.context.span_id, "016x")
             for span in span_exporter.get_finished_spans()
         }
-        seen_closed.append(event.span_id in finished_span_ids)
+        current_span_id = format(
+            trace.get_current_span().get_span_context().span_id,
+            "016x",
+        )
+        span_open.append(event.span_id not in finished_span_ids)
+        current_span_matches_event.append(current_span_id == event.span_id)
 
     hooks.on_workflow_completed(on_completed)
 
     await create_simple_workflow(hooks=hooks).execute()
 
-    assert seen_closed == [True]
+    assert span_open == [True]
+    assert current_span_matches_event == [True]
 
 
 @pytest.mark.asyncio
@@ -480,34 +487,66 @@ def failing_hook(event) -> None:
     result = await create_simple_workflow(hooks=hooks).execute()
 
     assert observed == ["ran"]
+    assert result.state.steps == ["HookNode"]
 
-    hook_error_spans = [
-        span for span in span_exporter.get_finished_spans() if span.name == "junjo.hook_error"
+    finished_spans = span_exporter.get_finished_spans()
+    assert all(span.name != "junjo.hook_error" for span in finished_spans)
+
+    spans = {span.name: span for span in finished_spans}
+    workflow_span = spans["Hook Workflow"]
+    hook_error_events = [
+        event for event in workflow_span.events if event.name == "junjo.hook_error"
     ]
-    assert len(hook_error_spans) == 1
-
-    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"]
-        == result.definition_id
-    )
-    assert "junjo.parent_executable_definition_id" not in hook_error_span.attributes
-    assert hook_error_span.attributes["junjo.executable_runtime_id"] == result.run_id
-    assert (
-        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
+    assert len(hook_error_events) == 1
+
+    hook_error_event = hook_error_events[0]
+    assert hook_error_event.attributes["junjo.hook.event"] == "workflow_started"
+    assert "failing_hook" in hook_error_event.attributes["junjo.hook.callback"]
+    assert hook_error_event.attributes["junjo.hook.error.type"] == "RuntimeError"
+    assert hook_error_event.attributes["junjo.hook.error.message"] == "bad hook"
+    assert hook_error_event.attributes["exception.type"] == "RuntimeError"
+    assert hook_error_event.attributes["exception.message"] == "bad hook"
+    assert "exception.stacktrace" in hook_error_event.attributes
+    assert workflow_span.status.status_code is StatusCode.UNSET
+
+
+@pytest.mark.asyncio
+async def test_terminal_hook_failures_are_recorded_on_the_terminal_span(
+    span_exporter: InMemorySpanExporter,
+) -> None:
+    hooks = Hooks()
+    span_open: list[bool] = []
+
+    def failing_hook(event) -> None:
+        finished_span_ids = {
+            format(span.context.span_id, "016x")
+            for span in span_exporter.get_finished_spans()
+        }
+        span_open.append(event.span_id not in finished_span_ids)
+        raise RuntimeError("bad terminal hook")
+
+    hooks.on_workflow_completed(failing_hook)
+
+    result = await create_simple_workflow(hooks=hooks).execute()
+
+    assert result.state.steps == ["HookNode"]
+    assert span_open == [True]
+
+    finished_spans = span_exporter.get_finished_spans()
+    assert all(span.name != "junjo.hook_error" for span in finished_spans)
+
+    workflow_span = next(span for span in finished_spans if span.name == "Hook Workflow")
+    hook_error_events = [
+        event for event in workflow_span.events if event.name == "junjo.hook_error"
+    ]
+    assert len(hook_error_events) == 1
+    hook_error_event = hook_error_events[0]
+    assert hook_error_event.attributes["junjo.hook.event"] == "workflow_completed"
+    assert hook_error_event.attributes["junjo.hook.error.type"] == "RuntimeError"
+    assert hook_error_event.attributes["junjo.hook.error.message"] == "bad terminal hook"
+    assert hook_error_event.attributes["exception.type"] == "RuntimeError"
+    assert hook_error_event.attributes["exception.message"] == "bad terminal hook"
+    assert workflow_span.status.status_code is StatusCode.UNSET
 
 
 @pytest.mark.asyncio