observability: phase 6.1 concurrency-safe observer (PR-A) (#22)

* observability: phase 6.1 concurrency-safe observer

Phase 6.0 left three architectural limitations: OTelObserver state
collided across concurrent invocations on a shared instance, OTel
attach tokens spanned event boundaries (LIFO violations under
interleaved fan-out, suppressed by try/except guards in round-7),
and LLM spans had no calling-node identity threaded through, so
parent resolution fell back to the OTel current-span context and
mis-attributed under concurrent fan-out + retry. Spec agent's ack
folded the three into one coherent refactor — they share the same
primitive ("parents come from the observer's internal maps within
a single event handler's scope") and split would ship intermediate
states that fail their own correctness story.

Three new ContextVars in observability/correlation.py carry the
calling-node identity: current_namespace_prefix (default ()),
current_fan_out_index (default None), current_attempt_index
(default 0). The engine sets the first two in the outer per-node
scope of _step_function_node / _step_subgraph_node /
_step_fan_out_node; attempt_index lives inside innermost so retry
middleware that re-enters the inner closure bumps the value per
attempt. _LlmEventState gains calling_* fields populated from the
ContextVars at dispatch time.

OTelObserver state is now outer-keyed by invocation_id, not
correlation_id. Resume preserves the cid across runs (per spec
§5.6 cross-run join), so a cid-keyed map would have the resumed
run's spans inherit the prior invocation's trace_id, violating
§5.1's "each invocation has its own trace_id". Caught during impl
by test_phase5_fixture_031_span_assertions. The cid is still set
as openarmature.correlation_id on every span — only the state-
scoping key changed.

Parent resolution moves entirely to internal maps. _OpenSpan no
longer carries an OTel context token; each event handler reads
the parent from per-invocation open_spans / subgraph_spans /
detached_roots and starts the span with
context=set_span_in_context(parent). The round-7 try/except
ValueError guards on cross-event detach are deleted (no tokens
cross events). The close-prior-correlation_id branch in
_handle_started is deleted (per-invocation scoping makes it
unreachable). "Concurrency model — NOT safe" warning and
cross-event attach-token notes come out of the class docstring.

LLM-span parent resolution uses calling-node identity directly:
_resolve_llm_parent looks up open_spans by the full _StackKey
(calling_namespace_prefix, calling_attempt_index,
calling_fan_out_index), then walks ancestors for subgraph dispatch
or detached roots, then falls back to the invocation span. Under
concurrent fan-out + retry the LLM span lands under exactly the
right calling node regardless of dispatch ordering or which
sibling instance's span happens to be on the OTel current-span
stack.

Tests added in tests/unit/test_observability_otel.py:

- shared-observer concurrent invocations (asyncio.gather × 5):
  N distinct trace_ids and clean per-invocation span trees.
- concurrent fan-out (4 instances) each calling LLM: every LLM
  span parents under its own calling instance, not a sibling.
- LIFO regression check under warnings.catch_warnings("error"):
  the path that previously needed try/except guards stays quiet
  under the new architecture rather than being suppressed.
- LLM call inside retried node (max_attempts=3, two failures +
  one success): each LLM span parents under its own attempt's
  span, not a hardcoded attempt 0.

387 tests pass (4 new). Pyright clean.

* otel: address PR #22 review

- Fix _fan_out_node_span_key hardcoding attempt_index=0. Fan-out
  nodes wrapped with retry middleware open at the bumped
  attempt_index, so the lookup missed under retry and the
  detached-instance Link wasn't added to the §4.4 cross-trace
  navigation. Replaced with _find_fan_out_node_span scan that
  finds the in-flight fan-out span at prefix regardless of
  attempt_index (only one such entry is open at a time).
- Update close_invocation docstring to be honest about the
  invocation_id sourcing constraint (CompiledGraph.invoke doesn't
  return it; the ContextVar is reset before control returns) and
  recommend shutdown() as the typical-production lifecycle hook.
  The deeper UX gap (long-lived observers without periodic
  shutdown accumulate per-invocation residue) is tracked in
  phase-6-1-conformance-fillin.md for follow-up.
- Fix stale "now-cid-scoped" wording in openai.py — observer
  state is invocation_id-keyed in the merged code.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -44,12 +44,18 @@
 from openarmature.observability.correlation import (
     _reset_active_dispatch,
     _reset_active_observers,
+    _reset_attempt_index,
     _reset_correlation_id,
+    _reset_fan_out_index,
     _reset_invocation_id,
+    _reset_namespace_prefix,
     _set_active_dispatch,
     _set_active_observers,
+    _set_attempt_index,
     _set_correlation_id,
+    _set_fan_out_index,
     _set_invocation_id,
+    _set_namespace_prefix,
 )
 
 from .edges import END, ConditionalEdge, EndSentinel, StaticEdge
@@ -558,51 +564,59 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             attempt_index = attempt_counter[0]
             attempt_counter[0] += 1
 
-            self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
-
+            # Calling-node identity for capability backends emitting
+            # from inside this attempt's scope (e.g., LLM provider's
+            # span hook). Per-attempt scope so retry middleware that
+            # re-enters innermost bumps the visible attempt_index.
+            attempt_token = _set_attempt_index(attempt_index)
             try:
-                partial = await node.run(s)
-            except Exception as e:
-                wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
-                self._dispatch_completed(
-                    context,
-                    current,
-                    namespace,
-                    step,
-                    s,
-                    error=wrapped,
-                    attempt_index=attempt_index,
-                )
-                raise
+                self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
+
+                try:
+                    partial = await node.run(s)
+                except Exception as e:
+                    wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                    self._dispatch_completed(
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=wrapped,
+                        attempt_index=attempt_index,
+                    )
+                    raise
+
+                try:
+                    merged = _merge_partial(s, partial, self.reducers, current)
+                except (ReducerError, StateValidationError) as e:
+                    self._dispatch_completed(
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=e,
+                        attempt_index=attempt_index,
+                    )
+                    raise
 
-            try:
-                merged = _merge_partial(s, partial, self.reducers, current)
-            except (ReducerError, StateValidationError) as e:
                 self._dispatch_completed(
                     context,
                     current,
                     namespace,
                     step,
                     s,
-                    error=e,
+                    post_state=merged,
                     attempt_index=attempt_index,
                 )
-                raise
-
-            self._dispatch_completed(
-                context,
-                current,
-                namespace,
-                step,
-                s,
-                post_state=merged,
-                attempt_index=attempt_index,
-            )
-            # Return the partial (not the merged state) so middleware sees
-            # the partial-update shape per pipeline-utilities §2. The
-            # engine's canonical merge against the original state happens
-            # below, after the chain returns.
-            return partial
+                # Return the partial (not the merged state) so middleware sees
+                # the partial-update shape per pipeline-utilities §2. The
+                # engine's canonical merge against the original state happens
+                # below, after the chain returns.
+                return partial
+            finally:
+                _reset_attempt_index(attempt_token)
 
         chain: ChainCall = compose_chain(
             list(self.middleware) + list(node.middleware),
@@ -612,12 +626,17 @@ async def innermost(s: Any) -> Mapping[str, Any]:
         # Spec observability §3 / Phase 6 LLM-span hook: capability
         # backends emitting from inside a node body (the
         # llm-provider span instrumentation in OpenAIProvider) need
-        # to find the observers active for THIS invocation. Set the
-        # ContextVar around the chain invocation; reset in
-        # ``try/finally`` so an exception escaping the chain still
-        # restores the prior value.
+        # to find the observers active for THIS invocation, which
+        # node is calling, and which fan-out instance (if any) the
+        # call belongs to. ``namespace_prefix`` and ``fan_out_index``
+        # are set in this outer scope (per-node, not per-attempt);
+        # ``attempt_index`` is set inside ``innermost`` per attempt.
+        # All four reset in ``try/finally`` so an exception escaping
+        # the chain still restores the prior values.
         observers_token = _set_active_observers(context.full_observers())
         dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
+        namespace_token = _set_namespace_prefix(namespace)
+        fan_out_token = _set_fan_out_index(context.fan_out_index)
         try:
             try:
                 final_partial = await chain(state)
@@ -628,6 +647,8 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                 # the chain unrecovered. Wrap as NodeException per §4.
                 raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
         finally:
+            _reset_fan_out_index(fan_out_token)
+            _reset_namespace_prefix(namespace_token)
             _reset_active_dispatch(dispatch_token)
             _reset_active_observers(observers_token)
         # Engine's canonical merge uses the ORIGINAL state per §2: "the
@@ -686,13 +707,18 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             list(self.middleware) + list(node.middleware),
             innermost,
         )
-        # Same active-observers scope as _step_function_node — parent
-        # middleware running before the descent should see the parent's
-        # observer set; the inner _invoke (called via ``node.run``)
-        # descends into its own context and sets a new scope from
-        # there.
+        # Same active-observers + calling-node scope as
+        # ``_step_function_node`` — parent middleware running before
+        # the descent should see the wrapper node's namespace +
+        # fan_out_index for any LLM-provider hook emissions.
+        # ``attempt_index`` defaults to 0 from the ContextVar; the
+        # subgraph wrapper has no engine-managed attempt counter
+        # (inner ``_step_function_node`` calls own their own).
+        namespace = context.namespace_prefix + (current,)
         observers_token = _set_active_observers(context.full_observers())
         dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
+        namespace_token = _set_namespace_prefix(namespace)
+        fan_out_token = _set_fan_out_index(context.fan_out_index)
 
         try:
             try:
@@ -707,6 +733,8 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                 # preserved.
                 raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
         finally:
+            _reset_fan_out_index(fan_out_token)
+            _reset_namespace_prefix(namespace_token)
             _reset_active_dispatch(dispatch_token)
             _reset_active_observers(observers_token)
         return _merge_partial(state, final_partial, self.reducers, current)
@@ -743,58 +771,68 @@ async def _step_fan_out_node(
         async def innermost(s: Any) -> Mapping[str, Any]:
             attempt_index = attempt_counter[0]
             attempt_counter[0] += 1
-            self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
+            attempt_token = _set_attempt_index(attempt_index)
             try:
-                partial = await node.run_with_context(s, context)
-            except RuntimeGraphError as e:
-                self._dispatch_completed(
-                    context, current, namespace, step, s, error=e, attempt_index=attempt_index
-                )
-                raise
-            except Exception as e:
-                wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
+                try:
+                    partial = await node.run_with_context(s, context)
+                except RuntimeGraphError as e:
+                    self._dispatch_completed(
+                        context, current, namespace, step, s, error=e, attempt_index=attempt_index
+                    )
+                    raise
+                except Exception as e:
+                    wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                    self._dispatch_completed(
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=wrapped,
+                        attempt_index=attempt_index,
+                    )
+                    raise wrapped from e
+
+                try:
+                    merged = _merge_partial(s, partial, self.reducers, current)
+                except (ReducerError, StateValidationError) as e:
+                    self._dispatch_completed(
+                        context, current, namespace, step, s, error=e, attempt_index=attempt_index
+                    )
+                    raise
+
                 self._dispatch_completed(
                     context,
                     current,
                     namespace,
                     step,
                     s,
-                    error=wrapped,
+                    post_state=merged,
                     attempt_index=attempt_index,
                 )
-                raise wrapped from e
-
-            try:
-                merged = _merge_partial(s, partial, self.reducers, current)
-            except (ReducerError, StateValidationError) as e:
-                self._dispatch_completed(
-                    context, current, namespace, step, s, error=e, attempt_index=attempt_index
-                )
-                raise
-
-            self._dispatch_completed(
-                context,
-                current,
-                namespace,
-                step,
-                s,
-                post_state=merged,
-                attempt_index=attempt_index,
-            )
-            return partial
+                return partial
+            finally:
+                _reset_attempt_index(attempt_token)
 
         chain: ChainCall = compose_chain(
             list(self.middleware) + list(node.middleware),
             innermost,
         )
 
         # Same observability §3 / LLM-span hook contract as
-        # _step_function_node: set the active observer set in scope
-        # around the chain invocation so capability backends emitting
-        # from inside the fan-out's parent dispatch (or any code
-        # running on its call stack) can find the observers.
+        # _step_function_node: set the active observer set, calling
+        # node identity, and dispatch scope around the chain
+        # invocation so capability backends emitting from inside the
+        # fan-out's parent dispatch (or any code running on its call
+        # stack) can find them. ``fan_out_index`` here is the parent
+        # context's view (the fan-out node from outside); per-instance
+        # values get set when the inner subgraph descends with the
+        # instance's index in its own context.
         observers_token = _set_active_observers(context.full_observers())
         dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
+        namespace_token = _set_namespace_prefix(namespace)
+        fan_out_token = _set_fan_out_index(context.fan_out_index)
         try:
             try:
                 final_partial = await chain(state)
@@ -803,6 +841,8 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             except Exception as e:
                 raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
         finally:
+            _reset_fan_out_index(fan_out_token)
+            _reset_namespace_prefix(namespace_token)
             _reset_active_dispatch(dispatch_token)
             _reset_active_observers(observers_token)
         merged_outer = _merge_partial(state, final_partial, self.reducers, current)

src/openarmature/llm/providers/openai.py

@@ -47,7 +47,12 @@
 
 from openarmature.graph.events import NodeEvent
 from openarmature.graph.state import State
-from openarmature.observability.correlation import current_dispatch
+from openarmature.observability.correlation import (
+    current_attempt_index,
+    current_dispatch,
+    current_fan_out_index,
+    current_namespace_prefix,
+)
 
 from ..errors import (
     ProviderAuthentication,
@@ -554,6 +559,15 @@ class _LlmEventState(State):
     LLM-span maps by it so concurrent ``complete()`` calls (e.g.,
     fan-out instances each calling the provider) don't collide on
     a single sentinel-namespace key.
+
+    ``calling_namespace_prefix``, ``calling_attempt_index``, and
+    ``calling_fan_out_index`` carry the calling node's identity so
+    the OTel observer can resolve the §5.5 "parent under calling
+    node" contract correctly under concurrent fan-out and retry.
+    Populated from the engine's ContextVars (set in
+    ``_step_*_node`` around node-body execution); fall back to
+    sentinel defaults (empty tuple, 0, ``None``) when the LLM
+    provider is called outside any node body.
     """
 
     call_id: str
@@ -571,6 +585,14 @@ class _LlmEventState(State):
     error_type: str | None = None
     error_message: str | None = None
     error_category: str | None = None
+    # Calling-node identity captured at dispatch time. The OTel
+    # observer reads these to look up the calling node's span in
+    # its (now-invocation_id-scoped) ``_open_spans`` map without relying on
+    # the OTel current-span context (which under concurrent fan-out
+    # can yield a sibling instance's span).
+    calling_namespace_prefix: tuple[str, ...] = ()
+    calling_attempt_index: int = 0
+    calling_fan_out_index: int | None = None
 
 
 def _make_llm_event(
@@ -611,6 +633,9 @@ def _make_llm_event(
         error_type=error_type,
         error_message=error_message,
         error_category=error_category,
+        calling_namespace_prefix=current_namespace_prefix(),
+        calling_attempt_index=current_attempt_index(),
+        calling_fan_out_index=current_fan_out_index(),
     )
     return NodeEvent(
         node_name="openarmature.llm.complete",

src/openarmature/observability/__init__.py

@@ -22,14 +22,20 @@
 
 from .correlation import (
     current_active_observers,
+    current_attempt_index,
     current_correlation_id,
     current_dispatch,
+    current_fan_out_index,
     current_invocation_id,
+    current_namespace_prefix,
 )
 
 __all__ = [
     "current_active_observers",
+    "current_attempt_index",
     "current_correlation_id",
     "current_dispatch",
+    "current_fan_out_index",
     "current_invocation_id",
+    "current_namespace_prefix",
 ]