From ec6ba6c386ecbf0db1c96643aae2449e2b88b43b Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 21 Aug 2026 00:58:08 -0700 Subject: [PATCH 1/2] Resolve an orphan provider span's parent structurally A call issued from branch or fan-out instance middleware resolved its parent against whatever dispatch spans happened to be registered when the observer drained the event. Dispatch spans are synthesized from inner NODE events, and a wrapper-issued call is enqueued before the wrapper's first inner node starts, so the answer 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. Observability section 10 covers parentage, so that was non-conforming rather than untidy. The parent is now resolved structurally: a call issued from a wrapper is inside that wrapper, so its nearest enclosing wrapper is that dispatch span whether or not the observer has materialized it. Any dispatch span the calling lineage sits inside is synthesized on demand, triggered by the first event that needs it rather than only by an inner node event. Ruled in the release coord thread; the section 6 sentence pinning the span's start time to the inner started event changes with it, and that wording is spec's to land. Two details. The synthesis walks the full calling namespace rather than its proper ancestors, because a call from branch middleware sits AT the parallel-branches namespace, so its branch's prefix is that namespace and the existing ancestor walk never reaches it. And a provider event carries the lineage chains but no subgraph identities, so a span it synthesizes would read an empty subgraph name where one synthesized from a node event reads the real value; the identity is backfilled from the first node event so that attribute cannot depend on which event arrived first either. Applies to both dispatch span kinds. They share one synthesis path with one trigger, so neither has an eager arm the other lacks. --- .../observability/otel/observer.py | 98 ++++++++++++++++++- 1 file changed, 95 insertions(+), 3 deletions(-) diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index dc456cf..34721f5 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -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.""" @@ -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, @@ -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, @@ -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, @@ -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, @@ -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, @@ -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, @@ -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, @@ -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)) continue # If ``prefix`` names a parallel-branches or fan-out NODE # (detected by an entry in the respective parent_node_name @@ -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) + 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, From ea9476b7c400fdc82ce4bc5e20c31c491e52535a Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 21 Aug 2026 01:08:44 -0700 Subject: [PATCH 2/2] Wire orphan-fallback fixtures 152 and 153 Both assert where a provider span lands when it is issued from a wrapper rather than the node body, so the calling node's span is not open and the span falls back to the nearest enclosing wrapper. 152's is the per-branch dispatch span; 153 nests a fan-out between them so its is the innermost instance span. THE WRAPPER YIELDS ON PURPOSE. An earlier version of this driver made both fixtures pass, and that passing was worthless: it depended on nothing yielding to the event loop between the provider call and the next node start. One `await asyncio.sleep(0)` moved 152's orphan to the invocation root and 153's to the branch dispatch span its own invariant forbids. The yield is the assertion, and the observer change it exercises lands separately. The routing invariant now checks routing. It previously counted orphans per dispatch span, which a full branch swap satisfies exactly as well as correct routing, and the span_tree cannot see the difference because the two orphan spans are name- and attribute-identical in the declared tree. The mock routes each branch's distinct request content to a distinct response id, so the id on the span identifies which branch issued it, and the expected mapping is derived from the fixture rather than hardcoded. Dispatch spans are told from node spans by the attributes only a synthesized dispatch span carries. Matching on name or on branch_name misfires both ways: an instance dispatch span reuses its fan-out node's name, and every node span inside a branch carries branch_name, so the sibling and absence claims accepted the guard node span they forbid. Subgraph references are collected in one helper covering all three spellings. Missing the parallel-branches one compiled a host subgraph before its branch targets and died on a bare KeyError from the builder. --- tests/conformance/test_observability.py | 512 ++++++++++++++++++++++-- 1 file changed, 488 insertions(+), 24 deletions(-) diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index d64db98..2d7703d 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -263,6 +263,12 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # an impl sourcing the event from `raw` surfaces the wire value there -- # so it needs a driver reading both surfaces off one invocation. "149-malformed-wire-counter-nulled-through-mapping-to-event-and-span", + # 152 / 153 (proposal 0084): where a provider span emitted from a + # WRAPPER lands when the calling node's span is not open. Their driver + # yields inside the wrapper on purpose; passing without that yield was + # the schedule-dependence, not the fix. + "152-otel-parallel-branch-orphan-llm-fallback", + "153-otel-mixed-nesting-orphan-llm-fallback", # v0.69.0 — proposal 0063 (tool-execution observability). A # calls_tool node enters the with_tool_call scope; the typed # ToolCallEvent / ToolCallFailedEvent drive the OTel tool span + @@ -446,30 +452,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "threading per_node_mw into add_parallel_branches_node, NOT a failure path in the 110 " "driver -- that driver is reached and would still see an unretried raise" ), - # Both were wired and then UN-wired: a driver exists in the PR history that - # makes them pass, and the passing was not worth having. The orphan provider - # call is enqueued before the branch's first inner node, so whether the - # per-branch dispatch span is registered when the observer resolves the - # parent depends on nothing yielding to the event loop in between. Inserting - # a single `await asyncio.sleep(0)` in the wrapper -- ordinary for real - # middleware -- moves 152's orphan to the invocation root and 153's to the - # `work` branch dispatch span, the parent 153 explicitly forbids. Measured, - # not reasoned. A green run would have certified a lucky interleaving. - # - # The lineage-key defect these exposed IS fixed (`_branch_dispatch_key` now - # normalizes chains shallower than the prefix, in both observers). That fix - # is necessary and not sufficient. Activating these needs the orphan parent - # resolved deterministically -- deferring the decision until the enclosing - # wrapper span is known -- which is an observer change, not a harness one. - "152-otel-parallel-branch-orphan-llm-fallback": ( - "orphan parent resolution is drain-schedule dependent; one `await asyncio.sleep(0)` in the " - "wrapper parents the orphan under the invocation root instead of the branch dispatch span. " - "Needs deterministic resolution in the observer, not a harness change" - ), - "153-otel-mixed-nesting-orphan-llm-fallback": ( - "same race as 152: under a yielding wrapper the orphan parents under the `work` branch " - "dispatch span, which this fixture's own invariant forbids" - ), # Proposal 0109 (spec v0.104.0) token-budget failure-path parity. } @@ -695,6 +677,9 @@ def _reject_unsupported_capability_gate(fixture_id: str, spec: Mapping[str, Any] "_run_token_budget_fixture": frozenset({"span_tree", "metrics", "invariants", "observers"}), # 149's driver: reads the typed-event and span halves off one invocation. "_run_typed_event_with_span_cases": frozenset({"span_tree", "observers", "invariants"}), + # 152 / 153's driver: span shape plus the absence, count and ordering claims + # the tree cannot express. + "_run_orphan_fallback_fixture": frozenset({"span_tree", "invariants"}), # `invariants` is documentary here; `observers` is deliberately ABSENT because # _run_llm_payload_case reads span_tree ONLY. Claiming observers is what let # 149 be re-routed here with its contains_event half silently dropped. @@ -917,6 +902,11 @@ async def test_observability_fixture(fixture_path: Path) -> None: "131-token-budget-on-structured-output-failure", }: await _run_token_budget_fixture(spec) + elif fixture_id in { + "152-otel-parallel-branch-orphan-llm-fallback", + "153-otel-mixed-nesting-orphan-llm-fallback", + }: + await _run_orphan_fallback_fixture(spec) elif fixture_id == "149-malformed-wire-counter-nulled-through-mapping-to-event-and-span": # Needs the request model bound independently of the mock; see the driver. await _run_typed_event_with_span_cases(spec) @@ -6821,6 +6811,480 @@ async def _run_typed_event_with_span_case(case: Mapping[str, Any]) -> None: _assert_invariants_recognized(case, _MALFORMED_COUNTER_INVARIANTS, "malformed-counter") +# Fixture 152 / 153 invariant names. Split by whether the case's `span_tree` +# already pins the claim, which is the test for documentary here. +# +# `_assert_span_tree_matches` matches each EXPECTED child against the actual +# children and does not reject extras, so it pins "X appears under Y" but NOT +# "nothing else appears under Y", and no count or ordering. That is why the +# absence, count and ordering claims below are implemented rather than recorded +# as documentary: the tree cannot express them. +_ORPHAN_FALLBACK_INVARIANTS = { + "orphan_llm_span_parents_under_branch_dispatch_span", + "orphan_llm_span_parents_under_innermost_fan_out_instance", + "orphan_llm_span_sibling_of_guard_node_span", + "orphan_llm_span_routed_to_correct_branch_by_name", + "orphan_llm_span_not_under_dispatcher_node_or_invocation", + "orphan_llm_span_not_under_work_branch_dispatch_span", + "orphan_llm_span_not_under_node_or_invocation", + "dispatch_spans_close_in_declaration_order", + "llm_provider_span_count", +} + +_LLM_SPAN = "openarmature.llm.complete" +_BRANCH_NAME_ATTR = "openarmature.node.branch_name" +# Attributes only a SYNTHESIZED dispatch span carries. Node spans inside a +# branch also carry `openarmature.node.branch_name`, and a fan-out instance +# dispatch span reuses its fan-out node's NAME, so neither name nor branch_name +# distinguishes a dispatch span from a node span. These do. +_BRANCH_DISPATCH_ATTR = "openarmature.parallel_branches.parent_node_name" +_INSTANCE_DISPATCH_ATTR = "openarmature.fan_out.parent_node_name" + + +def _is_branch_dispatch(span: Any) -> bool: + return _BRANCH_DISPATCH_ATTR in dict(span.attributes or {}) + + +def _is_instance_dispatch(span: Any) -> bool: + return _INSTANCE_DISPATCH_ATTR in dict(span.attributes or {}) + + +def _assert_orphan_fallback_invariants(case: Mapping[str, Any], spans: Sequence[Any]) -> None: + """Evaluate 152 / 153's orphan-parenting claims over the captured spans.""" + invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {}) + if not invariants: + return + + by_id = {s.context.span_id: s for s in spans} + + def _parent_of(span: Any) -> Any: + parent = span.parent + return by_id.get(parent.span_id) if parent is not None else None + + llm_spans = [s for s in spans if s.name == _LLM_SPAN] + # Positive anchor before any absence or parent claim: with no provider span + # recorded, every "the orphan is not under X" claim holds trivially and + # every "its parent is Y" claim is vacuous over an empty list. + assert llm_spans, ( + f"no {_LLM_SPAN} span was recorded, so the orphan-parenting claims would pass " + f"vacuously; got {sorted({s.name for s in spans})}" + ) + + expected_count = invariants.get("llm_provider_span_count") + if expected_count is not None: + assert len(llm_spans) == expected_count, ( + f"expected exactly {expected_count} {_LLM_SPAN} span(s); got {len(llm_spans)}" + ) + + if invariants.get("orphan_llm_span_parents_under_branch_dispatch_span"): + for llm in llm_spans: + parent = _parent_of(llm) + assert parent is not None and _is_branch_dispatch(parent), ( + f"the orphan {_LLM_SPAN} MUST parent under a per-branch DISPATCH span (one carrying " + f"{_BRANCH_DISPATCH_ATTR}); its parent is {parent.name if parent else None!r} with " + f"{sorted(dict(parent.attributes or {})) if parent else []}" + ) + + if invariants.get("orphan_llm_span_parents_under_innermost_fan_out_instance"): + for llm in llm_spans: + parent = _parent_of(llm) + assert parent is not None and _is_instance_dispatch(parent), ( + f"the orphan {_LLM_SPAN} MUST parent under the innermost fan-out INSTANCE dispatch " + f"span (one carrying {_INSTANCE_DISPATCH_ATTR}); its parent is " + f"{parent.name if parent else None!r} with " + f"{sorted(dict(parent.attributes or {})) if parent else []}" + ) + + if invariants.get("orphan_llm_span_sibling_of_guard_node_span"): + # The guard node span opens only AFTER the pre-phase wrapper call, so it + # cannot be the orphan's parent; it must be its SIBLING. + guards = [s for s in spans if s.name == "guard"] + assert guards, f"no `guard` node span was recorded; got {sorted({s.name for s in spans})}" + guard_parents = {g.parent.span_id for g in guards if g.parent is not None} + for llm in llm_spans: + parent = _parent_of(llm) + assert parent is not None and parent.context.span_id in guard_parents, ( + f"the orphan {_LLM_SPAN} MUST be a SIBLING of the guard node span (same parent), " + f"not a child of it; orphan parent={parent.name if parent else None!r}" + ) + + if invariants.get("orphan_llm_span_routed_to_correct_branch_by_name"): + # Each branch's OWN orphan lands under that branch. Counting one per + # branch does not express this: a full swap satisfies it exactly as well + # as correct routing, and the span_tree cannot see it either because the + # two orphan spans are name- and attribute-identical in the declared + # tree. The mock routes each branch's distinct request content to a + # distinct response id, so the id on the span identifies its issuer. + routed = { + dict(_parent_of(llm).attributes or {}).get(_BRANCH_NAME_ATTR): dict(llm.attributes or {}).get( + "gen_ai.response.id" + ) + for llm in llm_spans + if _parent_of(llm) is not None + } + expected_routing = _expected_branch_routing(case) + assert routed == expected_routing, ( + f"each branch's own orphan MUST land under that branch; expected {expected_routing}, got {routed}" + ) + + for absence_key, forbidden in ( + ("orphan_llm_span_not_under_dispatcher_node_or_invocation", None), + ("orphan_llm_span_not_under_node_or_invocation", None), + ("orphan_llm_span_not_under_work_branch_dispatch_span", "work"), + ): + if not invariants.get(absence_key): + continue + for llm in llm_spans: + parent = _parent_of(llm) + assert parent is not None, f"the orphan {_LLM_SPAN} has no parent span at all" + attrs = dict(parent.attributes or {}) + if forbidden is not None: + # The `work` BRANCH dispatch span specifically: a fan-out sits + # between it and the orphan, so parenting there means the + # fallback stopped one level too high. + assert not (_is_branch_dispatch(parent) and attrs.get(_BRANCH_NAME_ATTR) == forbidden), ( + f"the orphan {_LLM_SPAN} MUST NOT parent under the {forbidden!r} branch " + f"dispatch span; a fan-out instance sits between them" + ) + else: + assert parent.name != "openarmature.invocation", ( + f"the orphan {_LLM_SPAN} MUST NOT parent under the invocation root" + ) + # A NODE span is one that is NOT a synthesized dispatch span. + # Matching on name alone would misfire in both directions: an + # instance dispatch span reuses its fan-out node's name, and node + # names inside subgraphs are absent from `case["nodes"]`. + assert _is_branch_dispatch(parent) or _is_instance_dispatch(parent), ( + f"the orphan {_LLM_SPAN} MUST NOT parent under a NODE span; its parent is " + f"{parent.name!r} carrying {sorted(attrs)}" + ) + + order = cast("list[str] | None", invariants.get("dispatch_spans_close_in_declaration_order")) + if order is not None: + # Not expressible in `span_tree` at all: it is about END TIMES, not shape. + # Compared pairwise rather than by sorting, matching the existing + # implementation of this invariant name elsewhere in this file; sorting + # turns a tie into a stable-sort coin flip that reads as a pass. + dispatch = { + cast("str", dict(s.attributes or {}).get(_BRANCH_NAME_ATTR)): s + for s in spans + if _is_branch_dispatch(s) + } + missing = [name for name in order if name not in dispatch] + assert not missing, ( + f"expected one branch dispatch span per declared branch {order}; missing {missing}, " + f"got {sorted(dispatch)}" + ) + for earlier, later in zip(order, order[1:], strict=False): + assert dispatch[earlier].end_time <= dispatch[later].end_time, ( + f"branch dispatch spans MUST close in declaration order {order}; {earlier!r} closed " + f"at {dispatch[earlier].end_time} after {later!r} at {dispatch[later].end_time}" + ) + + +# --------------------------------------------------------------------------- +# Fixtures 152 / 153 — orphan LLM span parent resolution (proposal 0084, §5.5) +# --------------------------------------------------------------------------- +# +# Both assert where a provider span lands when it is issued from a WRAPPER +# rather than the node body, so the calling node's span is not open and the span +# falls back to the nearest enclosing wrapper. 152's is the per-branch DISPATCH +# span; 153 nests a fan-out between them so its is the innermost INSTANCE span. +# +# THE WRAPPER YIELDS ON PURPOSE. Both fixtures passed before the observer +# resolved this structurally, but only because nothing yielded to the event loop +# between the provider call and the next node start: dispatch spans are +# synthesized from inner node events, so a single `await asyncio.sleep(0)` moved +# 152's orphan to the invocation root and 153's to the `work` branch dispatch +# span. Passing without a yield was the defect wearing a green run. The yield is +# the assertion. + + +def _merged_subgraph_specs(case: Mapping[str, Any], spec: Mapping[str, Any]) -> dict[str, Any]: + """Subgraph declarations, from either level the corpus uses.""" + # conformance-adapter §5.4 documents `subgraphs:` as a FIXTURE TOP-LEVEL + # block and 153 puts it there, while 152 uses the case-level form that + # eighteen fixtures across four capabilities use and none of which also + # carries a top-level block. Spec has confirmed the case-level form is + # sanctioned and that §5.4 is what needs correcting, so accept both. + merged: dict[str, Any] = {} + for source in (spec, case): + merged.update(cast("dict[str, Any]", source.get("subgraphs") or {})) + return merged + + +def _subgraph_refs(node_spec: Mapping[str, Any]) -> set[str]: + """Every subgraph a node spec references, by any of the three spellings.""" + # Kept in one place so fan-out, plain-subgraph and parallel-branches + # references cannot drift apart. Missing the parallel-branches spelling made + # a host subgraph compile before its branch targets and die on a bare + # KeyError from the builder, which reads as a missing declaration. + refs: set[str] = set() + fan_out = cast("dict[str, Any]", node_spec.get("fan_out") or {}) + if isinstance(fan_out.get("subgraph"), str): + refs.add(cast("str", fan_out["subgraph"])) + if isinstance(node_spec.get("subgraph"), str): + refs.add(cast("str", node_spec["subgraph"])) + pb = cast("dict[str, Any]", node_spec.get("parallel_branches") or {}) + branches = cast("dict[str, Any]", pb.get("branches") or {}) + for branch_cfg in branches.values(): + if isinstance(cast("dict[str, Any]", branch_cfg).get("subgraph"), str): + refs.add(cast("str", cast("dict[str, Any]", branch_cfg)["subgraph"])) + return refs + + +def _wrapper_bearing_node(subgraph_spec: Mapping[str, Any]) -> tuple[str, dict[str, Any]] | None: + for node_name, node_spec in cast("dict[str, Any]", subgraph_spec.get("nodes") or {}).items(): + wrapper = cast("dict[str, Any] | None", node_spec.get("calls_llm_from_wrapper")) + if wrapper is not None: + return node_name, wrapper + return None + + +def _expected_branch_routing(case: Mapping[str, Any]) -> dict[str, str]: + """Map each branch name to the response id its own orphan call must return.""" + # Derived from the fixture, not hardcoded: the branch's subgraph declares the + # wrapper's request content, and the mock response whose echoed content + # matches is the one that branch must receive. + subgraphs = cast("dict[str, Any]", case.get("subgraphs") or {}) + branches: dict[str, Any] = {} + for node_spec in cast("dict[str, Any]", case["nodes"]).values(): + pb = cast("dict[str, Any] | None", cast("dict[str, Any]", node_spec).get("parallel_branches")) + if pb is not None: + branches = cast("dict[str, Any]", pb.get("branches") or {}) + break + routing: dict[str, str] = {} + for branch_name, branch_cfg in branches.items(): + sub = cast("dict[str, Any]", subgraphs.get(cast("str", branch_cfg.get("subgraph")) or "") or {}) + found = _wrapper_bearing_node(sub) + if found is None: + continue + content = _wrapper_request_content(found[1]) + for mock in cast("list[dict[str, Any]]", case.get("mock_llm") or []): + body = cast("dict[str, Any]", mock["body"]) + if _mock_matches_content(body, content): + routing[branch_name] = cast("str", body["id"]) + return routing + + +def _wrapper_request_content(wrapper_spec: Mapping[str, Any]) -> str: + messages = cast("list[dict[str, str]]", wrapper_spec.get("messages") or []) + return next((m["content"] for m in messages if m.get("role") == "user"), "guardrail check") + + +def _mock_matches_content(body: Mapping[str, Any], content: str) -> bool: + """Whether this canned response is the one for a request carrying `content`.""" + # 152's two responses differ only by id, and their `id` encodes the branch + # ("cc-152-a" for "guardrail a"). Match on that suffix so the routing is + # derived from the fixture rather than from queue order, which is + # nondeterministic under concurrent branches. + return cast("str", body.get("id", "")).endswith(content.rsplit(" ", 1)[-1]) + + +async def _run_orphan_fallback_fixture(spec: Mapping[str, Any]) -> None: + for case in cast("list[dict[str, Any]]", spec["cases"]): + case_name = cast("str", case["name"]) + try: + await _run_orphan_fallback_case(case, spec) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_orphan_fallback_case(case: Mapping[str, Any], spec: Mapping[str, Any]) -> None: + import asyncio # noqa: PLC0415 + import json # noqa: PLC0415 + + import httpx # noqa: PLC0415 + + from openarmature.llm import OpenAIProvider, UserMessage # noqa: PLC0415 + + from .adapter import build_graph # noqa: PLC0415 + + mocks = list(cast("list[dict[str, Any]]", case.get("mock_llm") or [])) + + def _handler(request: httpx.Request) -> httpx.Response: + # Routed by REQUEST CONTENT, not FIFO. Under concurrent branches the + # queue order is nondeterministic, so a FIFO mock cannot support the + # routing invariant: whichever branch called first would get the first + # id regardless of which one it was. + payload = cast("dict[str, Any]", json.loads(request.content.decode())) + sent = " ".join( + str(cast("dict[str, Any]", m).get("content", "")) + for m in cast("list[Any]", payload.get("messages") or []) + ) + for mock in mocks: + body = cast("dict[str, Any]", mock["body"]) + if any(_mock_matches_content(body, word) for word in sent.split()): + return httpx.Response( + 200, content=json.dumps(body).encode(), headers={"Content-Type": "application/json"} + ) + assert mocks, "mock_llm queue exhausted" + body = cast("dict[str, Any]", mocks[0]["body"]) + return httpx.Response( + 200, content=json.dumps(body).encode(), headers={"Content-Type": "application/json"} + ) + + provider = OpenAIProvider( + base_url="http://mock-llm.test", + model="test-model", + api_key="test", + transport=httpx.MockTransport(_handler), + ) + observer, exporter = _build_observer() + try: + + def _make_orphan_mw(wrapper_spec: Mapping[str, Any]) -> Any: + messages = (UserMessage(content=_wrapper_request_content(wrapper_spec)),) + phase = cast("str", wrapper_spec.get("phase", "pre")) + + async def _mw(state: Any, next_call: Any) -> Any: + if phase == "pre": + await provider.complete(list(messages)) + # See the note above the driver: the yield is the assertion. + await asyncio.sleep(0) + return await next_call(state) + result = await next_call(state) + await provider.complete(list(messages)) + await asyncio.sleep(0) + return result + + return _mw + + # ---- Attach each orphan wrapper to whatever ENCLOSES its subgraph, + # which is what the fixtures mean by "the nearest enclosing wrapper": a + # fan-out target takes INSTANCE middleware, a parallel branch takes + # BRANCH middleware. Not node middleware on the guard, which runs + # entirely inside the node span in both phases. + subgraph_specs = _merged_subgraph_specs(case, spec) + instance_mw: dict[str, dict[str, list[Any]]] = {} + branch_mw: dict[str, dict[str, list[Any]]] = {} + # Parallel-branches hosts live in the case AND in subgraphs, so a + # wrapper-bearing subgraph used as a branch of a nested pb node is found + # too; looking only at `case["nodes"]` tripped a misleading assertion. + host_nodes: list[tuple[str | None, str, dict[str, Any]]] = [] + for name, ns in cast("dict[str, Any]", case.get("nodes") or {}).items(): + host_nodes.append((None, name, cast("dict[str, Any]", ns))) + for sg_name, sg in subgraph_specs.items(): + sg_nodes = cast("dict[str, Any]", cast("dict[str, Any]", sg).get("nodes") or {}) + for name, ns in sg_nodes.items(): + host_nodes.append((sg_name, name, cast("dict[str, Any]", ns))) + for sg_name, sg_spec in subgraph_specs.items(): + found = _wrapper_bearing_node(cast("Mapping[str, Any]", sg_spec)) + if found is None: + continue + _node_name, wrapper_spec = found + mw = _make_orphan_mw(wrapper_spec) + attached = False + for host_sg, host_node, host_spec in host_nodes: + fan_out = cast("dict[str, Any]", host_spec.get("fan_out") or {}) + if fan_out.get("subgraph") == sg_name: + assert host_sg is not None, ( + f"fan-out node {host_node!r} targeting {sg_name!r} sits at the case level, " + f"where this driver has no instance-middleware seam" + ) + instance_mw.setdefault(host_sg, {}).setdefault(host_node, []).append(mw) + attached = True + host_pb = cast("dict[str, Any]", host_spec.get("parallel_branches") or {}) + branches = cast("dict[str, Any]", host_pb.get("branches") or {}) + for branch_name, branch_cfg in branches.items(): + if cast("dict[str, Any]", branch_cfg).get("subgraph") == sg_name: + assert host_sg is None, ( + f"parallel-branches node {host_node!r} is nested inside subgraph " + f"{host_sg!r}; this driver attaches branch middleware at the case level only" + ) + branch_mw.setdefault(host_node, {}).setdefault(branch_name, []).append(mw) + attached = True + assert attached, ( + f"subgraph {sg_name!r} declares `calls_llm_from_wrapper` but is neither a fan-out " + f"target nor a parallel branch, so it has no enclosing wrapper to orphan against" + ) + + # ---- Compile subgraphs innermost-first, so one referencing another sees + # it already compiled. + compiled: dict[str, Any] = {} + remaining = dict(subgraph_specs) + while remaining: + progressed = False + for sg_name, sg_spec in list(remaining.items()): + nodes = cast("dict[str, Any]", cast("dict[str, Any]", sg_spec).get("nodes") or {}) + needed: set[str] = set() + for node_spec in nodes.values(): + needed |= _subgraph_refs(cast("Mapping[str, Any]", node_spec)) + if not needed <= set(compiled): + continue + built_sg = build_graph( + cast("Mapping[str, Any]", sg_spec), + subgraphs=dict(compiled), + trace=[], + model_name=f"Sub_{sg_name}", + fan_out_instance_middleware=instance_mw.get(sg_name, {}), + ) + compiled[sg_name] = built_sg.builder.compile() + del remaining[sg_name] + progressed = True + assert progressed, ( + f"subgraph dependency cycle or missing declaration among {sorted(remaining)}; " + f"compiled so far: {sorted(compiled)}" + ) + + built = build_graph(case, subgraphs=compiled, trace=[], parallel_branches_branch_middleware=branch_mw) + graph = built.builder.compile() + graph.attach_observer(observer) + await graph.invoke(built.initial_state(cast("dict[str, Any]", case.get("initial_state") or {}))) + await graph.drain() + finally: + await provider.aclose() + observer.shutdown() + + spans = exporter.get_finished_spans() + expected = cast("dict[str, Any]", case["expected"]) + expected_tree = cast("list[dict[str, Any]] | None", expected.get("span_tree")) + if expected_tree is not None: + inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None) + assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}" + # NOT `_assert_error_span_extras` or `_check_payload_span_tree`: both + # assume unique span names, which 153 breaks (`inner_fan_out` names both + # the node span and its instance spans). Skipping them is only safe + # while neither fixture declares a key one of them solely reads, so + # assert that rather than leaving it to hold by luck. + _assert_no_skipped_span_tree_keys(expected_tree) + _assert_span_tree_matches(spans, [inv_root], expected_tree) + + _assert_invariants_recognized(case, _ORPHAN_FALLBACK_INVARIANTS, "orphan-fallback") + _assert_orphan_fallback_invariants(case, spans) + + +# Span-tree entry keys read ONLY by the two walkers this driver skips. +_SKIPPED_SPAN_TREE_KEYS = frozenset( + { + "attributes_absent", + "status_description", + "exception_recorded", + "input_parses_as_messages", + "attribute_truncated", + "attribute_does_not_contain", + "output_parses_as_object", + } +) + + +def _assert_no_skipped_span_tree_keys(expected_tree: Sequence[Mapping[str, Any]]) -> None: + """Fail if an entry declares a key whose only reader this driver skips.""" + + def _walk(entries: Sequence[Mapping[str, Any]], path: str) -> None: + for entry in entries: + here = f"{path}/{cast('str', entry.get('name') or '')}" + declared = sorted(_SKIPPED_SPAN_TREE_KEYS & set(entry)) + assert not declared, ( + f"{here} declares {declared}, which only a span-tree walker this driver skips would " + f"read, so the claim would run nowhere. Teach the walker name+attribute " + f"disambiguation before this fixture can carry those keys." + ) + _walk(cast("Sequence[Mapping[str, Any]]", entry.get("children") or []), here) + + _walk(expected_tree, "") + + async def _run_typed_event_chain_cases(spec: Mapping[str, Any], *, expect_failure: bool = False) -> None: """Iterate the multi-node-chain typed-event cases (067 success, 071 failure)."""