observability: phase 6.1 PR-C.2 — proposal 0013 + v0.10.0

Implement spec graph-engine + observability v0.10.0 (proposal
0013): non-detached fan-out instances now synthesize per-instance
dispatch spans nested between the fan-out node span and the
inner-node spans (mirroring the detached path's layout, but
landing in the parent trace rather than a fresh trace_id). The
fan-out node span carries the four §5.4 attributes (item_count /
concurrency / error_policy / parent_node_name) sourced from the
new ``NodeEvent.fan_out_config`` field.

Mechanism: typed ``FanOutEventConfig`` dataclass on the canonical
event payload (events.py). Engine populates eagerly at fan-out
entry in ``_step_fan_out_node`` — resolves item_count (from
``count`` mode or ``len(items_field)``) + concurrency once,
threads through every dispatch site (started, all four error-path
completed, deferred success-case completed). Pre-resolved values
flow into ``FanOutNode.run_with_context`` via new
``pre_resolved_count`` / ``pre_resolved_concurrency`` kwargs so
callable resolvers fire at most once per fan-out attempt scope
(retry middleware re-uses the outer-scope resolution).

The four contract corrections from PR #34's CoPilot review:

1. ``concurrency: int | None`` on the canonical payload (positive
   int or None for unbounded per pipeline-utilities §9.2). The
   ``None → 0`` translation lives only at the OTel attribute
   layer in ``_node_attrs`` per observability §5.4's bare-int
   sentinel convention.
2. All four ``FanOutEventConfig`` fields structurally required
   (frozen dataclass; partial resolution would raise on
   construction). The implementation-defined corner from
   spec-thread 03 (config resolution itself failing) doesn't
   surface in current fixtures; natural error propagation
   covers it.
3. Retried fan-out attempts carry ``fan_out_config``: the
   resolved config is constructed in the outer scope of
   ``_step_fan_out_node``; retry middleware re-enters
   ``innermost`` and re-uses the same outer-scope reference. No
   per-attempt re-resolution; partition is by node type, not
   event category.
4. ``parent_node_name`` is on per-instance spans only; the
   observer caches it from the fan-out node's ``started`` event
   in ``_InvState.fan_out_parent_node_name`` keyed by namespace
   prefix, applies at per-instance dispatch span synthesis, and
   clears at the fan-out node's ``completed`` event.

Observer changes (``observer.py``):

- ``_InvState`` gains ``fan_out_instance_spans`` (per-instance
  dispatch span store, keyed by ``prefix + (str(fan_out_index),)``)
  and ``fan_out_parent_node_name`` (cache).
- ``_handle_started`` populates the cache when fan-out node
  events land.
- ``_handle_completed`` closes per-instance dispatch spans on the
  fan-out node's own completion (children-before-parents
  ordering) and clears the cache entry.
- ``_sync_subgraph_spans`` routes non-detached fan-out instance
  namespaces through the new
  ``_open_fan_out_instance_dispatch_span`` helper instead of
  opening a shared subgraph span at the prefix.
- ``_resolve_parent_context`` finds the per-instance dispatch
  span before falling through, so inner-node events parent
  under their own per-instance dispatch span (not the shared
  fan-out node span).
- ``_node_attrs`` populates the three §5.4 fan-out node span
  attributes from ``event.fan_out_config`` when present.
- ``_drain_inv_state`` extended to drain
  ``fan_out_instance_spans`` in child→parent order.

Pin sites bumped to v0.10.0 (three-place sync per CLAUDE.md):
``openarmature-spec`` submodule pointer to ``ff86945``,
``pyproject.toml`` ``tool.openarmature.spec_version``,
``src/openarmature/__init__.py`` ``__spec_version__``,
``tests/test_smoke.py`` drift-guard literal.
``OTelObserver.spec_version`` updates automatically via PR-A's
``_read_spec_version``.

Tests:

- ``tests/conformance/test_observability.py`` removes
  ``006-otel-fan-out-instance-attribution`` from
  ``_DEFERRED_FIXTURES`` and adds ``_run_fixture_006`` driver.
  Verifies the fixture's expected tree: 1 fan-out NODE span
  with item_count/concurrency/error_policy attributes; 3
  per-instance dispatch spans with fan_out_index 0..2 and
  parent_node_name="process"; 3 compute spans (one per
  instance) parented under their own per-instance dispatch
  span.
- ``tests/conformance/adapter.py``:
  ``_TracingFanOutNode.run_with_context`` accepts and forwards
  the new pre-resolved kwargs (pyright override-compatibility).
- Existing fan-out tests (resolution-once-per-entry,
  concurrency-callable-once-per-entry, instance-middleware
  retry, fan-in ordering, etc.) stay green — pre-resolved-
  values plumbing didn't introduce double-resolution.

399 tests pass (was 392; net +7 from new fan-out path
verification + fixture 006 going from skip to pass + harness
work). 3 skipped (was 4; only fixture 010 remains, which is
PR-C.3's scope). Pyright clean.

PR-C.3 (observer ``prepare_sync`` + fixture 010) sits
independently behind its own architectural piece; lands in
either order. Phase 6.1 closes when both merge.

Author: chris-colinsky

openarmature-spec

@@ -36,7 +36,7 @@ Repository = "https://github.com/LunarCommand/openarmature-python"
 Specification = "https://github.com/LunarCommand/openarmature-spec"
 
 [tool.openarmature]
-spec_version = "0.9.0"
+spec_version = "0.10.0"
 
 [dependency-groups]
 dev = [

pyproject.toml

@@ -1,4 +1,4 @@
 """OpenArmature — workflow framework for LLM pipelines and tool-calling agents."""
 
 __version__ = "0.4.0rc0"
-__spec_version__ = "0.9.0"
+__spec_version__ = "0.10.0"

src/openarmature/__init__.py

@@ -67,8 +67,8 @@
     RuntimeGraphError,
     StateValidationError,
 )
-from .events import NodeEvent
-from .fan_out import FanOutNode
+from .events import FanOutEventConfig, NodeEvent
+from .fan_out import FanOutNode, _resolve_concurrency, _resolve_count
 from .middleware import ChainCall, Middleware, compose_chain
 from .nodes import Node
 from .observer import (
@@ -913,6 +913,28 @@ async def _step_fan_out_node(
         # hardcoded 0.
         attempt_counter: list[int] = [0]
 
+        # Resolve the fan-out config eagerly so the resolved values
+        # ride on every fan-out node event (per spec proposal 0013,
+        # v0.10.0: ``fan_out_config`` is populated on fan-out node
+        # events including retried attempts). For ``items_field``
+        # mode the count is ``len(parent_state.<items_field>)``; for
+        # ``count`` mode it's ``_resolve_count``. ``_resolve_concurrency``
+        # is pure regardless. Repeating these inside
+        # ``FanOutNode.run_with_context`` is cheap and matches the
+        # values surfaced here.
+        if node.config.items_field is not None:
+            items_attr: Any = getattr(state, node.config.items_field, [])
+            item_count = len(cast("list[Any]", items_attr)) if isinstance(items_attr, list) else 0
+        else:
+            item_count = _resolve_count(current, node.config, state)
+        concurrency_resolved: int | None = _resolve_concurrency(current, node.config, state)
+        fan_out_event_config = FanOutEventConfig(
+            item_count=item_count,
+            concurrency=concurrency_resolved,
+            error_policy=node.config.error_policy,
+            parent_node_name=current,
+        )
+
         # Cell holding the FINAL successful attempt's
         # (attempt_index, pre_state, merged); see same comment in
         # ``_step_function_node``.
@@ -923,12 +945,32 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             attempt_counter[0] += 1
             attempt_token = _set_attempt_index(attempt_index)
             try:
-                self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
+                self._dispatch_started(
+                    context,
+                    current,
+                    namespace,
+                    step,
+                    s,
+                    attempt_index=attempt_index,
+                    fan_out_config=fan_out_event_config,
+                )
                 try:
-                    partial = await node.run_with_context(s, context)
+                    partial = await node.run_with_context(
+                        s,
+                        context,
+                        pre_resolved_count=item_count,
+                        pre_resolved_concurrency=(concurrency_resolved,),
+                    )
                 except RuntimeGraphError as e:
                     self._dispatch_completed(
-                        context, current, namespace, step, s, error=e, attempt_index=attempt_index
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=e,
+                        attempt_index=attempt_index,
+                        fan_out_config=fan_out_event_config,
                     )
                     raise
                 except Exception as e:
@@ -941,14 +983,22 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                         s,
                         error=wrapped,
                         attempt_index=attempt_index,
+                        fan_out_config=fan_out_event_config,
                     )
                     raise wrapped from e
 
                 try:
                     merged = _merge_partial(s, partial, self.reducers, current)
                 except (ReducerError, StateValidationError) as e:
                     self._dispatch_completed(
-                        context, current, namespace, step, s, error=e, attempt_index=attempt_index
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=e,
+                        attempt_index=attempt_index,
+                        fan_out_config=fan_out_event_config,
                     )
                     raise
 
@@ -1021,6 +1071,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None:
                     final_pre_state,
                     post_state=final_merged,
                     attempt_index=final_attempt_index,
+                    fan_out_config=fan_out_event_config,
                 )
             else:
                 self._dispatch_completed(
@@ -1031,6 +1082,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None:
                     final_pre_state,
                     error=edge_error,
                     attempt_index=final_attempt_index,
+                    fan_out_config=fan_out_event_config,
                 )
 
         return _StepResult(state=merged_outer, finalize_completed=finalize_completed)
@@ -1044,6 +1096,7 @@ def _dispatch_started(
         pre_state: State,
         *,
         attempt_index: int = 0,
+        fan_out_config: FanOutEventConfig | None = None,
     ) -> None:
         _dispatch(
             context,
@@ -1058,6 +1111,7 @@ def _dispatch_started(
                 parent_states=context.parent_states_prefix,
                 attempt_index=attempt_index,
                 fan_out_index=context.fan_out_index,
+                fan_out_config=fan_out_config,
             ),
         )
 
@@ -1072,6 +1126,7 @@ def _dispatch_completed(
         post_state: State | None = None,
         error: RuntimeGraphError | None = None,
         attempt_index: int = 0,
+        fan_out_config: FanOutEventConfig | None = None,
     ) -> None:
         _dispatch(
             context,
@@ -1086,6 +1141,7 @@ def _dispatch_completed(
                 parent_states=context.parent_states_prefix,
                 attempt_index=attempt_index,
                 fan_out_index=context.fan_out_index,
+                fan_out_config=fan_out_config,
             ),
         )
 

src/openarmature/graph/compiled.py

@@ -16,6 +16,52 @@
 from .state import State
 
 
+@dataclass(frozen=True)
+class FanOutEventConfig:
+    """Spec §6 + §5.4 (per spec proposal 0013, v0.10.0):
+    fan-out node events carry the resolved configuration so backend
+    observers can attribute the fan-out node span (``item_count`` /
+    ``concurrency`` / ``error_policy``) and synthesize per-instance
+    spans with the right ``parent_node_name``.
+
+    Populated ONLY on ``started`` and ``completed`` events for a
+    fan-out node itself (partition by node type, not event category —
+    INCLUDES retried attempts of a fan-out node when retry middleware
+    wraps it). All other events leave ``NodeEvent.fan_out_config``
+    null.
+
+    Field shapes:
+
+    - ``item_count`` — non-negative int. The resolved instance count
+      per pipeline-utilities §9 (matches ``count_field`` value when
+      configured; matches ``len(items_field)`` in items_field mode).
+    - ``concurrency`` — positive int OR ``None`` (unbounded). Per
+      pipeline-utilities §9.2: zero or negative is rejected at config
+      resolution time as ``fan_out_invalid_concurrency``. Backend
+      mappings translate ``None`` to a sentinel at the attribute layer
+      (e.g., ``openarmature.fan_out.concurrency = 0`` per
+      observability §5.4) — that translation is observer-internal,
+      not engine-internal.
+    - ``error_policy`` — one of ``"fail_fast"`` or ``"collect"`` per
+      pipeline-utilities §9.4.
+    - ``parent_node_name`` — the fan-out node's name in the parent
+      graph. Carried here for caching by backend observers when
+      attributing per-instance spans (§5.4 mandates
+      ``openarmature.fan_out.parent_node_name`` on per-instance spans;
+      the engine surfaces the name once on the fan-out node's started
+      event, the observer caches and applies on every per-instance
+      span it synthesizes).
+
+    All four fields MUST be present when ``fan_out_config`` is
+    populated. Only ``concurrency`` is nullable.
+    """
+
+    item_count: int
+    concurrency: int | None
+    error_policy: str
+    parent_node_name: str
+
+
 @dataclass(frozen=True)
 class NodeEvent:
     """A single node-boundary event delivered to observers.
@@ -55,6 +101,9 @@ class NodeEvent:
       retries. `0` for nodes not wrapped by retry middleware.
     - `fan_out_index` is the 0-based index of this fan-out instance among
       its siblings. `None` for nodes not inside a fan-out.
+    - `fan_out_config` carries resolved fan-out configuration on events
+      from a fan-out NODE itself (per spec proposal 0013, v0.10.0). See
+      :class:`FanOutEventConfig`. ``None`` on every other event.
 
     Invariants:
     - On `started` events, `post_state` and `error` MUST both be None.
@@ -72,6 +121,7 @@ class NodeEvent:
     parent_states: tuple[State, ...]
     attempt_index: int = 0
     fan_out_index: int | None = None
+    fan_out_config: FanOutEventConfig | None = None
 
 
-__all__ = ["NodeEvent"]
+__all__ = ["FanOutEventConfig", "NodeEvent"]

src/openarmature/graph/events.py

@@ -107,16 +107,28 @@ async def run_with_context(
         self,
         state: ParentT,
         context: _InvocationContext,
+        *,
+        pre_resolved_count: int | None = None,
+        pre_resolved_concurrency: tuple[int | None] | None = None,
     ) -> Mapping[str, Any]:
         """Execute the fan-out and return the merged partial update.
 
         Per spec §9.1–§9.5: snapshot, resolve count + concurrency,
         build per-instance states, run concurrently with the configured
         error policy, fan-in collected/extra fields, write count_field
         and errors_field if configured.
+
+        ``pre_resolved_count`` / ``pre_resolved_concurrency`` are the
+        proposal-0013 v0.10.0 hooks: when the engine has already
+        resolved the config eagerly to populate
+        ``NodeEvent.fan_out_config`` for the fan-out node's events,
+        it passes the resolved values in so callable resolvers
+        aren't invoked twice. ``pre_resolved_concurrency`` is wrapped
+        in a 1-tuple to disambiguate "caller passed ``None``
+        (unbounded)" from "caller didn't pass anything."
         """
         cfg = self.config
-        instance_states = _build_instance_states(self.name, cfg, state)
+        instance_states = _build_instance_states(self.name, cfg, state, pre_resolved_count=pre_resolved_count)
         instance_count = len(instance_states)
 
         if instance_count == 0:
@@ -127,7 +139,10 @@ async def run_with_context(
             # reducer absorbs as a no-op).
             return _empty_fan_out_partial(cfg)
 
-        max_concurrency = _resolve_concurrency(self.name, cfg, state)
+        if pre_resolved_concurrency is not None:
+            max_concurrency = pre_resolved_concurrency[0]
+        else:
+            max_concurrency = _resolve_concurrency(self.name, cfg, state)
 
         # Per-instance task: build the instance_middleware chain, run
         # the subgraph against it, and return the per-instance partial
@@ -187,13 +202,22 @@ def _build_instance_states(
     node_name: str,
     cfg: FanOutConfig,
     parent_state: Any,
+    *,
+    pre_resolved_count: int | None = None,
 ) -> list[Any]:
     """Project parent state to per-instance subgraph states.
 
     Per spec §9.1:
     - items_field mode: one instance per item, item_field gets the item
     - count mode: ``count`` instances, item_field absent
     - both modes: inputs map parent fields onto subgraph state fields
+
+    ``pre_resolved_count`` (proposal-0013 hook): if the engine has
+    already resolved ``cfg.count`` to populate
+    ``NodeEvent.fan_out_config.item_count``, the resolved value is
+    passed in here so the callable resolver isn't invoked twice.
+    Only consulted in ``count`` mode; ``items_field`` mode reads
+    from the state field unchanged.
     """
     sub_state_cls = cfg.subgraph.state_cls
     parent_dump = parent_state.model_dump()
@@ -226,7 +250,10 @@ def _build_instance_states(
         return instances
 
     # count mode
-    resolved_count = _resolve_count(node_name, cfg, parent_state)
+    if pre_resolved_count is not None:
+        resolved_count = pre_resolved_count
+    else:
+        resolved_count = _resolve_count(node_name, cfg, parent_state)
     instances = []
     for _idx in range(resolved_count):
         init = {}