Skip to content

Resolve orphan provider span parents structurally, wire 152/153 #277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chris-colinsky wants to merge 2 commits into
base: main
Choose a base branch
from
+583 −27
Open

Resolve orphan provider span parents structurally, wire 152/153 #277

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ec6ba6c
Resolve an orphan provider span's parent structurally
chris-colinsky Aug 21, 2026
ea9476b
Wire orphan-fallback fixtures 152 and 153
chris-colinsky Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 95 additions & 3 deletions src/openarmature/observability/otel/observer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,14 +298,32 @@ def _subgraph_identity_at(event: NodeEvent, depth: int) -> str:
"""
# Spec observability §5.3 (coord thread
# clarify-subgraph-name-semantics).
# `getattr`, because a dispatch span may now be synthesized from a provider
# or tool event, which carries the lineage fields but not the identities.
# The empty-string fallback below is that case; `_backfill_subgraph_identity`
# fills it in from the first node event, so the attribute does not depend on
# which event happened to trigger synthesis.
identities = cast("tuple[str | None, ...]", getattr(event, "subgraph_identities", ()))
idx = depth - 1
if 0 <= idx < len(event.subgraph_identities):
identity = event.subgraph_identities[idx]
if 0 <= idx < len(identities):
identity = identities[idx]
if identity is not None:
return identity
return ""


def _backfill_subgraph_identity(open_span: Any, event: NodeEvent, depth: int) -> None:
"""Set ``openarmature.subgraph.name`` on an already-open dispatch span."""
# Dispatch spans can now be synthesized by a provider or tool event, which
# carries the lineage chains but no `subgraph_identities`, so the attribute
# would otherwise read empty whenever a wrapper-issued call happened to
# arrive first. That is the same class of schedule-dependence the on-demand
# synthesis exists to remove, one attribute down.
identity = _subgraph_identity_at(event, depth)
if identity:
open_span.span.set_attribute("openarmature.subgraph.name", identity)


def _empty_str_frozenset() -> frozenset[str]:
"""Typed empty frozenset factory for ``detached_subgraphs`` /
``detached_fan_outs`` defaults."""
Expand Down Expand Up @@ -1630,6 +1648,7 @@ def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
parent_ctx = self._resolve_llm_parent(
inv_state,
invocation_id,
event=event,
calling_namespace_prefix=event.namespace,
calling_attempt_index=event.attempt_index,
calling_fan_out_index=event.fan_out_index,
Expand Down Expand Up @@ -1857,6 +1876,7 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None:
parent_ctx = self._resolve_llm_parent(
inv_state,
invocation_id,
event=event,
calling_namespace_prefix=event.namespace,
calling_attempt_index=event.attempt_index,
calling_fan_out_index=event.fan_out_index,
Expand Down Expand Up @@ -2006,6 +2026,7 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non
parent_ctx = self._resolve_llm_parent(
inv_state,
invocation_id,
event=event,
calling_namespace_prefix=event.namespace,
calling_attempt_index=event.attempt_index,
calling_fan_out_index=event.fan_out_index,
Expand Down Expand Up @@ -2118,6 +2139,7 @@ def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None:
parent_ctx = self._resolve_llm_parent(
inv_state,
invocation_id,
event=event,
calling_namespace_prefix=event.namespace,
calling_attempt_index=event.attempt_index,
calling_fan_out_index=event.fan_out_index,
Expand Down Expand Up @@ -2231,6 +2253,7 @@ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None:
parent_ctx = self._resolve_llm_parent(
inv_state,
invocation_id,
event=event,
calling_namespace_prefix=event.namespace,
calling_attempt_index=event.attempt_index,
calling_fan_out_index=event.fan_out_index,
Expand Down Expand Up @@ -2268,6 +2291,7 @@ def _resolve_llm_parent(
inv_state: _InvState,
invocation_id: str,
*,
event: Any = None,
calling_namespace_prefix: tuple[str, ...],
calling_attempt_index: int,
calling_fan_out_index: int | None,
Expand Down Expand Up @@ -2306,6 +2330,14 @@ def _resolve_llm_parent(
# 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.
# ORPHAN PATH ONLY: the calling node's span is not open, so this is a
# call from a wrapper. Materialize any dispatch span the calling lineage
# sits inside before walking for it, so the answer does not depend on
# whether the wrapper's first inner node has drained yet.
if event is not None:
self._synthesize_call_site_wrapper_spans(
inv_state, cast("str | None", getattr(event, "correlation_id", None)), event
)
return self._resolve_enclosing_wrapper_context(
inv_state,
invocation_id,
Expand Down Expand Up @@ -2589,10 +2621,17 @@ def _sync_subgraph_spans(
branch_key = _branch_dispatch_key(
prefix, event.fan_out_index_chain, event.branch_name_chain, event.branch_name
)
if branch_key not in inv_state.parallel_branches_branch_spans:
open_branch = inv_state.parallel_branches_branch_spans.get(branch_key)
if open_branch is None:
self._open_parallel_branches_branch_dispatch_span(
inv_state, correlation_id, prefix, event
)
else:
# Already synthesized, possibly from a provider event, which
# carries the lineage but not the subgraph identities. Fill
# the identity in from this node event so the attribute does
# not depend on which event triggered synthesis.
_backfill_subgraph_identity(open_branch, event, len(prefix))
Comment on lines +2630 to +2634
continue
# If ``prefix`` names a parallel-branches or fan-out NODE
# (detected by an entry in the respective parent_node_name
Expand Down Expand Up @@ -2880,6 +2919,59 @@ def _open_detached_fan_out_instance_root(
inv_state.detached_roots[instance_key] = _OpenSpan(span=instance_root)
inv_state.fan_out_instance_root_prefixes.add(instance_key)

def _synthesize_call_site_wrapper_spans(
self,
inv_state: _InvState,
correlation_id: str | None,
event: Any,
) -> None:
"""Open any dispatch span the CALLING lineage sits inside that has not
been synthesized yet."""
# Spec observability §5.5 (Lineage-resolved parent), as ruled in the
# release-v0.17.0 coord thread: the parent is resolved STRUCTURALLY. A
# call issued from branch or instance middleware is inside that branch or
# instance, so its nearest enclosing wrapper is that dispatch span,
# whether or not the observer has materialized it yet.
#
# Without this, the parent depended on drain scheduling. Dispatch spans
# are synthesized from inner NODE events, and a wrapper-issued provider
# call is enqueued BEFORE the wrapper's first inner node starts. Whether
# the span existed at resolution time came down to whether anything
# yielded to the event loop in between: one `await asyncio.sleep(0)` in
# user middleware moved the span from its branch to the invocation root.
# §10 covers parentage, so that was non-conforming rather than untidy.
#
# Two differences from `_sync_subgraph_spans`, which does this for node
# events. It walks PROPER ancestors (`range(1, len(namespace))`), but a
# call from branch middleware sits AT the parallel-branches namespace, so
# its branch's prefix IS the full namespace and that walk never reaches
# it. And this opens dispatch spans only: no subgraph wrappers, no
# detached roots, and nothing is closed, since a provider event is not a
# position change.
namespace = cast("tuple[str, ...]", event.namespace)
fan_out_index_chain = cast("tuple[int | None, ...]", event.fan_out_index_chain)
branch_name_chain = cast("tuple[str | None, ...]", event.branch_name_chain)
branch_name = cast("str | None", event.branch_name)
Comment on lines +2951 to +2954
for depth in range(1, len(namespace) + 1):
prefix = namespace[:depth]
fi_axis = fan_out_index_chain[depth - 1] if depth - 1 < len(fan_out_index_chain) else None
if (
fi_axis is not None
and prefix[-1] not in self.detached_fan_outs
and prefix in inv_state.fan_out_parent_node_name
and _dispatch_key(prefix, fan_out_index_chain, branch_name_chain)
not in inv_state.fan_out_instance_spans
):
self._open_fan_out_instance_dispatch_span(inv_state, correlation_id, prefix, event)
if (
branch_name is not None
and prefix in inv_state.parallel_branches_parent_node_name
and branch_name in inv_state.parallel_branches_branch_names.get(prefix, frozenset())
and _branch_dispatch_key(prefix, fan_out_index_chain, branch_name_chain, branch_name)
not in inv_state.parallel_branches_branch_spans
):
self._open_parallel_branches_branch_dispatch_span(inv_state, correlation_id, prefix, event)

def _open_fan_out_instance_dispatch_span(
self,
inv_state: _InvState,
Expand Down
Loading