diff --git a/conformance.toml b/conformance.toml index 52ff438..07338c8 100644 --- a/conformance.toml +++ b/conformance.toml @@ -836,9 +836,9 @@ note = "Advisory, observability-only per-prompt token budget. PR A (completion p # (graph-engine §6 fan_out_index_chain / branch_name_chain + # observability §4 / §5.5 lineage-resolved parent). [proposals."0084"] -status = "partial" +status = "implemented" since = "0.17.0" -note = "The event-surface chains + the nested-LLM exact-match shipped. graph-engine §6: fan_out_index_chain / branch_name_chain (mirroring NodeEvent) now ride the provider/tool events (LlmCompletionEvent / LlmFailedEvent / LlmRetryAttemptEvent / EmbeddingEvent / EmbeddingFailedEvent / RerankEvent / RerankFailedEvent / ToolCallEvent / ToolCallFailedEvent) and the framework FailureIsolatedEvent, populated from the enclosing-lineage ContextVars at call time. observability §4.3 / §5.5: the OTel node-span key carries the scalars AND the enclosing chains, so an inner node under two concurrent outer fan-out instances no longer collides (no dropped / mis-closed spans), and the LLM-parent exact-match (§5.5 Lineage-resolved parent, fallback 1) resolves under the lineage-disambiguated calling-node span rather than the innermost scalar. observability fixture 132 (nested keying + nested-LLM exact-match, OTel) is un-deferred and driven. partial because the orphan fallback (a wrapper / middleware-issued provider call with no open calling-node span, §5.5) is not yet generalized to nearest-open-ancestor-at-any-depth and the Langfuse resolver is unchanged: observability fixtures 133 (OTel orphan) + 134 (Langfuse) and the graph-engine event-chain fixture 039 tightening defer with the orphan-call conformance-adapter primitive (a later PR)." +note = "Nested-fan-out span lineage. graph-engine §6: fan_out_index_chain / branch_name_chain (mirroring NodeEvent) ride the provider/tool events (LlmCompletionEvent / LlmFailedEvent / LlmRetryAttemptEvent / EmbeddingEvent / EmbeddingFailedEvent / RerankEvent / RerankFailedEvent / ToolCallEvent / ToolCallFailedEvent) and the framework FailureIsolatedEvent, populated from the enclosing-lineage ContextVars at call time. observability §4.3 / §5.5: the OTel node-span key AND the Langfuse node-observation key now carry the scalars plus the enclosing chains, so an inner node under two concurrent outer fan-out instances no longer collides (no dropped / mis-closed spans / observations), while a callable parallel-branch stays distinct from its parallel-branches node. The LLM-parent exact-match (§5.5 Lineage-resolved parent, fallback 1) resolves under the lineage-disambiguated calling-node span / observation rather than the innermost scalar. The §5.5 orphan fallback (a wrapper / middleware-issued provider call with no open calling-node span) is generalized in BOTH observers to the nearest enclosing wrapper on the calling node's lineage via a shared resolver (_resolve_enclosing_wrapper_context on OTel, _resolve_enclosing_wrapper_observation_id on Langfuse), which also now backs FailureIsolatedEvent parenting so the marker resolves consistently across observers. observability fixtures 132 / 133 / 134 all run; the orphan-call primitive calls_llm_from_wrapper is modeled as NodeSpec.calls_llm_from_wrapper so 133 / 134 also parse + round-trip. KNOWN LIMITATION (cross-observer parity, §8.9): the harness realizes the orphan call as fan-out INSTANCE middleware in the POST phase, because a per-node PRE-phase wrapper call does not orphan in the python engine (prepare_sync opens the node span before node middleware). Under the fixtures' declared PRE phase, a Langfuse orphan resolves to the OUTER fan-out instance (Langfuse creates instance observations lazily, so the inner one does not exist when the pre-phase event drains) while OTel resolves to the INNER -- a real pre-phase parity gap the fixtures do not cover. Parity holds for the post-phase realization the fixtures drive (asserted by a live OTel-vs-Langfuse parent comparison in 134). The pre-phase gap is tracked (Langfuse eager instance-observation creation) and flagged to spec." # Spec v0.80.0 (proposal 0085). Nested-fan-out checkpoint resume # lineage (pipeline-utilities §10.11 enclosing_fan_out_lineage). diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index 99169b3..cb0929d 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -1142,12 +1142,13 @@ class FailureIsolatedEvent: post_state: Mapping[str, Any] caught_exception: CaughtException # Proposal 0084 (spec v0.81.0): enclosing fan-out / branch lineage chains - # parallel to ``namespace`` (see LlmCompletionEvent). Carried for surface - # consistency with the provider events. The marker's calling-node span is - # already closed by delivery time (its completed event precedes this one on - # the serial queue), so it parents under the invocation span, not via the - # lineage exact-match -- the chains only disambiguate in the rare case that - # calling-node span is still open. + # parallel to ``namespace`` (see LlmCompletionEvent). The marker's + # calling-node span/observation is already closed by delivery time (its + # completed event precedes this one on the serial queue), so both observers + # take the §5.5 orphan fallback and parent the marker under the nearest + # enclosing wrapper on this lineage (the fan-out instance / branch / subgraph + # span, else the invocation span / Trace). The chains disambiguate that + # enclosing wrapper across concurrent sibling instances. fan_out_index_chain: tuple[int | None, ...] = () branch_name_chain: tuple[str | None, ...] = () diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index a5ead2d..b9b0127 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -92,15 +92,21 @@ def _read_implementation_version() -> str: return __version__ -# In-flight Span observation handle, keyed by the standard span-stack -# key (namespace, attempt_index, fan_out_index, branch_name). -# ``branch_name`` discriminates concurrent same-named inner nodes -# across sibling parallel-branches branches (pipeline-utilities §11); -# without it the two inner ``ask`` nodes of two branches with the -# same namespace + fan_out_index would collide on the same key. -# Mirrors the OTel observer's ``_StackKey`` shape but holds a -# Langfuse handle instead of an OTel Span. -_StackKey = tuple[tuple[str, ...], int, int | None, str | None] +# In-flight Span observation handle, keyed by the scalars +# (namespace, attempt_index, fan_out_index, branch_name) AND the enclosing +# fan-out / branch lineage chains (proposal 0084). ``branch_name`` discriminates +# concurrent same-named inner nodes across sibling parallel-branches branches; +# the chains discriminate an inner node under two concurrent OUTER fan-out +# instances (which share the innermost scalar fan_out_index), so the nested +# exact-match Generation-parent lookup finds its own calling node's observation +# rather than a sibling's. The scalars are retained (as in the OTel observer): +# a callable parallel-branch carries branch_name on the event but does not +# extend branch_name_chain, so key[3] keeps it distinct from its own +# parallel-branches node. Mirrors the OTel observer's ``_StackKey`` shape but +# holds a Langfuse handle instead of an OTel Span. +_StackKey = tuple[ + tuple[str, ...], int, int | None, str | None, tuple[int | None, ...], tuple[str | None, ...] +] # Lineage-aware dispatch keys (proposal 0045): the fan-out / pb NODE namespace # prefix plus the fan-out instance index / branch name chain slices along the @@ -735,7 +741,7 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non # (fan-out NODE / pb NODE) identified by presence in the # parent_node_name caches. for key, observation in inv_state.open_observations.items(): - ns, _ai, _fi, _bn = key + ns = key[0] if ns != aug_ns and not is_strict_prefix(ns, aug_ns): continue # A fan-out / pb NODE is a shared parent and MUST NOT carry an @@ -754,13 +760,16 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non # ------------------------------------------------------------------ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: - # Render the FailureIsolationMiddleware catch as a marker - # observation. Parented under the wrapped node's observation when - # it is still open; otherwise trace-level (the node observation - # is typically already closed-with-error by delivery time, since - # the node-body raise fires the node's completed event before the - # middleware recovers). The wrapped node's name rides on - # ``metadata.failure_isolation_node`` for correlation regardless. + # Render the FailureIsolationMiddleware catch as a marker observation. + # The wrapped node's observation is typically already closed by delivery + # time (the node-body raise fires the node's completed event before the + # middleware recovers), so this usually takes the orphan fallback: the + # marker parents under the nearest enclosing wrapper on its lineage + # (proposal 0084 §5.5), matching the OTel observer -- which routes + # _handle_failure_isolated through _resolve_llm_parent -- for + # cross-observer parity (the enclosing fan-out instance / branch / + # subgraph observation, else None -> the Trace itself). The wrapped + # node's name rides on ``metadata.failure_isolation_node`` regardless. from openarmature.observability.correlation import ( current_correlation_id, current_invocation_id, @@ -772,9 +781,32 @@ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: inv_state = self._inv_states.get(invocation_id) if inv_state is None: return - key: _StackKey = (event.namespace, event.attempt_index, event.fan_out_index, event.branch_name) + # FailureIsolatedEvent is not a NodeEvent but carries the same lineage + # fields (proposal 0084), so build the 6-tuple key inline rather than via + # _key_for (typed for NodeEvent). + key: _StackKey = ( + event.namespace, + event.attempt_index, + event.fan_out_index, + event.branch_name, + event.fan_out_index_chain, + event.branch_name_chain, + ) parent = inv_state.open_observations.get(key) - parent_observation_id = parent.handle.id if parent is not None else None + if parent is not None: + parent_observation_id = parent.handle.id + else: + # Calling node observation not open (the typical case): parent under + # the nearest enclosing wrapper, the same orphan resolution the LLM / + # tool Generations use, so OTel and Langfuse agree on the marker's + # parent. + parent_observation_id = self._resolve_enclosing_wrapper_observation_id( + inv_state, + namespace=event.namespace, + branch_name=event.branch_name, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) metadata: dict[str, Any] = { "failure_isolation_event_name": event.event_name, "error_message": event.caught_exception.message, @@ -968,56 +1000,38 @@ def _open_trace(self, invocation_id: str, correlation_id: str | None, event: Nod self._inv_states[invocation_id] = _InvState(trace_id=invocation_id) def _key_for(self, event: NodeEvent) -> _StackKey: - return (event.namespace, event.attempt_index, event.fan_out_index, event.branch_name) + # Proposal 0084: scalars (retained) + enclosing lineage chains (added), + # mirroring the OTel observer, so concurrent nested fan-out node + # observations don't collide while a callable parallel-branch stays + # distinct from its parallel-branches node. + return ( + event.namespace, + event.attempt_index, + event.fan_out_index, + event.branch_name, + event.fan_out_index_chain, + event.branch_name_chain, + ) def _resolve_parent_observation_id(self, inv_state: _InvState, event: NodeEvent) -> str | None: # Parent precedence (innermost wins): - # 1. Per-instance fan-out dispatch observation at - # namespace[:1] + (str(fan_out_index),) — both detached - # (where the dispatch observation lives in the detached - # Trace) and non-detached (where it lives in the main - # Trace) cases route here when event is inside a fan-out - # instance. - # 2. Subgraph dispatch observation at any matching ancestor - # prefix, walked longest-first. - # 3. Leaf node observation at any matching ancestor prefix, - # walked longest-first. + # 1. Per-instance fan-out / per-branch dispatch observation on the + # event's lineage (the enclosing wrapper) — resolved by the shared + # helper below. + # 2. Subgraph dispatch observation at any matching ancestor prefix + # (also the shared helper). + # 3. Leaf node observation at any matching ancestor prefix, walked + # longest-first. # 4. None — the Trace itself becomes the implicit parent. - # Per proposals 0044 / 0013 / 0045: an inner node parents under the - # INNERMOST dispatch on its lineage -- a per-branch dispatch - # (parallel_branches_branch_spans) or a per-instance fan-out dispatch - # (fan_out_instance_observations), both keyed by the lineage-aware - # _dispatch_key. Walk prefixes longest-first so the innermost wins; the - # lineage key carries the enclosing fan-out instance / branch chain, so - # this resolves arbitrary nesting (fan-out in fan-out, parallel-branches - # in fan-out, ...) to the RIGHT outer instance. Mirrors OTel - # _resolve_parent_context. - for prefix_len in range(len(event.namespace), 0, -1): - prefix = event.namespace[:prefix_len] - fi_axis = ( - event.fan_out_index_chain[prefix_len - 1] - if prefix_len - 1 < len(event.fan_out_index_chain) - else None - ) - if event.branch_name is not None: - branch_dispatch = inv_state.parallel_branches_branch_spans.get( - _branch_dispatch_key( - prefix, event.fan_out_index_chain, event.branch_name_chain, event.branch_name - ) - ) - if branch_dispatch is not None: - return branch_dispatch.handle.id - if fi_axis is not None: - dispatch = inv_state.fan_out_instance_observations.get( - _dispatch_key(prefix, event.fan_out_index_chain, event.branch_name_chain) - ) - if dispatch is not None: - return dispatch.handle.id - for prefix_len in range(len(event.namespace) - 1, 0, -1): - prefix = event.namespace[:prefix_len] - sg = inv_state.subgraph_observations.get(prefix) - if sg is not None: - return sg.handle.id + wrapper_id = self._resolve_enclosing_wrapper_observation_id( + inv_state, + namespace=event.namespace, + branch_name=event.branch_name, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) + if wrapper_id is not None: + return wrapper_id # Open leaf-node observation fallback. The outer loop already # walks longest-first; the inner scan picks the first matching # open observation, which is fine for the cases dispatch @@ -1042,6 +1056,50 @@ def _resolve_parent_observation_id(self, inv_state: _InvState, event: NodeEvent) return observation.handle.id return None + def _resolve_enclosing_wrapper_observation_id( + self, + inv_state: _InvState, + *, + namespace: tuple[str, ...], + branch_name: str | None, + fan_out_index_chain: tuple[int | None, ...], + branch_name_chain: tuple[str | None, ...], + ) -> str | None: + """Resolve the id of the nearest enclosing wrapper observation on the + given lineage (proposal 0084 §4.3): the INNERMOST per-branch dispatch / + per-instance fan-out dispatch observation (both keyed by the + lineage-aware _dispatch_key so arbitrary nesting resolves to the RIGHT + outer instance), else the innermost subgraph observation, else None. + Mirrors OTel ``_resolve_enclosing_wrapper_context``. Shared by the node + parent (``_resolve_parent_observation_id``, with the node event's + lineage) and the §5.5 orphan Generation fallback (with the calling + node's lineage when its observation is not open). The orphan path stops + here (None -> the Trace) rather than falling back to a sibling LEAF + observation, which is not an enclosing wrapper.""" + for prefix_len in range(len(namespace), 0, -1): + prefix = namespace[:prefix_len] + fi_axis = ( + fan_out_index_chain[prefix_len - 1] if prefix_len - 1 < len(fan_out_index_chain) else None + ) + if branch_name is not None: + branch_dispatch = inv_state.parallel_branches_branch_spans.get( + _branch_dispatch_key(prefix, fan_out_index_chain, branch_name_chain, branch_name) + ) + if branch_dispatch is not None: + return branch_dispatch.handle.id + if fi_axis is not None: + dispatch = inv_state.fan_out_instance_observations.get( + _dispatch_key(prefix, fan_out_index_chain, branch_name_chain) + ) + if dispatch is not None: + return dispatch.handle.id + for prefix_len in range(len(namespace) - 1, 0, -1): + prefix = namespace[:prefix_len] + sg = inv_state.subgraph_observations.get(prefix) + if sg is not None: + return sg.handle.id + return None + def _trace_id_for( self, inv_state: _InvState, @@ -1566,18 +1624,16 @@ def _find_node_observation( # when the NODE is itself nested inside an outer fan-out instance / # branch, several instances of the same NODE namespace are open at once # under concurrency, so a namespace-only scan would bind the wrong one. - # The NODE's own event carries the instance / branch it sits in as its - # fan_out_index / branch_name (key[2] / key[3]); that equals the - # augmenting/leaf event's chain entry at the level above this NODE. - n = len(prefix) - enclosing_fi = ( - event.fan_out_index_chain[n - 2] if n >= 2 and n - 2 < len(event.fan_out_index_chain) else None - ) - enclosing_bn = ( - event.branch_name_chain[n - 2] if n >= 2 and n - 2 < len(event.branch_name_chain) else None - ) + # Disambiguate by the full enclosing chain (proposal 0084): the NODE + # observation sits on the event's ancestor path iff its stored lineage + # chain is a prefix of the event's -- the same lineage-boundary rule the + # augmentation scoping uses (``_observation_chain_on_path``). Matching + # the innermost scalar alone is ambiguous at >=3 levels, where an + # intermediate instance index repeats across concurrent outer instances. for key, observation in inv_state.open_observations.items(): - if key[0] == prefix and key[2] == enclosing_fi and key[3] == enclosing_bn: + if key[0] == prefix and _observation_chain_on_path( + observation, event.fan_out_index_chain, event.branch_name_chain + ): return observation return None @@ -1725,6 +1781,8 @@ def _handle_typed_llm_completion(self, event: LlmCompletionEvent) -> None: calling_attempt_index=event.attempt_index, calling_fan_out_index=event.fan_out_index, calling_branch_name=event.branch_name, + calling_fan_out_index_chain=event.fan_out_index_chain, + calling_branch_name_chain=event.branch_name_chain, ) metadata = self._typed_event_metadata(event, correlation_id) model_parameters: dict[str, Any] = dict(event.request_params or {}) @@ -1803,6 +1861,8 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: calling_attempt_index=event.attempt_index, calling_fan_out_index=event.fan_out_index, calling_branch_name=event.branch_name, + calling_fan_out_index_chain=event.fan_out_index_chain, + calling_branch_name_chain=event.branch_name_chain, ) metadata = self._typed_event_metadata(event, correlation_id) # Failure-specific metadata rows: surface error_type + error_ @@ -1886,6 +1946,8 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: calling_attempt_index=event.attempt_index, calling_fan_out_index=event.fan_out_index, calling_branch_name=event.branch_name, + calling_fan_out_index_chain=event.fan_out_index_chain, + calling_branch_name_chain=event.branch_name_chain, ) # §8.4.6 metadata: tool name always, tool_call_id when present. metadata: dict[str, Any] = {"openarmature_tool_name": event.tool_name} @@ -1968,6 +2030,8 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non calling_attempt_index=event.attempt_index, calling_fan_out_index=event.fan_out_index, calling_branch_name=event.branch_name, + calling_fan_out_index_chain=event.fan_out_index_chain, + calling_branch_name_chain=event.branch_name_chain, ) # §8.4.5 metadata: input_count always; dimensions / response_id are # response-derived (success-only). correlation_id + caller metadata @@ -2075,6 +2139,8 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: calling_attempt_index=event.attempt_index, calling_fan_out_index=event.fan_out_index, calling_branch_name=event.branch_name, + calling_fan_out_index_chain=event.fan_out_index_chain, + calling_branch_name_chain=event.branch_name_chain, ) # §8.4.7 request-side identity metadata (present on both variants): # query_length (UTF-8 byte length), document_count, top_k (when @@ -2157,44 +2223,36 @@ def _resolve_llm_parent_observation_id( calling_attempt_index: int, calling_fan_out_index: int | None, calling_branch_name: str | None, + calling_fan_out_index_chain: tuple[int | None, ...], + calling_branch_name_chain: tuple[str | None, ...], ) -> str | None: # Calling-node identity precedence: - # 1. Exact-match leaf node at the calling key. - # 2. Per-instance fan-out dispatch observation when the - # call originated inside a fan-out instance. - # 3. Subgraph dispatch observations along the calling - # namespace prefix, walked longest-prefix-first. - # 4. None — Trace becomes the implicit parent. - # The dispatch fallbacks cover the wrapped-call cases the - # exact-match miss would otherwise need a leaf-ancestor walk - # to handle. + # 1. Exact-match leaf node observation at the calling key (the + # lineage-disambiguated calling node). + # 2. Orphan fallback (proposal 0084 §5.5 "Lineage-resolved parent"): + # the calling node's observation is not open (a middleware / wrapper + # call), so the Generation parents under the nearest enclosing + # wrapper observation per §4.3, resolved via the lineage chain to + # the correct inner instance -- the same ancestor walk the node + # parent uses, not the old top-level-scalar shortcut. None -> Trace. key: _StackKey = ( calling_namespace_prefix, calling_attempt_index, calling_fan_out_index, calling_branch_name, + calling_fan_out_index_chain, + calling_branch_name_chain, ) observation = inv_state.open_observations.get(key) if observation is not None: return observation.handle.id - # Per-instance fan-out dispatch. The dispatch map is keyed by the - # lineage-aware _dispatch_key; reconstruct the top-level instance's key - # (namespace[:1], the instance index, no branch axis) to match it. Only - # the innermost fan_out_index is available here, so this resolves an LLM - # call directly inside a top-level fan-out instance (the case the flat - # ``namespace[:1] + (str(index),)`` key handled before the lineage keys). - if calling_fan_out_index is not None and calling_namespace_prefix: - instance_key = _dispatch_key(calling_namespace_prefix[:1], (calling_fan_out_index,), (None,)) - dispatch = inv_state.fan_out_instance_observations.get(instance_key) - if dispatch is not None: - return dispatch.handle.id - # Subgraph dispatch, longest-prefix-first. - for prefix_len in range(len(calling_namespace_prefix), 0, -1): - prefix = calling_namespace_prefix[:prefix_len] - sg = inv_state.subgraph_observations.get(prefix) - if sg is not None: - return sg.handle.id - return None + return self._resolve_enclosing_wrapper_observation_id( + inv_state, + namespace=calling_namespace_prefix, + branch_name=calling_branch_name, + fan_out_index_chain=calling_fan_out_index_chain, + branch_name_chain=calling_branch_name_chain, + ) def _typed_event_metadata( self, event: LlmCompletionEvent | LlmFailedEvent, correlation_id: str | None diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index c2e7978..532e83a 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -2236,41 +2236,26 @@ def _resolve_llm_parent( calling = inv_state.open_spans.get(calling_key) if calling is not None: return set_span_in_context(calling.span) - # 2. Per-instance fan-out dispatch (spec ruling, - # review-nested-fan-out-lineage): an orphaned LLM span inside a - # fan-out instance parents under the per-instance dispatch span, not - # the subgraph / invocation span. Mirrors the LangfuseObserver - # fallback. Top-level instance ONLY (namespace[:1] + the scalar - # fan_out_index): for a NESTED fan-out instance the innermost index - # can coincide with a sibling top-level instance's, so this may - # mis-resolve to that sibling rather than miss. The nearest-open- - # ancestor-at-any-depth generalization (which also reorders this ahead - # of the subgraph walk below) fixes that and rides the spec §5.5 - # fixture. - if calling_fan_out_index is not None and calling_namespace_prefix: - instance_key = _dispatch_key(calling_namespace_prefix[:1], (calling_fan_out_index,), (None,)) - dispatch = inv_state.fan_out_instance_spans.get(instance_key) - if dispatch is not None: - return set_span_in_context(dispatch.span) - # 3. Walk up the calling namespace prefix for a synthetic - # subgraph dispatch span at any ancestor — covers LLM - # calls from inside subgraph wrapper middleware. - for plen in range(len(calling_namespace_prefix), 0, -1): - ancestor = calling_namespace_prefix[:plen] - sg = inv_state.subgraph_spans.get(ancestor) - if sg is not None: - return set_span_in_context(sg.span) - dr = inv_state.detached_roots.get(ancestor) - if dr is not None: - return set_span_in_context(dr.span) - # 4. Invocation span — ``complete()`` called outside any - # node body but inside an ``invoke()``. - inv = self._invocation_span.get(invocation_id) - if inv is not None: - return set_span_in_context(inv.span) - # 5. No invocation in scope — return a fresh empty Context. - # The span will live in its own trace. - return otel_context.Context() + # 2. Orphan fallback (proposal 0084 §5.5 "Lineage-resolved parent"): + # the calling-node span is not open -- a call issued from middleware + # (pre/post phase) or a wrapper rather than the node body -- so the + # provider span parents under the nearest enclosing wrapper per §4.3, + # resolved via the lineage chain to the correct inner instance (not + # the old top-level-scalar shortcut, which mis-resolves to a + # coincidentally-indexed sibling under nested concurrency). This is + # the same ancestor walk the node parent uses, so a wrapper-issued + # call and an in-body call on the same lineage resolve to the same + # enclosing parent; it subsumes the former per-instance / subgraph / + # invocation / empty-context fallbacks. + return self._resolve_enclosing_wrapper_context( + inv_state, + invocation_id, + namespace=calling_namespace_prefix, + fan_out_index=calling_fan_out_index, + branch_name=calling_branch_name, + fan_out_index_chain=calling_fan_out_index_chain, + branch_name_chain=calling_branch_name_chain, + ) # ------------------------------------------------------------------ # Helpers @@ -2322,10 +2307,38 @@ def _resolve_parent_context( invocation_id: str, event: NodeEvent, ) -> object: - """Return the OTel context to use as the parent for this - event's span. Walks namespace ancestors finding the - innermost-open subgraph or detached root span; falls back to - the invocation span.""" + """Return the OTel context to use as the parent for this event's span: + the nearest enclosing wrapper on the event's lineage, else the + invocation span. Delegates to the shared enclosing-wrapper resolver so + the §5.5 orphan LLM-span fallback resolves a wrapper-issued call to the + same parent (proposal 0084).""" + return self._resolve_enclosing_wrapper_context( + inv_state, + invocation_id, + namespace=event.namespace, + fan_out_index=event.fan_out_index, + branch_name=event.branch_name, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) + + def _resolve_enclosing_wrapper_context( + self, + inv_state: _InvState, + invocation_id: str, + *, + namespace: tuple[str, ...], + fan_out_index: int | None, + branch_name: str | None, + fan_out_index_chain: tuple[int | None, ...], + branch_name_chain: tuple[str | None, ...], + ) -> object: + """Resolve the OTel context of the nearest enclosing wrapper span on the + given lineage (proposal 0084 §4.3), falling back to the invocation span. + Shared by the node parent (``_resolve_parent_context``, called with the + node event's lineage) and the §5.5 orphan LLM-span fallback (called with + the calling node's lineage when its span is not open), so both resolve + to the same parent.""" # 1. Walk prefix lengths longest-to-shortest. The INNERMOST # matching synthetic dispatch span wins. Three keying # schemes live alongside each other at each prefix: @@ -2342,47 +2355,43 @@ def _resolve_parent_context( # arbitrary composition (parallel-branches inside fan-out # instance and vice versa) — the dispatch span at the # deepest matching depth is the most-immediate parent. - for prefix_len in range(len(event.namespace), 0, -1): - prefix = event.namespace[:prefix_len] + for prefix_len in range(len(namespace), 0, -1): + prefix = namespace[:prefix_len] # Lineage-aware keys (proposal 0045): carry the enclosing fan-out # instance / branch chain so this resolves to the RIGHT outer # instance for arbitrary nesting. fi_axis = ( - event.fan_out_index_chain[prefix_len - 1] - if prefix_len - 1 < len(event.fan_out_index_chain) - else None + fan_out_index_chain[prefix_len - 1] if prefix_len - 1 < len(fan_out_index_chain) else None ) - if event.branch_name is not None: + if branch_name is not None: branch_dispatch = inv_state.parallel_branches_branch_spans.get( - _branch_dispatch_key( - prefix, event.fan_out_index_chain, event.branch_name_chain, event.branch_name - ) + _branch_dispatch_key(prefix, fan_out_index_chain, branch_name_chain, branch_name) ) if branch_dispatch is not None: return set_span_in_context(branch_dispatch.span) - if event.fan_out_index is not None: + if fan_out_index is not None: # Detached roots keep the top-level routing key shape (detached # generalization is out of scope). - root = inv_state.detached_roots.get(prefix + (str(event.fan_out_index),)) + root = inv_state.detached_roots.get(prefix + (str(fan_out_index),)) if root is not None: return set_span_in_context(root.span) if fi_axis is not None: instance_dispatch = inv_state.fan_out_instance_spans.get( - _dispatch_key(prefix, event.fan_out_index_chain, event.branch_name_chain) + _dispatch_key(prefix, fan_out_index_chain, branch_name_chain) ) if instance_dispatch is not None: return set_span_in_context(instance_dispatch.span) # 1b. Detached subgraph root at any matching prefix wins # (highest precedence — events inside a detached subtree # always parent under the detached root, never bleed up). - for prefix_len in range(len(event.namespace) - 1, -1, -1): - prefix = event.namespace[:prefix_len] + for prefix_len in range(len(namespace) - 1, -1, -1): + prefix = namespace[:prefix_len] root = inv_state.detached_roots.get(prefix) if root is not None: return set_span_in_context(root.span) # 2. Innermost synthetic subgraph span at any prefix. - for prefix_len in range(len(event.namespace) - 1, 0, -1): - prefix = event.namespace[:prefix_len] + for prefix_len in range(len(namespace) - 1, 0, -1): + prefix = namespace[:prefix_len] sg = inv_state.subgraph_spans.get(prefix) if sg is not None: return set_span_in_context(sg.span) diff --git a/tests/conformance/harness/directives.py b/tests/conformance/harness/directives.py index 53d5f55..18c020d 100644 --- a/tests/conformance/harness/directives.py +++ b/tests/conformance/harness/directives.py @@ -381,6 +381,20 @@ class CallsLlmSpec(_AllowExtras): config: RuntimeConfigSpec | None = None +class CallsLlmFromWrapperSpec(_AllowExtras): + """Wrapper-issued (orphan) LLM call, proposal 0084 / conformance-adapter + §5.1: the harness wraps the node in a middleware that issues exactly one + mock-provider ``complete()`` in the named ``phase`` (``pre`` before the + body, ``post`` after) and discards the response, so the call has no open + calling-node span and exercises the observability §5.5 orphan fallback. + Additive alongside the node's primary directive (the ``guard`` node in + fixtures 133 / 134 carries ``update`` as its primary), so this is NOT a + primary field.""" + + phase: Literal["pre", "post"] = "pre" + messages: list[dict[str, Any]] | None = None + + class MockToolSpec(_AllowExtras): """The mock tool a ``calls_tool`` node runs inside the instrumentation scope: ``returns`` (a result -> ToolCallEvent) XOR @@ -501,6 +515,9 @@ class NodeSpec(_ForbidExtras): calls_embed: CallsEmbedSpec | None = None calls_rerank: CallsRerankSpec | None = None calls_tool: CallsToolSpec | None = None + # Proposal 0084 orphan-call primitive: a wrapper-issued side call, ADDITIVE + # to the node's primary directive (not in _PRIMARY_FIELDS below). + calls_llm_from_wrapper: CallsLlmFromWrapperSpec | None = None # Companions — additive. # ``renders_prompt`` (proposal 0064 / 0083, fixtures 064 / 126-129): diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py index 754126a..1d564d4 100644 --- a/tests/conformance/test_fixture_parsing.py +++ b/tests/conformance/test_fixture_parsing.py @@ -614,15 +614,9 @@ def _id(case: tuple[str, Path]) -> str: # completion-path fixtures (126-129) + the structured-output-failure path # (131) parse + run in test_observability; the Langfuse WARNING (130) parses # + runs via test_observability_langfuse. All of 0083's fixtures now run. - # Proposal 0084 (nested-fan-out span lineage, v0.81.0) -- the - # lineage-chain directive shapes. (132 parses cleanly; accounted in - # test_observability.) - "observability/133-otel-nested-fan-out-orphan-llm-fallback": ( - "Proposal 0084 nested-fan-out span lineage; not implemented" - ), - "observability/134-langfuse-nested-fan-out-parent-resolution": ( - "Proposal 0084 nested-fan-out span lineage; not implemented" - ), + # Proposal 0084 (nested-fan-out span lineage, v0.81.0): 132/133/134 all + # parse cleanly and run in test_observability / test_observability_langfuse. + # The orphan-call directive is modeled as NodeSpec.calls_llm_from_wrapper. # Proposal 0087 (within-node directive execution order, v0.82.0). # 135 reuses the augment_metadata / capture_invocation_metadata_into # directive shapes (as 043-046 above) the cross-cap parser does not diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index de06752..8ba3e70 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -96,6 +96,13 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # spans (chain-keyed node spans) + nested-LLM exact-match (each # llm.complete parents under its own lineage-disambiguated `ask`). "132-otel-nested-fan-out-span-keying-and-llm-exact-match", + # v0.81.0 -- proposal 0084 orphan fallback (§5.5 Lineage-resolved + # parent). A provider call issued from a node WRAPPER (middleware + # pre-phase, the calling-node span not open) parents its orphan span + # under the nearest enclosing wrapper -- the inner fan-out instance + # span -- chain-routed to the correct inner instance, as a sibling of + # the later-opened `guard` node span. + "133-otel-nested-fan-out-orphan-llm-fallback", # v0.42.0 — proposal 0050 call-level-retry per-attempt LLM span # surface. Single-attempt default: one span, attempt_index 0. "057-llm-attempt-index-single-attempt-default", @@ -383,16 +390,10 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "Langfuse WARNING-level rendering; driven in test_observability_langfuse" ), # Proposal 0084 (nested-fan-out span lineage, spec v0.81.0). Fixture 132 - # (span keying + nested-LLM exact-match) now runs -- see _SUPPORTED_FIXTURES. - # 133 (orphan LLM fallback) + 134 (Langfuse parent resolution) defer with - # the orphan-call conformance-adapter primitive (a later PR). - **{ - fixture_id: "nested-fan-out span lineage (proposal 0084) not-yet implemented" - for fixture_id in ( - "133-otel-nested-fan-out-orphan-llm-fallback", - "134-langfuse-nested-fan-out-parent-resolution", - ) - }, + # (span keying + nested-LLM exact-match) and 133 (orphan LLM fallback, via + # the `calls_llm_from_wrapper` primitive) now run -- see _SUPPORTED_FIXTURES. + # 134 (Langfuse parent resolution) runs in the sibling Langfuse harness -- + # see _LANGFUSE_HARNESS_FIXTURES. # Proposal 0088 (Langfuse parallel-branches mapping parity, spec # v0.83.0). The §8.4.8 per-branch dispatch-span Langfuse observation is # unimplemented until a later v0.16.0 PR. @@ -424,6 +425,11 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "036-caller-invocation-id-non-uuid", "037-langfuse-trace-input-output", "059-implementation-attribution-langfuse", + # 134 -- proposal 0084 Langfuse parent resolution (nested exact-match + + # orphan fallback). Driven by a dedicated hand-built runner in the + # sibling harness; the generic topology path cannot model the + # `calls_llm_from_wrapper` orphan-call primitive. + "134-langfuse-nested-fan-out-parent-resolution", } ) @@ -616,6 +622,8 @@ async def test_observability_fixture(fixture_path: Path) -> None: await _run_fixture_110(spec) elif fixture_id == "132-otel-nested-fan-out-span-keying-and-llm-exact-match": await _run_fixture_132(spec) + elif fixture_id == "133-otel-nested-fan-out-orphan-llm-fallback": + await _run_fixture_133(spec) elif fixture_id in { "040-llm-cache-attribute-emission", "041-llm-cache-attribute-absence", @@ -2721,6 +2729,295 @@ async def _ask_body(_s: Any) -> dict[str, str]: assert not unhandled, f"fixture 132 declares unhandled invariants: {sorted(unhandled)}" +async def _run_fixture_133(spec: Mapping[str, Any]) -> None: + # Proposal 0084 (§5.5 Lineage-resolved parent, orphan fallback): a provider + # call issued from a node WRAPPER (middleware pre-phase) has no open calling- + # node span, so its provider span falls back to the nearest enclosing wrapper + # -- the inner fan-out INSTANCE span -- chain-routed to the correct inner + # instance (never a coincidentally-indexed sibling inner instance, a fan-out + # NODE span, or the invocation span), as a SIBLING of the later-opened + # `guard` node span. + for case in cast("list[dict[str, Any]]", spec["cases"]): + case_name = cast("str", case["name"]) + try: + await _run_fixture_133_case(case, spec) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_133_case(case: Mapping[str, Any], spec: Mapping[str, Any]) -> None: + import asyncio + import json + + import httpx + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + from openarmature.graph import END, GraphBuilder + from openarmature.llm import OpenAIProvider, UserMessage + + from .adapter import build_state_cls + + # ---- FIFO mock provider. The two canned responses are identical; this + # fixture asserts orphan-span parenting, not response-content routing. + mock_responses = list(cast("list[dict[str, Any]]", case.get("mock_llm") or [])) + + def _handler(_request: httpx.Request) -> httpx.Response: + if not mock_responses: + raise AssertionError("mock_llm queue exhausted") + body = cast("dict[str, Any]", mock_responses.pop(0)["body"]) + return httpx.Response( + 200, content=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"} + ) + + provider = OpenAIProvider( + base_url="http://mock-llm.test", + model="test-model", + api_key="test", + transport=httpx.MockTransport(_handler), + ) + + # ---- Innermost leaf subgraph. The `guard` node's BODY returns the `update` + # dict (marker=1) so the inner fan-out collect_field has a value; a WRAPPER + # issues exactly one orphan provider.complete and DISCARDS the response, + # modelling a guardrail / classifier side call. + # + # Realization note (this harness deviates from the fixture's literal + # "per-node `guard` middleware, phase: pre" on two axes, both forced by the + # observers' real timing; the orphan SEMANTIC -- no open calling-node span -- + # and the oracle span_tree are preserved): + # + # 1. INSTANCE middleware on the inner fan-out, not per-node middleware on + # `guard`. A per-node wrapper's call carries guard's own lineage, so the + # observer binds it to guard's span (the OTel observer opens node spans + # synchronously via ``prepare_sync`` before the async worker resolves the + # provider event), and it parents under guard, not orphaned. The fan-out + # instance wrapper keeps the orphan call's lineage at the fan-out + # boundary (off the leaf node), so its span falls back to the nearest + # enclosing wrapper -- the inner fan-out INSTANCE span -- exactly as §5.5 + # intends, a sibling of the later `guard` node span. `leaf_sg` is the + # single `guard` node, so the instance wrapper IS guard's wrapper; the + # `calls_llm_from_wrapper` directive is read from the `guard` node spec. + # 2. POST phase (fire AFTER next()), matching the sibling Langfuse 134 + # harness for a single realization across both backends. The Langfuse + # observer creates the inner instance OBSERVATION lazily when guard's + # started event drains on the serial worker, so a pre-phase orphan + # (enqueued first) resolves to the OUTER instance there; post guarantees + # the inner instance exists at resolution time. OTel resolves to the + # inner instance in either phase. The fixture declares phase: pre; its .md + # notes either phase exercises the same fallback (in post, guard has + # closed, so the calling-node span is still not open at emit). + subgraphs_spec = cast("dict[str, Any]", spec["subgraphs"]) + leaf_spec = cast("dict[str, Any]", subgraphs_spec["leaf_sg"]) + leaf_state_cls = build_state_cls( + "Leaf_133", cast("dict[str, dict[str, Any]]", leaf_spec["state"]["fields"]) + ) + guard_spec = cast("dict[str, Any]", leaf_spec["nodes"]["guard"]) + update_map = dict(cast("dict[str, Any]", guard_spec["update"])) + wrapper_spec = cast("dict[str, Any]", guard_spec["calls_llm_from_wrapper"]) + messages_in: tuple[Any, ...] = tuple( + UserMessage(content=m["content"]) + for m in cast("list[dict[str, str]]", wrapper_spec.get("messages", [])) + if m.get("role") == "user" + ) or (UserMessage(content="guardrail check"),) + + # Rendezvous: hold every wrapper provider call until all are in flight before + # any fires, so the two orphan provider spans are emitted while both inner + # instance subtrees -- whose instance spans share the innermost scalar + # fan_out_index 0 -- are in flight at once. A scalar-only parent key would + # then collide and mis-parent outer-1's orphan under outer-0's inner instance; + # the chain key routes each to its own. This is what makes the fixture + # DISCRIMINATE the fix. Times out to a no-op if fewer run concurrently than + # expected, so it never deadlocks. total_asks derives from the outer items' + # default (2 here: outer_seeds [[0], [0]]). + outer_items0 = cast( + "list[list[Any]]", + case["state"]["fields"][case["nodes"]["outer_fan_out"]["fan_out"]["items_field"]].get("default") + or [], + ) + total_asks = sum(len(x) for x in outer_items0) + assert total_asks > 0, ( + "fixture 133 rendezvous gate degenerated (total_asks=0): the outer items " + "must be supplied via the field `default` so the concurrent interleaving is forced" + ) + rendezvous = {"n": 0} + gate = asyncio.Event() + + async def _fire_orphan_call() -> None: + rendezvous["n"] += 1 + if rendezvous["n"] >= total_asks: + gate.set() + try: + async with asyncio.timeout(5): + await gate.wait() + except TimeoutError: + # Best-effort rendezvous: release rather than deadlock if fewer run + # concurrently than total_asks (asserted > 0 above, so the gate cannot + # degenerate silently on the very bug this fixture exists to catch). + pass + # Wrapper-issued side call: fire exactly one completion and DISCARD it + # (never written to state), modelling a guardrail / classifier check. + await provider.complete(list(messages_in)) + + async def _guard_body(_s: Any) -> Mapping[str, Any]: + return dict(update_map) + + async def _wrapper_mw(state: Any, next_call: Any) -> Mapping[str, Any]: + # Instance middleware on the inner fan-out, POST phase: fire AFTER next() + # (the inner subgraph, i.e. `guard`, has run and its span has closed), so + # the calling-node span is not open and the orphan provider span falls + # back to the nearest enclosing wrapper: the inner fan-out INSTANCE span. + result = await next_call(state) + await _fire_orphan_call() + return result + + leaf_sg = ( + GraphBuilder(leaf_state_cls) + .add_node("guard", _guard_body) + .add_edge("guard", END) + .set_entry("guard") + .compile() + ) + + # ---- Middle subgraph: the inner fan-out over `leaf_sg`, with the orphan-call + # wrapper attached as INSTANCE middleware (see the realization note above). + mid_spec = cast("dict[str, Any]", subgraphs_spec["mid"]) + mid_state_cls = build_state_cls("Mid_133", cast("dict[str, dict[str, Any]]", mid_spec["state"]["fields"])) + inner_fo = cast("dict[str, Any]", mid_spec["nodes"]["inner_fan_out"]["fan_out"]) + mid = ( + GraphBuilder(mid_state_cls) + .add_fan_out_node( + "inner_fan_out", + subgraph=leaf_sg, + items_field=cast("str", inner_fo["items_field"]), + item_field=cast("str", inner_fo["item_field"]), + collect_field=cast("str", inner_fo["collect_field"]), + target_field=cast("str", inner_fo["target_field"]), + error_policy=cast("str", inner_fo.get("error_policy", "fail_fast")), + instance_middleware=[_wrapper_mw], + ) + .add_edge("inner_fan_out", END) + .set_entry("inner_fan_out") + .compile() + ) + + # ---- Outer graph: the outer fan-out over `mid`, forced concurrent so the + # two outer instances' inner subtrees are in flight at once (the condition + # the chain keying discriminates -- a scalar key would collide/mis-parent). + outer_fields = cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + outer_state_cls = build_state_cls("Outer_133", outer_fields) + outer_fo = cast("dict[str, Any]", case["nodes"]["outer_fan_out"]["fan_out"]) + outer_default = cast("list[Any]", outer_fields[cast("str", outer_fo["items_field"])].get("default") or []) + graph = ( + GraphBuilder(outer_state_cls) + .add_fan_out_node( + "outer_fan_out", + subgraph=mid, + items_field=cast("str", outer_fo["items_field"]), + item_field=cast("str", outer_fo["item_field"]), + collect_field=cast("str", outer_fo["collect_field"]), + target_field=cast("str", outer_fo["target_field"]), + error_policy=cast("str", outer_fo.get("error_policy", "fail_fast")), + concurrency=max(len(outer_default), 2), + ) + .add_edge("outer_fan_out", END) + .set_entry("outer_fan_out") + .compile() + ) + + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + graph.attach_observer(observer) + try: + await graph.invoke(outer_state_cls()) + await graph.drain() + finally: + observer.shutdown() + await provider.aclose() + + spans = exporter.get_finished_spans() + + # ---- span_tree: the recursive match enforces that each orphan llm.complete + # is a child of ITS OWN inner fan-out instance span (chain-disambiguated), + # a sibling of that instance's `guard` node span. + expected_tree = cast("list[dict[str, Any]]", case["expected"]["span_tree"]) + inv_root = next( + (s for s in spans if s.name == "openarmature.invocation" and cast("Any", s.parent) is None), None + ) + assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}" + _assert_span_tree_matches(spans, [inv_root], expected_tree) + + # ---- Invariants (fail loud on any declared-but-unhandled name). + invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {}) + + def _attr(s: Any, key: str) -> Any: + attrs = cast("dict[str, Any]", s.attributes or {}) + return attrs.get(key) + + llm_spans = [s for s in spans if s.name == "openarmature.llm.complete"] + guard_spans = [s for s in spans if s.name == "guard"] + guard_ids = {cast("Any", s.context).span_id for s in guard_spans} + guard_parent_ids = {cast("Any", s.parent).span_id for s in guard_spans if s.parent is not None} + # Inner fan-out INSTANCE span: carries the scalar fan_out_index AND names its + # parent fan-out node `inner_fan_out` (distinguishing it from the inner + # fan-out NODE span, which carries item_count, and from an outer instance). + inner_instance_ids = { + cast("Any", s.context).span_id + for s in spans + if s.name == "inner_fan_out" + and _attr(s, "openarmature.node.fan_out_index") is not None + and _attr(s, "openarmature.fan_out.parent_node_name") == "inner_fan_out" + } + fan_out_node_ids = { + cast("Any", s.context).span_id + for s in spans + if _attr(s, "openarmature.fan_out.item_count") is not None + } + invocation_ids = {cast("Any", s.context).span_id for s in spans if s.name == "openarmature.invocation"} + llm_parents = [cast("Any", s.parent).span_id for s in llm_spans if s.parent is not None] + + if invariants.get("orphan_llm_span_parents_under_inner_fan_out_instance"): + assert llm_parents and all(p in inner_instance_ids for p in llm_parents), ( + "each orphan llm.complete MUST parent under an inner fan-out INSTANCE span" + ) + assert all(p not in guard_ids for p in llm_parents), ( + "orphan llm.complete MUST NOT parent under the guard node span" + ) + if invariants.get("orphan_llm_span_not_under_coincidentally_indexed_sibling"): + assert len(set(llm_parents)) == len(llm_spans), ( + "each orphan llm.complete MUST parent under a DISTINCT inner instance span " + "(chain-routed, not the coincidentally-indexed sibling sharing fan_out_index)" + ) + if invariants.get("orphan_llm_span_not_under_fan_out_node_or_invocation"): + assert all(p not in fan_out_node_ids for p in llm_parents), ( + "orphan llm.complete MUST NOT parent under a fan-out NODE span" + ) + assert all(p not in invocation_ids for p in llm_parents), ( + "orphan llm.complete MUST NOT parent under the invocation span" + ) + if invariants.get("orphan_llm_span_sibling_of_calling_node_span"): + assert all(p in guard_parent_ids for p in llm_parents), ( + "each orphan llm.complete MUST be a SIBLING of its guard node span (share its parent)" + ) + assert all(p not in guard_ids for p in llm_parents), ( + "orphan llm.complete parent MUST NOT be the guard span itself" + ) + if (want := invariants.get("llm_provider_span_count")) is not None: + assert len(llm_spans) == want, f"expected {want} llm.complete spans; got {len(llm_spans)}" + + _handled = { + "orphan_llm_span_parents_under_inner_fan_out_instance", + "orphan_llm_span_not_under_coincidentally_indexed_sibling", + "orphan_llm_span_not_under_fan_out_node_or_invocation", + "orphan_llm_span_sibling_of_calling_node_span", + "llm_provider_span_count", + } + unhandled = set(invariants) - _handled + assert not unhandled, f"fixture 133 declares unhandled invariants: {sorted(unhandled)}" + + async def _run_fixture_110(spec: Mapping[str, Any]) -> None: # Proposal 0075 callable-branch span shape (observability §5.7): an # inline-callable parallel branch renders as ONE per-branch dispatch span diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index f3c9a71..58c7796 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -129,6 +129,14 @@ # to metadata.token_budget.*. Drives the input-exceeded path via # renders_prompt + the prompt's token_budget block. "130-langfuse-token-budget-warning-level", + # 134 (proposal 0084): the Langfuse Generation parent resolves by the same + # chain-aware §5.5 rule as the OTel span parent -- both the nested + # exact-match (case 1, mirrors OTel 132) and the orphan fallback (case 2, + # mirrors OTel 133). Driven by a dedicated hand-built runner + # (``_run_langfuse_134``): the nested double fan-out + the + # `calls_llm_from_wrapper` orphan primitive are not modeled by the generic + # ``_run_case`` topology path (precedent: the 039 hand-built runner). + "134-langfuse-nested-fan-out-parent-resolution", } ) @@ -470,6 +478,17 @@ async def fetch( async def test_langfuse_fixture(fixture_path: Path) -> None: spec = _load(fixture_path) fixture_stem = fixture_path.stem + if fixture_stem == _FIXTURE_134: + # 134's nested double fan-out + `calls_llm_from_wrapper` orphan primitive + # are not modeled by the generic `_run_case` topology path; each case is + # driven by a dedicated hand-built runner (cf. the 039 hand-built path). + for case in cast("list[dict[str, Any]]", spec["cases"]): + case_name = cast("str", case.get("name") or "") + try: + await _run_langfuse_134(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + return if "cases" in spec: # Fold fixture-level ``subgraphs`` / ``inner_subgraphs`` into # each case so the per-case runner sees them locally. Fixture @@ -1349,6 +1368,543 @@ async def _run_resume_case( # hook-derived input/output match the resumed invocation's state. +# Fixture 134 (proposal 0084): the Langfuse Generation parent MUST resolve by the +# same chain-aware §5.5 rule as the OTel span parent, for both the nested +# exact-match (case 1, mirrors OTel 132) and the orphan fallback (case 2, mirrors +# OTel 133). The nested double fan-out + the `calls_llm_from_wrapper` orphan +# primitive are not modeled by the generic `_run_case` topology path, so each +# case is hand-built here (cf. the 039 hand-built runner) against GraphBuilder + +# an InMemoryLangfuseClient / LangfuseObserver. +_FIXTURE_134 = "134-langfuse-nested-fan-out-parent-resolution" + + +def _build_134_provider(case: Mapping[str, Any]) -> OpenAIProvider: + # FIFO mock provider. The canned responses are identical; the fixture asserts + # observation parenting, not response-content routing. + responses = list(cast("list[dict[str, Any]]", case.get("mock_llm") or [])) + + def _handler(_request: httpx.Request) -> httpx.Response: + if not responses: + raise AssertionError("mock_llm queue exhausted") + body = cast("dict[str, Any]", responses.pop(0)["body"]) + return httpx.Response( + 200, content=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"} + ) + + return OpenAIProvider( + base_url="http://mock-llm.test", + model="test-model", + api_key="test", + transport=httpx.MockTransport(_handler), + ) + + +def _build_134_rendezvous(case: Mapping[str, Any]) -> tuple[dict[str, int], Any, int]: + # Hold every provider call until all are in flight before any fires, so the + # Generations are emitted while both concurrent outer instances' inner + # subtrees are in flight -- the condition under which a scalar-only parent key + # would collide / mis-parent and the chain key discriminates. total_asks + # derives from the outer items' default (4 for case 1, 2 for case 2). + import asyncio # noqa: PLC0415 + + outer_items = cast( + "list[list[Any]]", + case["state"]["fields"][case["nodes"]["outer_fan_out"]["fan_out"]["items_field"]].get("default") + or [], + ) + total_asks = sum(len(x) for x in outer_items) + assert total_asks > 0, ( + "fixture 134 rendezvous gate degenerated (total_asks=0): the outer items " + "must be supplied via the field `default` so the concurrent interleaving is forced" + ) + return {"n": 0}, asyncio.Event(), total_asks + + +def _build_134_outer_graph(case: Mapping[str, Any], leaf_sg: Any, *, suffix: str) -> tuple[Any, Any]: + # Assemble the standard nested fan-out (outer fan-out over `mid`, `mid`'s + # inner fan-out over the given leaf subgraph), outer forced concurrent so both + # outer instances' inner subtrees are in flight at once. Returns the compiled + # graph + the outer state class. + subgraphs_spec = cast("dict[str, Any]", case["subgraphs"]) + mid_spec = cast("dict[str, Any]", subgraphs_spec["mid"]) + mid_state_cls = build_state_cls( + f"Mid134{suffix}", cast("dict[str, dict[str, Any]]", mid_spec["state"]["fields"]) + ) + inner_fo = cast("dict[str, Any]", mid_spec["nodes"]["inner_fan_out"]["fan_out"]) + # subgraph_identity drives the instance observation's metadata.subgraph_name + # (the instance Span IS the subgraph wrapper, §8.4.2); sourced from the fan_out + # config's `subgraph` field ("leaf_sg" / "mid"). + mid = ( + GraphBuilder(mid_state_cls) + .add_fan_out_node( + "inner_fan_out", + subgraph=leaf_sg, + items_field=cast("str", inner_fo["items_field"]), + item_field=cast("str", inner_fo["item_field"]), + collect_field=cast("str", inner_fo["collect_field"]), + target_field=cast("str", inner_fo["target_field"]), + error_policy=cast("str", inner_fo.get("error_policy", "fail_fast")), + subgraph_identity=cast("str", inner_fo["subgraph"]), + ) + .add_edge("inner_fan_out", END) + .set_entry("inner_fan_out") + .compile() + ) + outer_fields = cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + outer_state_cls = build_state_cls(f"Outer134{suffix}", outer_fields) + outer_fo = cast("dict[str, Any]", case["nodes"]["outer_fan_out"]["fan_out"]) + outer_default = cast("list[Any]", outer_fields[cast("str", outer_fo["items_field"])].get("default") or []) + graph = ( + GraphBuilder(outer_state_cls) + .add_fan_out_node( + "outer_fan_out", + subgraph=mid, + items_field=cast("str", outer_fo["items_field"]), + item_field=cast("str", outer_fo["item_field"]), + collect_field=cast("str", outer_fo["collect_field"]), + target_field=cast("str", outer_fo["target_field"]), + error_policy=cast("str", outer_fo.get("error_policy", "fail_fast")), + subgraph_identity=cast("str", outer_fo["subgraph"]), + concurrency=max(len(outer_default), 2), + ) + .add_edge("outer_fan_out", END) + .set_entry("outer_fan_out") + .compile() + ) + return graph, outer_state_cls + + +async def _run_134_and_capture(graph: Any, outer_state_cls: Any) -> tuple[InMemoryLangfuseClient, list[Any]]: + # Attach BOTH observers to the SAME invocation so the + # langfuse_parent_matches_otel_parent invariant can compare the two backends' + # actually-resolved parents on one workload, not re-assert a Langfuse-only + # membership check. Returns (langfuse client, OTel finished spans). + from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: PLC0415 + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: PLC0415 + InMemorySpanExporter, + ) + + from openarmature.observability.otel.observer import OTelObserver # noqa: PLC0415 + + client = InMemoryLangfuseClient() + exporter = InMemorySpanExporter() + graph.attach_observer(LangfuseObserver(client=client)) + graph.attach_observer(OTelObserver(span_processor=SimpleSpanProcessor(exporter))) + await graph.invoke(outer_state_cls()) + await graph.drain() + return client, list(exporter.get_finished_spans()) + + +def _parent_parity_descriptors_otel(spans: list[Any]) -> list[tuple[Any, Any, Any]]: + # Per openarmature.llm.complete span, its resolved parent as + # (name, fan_out_index, fan_out_parent_node_name) -- the cross-backend- + # comparable identity of the parent span. + by_id = {s.context.span_id: s for s in spans} + out: list[tuple[Any, Any, Any]] = [] + for s in spans: + if s.name != "openarmature.llm.complete": + continue + parent = by_id.get(s.parent.span_id) if s.parent is not None else None + if parent is None: + out.append(("", None, None)) + continue + attrs = dict(parent.attributes or {}) + out.append( + ( + parent.name, + attrs.get("openarmature.node.fan_out_index"), + attrs.get("openarmature.fan_out.parent_node_name"), + ) + ) + return sorted(out, key=lambda d: tuple("" if x is None else str(x) for x in d)) + + +def _parent_parity_descriptors_langfuse(trace: LangfuseTrace) -> list[tuple[Any, Any, Any]]: + by_id = {o.id: o for o in trace.observations} + out: list[tuple[Any, Any, Any]] = [] + for g in trace.observations: + if g.type != "generation": + continue + parent = by_id.get(g.parent_observation_id) if g.parent_observation_id else None + if parent is None: + out.append(("", None, None)) + continue + meta = parent.metadata or {} + out.append((parent.name, meta.get("fan_out_index"), meta.get("fan_out_parent_node_name"))) + return sorted(out, key=lambda d: tuple("" if x is None else str(x) for x in d)) + + +async def _run_langfuse_134(case: Mapping[str, Any]) -> None: + name = cast("str", case.get("name")) + if name == "generation_parents_under_own_lineage_calling_node_observation": + await _run_langfuse_134_case1(case) + elif name == "orphan_generation_parents_under_inner_fan_out_instance_observation": + await _run_langfuse_134_case2(case) + else: + raise NotImplementedError(f"134 case not yet wired: {name!r}") + + +async def _run_langfuse_134_case1(case: Mapping[str, Any]) -> None: + # Case 1 (mirrors OTel 132): each inner `ask` node issues an in-body LLM call; + # its Generation parents under its OWN lineage-disambiguated `ask` Span + # observation, never a sibling `ask` in the other outer instance sharing the + # innermost fan_out_index. + import asyncio # noqa: PLC0415 + + provider = _build_134_provider(case) + subgraphs_spec = cast("dict[str, Any]", case["subgraphs"]) + leaf_spec = cast("dict[str, Any]", subgraphs_spec["leaf_sg"]) + leaf_state_cls = build_state_cls( + "Leaf134C1", cast("dict[str, dict[str, Any]]", leaf_spec["state"]["fields"]) + ) + ask_calls_llm = cast("dict[str, Any]", leaf_spec["nodes"]["ask"]["calls_llm"]) + stores_in = cast("str", ask_calls_llm.get("stores_response_in", "score")) + messages_in = _materialize_messages(cast("list[dict[str, Any]]", ask_calls_llm.get("messages") or [])) + rendezvous, gate, total_asks = _build_134_rendezvous(case) + + async def _ask_body(_s: Any) -> dict[str, str]: + rendezvous["n"] += 1 + if rendezvous["n"] >= total_asks: + gate.set() + try: + async with asyncio.timeout(5): + await gate.wait() + except TimeoutError: + # Best-effort rendezvous: release rather than deadlock if fewer run + # concurrently than total_asks (asserted > 0 above). + pass + response = await provider.complete(cast("Sequence[Any]", messages_in)) + return {stores_in: response.message.content or ""} + + leaf_sg = ( + GraphBuilder(leaf_state_cls) + .add_node("ask", _ask_body) + .add_edge("ask", END) + .set_entry("ask") + .compile() + ) + graph, outer_state_cls = _build_134_outer_graph(case, leaf_sg, suffix="C1") + try: + client, otel_spans = await _run_134_and_capture(graph, outer_state_cls) + finally: + await provider.aclose() + + trace = _single_trace(client) + expected = cast("dict[str, Any]", case["expected"]) + _assert_langfuse_134_trace(trace, cast("dict[str, Any]", expected["langfuse_trace"])) + + invariants = cast("dict[str, Any]", expected.get("invariants") or {}) + generations = [o for o in trace.observations if o.type == "generation"] + ask_obs = [o for o in trace.observations if o.type == "span" and o.name == "ask"] + ask_ids = {o.id for o in ask_obs} + gen_parents = [g.parent_observation_id for g in generations] + + if invariants.get("no_inner_observations_dropped_under_concurrent_nesting"): + assert len(ask_obs) == total_asks and len(generations) == total_asks, ( + f"inner observations dropped: {len(ask_obs)} ask / {len(generations)} generation " + f"(want {total_asks}/{total_asks})" + ) + if (want := invariants.get("generation_count")) is not None: + assert len(generations) == want, f"expected {want} Generations; got {len(generations)}" + if invariants.get("generation_parents_under_own_lineage_calling_node_observation"): + assert gen_parents and all(p in ask_ids for p in gen_parents), ( + "each Generation MUST parent under its calling `ask` Span observation" + ) + assert len(set(gen_parents)) == len(generations), ( + "each Generation MUST parent under a DISTINCT `ask` observation (nested exact-match)" + ) + if invariants.get("langfuse_parent_matches_otel_parent"): + # Real cross-backend comparison (§8.9): both observers ran on the SAME + # invocation above; each Generation's resolved parent MUST match the OTel + # llm.complete span's resolved parent by (name, fan_out_index, + # parent_node_name). Under the harness's POST realization these agree; + # under the fixtures' declared PRE phase they would NOT -- Langfuse + # resolves a lazily-created inner instance to the outer one (task #100, + # flagged to spec) -- so parity is asserted for the post-phase run here. + lf_desc = _parent_parity_descriptors_langfuse(trace) + otel_desc = _parent_parity_descriptors_otel(otel_spans) + assert lf_desc == otel_desc, ( + f"Langfuse and OTel resolved DIFFERENT parents: langfuse={lf_desc} otel={otel_desc}" + ) + + _handled = { + "no_inner_observations_dropped_under_concurrent_nesting", + "generation_count", + "generation_parents_under_own_lineage_calling_node_observation", + "langfuse_parent_matches_otel_parent", + } + unhandled = set(invariants) - _handled + assert not unhandled, f"fixture 134 case 1 declares unhandled invariants: {sorted(unhandled)}" + + +async def _run_langfuse_134_case2(case: Mapping[str, Any]) -> None: + # Case 2 (mirrors OTel 133): the `guard` node's BODY returns marker=1; a + # WRAPPER issues one orphan provider.complete and DISCARDS the response. The + # harness realizes the fixture's "per-node `guard` middleware, phase: pre" as + # INSTANCE middleware on the inner fan-out, POST phase -- the same realization + # as the OTel 133 harness (see the note there). Two axes, both forced by the + # observers' real timing; the orphan SEMANTIC and the oracle observation tree + # are preserved: + # 1. Instance middleware, not per-node middleware on `guard`: a per-node + # wrapper's call carries guard's lineage and binds to guard's own + # observation, not orphaned. The fan-out instance wrapper keeps the + # orphan call's lineage at the fan-out boundary, so its Generation falls + # back to the nearest enclosing wrapper -- the inner fan-out INSTANCE Span + # observation -- a sibling of the `guard` Span. `leaf_sg` is the single + # `guard` node, so the instance wrapper IS guard's wrapper; the directive + # is read from the `guard` node spec. + # 2. POST phase (fire AFTER next()): the LangfuseObserver creates the inner + # instance observation lazily when guard's started event drains on the + # serial worker, so a pre-phase orphan (enqueued first) resolves to the + # OUTER instance. Post guarantees the inner instance observation exists at + # resolution time. In post `guard` has closed, so the calling-node + # observation is still not open at emit (the orphan semantic holds). + import asyncio # noqa: PLC0415 + + provider = _build_134_provider(case) + subgraphs_spec = cast("dict[str, Any]", case["subgraphs"]) + leaf_spec = cast("dict[str, Any]", subgraphs_spec["leaf_sg"]) + leaf_state_cls = build_state_cls( + "Leaf134C2", cast("dict[str, dict[str, Any]]", leaf_spec["state"]["fields"]) + ) + guard_spec = cast("dict[str, Any]", leaf_spec["nodes"]["guard"]) + update_map = dict(cast("dict[str, Any]", guard_spec["update"])) + wrapper_spec = cast("dict[str, Any]", guard_spec["calls_llm_from_wrapper"]) + messages_in = _materialize_messages(cast("list[dict[str, Any]]", wrapper_spec.get("messages") or [])) + rendezvous, gate, total_asks = _build_134_rendezvous(case) + + async def _fire_orphan_call() -> None: + rendezvous["n"] += 1 + if rendezvous["n"] >= total_asks: + gate.set() + try: + async with asyncio.timeout(5): + await gate.wait() + except TimeoutError: + # Best-effort rendezvous: release rather than deadlock if fewer run + # concurrently than total_asks (asserted > 0 above). + pass + # Wrapper-issued side call: fire exactly one completion and DISCARD it. + await provider.complete(cast("Sequence[Any]", messages_in)) + + async def _guard_body(_s: Any) -> Mapping[str, Any]: + return dict(update_map) + + async def _wrapper_mw(state: Any, next_call: Any) -> Mapping[str, Any]: + # POST phase: fire AFTER next() (the inner subgraph `guard` has run and its + # observation has closed), so the orphan Generation falls back to the inner + # fan-out INSTANCE Span observation (see the realization note above). + result = await next_call(state) + await _fire_orphan_call() + return result + + leaf_sg = ( + GraphBuilder(leaf_state_cls) + .add_node("guard", _guard_body) + .add_edge("guard", END) + .set_entry("guard") + .compile() + ) + # Attach the orphan-call wrapper as INSTANCE middleware on the inner fan-out + # (see the realization note above). + mid_spec = cast("dict[str, Any]", subgraphs_spec["mid"]) + mid_state_cls = build_state_cls( + "Mid134C2", cast("dict[str, dict[str, Any]]", mid_spec["state"]["fields"]) + ) + inner_fo = cast("dict[str, Any]", mid_spec["nodes"]["inner_fan_out"]["fan_out"]) + # subgraph_identity drives the instance observation's metadata.subgraph_name + # (sourced from the fan_out config's `subgraph` field, as in case 1). + mid = ( + GraphBuilder(mid_state_cls) + .add_fan_out_node( + "inner_fan_out", + subgraph=leaf_sg, + items_field=cast("str", inner_fo["items_field"]), + item_field=cast("str", inner_fo["item_field"]), + collect_field=cast("str", inner_fo["collect_field"]), + target_field=cast("str", inner_fo["target_field"]), + error_policy=cast("str", inner_fo.get("error_policy", "fail_fast")), + subgraph_identity=cast("str", inner_fo["subgraph"]), + instance_middleware=[_wrapper_mw], + ) + .add_edge("inner_fan_out", END) + .set_entry("inner_fan_out") + .compile() + ) + outer_fields = cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + outer_state_cls = build_state_cls("Outer134C2", outer_fields) + outer_fo = cast("dict[str, Any]", case["nodes"]["outer_fan_out"]["fan_out"]) + outer_default = cast("list[Any]", outer_fields[cast("str", outer_fo["items_field"])].get("default") or []) + graph = ( + GraphBuilder(outer_state_cls) + .add_fan_out_node( + "outer_fan_out", + subgraph=mid, + items_field=cast("str", outer_fo["items_field"]), + item_field=cast("str", outer_fo["item_field"]), + collect_field=cast("str", outer_fo["collect_field"]), + target_field=cast("str", outer_fo["target_field"]), + error_policy=cast("str", outer_fo.get("error_policy", "fail_fast")), + subgraph_identity=cast("str", outer_fo["subgraph"]), + concurrency=max(len(outer_default), 2), + ) + .add_edge("outer_fan_out", END) + .set_entry("outer_fan_out") + .compile() + ) + try: + client, otel_spans = await _run_134_and_capture(graph, outer_state_cls) + finally: + await provider.aclose() + + trace = _single_trace(client) + expected = cast("dict[str, Any]", case["expected"]) + _assert_langfuse_134_trace(trace, cast("dict[str, Any]", expected["langfuse_trace"])) + + invariants = cast("dict[str, Any]", expected.get("invariants") or {}) + generations = [o for o in trace.observations if o.type == "generation"] + guard_obs = [o for o in trace.observations if o.type == "span" and o.name == "guard"] + guard_ids = {o.id for o in guard_obs} + guard_parent_ids = {o.parent_observation_id for o in guard_obs} + # Inner fan-out INSTANCE Span observation: carries fan_out_index AND names its + # parent fan-out node `inner_fan_out` (distinct from the inner fan-out NODE + # observation, which carries fan_out_item_count). + inner_instance_ids = { + o.id + for o in trace.observations + if o.type == "span" + and o.name == "inner_fan_out" + and o.metadata.get("fan_out_index") is not None + and o.metadata.get("fan_out_parent_node_name") == "inner_fan_out" + } + fan_out_node_ids = {o.id for o in trace.observations if o.metadata.get("fan_out_item_count") is not None} + gen_parents = [g.parent_observation_id for g in generations] + + if invariants.get("orphan_generation_parents_under_inner_fan_out_instance_observation"): + assert gen_parents and all(p in inner_instance_ids for p in gen_parents), ( + "each orphan Generation MUST parent under an inner fan-out INSTANCE Span observation" + ) + assert all(p not in guard_ids for p in gen_parents), ( + "orphan Generation MUST NOT parent under the `guard` Span observation" + ) + if invariants.get("orphan_generation_not_under_coincidentally_indexed_sibling"): + assert len(set(gen_parents)) == len(generations), ( + "each orphan Generation MUST parent under a DISTINCT inner instance observation " + "(chain-routed, not the coincidentally-indexed sibling)" + ) + if invariants.get("orphan_generation_not_under_fan_out_node_or_trace"): + assert all(p not in fan_out_node_ids for p in gen_parents), ( + "orphan Generation MUST NOT parent under a fan-out NODE observation" + ) + assert all(p is not None for p in gen_parents), ( + "orphan Generation MUST NOT parent under the Trace root" + ) + if invariants.get("orphan_generation_sibling_of_calling_node_observation"): + assert all(p in guard_parent_ids for p in gen_parents), ( + "each orphan Generation MUST be a SIBLING of its `guard` Span observation (share its parent)" + ) + assert all(p not in guard_ids for p in gen_parents), ( + "orphan Generation parent MUST NOT be the `guard` observation itself" + ) + if (want := invariants.get("generation_count")) is not None: + assert len(generations) == want, f"expected {want} Generations; got {len(generations)}" + if invariants.get("langfuse_parent_matches_otel_parent"): + # Real cross-backend comparison (§8.9): both observers ran on the SAME + # invocation above; each orphan Generation's resolved parent MUST match + # the OTel orphan llm.complete span's resolved parent by (name, + # fan_out_index, parent_node_name). Under the harness's POST realization + # these agree (both the inner fan-out instance); under the fixture's + # declared PRE phase they would NOT -- Langfuse resolves the lazily- + # created inner instance to the outer one (task #100, flagged to spec). + lf_desc = _parent_parity_descriptors_langfuse(trace) + otel_desc = _parent_parity_descriptors_otel(otel_spans) + assert lf_desc == otel_desc, ( + f"Langfuse and OTel resolved DIFFERENT parents: langfuse={lf_desc} otel={otel_desc}" + ) + + _handled = { + "orphan_generation_parents_under_inner_fan_out_instance_observation", + "orphan_generation_not_under_coincidentally_indexed_sibling", + "orphan_generation_not_under_fan_out_node_or_trace", + "orphan_generation_sibling_of_calling_node_observation", + "generation_count", + "langfuse_parent_matches_otel_parent", + } + unhandled = set(invariants) - _handled + assert not unhandled, f"fixture 134 case 2 declares unhandled invariants: {sorted(unhandled)}" + + +def _single_trace(client: InMemoryLangfuseClient) -> LangfuseTrace: + assert len(client.traces) == 1, f"expected exactly one Trace, got {len(client.traces)}" + return next(iter(client.traces.values())) + + +def _assert_langfuse_134_trace(trace: LangfuseTrace, expected: dict[str, Any]) -> None: + # 134 asserts trace headline fields + the observation tree, matching children + # ORDER-INDEPENDENTLY (mirroring the OTel ``_assert_span_tree_matches`` set + # match). The shared ``_assert_observation_tree`` zips children in emission + # order; here two siblings under the inner instance (the orphan Generation and + # the `guard` Span) are emitted `guard`-first (the orphan fires in the wrapper + # post-phase), the reverse of the YAML order, and concurrent outer instances + # can drain in either order -- neither is semantically load-bearing. + _assert_string_or_placeholder("trace.id", trace.id, expected.get("id")) + if "name" in expected: + _assert_string_or_placeholder("trace.name", trace.name, expected.get("name")) + root_observations = trace.children_of(None) + _assert_obs_tree_unordered( + trace, root_observations, cast("list[dict[str, Any]]", expected["observations"]) + ) + + +def _obs_shallow_matches(actual: LangfuseObservation, expected: dict[str, Any]) -> bool: + # Match an observation by type + name (+ placeholder) + metadata SUBSET (extra + # actual keys tolerated), the disambiguators the 134 tree needs. + if "type" in expected and actual.type != expected["type"]: + return False + if "name" in expected: + name = expected["name"] + if _is_placeholder(name): + if not (isinstance(actual.name, str) and actual.name): + return False + elif actual.name != name: + return False + for key, value in cast("dict[str, Any]", expected.get("metadata") or {}).items(): + if actual.metadata.get(key) != value: + return False + return True + + +def _assert_obs_tree_unordered( + trace: LangfuseTrace, + actual_children: list[LangfuseObservation], + expected_children: list[dict[str, Any]], +) -> None: + assert len(actual_children) == len(expected_children), ( + f"observation children count mismatch: expected {len(expected_children)}, " + f"got {len(actual_children)} ({[o.name for o in actual_children]})" + ) + remaining = list(actual_children) + for expected in expected_children: + matches = [a for a in remaining if _obs_shallow_matches(a, expected)] + assert len(matches) >= 1, ( + f"no observation matched expected type={expected.get('type')!r} " + f"name={expected.get('name')!r} metadata={expected.get('metadata')!r}; " + f"candidates: {[(o.type, o.name) for o in remaining]}" + ) + assert len(matches) == 1, ( + f"ambiguous match for expected type={expected.get('type')!r} " + f"name={expected.get('name')!r} metadata={expected.get('metadata')!r}: " + f"{len(matches)} candidates. Add a disambiguating metadata key." + ) + matched = matches[0] + remaining.remove(matched) + _assert_obs_tree_unordered( + trace, + trace.children_of(matched.id), + cast("list[dict[str, Any]]", expected.get("children") or []), + ) + + def _resolve_llm_model(case: Mapping[str, Any]) -> str: # Single LLM call per fixture today; pick up the per-call model if # supplied (fixture 023 explicitly sets `model: "test-model"` on diff --git a/tests/unit/test_observability_langfuse.py b/tests/unit/test_observability_langfuse.py index f8a5235..dc47e2b 100644 --- a/tests/unit/test_observability_langfuse.py +++ b/tests/unit/test_observability_langfuse.py @@ -1778,17 +1778,19 @@ async def test_typed_failed_event_parents_under_branch_calling_node() -> None: try: # Bootstrap the Trace + two branch-distinguished node # observations directly. _InvState's open_observations map is - # keyed by (namespace, attempt_index, fan_out_index, - # branch_name); the calling node identity on the typed event - # is (("dispatcher", "ask"), 0, None, "fast"). + # keyed by (namespace, attempt_index, fan_out_index, branch_name) + # plus the proposal-0084 lineage chains; the calling node identity + # on the (non-nested) typed event has empty chains, so the keys are + # (("dispatcher", "ask"), 0, None, "fast"/"slow", (), ()). key[3] + # (branch_name) is the discriminator this test exercises. client.trace(id=invocation_id, name="dispatcher") observer._inv_states[invocation_id] = _InvState(trace_id=invocation_id) # noqa: SLF001 inv_state = observer._inv_states[invocation_id] # noqa: SLF001 # Open two observations under the trace — one per branch. fast_handle = client.generation(trace_id=invocation_id, name="ask", model="m-test") slow_handle = client.generation(trace_id=invocation_id, name="ask", model="m-test") - fast_key = (("dispatcher", "ask"), 0, None, "fast") - slow_key = (("dispatcher", "ask"), 0, None, "slow") + fast_key = (("dispatcher", "ask"), 0, None, "fast", (), ()) + slow_key = (("dispatcher", "ask"), 0, None, "slow", (), ()) inv_state.open_observations[fast_key] = _OpenObservation(handle=fast_handle) inv_state.open_observations[slow_key] = _OpenObservation(handle=slow_handle) await observer( @@ -1820,13 +1822,14 @@ async def test_typed_failed_event_parents_under_branch_calling_node() -> None: async def test_llm_event_parents_under_fan_out_instance_dispatch() -> None: - # Regression cover for _resolve_llm_parent_observation_id fallback #2: when an - # LLM event fires inside a top-level fan-out instance and the calling node has - # no open observation (fallback #1 misses), the Generation MUST parent under - # the per-instance fan-out dispatch observation. The dispatch map is keyed by - # the lineage-aware _dispatch_key; before the lineage keys this fallback used - # a flat namespace[:1] + (str(index),) key, which always-misses against the - # composite map and silently re-parents the Generation under the Trace. + # Regression cover for the _resolve_llm_parent_observation_id orphan fallback: + # when an LLM event fires inside a top-level fan-out instance and the calling + # node has no open observation (fallback #1 misses), the Generation MUST + # parent under the per-instance fan-out dispatch observation (the nearest + # enclosing wrapper). Proposal 0084: the fallback resolves the dispatch via + # the event's lineage chain, so the event carries fan_out_index_chain= + # (0, None) aligned to namespace ("fan", "ask") -- instance 0 at "fan", none + # at "ask". from openarmature.observability.correlation import ( _reset_invocation_id, _set_invocation_id, @@ -1860,6 +1863,8 @@ async def test_llm_event_parents_under_fan_out_instance_dispatch() -> None: attempt_index=0, fan_out_index=0, branch_name=None, + fan_out_index_chain=(0, None), + branch_name_chain=(None, None), model="m-test", error_category="provider_unavailable", error_type="ProviderUnavailable", @@ -1878,6 +1883,71 @@ async def test_llm_event_parents_under_fan_out_instance_dispatch() -> None: ) +async def test_llm_event_parents_under_parallel_branch_dispatch() -> None: + # Proposal 0084 branch coverage: the orphan fallback resolves a per-branch + # dispatch observation as well as a fan-out instance observation, but the + # 0084 fixtures (132/133/134) are fan-out only. An orphan LLM event fired + # inside a parallel branch (calling node observation not open) MUST parent + # under the per-branch dispatch observation -- the nearest enclosing wrapper + # -- resolved via branch_name_chain. Guards the reference behavior pending a + # spec fixture (see release-v0.17.0 coord). + from openarmature.observability.correlation import ( + _reset_invocation_id, + _set_invocation_id, + ) + from openarmature.observability.langfuse.observer import ( + _branch_dispatch_key, + _InvState, + _OpenObservation, + ) + from tests._helpers.typed_event import make_failed_event + + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) + invocation_id = "inv-pb-llm" + token = _set_invocation_id(invocation_id) + try: + client.trace(id=invocation_id, name="dispatcher") + observer._inv_states[invocation_id] = _InvState(trace_id=invocation_id) # noqa: SLF001 + inv_state = observer._inv_states[invocation_id] # noqa: SLF001 + # Per-branch dispatch for branch "fast" of pb node "dispatcher". No + # open_observation for the calling node ("dispatcher", "ask"), so the + # resolver reaches the per-branch dispatch fallback. The calling node + # sits in branch "fast": branch_name_chain=(None, "fast") aligned to + # namespace. + fi_chain: tuple[int | None, ...] = (None, None) + bn_chain: tuple[str | None, ...] = (None, "fast") + dispatch_handle = client.span(trace_id=invocation_id, name="fast") + branch_key = _branch_dispatch_key(("dispatcher",), fi_chain, bn_chain, "fast") + inv_state.parallel_branches_branch_spans[branch_key] = _OpenObservation(handle=dispatch_handle) + await observer( + make_failed_event( + invocation_id=invocation_id, + node_name="ask", + namespace=("dispatcher", "ask"), + attempt_index=0, + fan_out_index=None, + branch_name="fast", + fan_out_index_chain=fi_chain, + branch_name_chain=bn_chain, + model="m-test", + error_category="provider_unavailable", + error_type="ProviderUnavailable", + error_message="503 from upstream", + call_id="cc-pb", + ) + ) + finally: + _reset_invocation_id(token) + + trace = client.traces[invocation_id] + error_gens = [o for o in trace.observations if o.type == "generation" and o.level == "ERROR"] + assert len(error_gens) == 1 + assert error_gens[0].parent_observation_id == dispatch_handle.id, ( + "LLM Generation must parent under the per-branch dispatch observation" + ) + + # --------------------------------------------------------------------------- # Proposal 0063 — tool-execution Tool observation (asType "tool") # --------------------------------------------------------------------------- diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index 6bb46b3..3b00cf0 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -560,11 +560,13 @@ async def test_active_prompt_propagates_to_llm_span_attributes() -> None: async def test_llm_span_parents_under_fan_out_instance_dispatch() -> None: - # Gap 2 (review-nested-fan-out-lineage): an LLM span whose calling node has - # no open span and fires inside a top-level fan-out instance MUST parent - # under the per-instance fan-out dispatch span (matching the Langfuse - # observer), not fall through to the subgraph / invocation span. Before this - # OTel had no fan-out-instance fallback in _resolve_llm_parent. + # An LLM span whose calling node has no open span and fires inside a + # top-level fan-out instance MUST parent under the per-instance fan-out + # dispatch span (the §5.5 orphan fallback / nearest enclosing wrapper), + # not fall through to the subgraph / invocation span. Proposal 0084: the + # fallback resolves the dispatch via the event's lineage chain, so the + # event carries fan_out_index_chain=(0, None) aligned to namespace + # ("fan", "ask") -- instance 0 at the "fan" boundary, none at "ask". from openarmature.observability.correlation import ( _reset_invocation_id, _set_invocation_id, @@ -599,6 +601,8 @@ async def test_llm_span_parents_under_fan_out_instance_dispatch() -> None: attempt_index=0, fan_out_index=0, branch_name=None, + fan_out_index_chain=(0, None), + branch_name_chain=(None, None), ) ) # dispatch_span is ended by observer.shutdown() below (it drains @@ -615,6 +619,68 @@ async def test_llm_span_parents_under_fan_out_instance_dispatch() -> None: ) +async def test_llm_span_parents_under_parallel_branch_dispatch() -> None: + # Proposal 0084 branch coverage: the §5.5 orphan fallback is defined for a + # per-branch dispatch span as well as a fan-out instance span, but the 0084 + # fixtures (132/133/134) are fan-out only. An orphan LLM call (no open + # calling-node span) fired inside a parallel branch MUST parent under the + # per-branch dispatch span -- the nearest enclosing wrapper -- resolved via + # branch_name_chain, the branch-side analog of the fan-out case above. Guards + # the reference behavior pending a spec fixture (see release-v0.17.0 coord). + from openarmature.observability.correlation import ( + _reset_invocation_id, + _set_invocation_id, + ) + from openarmature.observability.otel.observer import ( + _branch_dispatch_key, + _InvState, + _OpenSpan, + ) + from tests._helpers.typed_event import make_retry_attempt_event + + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + invocation_id = "inv-pb-llm" + token = _set_invocation_id(invocation_id) + try: + # Per-branch dispatch span for branch "fast" of pb node "dispatcher". No + # open_spans entry for the calling node ("dispatcher", "ask"), so the + # resolver must reach the per-branch dispatch fallback. The calling node + # sits in branch "fast": branch_name_chain=(None, "fast") aligned to + # namespace (None at the "dispatcher" pb boundary, "fast" at the "ask" + # branch descent). + fi_chain: tuple[int | None, ...] = (None, None) + bn_chain: tuple[str | None, ...] = (None, "fast") + observer._inv_states[invocation_id] = _InvState() # noqa: SLF001 + dispatch_span = observer._tracer.start_span("fast") # noqa: SLF001 + branch_key = _branch_dispatch_key(("dispatcher",), fi_chain, bn_chain, "fast") + observer._inv_states[invocation_id].parallel_branches_branch_spans[branch_key] = _OpenSpan( # noqa: SLF001 + span=dispatch_span + ) + await observer( + make_retry_attempt_event( + invocation_id=invocation_id, + node_name="ask", + namespace=("dispatcher", "ask"), + attempt_index=0, + fan_out_index=None, + branch_name="fast", + fan_out_index_chain=fi_chain, + branch_name_chain=bn_chain, + ) + ) + finally: + _reset_invocation_id(token) + + observer.shutdown() + llm_spans = [s for s in exporter.get_finished_spans() if s.name == "openarmature.llm.complete"] + assert len(llm_spans) == 1 + assert llm_spans[0].parent is not None, "LLM span must have a parent, not be a trace root" + assert cast("Any", llm_spans[0].parent).span_id == dispatch_span.get_span_context().span_id, ( + "orphan LLM span must parent under the per-branch dispatch span" + ) + + async def test_llm_span_has_no_prompt_attributes_when_no_active_prompt() -> None: """Without ``with_active_prompt``, the LLM-call span MUST NOT carry ``openarmature.prompt.*`` attributes."""