Migrate LlmUsageAccumulator to typed event filter

docs/examples/production-observability.md

@@ -85,10 +85,11 @@ sees the same logical events represented two ways.
 - **Queryable accumulator + `drain_events_for`**
   ([queryable observer pattern](../concepts/observability.md)).
   A third observer — `LlmUsageAccumulator` — subscribes to the
-  same event stream but only records the LLM-namespace events
-  carrying an `LlmEventPayload`. It accumulates per-invocation
-  token totals in memory, indexed by `current_invocation_id()`.
-  The terminal `persist` node calls
+  same event stream but only records the typed
+  `LlmCompletionEvent` variant (one event per successful LLM call;
+  outcome fields read directly off the event). It accumulates
+  per-invocation token totals in memory, indexed by
+  `event.invocation_id`. The terminal `persist` node calls
   `await graph.drain_events_for(current_invocation_id(), timeout=2.0)`
   to synchronize on the deliver loop, then reads the accumulator's
   bucket and drops it. Without the drain, the bucket might be
@@ -97,8 +98,32 @@ sees the same logical events represented two ways.
   a single-callable shape; the accumulator just exposes its own
   read methods (`get_bucket` / `drop`) that the persist node knows
   about. This is the canonical shape for per-invocation cost
-  attribution at request scope, replacing the round-trip-through-
-  State workarounds that pre-v0.12.0 deployments used.
+  attribution at request scope, in place of routing every token
+  count through State (a workaround that pollutes the state
+  schema with non-pipeline data).
+
+  The filter shape is `isinstance(event, LlmCompletionEvent)` —
+  one isinstance check against the typed event variant on the
+  observer event union. The provider also dual-emits a sentinel
+  `NodeEvent` pair during the transition period for backwards
+  compatibility with older accumulators; this example's
+  accumulator ignores the sentinel pair because the typed event
+  carries the same outcome data without the pair-join logic. New
+  accumulators should follow the isinstance-based filter shape
+  here; the CHANGELOG tracks when the sentinel emission is
+  removed.
+
+  `LlmCompletionEvent` is success-only by spec design. Failed LLM
+  calls flow through the exception path and do not emit the typed
+  event, so `bucket.call_count` reflects successful calls only.
+  This is the right semantic for a usage accumulator (failed
+  calls produce no tokens). A pipeline tracking attempt-level
+  failure rates needs a separate listener — either a custom
+  observer on the sentinel `NodeEvent` pair or a future
+  failure-event typed variant if and when that proposal lands.
+  Production code migrating an existing accumulator from the
+  sentinel pattern should expect this counting shift if it was
+  previously counting failure-path events.
 
 ## How to run
 
@@ -167,8 +192,15 @@ Trace id=<uuid>
 - **OTel spans block**: one line per captured span, sorted by
   start time. The relevant attributes shown are a curated subset
   for readability; the full attribute set is on each `Span` object
-  for any reader inspecting them programmatically. Note three
-  attribute families worth telling apart:
+  for any reader inspecting them programmatically. The
+  `openarmature.llm.complete` span name + the `gen_ai.usage.*`
+  attribute family come from the OTel observer's current
+  sentinel-`NodeEvent` handler — the OTel and Langfuse observers
+  have not yet migrated to consuming the typed `LlmCompletionEvent`
+  variant. Span names and attribute paths may shift when the
+  observer migration lands; the example's emitted span structure
+  tracks the current observer behavior. Note three attribute
+  families worth telling apart:
     - The root `openarmature.invocation` span carries
       `openarmature.graph.spec_version` plus the
       `openarmature.implementation.name` / `.version` attribution

examples/production-observability/main.py

@@ -49,9 +49,10 @@
   bucket and drops it. Without the drain, the bucket would be
   missing the most recent LLM event's tokens (the deliver loop
   hasn't reached them yet). This is the canonical shape for
-  per-invocation cost attribution at request scope, replacing the
-  round-trip-through-State workarounds that pre-v0.12.0 deployments
-  used. The pattern is convention-only at the observer level:
+  per-invocation cost attribution at request scope, in place of
+  routing every token count through State (a workaround pattern
+  that pollutes the state schema with non-pipeline data). The
+  pattern is convention-only at the observer level:
   ``Observer`` itself stays a single-callable protocol; the
   queryable accumulator just exposes its own read methods
   (``get_bucket`` / ``drop``) that the persist node knows about.
@@ -99,7 +100,7 @@
     CompiledGraph,
     GraphBuilder,
     InvocationCompletedEvent,
-    NodeEvent,
+    LlmCompletionEvent,
     NodeException,
     ObserverEvent,
     State,
@@ -112,7 +113,6 @@
     SystemMessage,
     UserMessage,
 )
-from openarmature.observability import LLM_NAMESPACE, LlmEventPayload
 from openarmature.observability.correlation import current_invocation_id
 from openarmature.observability.langfuse import (
     InMemoryLangfuseClient,
@@ -163,13 +163,31 @@ class BriefingState(State):
 # consume.  Convention only; openarmature does not ship a base class
 # for accumulators.
 #
-# The accumulator subscribes to every event but only records the LLM-
-# namespace ones (provider-emitted ``openarmature.llm.complete`` event
-# pair carrying an LlmEventPayload on ``pre_state``).  Per-invocation
-# isolation is by ``current_invocation_id()`` — read inside the
-# observer callback from the worker's Context, populated by the
-# engine at worker create time. Concurrent invocations on one
-# observer each get their own bucket.
+# The accumulator subscribes to every event but only records the
+# typed ``LlmCompletionEvent`` variant — one event per successful LLM
+# call, structured outcome fields read directly off the event without
+# the namespace-string-match + payload-narrow dance the legacy
+# sentinel pattern needed. The provider also dual-emits a sentinel
+# ``NodeEvent`` pair during the transition period for backwards
+# compatibility with older accumulators; this accumulator ignores
+# the sentinel pair because the typed event carries the same outcome
+# data without the pair-join logic. New accumulators should follow
+# the isinstance-based filter shape here; the CHANGELOG tracks when
+# the sentinel emission is removed.
+#
+# Per-invocation isolation is by ``LlmCompletionEvent.invocation_id``
+# — read directly off the event, no ContextVar lookup needed.
+# Concurrent invocations on one observer each get their own bucket.
+#
+# ``LlmCompletionEvent`` is success-only by spec design. Failed LLM
+# calls flow through the exception path and do NOT emit the typed
+# event, so ``bucket.call_count`` here reflects successful calls
+# only. This is the right semantic for a usage accumulator (failed
+# calls produce no tokens / cost). A pipeline tracking attempt-level
+# failure rates needs a separate listener — either a custom observer
+# on the sentinel ``NodeEvent`` pair, or a future
+# ``LlmCallFailedEvent`` typed variant if and when that proposal
+# lands.
 
 
 @dataclass
@@ -199,39 +217,40 @@ async def __call__(self, event: ObserverEvent) -> None:
         if isinstance(event, InvocationCompletedEvent):
             self._by_invocation.pop(event.invocation_id, None)
             return
-        if not isinstance(event, NodeEvent):
-            return
-        if event.namespace != LLM_NAMESPACE:
-            return
-        # Only the completed half of the pair carries the token counts.
-        if event.phase != "completed":
+        if not isinstance(event, LlmCompletionEvent):
             return
-        if not isinstance(event.pre_state, LlmEventPayload):
-            return
-        # NodeEvent doesn't carry invocation_id on the dataclass;
-        # observers read it from the ContextVar, which the
-        # deliver-loop worker's Context carries from the engine task
-        # at worker create-time (per-invocation worker, per-invocation
-        # Context).
-        invocation_id = current_invocation_id()
-        if invocation_id is None:
+        # call_count tracks successful LLM calls (the typed event is
+        # success-only by spec design). Spec contract has "call
+        # happened" and "usage reported" as INDEPENDENT — a provider
+        # may legitimately omit usage on a successful call. Create the
+        # bucket and increment call_count unconditionally so the
+        # counter reflects all successful calls; gate only the
+        # token-counting math on usage being populated.
+        bucket = self._by_invocation.setdefault(event.invocation_id, _UsageBucket())
+        bucket.call_count += 1
+        # The typed event's usage field is nullable per the spec
+        # contract ("may be null when the provider does not report
+        # usage"). Python's provider always passes a Usage instance
+        # (with all-None fields when not reported), but the defensive
+        # guard keeps the accumulator robust against future providers
+        # that exercise the null option. Calls without reported usage
+        # contribute zero tokens (the only honest value we can record).
+        usage = event.usage
+        if usage is None:
             return
-        payload = event.pre_state
-        bucket = self._by_invocation.setdefault(invocation_id, _UsageBucket())
-        if payload.prompt_tokens is not None:
-            bucket.prompt_tokens += payload.prompt_tokens
-        if payload.completion_tokens is not None:
-            bucket.completion_tokens += payload.completion_tokens
+        if usage.prompt_tokens is not None:
+            bucket.prompt_tokens += usage.prompt_tokens
+        if usage.completion_tokens is not None:
+            bucket.completion_tokens += usage.completion_tokens
         # Prefer the provider-reported total when present; otherwise
         # derive from prompt + completion when at least one is known.
-        # A payload with all three None (rare; provider didn't report
-        # usage at all) contributes zero, which is the only honest
-        # value we can record.
-        if payload.total_tokens is not None:
-            bucket.total_tokens += payload.total_tokens
-        elif payload.prompt_tokens is not None or payload.completion_tokens is not None:
-            bucket.total_tokens += (payload.prompt_tokens or 0) + (payload.completion_tokens or 0)
-        bucket.call_count += 1
+        # A usage record with all three None (rare; provider didn't
+        # report counts at all) contributes zero, which is the only
+        # honest value we can record.
+        if usage.total_tokens is not None:
+            bucket.total_tokens += usage.total_tokens
+        elif usage.prompt_tokens is not None or usage.completion_tokens is not None:
+            bucket.total_tokens += (usage.prompt_tokens or 0) + (usage.completion_tokens or 0)
 
     # Consumers MUST synchronize on ``drain_events_for`` before
     # calling ``get_bucket`` if completeness matters — without the