From 0a80474eb24a5a622b08310626e405d9c58f319c Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Wed, 19 Aug 2026 23:48:47 -0700 Subject: [PATCH 1/7] Normalize branch dispatch keys shallower than their prefix Per-branch dispatch spans are stored under a lineage-aware key carrying the enclosing fan-out-index and branch chains. The key builder truncated a chain longer than the namespace prefix but never padded one that was shorter, so a caller whose lineage is shallower than the prefix built a key denoting the same lineage as the registered one while being unequal to it. An orphan provider call issued from branch middleware is exactly that caller: it carries empty chains, looked up (prefix, (), (), branch) against a span registered as (prefix, (None,), (), branch), missed, and fell through the ancestor walk to the invocation root instead of parenting under the branch dispatch span. Both backends carried the defect. Each helper's docstring says it mirrors the other, but nothing enforced that, so the fix is applied and tested in both. --- src/openarmature/observability/langfuse/observer.py | 4 +++- src/openarmature/observability/otel/observer.py | 12 +++++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 8fac4e3..4edd49c 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -195,7 +195,9 @@ def _branch_dispatch_key( pb node) so a pb nested inside an outer fan-out instance doesn't collide across outer instances.""" n = len(prefix) - return (prefix, tuple(fan_out_index_chain[:n]), tuple(branch_name_chain[: n - 1]), branch_name) + fan_out = tuple(fan_out_index_chain[:n]) + (None,) * max(0, n - len(fan_out_index_chain)) + branches = tuple(branch_name_chain[: n - 1]) + (None,) * max(0, (n - 1) - len(branch_name_chain)) + return (prefix, fan_out, branches, branch_name) def _empty_str_frozenset() -> frozenset[str]: diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index ec51085..b48824d 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -393,7 +393,17 @@ def _branch_dispatch_key( across outer instances. Mirrors the LangfuseObserver helper of the same name.""" n = len(prefix) - return (prefix, tuple(fan_out_index_chain[:n]), tuple(branch_name_chain[: n - 1]), branch_name) + # Chains are normalized to the prefix DEPTH in both directions: truncated + # when longer, padded with None when shorter. Truncating alone was a defect. + # A caller whose lineage is shallower than the prefix -- an orphan provider + # call issued from branch middleware carries empty chains -- built + # `(prefix, (), (), branch)` while the span had been registered under + # `(prefix, (None,), (), branch)`. Those denote the same lineage, "no + # enclosing fan-out at that depth", and differed only as tuple keys, so the + # lookup missed and the orphan fell through to the invocation root. + fan_out = tuple(fan_out_index_chain[:n]) + (None,) * max(0, n - len(fan_out_index_chain)) + branches = tuple(branch_name_chain[: n - 1]) + (None,) * max(0, (n - 1) - len(branch_name_chain)) + return (prefix, fan_out, branches, branch_name) # Sorted object keys, no insignificant whitespace, UTF-8 output (per From d06fe51845412838bb354578c4817812d4a2c778 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Wed, 19 Aug 2026 23:51:00 -0700 Subject: [PATCH 2/7] Pin the branch dispatch key against both failure directions The padding fix has two ways to be wrong and no fixture covers either on the Langfuse side, where no activated fixture reaches that helper. Under-padding is the original defect: a shallow lineage builds a key unequal to the registered one. Over-padding is the tempting sloppy fix: replacing the chain with None entries collapses genuinely distinct enclosing lineages, so a parallel-branches node inside outer fan-out instance 0 would share a key with the same node inside instance 1. Both helpers are driven by the same tests, including one asserting they agree across a matrix of lineages. They are duplicated by design, one per backend, and each docstring claims to mirror the other while nothing checked it -- which is how the same defect came to sit in both. --- tests/unit/test_observability_otel.py | 62 +++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index b8c1bbe..3022fe8 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -5228,3 +5228,65 @@ def __str__(self) -> str: attrs = dict(span.attributes or {}) assert "openarmature.tool.call.result" in attrs assert "OPAQUE-RESULT" in attrs["openarmature.tool.call.result"] + + +# --- per-branch dispatch key normalization (proposal 0084 lineage keys) ------- + + +def _branch_key_impls() -> list[tuple[str, Any]]: + """Both copies of the lineage key builder, so drift between them fails.""" + from openarmature.observability.otel.observer import _branch_dispatch_key as otel_key + + impls: list[tuple[str, Any]] = [("otel", otel_key)] + langfuse = pytest.importorskip("openarmature.observability.langfuse.observer") + impls.append(("langfuse", langfuse._branch_dispatch_key)) # noqa: SLF001 + return impls + + +@pytest.mark.parametrize(("label", "key"), _branch_key_impls()) +def test_branch_dispatch_key_pads_chains_shallower_than_the_prefix(label: str, key: Any) -> None: + # An orphan provider call issued from branch middleware carries EMPTY + # lineage chains, while the dispatch span was registered from an inner node + # event whose chains are padded to the namespace depth. Both denote "no + # enclosing fan-out at that depth", so they MUST produce the same key; when + # they did not, the lookup missed and the orphan span fell through to the + # invocation root (conformance fixture 152). + prefix = ("dispatcher",) + registered = key(prefix, (None,), (), "branch_a") + from_orphan = key(prefix, (), (), "branch_a") + assert from_orphan == registered, f"{label}: shallow chains must normalize to the registered key" + + +@pytest.mark.parametrize(("label", "key"), _branch_key_impls()) +def test_branch_dispatch_key_still_discriminates_real_lineages(label: str, key: Any) -> None: + # The padding must not collapse genuinely different enclosing lineages: a pb + # node inside outer fan-out instance 0 and the same node inside instance 1 + # are different dispatch spans and must not share a key. + prefix = ("outer", "dispatcher") + assert key(prefix, (0, None), (), "b") != key(prefix, (1, None), (), "b"), ( + f"{label}: distinct enclosing fan-out instances must not collide" + ) + assert key(prefix, (None, None), ("x",), "b") != key(prefix, (None, None), ("y",), "b"), ( + f"{label}: distinct enclosing branch chains must not collide" + ) + assert key(prefix, (None, None), (), "a") != key(prefix, (None, None), (), "b"), ( + f"{label}: distinct branch names must not collide" + ) + + +def test_branch_dispatch_key_copies_agree() -> None: + # The two implementations are duplicated by design (one per backend) and + # their docstrings say each mirrors the other. Nothing enforced that, so the + # padding defect existed in both and was fixed in both by hand. + impls = _branch_key_impls() + assert len(impls) == 2, "expected both backends' key builders to be importable" + cases = [ + (("dispatcher",), (), (), "a"), + (("dispatcher",), (None,), (), "a"), + (("outer", "dispatcher"), (0,), (), "a"), + (("outer", "dispatcher"), (0, 1), ("x",), "a"), + ((), (), (), "a"), + ] + for args in cases: + results = {label: key(*args) for label, key in impls} + assert len(set(results.values())) == 1, f"key builders disagree on {args}: {results}" From f478b70ec9e71cf1916d91d6eae1eba6179c5b21 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Wed, 19 Aug 2026 23:51:14 -0700 Subject: [PATCH 3/7] 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. 152 was failing on the key defect fixed separately, and is the only fixture that discriminates it. The driver rides on the generic graph builder rather than copying fixture 133's, which is hardcoded to that fixture's subgraph and node names. It attaches each orphan wrapper to whatever encloses its subgraph, which is the fixtures' own claim about the fallback: instance middleware for a fan-out target, branch middleware for a parallel branch. Node middleware would not do, running entirely inside the node span in both phases. Nine invariants are implemented rather than recorded as documentary, because the tree matcher pins "X appears under Y" but tolerates extra children, so it can express neither the absence claims nor the count nor the branch close ordering. Subgraph declarations are read from the case as well as the fixture top level. Fifteen fixtures across six capabilities use the case-level form that conformance-adapter section 11 does not document; reading only the documented level is what made 152 fail with a bare KeyError. --- tests/conformance/test_observability.py | 390 +++++++++++++++++++++++- 1 file changed, 383 insertions(+), 7 deletions(-) diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 22b8fdb..72ac07d 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. 152's fallback is the + # per-branch dispatch span; 153 nests a fan-out between them so it is the + # innermost instance span. + "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,13 +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" ), - "152-otel-parallel-branch-orphan-llm-fallback": ( - "reuses fixture 133's orphan-fallback driver, which does not build the `subgraphs` block " - "152 adds (KeyError: 'subgraphs')" - ), - "153-otel-mixed-nesting-orphan-llm-fallback": ( - "same driver gap as 152, one nesting level deeper (KeyError: 'leaf_sg')" - ), # Proposal 0109 (spec v0.104.0) token-budget failure-path parity. } @@ -678,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/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. @@ -900,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) @@ -6804,6 +6811,375 @@ 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 not any count. That is why the absence and +# count 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" +_FAN_OUT_INDEX_ATTR = "openarmature.node.fan_out_index" + + +# Keys that ONLY `_assert_error_span_extras` reads out of a `span_tree` entry. +_ERROR_SPAN_EXTRA_KEYS = frozenset({"attributes_absent", "status_description", "exception_recorded"}) + + +def _assert_no_error_span_extra_keys(expected_tree: Sequence[Mapping[str, Any]]) -> None: + """Fail if a span_tree entry declares a key whose only reader is skipped.""" + + def _walk(entries: Sequence[Mapping[str, Any]], path: str) -> None: + for entry in entries: + name = cast("str", entry.get("name") or "") + here = f"{path}/{name}" + declared = sorted(_ERROR_SPAN_EXTRA_KEYS & set(entry)) + assert not declared, ( + f"{here} declares {declared}, which only `_assert_error_span_extras` reads, and " + f"this driver skips that helper because it cannot handle the duplicate span names " + f"153 emits. Teach the helper name+attribute disambiguation before this fixture " + f"can carry those keys, or the claim runs nowhere." + ) + _walk(cast("Sequence[Mapping[str, Any]]", entry.get("children") or []), here) + + _walk(expected_tree, "") + + +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 below holds trivially, + # and so does every "its parent is Y" claim (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)}" + ) + + def _is_branch_dispatch(span: Any) -> bool: + return _BRANCH_NAME_ATTR in dict(span.attributes or {}) + + 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; " + f"its parent is {parent.name if parent else None!r}" + ) + + if invariants.get("orphan_llm_span_parents_under_innermost_fan_out_instance"): + for llm in llm_spans: + parent = _parent_of(llm) + attrs = dict(parent.attributes or {}) if parent is not None else {} + assert parent is not None and _FAN_OUT_INDEX_ATTR in attrs, ( + f"the orphan {_LLM_SPAN} MUST parent under the innermost fan-out INSTANCE span " + f"(the one carrying {_FAN_OUT_INDEX_ATTR}); its parent is " + f"{parent.name if parent else None!r} with attributes {sorted(attrs)}" + ) + + 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, so no branch may + # collect two while another collects none. The span_tree cannot express + # this: it requires one under each branch but tolerates extras. + per_branch: dict[Any, int] = {} + for llm in llm_spans: + parent = _parent_of(llm) + if parent is not None and _is_branch_dispatch(parent): + per_branch[parent.context.span_id] = per_branch.get(parent.context.span_id, 0) + 1 + counts = sorted(per_branch.values()) + assert counts and set(counts) == {1}, ( + f"each branch dispatch span MUST collect exactly one orphan {_LLM_SPAN}; got " + f"per-branch counts {counts}" + ) + + 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" + 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. + is_forbidden = dict(parent.attributes or {}).get(_BRANCH_NAME_ATTR) == forbidden + assert not is_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" + ) + node_names = set(cast("dict[str, Any]", case.get("nodes") or {})) + assert not ( + parent.name in node_names and _BRANCH_NAME_ATTR not in dict(parent.attributes or {}) + ), f"the orphan {_LLM_SPAN} MUST NOT parent under a NODE span; its parent is {parent.name!r}" + + 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. + dispatch = [s for s in spans if _is_branch_dispatch(s) and s.name in set(order)] + assert len(dispatch) == len(order), ( + f"expected one dispatch span per declared branch {order}; got {[s.name for s in dispatch]}" + ) + actual = [s.name for s in sorted(dispatch, key=lambda s: cast("int", s.end_time))] + assert actual == order, ( + f"branch dispatch spans MUST close in declaration order {order}; closed as {actual}" + ) + + +# --------------------------------------------------------------------------- +# Fixtures 152 / 153 — orphan LLM span parent resolution under a +# parallel-branches dispatcher (proposal 0084, observability §5.5) +# --------------------------------------------------------------------------- +# +# Both assert where a provider span lands when it is emitted from a WRAPPER +# around the calling node rather than from the node body, so the calling node's +# span is not open at emit and the span must fall back to the nearest enclosing +# wrapper. 152 puts the guard directly in a branch, so the fallback is the +# BRANCH DISPATCH span; 153 nests a fan-out between them, so it is the innermost +# fan-out INSTANCE span. +# +# Fixture 133 covers the same rule for a pure fan-out topology through a +# hand-built driver. This one does NOT reuse that driver: it is hardcoded to +# 133's subgraph and node names and to a rendezvous keyed on `fan_out.items_field`. +# The generic `build_graph` already models parallel branches, subgraphs and +# nested fan-outs, and already exposes the two middleware seams the orphan call +# needs, so these two ride on it instead. + + +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 §11 documents `subgraphs:` as a FIXTURE TOP-LEVEL + # block, and 153 puts it there. 152 puts it inside the case, as fifteen + # fixtures across six capabilities do, none of which also carries a + # top-level block. Reading only the documented level is what made 152 fail + # with a bare `KeyError: 'subgraphs'`, which reads like a missing directive + # rather than a placement difference. Raised with spec; accept both. + merged: dict[str, Any] = {} + for source in (spec, case): + merged.update(cast("dict[str, Any]", source.get("subgraphs") or {})) + return merged + + +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 + + +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 json # noqa: PLC0415 + + import httpx # noqa: PLC0415 + + from openarmature.llm import OpenAIProvider, UserMessage # noqa: PLC0415 + + from .adapter import build_graph # noqa: PLC0415 + + # ---- FIFO mock provider. 152's two canned responses differ per branch so + # the routing invariant can discriminate which branch issued which call. + 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), + ) + + def _make_orphan_mw(wrapper_spec: Mapping[str, Any]) -> Any: + messages: tuple[Any, ...] = tuple( + UserMessage(content=m["content"]) + for m in cast("list[dict[str, str]]", wrapper_spec.get("messages") or []) + if m.get("role") == "user" + ) or (UserMessage(content="guardrail check"),) + # `phase` decides whether the side call fires before or after the + # wrapped step. Both fixtures declare `pre`, and it is read rather than + # assumed: in `pre` the calling node has not started, in `post` it has + # already closed, and the fallback target is the same either way only + # because neither leaves the calling node's span open. + 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)) + return await next_call(state) + result = await next_call(state) + await provider.complete(list(messages)) + 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 (153), a parallel branch takes BRANCH + # middleware (152). Not node middleware on the guard itself, which runs + # entirely inside the node span in both phases and so can never orphan. + # + # Phase is read from the directive but does not decide the parent here. + # Observer events are queued and drained, so the provider span is resolved + # retrospectively, by which point the branch's first inner node has started + # and the dispatch span exists. Both phases therefore resolve the same way, + # which is measured in the wiring tests rather than assumed. + subgraph_specs = _merged_subgraph_specs(case, spec) + instance_mw: dict[str, dict[str, list[Any]]] = {} + branch_mw: dict[str, dict[str, list[Any]]] = {} + 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_name, host_spec in subgraph_specs.items(): + for fo_node, fo_spec in cast("dict[str, Any]", host_spec.get("nodes") or {}).items(): + if cast("dict[str, Any]", fo_spec.get("fan_out") or {}).get("subgraph") == sg_name: + instance_mw.setdefault(host_name, {}).setdefault(fo_node, []).append(mw) + attached = True + for outer_node, outer_spec in cast("dict[str, Any]", case.get("nodes") or {}).items(): + pb = cast("dict[str, Any] | None", outer_spec.get("parallel_branches")) + if pb is None: + continue + for branch_name, branch_cfg in cast("dict[str, Any]", pb["branches"]).items(): + if branch_cfg.get("subgraph") == sg_name: + branch_mw.setdefault(outer_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]", sg_spec.get("nodes") or {}) + needed = { + cast("dict[str, Any]", n["fan_out"])["subgraph"] + for n in nodes.values() + if "fan_out" in n and "subgraph" in cast("dict[str, Any]", n["fan_out"]) + } | {cast("str", n["subgraph"]) for n in nodes.values() if isinstance(n.get("subgraph"), str)} + 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)}" + ) + + observer, exporter = _build_observer() + built = build_graph(case, subgraphs=compiled, trace=[], parallel_branches_branch_middleware=branch_mw) + graph = built.builder.compile() + graph.attach_observer(observer) + try: + 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`: that helper assumes unique span + # names, which 153 breaks (`inner_fan_out` names both the node span and + # its instance spans). Skipping it is only safe while neither fixture + # declares a key it is the sole reader of, so assert exactly that rather + # than leaving it to hold by luck: a pin bump adding `attributes_absent` + # would otherwise be dropped in silence. + _assert_no_error_span_extra_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) + + 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).""" From 37a7f5aa77f067492ee08a9e7e5178da4b0ebc10 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 20 Aug 2026 23:48:52 -0700 Subject: [PATCH 4/7] Re-defer 152/153 on a measured race, not a driver gap A driver for both exists earlier in this branch and makes them pass. 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. One `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, which 153's own invariant forbids. Activating them would have certified a lucky interleaving as conformance. The lineage-key defect they exposed is real and stays fixed. It is necessary and not sufficient: these need the orphan parent resolved deterministically, by deferring the decision until the enclosing wrapper span is known, which is an observer change rather than a harness one. The deferral strings now carry the measurement so the next author starts from it instead of rediscovering a driver gap that is not the obstacle. --- tests/conformance/test_observability.py | 407 ++---------------------- 1 file changed, 24 insertions(+), 383 deletions(-) diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 72ac07d..d64db98 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -263,12 +263,6 @@ 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. 152's fallback is the - # per-branch dispatch span; 153 nests a fan-out between them so it is the - # innermost instance span. - "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 + @@ -452,6 +446,30 @@ 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. } @@ -677,9 +695,6 @@ 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/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. @@ -902,11 +917,6 @@ 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) @@ -6811,375 +6821,6 @@ 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 not any count. That is why the absence and -# count 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" -_FAN_OUT_INDEX_ATTR = "openarmature.node.fan_out_index" - - -# Keys that ONLY `_assert_error_span_extras` reads out of a `span_tree` entry. -_ERROR_SPAN_EXTRA_KEYS = frozenset({"attributes_absent", "status_description", "exception_recorded"}) - - -def _assert_no_error_span_extra_keys(expected_tree: Sequence[Mapping[str, Any]]) -> None: - """Fail if a span_tree entry declares a key whose only reader is skipped.""" - - def _walk(entries: Sequence[Mapping[str, Any]], path: str) -> None: - for entry in entries: - name = cast("str", entry.get("name") or "") - here = f"{path}/{name}" - declared = sorted(_ERROR_SPAN_EXTRA_KEYS & set(entry)) - assert not declared, ( - f"{here} declares {declared}, which only `_assert_error_span_extras` reads, and " - f"this driver skips that helper because it cannot handle the duplicate span names " - f"153 emits. Teach the helper name+attribute disambiguation before this fixture " - f"can carry those keys, or the claim runs nowhere." - ) - _walk(cast("Sequence[Mapping[str, Any]]", entry.get("children") or []), here) - - _walk(expected_tree, "") - - -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 below holds trivially, - # and so does every "its parent is Y" claim (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)}" - ) - - def _is_branch_dispatch(span: Any) -> bool: - return _BRANCH_NAME_ATTR in dict(span.attributes or {}) - - 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; " - f"its parent is {parent.name if parent else None!r}" - ) - - if invariants.get("orphan_llm_span_parents_under_innermost_fan_out_instance"): - for llm in llm_spans: - parent = _parent_of(llm) - attrs = dict(parent.attributes or {}) if parent is not None else {} - assert parent is not None and _FAN_OUT_INDEX_ATTR in attrs, ( - f"the orphan {_LLM_SPAN} MUST parent under the innermost fan-out INSTANCE span " - f"(the one carrying {_FAN_OUT_INDEX_ATTR}); its parent is " - f"{parent.name if parent else None!r} with attributes {sorted(attrs)}" - ) - - 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, so no branch may - # collect two while another collects none. The span_tree cannot express - # this: it requires one under each branch but tolerates extras. - per_branch: dict[Any, int] = {} - for llm in llm_spans: - parent = _parent_of(llm) - if parent is not None and _is_branch_dispatch(parent): - per_branch[parent.context.span_id] = per_branch.get(parent.context.span_id, 0) + 1 - counts = sorted(per_branch.values()) - assert counts and set(counts) == {1}, ( - f"each branch dispatch span MUST collect exactly one orphan {_LLM_SPAN}; got " - f"per-branch counts {counts}" - ) - - 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" - 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. - is_forbidden = dict(parent.attributes or {}).get(_BRANCH_NAME_ATTR) == forbidden - assert not is_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" - ) - node_names = set(cast("dict[str, Any]", case.get("nodes") or {})) - assert not ( - parent.name in node_names and _BRANCH_NAME_ATTR not in dict(parent.attributes or {}) - ), f"the orphan {_LLM_SPAN} MUST NOT parent under a NODE span; its parent is {parent.name!r}" - - 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. - dispatch = [s for s in spans if _is_branch_dispatch(s) and s.name in set(order)] - assert len(dispatch) == len(order), ( - f"expected one dispatch span per declared branch {order}; got {[s.name for s in dispatch]}" - ) - actual = [s.name for s in sorted(dispatch, key=lambda s: cast("int", s.end_time))] - assert actual == order, ( - f"branch dispatch spans MUST close in declaration order {order}; closed as {actual}" - ) - - -# --------------------------------------------------------------------------- -# Fixtures 152 / 153 — orphan LLM span parent resolution under a -# parallel-branches dispatcher (proposal 0084, observability §5.5) -# --------------------------------------------------------------------------- -# -# Both assert where a provider span lands when it is emitted from a WRAPPER -# around the calling node rather than from the node body, so the calling node's -# span is not open at emit and the span must fall back to the nearest enclosing -# wrapper. 152 puts the guard directly in a branch, so the fallback is the -# BRANCH DISPATCH span; 153 nests a fan-out between them, so it is the innermost -# fan-out INSTANCE span. -# -# Fixture 133 covers the same rule for a pure fan-out topology through a -# hand-built driver. This one does NOT reuse that driver: it is hardcoded to -# 133's subgraph and node names and to a rendezvous keyed on `fan_out.items_field`. -# The generic `build_graph` already models parallel branches, subgraphs and -# nested fan-outs, and already exposes the two middleware seams the orphan call -# needs, so these two ride on it instead. - - -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 §11 documents `subgraphs:` as a FIXTURE TOP-LEVEL - # block, and 153 puts it there. 152 puts it inside the case, as fifteen - # fixtures across six capabilities do, none of which also carries a - # top-level block. Reading only the documented level is what made 152 fail - # with a bare `KeyError: 'subgraphs'`, which reads like a missing directive - # rather than a placement difference. Raised with spec; accept both. - merged: dict[str, Any] = {} - for source in (spec, case): - merged.update(cast("dict[str, Any]", source.get("subgraphs") or {})) - return merged - - -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 - - -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 json # noqa: PLC0415 - - import httpx # noqa: PLC0415 - - from openarmature.llm import OpenAIProvider, UserMessage # noqa: PLC0415 - - from .adapter import build_graph # noqa: PLC0415 - - # ---- FIFO mock provider. 152's two canned responses differ per branch so - # the routing invariant can discriminate which branch issued which call. - 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), - ) - - def _make_orphan_mw(wrapper_spec: Mapping[str, Any]) -> Any: - messages: tuple[Any, ...] = tuple( - UserMessage(content=m["content"]) - for m in cast("list[dict[str, str]]", wrapper_spec.get("messages") or []) - if m.get("role") == "user" - ) or (UserMessage(content="guardrail check"),) - # `phase` decides whether the side call fires before or after the - # wrapped step. Both fixtures declare `pre`, and it is read rather than - # assumed: in `pre` the calling node has not started, in `post` it has - # already closed, and the fallback target is the same either way only - # because neither leaves the calling node's span open. - 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)) - return await next_call(state) - result = await next_call(state) - await provider.complete(list(messages)) - 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 (153), a parallel branch takes BRANCH - # middleware (152). Not node middleware on the guard itself, which runs - # entirely inside the node span in both phases and so can never orphan. - # - # Phase is read from the directive but does not decide the parent here. - # Observer events are queued and drained, so the provider span is resolved - # retrospectively, by which point the branch's first inner node has started - # and the dispatch span exists. Both phases therefore resolve the same way, - # which is measured in the wiring tests rather than assumed. - subgraph_specs = _merged_subgraph_specs(case, spec) - instance_mw: dict[str, dict[str, list[Any]]] = {} - branch_mw: dict[str, dict[str, list[Any]]] = {} - 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_name, host_spec in subgraph_specs.items(): - for fo_node, fo_spec in cast("dict[str, Any]", host_spec.get("nodes") or {}).items(): - if cast("dict[str, Any]", fo_spec.get("fan_out") or {}).get("subgraph") == sg_name: - instance_mw.setdefault(host_name, {}).setdefault(fo_node, []).append(mw) - attached = True - for outer_node, outer_spec in cast("dict[str, Any]", case.get("nodes") or {}).items(): - pb = cast("dict[str, Any] | None", outer_spec.get("parallel_branches")) - if pb is None: - continue - for branch_name, branch_cfg in cast("dict[str, Any]", pb["branches"]).items(): - if branch_cfg.get("subgraph") == sg_name: - branch_mw.setdefault(outer_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]", sg_spec.get("nodes") or {}) - needed = { - cast("dict[str, Any]", n["fan_out"])["subgraph"] - for n in nodes.values() - if "fan_out" in n and "subgraph" in cast("dict[str, Any]", n["fan_out"]) - } | {cast("str", n["subgraph"]) for n in nodes.values() if isinstance(n.get("subgraph"), str)} - 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)}" - ) - - observer, exporter = _build_observer() - built = build_graph(case, subgraphs=compiled, trace=[], parallel_branches_branch_middleware=branch_mw) - graph = built.builder.compile() - graph.attach_observer(observer) - try: - 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`: that helper assumes unique span - # names, which 153 breaks (`inner_fan_out` names both the node span and - # its instance spans). Skipping it is only safe while neither fixture - # declares a key it is the sole reader of, so assert exactly that rather - # than leaving it to hold by luck: a pin bump adding `attributes_absent` - # would otherwise be dropped in silence. - _assert_no_error_span_extra_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) - - 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).""" From f1e6f264d94759489848a27245176f7cd0f2e8fa Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 20 Aug 2026 23:50:32 -0700 Subject: [PATCH 5/7] Dedupe the callable-branch dispatch span on its real key The guard compared a legacy `namespace + (branch_name,)` tuple against a dict keyed by the 4-tuple the opener actually stores, so it never matched. `_open_started_span` runs twice for a callable-branch started event, once from the engine task's `prepare_sync` and once from the async `__call__`, and the callable-branch arm returns before `open_spans` is written, so the usual idempotency check cannot short-circuit either. The second open overwrote the first in the registry; the overwritten span was never ended and never exported. Measured on a two-callable-branch graph: four opens for two branches. The exported span TREE is identical either way, which is why no conformance fixture catches it. What differs is the span published into the branch body as active, so a log record emitted from a callable branch carried a span id absent from the trace. The Langfuse observer's equivalent guard already used the shared key builder; this brings the two backends back into agreement. --- src/openarmature/observability/otel/observer.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index b48824d..dc456cf 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -1057,7 +1057,20 @@ def _open_started_span(self, event: NodeEvent) -> None: and event.parallel_branches_config is None and event.namespace in inv_state.parallel_branches_parent_node_name ): - branch_key = event.namespace + (event.branch_name,) + # Keyed the same way `_open_parallel_branches_branch_dispatch_span` + # STORES it, and the same way the Langfuse observer's equivalent + # guard already did. A legacy `namespace + (branch_name,)` tuple was + # compared against a dict keyed by the 4-tuple, so it never matched: + # `_open_started_span` runs twice for a callable-branch started event + # (once from the engine task's `prepare_sync`, once from the async + # `__call__`), and the second open overwrote the first in the dict. + # The overwritten span was never ended and never exported, so a log + # record emitted from the branch body carried a span id absent from + # the trace. The span TREE looked correct, which is why no fixture + # catches it. + branch_key = _branch_dispatch_key( + event.namespace, event.fan_out_index_chain, event.branch_name_chain, event.branch_name + ) if branch_key not in inv_state.parallel_branches_branch_spans: self._open_parallel_branches_branch_dispatch_span( inv_state, correlation_id, event.namespace, event From 9e4b8ae48bd42501ac7b2719014d3f06db7b27ed Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 20 Aug 2026 23:50:45 -0700 Subject: [PATCH 6/7] Keep the langfuse skip off the OTel module scope `pytest.importorskip` was called while building a parametrize argument, so it ran at module import and its skip is module-scoped. Without the langfuse extra the whole file collapsed to a single skip, taking ~104 unrelated OTel tests with it, including the tripwire that exists to catch drift between the two hand-duplicated key builders. Measured with the import blocked: 1 skipped before, 104 passed and 3 skipped after. CI installs all extras, which is what kept it silent. Also pins the callable-branch dispatch span against being opened twice. That one asserts on the dispatch-span registry rather than the exported tree, because the tree is identical under the defect. --- tests/unit/test_observability_otel.py | 103 ++++++++++++++++++++++---- 1 file changed, 88 insertions(+), 15 deletions(-) diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index 3022fe8..af7cfa7 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -5233,18 +5233,26 @@ def __str__(self) -> str: # --- per-branch dispatch key normalization (proposal 0084 lineage keys) ------- -def _branch_key_impls() -> list[tuple[str, Any]]: - """Both copies of the lineage key builder, so drift between them fails.""" - from openarmature.observability.otel.observer import _branch_dispatch_key as otel_key - - impls: list[tuple[str, Any]] = [("otel", otel_key)] - langfuse = pytest.importorskip("openarmature.observability.langfuse.observer") - impls.append(("langfuse", langfuse._branch_dispatch_key)) # noqa: SLF001 - return impls - - -@pytest.mark.parametrize(("label", "key"), _branch_key_impls()) -def test_branch_dispatch_key_pads_chains_shallower_than_the_prefix(label: str, key: Any) -> None: +# Both copies of the lineage key builder, parametrized by MODULE NAME rather than +# by imported function. Calling `pytest.importorskip` while building the +# parametrize argument runs it at module import, and its skip is module-scoped: +# without the langfuse extra the whole ~5,300-line OTel module would collapse to +# a single skip, taking ~95 unrelated tests with it. Importing inside the test +# body keeps the skip to the tests that actually need the extra. +_BRANCH_KEY_MODULES = [ + ("otel", "openarmature.observability.otel.observer"), + ("langfuse", "openarmature.observability.langfuse.observer"), +] + + +def _branch_key(module_name: str) -> Any: + module = pytest.importorskip(module_name) + return module._branch_dispatch_key # noqa: SLF001 + + +@pytest.mark.parametrize(("label", "module_name"), _BRANCH_KEY_MODULES) +def test_branch_dispatch_key_pads_chains_shallower_than_the_prefix(label: str, module_name: str) -> None: + key = _branch_key(module_name) # An orphan provider call issued from branch middleware carries EMPTY # lineage chains, while the dispatch span was registered from an inner node # event whose chains are padded to the namespace depth. Both denote "no @@ -5257,8 +5265,9 @@ def test_branch_dispatch_key_pads_chains_shallower_than_the_prefix(label: str, k assert from_orphan == registered, f"{label}: shallow chains must normalize to the registered key" -@pytest.mark.parametrize(("label", "key"), _branch_key_impls()) -def test_branch_dispatch_key_still_discriminates_real_lineages(label: str, key: Any) -> None: +@pytest.mark.parametrize(("label", "module_name"), _BRANCH_KEY_MODULES) +def test_branch_dispatch_key_still_discriminates_real_lineages(label: str, module_name: str) -> None: + key = _branch_key(module_name) # The padding must not collapse genuinely different enclosing lineages: a pb # node inside outer fan-out instance 0 and the same node inside instance 1 # are different dispatch spans and must not share a key. @@ -5278,7 +5287,7 @@ def test_branch_dispatch_key_copies_agree() -> None: # The two implementations are duplicated by design (one per backend) and # their docstrings say each mirrors the other. Nothing enforced that, so the # padding defect existed in both and was fixed in both by hand. - impls = _branch_key_impls() + impls = [(label, _branch_key(name)) for label, name in _BRANCH_KEY_MODULES] assert len(impls) == 2, "expected both backends' key builders to be importable" cases = [ (("dispatcher",), (), (), "a"), @@ -5290,3 +5299,67 @@ def test_branch_dispatch_key_copies_agree() -> None: for args in cases: results = {label: key(*args) for label, key in impls} assert len(set(results.values())) == 1, f"key builders disagree on {args}: {results}" + + +async def test_callable_branch_dispatch_span_is_opened_once() -> None: + # `_open_started_span` runs TWICE for a callable-branch started event: once + # from the engine task's `prepare_sync`, once from the async `__call__`. The + # dedup guard compared a legacy `namespace + (branch_name,)` tuple against a + # dict keyed by the 4-tuple `_BranchDispatchKey`, so it never matched and a + # second span was opened, overwriting the first. The overwritten span was + # never ended and never exported. + # + # The exported span TREE is identical either way, which is why no + # conformance fixture catches this. What differs is the span published into + # the branch body as the active span: under the defect it is the orphaned + # copy, so a log record emitted from the branch carries a span id that + # appears nowhere in the trace. This asserts on the dispatch-span registry + # rather than the tree, since the registry is where the overwrite happens. + from openarmature.graph.parallel_branches import BranchSpec + from openarmature.observability.otel.observer import _branch_dispatch_key + + class _S(State): + n: int = 0 + + async def _ca(_s: Any) -> dict[str, Any]: + return {} + + async def _cb(_s: Any) -> dict[str, Any]: + return {} + + opened: list[Any] = [] + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + original = observer._open_parallel_branches_branch_dispatch_span # noqa: SLF001 + + def _spy(*args: Any, **kwargs: Any) -> Any: + event = cast("Any", args[-1] if args else kwargs.get("event")) + assert event is not None, "spy received no event to key on" + opened.append( + _branch_dispatch_key( + event.namespace, + event.fan_out_index_chain, + event.branch_name_chain, + event.branch_name, + ) + ) + return original(*args, **kwargs) + + observer._open_parallel_branches_branch_dispatch_span = _spy # type: ignore[method-assign] # noqa: SLF001 + + graph = ( + GraphBuilder(_S) + .add_parallel_branches_node("pb", branches={"ca": BranchSpec(call=_ca), "cb": BranchSpec(call=_cb)}) + .add_edge("pb", END) + .set_entry("pb") + .compile() + ) + graph.attach_observer(observer) + await graph.invoke(_S()) + await graph.drain() + observer.shutdown() + + assert len(opened) == len(set(opened)), ( + f"each callable branch's dispatch span MUST be opened once; opened {opened}" + ) + assert len(opened) == 2, f"expected one dispatch span per callable branch; opened {opened}" From ab3919b6f47e2388d557f79ab1c705dc66b2035f Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 21 Aug 2026 00:19:31 -0700 Subject: [PATCH 7/7] Pin the branch-chain half of the dispatch key padding The existing padding test uses a depth-1 prefix, where the branch slice is `chain[:0]` and is empty whether padded or not, so it only ever exercised the fan-out half. Deleting the `branches` padding from both observers left the whole suite green; deleting it from one was caught only incidentally, by the agreement test noticing the copies diverged. A depth-2 prefix slices `chain[:1]`, so a caller whose branch chain is shorter than `n - 1` builds an unpadded key against a padded registration. Both backends are driven, and the discrimination test still rules out a padding that collapses distinct branch lineages. --- tests/unit/test_observability_otel.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index af7cfa7..9eee0e8 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -5265,6 +5265,28 @@ def test_branch_dispatch_key_pads_chains_shallower_than_the_prefix(label: str, m assert from_orphan == registered, f"{label}: shallow chains must normalize to the registered key" +@pytest.mark.parametrize(("label", "module_name"), _BRANCH_KEY_MODULES) +def test_branch_dispatch_key_pads_branch_chain_shallower_than_the_prefix( + label: str, module_name: str +) -> None: + key = _branch_key(module_name) + # The branch-name half of the same normalization, which the fan-out test + # above does not reach: it uses a depth-1 prefix, where the branch slice is + # `chain[:0]` and is empty whether padded or not. A depth-2 prefix slices + # `chain[:1]`, so a caller whose branch chain is shorter than `n - 1` builds + # an unpadded key while the span was registered with a padded one. + # + # Without this, deleting the `branches` padding line from BOTH copies leaves + # the entire suite green; deleting it from one is caught only incidentally, + # by the agreement test noticing the copies diverged. + prefix = ("outer", "dispatcher") + registered = key(prefix, (None, None), (None,), "branch_a") + from_orphan = key(prefix, (None, None), (), "branch_a") + assert from_orphan == registered, ( + f"{label}: a branch chain shallower than the prefix must normalize to the registered key" + ) + + @pytest.mark.parametrize(("label", "module_name"), _BRANCH_KEY_MODULES) def test_branch_dispatch_key_still_discriminates_real_lineages(label: str, module_name: str) -> None: key = _branch_key(module_name)