diff --git a/CHANGELOG.md b/CHANGELOG.md index d73a94e..3a90bad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The - **`current_fan_out_index()` inside fan-out instance middleware** now returns the executing instance's index (and `current_fan_out_index_chain()` its lineage) instead of `None`. The engine set the fan-out lineage ContextVars per-node, inside the inner subgraph, which left them unset in `instance_middleware` that wraps the subgraph from outside; they are now set around the instance-middleware chain. The documented `instance_middleware` use (`RetryMiddleware`) does not read the index, so no shipped behavior changes. This corrects the value seen by custom instance middleware that reads the index or calls `set_invocation_metadata`. - **Langfuse per-branch dispatch-span observation** (observability §4.3 / §8.4.2, proposals 0042 / 0044). The Langfuse observer now synthesizes a per-branch Span observation under a `parallel_branches` dispatcher node, so each branch's inner observations nest under their own branch span (a three-level dispatcher / per-branch-span / inner-nodes tree) instead of parenting directly under the dispatcher. The per-branch observation carries the OA-emitted `branch_name` alongside the caller baseline metadata and any per-branch augmentation, and the Generation observation now carries `branch_name` too. The OTel observer already produced this shape (proposal 0044 shipped OTel-only in v0.11.0); this brings the Langfuse mapping into line. Callable branches (proposal 0075) are unchanged. +- **Augmentation no longer lands on a shared-parent fan-out / parallel-branches node** (observability §3.4, proposal 0045). A key set via `set_invocation_metadata` inside a fan-out instance or a parallel-branches branch was incorrectly applied to the shared fan-out / dispatcher NODE span (the fork point) when the augmenting context executed at that node's own namespace, in addition to the per-instance / per-branch dispatch span where it belongs. Both the OTel and Langfuse observers now skip the shared-parent node in that case, matching the behavior already applied to strict-ancestor shared parents. The per-instance / per-branch dispatch spans and the lineage ancestors that carry the augmentation are unaffected. +- **Langfuse fan-out-instance dispatch nested below the top level** (observability §5.4, proposal 0013). The Langfuse observer's per-instance dispatch synthesis and parent resolution are now prefix-general, so a fan-out node sitting inside a serial subgraph wrapper (rather than at the top namespace level) gets its per-instance dispatch observation synthesized and its inner observations parented under it. This matches the OTel observer, which already resolved across every namespace prefix. ## [0.15.0] — 2026-06-22 diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 05af4c4..1858b2b 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -669,12 +669,14 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non # parent_node_name caches. for key, observation in inv_state.open_observations.items(): ns, _ai, _fi, _bn = key - if ns == aug_ns: - if _observation_chain_on_path(observation, aug_fi_chain, aug_bn_chain): - observation.handle.update(metadata=metadata_delta) - continue - if not is_strict_prefix(ns, aug_ns): + if ns != aug_ns and not is_strict_prefix(ns, aug_ns): continue + # A fan-out / pb NODE is a shared parent and MUST NOT carry an + # instance's / branch's augmentation (proposal 0045 §3.4). This skip + # applies whether the NODE sits strictly above the augmenter OR at + # the augmenter's own namespace: an instance/branch executes AT the + # fan-out/pb node's namespace, so ns == aug_ns also matches the shared + # NODE (its per-instance dispatch is the one updated, separately above). if ns in inv_state.fan_out_parent_node_name or ns in inv_state.parallel_branches_parent_node_name: continue if _observation_chain_on_path(observation, aug_fi_chain, aug_bn_chain): @@ -914,19 +916,24 @@ def _resolve_parent_observation_id(self, inv_state: _InvState, event: NodeEvent) # 3. Leaf node observation at any matching ancestor prefix, # walked longest-first. # 4. None — the Trace itself becomes the implicit parent. - # Per proposal 0044: an inner branch node parents under its per-branch - # dispatch observation (longest-first; innermost branch wins). - if event.branch_name is not None: - for prefix_len in range(len(event.namespace) - 1, 0, -1): - prefix = event.namespace[:prefix_len] + # Per proposals 0044 / 0013: an inner node parents under the INNERMOST + # dispatch on its lineage -- a per-branch dispatch + # (parallel_branches_branch_spans, keyed prefix + (branch_name,)) or a + # per-instance fan-out dispatch (fan_out_instance_observations, keyed + # prefix + (str(fan_out_index),)). Walk prefixes longest-first so the + # innermost wins; this resolves arbitrary nesting (fan-out in fan-out, + # parallel-branches in fan-out, ...). Mirrors the OTel + # _resolve_parent_context. + for prefix_len in range(len(event.namespace), 0, -1): + prefix = event.namespace[:prefix_len] + if event.branch_name is not None: branch_dispatch = inv_state.parallel_branches_branch_spans.get(prefix + (event.branch_name,)) if branch_dispatch is not None: return branch_dispatch.handle.id - if event.fan_out_index is not None and event.namespace: - instance_key = event.namespace[:1] + (str(event.fan_out_index),) - dispatch = inv_state.fan_out_instance_observations.get(instance_key) - if dispatch is not None: - return dispatch.handle.id + if event.fan_out_index is not None: + dispatch = inv_state.fan_out_instance_observations.get(prefix + (str(event.fan_out_index),)) + if dispatch is not None: + return dispatch.handle.id for prefix_len in range(len(event.namespace) - 1, 0, -1): prefix = event.namespace[:prefix_len] sg = inv_state.subgraph_observations.get(prefix) @@ -1007,39 +1014,38 @@ def _sync_subgraph_observations( prefix = namespace[:depth] if prefix in inv_state.subgraph_observations: continue - # Non-detached per-instance dispatch for the current - # event's own fan-out instance gets opened below; skip - # the regular subgraph path here so we don't double-open. + # The per-instance dispatch for this event's own instance is opened + # below; skip the regular subgraph path so we don't double-open. Runs + # at ANY depth -- the fan-out may sit inside another fan-out instance + # or a subgraph wrapper -- mirroring the OTel observer. if ( - depth == 1 - and event.fan_out_index is not None + event.fan_out_index is not None and (prefix + (str(event.fan_out_index),)) in inv_state.fan_out_instance_observations ): continue - # Detached subgraph: the first segment matches a - # configured detached_subgraphs name → mint a fresh - # detached Trace + open the dispatch observation in it. + # Detached subgraph: kept top-level (depth == 1). _trace_id_for routes + # detached events by namespace[:1], so a nested detached unit would + # partially detach (its dispatch in the new Trace, inner nodes in the + # main one). Nested-detached support rides with the deferred + # nested-dispatch-keying fix that generalizes _trace_id_for too. if depth == 1 and prefix[0] in self.detached_subgraphs: self._open_detached_subgraph_trace(inv_state, correlation_id, prefix, event) continue - # Detached fan-out: the fan-out instance gets its own - # Trace per spec §8.5. The fan-out node's Span observation - # in the parent Trace already exists (opened on the - # fan-out node's started event); the detached dispatch - # observation goes into the new Trace. + # Detached fan-out: the fan-out instance gets its own Trace per spec + # §8.5. The fan-out node's Span observation in the parent Trace + # already exists; the detached dispatch goes into the new Trace. Kept + # top-level for the same reason as detached subgraphs above. if depth == 1 and event.fan_out_index is not None and prefix[0] in self.detached_fan_outs: self._open_detached_fan_out_instance_trace(inv_state, correlation_id, prefix, event) continue - # Non-detached fan-out: synthesize per-instance dispatch - # observation under the fan-out node observation (proposal - # 0013 v0.10.0). Only triggers when the inner event is - # inside a fan-out instance AND the fan-out node's - # parent_node_name has been cached (i.e., the fan-out - # node's own started event was seen). + # Non-detached fan-out: synthesize the per-instance dispatch + # observation under the fan-out node observation (proposal 0013 + # v0.10.0). The fan_out_parent_node_name cache match self-gates to + # the fan-out node's namespace, so this runs at any depth (nested + # fan-out, or a fan-out inside a subgraph wrapper / branch). if ( - depth == 1 - and event.fan_out_index is not None - and prefix[0] not in self.detached_fan_outs + event.fan_out_index is not None + and prefix[-1] not in self.detached_fan_outs and prefix in inv_state.fan_out_parent_node_name ): self._open_fan_out_instance_dispatch_observation(inv_state, correlation_id, prefix, event) @@ -1440,14 +1446,16 @@ def _close_parallel_branches_branch_dispatch_observation( def _find_node_observation( self, inv_state: _InvState, prefix: tuple[str, ...] ) -> _OpenObservation | None: - # Find a NODE's own open leaf observation at the given prefix (the - # fan-out or parallel-branches NODE, whose per-instance / per-branch - # dispatches parent under it). Retry middleware wrapping the node bumps - # the attempt_index; this scans for any entry at ``prefix`` with - # ``fan_out_index is None``. Only one such entry is open at a time - # (retry opens and closes within an attempt's lifecycle). + # Find a NODE's own open leaf observation at ``prefix`` (the fan-out or + # parallel-branches NODE, whose per-instance / per-branch dispatches + # parent under it). Scan by namespace only: the NODE may itself carry an + # outer fan_out_index / branch_name when it is nested inside another + # fan-out instance or branch, so filtering on fan_out_index is None would + # miss it. Only one entry per namespace is open at a time (retry opens + # and closes attempts serially), so the scan is unambiguous. Mirrors the + # OTel _find_fan_out_node_span. for key, observation in inv_state.open_observations.items(): - if key[0] == prefix and key[2] is None: + if key[0] == prefix: return observation return None diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index bb6854c..44f9fdd 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -1190,20 +1190,17 @@ def _collect_augmentation_targets( # parent_node_name caches) for key, open_span in inv_state.open_spans.items(): ns, _ai, _fi, _bn = key - if ns == aug_ns: - # Same context — must have matching chain to be the - # augmenter's own attempt rather than a sibling - # instance's same-named node. - if _span_chain_on_path(open_span, aug_fi_chain, aug_bn_chain): - targets.append(open_span.span) + if ns != aug_ns and not is_strict_prefix(ns, aug_ns): continue - if not is_strict_prefix(ns, aug_ns): - continue - # Shared-parent check: if this NODE is a fan-out node or - # a parallel-branches node (dispatcher), it's a shared - # parent and MUST NOT be updated regardless of cardinality - # (§3.4 — the structural classification governs, not the - # live sibling count). + # Shared-parent check: a fan-out NODE or parallel-branches NODE + # (dispatcher) is a shared parent and MUST NOT be updated regardless + # of cardinality (§3.4 — the structural classification governs, not + # the live sibling count). This applies whether the NODE is a strict + # ancestor OR sits at the augmenter's own namespace: an + # instance/branch executes AT the fan-out/pb node's namespace, so + # ns == aug_ns matches the shared NODE too (its per-instance dispatch + # span is the one updated, separately above). The chain check below + # still excludes sibling instances' same-named nodes. if ns in inv_state.fan_out_parent_node_name or ns in inv_state.parallel_branches_parent_node_name: continue if _span_chain_on_path(open_span, aug_fi_chain, aug_bn_chain): diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index 879ae08..a0addf0 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -18,6 +18,7 @@ import copy import json from collections.abc import Callable, Mapping, Sequence +from contextvars import ContextVar from datetime import UTC, datetime from pathlib import Path from typing import Any, cast @@ -26,7 +27,7 @@ import pytest import yaml -from openarmature.graph import END, GraphBuilder +from openarmature.graph import END, ExplicitMapping, GraphBuilder from openarmature.llm import OpenAIProvider from openarmature.llm.response import RuntimeConfig from openarmature.observability.langfuse import ( @@ -108,17 +109,14 @@ # proposal 0044) that inner branch nodes parent under, ported from the # OTel observer's parallel_branches_branch_spans machinery. "030-caller-metadata-parallel-branches-per-branch", - # 039 (nested-lineage augmentation, proposal 0045) stays deferred: the - # three cases need harness extensions the existing primitives lack. - # Cases 1 + 3 (nested fan-out / fan-out-in-serial) need the fan-out - # augment middleware to read items_field from the executing subgraph's - # RUNTIME state (the outer instance's threaded inner_seed), not the - # build-time initial_state the current _make_augment_instance_middleware - # captures. Case 2 (pb-inside-fan-out) needs a new - # augment_metadata_from_outer_item factory AND depends on 030's - # per-branch dispatch span landing first. 0045's contract IS exercised - # at unit level via the OTel observer's - # ``test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span``. + # 039 (nested-lineage augmentation, proposal 0045): the LangfuseObserver + # gained prefix-general fan-out-instance dispatch (so a fan-out under a + # serial wrapper parents correctly) and skips shared-parent NODEs in the + # augmentation walk (0045 §3.4 MUST-NOT). Case 3 (fan-out in a serial + # subgraph) is wired via the dedicated hand-built _build_039_graph runner; + # cases 1 + 2 are TEMPORARILY deferred via _DEFERRED_CASES pending the + # shared nested-dispatch-keying fix (see that note). + "039-nested-lineage-augmentation", } ) @@ -127,10 +125,24 @@ # ``(fixture_stem, case_name)``. The case-loop in the runner ``continue``s # past matching cases — NOT ``pytest.skip``, which would skip the whole # fixture's test invocation and hide the surrounding cases that DO run. -# Currently empty; the harness covers every activated case. Kept as a -# named hook so future per-case deferrals don't need to re-introduce the -# pattern. -_DEFERRED_CASES: frozenset[tuple[str, str]] = frozenset() +_DEFERRED_CASES: frozenset[tuple[str, str]] = frozenset( + { + # 039 cases 1 + 2 are TEMPORARILY deferred pending one deeper observer + # fix shared by both: dispatch keys + # (fan_out_instance_observations / parallel_branches_branch_spans) are + # namespace-local and do NOT encode the enclosing fan-out instance, so a + # dispatch INSIDE an outer fan-out instance collides across instances -- + # case 1's inner instance dispatch and case 2's per-branch dispatch both + # reparent the second outer instance's events under the first's dispatch. + # The fix (thread the enclosing fan_out_index_chain / branch_name_chain + # into the dispatch keys, across synthesis + resolution + the + # augmentation walk, in both observers) is its own focused effort + spec + # coordination. Case 3 (single fan-out level under a serial wrapper) does + # not nest dispatches, so it is wired. See _build_039_graph. + ("039-nested-lineage-augmentation", "inner_fan_out_augmenter_propagates_to_outer_dispatch_span"), + ("039-nested-lineage-augmentation", "parallel_branch_augmenter_propagates_to_outer_fan_out_instance"), + } +) # Mocks the spec fixture 037 references for ``trace_input_from_state`` / @@ -469,11 +481,11 @@ async def test_langfuse_fixture(fixture_path: Path) -> None: if fixture_inner_subgraphs is not None and "inner_subgraphs" not in case: case["inner_subgraphs"] = fixture_inner_subgraphs try: - await _run_case(case) + await _run_case(case, fixture_stem=fixture_stem) except AssertionError as e: raise AssertionError(f"case {case_name!r}: {e}") from e else: - await _run_case(spec) + await _run_case(spec, fixture_stem=fixture_stem) def _has_topology_constructs(case: Mapping[str, Any]) -> bool: @@ -640,6 +652,151 @@ def _build_inner_subgraph_with_llm( return builder.compile() +# Fixture 039 (nested-lineage augmentation) declares nested fan-out graphs the +# generic cross-cap adapter can't construct (a fan-out inside a subgraph wrapper +# / another fan-out, and a per-item sub-field as the inner fan-out's items +# source). Each case is hand-built here against the engine's GraphBuilder -- +# mirroring the dedicated 044 builder on the OTel side -- then driven through the +# shared observer + assertion path. The expected langfuse_trace in the YAML +# stays the oracle. +_FIXTURE_039 = "039-nested-lineage-augmentation" + + +def _build_039_graph( + case: Mapping[str, Any], + *, + provider: OpenAIProvider | None, + prompt_manager: PromptManager | None, +) -> tuple[Any, Any]: + """Dispatch a 039 case to its hand-built graph; return (graph, factory).""" + name = cast("str", case.get("name")) + if name == "fan_out_in_serial_subgraph_augmenter_propagates_to_wrapper_span": + return _build_039_case3(case, provider=provider, prompt_manager=prompt_manager) + raise NotImplementedError(f"039 case not yet wired: {name!r}") + + +def _build_039_case3( + case: Mapping[str, Any], + *, + provider: OpenAIProvider | None, + prompt_manager: PromptManager | None, +) -> tuple[Any, Any]: + # Case 3: a serial subgraph wrapper (`wrap`) descends into `wrapped_fan_out`, + # whose `pick` fan-out runs per-product; each instance augments note=. + # The wrapper span must carry the augmentation (last-writer) per 0045's + # lineage-aware rule, the fan-out NODE must not. + # The fan-out places each outer product into per_product's item_field slot; + # the augment middleware reads from it. per_product's declared state + # ({picked}) lacks the slot, so inject it (mirrors _synthesize_fan_out_ + # aggregation on the generic 029 path). + assert provider is not None, "039 cases declare mock_llm, so the provider must be set" + per_product_spec = copy.deepcopy(cast("dict[str, Any]", case["inner_subgraphs"]["per_product"])) + per_product_spec.setdefault("state", {}).setdefault("fields", {}).setdefault( + "oa_fan_out_item", {"type": "dict", "default": {}} + ) + per_product = _build_inner_subgraph_with_llm( + per_product_spec, + provider=provider, + prompt_manager=prompt_manager, + render_variables={}, + ) + wrap_state_cls = build_state_cls( + "Wrapped039C3", + { + "picks": {"type": "list", "reducer": "append", "default": []}, + "products": {"type": "list", "default": []}, + "oa_fan_out_item": {"type": "dict", "default": {}}, + }, + ) + wrap_builder: GraphBuilder[Any] = GraphBuilder(wrap_state_cls) + wrap_builder.set_entry("pick") + wrap_builder.add_fan_out_node( + "pick", + subgraph=per_product, + items_field="products", + item_field="oa_fan_out_item", + collect_field="picked", + target_field="picks", + instance_middleware=[_make_augment_instance_middleware({"note": "id"}, "oa_fan_out_item")], + ) + wrap_builder.add_edge("pick", END) + wrapped_fan_out = wrap_builder.compile() + + outer_state_cls = build_state_cls( + "Outer039C3", + {"result": {"type": "list", "default": []}, "products": {"type": "list", "default": []}}, + ) + outer_builder: GraphBuilder[Any] = GraphBuilder(outer_state_cls) + outer_builder.set_entry("wrap") + outer_builder.add_subgraph_node( + "wrap", + wrapped_fan_out, + ExplicitMapping(inputs={"products": "products"}, outputs={"result": "picks"}), + ) + outer_builder.add_edge("wrap", END) + graph = outer_builder.compile() + initial = cast("dict[str, Any]", case.get("initial_state") or {}) + return graph, (lambda: outer_state_cls(**initial)) + + +# Proposal 0045 §3.4: a key set via set_invocation_metadata inside a fan-out +# instance / parallel-branches branch lands ONLY on the dispatch ancestors on +# the augmenter's call-stack path -- NOT on the shared fan-out/pb NODE, sibling +# instances, or (inside a dispatch) the Trace. The tree asserter is subset-based +# (extra keys tolerated), which can't catch a MUST-NOT violation, so 039 +# additionally enforces that augmented keys absent from an observation's expected +# metadata are absent in the actual. Scoped to 039 via this ContextVar so the +# established subset semantics for the other fixtures are unchanged. +_AUGMENT_KEYS_UNDER_TEST: ContextVar[frozenset[str]] = ContextVar( + "augment_keys_under_test", default=frozenset() +) + + +def _collect_augment_keys(case: Mapping[str, Any]) -> frozenset[str]: + """Collect the metadata keys augment directives set, anywhere in the case's + topology (fan-out / parallel-branches augment blocks at any nesting).""" + keys: set[str] = set() + directives = ("augment_metadata_from_field", "augment_metadata_from_outer_item", "augment_metadata") + + def _harvest(block: Any) -> None: + if not isinstance(block, dict): + return + for directive in directives: + mapping = cast("dict[str, Any]", block).get(directive) + if isinstance(mapping, dict): + keys.update(cast("dict[str, Any]", mapping).keys()) + + def _walk(spec: Mapping[str, Any]) -> None: + for node in cast("dict[str, Any]", spec.get("nodes") or {}).values(): + if not isinstance(node, dict): + continue + node_dict = cast("dict[str, Any]", node) + _harvest(node_dict.get("fan_out")) + pb = cast("dict[str, Any] | None", node_dict.get("parallel_branches")) + for branch in cast("dict[str, Any]", (pb or {}).get("branches") or {}).values(): + _harvest(branch) + for collection in ("subgraphs", "inner_subgraphs"): + for sub in cast("dict[str, Any]", spec.get(collection) or {}).values(): + if isinstance(sub, dict): + _walk(cast("Mapping[str, Any]", sub)) + + _walk(case) + return frozenset(keys) + + +def _assert_augment_keys_not_leaked( + label: str, actual: Mapping[str, Any], expected: Mapping[str, Any] +) -> None: + # Proposal 0045 §3.4 MUST-NOT: an augmented key absent from the expected + # metadata (a shared fan-out/pb NODE, a sibling, or the Trace inside a + # dispatch) MUST also be absent in the actual. Complements the subset matcher. + for key in _AUGMENT_KEYS_UNDER_TEST.get(): + if key not in expected: + assert key not in actual, ( + f"{label}: MUST NOT carry augmented key {key!r} (proposal 0045 §3.4); got {actual.get(key)!r}" + ) + + def _resolve_detached_wrapper_names(case: Mapping[str, Any]) -> frozenset[str]: """Translate fixture-level ``detached_subgraphs`` (a list of SUBGRAPH IDENTITY names) into the set of WRAPPER NODE names the observer keys @@ -662,7 +819,11 @@ def _resolve_detached_wrapper_names(case: Mapping[str, Any]) -> frozenset[str]: return frozenset(wrappers) -async def _run_case(case: Mapping[str, Any]) -> None: +async def _run_case(case: Mapping[str, Any], *, fixture_stem: str | None = None) -> None: + # 039 additionally enforces proposal 0045's MUST-NOT scoping (an augmented + # key absent from an observation's expected metadata must be absent in the + # actual); other fixtures keep the established subset semantics. + _AUGMENT_KEYS_UNDER_TEST.set(_collect_augment_keys(case) if fixture_stem == _FIXTURE_039 else frozenset()) # ---- Mock LLM transport (if the graph has an LLM call) mock_responses = cast("list[dict[str, Any]] | None", case.get("mock_llm")) transport = _build_mock_llm_handler(mock_responses) if mock_responses else None @@ -692,7 +853,13 @@ async def _run_case(case: Mapping[str, Any]) -> None: # ``adapter.build_graph`` machinery for subgraph / fan_out shapes; # LLM/prompt fixtures (022/023/024) use the simpler hand-rolled # per-node build that knows about ``calls_llm`` / ``renders_prompt``. - if _has_topology_constructs(case): + if fixture_stem == _FIXTURE_039: + # 039's nested fan-out graphs are hand-built (the generic adapter can't + # construct them); see _build_039_graph. + graph, initial_state_factory = _build_039_graph( + case, provider=provider, prompt_manager=prompt_manager + ) + elif _has_topology_constructs(case): # The topology fixtures (031/032/033) use inner-node test-seam # directives the cross-capability adapter doesn't translate # (``update_pure_from_state`` computes a value the assertions @@ -1325,6 +1492,7 @@ def _assert_trace( f"trace.metadata.invocation_id: raw trace.id {trace.id!r} != {expected_invocation_id!r}" ) _assert_metadata_subset("trace.metadata", trace.metadata, expected_metadata) + _assert_augment_keys_not_leaked("trace.metadata", trace.metadata, expected_metadata) # Proposal 0043 (§8.4.1 trace.input/output sourcing). Fixtures that # opt in supply these as YAML maps; older fixtures leave them absent. if "input" in expected: @@ -1452,6 +1620,9 @@ def _assert_observation( ) expected_metadata = cast("dict[str, Any]", expected.get("metadata") or {}) _assert_metadata_subset(f"observation[{actual.name}].metadata", actual.metadata, expected_metadata) + _assert_augment_keys_not_leaked( + f"observation[{actual.name}].metadata", actual.metadata, expected_metadata + ) expected_children = cast("list[dict[str, Any]]", expected.get("children") or []) actual_children = trace.children_of(actual.id)