From cf2aa6f2a85fe42adb8dea0b4cadc9db4a8899b2 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 31 May 2026 13:29:44 -0700 Subject: [PATCH 1/2] Implement proposal 0045 (nested-lineage augmentation) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec proposal 0045 (observability §3.4) rewrites the per-async-context augmentation boundary as three lineage-aware rules: the augmenter's call-stack ancestor chain MUST update (every strict dispatch ancestor on the augmenter's specific path); siblings at any depth MUST NOT; shared parents (fan-out NODE, parallel-branches NODE, invocation span when inside a dispatch boundary) MUST NOT. Engine: tracks per-depth lineage chains parallel to namespace_prefix. descend_into_subgraph extends both chains by None; fan-out instance descent extends fan_out_index_chain with the instance index; pb branch descent extends branch_name_chain with the branch name. Chain ContextVars driven at every leaf-node execution site so set_invocation_metadata sees the full chain. Events: NodeEvent and MetadataAugmentationEvent grow fan_out_index_chain and branch_name_chain fields populated from context. Backwards-compatible — scalar fan_out_index and branch_name fields stay as innermost values. OTel and Langfuse observers: _collect_augmentation_targets and _handle_metadata_augmentation rewrite against the three-step boundary decision tree. Both observers store the chain on _OpenSpan / _OpenObservation so the comparison is local rather than re-derived per event. Shared-parent identification uses the parent_node_name caches structurally — pb cache mirror added to the Langfuse observer. Invocation span update: preserved per fixture 034 (outermost-serial case where the augmenter has no fan-out or pb dispatch on its call-stack path). When inside any dispatch boundary, the invocation span MUST NOT update per §3.4's shared-parent rule. Spec pin v0.36.0 → v0.37.0. Conformance fixture 039 stays deferred in the Langfuse harness — needs runtime-state item-list lookup for nested fan-outs plus a new augment_metadata_from_outer_item directive for case 2. Behavioral contract verified at unit level via test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span. --- CHANGELOG.md | 1 + conformance.toml | 2 +- openarmature-spec | 2 +- pyproject.toml | 2 +- src/openarmature/AGENTS.md | 4 +- src/openarmature/__init__.py | 2 +- src/openarmature/graph/compiled.py | 34 +++ src/openarmature/graph/events.py | 21 ++ src/openarmature/graph/observer.py | 42 +++- src/openarmature/graph/parallel_branches.py | 1 + src/openarmature/observability/correlation.py | 62 +++++ .../observability/langfuse/observer.py | 184 ++++++++++---- src/openarmature/observability/metadata.py | 7 + .../observability/otel/observer.py | 230 ++++++++++++++---- .../test_observability_langfuse.py | 25 ++ tests/test_smoke.py | 2 +- tests/unit/test_observability_otel.py | 142 +++++++++++ 17 files changed, 658 insertions(+), 105 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c405a41..ae5348d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ### Added +- **Nested-lineage augmentation containment scope** (proposal 0045, observability §3.4, spec v0.37.0). The per-async-context augmentation boundary rewrites as three lineage-aware rules: the augmenter's call-stack ancestor chain MUST update (every strict dispatch ancestor on the path — each outer fan-out instance dispatch span, each outer parallel-branches branch dispatch span, each outer serial subgraph wrapper); siblings at any depth MUST NOT; shared parents (fan-out NODE, parallel-branches NODE, invocation span) MUST NOT. Engine-side: tracks per-depth lineage chains (`fan_out_index_chain` / `branch_name_chain`) parallel to `namespace_prefix`, available on `NodeEvent` and `MetadataAugmentationEvent`. Observer-side: `OTelObserver._collect_augmentation_targets` and `LangfuseObserver._handle_metadata_augmentation` rewrite against the three-step boundary decision tree. Single-level behavior (fixtures 029 / 030 / 034) is unchanged. - **`LangfuseObserver` Trace input/output sourcing** (proposal 0043, observability §8.4.1). New observer construction knobs populate `trace.input` and `trace.output` per the three-lever decision tree: - **`disable_state_payload: bool = True`** — privacy knob symmetric to `disable_llm_payload`. When ON (default), Trace fields receive the minimal stub `{entry_node, correlation_id}` / `{final_node, status}`; when OFF, the raw state object is serialized. - **`trace_input_from_state` / `trace_output_from_state`** — optional caller hooks returning the domain-shaped value to use for `trace.input` / `trace.output`. Returning `None` falls through to the next applicable lever. diff --git a/conformance.toml b/conformance.toml index 981bcd7f..084b0359 100644 --- a/conformance.toml +++ b/conformance.toml @@ -32,7 +32,7 @@ [manifest] implementation = "openarmature-python" -spec_pin = "v0.36.0" +spec_pin = "v0.37.0" # Status values: # implemented — shipped behavior matches the proposal's contract diff --git a/openarmature-spec b/openarmature-spec index 3b21d88f..253bff28 160000 --- a/openarmature-spec +++ b/openarmature-spec @@ -1 +1 @@ -Subproject commit 3b21d88f490d473cb9baadafded0dd58ce244ba8 +Subproject commit 253bff28b13f4126a28054dee33ea833a987fa2c diff --git a/pyproject.toml b/pyproject.toml index 8b507dfa..46da2024 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,7 +58,7 @@ Specification = "https://github.com/LunarCommand/openarmature-spec" openarmature = "openarmature.cli:main" [tool.openarmature] -spec_version = "0.36.0" +spec_version = "0.37.0" [dependency-groups] dev = [ diff --git a/src/openarmature/AGENTS.md b/src/openarmature/AGENTS.md index 852ee55a..e9e5d7e2 100644 --- a/src/openarmature/AGENTS.md +++ b/src/openarmature/AGENTS.md @@ -1,6 +1,6 @@ # OpenArmature — Agent documentation -*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.36.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.* +*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.37.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.* ## TL;DR @@ -10,7 +10,7 @@ OpenArmature is a workflow framework for LLM pipelines and tool-calling agents ## Capability contracts -_Sourced from openarmature-spec v0.36.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._ +_Sourced from openarmature-spec v0.37.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._ ### Capability: `graph-engine` diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py index 4ef2698f..5e2b26dd 100644 --- a/src/openarmature/__init__.py +++ b/src/openarmature/__init__.py @@ -25,4 +25,4 @@ """ __version__ = "0.10.0" -__spec_version__ = "0.36.0" +__spec_version__ = "0.37.0" diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index eacac32d..80b97c2a 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -64,15 +64,19 @@ from openarmature.observability.correlation import ( _reset_active_dispatch, _reset_active_observers, + _reset_branch_name_chain, _reset_correlation_id, _reset_fan_out_index, + _reset_fan_out_index_chain, _reset_invocation_id, _reset_namespace_prefix, _set_active_dispatch, _set_active_observer_span, _set_active_observers, + _set_branch_name_chain, _set_correlation_id, _set_fan_out_index, + _set_fan_out_index_chain, _set_invocation_id, _set_namespace_prefix, current_active_observer_span, @@ -1438,6 +1442,11 @@ async def innermost(s: Any) -> Mapping[str, Any]: dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) namespace_token = _set_namespace_prefix(namespace) fan_out_token = _set_fan_out_index(context.fan_out_index) + # Per proposal 0045 (v0.37.0): drive the per-depth chain + # ContextVars from the context so ``set_invocation_metadata`` + # sees the full lineage chain at augmentation time. + fan_out_chain_token = _set_fan_out_index_chain(context.fan_out_index_chain) + branch_chain_token = _set_branch_name_chain(context.branch_name_chain) try: try: final_partial = await chain(state) @@ -1459,6 +1468,8 @@ async def innermost(s: Any) -> Mapping[str, Any]: # the chain unrecovered. Wrap as NodeException per §4. raise NodeException(node_name=current, cause=e, recoverable_state=state) from e finally: + _reset_branch_name_chain(branch_chain_token) + _reset_fan_out_index_chain(fan_out_chain_token) _reset_fan_out_index(fan_out_token) _reset_namespace_prefix(namespace_token) _reset_active_dispatch(dispatch_token) @@ -1579,6 +1590,9 @@ async def innermost(s: Any) -> Mapping[str, Any]: dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) namespace_token = _set_namespace_prefix(namespace) fan_out_token = _set_fan_out_index(context.fan_out_index) + # Per proposal 0045: drive per-depth chain ContextVars. + fan_out_chain_token = _set_fan_out_index_chain(context.fan_out_index_chain) + branch_chain_token = _set_branch_name_chain(context.branch_name_chain) try: try: @@ -1593,6 +1607,8 @@ async def innermost(s: Any) -> Mapping[str, Any]: # preserved. raise NodeException(node_name=current, cause=e, recoverable_state=state) from e finally: + _reset_branch_name_chain(branch_chain_token) + _reset_fan_out_index_chain(fan_out_chain_token) _reset_fan_out_index(fan_out_token) _reset_namespace_prefix(namespace_token) _reset_active_dispatch(dispatch_token) @@ -1868,6 +1884,9 @@ async def innermost(s: Any) -> Mapping[str, Any]: dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) namespace_token = _set_namespace_prefix(namespace) fan_out_token = _set_fan_out_index(context.fan_out_index) + # Per proposal 0045: drive per-depth chain ContextVars. + fan_out_chain_token = _set_fan_out_index_chain(context.fan_out_index_chain) + branch_chain_token = _set_branch_name_chain(context.branch_name_chain) # Per spec §10.11 the ``fan_out_progress`` entry is "in-flight # only"; the fan-out's own completion save below is the last # point where the entry is needed (proposal 0009: that save @@ -1895,6 +1914,8 @@ async def innermost(s: Any) -> Mapping[str, Any]: except Exception as e: raise NodeException(node_name=current, cause=e, recoverable_state=state) from e finally: + _reset_branch_name_chain(branch_chain_token) + _reset_fan_out_index_chain(fan_out_chain_token) _reset_fan_out_index(fan_out_token) _reset_namespace_prefix(namespace_token) _reset_active_dispatch(dispatch_token) @@ -2084,6 +2105,9 @@ async def innermost(s: Any) -> Mapping[str, Any]: dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) namespace_token = _set_namespace_prefix(namespace) fan_out_token = _set_fan_out_index(context.fan_out_index) + # Per proposal 0045: drive per-depth chain ContextVars. + fan_out_chain_token = _set_fan_out_index_chain(context.fan_out_index_chain) + branch_chain_token = _set_branch_name_chain(context.branch_name_chain) try: try: final_partial = await chain(state) @@ -2092,6 +2116,8 @@ async def innermost(s: Any) -> Mapping[str, Any]: except Exception as e: raise NodeException(node_name=current, cause=e, recoverable_state=state) from e finally: + _reset_branch_name_chain(branch_chain_token) + _reset_fan_out_index_chain(fan_out_chain_token) _reset_fan_out_index(fan_out_token) _reset_namespace_prefix(namespace_token) _reset_active_dispatch(dispatch_token) @@ -2173,6 +2199,11 @@ def _dispatch_started( fan_out_config=fan_out_config, parallel_branches_config=parallel_branches_config, branch_name=current_branch_name(), + # Per proposal 0045: per-depth lineage chains so + # observers can identify the augmenter's call-stack + # ancestor path under nested dispatch. + fan_out_index_chain=context.fan_out_index_chain, + branch_name_chain=context.branch_name_chain, subgraph_identities=context.subgraph_identities, caller_invocation_metadata=current_invocation_metadata(), ), @@ -2210,6 +2241,9 @@ def _dispatch_completed( fan_out_config=fan_out_config, parallel_branches_config=parallel_branches_config, branch_name=current_branch_name(), + # Per proposal 0045: per-depth lineage chains. + fan_out_index_chain=context.fan_out_index_chain, + branch_name_chain=context.branch_name_chain, subgraph_identities=context.subgraph_identities, caller_invocation_metadata=current_invocation_metadata(), ), diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index 9ee14268..aa145ca4 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -257,6 +257,17 @@ class NodeEvent: # simultaneously when a branch's subgraph contains a fan-out # (and vice versa). branch_name: str | None = None + # Per proposal 0045 (v0.37.0): per-depth lineage chains parallel + # to ``namespace``. Position ``i`` is the fan_out_index (or + # branch_name) at the dispatch boundary leading to namespace + # depth ``i+1`` — or ``None`` when that boundary is a subgraph + # wrapper (not a fan-out, not a parallel-branches branch). + # ``fan_out_index`` and ``branch_name`` above carry the + # INNERMOST values; the chains carry the full lineage so + # observers can apply the §3.4 lineage-aware boundary rule + # without re-deriving it from successive events. + fan_out_index_chain: tuple[int | None, ...] = () + branch_name_chain: tuple[str | None, ...] = () # Per observability §5.3 + the coord-thread # ``clarify-subgraph-name-semantics`` resolution: chain of # compiled-subgraph identities parallel to the wrapper-depth @@ -328,6 +339,16 @@ class MetadataAugmentationEvent: attempt_index: int = 0 fan_out_index: int | None = None branch_name: str | None = None + # Per proposal 0045 (v0.37.0): the augmenter's per-depth lineage + # chain. Two parallel tuples indexed by namespace position — + # position ``i`` is the fan_out_index (or branch_name) at + # namespace depth ``i+1``, or ``None`` if that depth's dispatch + # boundary is not a fan-out instance (not a parallel-branches + # branch). Required by §3.4's lineage-aware boundary rule so + # observers can identify the augmenter's call-stack ancestor + # chain rather than only the innermost dispatch. + fan_out_index_chain: tuple[int | None, ...] = () + branch_name_chain: tuple[str | None, ...] = () # Spec: realizes observability §8.4.1 *Trace input/output sourcing* diff --git a/src/openarmature/graph/observer.py b/src/openarmature/graph/observer.py index ec8a170f..da5fc533 100644 --- a/src/openarmature/graph/observer.py +++ b/src/openarmature/graph/observer.py @@ -416,6 +416,18 @@ class _InvocationContext: # instance, and absent (None) for nodes outside any fan-out. fan_out_index: int | None = None + # Per proposal 0045 (v0.37.0): per-depth lineage chains. Mirror + # ``namespace_prefix`` depth — position ``i`` is the + # fan_out_index (resp. branch_name) for the dispatch boundary + # at namespace depth ``i+1``, or ``None`` if that boundary is a + # subgraph wrapper / serial node (not a fan-out, not a + # parallel-branches branch). The chains are extended by one + # entry at every ``descend_into_*`` call; the engine then drives + # the chain ContextVars from these fields at every node-execution + # site so ``set_invocation_metadata`` sees the full chain. + fan_out_index_chain: tuple[int | None, ...] = () + branch_name_chain: tuple[str | None, ...] = () + # ---------------------------------------------------------------- # Checkpointing fields (spec pipeline-utilities §10) # @@ -551,6 +563,11 @@ def descend_into_subgraph( parent_states_prefix=self.parent_states_prefix + (parent_state,), subgraph_identities=self.subgraph_identities + (subgraph_identity,), fan_out_index=self.fan_out_index, + # Per proposal 0045: subgraph wrappers don't add a + # fan-out or branch axis — extend both chains by + # ``None`` at this depth. + fan_out_index_chain=self.fan_out_index_chain + (None,), + branch_name_chain=self.branch_name_chain + (None,), invocation_id=self.invocation_id, correlation_id=self.correlation_id, checkpointer=self.checkpointer, @@ -602,6 +619,11 @@ def descend_into_fan_out_instance( parent_states_prefix=self.parent_states_prefix + (parent_state,), subgraph_identities=self.subgraph_identities + (subgraph_identity,), fan_out_index=fan_out_index, + # Per proposal 0045: fan-out instance descent extends the + # fan_out_index chain with the instance's index; the + # branch chain extends with ``None`` (no branch axis here). + fan_out_index_chain=self.fan_out_index_chain + (fan_out_index,), + branch_name_chain=self.branch_name_chain + (None,), invocation_id=self.invocation_id, correlation_id=self.correlation_id, checkpointer=self.checkpointer, @@ -626,6 +648,8 @@ def descend_into_parallel_branch( parallel_branches_node_name: str, parent_state: State, sub_attached: tuple[SubscribedObserver, ...], + *, + branch_name: str, ) -> _InvocationContext: """Build the context for one parallel-branches branch's subgraph invocation. @@ -637,11 +661,13 @@ def descend_into_parallel_branch( inner events nest under it (mirrors ``descend_into_fan_out_instance``'s namespace stamping). - Branch identity lives on the - ``observability.correlation._branch_name_var`` ContextVar - rather than on the descend context; set inside the - branch's task closure so ``copy_context`` inherits it - through the subgraph's execution. + Branch identity (the SCALAR innermost branch_name) lives on + the ``observability.correlation._branch_name_var`` ContextVar + — set inside the branch's task closure so ``copy_context`` + inherits it through the subgraph's execution. The PER-DEPTH + ``branch_name_chain`` (proposal 0045) is extended here on the + context so the engine can drive the chain ContextVar at + every inner-node execution site. Per §11.9 / §10.7 atomic-restart: drops the checkpointer and pending_resume_states (a crash mid-dispatch re-runs the @@ -662,6 +688,12 @@ def descend_into_parallel_branch( # needed) is a future addition. subgraph_identities=self.subgraph_identities + (None,), fan_out_index=self.fan_out_index, + # Per proposal 0045: parallel-branches branch descent + # extends the branch chain with this branch's name; the + # fan_out_index chain extends with ``None`` (no fan-out + # axis here). + fan_out_index_chain=self.fan_out_index_chain + (None,), + branch_name_chain=self.branch_name_chain + (branch_name,), invocation_id=self.invocation_id, correlation_id=self.correlation_id, checkpointer=None, diff --git a/src/openarmature/graph/parallel_branches.py b/src/openarmature/graph/parallel_branches.py index 24c4f6fb..b33ec018 100644 --- a/src/openarmature/graph/parallel_branches.py +++ b/src/openarmature/graph/parallel_branches.py @@ -161,6 +161,7 @@ async def run_branch(branch_name: str, spec: BranchSpec[Any]) -> Mapping[str, An parallel_branches_node_name=self.name, parent_state=state, sub_attached=tuple(spec.subgraph._attached_observers), # noqa: SLF001 + branch_name=branch_name, ) async def innermost(s: Any) -> Mapping[str, Any]: diff --git a/src/openarmature/observability/correlation.py b/src/openarmature/observability/correlation.py index 1ea67a2b..20174426 100644 --- a/src/openarmature/observability/correlation.py +++ b/src/openarmature/observability/correlation.py @@ -363,6 +363,62 @@ def _reset_branch_name(token: Token[str | None]) -> None: _branch_name_var.reset(token) +# Per proposal 0045 (v0.37.0): the per-depth lineage chain. Mirrors +# the namespace_prefix's depth: position N is the fan_out_index (or +# branch_name) at namespace depth N+1, or None if that depth's +# dispatch boundary is not a fan-out instance (not a parallel- +# branches branch). Required by the augmentation containment rule +# (§3.4) so observers can identify the augmenter's call-stack +# ancestor chain rather than only the innermost dispatch. The +# scalar ``_fan_out_index_var`` / ``_branch_name_var`` above are +# kept (innermost values) so existing callers continue to work. +_fan_out_index_chain_var: ContextVar[tuple[int | None, ...]] = ContextVar( + "openarmature.fan_out_index_chain", default=() +) + + +def current_fan_out_index_chain() -> tuple[int | None, ...]: + """Return the per-depth fan_out_index chain for the currently + executing node, or the empty tuple outside any node body. Each + position corresponds to a depth in ``current_namespace_prefix()``; + a non-``None`` entry at position ``i`` means the dispatch boundary + at depth ``i+1`` is a fan-out instance with that index. + """ + return _fan_out_index_chain_var.get() + + +def _set_fan_out_index_chain(value: tuple[int | None, ...]) -> Token[tuple[int | None, ...]]: + """Set the chain. Internal — engine-only.""" + return _fan_out_index_chain_var.set(value) + + +def _reset_fan_out_index_chain(token: Token[tuple[int | None, ...]]) -> None: + _fan_out_index_chain_var.reset(token) + + +_branch_name_chain_var: ContextVar[tuple[str | None, ...]] = ContextVar( + "openarmature.branch_name_chain", default=() +) + + +def current_branch_name_chain() -> tuple[str | None, ...]: + """Return the per-depth branch_name chain. Mirrors + ``current_fan_out_index_chain()`` — a non-``None`` entry at + position ``i`` means the dispatch boundary at depth ``i+1`` is a + parallel-branches branch with that name. + """ + return _branch_name_chain_var.get() + + +def _set_branch_name_chain(value: tuple[str | None, ...]) -> Token[tuple[str | None, ...]]: + """Set the chain. Internal — engine-only.""" + return _branch_name_chain_var.set(value) + + +def _reset_branch_name_chain(token: Token[tuple[str | None, ...]]) -> None: + _branch_name_chain_var.reset(token) + + _attempt_index_var: ContextVar[int] = ContextVar("openarmature.attempt_index", default=0) @@ -456,9 +512,11 @@ def _reset_active_observer_span(token: Token[object | None]) -> None: "current_active_observers", "current_attempt_index", "current_branch_name", + "current_branch_name_chain", "current_correlation_id", "current_dispatch", "current_fan_out_index", + "current_fan_out_index_chain", "current_invocation_id", "current_namespace_prefix", "validate_invocation_id", @@ -471,8 +529,10 @@ def _reset_active_observer_span(token: Token[object | None]) -> None: "_reset_active_observers", "_reset_attempt_index", "_reset_branch_name", + "_reset_branch_name_chain", "_reset_correlation_id", "_reset_fan_out_index", + "_reset_fan_out_index_chain", "_reset_invocation_id", "_reset_namespace_prefix", "_set_active_dispatch", @@ -480,8 +540,10 @@ def _reset_active_observer_span(token: Token[object | None]) -> None: "_set_active_observers", "_set_attempt_index", "_set_branch_name", + "_set_branch_name_chain", "_set_correlation_id", "_set_fan_out_index", + "_set_fan_out_index_chain", "_set_invocation_id", "_set_namespace_prefix", ] diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 8dafe433..c150dfc6 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -34,7 +34,7 @@ MetadataAugmentationEvent, NodeEvent, ) -from openarmature.observability.lineage import is_prefix_or_equal, is_strict_prefix +from openarmature.observability.lineage import is_strict_prefix from openarmature.observability.llm_event import LLM_NAMESPACE, LlmEventPayload from .client import ( @@ -77,9 +77,39 @@ def _read_spec_version() -> str: @dataclass class _OpenObservation: - """An in-flight Langfuse observation pinned in the observer's state.""" + """An in-flight Langfuse observation pinned in the observer's state. + + Per proposal 0045: carries the observation's own + ``fan_out_index_chain`` and ``branch_name_chain`` so the + augmentation walk can apply §3.4's lineage-aware boundary rule + (mirror of the OTel observer's ``_OpenSpan``).""" handle: LangfuseSpanHandle | LangfuseGenerationHandle + fan_out_index_chain: tuple[int | None, ...] = () + branch_name_chain: tuple[str | None, ...] = () + + +def _observation_chain_on_path( + open_obs: _OpenObservation, + aug_fi_chain: tuple[int | None, ...], + aug_bn_chain: tuple[str | None, ...], +) -> bool: + """Mirror of the OTel observer's ``_span_chain_on_path`` for + Langfuse observations. Returns True iff the observation's chain + is a prefix-match of the augmenter's chain.""" + obs_fi = open_obs.fan_out_index_chain + obs_bn = open_obs.branch_name_chain + if len(obs_fi) > len(aug_fi_chain): + return False + if len(obs_bn) > len(aug_bn_chain): + return False + for i in range(len(obs_fi)): + if obs_fi[i] != aug_fi_chain[i]: + return False + for i in range(len(obs_bn)): + if obs_bn[i] != aug_bn_chain[i]: + return False + return True def _empty_str_frozenset() -> frozenset[str]: @@ -185,6 +215,14 @@ class _InvState: # per-instance dispatch observation can attach # metadata.fan_out_parent_node_name). fan_out_parent_node_name: dict[tuple[str, ...], str] = field(default_factory=dict[tuple[str, ...], str]) + # Per proposal 0045: structural identification of parallel- + # branches NODE namespaces. Populated on a pb NODE's started + # event (whichever events carry ``parallel_branches_config``); + # consulted by the augmentation walk to skip the pb NODE itself + # as a shared parent (§3.4's structural classification). + parallel_branches_parent_node_name: dict[tuple[str, ...], str] = field( + default_factory=dict[tuple[str, ...], str] + ) # Side-cache: accumulator for `metadata.detached_child_trace_ids` # on dispatch observations that spawn detached children. Keyed by # the dispatch observation's prefix (the fan-out node's namespace, @@ -336,6 +374,16 @@ def _open_started_observation(self, event: NodeEvent) -> None: if event.fan_out_config is not None and event.fan_out_index is None: inv_state.fan_out_parent_node_name[event.namespace] = event.fan_out_config.parent_node_name + # Per proposal 0045: mirror cache for parallel-branches NODE + # identification (used by the augmentation shared-parent + # check). No additional ``branch_name is None`` filter — the + # ``*_config`` field is itself only populated on a NODE's own + # events. + if event.parallel_branches_config is not None: + inv_state.parallel_branches_parent_node_name[event.namespace] = ( + event.parallel_branches_config.parent_node_name + ) + key = self._key_for(event) if key in inv_state.open_observations: # Idempotent: a second started for the same (namespace, @@ -357,7 +405,11 @@ def _open_started_observation(self, event: NodeEvent) -> None: metadata=metadata, parent_observation_id=parent_observation_id, ) - inv_state.open_observations[key] = _OpenObservation(handle=handle) + inv_state.open_observations[key] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) def _handle_completed(self, event: NodeEvent) -> None: from openarmature.observability.correlation import current_invocation_id @@ -397,6 +449,10 @@ def _handle_completed(self, event: NodeEvent) -> None: # rather than appending to the previous iteration's # accumulator and overwriting the prior link metadata. inv_state.detached_child_trace_ids.pop(event.namespace, None) + # Per proposal 0045: clean up the pb cache on a pb NODE's own + # completion. Same shape as the fan-out cleanup above. + if event.parallel_branches_config is not None: + inv_state.parallel_branches_parent_node_name.pop(event.namespace, None) key = self._key_for(event) observation = inv_state.open_observations.pop(key, None) @@ -446,53 +502,68 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non if invocation_id is None or not event.entries: return inv_state = self._inv_states.get(invocation_id) - aug_fi = event.fan_out_index - aug_bn = event.branch_name aug_ns = event.namespace + aug_fi_chain = event.fan_out_index_chain + aug_bn_chain = event.branch_name_chain metadata_delta = dict(event.entries) - # Trace.metadata: only for outermost-serial. The Trace is - # shared across siblings, so a fan-out instance's per-item - # productId would leak across siblings if it updated the - # Trace — 029's fixture explicitly rules that out. - if aug_fi is None and aug_bn is None: + # Trace.metadata: only when the augmenter sits in OUTERMOST + # SERIAL context (no fan-out instance and no parallel-branches + # branch on its call-stack path). Per §3.4 the Trace is a + # shared parent inside any dispatch boundary — siblings would + # leak — so only the no-dispatch-on-path case writes. + outermost_serial = all(fi is None for fi in aug_fi_chain) and all(bn is None for bn in aug_bn_chain) + if outermost_serial: self.client.update_trace(id=invocation_id, metadata=metadata_delta) if inv_state is None: return - # Subgraph wrapper observations on the ancestor path - # (outermost-serial only — inside a fan-out instance the - # subgraph wrapper above the fan-out node is sibling-shared). - if aug_fi is None and aug_bn is None: - for prefix, observation in inv_state.subgraph_observations.items(): - if is_strict_prefix(prefix, aug_ns): - observation.handle.update(metadata=metadata_delta) + # Per proposal 0045: parallel walk of the OTel observer's + # _collect_augmentation_targets — subgraph wrappers on the + # call-stack path (chain prefix-matches), fan-out instance + # dispatch observations whose dispatch position matches the + # augmenter's chain, and open NODE observations on the path + # (skipping fan-out / pb shared-parent NODEs). - # Fan-out instance dispatch observation(s) when the augmenter - # is inside a fan-out instance. Keys are anchor_ns + - # (str(fan_out_index),) per - # ``_open_fan_out_instance_dispatch_observation``. - if aug_fi is not None: - fi_str = str(aug_fi) - for key, observation in inv_state.fan_out_instance_observations.items(): - if not key or key[-1] != fi_str: - continue - anchor_ns = key[:-1] - if is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns: - observation.handle.update(metadata=metadata_delta) + # Subgraph wrapper observations on the path. + for prefix, observation in inv_state.subgraph_observations.items(): + if not is_strict_prefix(prefix, aug_ns): + continue + if _observation_chain_on_path(observation, aug_fi_chain, aug_bn_chain): + observation.handle.update(metadata=metadata_delta) - # Open node observations on the augmenter's call stack. Match - # on ``fan_out_index`` AND ``branch_name`` to skip sibling - # instances / branches; namespace must prefix (or equal) the - # augmenter's. + # Fan-out instance dispatch observations. + for key, observation in inv_state.fan_out_instance_observations.items(): + if not key: + continue + anchor_ns = key[:-1] + fi_str = key[-1] + if not (is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns): + continue + chain_pos = len(anchor_ns) - 1 + if chain_pos < 0 or chain_pos >= len(aug_fi_chain): + continue + aug_fi_at_pos = aug_fi_chain[chain_pos] + if aug_fi_at_pos is None or str(aug_fi_at_pos) != fi_str: + continue + observation.handle.update(metadata=metadata_delta) + + # Open NODE observations. Same as augmenter or strict + # ancestor on the path; skip shared-parent NODE observations + # (fan-out NODE / pb NODE) identified by presence in the + # parent_node_name caches. for key, observation in inv_state.open_observations.items(): - ns, _ai, fi, bn = key - if fi != aug_fi: + 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): continue - if bn != aug_bn: + if ns in inv_state.fan_out_parent_node_name or ns in inv_state.parallel_branches_parent_node_name: continue - if is_prefix_or_equal(ns, aug_ns): + if _observation_chain_on_path(observation, aug_fi_chain, aug_bn_chain): observation.handle.update(metadata=metadata_delta) # ------------------------------------------------------------------ @@ -810,7 +881,13 @@ def _open_subgraph_observation( metadata=metadata, parent_observation_id=parent_observation_id, ) - inv_state.subgraph_observations[prefix] = _OpenObservation(handle=handle) + # Per proposal 0045: chain sliced to wrapper depth. + chain_len = len(prefix) + inv_state.subgraph_observations[prefix] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) def _open_fan_out_instance_dispatch_observation( self, @@ -845,7 +922,13 @@ def _open_fan_out_instance_dispatch_observation( parent_observation_id=parent_observation_id, ) instance_key = prefix + (str(event.fan_out_index),) - inv_state.fan_out_instance_observations[instance_key] = _OpenObservation(handle=handle) + # Per proposal 0045: chain sliced to instance-dispatch depth. + chain_len = len(prefix) + inv_state.fan_out_instance_observations[instance_key] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) def _open_detached_subgraph_trace( self, @@ -963,7 +1046,15 @@ def _open_detached_subgraph_trace( metadata=dispatch_metadata, parent_observation_id=None, ) - inv_state.subgraph_observations[prefix] = _OpenObservation(handle=handle) + # Per proposal 0045: detached subgraph wrapper sits in its own + # trace; chain still mirrors the parent-trace path so the + # augmentation lookup is consistent with non-detached. + chain_len = len(prefix) + inv_state.subgraph_observations[prefix] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) inv_state.detached_traces[prefix] = detached_trace_id def _open_detached_fan_out_instance_trace( @@ -1040,7 +1131,12 @@ def _open_detached_fan_out_instance_trace( parent_observation_id=None, ) instance_key = prefix + (str(event.fan_out_index),) - inv_state.fan_out_instance_observations[instance_key] = _OpenObservation(handle=handle) + chain_len = len(prefix) + inv_state.fan_out_instance_observations[instance_key] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) inv_state.detached_traces[instance_key] = detached_trace_id inv_state.fan_out_instance_root_prefixes.add(instance_key) @@ -1213,7 +1309,11 @@ def _handle_llm_event(self, event: NodeEvent) -> None: parent_observation_id=parent_observation_id, prompt=self._resolve_prompt_link(payload), ) - inv_state.open_llm_observations[payload.call_id] = _OpenObservation(handle=handle) + inv_state.open_llm_observations[payload.call_id] = _OpenObservation( + handle=handle, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) return # completed: pop the started handle and finalize. diff --git a/src/openarmature/observability/metadata.py b/src/openarmature/observability/metadata.py index fc3f2d1f..090b40e3 100644 --- a/src/openarmature/observability/metadata.py +++ b/src/openarmature/observability/metadata.py @@ -168,8 +168,10 @@ def set_invocation_metadata(**entries: AttributeValue) -> None: from .correlation import ( current_attempt_index, current_branch_name, + current_branch_name_chain, current_dispatch, current_fan_out_index, + current_fan_out_index_chain, current_namespace_prefix, ) @@ -182,6 +184,11 @@ def set_invocation_metadata(**entries: AttributeValue) -> None: attempt_index=current_attempt_index(), fan_out_index=current_fan_out_index(), branch_name=current_branch_name(), + # Per proposal 0045: the per-depth chains let observers walk + # the augmenter's call-stack ancestor path rather than only + # seeing the innermost dispatch. + fan_out_index_chain=current_fan_out_index_chain(), + branch_name_chain=current_branch_name_chain(), ) dispatch(event) diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index 549bd623..38a853ec 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -102,7 +102,7 @@ MetadataAugmentationEvent, NodeEvent, ) -from openarmature.observability.lineage import is_prefix_or_equal, is_strict_prefix +from openarmature.observability.lineage import is_strict_prefix from openarmature.observability.llm_event import LLM_NAMESPACE, LlmEventPayload # Span-stack key shape: @@ -202,9 +202,47 @@ class _OpenSpan: """An in-flight span. No OTel context token: the new architecture resolves parents from the observer's internal maps within a single event handler's scope, so no token needs to live across - events.""" + events. + + Per proposal 0045: carries the span's own ``fan_out_index_chain`` + and ``branch_name_chain`` so the augmentation walk can apply + §3.4's lineage-aware boundary rule without re-deriving the chain + from successive events.""" span: Span + fan_out_index_chain: tuple[int | None, ...] = () + branch_name_chain: tuple[str | None, ...] = () + + +def _span_chain_on_path( + open_span: _OpenSpan, + aug_fi_chain: tuple[int | None, ...], + aug_bn_chain: tuple[str | None, ...], +) -> bool: + """Return True iff ``open_span``'s chain is a prefix-match of the + augmenter's chain — i.e., the span sits on the augmenter's + call-stack ancestor path. Per proposal 0045 §3.4: + + - A span shorter than the augmenter (chain prefix-matches) is an + ancestor on the path. + - A span at the same depth (chain exact-matches) is the augmenter + itself (or a descendant sharing the mutated mapping). + - A span deeper than the augmenter, OR with a position-mismatch + anywhere, is a sibling and MUST NOT be updated. + """ + span_fi = open_span.fan_out_index_chain + span_bn = open_span.branch_name_chain + if len(span_fi) > len(aug_fi_chain): + return False + if len(span_bn) > len(aug_bn_chain): + return False + for i in range(len(span_fi)): + if span_fi[i] != aug_fi_chain[i]: + return False + for i in range(len(span_bn)): + if span_bn[i] != aug_bn_chain[i]: + return False + return True # Sorted object keys, no insignificant whitespace, UTF-8 output (per @@ -645,7 +683,11 @@ def _open_started_span(self, event: NodeEvent) -> None: kind=SpanKind.INTERNAL, attributes=self._node_attrs(event, correlation_id), ) - inv_state.open_spans[self._key_for(event)] = _OpenSpan(span=span) + inv_state.open_spans[self._key_for(event)] = _OpenSpan( + span=span, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) def _handle_completed(self, event: NodeEvent) -> None: """Close the matching span, applying §4.2 status mapping.""" @@ -771,62 +813,118 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non def _collect_augmentation_targets( self, invocation_id: str, event: MetadataAugmentationEvent ) -> list[Span]: + """Collect open spans on the augmenter's call-stack ancestor + chain per proposal 0045 §3.4. Three-step boundary decision + tree per open span: + + 1. Same context as augmenter (or descendant sharing the + mutated mapping) — update. + 2. Strict dispatch ancestor on the augmenter's call-stack path + (each outer fan-out instance dispatch, each outer parallel- + branches branch dispatch, each outer serial-subgraph wrapper + on the path) — update. + 3. Sibling at any depth, OR shared parent at any depth (the + fan-out NODE itself, the parallel-branches NODE itself, the + invocation span when augmenter is non-root) — do not update. + + Chain match: for an open span to be on the augmenter's path, + the span's own per-depth lineage chain MUST prefix-match the + augmenter's chain at every position the span occupies. Where + positions disagree, the span is a sibling. + """ targets: list[Span] = [] - aug_fi = event.fan_out_index - aug_bn = event.branch_name aug_ns = event.namespace + aug_fi_chain = event.fan_out_index_chain + aug_bn_chain = event.branch_name_chain inv_state = self._inv_states.get(invocation_id) - if aug_fi is None and aug_bn is None: - # Outermost-serial context: the invocation span is in scope. + + # Invocation span: included only when the augmenter is in + # OUTERMOST SERIAL context — no fan-out instance and no + # parallel-branches branch on its call-stack path. Subgraph + # wrappers (chain entries with None at both axes) don't + # introduce the shared-parent semantics; only fan-out and + # parallel-branches do. Mirrors fixture 034 and matches the + # 0045 §3.4 statement that "existing single-level fixtures + # remain unchanged." + outermost_serial = all(fi is None for fi in aug_fi_chain) and all(bn is None for bn in aug_bn_chain) + if outermost_serial: inv_open = self._invocation_span.get(invocation_id) if inv_open is not None: targets.append(inv_open.span) - if inv_state is not None: - # Subgraph wrapper spans on the ancestor path. - for prefix, open_span in inv_state.subgraph_spans.items(): - if is_strict_prefix(prefix, aug_ns): - targets.append(open_span.span) + if inv_state is None: return targets - if aug_fi is not None: - # Fan-out instance dispatch span(s) on the ancestor path. - # Keys are anchor_ns + (str(fan_out_index),) per - # ``_open_fan_out_instance_dispatch_span``. - fi_str = str(aug_fi) - for key, open_span in inv_state.fan_out_instance_spans.items(): - if not key or key[-1] != fi_str: - continue - anchor_ns = key[:-1] - if is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns: - targets.append(open_span.span) - if aug_bn is not None: - # Per-branch dispatch span(s) on the ancestor path (per - # spec §3.4's "outer parallel-branches branch" ancestor- - # chain rule, formalized in proposal 0045). Keys are - # anchor_ns + (branch_name,) per - # ``_open_parallel_branches_branch_dispatch_span``. - # Same shape as the fan-out branch above. - for key, open_span in inv_state.parallel_branches_branch_spans.items(): - if not key or key[-1] != aug_bn: - continue - anchor_ns = key[:-1] - if is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns: - targets.append(open_span.span) - # Open node spans on the augmenter's call stack. Match on - # ``fan_out_index`` AND ``branch_name`` to skip sibling - # instances / branches; namespace must prefix (or equal) the - # augmenter's. The BN discriminator is what keeps two - # concurrent same-named inner nodes across sibling - # parallel-branches branches from leaking augmentation to - # each other. + + # Subgraph wrapper spans on the path: prefix-of-aug-namespace + + # chain prefix-matches aug's chain. + for prefix, open_span in inv_state.subgraph_spans.items(): + if not is_strict_prefix(prefix, aug_ns): + continue + if _span_chain_on_path(open_span, aug_fi_chain, aug_bn_chain): + targets.append(open_span.span) + + # Fan-out instance dispatch spans: keyed by anchor_ns + + # (str(fi),). The dispatch represents the descent at namespace + # position len(anchor_ns)-1 — i.e., chain position + # ``len(anchor_ns)-1`` should match the dispatch's fi. + for key, open_span in inv_state.fan_out_instance_spans.items(): + if not key: + continue + anchor_ns = key[:-1] + fi_str = key[-1] + if not (is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns): + continue + chain_pos = len(anchor_ns) - 1 + if chain_pos < 0 or chain_pos >= len(aug_fi_chain): + continue + aug_fi_at_pos = aug_fi_chain[chain_pos] + if aug_fi_at_pos is None or str(aug_fi_at_pos) != fi_str: + continue + targets.append(open_span.span) + + # Per-branch dispatch spans: keyed by anchor_ns + (branch_name,). + # Mirror logic to fan-out above, against the branch chain. + for key, open_span in inv_state.parallel_branches_branch_spans.items(): + if not key: + continue + anchor_ns = key[:-1] + bn_str = key[-1] + if not (is_strict_prefix(anchor_ns, aug_ns) or anchor_ns == aug_ns): + continue + chain_pos = len(anchor_ns) - 1 + if chain_pos < 0 or chain_pos >= len(aug_bn_chain): + continue + if aug_bn_chain[chain_pos] != bn_str: + continue + targets.append(open_span.span) + + # Open NODE spans: same context (aug's own attempt span), or + # strict ancestor on the augmenter's path. Skip: + # - sibling NODE spans (chain mismatch at some position) + # - shared-parent NODE spans (fan-out NODE / pb NODE + # identified structurally by their presence in the + # parent_node_name caches) for key, open_span in inv_state.open_spans.items(): - ns, _ai, fi, bn = key - if fi != aug_fi: + 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) + continue + if not is_strict_prefix(ns, aug_ns): continue - if bn != aug_bn: + # 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). + if ns in inv_state.fan_out_parent_node_name or ns in inv_state.parallel_branches_parent_node_name: continue - if is_prefix_or_equal(ns, aug_ns): + if _span_chain_on_path(open_span, aug_fi_chain, aug_bn_chain): targets.append(open_span.span) + return targets # ------------------------------------------------------------------ @@ -1032,7 +1130,11 @@ def _handle_llm_event(self, event: NodeEvent) -> None: kind=SpanKind.CLIENT, attributes=attrs, ) - inv_state.open_llm_spans[payload.call_id] = _OpenSpan(span=span) + inv_state.open_llm_spans[payload.call_id] = _OpenSpan( + span=span, + fan_out_index_chain=event.fan_out_index_chain, + branch_name_chain=event.branch_name_chain, + ) elif event.phase == "completed": open_span = inv_state.open_llm_spans.pop(payload.call_id, None) if open_span is None: @@ -1394,7 +1496,16 @@ def _open_subgraph_span( kind=SpanKind.INTERNAL, attributes=attrs, ) - inv_state.subgraph_spans[prefix] = _OpenSpan(span=span) + # Per proposal 0045: the subgraph wrapper sits at namespace + # depth ``len(prefix)``; its chain is the slice of the + # inner event's chain up to that position. Both chain + # entries at the wrapper's position are ``None`` (subgraph + # wrappers don't introduce a fan-out or branch axis). + inv_state.subgraph_spans[prefix] = _OpenSpan( + span=span, + fan_out_index_chain=event.fan_out_index_chain[: len(prefix)], + branch_name_chain=event.branch_name_chain[: len(prefix)], + ) def _close_subgraph_span(self, inv_state: _InvState, prefix: tuple[str, ...]) -> None: open_span = inv_state.subgraph_spans.pop(prefix, None) @@ -1587,7 +1698,16 @@ def _open_fan_out_instance_dispatch_span( attributes=attrs, ) instance_key = prefix + (str(event.fan_out_index),) - inv_state.fan_out_instance_spans[instance_key] = _OpenSpan(span=instance_span) + # Per proposal 0045: this dispatch span sits AT the descent + # boundary into the fan-out instance. Its chain is the slice + # of the inner event's chain up to and including the + # boundary's own position. + chain_len = len(prefix) + inv_state.fan_out_instance_spans[instance_key] = _OpenSpan( + span=instance_span, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) def _close_fan_out_instance_dispatch_span(self, inv_state: _InvState, key: tuple[str, ...]) -> None: open_span = inv_state.fan_out_instance_spans.pop(key, None) @@ -1659,7 +1779,15 @@ def _open_parallel_branches_branch_dispatch_span( attributes=attrs, ) branch_key = prefix + (event.branch_name,) - inv_state.parallel_branches_branch_spans[branch_key] = _OpenSpan(span=branch_span) + # Per proposal 0045: this dispatch span sits at the descent + # boundary into this branch. Its chain is the slice up to + # and including the boundary's own position. + chain_len = len(prefix) + inv_state.parallel_branches_branch_spans[branch_key] = _OpenSpan( + span=branch_span, + fan_out_index_chain=event.fan_out_index_chain[:chain_len], + branch_name_chain=event.branch_name_chain[:chain_len], + ) def _close_parallel_branches_branch_dispatch_span( self, inv_state: _InvState, key: tuple[str, ...] diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index cb2d86f5..f3d20330 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -102,6 +102,31 @@ # Both fixtures land once spec settles the dispatch-span # shape AND the adapter learns to infer fan-out aggregation # defaults from inner subgraphs. + # 039 (nested-lineage augmentation) — proposal 0045 v0.37.0. + # Stays deferred in PR 11. The three cases require harness + # extensions the existing primitives don't cover: + # - Case 1 + 3 (nested fan-out + fan-out-in-serial): the + # ``augment_metadata_from_field`` middleware captures the + # items list at fixture-build time from ``initial_state``; + # nested fan-outs read items from the OUTER instance's + # subgraph state (the ``inner_seed`` field threaded through + # per_product), which the current item-capture path can't + # resolve. Needs a runtime-state item-list lookup. + # - Case 2 (pb-inside-fan-out): introduces a new directive + # ``augment_metadata_from_outer_item: {key: field}`` — the + # pb branch's middleware sources the augmentation value + # from the SURROUNDING outer fan-out instance's item, not + # from a static value or the branch's own state. Needs a + # new middleware factory + a way to capture the outer-item + # reference at fixture-build time. + # 0045's behavioral contract IS exercised at unit level via the + # OTel observer's + # ``test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span`` + # regression test, and the cross-impl coverage the spec relies + # on for nested cases is already in place. Activating 039 against the + # Langfuse harness is a follow-up PR — the harness extensions + # above are non-trivial and orthogonal to the engine + observer + # changes 0045 actually mandates. } ) diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 60d9bbf8..465f81af 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -9,7 +9,7 @@ def test_package_versions() -> None: assert openarmature.__version__ == "0.10.0" - assert openarmature.__spec_version__ == "0.36.0" + assert openarmature.__spec_version__ == "0.37.0" def test_spec_version_matches_pyproject() -> None: diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index 138e06fd..4fb5bbb0 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -2791,3 +2791,145 @@ async def _policy_ask(_s: _BranchS) -> dict[str, str]: assert "openarmature.user.audit_kind" not in dispatch_spans_by_branch["policy_audit"], ( "sibling branch's dispatch span MUST NOT receive the augmenter's key" ) + + +async def test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span() -> None: + # Spec proposal 0045 §3.4 lineage-aware containment rule. + # Topology: outer fan-out wrapping a serial subgraph that + # contains a leaf. The leaf augments a per-item key. Augment + # targets per §3.4: + # + # - Outer instance #1's dispatch span MUST receive + # ``group="item-200"`` (rule 2, strict ancestor on the path). + # - Outer instance #0's dispatch span MUST receive + # ``group="item-100"`` (rule 2, its own subtree). + # - Outer instance #0 and #1's dispatch spans MUST NOT receive + # each other's value (rule 3, siblings). + # - The outer fan-out NODE span MUST NOT receive any group key + # (rule 3, shared parent). + # - The invocation span MUST NOT receive any group key (rule 3, + # shared parent — augmenter is inside a fan-out instance). + # + # The chain at the augmenter is ``(K,)`` where K is the outer + # instance's index — the per-depth tracking that 0045 requires + # is exercised by the resolver picking the matching outer + # dispatch span (and skipping the sibling) on each leaf's + # augmentation. + import asyncio + + from openarmature.observability.metadata import set_invocation_metadata + + class _OuterS(State): + items: list[int] = [] + results: Annotated[list[int], append] = [] + + class _MidS(State): + item: int = 0 + out: int = 0 + + class _LeafS(State): + item: int = 0 + out: int = 0 + + async def _leaf(s: _LeafS) -> dict[str, int]: + await asyncio.sleep(0) + # Augment with a per-item key so we can detect which dispatch + # span the augmentation lands on. + set_invocation_metadata(group=f"item-{s.item}") + return {"out": s.item} + + leaf_subgraph = ( + GraphBuilder(_LeafS).add_node("leaf", _leaf).add_edge("leaf", END).set_entry("leaf").compile() + ) + + # Mid-level: a serial subgraph wrapping the leaf. Threads + # ``item`` straight through and exposes ``out``. + async def _mid_passthrough(_s: _MidS) -> dict[str, int]: + return {} + + from openarmature.graph.projection import ExplicitMapping + + mid_subgraph = ( + GraphBuilder(_MidS) + .add_subgraph_node( + "leaf_wrap", + leaf_subgraph, + projection=ExplicitMapping[_MidS, _LeafS](inputs={"item": "item"}, outputs={"out": "out"}), + ) + .add_node("noop", _mid_passthrough) + .add_edge("leaf_wrap", "noop") + .add_edge("noop", END) + .set_entry("leaf_wrap") + .compile() + ) + + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + g = ( + GraphBuilder(_OuterS) + .add_fan_out_node( + "outer_fan", + subgraph=mid_subgraph, + collect_field="out", + target_field="results", + items_field="items", + item_field="item", + ) + .add_edge("outer_fan", END) + .set_entry("outer_fan") + .compile() + ) + g.attach_observer(observer) + try: + await g.invoke(_OuterS(items=[100, 200])) + await g.drain() + finally: + observer.shutdown() + + spans = exporter.get_finished_spans() + # Outer per-instance dispatch spans (name = "outer_fan" with + # ``fan_out_index`` ∈ {0, 1} and ``fan_out.parent_node_name``). + outer_dispatches = { + dict(s.attributes or {}).get("openarmature.node.fan_out_index"): s + for s in spans + if s.name == "outer_fan" + and dict(s.attributes or {}).get("openarmature.node.fan_out_index") is not None + and "openarmature.fan_out.parent_node_name" in dict(s.attributes or {}) + } + assert outer_dispatches.keys() == {0, 1}, ( + f"expected two outer fan-out instance dispatch spans; got {outer_dispatches.keys()!r}" + ) + + # Each outer dispatch carries the leaf's per-item group value (its + # own subtree's augmentation propagated outward via the lineage- + # aware boundary rule). + outer0_group = dict(outer_dispatches[0].attributes or {}).get("openarmature.user.group") + outer1_group = dict(outer_dispatches[1].attributes or {}).get("openarmature.user.group") + assert outer0_group == "item-100", ( + f"outer instance #0's dispatch span MUST carry its leaf's augmented value; got {outer0_group!r}" + ) + assert outer1_group == "item-200", ( + f"outer instance #1's dispatch span MUST carry its leaf's augmented value; got {outer1_group!r}" + ) + + # The outer fan-out NODE span (shared parent of both instances — + # the one without ``fan_out_index`` on its attributes) MUST NOT + # carry any augmented group key. + outer_node_spans = [ + s + for s in spans + if s.name == "outer_fan" and dict(s.attributes or {}).get("openarmature.node.fan_out_index") is None + ] + assert len(outer_node_spans) >= 1 + for outer_node in outer_node_spans: + assert "openarmature.user.group" not in dict(outer_node.attributes or {}), ( + "outer fan-out NODE span (shared parent) MUST NOT receive any augmented group key" + ) + + # The invocation span MUST NOT carry it (augmenter is inside a + # fan-out → invocation is a shared parent). + inv_spans = [s for s in spans if s.name == "openarmature.invocation"] + assert len(inv_spans) == 1 + assert "openarmature.user.group" not in dict(inv_spans[0].attributes or {}), ( + "invocation span MUST NOT receive augmenter's key when inside a fan-out instance" + ) From 4f5dcd5d95cd02b717625656d914e341c1ff393a Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sun, 31 May 2026 15:31:46 -0700 Subject: [PATCH 2/2] Add proposal 0045 entry to conformance manifest CI check_conformance_manifest.py flagged that proposal 0045 (accepted in spec v0.37.0) had no entry in conformance.toml. Add the entry alongside 0044 with status=implemented and the same v0.11.0 ``since`` as the other proposals in the batch. --- conformance.toml | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/conformance.toml b/conformance.toml index 084b0359..f67c637f 100644 --- a/conformance.toml +++ b/conformance.toml @@ -215,3 +215,15 @@ since = "0.11.0" [proposals."0044"] status = "implemented" since = "0.11.0" + +# Spec v0.37.0 (proposal 0045). Engine-side per-depth lineage chains +# + observer-side three-step boundary decision tree implemented. +# Single-level fixtures 029 / 030 / 034 stay unchanged per 0045's +# backward-compat note. Nested-case fixture 039 stays deferred in the +# Langfuse harness — needs runtime-state item-list lookup for nested +# fan-outs plus an ``augment_metadata_from_outer_item`` directive. +# Behavioral contract verified at unit level via +# ``test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span``. +[proposals."0045"] +status = "implemented" +since = "0.11.0"