Implement 0040 open-span metadata update

Mid-invocation set_invocation_metadata now emits a
MetadataAugmentationEvent on the engine's serial observer queue.
Both observers match the augmenter's lineage tuple
(namespace, attempt_index, fan_out_index, branch_name) against
their open observations and apply the entries in place:
span.set_attribute on OTel, observation.update(metadata=...) on
Langfuse. Outermost-serial also fires trace.update so the augmented
keys land at trace.metadata.<key>. Sibling instances and ancestors
above the innermost containment are skipped per spec sec 3.4.

OTel's open-span _StackKey widened to include branch_name so
concurrent same-named inner nodes across sibling parallel-branches
branches no longer collide. Same widening on the Langfuse observer
plus the LLM event payload's calling_branch_name field. Shared
tuple-prefix helpers extracted to observability/lineage.py.

Conformance: activate 028's mid-invocation rejection case (helper
raises at the call site on a reserved key) and 034 outermost-serial
fixture via the Langfuse harness. Fix the pre-existing 005 flake by
resetting the OTel global tracer Once primitive before
set_tracer_provider. 029 and 030 stay deferred with documented
fixture-shape gaps; the augmentation mechanism itself is covered
end-to-end by new unit tests in test_observability_otel and
test_observability_langfuse.

Two coord threads opened for follow-up: nested fan-out lineage
scope and parallel-branches per-branch dispatch span shape.

Author: chris-colinsky

examples/00-hello-world/main.py

@@ -49,6 +49,7 @@
     END,
     CompiledGraph,
     GraphBuilder,
+    MetadataAugmentationEvent,
     NodeEvent,
     State,
     append,
@@ -194,14 +195,18 @@ def route(state: PipelineState) -> str:
     return state.classification.intent
 
 
-async def trace(event: NodeEvent) -> None:
+async def trace(event: NodeEvent | MetadataAugmentationEvent) -> None:
     # OpenAIProvider emits NodeEvent-shaped events for LLM-span
     # tracking under a sentinel namespace; those have post_state=None.
+    # ``set_invocation_metadata`` from within a node body emits a
+    # MetadataAugmentationEvent; this tracer ignores those.
     # Filter to events that carry a PipelineState snapshot before
-    # reading it. The isinstance check both narrows the type for
+    # reading it. The isinstance checks both narrow the type for
     # static checkers (post_state is typed as the base State, not
-    # PipelineState) and acts as a defensive guard against any
+    # PipelineState) and act as a defensive guard against any
     # foreign-state observer event the engine might dispatch.
+    if not isinstance(event, NodeEvent):
+        return
     if event.phase == "completed" and event.error is None and isinstance(event.post_state, PipelineState):
         print(f"{event.node_name}: sources={event.post_state.sources}")
 

examples/03-observer-hooks/main.py

@@ -54,6 +54,7 @@
     CompiledGraph,
     ExplicitMapping,
     GraphBuilder,
+    MetadataAugmentationEvent,
     NodeEvent,
     Observer,
     State,
@@ -186,12 +187,18 @@ def build_review_subgraph() -> CompiledGraph[ReviewState]:
 # fire on every invocation of the compiled graph until removed.
 
 
-async def console_tracer(event: NodeEvent) -> None:
+async def console_tracer(event: NodeEvent | MetadataAugmentationEvent) -> None:
     """Print one structured line per node boundary to stderr.
 
     Format: `[step=N] namespace.path → fields_changed_in_this_step`
     On error, format flips to `... ✗ error_category`.
+
+    Mid-invocation ``set_invocation_metadata`` augmentations also
+    reach observers as ``MetadataAugmentationEvent`` instances; this
+    tracer ignores them.
     """
+    if isinstance(event, MetadataAugmentationEvent):
+        return
     namespace = ".".join(event.namespace)
     if event.error is not None:
         print(
@@ -232,7 +239,9 @@ def __init__(self) -> None:
         self.errors: int = 0
         self.namespaces: set[tuple[str, ...]] = set()
 
-    async def __call__(self, event: NodeEvent) -> None:
+    async def __call__(self, event: NodeEvent | MetadataAugmentationEvent) -> None:
+        if isinstance(event, MetadataAugmentationEvent):
+            return
         self.events += 1
         if event.error is not None:
             self.errors += 1

examples/04-nested-subgraphs/main.py

@@ -49,6 +49,7 @@
     CompiledGraph,
     ExplicitMapping,
     GraphBuilder,
+    MetadataAugmentationEvent,
     NodeEvent,
     State,
     append,
@@ -349,7 +350,9 @@ def _fmt_state(state: Any) -> str:
     return " ".join(parts) if parts else "(empty)"
 
 
-async def depth_observer(event: NodeEvent) -> None:
+async def depth_observer(event: NodeEvent | MetadataAugmentationEvent) -> None:
+    if isinstance(event, MetadataAugmentationEvent):
+        return
     depth = len(event.namespace)
     indent = "  " * (depth - 1)
     ns = " > ".join(event.namespace)

examples/05-fan-out-with-retry/main.py

@@ -78,6 +78,7 @@
     END,
     CompiledGraph,
     GraphBuilder,
+    MetadataAugmentationEvent,
     NodeEvent,
     State,
     append,
@@ -296,7 +297,7 @@ def build_graph(error_policy: str = "fail_fast") -> CompiledGraph[BatchState]:
     )
 
 
-async def fan_out_config_observer(event: NodeEvent) -> None:
+async def fan_out_config_observer(event: NodeEvent | MetadataAugmentationEvent) -> None:
     """Print the fan-out node's resolved config when its dispatch event
     fires.
 
@@ -308,6 +309,8 @@ async def fan_out_config_observer(event: NodeEvent) -> None:
     ``concurrency`` are callable resolvers whose value isn't visible
     in code.
     """
+    if not isinstance(event, NodeEvent):
+        return
     if event.fan_out_config is None:
         return
     if event.phase != "started":

examples/06-parallel-branches/main.py

@@ -70,6 +70,7 @@
     BranchSpec,
     CompiledGraph,
     GraphBuilder,
+    MetadataAugmentationEvent,
     NodeEvent,
     State,
     append,
@@ -240,7 +241,7 @@ async def present(s: ArticleState) -> Mapping[str, Any]:
     return {"trace": ["present"]}
 
 
-async def branch_attribution_observer(event: NodeEvent) -> None:
+async def branch_attribution_observer(event: NodeEvent | MetadataAugmentationEvent) -> None:
     """Print which branch each inner-node event came from.
 
     NodeEvent carries ``branch_name`` on events from nodes that
@@ -250,6 +251,8 @@ async def branch_attribution_observer(event: NodeEvent) -> None:
     observer skips events with no branch attribution and prints
     ``(branch=…) node_name`` for the rest.
     """
+    if not isinstance(event, NodeEvent):
+        return
     if event.branch_name is None or event.phase != "started":
         return
     print(f"  [observer] (branch={event.branch_name}) inner node {event.node_name!r} started")

src/openarmature/graph/__init__.py

@@ -35,7 +35,7 @@
     StateValidationError,
     UnreachableNode,
 )
-from .events import NodeEvent
+from .events import MetadataAugmentationEvent, NodeEvent
 from .fan_out import FanOutConfig, FanOutNode
 from .middleware import (
     Middleware,
@@ -78,6 +78,7 @@
     "GraphBuilder",
     "GraphError",
     "MappingReferencesUndeclaredField",
+    "MetadataAugmentationEvent",
     "Middleware",
     "MultipleOutgoingEdges",
     "NextCall",

src/openarmature/graph/events.py

@@ -231,4 +231,46 @@ class NodeEvent:
     caller_invocation_metadata: Mapping[str, AttributeValue] = field(default_factory=lambda: _EMPTY_METADATA)
 
 
-__all__ = ["FanOutEventConfig", "NodeEvent"]
+# Spec: realizes observability §3.4 + graph-engine §6 augmentation
+# event mechanism (proposal 0040). Emitted by
+# ``set_invocation_metadata`` when called mid-invocation; carries the
+# delta + the augmenting context's lineage identity so observers can
+# resolve which of their open observations belong to the augmenting
+# context's subtree and apply the entries in place.
+@dataclass(frozen=True)
+class MetadataAugmentationEvent:
+    """A metadata-augmentation event delivered to observers.
+
+    Emitted by :func:`openarmature.observability.metadata.set_invocation_metadata`
+    when called mid-invocation. Carries:
+
+    - ``entries``: the delta merged into the per-async-context
+      invocation metadata mapping by the call. Read-only view.
+    - ``namespace`` / ``attempt_index`` / ``fan_out_index`` /
+      ``branch_name``: the four lineage fields that jointly identify
+      the augmenting execution context (the calling node's identity
+      tuple). When ``set_invocation_metadata`` is called from outside
+      a node body, ``namespace`` is the empty tuple, ``attempt_index``
+      is ``0``, and both ``fan_out_index`` and ``branch_name`` are
+      ``None`` — the invocation-level identity.
+
+    Distinct from :class:`NodeEvent` because there is no node phase,
+    no pre/post state, and no error: this event reports a side-channel
+    augmentation, not a node-attempt boundary. Per graph-engine §6 the
+    event is NOT subject to the observer ``phases`` filter (which only
+    governs ``NodeEvent`` phases); the delivery worker forwards it to
+    every subscribed observer. Observers that handle it iterate their
+    open observations whose lineage is an ancestor of (or equal to)
+    the augmenting context's lineage and apply the entries as
+    ``openarmature.user.<key>`` (OTel, §5.6) /
+    ``metadata.<key>`` (Langfuse, §8.4.1+§8.4.2).
+    """
+
+    entries: Mapping[str, AttributeValue]
+    namespace: tuple[str, ...]
+    attempt_index: int = 0
+    fan_out_index: int | None = None
+    branch_name: str | None = None
+
+
+__all__ = ["FanOutEventConfig", "MetadataAugmentationEvent", "NodeEvent"]

src/openarmature/graph/observer.py

@@ -34,19 +34,20 @@
 from dataclasses import dataclass, field
 from typing import Any, Literal, Protocol
 
-from .events import NodeEvent
+from .events import MetadataAugmentationEvent, NodeEvent
 from .state import State
 
 
 class Observer(Protocol):
-    """The shape of a callable that receives node-boundary events.
+    """The shape of a callable that receives observer events.
 
     `Observer` is a structural Protocol; any async callable matching the
     signature qualifies, no subclass required. Plain functions, bound
     methods, and class instances with `__call__` all work::
 
-        async def log_observer(event: NodeEvent) -> None:
-            print(event.node_name, event.phase)
+        async def log_observer(event: NodeEvent | MetadataAugmentationEvent) -> None:
+            if isinstance(event, NodeEvent):
+                print(event.node_name, event.phase)
 
         compiled.attach_observer(log_observer)
 
@@ -63,6 +64,27 @@ async def log_observer(event: NodeEvent) -> None:
     conformance doesn't pin you to that name; any of `event`, `_event`,
     `e`, etc. matches.
 
+    Two event variants reach observers (graph-engine §6 + proposal
+    0040). The signature is the union; observers ``isinstance``-narrow
+    on the first line and choose which variants they handle.
+
+    - :class:`NodeEvent` — the started/completed/checkpoint phase
+      events. Subject to the ``phases`` filter on
+      :class:`SubscribedObserver`; observers whose phase set excludes
+      ``event.phase`` do NOT receive it.
+    - :class:`MetadataAugmentationEvent` — emitted by
+      :func:`openarmature.observability.metadata.set_invocation_metadata`
+      when called mid-invocation. Carries the augmenting context's
+      lineage tuple (``namespace``, ``attempt_index``,
+      ``fan_out_index``, ``branch_name``) so rich backends can update
+      their open observations in place
+      (``span.set_attribute(openarmature.user.<key>, v)`` for OTel,
+      ``observation.update(metadata=...)`` for Langfuse). Per spec §6
+      this variant is NOT subject to the ``phases`` filter — every
+      subscribed observer sees it and isinstance-narrows to decide
+      whether to act. Simple user observers typically early-return
+      after ``isinstance(event, NodeEvent)`` checks.
+
     Optional ``prepare_sync`` extension
     -----------------------------------
     An observer MAY additionally define a synchronous method::
@@ -81,9 +103,13 @@ def prepare_sync(self, event: NodeEvent, /) -> None: ...
     the synchronous prep entirely; observers that do define it run
     only for ``"started"``-phase events, with errors warned-not-
     propagated (same isolation contract as the async path).
+    ``prepare_sync`` is never invoked for
+    :class:`MetadataAugmentationEvent` (the synchronous-prep contract
+    is anchored on the ``started`` phase, which only ``NodeEvent``
+    carries).
     """
 
-    async def __call__(self, event: NodeEvent, /) -> None: ...
+    async def __call__(self, event: NodeEvent | MetadataAugmentationEvent, /) -> None: ...
 
 
 # Per spec v0.6.0 §6: the two valid phase strings. Used as the default
@@ -200,15 +226,22 @@ class _QueuedItem:
     receive it. The list is computed at dispatch time so events from
     different depths in nested subgraphs carry the correct observer chain
     without the worker needing to know the graph topology.
+
+    ``event`` is the union of ``NodeEvent`` (started / completed /
+    checkpoint phases) and ``MetadataAugmentationEvent`` (proposal
+    0040, side-channel augmentation). The delivery worker branches by
+    type to apply the right delivery contract (phase-filter for
+    ``NodeEvent``, no filter for the augmentation event).
     """
 
-    event: NodeEvent
+    event: NodeEvent | MetadataAugmentationEvent
     observers: tuple[SubscribedObserver, ...]
 
 
 # A sentinel value the engine puts on the queue to signal the worker to
 # return after draining the events ahead of it. None is unambiguous —
-# observers receive `NodeEvent` instances, never None.
+# the queue carries `NodeEvent` and `MetadataAugmentationEvent` instances
+# wrapped in `_QueuedItem`, never None.
 _DRAIN_SENTINEL = None
 
 
@@ -587,16 +620,29 @@ def take_step(self) -> int:
         return n
 
 
-def _dispatch(context: _InvocationContext, event: NodeEvent) -> None:
-    """Enqueue a node event for the delivery worker.
-
-    For ``"started"``-phase events, also call any subscribed observer's
-    optional ``prepare_sync(event)`` synchronously — in the engine task,
-    BEFORE queueing — so observers that need to publish per-event state
-    the engine itself reads in the same engine-task scope (e.g., the
-    OTel observer setting ``current_active_observer_span`` for the
-    engine to attach into the OTel context) can do so before the node
-    body runs.
+def _dispatch(
+    context: _InvocationContext,
+    event: NodeEvent | MetadataAugmentationEvent,
+) -> None:
+    """Enqueue an event for the delivery worker.
+
+    Handles two event variants:
+
+    - :class:`NodeEvent`: the started/completed/checkpoint pair model.
+      For ``"started"``-phase events, also calls any subscribed
+      observer's optional ``prepare_sync(event)`` synchronously — in
+      the engine task, BEFORE queueing — so observers that need to
+      publish per-event state the engine itself reads in the same
+      engine-task scope (e.g., the OTel observer setting
+      ``current_active_observer_span`` for the engine to attach into
+      the OTel context) can do so before the node body runs.
+    - :class:`MetadataAugmentationEvent` (proposal 0040): a side-
+      channel augmentation event emitted by
+      ``set_invocation_metadata`` mid-invocation. Bypasses the
+      ``prepare_sync`` branch entirely — the sync-prep contract is
+      anchored on ``"started"``, which only ``NodeEvent`` carries.
+      Queued onto the same serial worker so observers see it in
+      strict order with the surrounding node events.
 
     Phase-gated forwarding: ``prepare_sync`` only fires when ``"started"``
     is in the subscribed observer's ``phases`` set, mirroring how the
@@ -616,7 +662,7 @@ def _dispatch(context: _InvocationContext, event: NodeEvent) -> None:
     observers = context.full_observers()
     if not observers:
         return
-    if event.phase == "started":
+    if isinstance(event, NodeEvent) and event.phase == "started":
         for subscribed in observers:
             if "started" not in subscribed.phases:
                 continue
@@ -686,9 +732,15 @@ async def deliver_loop(
       each).
     - No observer receives event N+1 until everyone has finished N
       (the loop processes one item fully before pulling the next).
-    - Observers whose ``phases`` set excludes the event's phase do
-      NOT receive it. Phase filter applies at delivery, not dispatch;
-      the engine still produces both events for every attempt.
+    - For :class:`NodeEvent`, observers whose ``phases`` set excludes
+      the event's phase do NOT receive it. Phase filter applies at
+      delivery, not dispatch; the engine still produces both events
+      for every attempt.
+    - For :class:`MetadataAugmentationEvent` (proposal 0040), the
+      ``phases`` filter is bypassed entirely — the event isn't a
+      node-phase event, so every subscribed observer receives it
+      regardless of ``phases``. Observers ``isinstance``-narrow on
+      the first line and choose whether to act.
     - Observer exceptions don't propagate, don't break siblings,
       don't block subsequent events. Reported via ``warnings.warn``.
 
@@ -698,11 +750,12 @@ async def deliver_loop(
         item = await queue.get()
         if item is None:
             return
+        event = item.event
         for subscribed in item.observers:
-            if item.event.phase not in subscribed.phases:
+            if isinstance(event, NodeEvent) and event.phase not in subscribed.phases:
                 continue
             try:
-                await subscribed.observer(item.event)
+                await subscribed.observer(event)
             except Exception as e:
                 warnings.warn(
                     f"observer raised {type(e).__name__}: {e}",

src/openarmature/llm/providers/openai.py

@@ -55,6 +55,7 @@
 from openarmature.graph.events import NodeEvent
 from openarmature.observability.correlation import (
     current_attempt_index,
+    current_branch_name,
     current_dispatch,
     current_fan_out_index,
     current_namespace_prefix,
@@ -1256,6 +1257,7 @@ def _make_llm_event(
         calling_namespace_prefix=current_namespace_prefix(),
         calling_attempt_index=current_attempt_index(),
         calling_fan_out_index=current_fan_out_index(),
+        calling_branch_name=current_branch_name(),
         active_prompt=active_prompt,
         active_prompt_group=active_prompt_group,
         input_messages=input_messages,