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..9d226d0f 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 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: - `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(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 + 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,13 +127,15 @@ 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 --- 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= @@ -133,12 +157,33 @@ 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 - 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 6be228d2..9a18ba3a 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 + ``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 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,16 @@ InMemorySpanExporter, ) -from openarmature.graph import END, GraphBuilder, NodeException, State +from openarmature.graph import ( + END, + CompiledGraph, + GraphBuilder, + InvocationCompletedEvent, + NodeEvent, + NodeException, + ObserverEvent, + State, +) from openarmature.graph.middleware import TimingMiddleware, TimingRecord from openarmature.llm import ( LlmProviderError, @@ -88,6 +112,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 +151,120 @@ 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: + # 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: + 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 + # 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 + # 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) + + # 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) + + +# 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,26 +335,103 @@ 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.""" + # 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 build_graph() for the initialization pattern" + ) + invocation_id = current_invocation_id() + 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 + # 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). + + 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_edge("respond", END) + .add_node("persist", persist) + .add_edge("respond", "persist") + .add_edge("persist", END) .set_entry("respond") .compile() ) + _compiled_graph.attach_observer(_accumulator) + return _compiled_graph # --------------------------------------------------------------------------- @@ -269,8 +486,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] = [] @@ -278,18 +530,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 @@ -349,8 +591,17 @@ async def main() -> None: span_exporter = InMemorySpanExporter() langfuse_client = InMemoryLangfuseClient() + # build_graph() owns the accumulator construction + attachment + # plus the module-level singleton wiring. main() attaches the + # observability-stack observers on top. graph = build_graph() - 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) + graph.attach_observer(otel_observer) graph.attach_observer(_build_langfuse_observer(langfuse_client)) # Caller-supplied multi-tenant metadata. Both observers pick @@ -399,8 +650,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: