Wire subgraph_identity through engine + observers (Option A)

Implements the coord-thread `clarify-subgraph-name-semantics`
resolution (msg 02): `metadata.subgraph_name` carries the compiled
subgraph's identity rather than the wrapper node name.

Engine surface:

- `SubgraphNode` gains `subgraph_identity: str | None = None`
  (optional, BC-preserving).
- `FanOutConfig` gains `subgraph_identity: str | None = None` for
  the symmetric fan-out case; threaded through
  `GraphBuilder.add_fan_out_node`.
- `_InvocationContext` gains `subgraph_identities:
  tuple[str | None, ...]`, parallel to `namespace_prefix` — index
  `i` is the identity of the wrapper at `namespace_prefix[i]`.
  `descend_into_subgraph` / `descend_into_fan_out_instance` /
  `descend_into_parallel_branch` extend the chain.
- `NodeEvent.subgraph_identities: tuple[str | None, ...] = ()`
  carries the chain at event-emission time so observers can read
  the identity at each wrapper depth.

Observer surface:

- Langfuse `LangfuseObserver` and OTel `OTelObserver` both gain
  `_subgraph_identity_at(event, depth)` and emit identity (or empty
  string when None) as `metadata.subgraph_name` / `openarmature.
  subgraph.name` on wrapper observations / spans, per-instance
  fan-out dispatch observations, and detached-trace wrapper
  observations.
- Detached subgraph's wrapper observation name now also uses the
  identity (falling back to the wrapper node name when None).
- Langfuse wrapper observation synthesis now carries `namespace`,
  `step`, and `attempt_index` metadata (step from the first inner
  event; attempt_index hardcoded 0 since wrappers don't retry).

Conformance adapter:

- `_TracingSubgraphNode` and `_add_fan_out_node` set
  `subgraph_identity` from the fixture's `subgraphs:` block key
  when compiling, so every fixture-built subgraph carries its
  declared identity.

Updates the pre-existing OTel-side test on fixture 002 to assert
`openarmature.subgraph.name == "inner"` (identity) rather than
`"outer_sub"` (wrapper node name) — the prior assertion was
inconsistent with Option A semantics; the spec-side fixture 002
YAML doesn't assert this attribute, so the python test was a
stricter local check that needed correction.

Langfuse fixtures 031/032/033 stay deferred pending two additional
spec/fixture ambiguities surfaced while wiring this up: (a) the
`step` value on `outer_out` (fixture 031 says 2, but graph-engine
§6 says inner-subgraph node executions increment the same counter
so the engine emits 3) and (b) whether the `namespace` metadata
on detached-trace inner observations should be rewritten to use
the subgraph identity at the wrapper position. Both queued in
coord thread `clarify-subgraph-name-semantics` msg 03.

Author: chris-colinsky

src/openarmature/graph/builder.py

@@ -148,6 +148,7 @@ def add_fan_out_node[ChildT: State](
         instance_middleware: Iterable[Middleware] | None = None,
         errors_field: str | None = None,
         middleware: Iterable[Middleware] | None = None,
+        subgraph_identity: str | None = None,
     ) -> Self:
         """Register a fan-out node.
 
@@ -262,6 +263,7 @@ def add_fan_out_node[ChildT: State](
             extra_outputs=dict(extra_outputs or {}),
             instance_middleware=tuple(instance_middleware or ()),
             errors_field=errors_field,
+            subgraph_identity=subgraph_identity,
         )
         # FanOutNode satisfies the Node[StateT] structural protocol (run
         # returns a partial update; name and middleware are present),

src/openarmature/graph/compiled.py

@@ -1987,6 +1987,7 @@ def _dispatch_started(
                 fan_out_index=context.fan_out_index,
                 fan_out_config=fan_out_config,
                 branch_name=current_branch_name(),
+                subgraph_identities=context.subgraph_identities,
             ),
         )
 
@@ -2020,6 +2021,7 @@ def _dispatch_completed(
                 fan_out_index=context.fan_out_index,
                 fan_out_config=fan_out_config,
                 branch_name=current_branch_name(),
+                subgraph_identities=context.subgraph_identities,
             ),
         )
 
@@ -2202,5 +2204,6 @@ async def _maybe_save_checkpoint(
                 parent_states=context.parent_states_prefix,
                 attempt_index=attempt_index,
                 fan_out_index=None,
+                subgraph_identities=context.subgraph_identities,
             ),
         )

src/openarmature/graph/events.py

@@ -191,6 +191,20 @@ class NodeEvent:
     # simultaneously when a branch's subgraph contains a fan-out
     # (and vice versa).
     branch_name: str | None = None
+    # Per observability §5.3 + the coord-thread
+    # ``clarify-subgraph-name-semantics`` resolution: chain of
+    # compiled-subgraph identities parallel to the wrapper-depth
+    # positions of ``namespace``. Index ``i`` is the identity for
+    # the wrapper at ``namespace[i]`` (or ``None`` when that
+    # wrapper has no tracked identity); chain length equals the
+    # depth of wrapper nesting (always ``< len(namespace)`` since
+    # the last element of ``namespace`` is the current node, not
+    # a wrapper). Observers read by depth and emit it as
+    # ``observation.metadata.subgraph_name`` (Langfuse) /
+    # ``openarmature.subgraph.name`` (OTel), falling back to the
+    # empty string when ``None`` per §5.3's "if the implementation
+    # tracks one" clause.
+    subgraph_identities: tuple[str | None, ...] = ()
 
 
 __all__ = ["FanOutEventConfig", "NodeEvent"]

src/openarmature/graph/fan_out.py

@@ -80,6 +80,15 @@ class FanOutConfig:
     extra_outputs: Mapping[str, str] = field(default_factory=dict[str, str])
     instance_middleware: tuple[Middleware, ...] = ()
     errors_field: str | None = None
+    # The identity of the compiled inner subgraph (the key under
+    # which the subgraph is declared in a ``subgraphs:`` registry).
+    # Threaded onto every per-instance event so observers can emit
+    # ``observation.metadata.subgraph_name`` on each per-instance
+    # dispatch observation (Langfuse) /
+    # ``openarmature.subgraph.name`` on the corresponding span
+    # (OTel). Optional and BC-preserving — direct callers that don't
+    # supply it get the empty-string fallback per observability §5.3.
+    subgraph_identity: str | None = None
 
 
 @dataclass(frozen=True)
@@ -271,6 +280,7 @@ async def run_instance(idx: int, instance_state: ChildT) -> Mapping[str, Any]:
                 parent_state=state,
                 sub_attached=tuple(cfg.subgraph._attached_observers),  # noqa: SLF001
                 fan_out_index=idx,
+                subgraph_identity=cfg.subgraph_identity,
             )
 
             async def innermost(s: ChildT) -> Mapping[str, Any]:

src/openarmature/graph/observer.py

@@ -336,6 +336,18 @@ class _InvocationContext:
     step_counter: list[int] = field(default_factory=lambda: [0])
     namespace_prefix: tuple[str, ...] = ()
     parent_states_prefix: tuple[State, ...] = ()
+    # Per observability §5.3 + the coord-thread `clarify-subgraph-name-
+    # semantics` resolution. Parallel to ``namespace_prefix`` — index
+    # ``i`` is the compiled-subgraph identity for the wrapper at
+    # ``namespace_prefix[i]``, or ``None`` for wrappers constructed
+    # without an identity. Used by observers to emit
+    # ``metadata.subgraph_name`` (Langfuse) and
+    # ``openarmature.subgraph.name`` (OTel) on the wrapper observation
+    # / span at each depth. The chain shape lets nested subgraphs
+    # carry distinct identities at distinct depths even though
+    # v0.10.0's conformance fixtures only exercise single-level
+    # nesting.
+    subgraph_identities: tuple[str | None, ...] = ()
     # Per pipeline-utilities §9 + graph-engine §6: nodes inside a
     # fan-out instance fire events tagged with the instance's 0-based
     # index. Set when descending into a fan-out instance, inherited
@@ -426,6 +438,8 @@ def descend_into_subgraph(
         subgraph_node_name: str,
         parent_state: State,
         sub_attached: tuple[SubscribedObserver, ...],
+        *,
+        subgraph_identity: str | None = None,
     ) -> _InvocationContext:
         """Build the context for a subgraph-as-node call.
 
@@ -447,6 +461,7 @@ def descend_into_subgraph(
             step_counter=self.step_counter,
             namespace_prefix=self.namespace_prefix + (subgraph_node_name,),
             parent_states_prefix=self.parent_states_prefix + (parent_state,),
+            subgraph_identities=self.subgraph_identities + (subgraph_identity,),
             fan_out_index=self.fan_out_index,
             invocation_id=self.invocation_id,
             correlation_id=self.correlation_id,
@@ -466,6 +481,8 @@ def descend_into_fan_out_instance(
         parent_state: State,
         sub_attached: tuple[SubscribedObserver, ...],
         fan_out_index: int,
+        *,
+        subgraph_identity: str | None = None,
     ) -> _InvocationContext:
         """Build the context for one fan-out instance's subgraph invocation.
 
@@ -491,6 +508,7 @@ def descend_into_fan_out_instance(
             step_counter=self.step_counter,
             namespace_prefix=self.namespace_prefix + (fan_out_node_name,),
             parent_states_prefix=self.parent_states_prefix + (parent_state,),
+            subgraph_identities=self.subgraph_identities + (subgraph_identity,),
             fan_out_index=fan_out_index,
             invocation_id=self.invocation_id,
             correlation_id=self.correlation_id,
@@ -541,6 +559,12 @@ def descend_into_parallel_branch(
             step_counter=self.step_counter,
             namespace_prefix=self.namespace_prefix + (parallel_branches_node_name,),
             parent_states_prefix=self.parent_states_prefix + (parent_state,),
+            # Parallel-branches don't reify a single inner subgraph
+            # identity at the wrapper position — each branch can hold a
+            # different subgraph — so we extend the chain with ``None``
+            # at this depth. Per-branch identity handling (if ever
+            # needed) is a future addition.
+            subgraph_identities=self.subgraph_identities + (None,),
             fan_out_index=self.fan_out_index,
             invocation_id=self.invocation_id,
             correlation_id=self.correlation_id,

src/openarmature/graph/subgraph.py

@@ -50,6 +50,17 @@ class SubgraphNode[ParentT: State, ChildT: State]:
         default_factory=FieldNameMatching[ParentT, ChildT]
     )
     middleware: tuple[Middleware, ...] = field(default_factory=tuple[Middleware, ...])
+    # The compiled subgraph's identity (the registry key under which
+    # the subgraph is declared, distinct from the wrapper node's
+    # ``name`` in the parent graph). Optional and BC-preserving:
+    # callers that don't pass it get an empty string emitted as the
+    # observability §5.3 ``subgraph_name`` attribute (matching the
+    # spec's "if the implementation tracks one" fallback). Setting
+    # it lets dashboards filter or aggregate across observations from
+    # the same compiled subgraph wrapped under different node names
+    # (e.g., a ``validator`` subgraph used as both ``validate_input``
+    # and ``validate_output``).
+    subgraph_identity: str | None = None
 
     async def run(
         self,
@@ -116,6 +127,7 @@ async def run(
                 subgraph_node_name=self.name,
                 parent_state=state,
                 sub_attached=tuple(self.compiled._attached_observers),
+                subgraph_identity=self.subgraph_identity,
             )
             sub_final = await self.compiled._invoke(sub_initial, child_context)
         return self.projection.project_out(sub_final, state, self.compiled.state_cls)

src/openarmature/observability/langfuse/observer.py

@@ -80,6 +80,28 @@ def _empty_str_frozenset() -> frozenset[str]:
     return frozenset()
 
 
+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
+    identity is tracked at that depth.
+
+    Per observability §5.3 + the coord-thread
+    ``clarify-subgraph-name-semantics`` resolution: the empty-string
+    fallback matches the spec's "if the implementation tracks one"
+    clause for implementations / direct ``SubgraphNode(...)`` callers
+    that don't wire an identity through. Conformance fixtures
+    031/032/033 lock identity as the required value; the empty-string
+    path keeps direct callers conformant with §5.3 but failing those
+    fixtures.
+    """
+    idx = depth - 1
+    if 0 <= idx < len(event.subgraph_identities):
+        identity = event.subgraph_identities[idx]
+        if identity is not None:
+            return identity
+    return ""
+
+
 @dataclass
 class _InvState:
     """Per-invocation state, isolated by invocation_id.
@@ -453,7 +475,7 @@ def _sync_subgraph_observations(
             # configured detached_subgraphs name → mint a fresh
             # detached Trace + open the dispatch observation in it.
             if depth == 1 and prefix[0] in self.detached_subgraphs:
-                self._open_detached_subgraph_trace(inv_state, correlation_id, prefix)
+                self._open_detached_subgraph_trace(inv_state, correlation_id, prefix, event)
                 continue
             # Detached fan-out: the fan-out instance gets its own
             # Trace per spec §8.5. The fan-out node's Span observation
@@ -478,13 +500,14 @@ def _sync_subgraph_observations(
                 self._open_fan_out_instance_dispatch_observation(inv_state, correlation_id, prefix, event)
                 continue
             # Plain non-detached subgraph dispatch.
-            self._open_subgraph_observation(inv_state, correlation_id, prefix)
+            self._open_subgraph_observation(inv_state, correlation_id, prefix, event)
 
     def _open_subgraph_observation(
         self,
         inv_state: _InvState,
         correlation_id: str | None,
         prefix: tuple[str, ...],
+        event: NodeEvent,
     ) -> None:
         # Parent is the nearest enclosing subgraph dispatch (if any),
         # else None (the Trace is the implicit parent for top-level
@@ -496,7 +519,18 @@ def _open_subgraph_observation(
             if sg is not None:
                 parent_observation_id = sg.handle.id
                 break
-        metadata: dict[str, Any] = {"subgraph_name": prefix[-1]}
+        # Subgraph wrappers don't dispatch their own events, so the
+        # synthetic wrapper observation inherits its scalar metadata
+        # from the FIRST inner event that triggered the synthesis.
+        # ``attempt_index`` is hardcoded to 0: the wrapper has no
+        # engine-managed retry counter of its own (inner nodes own
+        # their own attempt_index independently).
+        metadata: dict[str, Any] = {
+            "namespace": list(prefix),
+            "step": event.step,
+            "attempt_index": 0,
+            "subgraph_name": _subgraph_identity_at(event, len(prefix)),
+        }
         if correlation_id is not None:
             metadata["correlation_id"] = correlation_id
         handle = self.client.span(
@@ -519,9 +553,16 @@ def _open_fan_out_instance_dispatch_observation(
         fan_out_open = self._find_fan_out_node_observation(inv_state, prefix)
         parent_observation_id = fan_out_open.handle.id if fan_out_open is not None else None
         parent_node_name = inv_state.fan_out_parent_node_name.get(prefix, prefix[-1])
+        # Per-instance dispatch is synthesized from the first inner
+        # event inside the instance subtree; inherit scalar metadata
+        # from that event (same pattern as ``_open_subgraph_observation``).
         metadata: dict[str, Any] = {
+            "namespace": list(prefix),
+            "step": event.step,
+            "attempt_index": 0,
             "fan_out_parent_node_name": parent_node_name,
             "fan_out_index": event.fan_out_index,
+            "subgraph_name": _subgraph_identity_at(event, len(prefix)),
         }
         if correlation_id is not None:
             metadata["correlation_id"] = correlation_id
@@ -539,6 +580,7 @@ def _open_detached_subgraph_trace(
         inv_state: _InvState,
         correlation_id: str | None,
         prefix: tuple[str, ...],
+        event: NodeEvent,
     ) -> None:
         # Mint a fresh Trace for the detached subtree. The main Trace's
         # dispatch observation surfaces the link via
@@ -604,16 +646,26 @@ 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
-        self.client.trace(id=detached_trace_id, name=prefix[-1], metadata=detached_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
+        # ``clarify-subgraph-name-semantics`` and fixture 033's
+        # expected shape, its name and ``metadata.subgraph_name`` use
+        # the compiled-subgraph identity (e.g., ``"long_running_workflow"``)
+        # rather than the wrapper node name. Falls back to the wrapper
+        # node name when identity is empty (observability §5.3's
+        # "if the implementation tracks one" clause).
+        wrapper_obs_name = identity or prefix[-1]
+        self.client.trace(id=detached_trace_id, name=wrapper_obs_name, metadata=detached_metadata)
         dispatch_metadata: dict[str, Any] = {
-            "subgraph_name": prefix[-1],
+            "subgraph_name": identity,
             "detached": True,
         }
         if correlation_id is not None:
             dispatch_metadata["correlation_id"] = correlation_id
         handle = self.client.span(
             trace_id=detached_trace_id,
-            name=prefix[-1],
+            name=wrapper_obs_name,
             metadata=dispatch_metadata,
             parent_observation_id=None,
         )

src/openarmature/observability/otel/observer.py

@@ -148,6 +148,25 @@ def _read_spec_version() -> str:
     return __spec_version__
 
 
+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
+    identity is tracked at that depth.
+
+    Per observability §5.3 + the coord-thread
+    ``clarify-subgraph-name-semantics`` resolution: empty-string
+    fallback matches the spec's "if the implementation tracks one"
+    clause for callers using ``SubgraphNode(name=..., compiled=...)``
+    without supplying ``subgraph_identity``.
+    """
+    idx = depth - 1
+    if 0 <= idx < len(event.subgraph_identities):
+        identity = event.subgraph_identities[idx]
+        if identity is not None:
+            return identity
+    return ""
+
+
 def _empty_str_frozenset() -> frozenset[str]:
     """Typed empty frozenset factory for ``detached_subgraphs`` /
     ``detached_fan_outs`` defaults."""
@@ -1033,7 +1052,7 @@ def _sync_subgraph_spans(
             # If this prefix's first segment is configured as a
             # detached subgraph, mint a fresh trace.
             if depth == 1 and prefix[0] in self.detached_subgraphs:
-                self._open_detached_subgraph_root(inv_state, invocation_id, correlation_id, prefix)
+                self._open_detached_subgraph_root(inv_state, invocation_id, correlation_id, prefix, event)
                 continue
             # If this is a fan-out instance namespace (event.fan_out_index
             # populated, prefix == namespace[:1]), and the fan-out
@@ -1057,14 +1076,15 @@ def _sync_subgraph_spans(
             ):
                 self._open_fan_out_instance_dispatch_span(inv_state, correlation_id, prefix, event)
                 continue
-            self._open_subgraph_span(inv_state, invocation_id, correlation_id, prefix)
+            self._open_subgraph_span(inv_state, invocation_id, correlation_id, prefix, event)
 
     def _open_subgraph_span(
         self,
         inv_state: _InvState,
         invocation_id: str,
         correlation_id: str | None,
         prefix: tuple[str, ...],
+        event: NodeEvent,
     ) -> None:
         """Open a synthetic subgraph dispatch span for the given
         namespace prefix. Parent is the next-outer subgraph span (or
@@ -1088,7 +1108,7 @@ def _open_subgraph_span(
                 parent_ctx = set_span_in_context(inv.span)
         attrs: dict[str, Any] = {
             "openarmature.node.name": prefix[-1],
-            "openarmature.subgraph.name": prefix[-1],
+            "openarmature.subgraph.name": _subgraph_identity_at(event, len(prefix)),
         }
         if correlation_id is not None:
             attrs["openarmature.correlation_id"] = correlation_id
@@ -1114,6 +1134,7 @@ def _open_detached_subgraph_root(
         invocation_id: str,
         correlation_id: str | None,
         prefix: tuple[str, ...],
+        event: NodeEvent,
     ) -> None:
         """Mint a fresh trace for a detached subgraph entry. The
         detached root span lives in the new trace; the parent trace's
@@ -1141,7 +1162,7 @@ def _open_detached_subgraph_root(
             parent_ctx_for_dispatch = set_span_in_context(inv.span)
         attrs_parent: dict[str, Any] = {
             "openarmature.node.name": prefix[-1],
-            "openarmature.subgraph.name": prefix[-1],
+            "openarmature.subgraph.name": _subgraph_identity_at(event, len(prefix)),
         }
         if correlation_id is not None:
             attrs_parent["openarmature.correlation_id"] = correlation_id
@@ -1264,6 +1285,7 @@ def _open_fan_out_instance_dispatch_span(
             "openarmature.node.name": prefix[-1],
             "openarmature.fan_out.parent_node_name": parent_node_name,
             "openarmature.node.fan_out_index": event.fan_out_index,
+            "openarmature.subgraph.name": _subgraph_identity_at(event, len(prefix)),
         }
         if correlation_id is not None:
             attrs["openarmature.correlation_id"] = correlation_id