observability: PR #19 round-3 review followups

Seven Copilot/code-quality findings addressed:

- **`_close_subgraph_span` empty-except comment** — added inline
  rationale inside the `except ValueError` block (the explanation
  was already above the `try`; rule wants it on the empty branch).
- **Concurrency limitation documented** — class docstring now spells
  out that ``OTelObserver`` instances are NOT safe across concurrent
  invocations (overlapping namespaces collide; close-prior-
  correlation_id closes other in-flight invocations). Recommended
  pattern: one observer per ``CompiledGraph`` for sequential
  workloads, fresh per-invocation observers for parallel services.
  Phase 6.1 tracks the proper correlation_id-scoped state fix.
- **Module docstring drift** — rewrote to reflect the
  ``(namespace, attempt_index, fan_out_index)`` leaf-span key (not
  the prior ``(trace_id, ...)`` shape) plus the dedicated dicts
  for subgraph dispatch, detached roots, and invocation spans.
- **Real ``invocation_id`` on the invocation span** — added
  ``current_invocation_id()`` ContextVar reader to ``correlation.py``
  mirroring ``current_correlation_id``; engine sets it in
  ``invoke()`` before worker creation. ``OTelObserver`` reads it
  when opening the invocation span and omits the attribute when
  None (replaces the prior ``"<unset>"`` sentinel).
- **`_make_llm_event` phase typing** — tightened ``phase: str``
  to ``Literal["started", "completed"]``; dropped the corresponding
  ``cast``.
- **Test-isolation for global TracerProvider** — saved + restored
  the prior global provider in both
  ``test_observer_uses_private_provider_not_global`` and the
  fixture 005 driver (the only sites that call
  ``set_tracer_provider``).
- **`test_correlation_id_and_invocation_id_are_structurally_distinct`** —
  rewrote to drive a real invocation with a checkpointer and assert
  ``record.correlation_id != record.invocation_id`` deterministically;
  also cross-checks via the new ``current_invocation_id()`` reader
  inside the node body. The prior assertion was on stdlib
  ``uuid.uuid4()``, not framework behavior.

The Phase 6.1 conformance-fillin tracker (in openarmature-coord) is
updated to add the concurrency-safe state scoping work alongside
the deferred fixtures.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -45,9 +45,11 @@
     _reset_active_dispatch,
     _reset_active_observers,
     _reset_correlation_id,
+    _reset_invocation_id,
     _set_active_dispatch,
     _set_active_observers,
     _set_correlation_id,
+    _set_invocation_id,
 )
 
 from .edges import END, ConditionalEdge, EndSentinel, StaticEdge
@@ -388,6 +390,7 @@ async def invoke(
         # public ``invoke``, so they don't re-set; see §3.1's
         # "per-invocation is OUTERMOST invoke" wording).
         correlation_token = _set_correlation_id(resolved_correlation_id)
+        invocation_token = _set_invocation_id(invocation_id)
         worker = asyncio.create_task(deliver_loop(queue))
         self._active_workers.add(worker)
         # Auto-prune: when the worker completes (after the sentinel is
@@ -397,6 +400,7 @@ async def invoke(
         try:
             return await self._invoke(starting_state, context)
         finally:
+            _reset_invocation_id(invocation_token)
             _reset_correlation_id(correlation_token)
             # Sentinel terminates the worker after it processes events
             # already on the queue (including any error event we just

src/openarmature/llm/providers/openai.py

@@ -39,7 +39,7 @@
 
 import json
 from collections.abc import Sequence
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 import httpx
 from pydantic import ValidationError
@@ -525,7 +525,7 @@ def _looks_like_model_not_loaded(message: object) -> bool:
 
 
 def _make_llm_event(
-    phase: str,
+    phase: Literal["started", "completed"],
     *,
     model: str,
     finish_reason: FinishReason | None = None,
@@ -567,7 +567,7 @@ def _make_llm_event(
         node_name="openarmature.llm.complete",
         namespace=("openarmature.llm.complete",),
         step=-1,
-        phase=cast("Any", phase),
+        phase=phase,
         # ``pre_state`` is overloaded here as the LLM-event payload
         # carrier — see the docstring above.
         pre_state=cast("Any", {"llm_event": payload}),

src/openarmature/observability/__init__.py

@@ -24,10 +24,12 @@
     current_active_observers,
     current_correlation_id,
     current_dispatch,
+    current_invocation_id,
 )
 
 __all__ = [
     "current_active_observers",
     "current_correlation_id",
     "current_dispatch",
+    "current_invocation_id",
 ]

src/openarmature/observability/correlation.py

@@ -76,6 +76,46 @@ def _reset_correlation_id(token: Token[str | None]) -> None:
     _correlation_id_var.reset(token)
 
 
+# ---------------------------------------------------------------------------
+# Invocation ID — spec observability §5.1
+#
+# The framework-generated UUIDv4 that ties spans of one invocation
+# together within a single backend. Distinct from ``correlation_id``
+# (which is the cross-backend join key, caller-supplied or auto-
+# generated). Engine sets this ContextVar in ``invoke()`` BEFORE the
+# delivery worker is created so the worker's captured context sees
+# the right value.
+# ---------------------------------------------------------------------------
+
+
+_invocation_id_var: ContextVar[str | None] = ContextVar("openarmature.invocation_id", default=None)
+
+
+def current_invocation_id() -> str | None:
+    """Return the engine-minted invocation ID for the current
+    invocation, or ``None`` if no openarmature invocation is in
+    scope.
+
+    Per spec observability §5.1 every invocation produces a unique
+    UUIDv4 ``invocation_id``, framework-generated, surfaced as the
+    ``openarmature.invocation_id`` attribute on the invocation span
+    + on every per-backend record. This is the public reader for
+    backend mappings (OTel, future Langfuse) that need to populate
+    that attribute.
+    """
+    return _invocation_id_var.get()
+
+
+def _set_invocation_id(value: str) -> Token[str | None]:
+    """Set the invocation ID for the current invocation. Internal —
+    engine-only."""
+    return _invocation_id_var.set(value)
+
+
+def _reset_invocation_id(token: Token[str | None]) -> None:
+    _invocation_id_var.reset(token)
+
+
 # ---------------------------------------------------------------------------
 # Active observer set — for capability backends emitting from outside the
 # engine's per-step path (llm-provider span hook in Phase 6, future
@@ -170,14 +210,17 @@ def _reset_active_dispatch(
     "current_active_observers",
     "current_correlation_id",
     "current_dispatch",
+    "current_invocation_id",
     # Engine-internal lifecycle helpers — exported so the engine in
     # ``openarmature.graph.compiled`` can drive set/reset without
     # pyright's strict ``reportUnusedFunction`` flagging them as
     # dead. Underscore-prefixed; not part of the user-facing API.
     "_reset_active_dispatch",
     "_reset_active_observers",
     "_reset_correlation_id",
+    "_reset_invocation_id",
     "_set_active_dispatch",
     "_set_active_observers",
     "_set_correlation_id",
+    "_set_invocation_id",
 ]

src/openarmature/observability/otel/observer.py

@@ -3,10 +3,27 @@
 The observer subscribes to all three §6 phases (``started``,
 ``completed``, ``checkpoint_saved``) plus the LLM-provider events the
 ``OpenAIProvider`` enqueues from inside node bodies. On a ``started``
-event it opens a span and pushes it onto an in-flight map keyed by
-``(trace_id, namespace, attempt_index, fan_out_index)``; on the
-matching ``completed`` event it pops the span, applies §4.2 status
-mapping, and closes it.
+event it opens a leaf span and pushes it onto an in-flight map keyed
+by ``(namespace, attempt_index, fan_out_index)``; on the matching
+``completed`` event it pops the span, applies §4.2 status mapping,
+and closes it.
+
+Subtree isolation lives in dedicated dicts rather than the leaf-span
+key:
+
+- ``_subgraph_spans`` — synthetic subgraph dispatch spans (the
+  engine wrapper is transparent per fixture 013, but observability
+  §4.5 mandates a span). Keyed by namespace prefix. Open lazily on
+  the first deeper-namespace event, close when subsequent events
+  leave the prefix.
+- ``_detached_roots`` — root spans for detached subgraphs (§4.4)
+  and per-instance detached fan-out roots. Each lives in its own
+  fresh ``trace_id``; the parent's dispatch span carries an OTel
+  :class:`Link` to the detached trace.
+- ``_invocation_span`` — root invocation span keyed by
+  ``correlation_id``. Closed eagerly when a new correlation_id's
+  first event arrives (or explicitly via
+  :meth:`close_invocation` / :meth:`shutdown`).
 
 Spans are emitted through a **private** :class:`TracerProvider`
 constructed by this observer — never the OTel global. Per spec §6
@@ -18,9 +35,11 @@
 Detached trace mode (§4.4) is implemented by minting a fresh
 :class:`SpanContext` with a new ``trace_id`` when entering a
 configured-detached subgraph or fan-out; the parent's dispatch span
-carries an OTel :class:`Link` to the detached trace. The span-stack
-key includes ``trace_id`` so detached sub-trees and the parent trace
-maintain separate stacks naturally.
+carries an OTel :class:`Link` to the detached trace. Inner-event
+parent resolution checks the per-fan-out-instance key
+(``namespace[:1] + (str(fan_out_index),)``) before the generic
+prefix scan, so per-instance detached roots win without depending
+on attach-then-resolve ordering.
 """
 
 from __future__ import annotations
@@ -111,6 +130,21 @@ class OTelObserver:
       canonical source of LLM spans.
     - ``spec_version`` — string surfaced as
       ``openarmature.graph.spec_version`` on the invocation span.
+
+    **Concurrency model.** A single ``OTelObserver`` instance is
+    SAFE for sequential invocations on the same graph (one
+    ``invoke()`` followed by another, with the observer reused
+    between). It is NOT safe to share an instance across CONCURRENT
+    invocations — the internal span state (``_open_spans``,
+    ``_subgraph_spans``, ``_detached_roots``, ``_invocation_span``)
+    is keyed without per-invocation scoping, so overlapping
+    namespaces collide and the close-prior-correlation_id logic in
+    ``_handle_started`` would close another in-flight invocation's
+    span. Recommended pattern: one observer per ``CompiledGraph``
+    instance for sequential workloads; for ASGI / batch / parallel
+    invocation services, attach a fresh observer per invocation
+    (via ``invoke(observers=[...])``) until the Phase 6.1
+    correlation_id-scoped state lands.
     """
 
     span_processor: SpanProcessor
@@ -363,18 +397,22 @@ def _handle_llm_event(self, event: NodeEvent) -> None:
 
     def _open_invocation_span(self, correlation_id: str, event: NodeEvent) -> None:
         """Open the root invocation span for a new invocation."""
+        from openarmature.observability.correlation import current_invocation_id
+
         # The first event we receive carries the entry node's
         # name; treat it as the invocation's entry_node attribute.
+        # ``invocation_id`` is set by the engine in ``invoke()`` via
+        # the ``current_invocation_id`` ContextVar; if the observer
+        # somehow runs outside an engine invocation (None) we omit
+        # the attribute rather than emitting a misleading sentinel.
         attrs: dict[str, Any] = {
-            "openarmature.invocation_id": "<unset>",  # set below from context
             "openarmature.graph.entry_node": event.node_name,
             "openarmature.graph.spec_version": self.spec_version,
             "openarmature.correlation_id": correlation_id,
         }
-        # We don't have the engine's invocation_id directly without
-        # threading it through. Phase 6 follow-up could surface it
-        # via the correlation module; for now leave it blank for
-        # the conformance fixtures that don't strictly need it.
+        invocation_id = current_invocation_id()
+        if invocation_id is not None:
+            attrs["openarmature.invocation_id"] = invocation_id
         span = self._tracer.start_span(
             name="openarmature.invocation",
             kind=SpanKind.INTERNAL,
@@ -535,6 +573,10 @@ def _close_subgraph_span(self, prefix: tuple[str, ...]) -> None:
             try:
                 detach(cast("Any", open_span.token))
             except ValueError:
+                # Token was created in a different OTel context —
+                # cross-context detach raises here. The span has
+                # ended; the leaked context entry is cosmetic and
+                # unwinds when the worker task exits.
                 pass
 
     def _open_detached_subgraph_root(self, prefix: tuple[str, ...]) -> None:

tests/conformance/test_observability.py

@@ -335,54 +335,56 @@ async def _run_fixture_005_case(case: Mapping[str, Any]) -> None:
     )
 
     # Optional second exporter on the OTel global provider — sub-case 3.
+    # Save the prior global provider so we can restore it after the
+    # case (otherwise the global state leaks to subsequent tests).
     global_exporter: InMemorySpanExporter | None = None
+    prior_global = otel_trace.get_tracer_provider() if caller_global_active else None
     if caller_global_active:
         global_exporter = InMemorySpanExporter()
         global_provider = TracerProvider()
         global_provider.add_span_processor(SimpleSpanProcessor(global_exporter))
         otel_trace.set_tracer_provider(global_provider)
 
-    private_exporter = InMemorySpanExporter()
-    observer = OTelObserver(
-        span_processor=SimpleSpanProcessor(private_exporter),
-        disable_llm_spans=disable_llm_spans,
-    )
-
-    # Build a graph whose entry node calls a mock LLM provider.
-    graph, _ = _build_graph_with_mock_llm(case)
-    graph.attach_observer(observer)
+    try:
+        private_exporter = InMemorySpanExporter()
+        observer = OTelObserver(
+            span_processor=SimpleSpanProcessor(private_exporter),
+            disable_llm_spans=disable_llm_spans,
+        )
 
-    # Drive the graph. The ``calls_llm`` node body reads the mock
-    # responses set up in ``_build_graph_with_mock_llm`` (httpx
-    # MockTransport keyed off the response queue).
-    initial_state_cls = graph.state_cls
-    final = await graph.invoke(initial_state_cls())
-    await graph.drain()
-    observer.shutdown()
-    private_spans = private_exporter.get_finished_spans()
+        # Build a graph whose entry node calls a mock LLM provider.
+        graph, _ = _build_graph_with_mock_llm(case)
+        graph.attach_observer(observer)
 
-    # Sub-case 3: external span emitted by the harness through the
-    # global tracer (simulating auto-instrumentation).
-    if caller_global_active:
-        global_tracer = otel_trace.get_tracer("external-instrumentation")
-        with global_tracer.start_as_current_span("external.llm.call"):
-            pass
-        assert global_exporter is not None
-        global_spans = global_exporter.get_finished_spans()
-        assert len(global_spans) == 1, (
-            f"global exporter MUST see exactly one external span; got {len(global_spans)}"
-        )
-        assert global_spans[0].name == "external.llm.call"
-        # The load-bearing isolation check.
-        for s in global_spans:
-            assert not s.name.startswith("openarmature."), (
-                f"openarmature spans MUST NOT leak to the global provider; got {s.name!r}"
+        # Drive the graph. The ``calls_llm`` node body reads the mock
+        # responses set up in ``_build_graph_with_mock_llm`` (httpx
+        # MockTransport keyed off the response queue).
+        initial_state_cls = graph.state_cls
+        final = await graph.invoke(initial_state_cls())
+        await graph.drain()
+        observer.shutdown()
+        private_spans = private_exporter.get_finished_spans()
+
+        # Sub-case 3: external span emitted by the harness through the
+        # global tracer (simulating auto-instrumentation).
+        if caller_global_active:
+            global_tracer = otel_trace.get_tracer("external-instrumentation")
+            with global_tracer.start_as_current_span("external.llm.call"):
+                pass
+            assert global_exporter is not None
+            global_spans = global_exporter.get_finished_spans()
+            assert len(global_spans) == 1, (
+                f"global exporter MUST see exactly one external span; got {len(global_spans)}"
             )
-        # Reset the global provider so the next test in the file
-        # doesn't see a stale state. The OTel SDK doesn't expose a
-        # public reset; ``set_tracer_provider`` with a fresh empty
-        # provider is the closest equivalent.
-        otel_trace.set_tracer_provider(TracerProvider())
+            assert global_spans[0].name == "external.llm.call"
+            # The load-bearing isolation check.
+            for s in global_spans:
+                assert not s.name.startswith("openarmature."), (
+                    f"openarmature spans MUST NOT leak to the global provider; got {s.name!r}"
+                )
+    finally:
+        if caller_global_active and prior_global is not None:
+            otel_trace.set_tracer_provider(prior_global)
 
     # Common assertions: the LLM span presence/absence + (when
     # present) attributes + parent-child to the calling node.