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..dc456cf 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 @@ -1047,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 diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 22b8fdb..d64db98 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -446,12 +446,29 @@ 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": ( - "reuses fixture 133's orphan-fallback driver, which does not build the `subgraphs` block " - "152 adds (KeyError: 'subgraphs')" + "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 driver gap as 152, one nesting level deeper (KeyError: 'leaf_sg')" + "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. } diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index b8c1bbe..9eee0e8 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -5228,3 +5228,160 @@ 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) ------- + + +# 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 + # 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", "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) + # 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 = [(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"), + (("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}" + + +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}"