observability: PR #19 round-7 review followups (#20)

* observability: PR #19 round-7 review followups

- Guard cross-context detach in _handle_completed and
  _handle_llm_event so out-of-LIFO completion under concurrent
  fan-out doesn't crash the runtime (proper fix in Phase 6.1).
- Add call_id (UUIDv4) on _LlmEventState so concurrent LLM
  calls don't collide on the constant LLM-namespace key in
  _open_llm_spans.
- Switch importorskip target to opentelemetry.sdk.trace in
  both observability test modules so a partial install (API
  but no SDK) skips cleanly instead of erroring at collection.
- Drain leaf spans, LLM spans, detached roots, and synthetic
  subgraph dispatch spans on OTelObserver.shutdown() — now
  matches the docstring contract regardless of where the
  invocation ended.
- Snapshot + restore root logger handlers/filters in
  test_install_log_bridge_is_idempotent so it doesn't leak
  state into other tests.

* otel: address PR #20 review

- Add inline comment in _drain_open_span's except ValueError so
  CodeQL/code-quality stop flagging the bare swallow. Matches the
  existing pattern at the four other cross-context-detach sites.
- Fix shutdown() ordering. The docstring promised child->parent
  closure but iterating `list(map.keys())` follows insertion
  order (parents first), so children outlived their parents in
  traces. Now drains _open_llm_spans (deepest leaves) before
  _open_spans; sorts the depth-bearing maps (_open_spans,
  _detached_roots, _subgraph_spans) by -len(namespace) so deeper
  entries close first. _open_llm_spans (call_id keys) and
  _invocation_span (one entry per cid) need no within-map sort.

Author: chris-colinsky

src/openarmature/llm/providers/openai.py

@@ -38,6 +38,7 @@
 from __future__ import annotations
 
 import json
+import uuid
 from collections.abc import Sequence
 from typing import Any, Literal, cast
 
@@ -203,21 +204,30 @@ async def complete(
         # ``None`` outside an openarmature invocation (direct
         # provider use in scripts/tests), in which case the call
         # proceeds without span emission.
+        #
+        # ``call_id`` is minted once per ``complete()`` call and
+        # threaded through both events of the pair. Backend
+        # observers key their in-flight LLM-span maps by it so
+        # concurrent ``complete()`` calls (e.g., fan-out instances
+        # each calling this provider) don't collide on the
+        # constant ``("openarmature.llm.complete",)`` sentinel.
         dispatch = current_dispatch()
+        call_id = str(uuid.uuid4())
         if dispatch is not None:
-            dispatch(_make_llm_event("started", model=self.model))
+            dispatch(_make_llm_event("started", call_id=call_id, model=self.model))
 
         try:
             response = await self._do_complete(body)
         except Exception as exc:
             if dispatch is not None:
-                dispatch(_make_llm_event("completed", model=self.model, error=exc))
+                dispatch(_make_llm_event("completed", call_id=call_id, model=self.model, error=exc))
             raise
 
         if dispatch is not None:
             dispatch(
                 _make_llm_event(
                     "completed",
+                    call_id=call_id,
                     model=self.model,
                     finish_reason=response.finish_reason,
                     usage=response.usage,
@@ -537,8 +547,16 @@ class _LlmEventState(State):
     Langfuse / Datadog adapters) recognize the
     ``("openarmature.llm.complete",)`` namespace sentinel and read
     these fields directly via attribute access.
+
+    ``call_id`` is the per-call disambiguator: a UUIDv4 minted in
+    ``OpenAIProvider.complete`` and shared between the started /
+    completed event pair. Backend observers key their in-flight
+    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.
     """
 
+    call_id: str
     model: str
     finish_reason: str | None = None
     prompt_tokens: int | None = None
@@ -558,6 +576,7 @@ class _LlmEventState(State):
 def _make_llm_event(
     phase: Literal["started", "completed"],
     *,
+    call_id: str,
     model: str,
     finish_reason: FinishReason | None = None,
     usage: Usage | None = None,
@@ -569,6 +588,9 @@ def _make_llm_event(
     span instead of a node span. Backend-specific attribute extraction
     reads ``model``, ``finish_reason``, and ``usage`` from
     ``pre_state`` directly via attribute access.
+
+    ``call_id`` MUST be the same string on the started/completed
+    pair so the observer can match them under concurrency.
     """
     error_type: str | None = None
     error_message: str | None = None
@@ -580,6 +602,7 @@ def _make_llm_event(
         if isinstance(category, str):
             error_category = category
     payload = _LlmEventState(
+        call_id=call_id,
         model=model,
         finish_reason=finish_reason,
         prompt_tokens=usage.prompt_tokens if usage is not None else None,

src/openarmature/observability/otel/observer.py

@@ -181,6 +181,16 @@ class OTelObserver:
     _invocation_span: dict[str, _OpenSpan] = field(
         init=False, repr=False, default_factory=dict[str, _OpenSpan]
     )
+    # Per-LLM-call span tracking, keyed by ``call_id`` (UUIDv4
+    # minted in ``OpenAIProvider.complete`` per call). Concurrent
+    # ``complete()`` calls under fan-out instances would collide on
+    # a constant ``_LLM_NAMESPACE`` key in ``_open_spans``; the
+    # call_id-keyed dict disambiguates them. The deeper
+    # parent-attribution issue under concurrency (calling node's
+    # context not threaded through) is Phase 6.1 work.
+    _open_llm_spans: dict[str, _OpenSpan] = field(
+        init=False, repr=False, default_factory=dict[str, _OpenSpan]
+    )
     # Synthetic subgraph dispatch spans: the engine wrapper for
     # ``add_subgraph_node`` is transparent (graph-engine fixture 013
     # — no started/completed events of its own), but observability
@@ -306,7 +316,19 @@ def _handle_completed(self, event: NodeEvent) -> None:
         span.end()
         token = open_span.token
         if token is not None:
-            detach(cast("Any", token))
+            try:
+                detach(cast("Any", token))
+            except ValueError:
+                # Out-of-LIFO detach: under interleaved
+                # fan-out-instance events, ``completed`` for an
+                # earlier-started span can arrive while a
+                # later-started span's token is on top of the OTel
+                # context stack. The proper fix is Phase 6.1's
+                # "don't hold attach tokens across event
+                # boundaries"; the guard here keeps Phase 6.0
+                # robust to the interleave without corrupting
+                # subsequent span attribution.
+                pass
         # If this was a detached root, drop the root entry so a
         # subsequent re-entry mints a fresh trace.
         self._detached_roots.pop(event.namespace, None)
@@ -371,9 +393,9 @@ def _handle_llm_event(self, event: NodeEvent) -> None:
                 attributes=attrs,
             )
             token = attach(set_span_in_context(span))
-            self._open_spans[self._llm_key()] = _OpenSpan(span=span, token=token)
+            self._open_llm_spans[payload.call_id] = _OpenSpan(span=span, token=token)
         elif event.phase == "completed":
-            open_span = self._open_spans.pop(self._llm_key(), None)
+            open_span = self._open_llm_spans.pop(payload.call_id, None)
             if open_span is None:
                 return
             span = open_span.span
@@ -397,7 +419,16 @@ def _handle_llm_event(self, event: NodeEvent) -> None:
             else:
                 span.set_status(Status(StatusCode.OK))
             span.end()
-            detach(cast("Any", open_span.token))
+            try:
+                detach(cast("Any", open_span.token))
+            except ValueError:
+                # See ``_handle_completed`` for the rationale —
+                # out-of-LIFO detach under concurrent fan-out
+                # instance events is a known Phase 6.0 limitation;
+                # the proper fix is Phase 6.1's
+                # "don't hold attach tokens across event
+                # boundaries."
+                pass
 
     # ------------------------------------------------------------------
     # Helpers
@@ -432,11 +463,6 @@ def _open_invocation_span(self, correlation_id: str, event: NodeEvent) -> None:
     def _key_for(self, event: NodeEvent) -> _StackKey:
         return (event.namespace, event.attempt_index, event.fan_out_index)
 
-    def _llm_key(self) -> _StackKey:
-        """LLM events are unique within a node body — only one LLM
-        call can be open at a time per active span scope."""
-        return (_LLM_NAMESPACE, 0, None)
-
     def _resolve_parent_context(self, event: NodeEvent) -> object:
         """Return the OTel context to use as the parent for this
         event's span. Walks namespace ancestors finding the
@@ -718,6 +744,24 @@ def _close_detached_root(self, prefix: tuple[str, ...]) -> None:
                 # context entry leaks cosmetically.
                 pass
 
+    @staticmethod
+    def _drain_open_span(open_span: _OpenSpan) -> None:
+        """Close an open span as an orphan during shutdown: OK
+        status, end, and try-detach the token. No paired completed
+        event will arrive, so we don't have an error category to
+        record. Cross-context detach is swallowed (the worker that
+        owns the context eventually unwinds)."""
+        open_span.span.set_status(Status(StatusCode.OK))
+        open_span.span.end()
+        if open_span.token is not None:
+            try:
+                detach(cast("Any", open_span.token))
+            except ValueError:
+                # Cross-context detach during shutdown — token
+                # created in a different OTel context. Span has
+                # ended; ignore.
+                pass
+
     def _fan_out_node_span_key(self, prefix: tuple[str, ...]) -> _StackKey:
         """Build the lookup key for a fan-out node's own span (the
         parent dispatch span). Fan-out node has no attempt_index ≠ 0
@@ -782,8 +826,28 @@ def _close_invocation_span(self, correlation_id: str) -> None:
         open_span.span.end()
 
     def shutdown(self) -> None:
-        """Close any still-open invocation/detached spans and shut
-        down the underlying provider."""
+        """Close any still-open spans and shut down the underlying
+        provider. Walks state maps in child→parent order — LLM
+        spans (deepest leaves) before leaf node spans before
+        subgraph dispatch / detached roots before invocation spans
+        — and within each depth-bearing map sorts by namespace
+        depth (deepest first) so children end before their
+        parents. ``_open_llm_spans`` (call_id keys) and
+        ``_invocation_span`` (one entry per correlation_id) carry
+        no internal nesting and drain in insertion order.
+        Idempotent."""
+        for call_id in list(self._open_llm_spans.keys()):
+            open_span = self._open_llm_spans.pop(call_id, None)
+            if open_span is not None:
+                self._drain_open_span(open_span)
+        for key in sorted(self._open_spans.keys(), key=lambda k: -len(k[0])):
+            open_span = self._open_spans.pop(key, None)
+            if open_span is not None:
+                self._drain_open_span(open_span)
+        for prefix in sorted(self._detached_roots.keys(), key=lambda k: -len(k)):
+            self._close_detached_root(prefix)
+        for prefix in sorted(self._subgraph_spans.keys(), key=lambda k: -len(k)):
+            self._close_subgraph_span(prefix)
         for cid in list(self._invocation_span.keys()):
             self._close_invocation_span(cid)
         self._provider.shutdown()

tests/conformance/test_observability.py

@@ -38,7 +38,7 @@
 # import time when the extras are missing, which would fail
 # collection rather than skipping cleanly. Mirrors the pattern in
 # ``tests/unit/test_observability_otel.py``.
-pytest.importorskip("opentelemetry.trace")
+pytest.importorskip("opentelemetry.sdk.trace")
 
 from openarmature.observability.otel import OTelObserver  # noqa: E402
 

tests/unit/test_observability_otel.py

@@ -26,7 +26,7 @@
 import pytest
 
 # Skip the entire module if otel extras aren't installed.
-pytest.importorskip("opentelemetry.trace")
+pytest.importorskip("opentelemetry.sdk.trace")
 
 from opentelemetry import trace as otel_trace
 from opentelemetry.sdk.trace import TracerProvider
@@ -274,7 +274,7 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
         namespace=("openarmature.llm.complete",),
         step=-1,
         phase="started",
-        pre_state=_LlmEventState(model="test-m"),
+        pre_state=_LlmEventState(call_id="test-call-1", model="test-m"),
         post_state=None,
         error=None,
         parent_states=(),
@@ -284,7 +284,7 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
         namespace=("openarmature.llm.complete",),
         step=-1,
         phase="completed",
-        pre_state=_LlmEventState(model="test-m", finish_reason="stop"),
+        pre_state=_LlmEventState(call_id="test-call-1", model="test-m", finish_reason="stop"),
         post_state=None,
         error=None,
         parent_states=(),
@@ -349,9 +349,19 @@ def test_install_log_bridge_is_idempotent() -> None:
     LoggingHandler on the root logger."""
     from opentelemetry.sdk._logs import LoggerProvider
 
-    provider = LoggerProvider()
-    install_log_bridge(provider)
-    handler_count_before = len(logging.getLogger().handlers)
-    install_log_bridge(provider)
-    handler_count_after = len(logging.getLogger().handlers)
-    assert handler_count_before == handler_count_after
+    root = logging.getLogger()
+    prior_handlers = list(root.handlers)
+    prior_filters = list(root.filters)
+    try:
+        provider = LoggerProvider()
+        install_log_bridge(provider)
+        handler_count_before = len(root.handlers)
+        install_log_bridge(provider)
+        handler_count_after = len(root.handlers)
+        assert handler_count_before == handler_count_after
+    finally:
+        # install_log_bridge mutates the process-wide root logger;
+        # restore the prior handler + filter set so this test does
+        # not leak state into others.
+        root.handlers[:] = prior_handlers
+        root.filters[:] = prior_filters