observability: spec v0.7 OTel mapping (proposal 0007) — phase 6.0

.github/workflows/ci.yml

@@ -36,7 +36,14 @@ jobs:
       # version developers run locally.
 
       - name: Sync deps
-        run: uv sync --frozen
+        # ``--all-extras`` installs every optional-dependency group
+        # (currently ``[otel]``) so pyright type-checks the
+        # OTel-using modules with real types and tests covering
+        # extras-gated code paths run end-to-end. Without this,
+        # ``opentelemetry.*`` symbols come back Unknown and pyright
+        # produces hundreds of false-positive errors on
+        # ``tests/unit/test_observability_otel.py``.
+        run: uv sync --frozen --all-extras
 
       - name: Lint (ruff check)
         run: uv run ruff check .

pyproject.toml

@@ -16,6 +16,21 @@ dependencies = [
     "httpx>=0.27",
 ]
 
+[project.optional-dependencies]
+# Spec observability §6 OTel mapping. Optional per charter §3.1
+# principle 5 — backend mappings are pluggable rather than core
+# dependencies. Plan to lift into a sibling ``openarmature-otel``
+# package at v1.0 launch alongside ``openarmature-eval``.
+otel = [
+    # Upper bound guards against the SDK 2.x release that removes
+    # ``opentelemetry.sdk._logs.LoggingHandler`` (currently emits a
+    # DeprecationWarning). Migration to
+    # ``opentelemetry-instrumentation-logging`` lands in Phase 6.1
+    # before bumping the upper bound.
+    "opentelemetry-api>=1.27,<2",
+    "opentelemetry-sdk>=1.27,<2",
+]
+
 [project.urls]
 Repository = "https://github.com/LunarCommand/openarmature-python"
 Specification = "https://github.com/LunarCommand/openarmature-spec"

src/openarmature/graph/compiled.py

@@ -41,6 +41,14 @@
     CheckpointRecord,
     NodePosition,
 )
+from openarmature.observability.correlation import (
+    _reset_active_dispatch,
+    _reset_active_observers,
+    _reset_correlation_id,
+    _set_active_dispatch,
+    _set_active_observers,
+    _set_correlation_id,
+)
 
 from .edges import END, ConditionalEdge, EndSentinel, StaticEdge
 from .errors import (
@@ -369,6 +377,17 @@ async def invoke(
             pending_resume_states=pending_resume_states,
             resume_invocation=resume_invocation,
         )
+        # Spec observability §3.1: the correlation_id MUST be readable
+        # from anywhere within the invocation's async call tree via the
+        # language's idiomatic context primitive. Set the ContextVar
+        # BEFORE creating the delivery worker so the worker's captured
+        # context sees the correlation_id (asyncio.create_task snapshots
+        # the current Context at creation time). Reset on return so
+        # subsequent invocations get a fresh slate. Nested ``invoke()``
+        # calls (subgraph-as-node uses ``_invoke`` directly, not the
+        # 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)
         worker = asyncio.create_task(deliver_loop(queue))
         self._active_workers.add(worker)
         # Auto-prune: when the worker completes (after the sentinel is
@@ -378,6 +397,7 @@ async def invoke(
         try:
             return await self._invoke(starting_state, context)
         finally:
+            _reset_correlation_id(correlation_token)
             # Sentinel terminates the worker after it processes events
             # already on the queue (including any error event we just
             # dispatched on the failure path). Drain semantics live on
@@ -585,14 +605,27 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             innermost,
         )
 
+        # 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.
+        observers_token = _set_active_observers(context.full_observers())
+        dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
         try:
-            final_partial = await chain(state)
-        except RuntimeGraphError:
-            raise
-        except Exception as e:
-            # A raw exception (node-raised or middleware-raised) escaped
-            # the chain unrecovered. Wrap as NodeException per §4.
-            raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+            try:
+                final_partial = await chain(state)
+            except RuntimeGraphError:
+                raise
+            except Exception as e:
+                # A raw exception (node-raised or middleware-raised) escaped
+                # the chain unrecovered. Wrap as NodeException per §4.
+                raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+        finally:
+            _reset_active_dispatch(dispatch_token)
+            _reset_active_observers(observers_token)
         # Engine's canonical merge uses the ORIGINAL state per §2: "the
         # transformed state is passed to ``next``, NOT to the engine's
         # merge step." If middleware transformed state mid-chain, the
@@ -649,18 +682,29 @@ 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.
+        observers_token = _set_active_observers(context.full_observers())
+        dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
 
         try:
-            final_partial = await chain(state)
-        except RuntimeGraphError:
-            raise
-        except Exception as e:
-            # Same wrap as _step_function_node: a raw exception escaping
-            # the parent's middleware chain (e.g., a middleware bug or a
-            # projection error) becomes NodeException tagged with the
-            # SubgraphNode's wrapper name so §4 recoverable_state is
-            # preserved.
-            raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+            try:
+                final_partial = await chain(state)
+            except RuntimeGraphError:
+                raise
+            except Exception as e:
+                # Same wrap as _step_function_node: a raw exception escaping
+                # the parent's middleware chain (e.g., a middleware bug or a
+                # projection error) becomes NodeException tagged with the
+                # SubgraphNode's wrapper name so §4 recoverable_state is
+                # preserved.
+                raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+        finally:
+            _reset_active_dispatch(dispatch_token)
+            _reset_active_observers(observers_token)
         return _merge_partial(state, final_partial, self.reducers, current)
 
     async def _step_fan_out_node(
@@ -740,12 +784,23 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             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.
+        observers_token = _set_active_observers(context.full_observers())
+        dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event))
         try:
-            final_partial = await chain(state)
-        except RuntimeGraphError:
-            raise
-        except Exception as e:
-            raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+            try:
+                final_partial = await chain(state)
+            except RuntimeGraphError:
+                raise
+            except Exception as e:
+                raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+        finally:
+            _reset_active_dispatch(dispatch_token)
+            _reset_active_observers(observers_token)
         merged_outer = _merge_partial(state, final_partial, self.reducers, current)
         # Spec §10.3 + §10.7: the fan-out's own completion DOES save —
         # one record once the fan-out as a whole has finished and

src/openarmature/llm/providers/openai.py

@@ -44,6 +44,9 @@
 import httpx
 from pydantic import ValidationError
 
+from openarmature.graph.events import NodeEvent
+from openarmature.observability.correlation import current_dispatch
+
 from ..errors import (
     ProviderAuthentication,
     ProviderInvalidModel,
@@ -190,6 +193,41 @@ async def complete(
         validate_tools(tools)
         body = self._build_request_body(messages, tools, config)
 
+        # Spec observability §5.5 LLM provider span: when an
+        # observability backend is active in the current invocation,
+        # emit a started/completed event pair around the wire call so
+        # the backend can build a span. Queue-mediated dispatch
+        # preserves spec §6 serial event ordering across all event
+        # sources within an invocation. ``current_dispatch()`` returns
+        # ``None`` outside an openarmature invocation (direct
+        # provider use in scripts/tests), in which case the call
+        # proceeds without span emission.
+        dispatch = current_dispatch()
+        if dispatch is not None:
+            dispatch(_make_llm_event("started", 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))
+            raise
+
+        if dispatch is not None:
+            dispatch(
+                _make_llm_event(
+                    "completed",
+                    model=self.model,
+                    finish_reason=response.finish_reason,
+                    usage=response.usage,
+                )
+            )
+        return response
+
+    async def _do_complete(self, body: dict[str, Any]) -> Response:
+        """Wire-call helper: separated from ``complete()`` so the
+        LLM-provider span hook in ``complete()`` can wrap success and
+        failure paths uniformly."""
         try:
             resp = await self._client.post("/v1/chat/completions", json=body)
         except httpx.HTTPError as exc:
@@ -481,6 +519,64 @@ def _looks_like_model_not_loaded(message: object) -> bool:
     return "not loaded" in lower or "loading" in lower
 
 
+# ---------------------------------------------------------------------------
+# Observability §5.5 LLM provider span event helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_llm_event(
+    phase: str,
+    *,
+    model: str,
+    finish_reason: FinishReason | None = None,
+    usage: Usage | None = None,
+    error: BaseException | None = None,
+) -> NodeEvent:
+    """Build a NodeEvent-shaped record for the engine's delivery
+    queue. The OTel observer (or any backend mapping) recognises the
+    sentinel ``node_name`` and ``namespace`` and emits an LLM-specific
+    span instead of a node span. Backend-specific attribute extraction
+    reads ``model``, ``finish_reason``, and ``usage`` from
+    ``pre_state``'s ``llm_event`` payload.
+
+    The pre_state field is reused as the carrier for LLM event detail
+    because NodeEvent's shape is fixed (graph-engine §6) and adding
+    ad-hoc fields would break observers that pattern-match on the
+    existing shape. Backend mappings know to inspect
+    ``event.pre_state['llm_event']`` when the namespace is
+    ``("openarmature.llm.complete",)``.
+    """
+    payload: dict[str, Any] = {"model": model}
+    if finish_reason is not None:
+        payload["finish_reason"] = finish_reason
+    if usage is not None:
+        payload["prompt_tokens"] = usage.prompt_tokens
+        payload["completion_tokens"] = usage.completion_tokens
+        payload["total_tokens"] = usage.total_tokens
+    if error is not None:
+        # The engine's NodeEvent.error type is RuntimeGraphError, but
+        # llm-provider errors aren't graph-engine §4 categories. Carry
+        # the exception detail in the payload instead so backends can
+        # surface it without our needing to wrap as RuntimeGraphError.
+        payload["error_type"] = type(error).__name__
+        payload["error_message"] = str(error)
+        category = getattr(error, "category", None)
+        if isinstance(category, str):
+            payload["error_category"] = category
+    return NodeEvent(
+        node_name="openarmature.llm.complete",
+        namespace=("openarmature.llm.complete",),
+        step=-1,
+        phase=cast("Any", phase),
+        # ``pre_state`` is overloaded here as the LLM-event payload
+        # carrier — see the docstring above.
+        pre_state=cast("Any", {"llm_event": payload}),
+        post_state=None,
+        error=None,
+        parent_states=(),
+    )
+
+
 __all__ = [
     "OpenAIProvider",
 ]

src/openarmature/observability/__init__.py

@@ -0,0 +1,28 @@
+"""openarmature.observability — cross-backend observability surface.
+
+Two layers:
+
+- **Core** (this module + ``correlation.py``): always available, no
+  extra dependencies. Exposes :func:`current_correlation_id` and
+  :func:`current_active_observers` — the spec observability §3
+  ``ContextVar`` primitives that every backend mapping consumes.
+- **Backend mappings** (under ``observability.otel`` and future
+  ``observability.langfuse`` etc.): gated behind optional
+  dependencies (``pip install openarmature[otel]``). Importing the
+  subpackage without the extras installed raises an informative
+  ``ImportError`` pointing the caller at the install command.
+
+The split mirrors charter §3.1 principle 5: core defines the
+contracts; specific backends implement them. At v1.0 launch the
+backend mappings will lift into sibling packages
+(``openarmature-otel``, ``openarmature-langfuse``) — until then
+they live here under per-backend subpackages so the layering is
+established up front.
+"""
+
+from .correlation import current_active_observers, current_correlation_id
+
+__all__ = [
+    "current_active_observers",
+    "current_correlation_id",
+]