Add caller-supplied invocation metadata (proposal 0034)

Implements observability §3.4 + §5.6 + §8.4.1/§8.4.2: callers pass
``invoke(metadata={...})`` and the framework propagates the entries
to every observability backend.

Public surface:

- ``compiled.invoke(metadata=...)`` accepts a per-invocation
  ``dict[str, AttributeValue]`` mapping where values are OTel
  scalars or homogeneous arrays (str/int/float/bool).
- ``openarmature.observability.set_invocation_metadata(**entries)``
  augments the in-scope mapping mid-invocation (additive merge;
  existing keys overwritten).
- ``openarmature.observability.current_invocation_metadata()``
  reads the in-scope mapping (returns empty MappingProxyType
  outside an invocation).
- Synchronous validation at the API boundary rejects keys under
  the reserved ``openarmature.*`` / ``gen_ai.*`` prefixes and
  rejects non-OTel-compatible value types (``ValueError`` before
  any work begins).

Engine internals:

- ContextVar lifecycle in
  ``openarmature.observability.metadata``; engine drives
  ``_set_invocation_metadata`` / ``_reset_invocation_metadata``
  around the outermost ``invoke()`` call.
- ``NodeEvent.caller_invocation_metadata`` carries a dispatch-time
  snapshot (the deliver_loop task's Context is frozen at invoke
  time, so observers can't re-read the live ContextVar safely).
- ``LlmEventPayload.caller_invocation_metadata`` mirrors the
  pattern for LLM provider events.

OTel observer (§5.6): emits each entry as
``openarmature.user.<key>`` on every span — invocation, node,
subgraph wrapper, fan-out instance dispatch, LLM provider,
detached roots, checkpoint-migrate, checkpoint-save. Cross-cutting
attribute family parallel to ``openarmature.correlation_id``.

Langfuse observer (§8.4.1 + §8.4.2): merges each entry as a
top-level key into ``trace.metadata`` (at trace open) and into
every observation's ``metadata`` bag (leaf nodes, subgraph
wrappers, fan-out instance dispatches, detached-trace wrappers
+ link observations, LLM generations).

Conformance fixtures (proposal 0035's full set is 026 / 027 / 028
/ 029 / 030):

- 026 (OTel cross-cutting) and 027 (Langfuse top-level merge)
  activated.
- 028 (boundary rejection) activated via a dedicated
  ``_run_fixture_028`` runner that attaches both observers and
  asserts ``ValueError`` + no spans + no Langfuse traces.
- 029 (fan-out per-instance) and 030 (parallel-branches per-branch)
  stay deferred — they need an ``augment_metadata_from_field``
  harness primitive that calls ``set_invocation_metadata`` per
  fan-out instance / per parallel branch. The augmentation surface
  is already exercised by unit tests; the conformance fixtures
  un-defer in a follow-up.

Adds 25 focused unit tests covering boundary validation, ContextVar
lifecycle, augmentation merge / overwrite, reserved-namespace
rejection, ``invoke()``-boundary rejection, mid-invocation
augmentation persisting to subsequent nodes,
``openarmature.user.*`` emission on every OTel span, and Langfuse
top-level merge on trace + every observation.

The OTel-side LLM-payload runner was extended to handle multi-node
graphs where ``calls_llm`` is on a non-entry node (fixture 026 has
a ``prep`` step before the LLM call).

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -78,6 +78,12 @@
     current_active_observer_span,
     current_attempt_index,
 )
+from openarmature.observability.metadata import (
+    _reset_invocation_metadata,
+    _set_invocation_metadata,
+    current_invocation_metadata,
+    validate_invocation_metadata,
+)
 
 from .edges import END, ConditionalEdge, EndSentinel, StaticEdge
 from .errors import (
@@ -767,6 +773,7 @@ async def invoke(
         *,
         correlation_id: str | None = None,
         resume_invocation: str | None = None,
+        metadata: Mapping[str, Any] | None = None,
     ) -> StateT:
         """Run the graph from ``initial_state`` to END and return the
         final state.
@@ -805,8 +812,33 @@ async def invoke(
           own retry logic if transient backend failures should be
           reattempted.
 
+        **Caller-supplied invocation metadata (proposal 0034).**
+
+        - ``metadata`` is an optional mapping of arbitrary
+          ``key → value`` entries the framework propagates to every
+          observability backend. Values MUST be OTel-attribute-
+          compatible scalars (``str`` / ``int`` / ``float`` / ``bool``)
+          or homogeneous arrays of those types. Keys MUST NOT use
+          the ``openarmature.*`` or ``gen_ai.*`` reserved namespaces.
+          Validation runs synchronously at the API boundary; rule
+          violations raise ``ValueError`` BEFORE any work begins.
+        - Per spec §5.6 the OTel observer emits each entry as an
+          ``openarmature.user.<key>`` cross-cutting span attribute on
+          every span and OTel log record. The Langfuse observer
+          merges each entry into ``trace.metadata`` AND every
+          ``observation.metadata`` (top level, sibling to
+          ``correlation_id``).
+        - Mid-invocation augmentation via
+          :func:`openarmature.observability.set_invocation_metadata`
+          merges into the same ContextVar with the same validation
+          rules; affects spans emitted AFTER the call returns.
+
         Raises one of the runtime error categories on failure.
         """
+        # Validate caller-supplied metadata at the API boundary so any
+        # rule violation surfaces synchronously before the worker task
+        # is created or any node body runs.
+        validated_metadata = validate_invocation_metadata(metadata)
 
         invocation_scoped = tuple(_coerce_subscribed(o) for o in (observers or ()))
         queue: asyncio.Queue[_QueuedItem | None] = asyncio.Queue()
@@ -943,6 +975,7 @@ async def invoke(
         # "per-invocation is OUTERMOST invoke" wording).
         correlation_token = _set_correlation_id(resolved_correlation_id)
         invocation_token = _set_invocation_id(invocation_id)
+        metadata_token = _set_invocation_metadata(validated_metadata)
         worker = asyncio.create_task(deliver_loop(queue, context.drain_counters))
         self._active_workers[worker] = context
         # Auto-prune: when the worker completes (after the sentinel is
@@ -978,6 +1011,7 @@ async def invoke(
         try:
             return await self._invoke(starting_state, context)
         finally:
+            _reset_invocation_metadata(metadata_token)
             _reset_invocation_id(invocation_token)
             _reset_correlation_id(correlation_token)
             # Sentinel terminates the worker after it processes events
@@ -1988,6 +2022,7 @@ def _dispatch_started(
                 fan_out_config=fan_out_config,
                 branch_name=current_branch_name(),
                 subgraph_identities=context.subgraph_identities,
+                caller_invocation_metadata=current_invocation_metadata(),
             ),
         )
 
@@ -2022,6 +2057,7 @@ def _dispatch_completed(
                 fan_out_config=fan_out_config,
                 branch_name=current_branch_name(),
                 subgraph_identities=context.subgraph_identities,
+                caller_invocation_metadata=current_invocation_metadata(),
             ),
         )
 
@@ -2205,5 +2241,6 @@ async def _maybe_save_checkpoint(
                 attempt_index=attempt_index,
                 fan_out_index=None,
                 subgraph_identities=context.subgraph_identities,
+                caller_invocation_metadata=current_invocation_metadata(),
             ),
         )

src/openarmature/graph/events.py

@@ -13,12 +13,21 @@
 Frozen dataclass; observers receive a snapshot, not a live handle.
 """
 
-from dataclasses import dataclass
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from types import MappingProxyType
 from typing import Any, Literal
 
+from openarmature.observability.metadata import AttributeValue
+
 from .errors import RuntimeGraphError
 from .state import State
 
+# Sentinel empty metadata mapping for events constructed without a
+# live caller-metadata snapshot (test helpers, synthetic events).
+# Read-only proxy keeps the default allocation-free.
+_EMPTY_METADATA: MappingProxyType[str, AttributeValue] = MappingProxyType({})
+
 
 # Spec: realizes observability §5.4 fan-out attributes via the
 # event-payload mechanism added by proposal 0013 (v0.10.0). Backend
@@ -205,6 +214,21 @@ class NodeEvent:
     # empty string when ``None`` per §5.3's "if the implementation
     # tracks one" clause.
     subgraph_identities: tuple[str | None, ...] = ()
+    # Per observability §3.4 + §5.6 (proposal 0034): snapshot of the
+    # caller-supplied invocation metadata at event-construction
+    # time. The engine reads ``current_invocation_metadata()`` when
+    # it constructs the event (in the engine task / node body's
+    # Context); the observer reads from the snapshot on the event
+    # rather than re-reading the ContextVar at observer time —
+    # critical because the observer runs on the engine's
+    # ``deliver_loop`` task whose Context is frozen at invoke time
+    # (asyncio.create_task copies the parent Context at task
+    # creation), so the live ContextVar value in the deliver_loop
+    # would NOT reflect mid-invocation augmentations made by node
+    # bodies running in the main engine task. Observers emit each
+    # entry as ``openarmature.user.<key>`` (OTel, §5.6) /
+    # ``metadata.<key>`` (Langfuse, §8.4.1+§8.4.2).
+    caller_invocation_metadata: Mapping[str, AttributeValue] = field(default_factory=lambda: _EMPTY_METADATA)
 
 
 __all__ = ["FanOutEventConfig", "NodeEvent"]

src/openarmature/llm/providers/openai.py

@@ -60,6 +60,7 @@
     current_namespace_prefix,
 )
 from openarmature.observability.llm_event import LlmEventPayload
+from openarmature.observability.metadata import current_invocation_metadata
 
 # ``current_prompt_group`` / ``current_prompt_result`` are imported
 # lazily inside :meth:`OpenAIProvider.complete` to avoid a module-load
@@ -1264,6 +1265,7 @@ def _make_llm_event(
         response_id=response_id,
         response_model=response_model,
         genai_system=genai_system,
+        caller_invocation_metadata=dict(current_invocation_metadata()),
     )
     return NodeEvent(
         node_name="openarmature.llm.complete",

src/openarmature/observability/__init__.py

@@ -41,6 +41,16 @@
 # ``opentelemetry-sdk`` dependency) along.
 from .llm_event import LLM_NAMESPACE, LlmEventPayload
 
+# v0.10.0 (proposal 0034): caller-supplied invocation metadata surface.
+# `set_invocation_metadata` is the public augmentation helper users
+# call from inside node bodies / middleware / observers;
+# `current_invocation_metadata` is the public reader observers and
+# capability code consume.
+from .metadata import (
+    current_invocation_metadata,
+    set_invocation_metadata,
+)
+
 __all__ = [
     "LLM_NAMESPACE",
     "LlmEventPayload",
@@ -50,5 +60,7 @@
     "current_dispatch",
     "current_fan_out_index",
     "current_invocation_id",
+    "current_invocation_metadata",
     "current_namespace_prefix",
+    "set_invocation_metadata",
 ]

src/openarmature/observability/langfuse/observer.py

@@ -24,6 +24,7 @@
 
 import json
 import uuid
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, cast
 
@@ -80,6 +81,28 @@ def _empty_str_frozenset() -> frozenset[str]:
     return frozenset()
 
 
+def _apply_caller_metadata(metadata: dict[str, Any], caller_metadata: Mapping[str, Any]) -> None:
+    """Merge caller-supplied invocation metadata into a Trace's or
+    Observation's metadata bag at top level per observability §8.4.1
+    + §8.4.2 (proposal 0034).
+
+    Top-level placement is by spec: Langfuse UI filters on
+    ``metadata.<key>`` directly, so caller-supplied entries become
+    siblings to ``correlation_id`` / ``entry_node`` rather than
+    nested under a ``user`` sub-object.
+
+    Reserved-key collision with §8.4.1 / §8.4.2 keys
+    (``correlation_id``, ``entry_node``, ``spec_version``,
+    ``namespace``, etc.) is not currently checked here: the spec
+    permits the rejection to happen at either boundary, and the
+    ``invoke()`` API-boundary validation already rejects
+    ``openarmature.*`` / ``gen_ai.*`` prefixed keys. Per-Langfuse-
+    backend collision rejection is queued as a follow-up.
+    """
+    for key, value in caller_metadata.items():
+        metadata[key] = value
+
+
 def _subgraph_identity_at(event: NodeEvent, depth: int) -> str:
     """Return the compiled-subgraph identity for the wrapper at the
     given 1-based namespace depth, or the empty string when no
@@ -366,6 +389,7 @@ def _open_trace(self, invocation_id: str, correlation_id: str | None, event: Nod
         }
         if correlation_id is not None:
             metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         # §8.6 trace name: caller-supplied invocation label takes
         # precedence; entry-node name is the spec-recommended fallback.
         # The caller-supplied path lands in proposal 0034 (PR 4) — for
@@ -533,6 +557,7 @@ def _open_subgraph_observation(
         }
         if correlation_id is not None:
             metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         handle = self.client.span(
             trace_id=inv_state.trace_id,
             name=prefix[-1],
@@ -566,6 +591,7 @@ def _open_fan_out_instance_dispatch_observation(
         }
         if correlation_id is not None:
             metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         handle = self.client.span(
             trace_id=inv_state.trace_id,
             name=prefix[-1],
@@ -622,6 +648,7 @@ def _open_detached_subgraph_trace(
         }
         if correlation_id is not None:
             link_metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(link_metadata, event.caller_invocation_metadata)
         parent_observation_id: str | None = None
         for plen in range(len(prefix) - 1, 0, -1):
             outer = prefix[:plen]
@@ -646,6 +673,7 @@ def _open_detached_subgraph_trace(
         detached_metadata: dict[str, Any] = {"detached_from_invocation_id": inv_state.trace_id}
         if correlation_id is not None:
             detached_metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(detached_metadata, event.caller_invocation_metadata)
         identity = _subgraph_identity_at(event, len(prefix))
         # The detached trace's wrapper observation IS the migrated
         # SubgraphNode wrapper. Per the resolution in coord thread
@@ -673,6 +701,7 @@ def _open_detached_subgraph_trace(
         }
         if correlation_id is not None:
             dispatch_metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(dispatch_metadata, event.caller_invocation_metadata)
         handle = self.client.span(
             trace_id=detached_trace_id,
             name=wrapper_obs_name,
@@ -715,6 +744,7 @@ def _open_detached_fan_out_instance_trace(
             }
             if correlation_id is not None:
                 link_metadata["correlation_id"] = correlation_id
+            _apply_caller_metadata(link_metadata, event.caller_invocation_metadata)
             fan_out_open.handle.update(metadata=link_metadata)
         # Open the detached Trace + per-instance dispatch observation.
         detached_metadata: dict[str, Any] = {
@@ -723,6 +753,7 @@ def _open_detached_fan_out_instance_trace(
         }
         if correlation_id is not None:
             detached_metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(detached_metadata, event.caller_invocation_metadata)
         self.client.trace(
             id=detached_trace_id,
             name=prefix[-1],
@@ -736,6 +767,7 @@ def _open_detached_fan_out_instance_trace(
         }
         if correlation_id is not None:
             dispatch_metadata["correlation_id"] = correlation_id
+        _apply_caller_metadata(dispatch_metadata, event.caller_invocation_metadata)
         handle = self.client.span(
             trace_id=detached_trace_id,
             name=prefix[-1],
@@ -866,6 +898,7 @@ def _observation_metadata(self, event: NodeEvent, correlation_id: str | None) ->
             metadata["fan_out_item_count"] = cfg.item_count
             metadata["fan_out_concurrency"] = 0 if cfg.concurrency is None else cfg.concurrency
             metadata["fan_out_error_policy"] = cfg.error_policy
+        _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         return metadata
 
     # ------------------------------------------------------------------
@@ -1005,6 +1038,7 @@ def _llm_metadata_and_payload(
         active_group = payload.active_prompt_group
         if active_group is not None:
             metadata["prompt_group_name"] = active_group.group_name
+        _apply_caller_metadata(metadata, payload.caller_invocation_metadata)
 
         model_parameters: dict[str, Any] = {}
         request_params = payload.request_params or {}

src/openarmature/observability/llm_event.py

@@ -37,7 +37,7 @@
 
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 # Sentinel namespace the LLM provider emits to signal "this is an LLM
 # event, not a regular node event." Backend mappings (the OTel observer
@@ -112,6 +112,18 @@ class LlmEventPayload(BaseModel):
     response_id: str | None = None
     response_model: str | None = None
     genai_system: str = "openai"
+    # Per proposal 0034 / observability §3.4 + §5.6: snapshot of
+    # caller-supplied invocation metadata captured at LLM-event
+    # dispatch time (in the calling node's Context). Backend
+    # observers read from the snapshot rather than re-reading the
+    # ContextVar at observer time — the OTel + Langfuse observers
+    # run on the engine's ``deliver_loop`` task whose Context is
+    # frozen at invoke time, so mid-invocation augmentations made
+    # by node bodies running in the main engine task are NOT visible
+    # there. The snapshot pattern mirrors the existing
+    # ``calling_namespace_prefix`` / ``calling_attempt_index`` /
+    # ``calling_fan_out_index`` fields.
+    caller_invocation_metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 __all__ = ["LLM_NAMESPACE", "LlmEventPayload"]