observability: PR #19 round-5 review followups

Two Copilot findings addressed:

- **Log-bridge filter on the root logger** — ``install_log_bridge``
  was attaching ``_CorrelationIdFilter`` to the OTel
  ``LoggingHandler`` only. Pre-existing stdout / file / third-party
  handlers registered ahead of it on the root logger would see log
  records WITHOUT ``openarmature.correlation_id``, violating spec
  §7's "every log record emitted during an invocation MUST carry
  the attribute." Moved the filter to ``root.addFilter(...)`` so
  it runs at the LOGGER level, applying to every record regardless
  of which handler processes it. Idempotency is now tracked
  separately for handler vs filter installation.
- **Typed LLM event payload** — ``_make_llm_event`` previously
  stuffed a plain dict into ``NodeEvent.pre_state`` with
  ``cast("Any", ...)``, violating the documented
  ``pre_state: State`` contract. Any observer calling
  ``event.pre_state.model_dump()`` on an LLM event would crash.
  Wrapped the payload in a dedicated ``_LlmEventState(State)``
  Pydantic subclass with the LLM fields (model, finish_reason,
  prompt/completion/total tokens, optional error_type / message /
  category). The OTel observer reads attributes directly via
  ``isinstance(event.pre_state, _LlmEventState)`` and attribute
  access. Schema is honored.

Author: chris-colinsky

src/openarmature/llm/providers/openai.py

@@ -45,6 +45,7 @@
 from pydantic import ValidationError
 
 from openarmature.graph.events import NodeEvent
+from openarmature.graph.state import State
 from openarmature.observability.correlation import current_dispatch
 
 from ..errors import (
@@ -524,6 +525,36 @@ def _looks_like_model_not_loaded(message: object) -> bool:
 # ---------------------------------------------------------------------------
 
 
+class _LlmEventState(State):
+    """Typed payload for LLM-provider span events. Subclasses
+    :class:`openarmature.graph.state.State` so the
+    ``NodeEvent.pre_state: State`` contract holds — observers
+    calling ``event.pre_state.model_dump()`` (or any other
+    Pydantic-on-State method) work without the raw-dict overload
+    that previously violated the schema.
+
+    Backend mappings (the OTel observer in this repo, future
+    Langfuse / Datadog adapters) recognize the
+    ``("openarmature.llm.complete",)`` namespace sentinel and read
+    these fields directly via attribute access.
+    """
+
+    model: str
+    finish_reason: str | None = None
+    prompt_tokens: int | None = None
+    completion_tokens: int | None = None
+    total_tokens: int | None = None
+    # On error responses the provider caller doesn't have a
+    # graph-engine §4 ``RuntimeGraphError`` to put in
+    # ``NodeEvent.error``, so we surface the failure detail through
+    # these fields instead. ``error_category`` is the canonical §7
+    # llm-provider category (``provider_unavailable``, etc.) when
+    # the failed exception carries one.
+    error_type: str | None = None
+    error_message: str | None = None
+    error_category: str | None = None
+
+
 def _make_llm_event(
     phase: Literal["started", "completed"],
     *,
@@ -537,40 +568,33 @@ def _make_llm_event(
     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",)``.
+    ``pre_state`` directly via attribute access.
     """
-    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
+    error_type: str | None = None
+    error_message: str | None = None
+    error_category: str | None = None
     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)
+        error_type = type(error).__name__
+        error_message = str(error)
         category = getattr(error, "category", None)
         if isinstance(category, str):
-            payload["error_category"] = category
+            error_category = category
+    payload = _LlmEventState(
+        model=model,
+        finish_reason=finish_reason,
+        prompt_tokens=usage.prompt_tokens if usage is not None else None,
+        completion_tokens=usage.completion_tokens if usage is not None else None,
+        total_tokens=usage.total_tokens if usage is not None else None,
+        error_type=error_type,
+        error_message=error_message,
+        error_category=error_category,
+    )
     return NodeEvent(
         node_name="openarmature.llm.complete",
         namespace=("openarmature.llm.complete",),
         step=-1,
         phase=phase,
-        # ``pre_state`` is overloaded here as the LLM-event payload
-        # carrier — see the docstring above.
-        pre_state=cast("Any", {"llm_event": payload}),
+        pre_state=payload,
         post_state=None,
         error=None,
         parent_states=(),

src/openarmature/observability/otel/logs.py

@@ -47,13 +47,25 @@ def install_log_bridge(
 ) -> None:
     """Wire the stdlib root logger to the supplied OTel
     :class:`LoggerProvider`. Adds a
-    :class:`opentelemetry.sdk._logs.LoggingHandler` and an
-    :class:`_CorrelationIdFilter` so every log record emitted from
-    anywhere within an invocation carries the active
-    ``trace_id``/``span_id`` + ``openarmature.correlation_id``.
-
-    Idempotent: re-calling with a previously-installed provider is
-    a no-op (we check for an existing OA handler before adding).
+    :class:`opentelemetry.sdk._logs.LoggingHandler` for OTel-native
+    ``trace_id`` / ``span_id`` bridging, AND attaches an
+    :class:`_CorrelationIdFilter` directly to the ROOT LOGGER (not
+    the handler) so the ``openarmature.correlation_id`` attribute
+    lands on every log record emitted during an invocation —
+    including records consumed by pre-existing stdout / file /
+    third-party handlers the user already had configured.
+
+    Filter-on-the-root-logger placement matters per spec §7:
+    "log records emitted from anywhere within an invocation MUST
+    carry ``openarmature.correlation_id``." A handler-level filter
+    would only modify records flowing through THAT handler, so a
+    user's existing stdout handler would see records without the
+    attribute. The root-logger filter applies to every record,
+    regardless of which handler eventually processes it.
+
+    Idempotent: re-calling is a no-op (we check for the existing
+    OA-tagged handler AND for an existing filter instance on the
+    root logger).
 
     The user retains responsibility for providing the
     :class:`LoggerProvider` (typically built with their preferred
@@ -63,17 +75,23 @@ def install_log_bridge(
     from opentelemetry.sdk._logs import LoggingHandler  # type: ignore[attr-defined]
 
     root = logging.getLogger()
-    # Idempotency check — don't double-install on repeated calls.
-    for handler in root.handlers:
-        if isinstance(handler, LoggingHandler) and getattr(handler, "_openarmature_installed", False):
-            return
-    handler = LoggingHandler(level=level, logger_provider=provider)
-    # Direct assignment isn't typed on LoggingHandler; route through
-    # ``setattr`` to avoid pyright's strict attribute-access check
-    # without losing the idempotency-marker behavior.
-    object.__setattr__(handler, "_openarmature_installed", True)
-    handler.addFilter(_CorrelationIdFilter())
-    root.addHandler(handler)
+    # Idempotency #1: don't double-add the OTel LoggingHandler.
+    handler_already_installed = any(
+        isinstance(h, LoggingHandler) and getattr(h, "_openarmature_installed", False) for h in root.handlers
+    )
+    if not handler_already_installed:
+        handler = LoggingHandler(level=level, logger_provider=provider)
+        # Direct assignment isn't typed on LoggingHandler; route
+        # through ``object.__setattr__`` to avoid pyright's strict
+        # attribute-access check without losing the idempotency-
+        # marker behavior.
+        object.__setattr__(handler, "_openarmature_installed", True)
+        root.addHandler(handler)
+    # Idempotency #2: don't double-add the correlation_id filter to
+    # the root logger.
+    filter_already_installed = any(isinstance(f, _CorrelationIdFilter) for f in root.filters)
+    if not filter_already_installed:
+        root.addFilter(_CorrelationIdFilter())
 
 
 __all__ = [

src/openarmature/observability/otel/observer.py

@@ -341,17 +341,24 @@ def _emit_checkpoint_save_span(self, event: NodeEvent) -> None:
     def _handle_llm_event(self, event: NodeEvent) -> None:
         """LLM provider span per spec §5.5 — parented to the node
         span that invoked the provider."""
-        # The LLM provider's pre_state carries the event payload
-        # (model, finish_reason, usage, error detail) since
-        # NodeEvent's shape is fixed. See
-        # ``openarmature.llm.providers.openai._make_llm_event``.
-        payload = cast(
-            "dict[str, Any]",
-            cast("dict[str, Any]", event.pre_state).get("llm_event", {}),
-        )
+        # ``pre_state`` is a typed ``_LlmEventState`` Pydantic
+        # subclass — see
+        # ``openarmature.llm.providers.openai._LlmEventState``. We
+        # read attributes directly rather than treating the State
+        # as a dict, preserving the ``NodeEvent.pre_state: State``
+        # contract for any other observers that consume the event.
+        # Lazy import to avoid the otel→llm package dependency at
+        # module load time.
+        from openarmature.llm.providers.openai import _LlmEventState
+
+        if not isinstance(event.pre_state, _LlmEventState):
+            # Defensive — callers other than the OpenAIProvider hook
+            # shouldn't dispatch through the LLM_NAMESPACE sentinel.
+            return
+        payload = event.pre_state
         if event.phase == "started":
             parent_ctx = self._current_span_context()
-            attrs: dict[str, Any] = {"openarmature.llm.model": payload["model"]}
+            attrs: dict[str, Any] = {"openarmature.llm.model": payload.model}
             from openarmature.observability.correlation import current_correlation_id
 
             cid = current_correlation_id()
@@ -370,27 +377,23 @@ def _handle_llm_event(self, event: NodeEvent) -> None:
             if open_span is None:
                 return
             span = open_span.span
-            if "finish_reason" in payload:
-                span.set_attribute("openarmature.llm.finish_reason", payload["finish_reason"])
-            for usage_field in (
-                "prompt_tokens",
-                "completion_tokens",
-                "total_tokens",
-            ):
-                if payload.get(usage_field) is not None:
-                    span.set_attribute(
-                        f"openarmature.llm.usage.{usage_field}",
-                        payload[usage_field],
-                    )
-            if "error_type" in payload:
+            if payload.finish_reason is not None:
+                span.set_attribute("openarmature.llm.finish_reason", payload.finish_reason)
+            if payload.prompt_tokens is not None:
+                span.set_attribute("openarmature.llm.usage.prompt_tokens", payload.prompt_tokens)
+            if payload.completion_tokens is not None:
+                span.set_attribute("openarmature.llm.usage.completion_tokens", payload.completion_tokens)
+            if payload.total_tokens is not None:
+                span.set_attribute("openarmature.llm.usage.total_tokens", payload.total_tokens)
+            if payload.error_type is not None:
                 span.set_status(
                     Status(
                         StatusCode.ERROR,
-                        description=payload.get("error_category", payload["error_type"]),
+                        description=payload.error_category or payload.error_type,
                     )
                 )
-                if "error_category" in payload:
-                    span.set_attribute("openarmature.error.category", payload["error_category"])
+                if payload.error_category is not None:
+                    span.set_attribute("openarmature.error.category", payload.error_category)
             else:
                 span.set_status(Status(StatusCode.OK))
             span.end()

tests/unit/test_observability_otel.py

@@ -262,6 +262,8 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
     # LLM event through the observer's __call__ and assert no span was
     # produced. This isolates the disable_llm_spans branch from the
     # provider's own queue-dispatch wiring.
+    from openarmature.llm.providers.openai import _LlmEventState
+
     exporter = InMemorySpanExporter()
     observer = OTelObserver(
         span_processor=SimpleSpanProcessor(exporter),
@@ -272,7 +274,7 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
         namespace=("openarmature.llm.complete",),
         step=-1,
         phase="started",
-        pre_state={"llm_event": {"model": "test-m"}},  # type: ignore[arg-type]
+        pre_state=_LlmEventState(model="test-m"),
         post_state=None,
         error=None,
         parent_states=(),
@@ -282,12 +284,7 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
         namespace=("openarmature.llm.complete",),
         step=-1,
         phase="completed",
-        pre_state={  # type: ignore[arg-type]
-            "llm_event": {
-                "model": "test-m",
-                "finish_reason": "stop",
-            }
-        },
+        pre_state=_LlmEventState(model="test-m", finish_reason="stop"),
         post_state=None,
         error=None,
         parent_states=(),