From 508ac05e56a4ed169631cfc4cf39fadec7562c42 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 5 Jun 2026 18:06:34 -0700 Subject: [PATCH 1/4] Extend production-observability with accumulator pattern Two pre-release polish items for the v0.12.0 cycle. CHANGELOG: the Unreleased Changed entry for the spec-pin advance originally said proposal 0052 "lands in a follow-on PR of this cycle" and 0054 "lands in a follow-on PR". Both have since landed (PRs 131 + 132). Rewrites the bullet to factually describe the final state: three proposals (0048, 0052, 0054) ship as fully implemented, two (0051, 0053) ship as textual-only. production-observability example: adds an LlmUsageAccumulator class plus a terminal persist node that demonstrates the queryable observer + drain_events_for pattern end-to-end. The accumulator subscribes to the LLM-namespace event stream, accumulates per- invocation token totals via current_invocation_id() bucket keys, and exposes convention-only get_bucket / drop methods. The persist node calls drain_events_for to synchronize on the deliver loop before reading the bucket so the rollup reflects every LLM call in the invocation, drops the bucket per the explicit-cleanup discipline, and prints a cost summary. The graph grows from respond -> END to respond -> persist -> END. Module-level singletons (_accumulator + _compiled_graph) keep the persist node closure-free and follow the existing _provider_instance precedent. Walkthrough doc updates the H1, overview, what-it-teaches list, captured-output sample, and reading-the-output walkthrough to cover the new pattern. --- CHANGELOG.md | 2 +- docs/examples/production-observability.md | 42 ++++- examples/production-observability/main.py | 206 ++++++++++++++++++++-- 3 files changed, 231 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a2d56844..d850fc78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ### Changed -- **Pinned spec advanced from v0.38.0 to v0.46.0.** Submodule + `[tool.openarmature].spec_version` + `conformance.toml` `spec_pin` advance together. Absorbs eight new proposals (0047-0054) into the conformance manifest. Two of them ship as part of the v0.12.0 cycle as textual-only acknowledgments with no code change required: **proposal 0051** (observability §8.4.1 Langfuse `trace.input` / `trace.output` implementation-surface caveat — documents that vendor SDK round-trip is required to project caller-side trace I/O updates onto the wire; the v0.11.0 (proposal 0043) caller-hook shape already matches the documented behavior) and **proposal 0053** (observability §3.4 shared-parent boundary clarification — tightens the structural-shared-parent classification to predicate the invocation span on whether at least one fan-out or parallel-branches dispatch is on the augmenter's call-stack path; behavior already matches via fixtures 034 + 039). Two more ship as implemented in this cycle: **proposal 0048** (read-symmetric metadata + queryable observer pattern docs — see *Added* below) and **proposal 0052** (implementation attribution attributes — landing in a follow-on PR of this cycle). The remaining proposals are marked `not-yet` in the conformance manifest with roadmap targets: 0047 + 0049 (v0.13.0 LLM provider hardening batch), 0050 (v0.14.0 retry & reliability batch), 0054 (per-invocation observer event drain — bundled into this v0.12.0 cycle alongside 0048 per the §9.4 accumulator-lifecycle pairing; lands in a follow-on PR). +- **Pinned spec advanced from v0.38.0 to v0.46.0.** Submodule + `[tool.openarmature].spec_version` + `conformance.toml` `spec_pin` advance together. Absorbs eight new proposals (0047-0054) into the conformance manifest. Two ship as textual-only acknowledgments with no code change required: **proposal 0051** (observability §8.4.1 Langfuse `trace.input` / `trace.output` implementation-surface caveat — documents that vendor SDK round-trip is required to project caller-side trace I/O updates onto the wire; the v0.11.0 (proposal 0043) caller-hook shape already matches the documented behavior) and **proposal 0053** (observability §3.4 shared-parent boundary clarification — tightens the structural-shared-parent classification to predicate the invocation span on whether at least one fan-out or parallel-branches dispatch is on the augmenter's call-stack path; behavior already matches via fixtures 034 + 039). Three ship as fully implemented this cycle: **proposal 0048** (read-symmetric metadata + queryable observer pattern docs — see *Added* below), **proposal 0052** (implementation attribution attributes — see *Added* below), and **proposal 0054** (per-invocation observer event drain — see *Added* below; bundled with 0048 as the §9.4 accumulator-lifecycle pair). The remaining proposals are marked `not-yet` in the conformance manifest with roadmap targets: 0047 + 0049 (v0.13.0 LLM provider hardening batch) and 0050 (v0.14.0 retry & reliability batch). - **README and docs homepage refreshed around reasons-to-choose.** Replaced the 10-bullet "Why OpenArmature" feature inventory in `README.md` with 5 differentiating reasons (LLM-infused workflows to agents on one engine; crash-safe resume by contract; destination-pluggable observability with OTel + Langfuse, no SaaS lock-in; compile-time topology checks; spec + conformance). The docs homepage (`docs/index.md`) card grid carries the same five plus a sixth card retained from the previous grid for async-first / LLM-agnostic: workflows-to-agents, crash-safe, pluggable observability, bad-graphs-don't-compile, parallelism (fan-out + parallel-branches + nested correctness), async-first. - **Docs sweep: stale references and em-dash normalization.** Fixed three definite stale references (`spec_version='0.26.0'` in the Langfuse example output now reads `'0.38.0'`; the dangling `v0.16.1` qualifier dropped from the parallel-branches concept page; `compiled.attach_observer` corrected to `graph.attach_observer` in `non-obvious-shapes.md` for variable-name consistency with the rest of the docs). Swept em dashes out of the user-facing docs (130 instances across 17 files) per the convention set during the patterns expansion. mkdocs strict build clean; no broken intra-docs links. - **The checkpointing-and-migration example grows a crash-and-resume drama.** The first invoke of the v1 graph now hits a simulated transient failure inside `size_crew` (raises a `RuntimeError` on its first attempt only). The example catches `NodeException` at the `invoke()` boundary, prints what's saved on disk (`define_objective`'s position is already in `completed_positions`), then re-invokes with `resume_invocation=`. The retried `size_crew` succeeds, `draft_timeline` runs, and the pipeline finishes - dramatizing the synchronous-checkpoint-by-contract reliability claim from the README pitch. The existing v1->v2 migration phase rides on top of the crash-survived checkpoint, so both reliability stories compose in one demo. Walk-through doc rewritten to cover both phases. diff --git a/docs/examples/production-observability.md b/docs/examples/production-observability.md index 2dedec6d..3548abb2 100644 --- a/docs/examples/production-observability.md +++ b/docs/examples/production-observability.md @@ -1,4 +1,4 @@ -# Production observability with dual observers and timing middleware +# Production observability with dual observers, timing middleware, and per-invocation cost rollup !!! info "Source" [https://github.com/LunarCommand/openarmature-python/blob/main/examples/production-observability/main.py](https://github.com/LunarCommand/openarmature-python/blob/main/examples/production-observability/main.py){target="_blank" rel="noopener"} @@ -7,14 +7,19 @@ A single-turn lunar-mission Q&A endpoint instrumented the way you'd ship it: BOTH OTel and Langfuse observers attached to the same graph, caller hooks deriving domain-shaped `trace.input` / `trace.output` from State, the built-in `TimingMiddleware` -recording per-node duration, and multi-tenant caller-supplied -metadata propagating to both observers in one `invoke()` call. +recording per-node duration, multi-tenant caller-supplied +metadata propagating to both observers in one `invoke()` call, AND +a third queryable-accumulator observer that a terminal `persist` +node reads at request scope after synchronizing on the deliver +loop with `drain_events_for`. ## Overview -One node, one LLM call, two production-grade observability -backends. The pipeline takes a question, calls the LLM, returns the -answer. The interesting part is the observability wiring: +Two nodes (`respond` then `persist`), one LLM call, three observers +attached at compile time. The pipeline takes a question, calls the +LLM, returns the answer, then synchronizes on the observer queue +and rolls up token cost. The interesting part is the observability +wiring: - `OTelObserver` attached with an `InMemorySpanExporter` (production swaps this for `BatchSpanProcessor` + @@ -77,6 +82,23 @@ sees the same logical events represented two ways. `InMemorySpanExporter` records every Span. Production deployments swap each for a real exporter / SDK adapter; the observer call surface doesn't change. +- **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 + `await graph.drain_events_for(state.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 + missing the most-recent LLM event's tokens (the deliver loop + hasn't reached them yet). The `Observer` protocol itself stays + 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. ## How to run @@ -105,6 +127,7 @@ request id: feature flag:v2-canary [timing] respond: 1234.5ms (success) +[persist] LLM usage: prompt=42, completion=38, total=80 across 1 call(s) answer: The primary objective of Apollo 11 was ... model: gpt-4o-mini-2024-07-18 @@ -133,6 +156,13 @@ Trace id= `TimingMiddleware` callback as soon as the respond chain returns. `outcome` is `"success"` here; a `ProviderRateLimit` would surface as `outcome="exception"` with `exception_category="provider_rate_limit"`. +- **`[persist] LLM usage: ...`**: emitted by the `persist` node + after it drains the deliver loop and reads the + `LlmUsageAccumulator`'s bucket for this invocation. If the drain + times out (slow / hung observer), the persist line is prefixed by + a `[persist] drain incomplete: N events still pending after 2.0s` + surface — the production version of that log would also flip an + SLO-breach metric. - **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 diff --git a/examples/production-observability/main.py b/examples/production-observability/main.py index 6be228d2..2104905f 100644 --- a/examples/production-observability/main.py +++ b/examples/production-observability/main.py @@ -41,12 +41,26 @@ ``LangfuseSDKAdapter(Langfuse(...))`` and ``BatchSpanProcessor(OTLPSpanExporter(...))`` respectively; the observer call surface doesn't change. +- **Queryable accumulator observer + per-invocation drain.** A + third observer (``LlmUsageAccumulator``) rolls up LLM token + totals per invocation. A terminal ``persist`` node calls + ``graph.drain_events_for(state.invocation_id)`` to synchronize on + the deliver loop, then reads the accumulator's 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: ``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. Complementary to the observer-hooks example (three observers side-by-side) and the langfuse-observability example (Langfuse observer + LangfusePromptBackend prompt linkage). This example's -headline is the production-shape wiring, not the hook surface or -the prompt linkage. +headline is the production-shape wiring + per-invocation cost +attribution, not the hook surface or the prompt linkage. **Configuration** (env vars; OpenAI defaults shown): @@ -70,6 +84,7 @@ import os import sys import uuid +from dataclasses import dataclass from typing import Any from opentelemetry.sdk.resources import Resource @@ -79,7 +94,15 @@ InMemorySpanExporter, ) -from openarmature.graph import END, GraphBuilder, NodeException, State +from openarmature.graph import ( + END, + CompiledGraph, + GraphBuilder, + NodeEvent, + NodeException, + ObserverEvent, + State, +) from openarmature.graph.middleware import TimingMiddleware, TimingRecord from openarmature.llm import ( LlmProviderError, @@ -88,6 +111,8 @@ SystemMessage, UserMessage, ) +from openarmature.observability import LLM_NAMESPACE, LlmEventPayload +from openarmature.observability.correlation import current_invocation_id from openarmature.observability.langfuse import ( InMemoryLangfuseClient, LangfuseObservation, @@ -125,6 +150,101 @@ class BriefingState(State): model_used: str = "" +# --------------------------------------------------------------------------- +# Queryable accumulator observer (per-invocation LLM token rollup) +# --------------------------------------------------------------------------- +# A third observer alongside the OTel + Langfuse pair. Its job is to +# accumulate per-invocation LLM token usage in memory so a terminal +# persist node can read the totals at request scope (rather than +# round-tripping every count through State). The Observer protocol is +# a single async callable; the accumulator adds its own read methods +# (``get_bucket`` / ``drop``) on the instance for the persist node to +# 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. + + +@dataclass +class _UsageBucket: + prompt_tokens: int = 0 + completion_tokens: int = 0 + total_tokens: int = 0 + call_count: int = 0 + + +class LlmUsageAccumulator: + """Per-invocation LLM token rollup.""" + + def __init__(self) -> None: + # Concurrent invocations on one observer each land in their + # own bucket. Production deployments with high concurrency + # would want an eviction policy on top to bound memory; the + # demo's persist node drops the bucket explicitly after read. + self._by_invocation: dict[str, _UsageBucket] = {} + + async def __call__(self, event: ObserverEvent) -> None: + 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": + 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: + 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 payload.total_tokens is not None: + bucket.total_tokens += payload.total_tokens + bucket.call_count += 1 + + # Consumers MUST synchronize on ``drain_events_for`` before + # calling ``get_bucket`` if completeness matters — without the + # drain the deliver loop may still hold pending events whose + # tokens have not been added yet. ``None`` is returned when + # nothing has been recorded yet (e.g., an invocation with no + # LLM calls). + def get_bucket(self, invocation_id: str) -> _UsageBucket | None: + """Read the accumulated bucket for an invocation.""" + return self._by_invocation.get(invocation_id) + + # The accumulator does NOT auto-drop on + # ``InvocationCompletedEvent`` — a terminal node legitimately + # needs to read the bucket BEFORE the invocation completes, and + # auto-drop would race the read. Callers invoke ``drop()`` + # explicitly after reading. + def drop(self, invocation_id: str) -> None: + """Release the bucket for an invocation.""" + self._by_invocation.pop(invocation_id, None) + + +# Module-level singletons make the persist node closure-free and +# match how ``_provider_instance`` is handled. In an application +# server, these would live on a request-scoped or app-scoped +# container instead. +_accumulator: LlmUsageAccumulator | None = None +_compiled_graph: CompiledGraph[BriefingState] | None = None + + # --------------------------------------------------------------------------- # Caller hooks for Langfuse trace.input / trace.output # --------------------------------------------------------------------------- @@ -195,23 +315,78 @@ async def respond(state: BriefingState) -> dict[str, Any]: } +# Terminal node. State is intentionally unused — this node's job is +# to synchronize on the observer deliver loop and report a derived +# rollup, not to read or modify pipeline state. +# +# ``drain_events_for`` blocks until every event dispatched up to this +# point has reached every attached observer. Without it the +# accumulator's bucket may still be missing the most-recent LLM +# event's tokens — the deliver loop hasn't processed them yet when +# the node body runs. Snapshot semantic: the drain awaits only +# events dispatched BEFORE the call (this node's own ``started`` +# event included), not events that fire after the call begins +# (notably this node's own ``completed`` event, which only fires +# after the body returns — that's how the call avoids deadlocking +# on itself). +# +# Default timeout is 5.0s; the demo tightens to 2.0s so a stuck +# observer surfaces fast. Production teams pick the threshold against +# their observer SLO. Returning a timeout summary instead of raising +# lets the caller record an SLO breach and proceed with whatever +# data is available, rather than failing the whole invocation. +async def persist(_state: BriefingState) -> dict[str, Any]: + """Drain the deliver loop, read the LLM-usage rollup, drop the bucket.""" + assert _compiled_graph is not None + assert _accumulator is not None + invocation_id = current_invocation_id() + assert invocation_id is not None + summary = await _compiled_graph.drain_events_for(invocation_id, timeout=2.0) + if summary.timeout_reached: + # Production: emit an SLO-breach metric. Demo: surface the + # gap inline so a reader sees what an incomplete drain looks + # like. + print(f"[persist] drain incomplete: {summary.undelivered_count} events still pending after 2.0s") + bucket = _accumulator.get_bucket(invocation_id) + _accumulator.drop(invocation_id) + if bucket is None: + print("[persist] no LLM usage recorded for this invocation") + return {} + # In production, this is where you'd write the canonical + # invocation artifact to durable storage: a JSON record with the + # answer + per-invocation token cost + caller metadata + trace + # IDs for cross-system join. The demo prints the rollup so the + # pattern is legible. + print( + f"[persist] LLM usage: prompt={bucket.prompt_tokens}, " + f"completion={bucket.completion_tokens}, total={bucket.total_tokens} " + f"across {bucket.call_count} call(s)" + ) + return {} + + # --------------------------------------------------------------------------- # Graph # --------------------------------------------------------------------------- -def build_graph(): - """Single-node graph: respond -> END. +def build_graph() -> CompiledGraph[BriefingState]: + """Two-node graph: respond -> persist -> END. TimingMiddleware wraps the respond node so wall-clock duration - is captured per call. No other middleware (RetryMiddleware lives - in the fan-out-with-retry / parallel-branches examples; this - one's scope is observability).""" + is captured per call. The persist node runs synchronously after + respond returns; it drains the deliver loop for the current + invocation, reads the LLM-usage accumulator's bucket, drops the + bucket, and prints a cost summary. No other middleware + (RetryMiddleware lives in the fan-out-with-retry / parallel- + branches examples; this one's scope is observability).""" timing = TimingMiddleware(node_name="respond", on_complete=_emit_timing) return ( GraphBuilder(BriefingState) .add_node("respond", respond, middleware=[timing]) - .add_edge("respond", END) + .add_node("persist", persist) + .add_edge("respond", "persist") + .add_edge("persist", END) .set_entry("respond") .compile() ) @@ -349,9 +524,16 @@ async def main() -> None: span_exporter = InMemorySpanExporter() langfuse_client = InMemoryLangfuseClient() - graph = build_graph() - graph.attach_observer(_build_otel_observer(span_exporter)) - graph.attach_observer(_build_langfuse_observer(langfuse_client)) + # Module-level singletons populated before invoke so the persist + # node (which is plain-async, no closure) can reach the graph for + # drain_events_for and the accumulator for the read + drop. + global _accumulator, _compiled_graph + _accumulator = LlmUsageAccumulator() + _compiled_graph = build_graph() + _compiled_graph.attach_observer(_build_otel_observer(span_exporter)) + _compiled_graph.attach_observer(_build_langfuse_observer(langfuse_client)) + _compiled_graph.attach_observer(_accumulator) + graph = _compiled_graph # Caller-supplied multi-tenant metadata. Both observers pick # the entries up: OTel attaches them as ``openarmature.user.*`` From f01273074d61c1cd8185b73be81e28ea25ea2e34 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 5 Jun 2026 18:17:15 -0700 Subject: [PATCH 2/4] Surface invocation-span attribution attrs in OTel formatter The example's _format_otel_spans excluded the root openarmature.invocation span from its captured-output listing because two issues kept it from landing in the in-memory exporter and from showing usefully even when it did: 1. The OTel observer's shutdown() was never called, so the root invocation span stayed open and never moved into the exporter's finished-spans list. Adds otel_observer.shutdown() to the finally block after drain(), mirroring the pattern in the OTel unit tests. 2. The formatter's curated key set didn't include the invocation-level attributes the new span carries (openarmature.graph.entry_node, .spec_version, openarmature.implementation.name + .version). The formatter now picks the right key set based on span name: the invocation span surfaces its four invocation-level attrs only, inner-node spans surface the per-node + cross-cutting user.* + GenAI semconv attrs. Skipping cross-cutting attrs on the invocation line avoids repeating data that appears three more times below. Net visible change: the captured-OTel-spans block now opens with a [openarmature.invocation] line carrying implementation_name='openarmature-python' + implementation_version + spec_version + entry_node. Operators filtering traces by library version in Phoenix / Datadog / Honeycomb / Tempo / HyperDX read these directly from the root invocation span. Walkthrough doc's reading-the-output bullet now distinguishes the three OTel attribute families (invocation-level 5.1, cross-cutting 5.6, GenAI semconv) and explains why the invocation span only closes on observer shutdown(). --- docs/examples/production-observability.md | 25 +++++++-- examples/production-observability/main.py | 67 ++++++++++++++++++----- 2 files changed, 72 insertions(+), 20 deletions(-) diff --git a/docs/examples/production-observability.md b/docs/examples/production-observability.md index 3548abb2..d9f91bdb 100644 --- a/docs/examples/production-observability.md +++ b/docs/examples/production-observability.md @@ -132,9 +132,10 @@ answer: The primary objective of Apollo 11 was ... model: gpt-4o-mini-2024-07-18 --- captured OTel spans --- - [openarmature.invocation] 1240.0ms openarmature.user.tenantId='demo-acme', ... + [openarmature.invocation] 1240.0ms openarmature.graph.entry_node='respond', openarmature.graph.spec_version='0.46.0', openarmature.implementation.name='openarmature-python', openarmature.implementation.version='0.12.0' [respond] 1235.0ms openarmature.node.name='respond', openarmature.user.tenantId='demo-acme', ... - [openarmature.llm.complete] 1200.0ms gen_ai.system='openai', gen_ai.usage.input_tokens=42, ... + [openarmature.llm.complete] 1200.0ms openarmature.user.tenantId='demo-acme', gen_ai.system='openai', gen_ai.usage.input_tokens=42, ... + [persist] 2.0ms openarmature.node.name='persist', openarmature.user.tenantId='demo-acme', ... --- captured Langfuse trace --- Trace id= @@ -166,9 +167,23 @@ Trace id= - **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 the - `openarmature.user.*` attributes appearing on every span (the - cross-cutting attribute propagation from `invoke(metadata=...)`). + for any reader inspecting them programmatically. Note three + attribute families worth telling apart: + - The root `openarmature.invocation` span carries + `openarmature.graph.spec_version` plus the + `openarmature.implementation.name` / `.version` attribution + attributes. These are invocation-span-only (per spec §5.1) — + operators filtering by library version use these. + - The `openarmature.user.*` attributes appear on every span, + reflecting the cross-cutting propagation from + `invoke(metadata=...)`. + - `gen_ai.usage.*` lands on the LLM span only, sourced from the + provider's wire response. + + The invocation span only lands in the exporter after the OTel + observer's `shutdown()` is called (closing the root span). The + demo calls it after `drain()` in the `finally` block; production + long-running processes call it at process exit. - **Langfuse trace block**: the same invocation as seen by the Langfuse data model. `trace.input` / `trace.output` come from the caller hooks (`{"question": ...}` / `{"answer": ..., "model": ...}`) diff --git a/examples/production-observability/main.py b/examples/production-observability/main.py index 2104905f..ec92657e 100644 --- a/examples/production-observability/main.py +++ b/examples/production-observability/main.py @@ -444,8 +444,43 @@ def _build_langfuse_observer(client: InMemoryLangfuseClient) -> LangfuseObserver # have ingested. +# Invocation-span-only attributes (spec 5.1). Surface these only on +# the root ``openarmature.invocation`` span line; inner spans don't +# carry them (they're invocation-level constants, not cross-cutting +# 5.6 attributes). +_INVOCATION_SPAN_KEYS = ( + "openarmature.graph.entry_node", + "openarmature.graph.spec_version", + "openarmature.implementation.name", + "openarmature.implementation.version", +) + +# Per-node + cross-cutting attributes (5.6 + GenAI semconv). Surface +# these on inner-node spans only; they propagate to the invocation +# span too but showing them there is redundant once they appear on +# every node line below. +_INNER_SPAN_KEYS = ( + "openarmature.node.name", + "openarmature.user.tenantId", + "openarmature.user.requestId", + "openarmature.user.featureFlag", + "gen_ai.system", + "gen_ai.usage.input_tokens", + "gen_ai.usage.output_tokens", +) + + def _format_otel_spans(spans: list[ReadableSpan]) -> str: - """One line per span: name, duration, key attributes.""" + """One line per span: name, duration, key attributes. + + The ``openarmature.invocation`` root span closes on observer + ``shutdown()`` and surfaces only its invocation-level + attributes (spec 5.1 — entry_node, spec_version, implementation + name + version). Inner-node spans surface the cross-cutting + caller metadata + GenAI semconv attributes; printing them on + the invocation line too would just repeat data shown three + more times below. + """ if not spans: return " (no spans captured)" lines: list[str] = [] @@ -453,18 +488,8 @@ def _format_otel_spans(spans: list[ReadableSpan]) -> str: spans_sorted = sorted(spans, key=lambda s: s.start_time or 0) for span in spans_sorted: attrs = span.attributes or {} - # Pull a few interesting attributes for the summary; the - # full set is in span.attributes for any reader who wants it. - keys_of_interest = ( - "openarmature.node.name", - "openarmature.user.tenantId", - "openarmature.user.requestId", - "openarmature.user.featureFlag", - "gen_ai.system", - "gen_ai.usage.input_tokens", - "gen_ai.usage.output_tokens", - ) - relevant = {k: v for k in keys_of_interest if (v := attrs.get(k)) is not None} + keys = _INVOCATION_SPAN_KEYS if span.name == "openarmature.invocation" else _INNER_SPAN_KEYS + relevant = {k: v for k in keys if (v := attrs.get(k)) is not None} duration_ms = 0.0 if span.start_time is not None and span.end_time is not None: duration_ms = (span.end_time - span.start_time) / 1_000_000.0 @@ -530,7 +555,13 @@ async def main() -> None: global _accumulator, _compiled_graph _accumulator = LlmUsageAccumulator() _compiled_graph = build_graph() - _compiled_graph.attach_observer(_build_otel_observer(span_exporter)) + # Keep the OTel observer reachable so we can ``shutdown()`` it + # after drain — the root ``openarmature.invocation`` span only + # closes on shutdown, and the in-memory exporter only surfaces + # closed spans through ``get_finished_spans()``. Production + # deployments do the same dance at process exit. + otel_observer = _build_otel_observer(span_exporter) + _compiled_graph.attach_observer(otel_observer) _compiled_graph.attach_observer(_build_langfuse_observer(langfuse_client)) _compiled_graph.attach_observer(_accumulator) graph = _compiled_graph @@ -581,8 +612,14 @@ async def main() -> None: finally: # drain() is required for short-lived processes: invoke() # returns when the graph reaches END regardless of whether - # the observer queue has finished draining. + # the observer queue has finished draining. shutdown() on + # the OTel observer closes the root ``openarmature.invocation`` + # span so it lands in the exporter alongside the per-node + # spans; the Langfuse observer has no analog because it + # writes Trace + Observation entities synchronously through + # the client. await graph.drain() + otel_observer.shutdown() await _get_provider().aclose() if final is not None: From c5952e0a038ba50cba7b3bafc8ad3c27bc58fc45 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 5 Jun 2026 18:24:03 -0700 Subject: [PATCH 3/4] Tighten accumulator example per PR review Eight PR review threads, addressing four distinct issues. state.invocation_id -> current_invocation_id() in the example module docstring and walkthrough doc. The runnable persist() uses current_invocation_id() because State has no invocation_id field by default; the docstring snippets had drifted to the wrong shape. assert -> RuntimeError in persist(). The three runtime preconditions (_compiled_graph not None, _accumulator not None, current_invocation_id() not None) now raise explicit RuntimeError so the failure mode stays informative under python -O, which strips asserts and would otherwise produce silent None dereferences. InvocationCompletedEvent backstop cleanup in the accumulator. persist()'s drop is the fast path; if drain_events_for times out and the deliver loop later processes late-arriving LLM events, setdefault() would recreate a bucket that nothing ever cleans up. Adding InvocationCompletedEvent handling at the top of __call__ drops any leftover bucket on invocation completion. The drop is idempotent so it composes with persist()'s drop without harm. Defensive total_tokens derivation. LlmEventPayload makes all three usage fields optional; providers that emit prompt + completion but no total (anything non-OpenAI in practice) would leave bucket.total_tokens at zero while the sub-fields are correct. Now derives total from prompt + completion when total is None on the payload. build_graph() self-contained per the demo convention. Previously, persist() depended on _compiled_graph + _accumulator module globals that only main() populated, so a copy-pasting reader doing `graph = build_graph(); await graph.invoke(...)` would hit RuntimeError at persist time. build_graph() now owns the accumulator construction, the graph attach, and the global wiring. main() drops the duplicate construction and just attaches OTel + Langfuse on top. --- docs/examples/production-observability.md | 2 +- examples/production-observability/main.py | 72 +++++++++++++++++------ 2 files changed, 54 insertions(+), 20 deletions(-) diff --git a/docs/examples/production-observability.md b/docs/examples/production-observability.md index d9f91bdb..b24ed666 100644 --- a/docs/examples/production-observability.md +++ b/docs/examples/production-observability.md @@ -89,7 +89,7 @@ sees the same logical events represented two ways. carrying an `LlmEventPayload`. It accumulates per-invocation token totals in memory, indexed by `current_invocation_id()`. The terminal `persist` node calls - `await graph.drain_events_for(state.invocation_id, timeout=2.0)` + `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 missing the most-recent LLM event's tokens (the deliver loop diff --git a/examples/production-observability/main.py b/examples/production-observability/main.py index ec92657e..b5b8bcf9 100644 --- a/examples/production-observability/main.py +++ b/examples/production-observability/main.py @@ -44,7 +44,7 @@ - **Queryable accumulator observer + per-invocation drain.** A third observer (``LlmUsageAccumulator``) rolls up LLM token totals per invocation. A terminal ``persist`` node calls - ``graph.drain_events_for(state.invocation_id)`` to synchronize on + ``graph.drain_events_for(current_invocation_id())`` to synchronize on the deliver loop, then reads the accumulator's 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 @@ -98,6 +98,7 @@ END, CompiledGraph, GraphBuilder, + InvocationCompletedEvent, NodeEvent, NodeException, ObserverEvent, @@ -190,6 +191,14 @@ def __init__(self) -> None: self._by_invocation: dict[str, _UsageBucket] = {} async def __call__(self, event: ObserverEvent) -> None: + # Backstop cleanup: drop any leftover bucket on invocation + # completion. persist() normally drops first, but if its + # drain_events_for timed out, late-delivered LLM events + # could recreate the bucket via setdefault(). Dropping + # again here is idempotent. + if isinstance(event, InvocationCompletedEvent): + self._by_invocation.pop(event.invocation_id, None) + return if not isinstance(event, NodeEvent): return if event.namespace != LLM_NAMESPACE: @@ -213,8 +222,15 @@ async def __call__(self, event: ObserverEvent) -> None: bucket.prompt_tokens += payload.prompt_tokens if payload.completion_tokens is not None: bucket.completion_tokens += payload.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 # Consumers MUST synchronize on ``drain_events_for`` before @@ -337,10 +353,18 @@ async def respond(state: BriefingState) -> dict[str, Any]: # data is available, rather than failing the whole invocation. async def persist(_state: BriefingState) -> dict[str, Any]: """Drain the deliver loop, read the LLM-usage rollup, drop the bucket.""" - assert _compiled_graph is not None - assert _accumulator is not None + # Use explicit RuntimeError rather than ``assert`` so the failure + # mode stays informative under ``python -O`` (which strips asserts + # and would otherwise turn these into silent ``None`` dereferences + # at the next attribute access). + if _compiled_graph is None or _accumulator is None: + raise RuntimeError( + "persist node requires _compiled_graph and _accumulator to be set " + "before invoke() — see main() for the initialization pattern" + ) invocation_id = current_invocation_id() - assert invocation_id is not None + if invocation_id is None: + raise RuntimeError("persist node called outside an active invocation") summary = await _compiled_graph.drain_events_for(invocation_id, timeout=2.0) if summary.timeout_reached: # Production: emit an SLO-breach metric. Demo: surface the @@ -374,14 +398,26 @@ def build_graph() -> CompiledGraph[BriefingState]: """Two-node graph: respond -> persist -> END. TimingMiddleware wraps the respond node so wall-clock duration - is captured per call. The persist node runs synchronously after + is captured per call. The persist node runs synchronously after respond returns; it drains the deliver loop for the current invocation, reads the LLM-usage accumulator's bucket, drops the - bucket, and prints a cost summary. No other middleware + bucket, and prints a cost summary. No other middleware (RetryMiddleware lives in the fan-out-with-retry / parallel- - branches examples; this one's scope is observability).""" + branches examples; this one's scope is observability). + + The ``LlmUsageAccumulator`` is constructed, attached to the + compiled graph, and registered to the module-level singletons + so the persist node (which reads from globals to stay closure- + free) can reach it without help from the caller. The factory + is self-contained — ``graph = build_graph(); await graph.invoke(...)`` + works on its own. OTel + Langfuse observers are NOT attached + here; those are observability-stack choices made at the call + site, and ``main()`` attaches them after build_graph() returns. + """ + global _accumulator, _compiled_graph timing = TimingMiddleware(node_name="respond", on_complete=_emit_timing) - return ( + _accumulator = LlmUsageAccumulator() + _compiled_graph = ( GraphBuilder(BriefingState) .add_node("respond", respond, middleware=[timing]) .add_node("persist", persist) @@ -390,6 +426,8 @@ def build_graph() -> CompiledGraph[BriefingState]: .set_entry("respond") .compile() ) + _compiled_graph.attach_observer(_accumulator) + return _compiled_graph # --------------------------------------------------------------------------- @@ -549,22 +587,18 @@ async def main() -> None: span_exporter = InMemorySpanExporter() langfuse_client = InMemoryLangfuseClient() - # Module-level singletons populated before invoke so the persist - # node (which is plain-async, no closure) can reach the graph for - # drain_events_for and the accumulator for the read + drop. - global _accumulator, _compiled_graph - _accumulator = LlmUsageAccumulator() - _compiled_graph = build_graph() + # build_graph() owns the accumulator construction + attachment + # plus the module-level singleton wiring. main() attaches the + # observability-stack observers on top. + graph = build_graph() # Keep the OTel observer reachable so we can ``shutdown()`` it # after drain — the root ``openarmature.invocation`` span only # closes on shutdown, and the in-memory exporter only surfaces - # closed spans through ``get_finished_spans()``. Production + # closed spans through ``get_finished_spans()``. Production # deployments do the same dance at process exit. otel_observer = _build_otel_observer(span_exporter) - _compiled_graph.attach_observer(otel_observer) - _compiled_graph.attach_observer(_build_langfuse_observer(langfuse_client)) - _compiled_graph.attach_observer(_accumulator) - graph = _compiled_graph + graph.attach_observer(otel_observer) + graph.attach_observer(_build_langfuse_observer(langfuse_client)) # Caller-supplied multi-tenant metadata. Both observers pick # the entries up: OTel attaches them as ``openarmature.user.*`` From 88903ec1c4ec9ef9f963343bf80505cd4d37c26a Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 5 Jun 2026 18:36:34 -0700 Subject: [PATCH 4/4] Reconcile docstrings + comments with refactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four stale-text findings from a second PR review pass — all caused by the previous review pass changing behavior without fully sweeping the surrounding documentation. Module docstring snippet now shows the full call shape: ``await graph.drain_events_for(current_invocation_id(), timeout=2.0)``, matching the runnable persist() pattern (the previous snippet stripped the await + timeout for brevity but under-described the API). The accumulator's drop() comment block was rewritten to describe the actual two-step lifecycle: fast-path explicit drop after read by the terminal node, plus the InvocationCompletedEvent backstop that the prior pass added. The old comment claimed "does NOT auto-drop on InvocationCompletedEvent" which directly contradicted the implementation. RuntimeError message in persist() now points readers at build_graph() instead of main() for the initialization pattern — the prior pass moved the singleton wiring into build_graph but left the error message pointing at the old call site. Walkthrough doc's "three observers attached at compile time" becomes "three observers attached before invoke", which is honest for both the build_graph-side accumulator attachment and the main-side OTel + Langfuse attachments. attach_observer happens after compile() in the OA API regardless of which function calls it. --- docs/examples/production-observability.md | 2 +- examples/production-observability/main.py | 38 +++++++++++++---------- 2 files changed, 22 insertions(+), 18 deletions(-) diff --git a/docs/examples/production-observability.md b/docs/examples/production-observability.md index b24ed666..9d226d0f 100644 --- a/docs/examples/production-observability.md +++ b/docs/examples/production-observability.md @@ -16,7 +16,7 @@ loop with `drain_events_for`. ## Overview Two nodes (`respond` then `persist`), one LLM call, three observers -attached at compile time. The pipeline takes a question, calls the +attached before invoke. The pipeline takes a question, calls the LLM, returns the answer, then synchronizes on the observer queue and rolls up token cost. The interesting part is the observability wiring: diff --git a/examples/production-observability/main.py b/examples/production-observability/main.py index b5b8bcf9..9a18ba3a 100644 --- a/examples/production-observability/main.py +++ b/examples/production-observability/main.py @@ -44,17 +44,17 @@ - **Queryable accumulator observer + per-invocation drain.** A third observer (``LlmUsageAccumulator``) rolls up LLM token totals per invocation. A terminal ``persist`` node calls - ``graph.drain_events_for(current_invocation_id())`` to synchronize on - the deliver loop, then reads the accumulator's 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: ``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. + ``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 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: + ``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. Complementary to the observer-hooks example (three observers side-by-side) and the langfuse-observability example (Langfuse @@ -243,11 +243,15 @@ def get_bucket(self, invocation_id: str) -> _UsageBucket | None: """Read the accumulated bucket for an invocation.""" return self._by_invocation.get(invocation_id) - # The accumulator does NOT auto-drop on - # ``InvocationCompletedEvent`` — a terminal node legitimately - # needs to read the bucket BEFORE the invocation completes, and - # auto-drop would race the read. Callers invoke ``drop()`` - # explicitly after reading. + # Bucket lifecycle is two-step. Fast path: a terminal node calls + # ``drop()`` immediately after reading via ``get_bucket()`` — + # that's the normal case and runs while the invocation is still + # active. Backstop: the accumulator's ``__call__`` also drops + # any bucket still present when ``InvocationCompletedEvent`` + # arrives. The backstop closes the leak where + # ``drain_events_for`` times out, the terminal node drops a + # stale bucket, then late-delivered LLM events ``setdefault()`` + # a fresh bucket that nothing would otherwise clean up. def drop(self, invocation_id: str) -> None: """Release the bucket for an invocation.""" self._by_invocation.pop(invocation_id, None) @@ -360,7 +364,7 @@ async def persist(_state: BriefingState) -> dict[str, Any]: if _compiled_graph is None or _accumulator is None: raise RuntimeError( "persist node requires _compiled_graph and _accumulator to be set " - "before invoke() — see main() for the initialization pattern" + "before invoke() — see build_graph() for the initialization pattern" ) invocation_id = current_invocation_id() if invocation_id is None: