Implement proposal 0045 (nested-lineage augmentation) (#104)

* Implement proposal 0045 (nested-lineage augmentation)

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.

* 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.

Author: chris-colinsky

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.

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
@@ -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"

openarmature-spec

@@ -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 = [

pyproject.toml

@@ -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`
 

src/openarmature/AGENTS.md

@@ -25,4 +25,4 @@
 """
 
 __version__ = "0.10.0"
-__spec_version__ = "0.36.0"
+__spec_version__ = "0.37.0"

src/openarmature/__init__.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(),
             ),

src/openarmature/graph/compiled.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*

src/openarmature/graph/events.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,

src/openarmature/graph/observer.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]: