From 529a66825b607c6d01b0e611aa33e9b22f9eac9a Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 25 Jun 2026 15:25:37 -0700 Subject: [PATCH 1/3] Fix fan-out dispatch and augmentation scoping Two observer corrections surfaced while wiring the nested-lineage conformance fixture (proposal 0045): - The augmentation walk's same-namespace arm now skips shared-parent fan-out / parallel-branches NODE spans, in both the OTel and Langfuse observers. A key set via set_invocation_metadata inside a fan-out instance or branch was wrongly applied to the shared fork NODE when the augmenting context executed at that node's own namespace, in addition to the dispatch span where it belongs. This violated the observability 3.4 MUST-NOT; the strict-ancestor arm already skipped shared parents. Per-instance / per-branch dispatch spans and lineage ancestors are unaffected. - The Langfuse per-instance fan-out dispatch synthesis and parent resolution are now prefix-general, so a fan-out nested below the top namespace level (inside a serial subgraph wrapper) gets its dispatch observation synthesized and its inner observations parented under it, matching the OTel observer. --- CHANGELOG.md | 2 + .../observability/langfuse/observer.py | 97 ++++++++++--------- .../observability/otel/observer.py | 23 ++--- 3 files changed, 63 insertions(+), 59 deletions(-) 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..e65f864 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,35 @@ 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. - if depth == 1 and prefix[0] in self.detached_subgraphs: + # Detached subgraph: detached_subgraphs holds bare node names, so + # match on prefix[-1] (the node-name segment) at any depth (at depth 1 + # this coincides with prefix[0], so the depth-1 behavior is unchanged). + if prefix[-1] 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. - if depth == 1 and event.fan_out_index is not None and prefix[0] in self.detached_fan_outs: + # 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. + if event.fan_out_index is not None and prefix[-1] 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 +1443,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): From 6c2cf60db7a58c7f17bb5d0f736d13f0f590f355 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 25 Jun 2026 15:26:38 -0700 Subject: [PATCH 2/3] Wire 039 case 3 conformance, defer cases 1+2 Activate the nested-lineage fixture (039) in the Langfuse conformance runner. Case 3 (a fan-out inside a serial subgraph wrapper) is built by a dedicated hand-built runner -- the generic adapter cannot construct nested fan-out graphs -- and asserted against the fixture's expected trace. A ContextVar-scoped negative check enforces proposal 0045's MUST-NOT: an augmented key absent from an observation's expected metadata must be absent in the actual, which the subset matcher alone cannot catch. Cases 1 and 2 are temporarily deferred via _DEFERRED_CASES. Both need a shared observer fix: dispatch keys do not encode the enclosing fan-out instance, so a dispatch inside an outer instance collides across instances. Tracked as a separate effort. --- .../test_observability_langfuse.py | 211 ++++++++++++++++-- 1 file changed, 191 insertions(+), 20 deletions(-) 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) From 4816a1a8ecee339bf275d7903aed61443875bc2d Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 25 Jun 2026 15:59:40 -0700 Subject: [PATCH 3/3] Keep detached dispatch synthesis top-level From CoPilot review of #193: the prefix-general change over-generalized the detached subgraph / fan-out synthesis arms to any depth, but _trace_id_for still routes detached events by namespace[:1]. A nested detached fan-out would partially detach -- its dispatch in the new Trace but inner nodes in the main one. Re-gate both detached arms to depth == 1; only the non-detached fan-out arm and the dedup need to be prefix-general (what case 3 exercises). A nested detached fan-out now gets no synthesis, consistent with the prior behavior, until the deferred nested-dispatch-keying fix generalizes _trace_id_for too. --- .../observability/langfuse/observer.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index e65f864..1858b2b 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -1023,16 +1023,19 @@ def _sync_subgraph_observations( and (prefix + (str(event.fan_out_index),)) in inv_state.fan_out_instance_observations ): continue - # Detached subgraph: detached_subgraphs holds bare node names, so - # match on prefix[-1] (the node-name segment) at any depth (at depth 1 - # this coincides with prefix[0], so the depth-1 behavior is unchanged). - if prefix[-1] in self.detached_subgraphs: + # 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; the detached dispatch goes into the new Trace. - if event.fan_out_index is not None and prefix[-1] in self.detached_fan_outs: + # 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 the per-instance dispatch