From b3fa899649eca7fd03cf75a5a5cf98595a73acee Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 18:44:34 -0700
Subject: [PATCH 1/6] =?UTF-8?q?observability:=20phase=206.1=20PR-C.2=20?=
 =?UTF-8?q?=E2=80=94=20proposal=200013=20+=20v0.10.0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 openarmature-spec                             |   2 +-
 pyproject.toml                                |   2 +-
 src/openarmature/__init__.py                  |   2 +-
 src/openarmature/graph/compiled.py            |  68 +++++++-
 src/openarmature/graph/events.py              |  52 +++++-
 src/openarmature/graph/fan_out.py             |  33 +++-
 .../observability/otel/observer.py            | 155 +++++++++++++++++-
 tests/conformance/adapter.py                  |  10 +-
 tests/conformance/test_observability.py       | 111 ++++++++++++-
 tests/test_smoke.py                           |   2 +-
 10 files changed, 415 insertions(+), 22 deletions(-)

diff --git a/openarmature-spec b/openarmature-spec
index 78fe9b4..ff86945 160000
--- a/openarmature-spec
+++ b/openarmature-spec
@@ -1 +1 @@
-Subproject commit 78fe9b4b0f53deafb89716ad0ffe6151d175db4a
+Subproject commit ff86945747c9d4767e5ededad62bab1c9c4e244a
diff --git a/pyproject.toml b/pyproject.toml
index 16591da..ef9c6cb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -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 = [
diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py
index 74cf11d..b0e8dc4 100644
--- a/src/openarmature/__init__.py
+++ b/src/openarmature/__init__.py
@@ -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"
diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py
index f222eb6..201060e 100644
--- a/src/openarmature/graph/compiled.py
+++ b/src/openarmature/graph/compiled.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,6 +983,7 @@ 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
 
@@ -948,7 +991,14 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                     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,
             ),
         )
 
diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py
index 205fa1d..67595c4 100644
--- a/src/openarmature/graph/events.py
+++ b/src/openarmature/graph/events.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"]
diff --git a/src/openarmature/graph/fan_out.py b/src/openarmature/graph/fan_out.py
index d97e278..97790c4 100644
--- a/src/openarmature/graph/fan_out.py
+++ b/src/openarmature/graph/fan_out.py
@@ -107,6 +107,9 @@ 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.
 
@@ -114,9 +117,18 @@ async def run_with_context(
         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,6 +202,8 @@ 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.
 
@@ -194,6 +211,13 @@ def _build_instance_states(
     - 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 = {}
diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py
index 5960a2f..73d8551 100644
--- a/src/openarmature/observability/otel/observer.py
+++ b/src/openarmature/observability/otel/observer.py
@@ -131,6 +131,25 @@ class _InvState:
     subgraph_spans: dict[tuple[str, ...], _OpenSpan] = field(default_factory=dict[tuple[str, ...], _OpenSpan])
     detached_roots: dict[tuple[str, ...], _OpenSpan] = field(default_factory=dict[tuple[str, ...], _OpenSpan])
     fan_out_instance_root_prefixes: set[tuple[str, ...]] = field(default_factory=set[tuple[str, ...]])
+    # Per spec §5.4 + proposal 0013 (v0.10.0): non-detached fan-outs
+    # synthesize per-instance dispatch spans nested between the fan-out
+    # node span and the inner-node spans. Keyed by ``prefix +
+    # (str(fan_out_index),)`` mirroring the detached path's keying in
+    # ``detached_roots``. Lives in the parent trace (not a fresh
+    # trace_id, unlike detached). Closed when the fan-out node's
+    # ``completed`` event fires.
+    fan_out_instance_spans: dict[tuple[str, ...], _OpenSpan] = field(
+        default_factory=dict[tuple[str, ...], _OpenSpan]
+    )
+    # ``parent_node_name`` cache (per spec proposal 0013 correction #4):
+    # the fan-out node's name surfaces on its own ``started`` event via
+    # ``NodeEvent.fan_out_config.parent_node_name``. The observer caches
+    # it keyed by the fan-out node's namespace prefix so when
+    # ``_open_fan_out_instance_dispatch_span`` synthesizes a per-instance
+    # dispatch span, it can attach
+    # ``openarmature.fan_out.parent_node_name`` even though the inner
+    # event itself doesn't carry ``fan_out_config``.
+    fan_out_parent_node_name: dict[tuple[str, ...], str] = field(default_factory=dict[tuple[str, ...], str])
 
 
 @dataclass
@@ -253,6 +272,15 @@ def _handle_started(self, event: NodeEvent) -> None:
         if invocation_id not in self._invocation_span:
             self._open_invocation_span(invocation_id, correlation_id, event)
 
+        # Per spec proposal 0013 (v0.10.0): cache the fan-out node's
+        # ``parent_node_name`` from its ``started`` event so
+        # ``_open_fan_out_instance_dispatch_span`` can attach
+        # ``openarmature.fan_out.parent_node_name`` to per-instance
+        # spans. Inner events from inside the fan-out's instances
+        # don't carry ``fan_out_config`` themselves; the cache bridges.
+        if event.fan_out_config is not None and event.fan_out_index is None:
+            inv_state.fan_out_parent_node_name[event.namespace] = event.fan_out_config.parent_node_name
+
         # Synthesize subgraph dispatch spans for any ancestor namespace
         # prefix that doesn't have one yet (per observability §4.5).
         # Also closes subgraph spans we've left.
@@ -286,6 +314,16 @@ def _handle_completed(self, event: NodeEvent) -> None:
             for key in list(inv_state.detached_roots.keys()):
                 if len(key) > len(event.namespace) and key[: len(event.namespace)] == event.namespace:
                     self._close_detached_root(inv_state, key)
+        # Per spec proposal 0013 (v0.10.0): if this is the fan-out
+        # node's own completion (non-detached path), close all
+        # per-instance dispatch spans synthesized for this fan-out.
+        # Children-before-parents close ordering: per-instance
+        # dispatch spans before the fan-out node's own span.
+        if event.fan_out_index is None and event.fan_out_config is not None:
+            for key in list(inv_state.fan_out_instance_spans.keys()):
+                if len(key) > len(event.namespace) and key[: len(event.namespace)] == event.namespace:
+                    self._close_fan_out_instance_dispatch_span(inv_state, key)
+            inv_state.fan_out_parent_node_name.pop(event.namespace, None)
         key = self._key_for(event)
         open_span = inv_state.open_spans.pop(key, None)
         if open_span is None:
@@ -498,6 +536,16 @@ def _resolve_parent_context(
             root = inv_state.detached_roots.get(instance_key)
             if root is not None:
                 return set_span_in_context(root.span)
+            # 1a'. Non-detached per-instance dispatch span (proposal
+            #      0013, v0.10.0). Same keying as the detached path
+            #      but lives in the parent trace under the fan-out
+            #      node's span. Inner-node events from inside a
+            #      non-detached fan-out instance parent under THIS
+            #      per-instance dispatch span (not the shared fan-out
+            #      node span at ``namespace[:1]``).
+            instance_dispatch = inv_state.fan_out_instance_spans.get(instance_key)
+            if instance_dispatch is not None:
+                return set_span_in_context(instance_dispatch.span)
         # 1b. Detached subgraph root at any matching prefix wins
         #     (highest precedence — events inside a detached subtree
         #     always parent under the detached root, never bleed up).
@@ -568,6 +616,21 @@ def _sync_subgraph_spans(
                 continue
             if prefix in inv_state.detached_roots:
                 continue
+            # Per spec proposal 0013 (v0.10.0): for non-detached
+            # fan-out instances, the per-instance dispatch span lives
+            # at ``prefix + (str(fan_out_index),)`` in
+            # ``fan_out_instance_spans``. If the per-instance dispatch
+            # span for THIS event's instance is already open, the
+            # fan-out node span at ``prefix`` is already open too
+            # (synthesized as part of the fan-out node's started
+            # event), so we don't open a new subgraph span at
+            # ``prefix``.
+            if (
+                depth == 1
+                and event.fan_out_index is not None
+                and (prefix + (str(event.fan_out_index),)) in inv_state.fan_out_instance_spans
+            ):
+                continue
             # 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:
@@ -579,6 +642,22 @@ def _sync_subgraph_spans(
             if depth == 1 and event.fan_out_index is not None and prefix[0] in self.detached_fan_outs:
                 self._open_detached_fan_out_instance_root(inv_state, correlation_id, prefix, event)
                 continue
+            # Per spec §5.4 + proposal 0013: non-detached fan-out
+            # instances get a synthetic per-instance dispatch span
+            # under the fan-out node span. Triggered by an event
+            # from inside a fan-out instance (event.fan_out_index
+            # populated) at depth 1 (the fan-out node's namespace
+            # prefix), where the parent_node_name cache has an
+            # entry — i.e., the fan-out node's started event has
+            # been seen.
+            if (
+                depth == 1
+                and event.fan_out_index is not None
+                and prefix[0] not in self.detached_fan_outs
+                and prefix in inv_state.fan_out_parent_node_name
+            ):
+                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)
 
     def _open_subgraph_span(
@@ -747,6 +826,63 @@ def _open_detached_fan_out_instance_root(
         inv_state.detached_roots[instance_key] = _OpenSpan(span=instance_root)
         inv_state.fan_out_instance_root_prefixes.add(instance_key)
 
+    def _open_fan_out_instance_dispatch_span(
+        self,
+        inv_state: _InvState,
+        correlation_id: str | None,
+        prefix: tuple[str, ...],
+        event: NodeEvent,
+    ) -> None:
+        """Per-instance dispatch span for a non-detached fan-out
+        (per spec §5.4 + proposal 0013, v0.10.0). Mirror of
+        ``_open_detached_fan_out_instance_root`` but lives in the
+        parent trace (no fresh trace_id).
+
+        Parents under the fan-out node span at ``prefix``. Span name
+        is the fan-out node's name; attributes are
+        ``openarmature.fan_out.parent_node_name`` (looked up from the
+        cache populated when the fan-out node's ``started`` event
+        landed), ``openarmature.node.fan_out_index`` (from
+        ``event.fan_out_index``), and the correlation_id if set.
+
+        Stored in ``inv_state.fan_out_instance_spans`` keyed by
+        ``prefix + (str(fan_out_index),)``. Closed when the fan-out
+        node's own ``completed`` event fires (children-before-parents
+        ordering).
+        """
+        # Find the fan-out node's open span (latest attempt under
+        # retry) to use as parent.
+        fan_out_open = self._find_fan_out_node_span(inv_state, prefix)
+        parent_ctx: object
+        if fan_out_open is not None:
+            parent_ctx = set_span_in_context(fan_out_open.span)
+        else:
+            parent_ctx = otel_context.Context()
+
+        parent_node_name = inv_state.fan_out_parent_node_name.get(prefix, prefix[-1])
+        attrs: dict[str, Any] = {
+            "openarmature.node.name": prefix[-1],
+            "openarmature.fan_out.parent_node_name": parent_node_name,
+            "openarmature.node.fan_out_index": event.fan_out_index,
+        }
+        if correlation_id is not None:
+            attrs["openarmature.correlation_id"] = correlation_id
+        instance_span = self._tracer.start_span(
+            name=prefix[-1],
+            context=cast("Any", parent_ctx),
+            kind=SpanKind.INTERNAL,
+            attributes=attrs,
+        )
+        instance_key = prefix + (str(event.fan_out_index),)
+        inv_state.fan_out_instance_spans[instance_key] = _OpenSpan(span=instance_span)
+
+    def _close_fan_out_instance_dispatch_span(self, inv_state: _InvState, key: tuple[str, ...]) -> None:
+        open_span = inv_state.fan_out_instance_spans.pop(key, None)
+        if open_span is None:
+            return
+        open_span.span.set_status(Status(StatusCode.OK))
+        open_span.span.end()
+
     def _close_detached_root(self, inv_state: _InvState, prefix: tuple[str, ...]) -> None:
         inv_state.fan_out_instance_root_prefixes.discard(prefix)
         open_span = inv_state.detached_roots.pop(prefix, None)
@@ -789,6 +925,18 @@ def _node_attrs(self, event: NodeEvent, correlation_id: str | None) -> dict[str,
             attrs["openarmature.node.fan_out_index"] = event.fan_out_index
         if correlation_id is not None:
             attrs["openarmature.correlation_id"] = correlation_id
+        # Per spec §5.4 + proposal 0013 (v0.10.0): fan-out node spans
+        # carry item_count / concurrency / error_policy. ``concurrency``
+        # is ``int | None`` on the event payload (spec §9.2 canonical
+        # type); the §5.4 attribute is a bare int with ``0`` as the
+        # "unbounded" sentinel (OTel attribute primitives can't carry
+        # null cleanly), so this is the OTel-attribute-layer
+        # translation.
+        if event.fan_out_config is not None:
+            cfg = event.fan_out_config
+            attrs["openarmature.fan_out.item_count"] = cfg.item_count
+            attrs["openarmature.fan_out.concurrency"] = 0 if cfg.concurrency is None else cfg.concurrency
+            attrs["openarmature.fan_out.error_policy"] = cfg.error_policy
         return attrs
 
     # ------------------------------------------------------------------
@@ -830,8 +978,9 @@ def _drain_inv_state(self, inv_state: _InvState) -> None:
         """Close any still-open spans in a per-invocation state
         container in child→parent order. LLM spans (deepest leaves)
         → leaf node spans (sorted deepest-first by namespace) →
-        detached roots → subgraph dispatch spans. Matches the
-        ordering used in ``shutdown``."""
+        non-detached fan-out per-instance dispatch spans → detached
+        roots → subgraph dispatch spans. Matches the ordering used in
+        ``shutdown``."""
         for call_id in list(inv_state.open_llm_spans.keys()):
             open_span = inv_state.open_llm_spans.pop(call_id, None)
             if open_span is not None:
@@ -840,6 +989,8 @@ def _drain_inv_state(self, inv_state: _InvState) -> None:
             open_span = inv_state.open_spans.pop(key, None)
             if open_span is not None:
                 self._drain_open_span(open_span)
+        for key in sorted(inv_state.fan_out_instance_spans.keys(), key=lambda k: -len(k)):
+            self._close_fan_out_instance_dispatch_span(inv_state, key)
         for prefix in sorted(inv_state.detached_roots.keys(), key=lambda k: -len(k)):
             self._close_detached_root(inv_state, prefix)
         for prefix in sorted(inv_state.subgraph_spans.keys(), key=lambda k: -len(k)):
diff --git a/tests/conformance/adapter.py b/tests/conformance/adapter.py
index 322d066..b51b31d 100644
--- a/tests/conformance/adapter.py
+++ b/tests/conformance/adapter.py
@@ -370,9 +370,17 @@ async def run_with_context(
         self,
         state: State,
         context: _InvocationContext,
+        *,
+        pre_resolved_count: int | None = None,
+        pre_resolved_concurrency: tuple[int | None] | None = None,
     ) -> Mapping[str, Any]:
         self.trace_list.append(self.name)
-        return await super().run_with_context(state, context)
+        return await super().run_with_context(
+            state,
+            context,
+            pre_resolved_count=pre_resolved_count,
+            pre_resolved_concurrency=pre_resolved_concurrency,
+        )
 
 
 @dataclass(frozen=True)
diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index abebc3d..753e03b 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -72,6 +72,7 @@
         "003-otel-error-status",
         "004-otel-routing-error-attribution",
         "005-otel-llm-provider-span-nested",
+        "006-otel-fan-out-instance-attribution",
         "007-otel-retry-attempt-spans",
         "008-otel-detached-trace-mode",
         "009-otel-correlation-id-cross-cutting",
@@ -81,11 +82,6 @@
 
 
 _DEFERRED_FIXTURES: dict[str, str] = {
-    "006-otel-fan-out-instance-attribution": (
-        "Needs non-detached fan-out per-instance dispatch span synthesis + "
-        "FanOutConfig metadata surfacing per spec §5.4 (current observer opens one "
-        "shared synthetic span at the namespace prefix, not per-instance). Lands in PR-C.2."
-    ),
     "010-otel-log-correlation": (
         "Needs synchronous observer prep hook (prepare_sync) so the engine task can "
         "attach the observer's span to OTel context for the duration of node-body "
@@ -139,6 +135,8 @@ async def test_observability_fixture(fixture_path: Path) -> None:
         await _run_fixture_004(spec)
     elif fixture_id == "005-otel-llm-provider-span-nested":
         await _run_fixture_005(spec)
+    elif fixture_id == "006-otel-fan-out-instance-attribution":
+        await _run_fixture_006(spec)
     elif fixture_id == "007-otel-retry-attempt-spans":
         await _run_fixture_007(spec)
     elif fixture_id == "008-otel-detached-trace-mode":
@@ -455,6 +453,109 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None:
 # ---------------------------------------------------------------------------
 
 
+async def _run_fixture_006(spec: Mapping[str, Any]) -> None:
+    """Spec §5.4 + proposal 0013 (v0.10.0): non-detached fan-out
+    instances synthesize per-instance dispatch spans nested between
+    the fan-out node span and the inner-node spans. The fan-out node
+    span carries ``item_count`` / ``concurrency`` / ``error_policy``
+    from ``NodeEvent.fan_out_config``; per-instance spans carry
+    ``fan_out_index`` and ``parent_node_name``."""
+    # Subgraphs declared at the spec level (outside ``cases:``) — the
+    # ``worker`` subgraph used by every case in this fixture lives
+    # there. Compile once and reuse across cases.
+    _patch_unsupported_directives(spec)
+    subgraphs = _compile_subgraphs(spec)
+    cases = cast("list[dict[str, Any]]", spec["cases"])
+    for case in cases:
+        case_name = cast("str", case["name"])
+        try:
+            await _run_fixture_006_case(case, subgraphs)
+        except AssertionError as e:
+            raise AssertionError(f"case {case_name!r}: {e}") from e
+
+
+async def _run_fixture_006_case(case: Mapping[str, Any], subgraphs: Mapping[str, Any]) -> None:
+    _patch_unsupported_directives(case)
+
+    observer, exporter = _build_observer()
+    trace_log: list[str] = []
+    built = build_graph(case, subgraphs=dict(subgraphs), trace=trace_log)
+    compiled = built.builder.compile()
+    compiled.attach_observer(observer)
+    initial_state = built.initial_state(case.get("initial_state", {}))
+    await compiled.invoke(initial_state)
+    await compiled.drain()
+    observer.shutdown()
+    spans = exporter.get_finished_spans()
+
+    # Build a span_id → span map for parent-by-id navigation.
+    by_id: dict[int, Any] = {}
+    for s in spans:
+        if s.context is not None:
+            by_id[s.context.span_id] = s
+
+    # Span-tree shape per fixture:
+    #   invocation
+    #   └─ process (fan-out NODE) — item_count=3, concurrency=2, error_policy="collect"
+    #      ├─ process (instance, fan_out_index=0) — parent_node_name="process"
+    #      │  └─ compute
+    #      ├─ process (instance, fan_out_index=1) — parent_node_name="process"
+    #      │  └─ compute
+    #      └─ process (instance, fan_out_index=2) — parent_node_name="process"
+    #         └─ compute
+    process_spans = [s for s in spans if s.name == "process"]
+    # Expect 4: 1 fan-out node + 3 per-instance dispatch spans.
+    assert len(process_spans) == 4, (
+        f"expected 4 'process' spans (1 fan-out node + 3 per-instance dispatch); got {len(process_spans)}"
+    )
+    # Fan-out node span carries the three §5.4 attributes.
+    fan_out_node_spans = [
+        s
+        for s in process_spans
+        if dict(s.attributes or {}).get("openarmature.fan_out.item_count") is not None
+    ]
+    assert len(fan_out_node_spans) == 1, (
+        f"expected exactly 1 fan-out NODE span (with item_count attribute); got {len(fan_out_node_spans)}"
+    )
+    fan_out_node_span = fan_out_node_spans[0]
+    fan_out_attrs = dict(fan_out_node_span.attributes or {})
+    assert fan_out_attrs.get("openarmature.fan_out.item_count") == 3
+    assert fan_out_attrs.get("openarmature.fan_out.concurrency") == 2
+    assert fan_out_attrs.get("openarmature.fan_out.error_policy") == "collect"
+
+    # Per-instance dispatch spans: 3 of them, each with
+    # fan_out_index 0..2 and parent_node_name "process".
+    per_instance_spans = [s for s in process_spans if s != fan_out_node_span]
+    assert len(per_instance_spans) == 3
+    fan_out_indices: set[int] = set()
+    for s in per_instance_spans:
+        attrs = dict(s.attributes or {})
+        idx = attrs.get("openarmature.node.fan_out_index")
+        assert isinstance(idx, int), f"per-instance span MUST carry fan_out_index; got attrs={attrs}"
+        fan_out_indices.add(idx)
+        assert attrs.get("openarmature.fan_out.parent_node_name") == "process", (
+            f"per-instance span MUST carry parent_node_name='process'; got {attrs}"
+        )
+        # Each per-instance dispatch span parents under the fan-out
+        # node span (proposal 0013 + §5.4 nesting).
+        assert s.parent is not None and s.parent.span_id == fan_out_node_span.context.span_id, (
+            "per-instance dispatch span MUST parent under the fan-out node span"
+        )
+    assert fan_out_indices == {0, 1, 2}, (
+        f"per-instance fan_out_index range MUST be 0..2; got {sorted(fan_out_indices)}"
+    )
+
+    # Each per-instance dispatch span has a 'compute' child (the
+    # inner-node work).
+    per_instance_ids = {s.context.span_id for s in per_instance_spans}
+    compute_spans = [s for s in spans if s.name == "compute"]
+    assert len(compute_spans) == 3, f"expected 3 compute spans; got {len(compute_spans)}"
+    for cs in compute_spans:
+        assert cs.parent is not None and cs.parent.span_id in per_instance_ids, (
+            "compute span MUST parent under a per-instance dispatch span"
+        )
+
+
 async def _run_fixture_007(spec: Mapping[str, Any]) -> None:
     """Two sub-cases:
 
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index cd55887..aa01689 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -3,4 +3,4 @@
 
 def test_package_versions() -> None:
     assert openarmature.__version__ == "0.4.0rc0"
-    assert openarmature.__spec_version__ == "0.9.0"
+    assert openarmature.__spec_version__ == "0.10.0"

From 53b4988791d6cc0e6251ff0ba8741f3082f348cb Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 19:16:00 -0700
Subject: [PATCH 2/6] otel: address PR #26 review
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Move fan_out runtime imports from compiled.py module top into
  function scope to break the textual import cycle CodeQL's
  py/cyclic-import rule flagged. fan_out has a TYPE_CHECKING
  back-reference to compiled (no runtime issue), but the static
  analyzer doesn't see the gate. Lazy-imports inside _invoke
  (FanOutNode for the isinstance check) and _step_fan_out_node
  (_resolve_concurrency / _resolve_count). Type-only FanOutNode
  reference moves to TYPE_CHECKING with `from __future__ import
  annotations` so signatures resolve without the runtime import.
- Fix _drain_inv_state ordering: per-instance dispatch spans in
  fan_out_instance_spans are children of the fan-out NODE span
  (which lives in open_spans at depth 1). The previous shape
  closed all of open_spans before draining
  fan_out_instance_spans, ending the parent before its children
  during shutdown/abandon. Now drains open_spans in two phases
  (deep >= 2 / shallow = 1) with the per-instance drain between,
  so children-before-parents holds.
- Remove dead `by_id` map from _run_fixture_006_case —
  scaffolding from an earlier draft; current assertions don't
  use it.
---
 src/openarmature/graph/compiled.py            | 31 +++++++++++++++++--
 .../observability/otel/observer.py            | 20 +++++++++++-
 tests/conformance/test_observability.py       |  6 ----
 3 files changed, 48 insertions(+), 9 deletions(-)

diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py
index 201060e..ef2a25a 100644
--- a/src/openarmature/graph/compiled.py
+++ b/src/openarmature/graph/compiled.py
@@ -21,12 +21,26 @@
 `cast(MyState, ...)` at the call site.
 """
 
+from __future__ import annotations
+
 import asyncio
 import time
 import uuid
 from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, field
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
+
+if TYPE_CHECKING:
+    # ``FanOutNode`` lives in ``.fan_out`` which has a TYPE_CHECKING
+    # back-reference to ``CompiledGraph`` here. Importing at module
+    # top would create a textual cycle CodeQL's
+    # ``py/cyclic-import`` rule flags (no runtime issue —
+    # ``fan_out``'s ``compiled`` import is itself TYPE_CHECKING-gated
+    # — but the static analyzer doesn't see that). Type annotations
+    # use the string form via ``from __future__ import annotations``;
+    # runtime use (the ``isinstance`` check in ``_invoke``) imports
+    # lazily inside the function.
+    from .fan_out import FanOutNode
 
 from pydantic import ValidationError
 
@@ -68,7 +82,6 @@
     StateValidationError,
 )
 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 (
@@ -519,6 +532,12 @@ async def _invoke(
                 current = skip_target
                 continue
 
+            # Lazy import: keeps the textual cycle off the module
+            # graph (``fan_out`` has a TYPE_CHECKING back-reference
+            # to this module). Function-scope import is cheap once
+            # cached; this branch fires once per fan-out step.
+            from .fan_out import FanOutNode  # noqa: PLC0415
+
             if isinstance(node, FanOutNode):
                 # Fan-out nodes are recognized as a distinct node type
                 # per pipeline-utilities §9. Dispatched through
@@ -922,6 +941,14 @@ async def _step_fan_out_node(
         # is pure regardless. Repeating these inside
         # ``FanOutNode.run_with_context`` is cheap and matches the
         # values surfaced here.
+        # Lazy import: function-scope to avoid a module-top
+        # textual cycle CodeQL flags. ``fan_out`` has a
+        # TYPE_CHECKING back-reference to this module, so the
+        # static-analyzer view of an importable cycle goes away
+        # when the engine doesn't reach into ``fan_out`` at module
+        # load time. Fires once per fan-out step.
+        from .fan_out import _resolve_concurrency, _resolve_count  # noqa: PLC0415
+
         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
diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py
index 73d8551..1d65fa0 100644
--- a/src/openarmature/observability/otel/observer.py
+++ b/src/openarmature/observability/otel/observer.py
@@ -985,12 +985,30 @@ def _drain_inv_state(self, inv_state: _InvState) -> None:
             open_span = inv_state.open_llm_spans.pop(call_id, None)
             if open_span is not None:
                 self._drain_open_span(open_span)
-        for key in sorted(inv_state.open_spans.keys(), key=lambda k: -len(k[0])):
+        # Inner-node spans (depth >= 2) drain first — these include
+        # the inner-node bodies inside fan-out instances, which are
+        # children of the per-instance dispatch spans.
+        for key in sorted(
+            (k for k in inv_state.open_spans if len(k[0]) >= 2),
+            key=lambda k: -len(k[0]),
+        ):
             open_span = inv_state.open_spans.pop(key, None)
             if open_span is not None:
                 self._drain_open_span(open_span)
+        # Per-instance dispatch spans (children of the fan-out NODE
+        # span) drain BEFORE depth-1 entries of ``open_spans`` —
+        # otherwise the fan-out NODE span (depth 1) would end
+        # before its per-instance children, violating
+        # children-before-parents close ordering.
         for key in sorted(inv_state.fan_out_instance_spans.keys(), key=lambda k: -len(k)):
             self._close_fan_out_instance_dispatch_span(inv_state, key)
+        # Remaining depth-1 entries in ``open_spans`` drain last —
+        # these are the fan-out NODE span itself + any sibling
+        # depth-1 leaf nodes still open.
+        for key in list(inv_state.open_spans.keys()):
+            open_span = inv_state.open_spans.pop(key, None)
+            if open_span is not None:
+                self._drain_open_span(open_span)
         for prefix in sorted(inv_state.detached_roots.keys(), key=lambda k: -len(k)):
             self._close_detached_root(inv_state, prefix)
         for prefix in sorted(inv_state.subgraph_spans.keys(), key=lambda k: -len(k)):
diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 753e03b..5b20f54 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -488,12 +488,6 @@ async def _run_fixture_006_case(case: Mapping[str, Any], subgraphs: Mapping[str,
     observer.shutdown()
     spans = exporter.get_finished_spans()
 
-    # Build a span_id → span map for parent-by-id navigation.
-    by_id: dict[int, Any] = {}
-    for s in spans:
-        if s.context is not None:
-            by_id[s.context.span_id] = s
-
     # Span-tree shape per fixture:
     #   invocation
     #   └─ process (fan-out NODE) — item_count=3, concurrency=2, error_policy="collect"

From 793701432fde58aae5fe833a9e266e872dce6751 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 19:41:22 -0700
Subject: [PATCH 3/6] codeql: suppress py/unsafe-cyclic-import

Both compiled.py and fan_out.py reference each other's types
only via TYPE_CHECKING blocks. CodeQL's analyzer flags the
textual cycle regardless of the gate; no runtime cycle exists.
Removing either side breaks pyright's type resolution for
generics across the boundary. Pyright's reportImportCycles
already covers genuine runtime cycles at type-check time, so
dropping this CodeQL rule loses no signal.
---
 .github/codeql/codeql-config.yml | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml
index 6747e29..909e9d4 100644
--- a/.github/codeql/codeql-config.yml
+++ b/.github/codeql/codeql-config.yml
@@ -48,3 +48,23 @@ query-filters:
   # so dropping the CodeQL rule doesn't lose signal.
   - exclude:
       id: py/unused-import
+  # ``py/unsafe-cyclic-import`` flags the textual import shape
+  # without honoring ``if TYPE_CHECKING:`` gates. The two cases
+  # in this codebase are mutual-typing pairs:
+  #
+  # - ``graph/compiled.py`` and ``graph/fan_out.py`` reference
+  #   each other's types only via TYPE_CHECKING blocks
+  #   (``compiled`` imports ``FanOutNode`` for type annotations;
+  #   ``fan_out`` imports ``CompiledGraph`` for the
+  #   ``FanOutConfig.subgraph`` field annotation). Both sides use
+  #   ``from __future__ import annotations`` so annotations are
+  #   strings; no runtime cycle exists.
+  #
+  # The canonical Python workaround for typed-module pairs is
+  # exactly this TYPE_CHECKING gating. Removing either side
+  # breaks pyright's type resolution for generics across the
+  # boundary. Pyright's ``reportImportCycles`` already covers the
+  # genuine runtime-cycle cases at type-check time, so dropping
+  # the CodeQL rule doesn't lose signal.
+  - exclude:
+      id: py/unsafe-cyclic-import

From 3866f819650944ebded0b10f4d8fbb0df2af6be2 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 19:42:42 -0700
Subject: [PATCH 4/6] fan-out: surface resolver failures with event pair
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

PR-C.2 hoisted item_count / concurrency resolution out of
``run_with_context`` to step entry so the eager
``FanOutEventConfig`` could ride on every fan-out node event.
That moved resolver failures (callable raises, ``getattr`` on
malformed state, items_field non-list) outside ``innermost``'s
``except Exception → NodeException`` block, dropping the
started/completed event pair and the NodeException wrap that
the inner path used to provide.

Re-establish the contract by wrapping the resolution block:
any failure now dispatches a started+completed pair with
``fan_out_config=None`` (we never built one) and surfaces as
NodeException. Also drops the silent 0-coercion when
``items_field`` resolves to a non-list — raises with the same
TypeError text ``_build_instance_states`` produces, so
fan_out_config never reports a misleading ``item_count=0``.
---
 src/openarmature/graph/compiled.py | 88 ++++++++++++++++++++++++++----
 1 file changed, 76 insertions(+), 12 deletions(-)

diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py
index ef2a25a..c830e16 100644
--- a/src/openarmature/graph/compiled.py
+++ b/src/openarmature/graph/compiled.py
@@ -949,18 +949,82 @@ async def _step_fan_out_node(
         # load time. Fires once per fan-out step.
         from .fan_out import _resolve_concurrency, _resolve_count  # noqa: PLC0415
 
-        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,
-        )
+        # Resolver failures (callable count/concurrency raising,
+        # ``getattr`` on a malformed state, etc.) used to land inside
+        # ``innermost``'s ``except Exception → NodeException`` block
+        # below and produce a started/completed event pair via the
+        # surrounding dispatches. Hoisting resolution out of
+        # ``run_with_context`` for the eager ``FanOutEventConfig``
+        # build moved them past that scope, so re-establish the
+        # contract here: surface a started/completed pair with
+        # ``fan_out_config=None`` (we never built one) and raise as
+        # ``NodeException``.
+        try:
+            if node.config.items_field is not None:
+                items_attr: Any = getattr(state, node.config.items_field, [])
+                if not isinstance(items_attr, list):
+                    raise NodeException(
+                        node_name=current,
+                        cause=TypeError(f"items_field {node.config.items_field!r} is not a list at runtime"),
+                        recoverable_state=state,
+                    )
+                item_count = len(cast("list[Any]", items_attr))
+            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,
+            )
+        except NodeException as resolution_error:
+            self._dispatch_started(
+                context,
+                current,
+                namespace,
+                step,
+                state,
+                attempt_index=0,
+                fan_out_config=None,
+            )
+            self._dispatch_completed(
+                context,
+                current,
+                namespace,
+                step,
+                state,
+                error=resolution_error,
+                attempt_index=0,
+                fan_out_config=None,
+            )
+            raise
+        except Exception as resolution_error:
+            wrapped = NodeException(
+                node_name=current,
+                cause=resolution_error,
+                recoverable_state=state,
+            )
+            self._dispatch_started(
+                context,
+                current,
+                namespace,
+                step,
+                state,
+                attempt_index=0,
+                fan_out_config=None,
+            )
+            self._dispatch_completed(
+                context,
+                current,
+                namespace,
+                step,
+                state,
+                error=wrapped,
+                attempt_index=0,
+                fan_out_config=None,
+            )
+            raise wrapped from resolution_error
 
         # Cell holding the FINAL successful attempt's
         # (attempt_index, pre_state, merged); see same comment in

From bf8b2001f16fd3fa2d8a661065154ca353f2eae1 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 19:49:37 -0700
Subject: [PATCH 5/6] tests: drop redundant Protocol assertion
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The ``_: ProjectionStrategy[S, Inner] = BoomProjection()`` line was
a defensive belt-and-suspenders structural-conformance check, but
the call-site ``add_subgraph_node(projection=BoomProjection())``
exercises the same Protocol check via its parameter annotation —
pyright catches a mismatch there without needing the explicit
assignment. Drop the line and update the comment to point at the
call site.
---
 tests/unit/test_runtime_errors.py | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/tests/unit/test_runtime_errors.py b/tests/unit/test_runtime_errors.py
index 0b7f213..acd0c8b 100644
--- a/tests/unit/test_runtime_errors.py
+++ b/tests/unit/test_runtime_errors.py
@@ -105,7 +105,7 @@ async def test_subgraph_projection_error_wrapped_as_node_exception() -> None:
     NodeException tagged with the subgraph wrapper's name so callers see
     a uniform error contract."""
 
-    from openarmature.graph import NodeException, ProjectionStrategy
+    from openarmature.graph import NodeException
 
     class Inner(State):
         x: int = 0
@@ -115,8 +115,9 @@ async def _inner_node(_s: Inner) -> dict[str, Any]:
 
     inner_g = GraphBuilder(Inner).add_node("i", _inner_node).add_edge("i", END).set_entry("i").compile()
 
-    # Parameter names match the ProjectionStrategy Protocol exactly so
-    # pyright's strict structural conformance check passes.
+    # Parameter names match the ProjectionStrategy Protocol exactly so the
+    # call-site ``projection=BoomProjection()`` below passes pyright's strict
+    # structural conformance check.
     class BoomProjection:
         def project_in(self, parent_state: S, subgraph_state_cls: type[Inner]) -> Inner:
             raise RuntimeError("project_in boom")
@@ -129,8 +130,6 @@ def project_out(
         ) -> dict[str, Any]:
             return {}
 
-    _: ProjectionStrategy[S, Inner] = BoomProjection()
-
     g = (
         GraphBuilder(S)
         .add_subgraph_node("sub", inner_g, projection=BoomProjection())

From 81046e8aaa0646789bfb72e6cbf99d3e39597dc7 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 9 May 2026 19:56:53 -0700
Subject: [PATCH 6/6] tests: comment LLM step sentinel + parametrize list
 factories

- Annotate ``step=-1`` in the synthetic LLM-event observer test
  to point readers at ``OpenAIProvider._llm_event`` (openai.py:643)
  where the same sentinel is minted for production LLM-provider
  span events that aren't tied to graph step sequencing.
- ``_ParentState`` in the fan-out gating test now uses
  ``Field(default_factory=list[int])`` instead of the bare ``[]``
  default, matching the parametrized factory shape used in
  ``test_checkpoint.py``'s ``_ParentState`` and the surrounding
  pyright-strict expectation that field types are fully known.
---
 tests/unit/test_observability_otel.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py
index 94ad62e..20e1474 100644
--- a/tests/unit/test_observability_otel.py
+++ b/tests/unit/test_observability_otel.py
@@ -24,6 +24,7 @@
 import logging
 
 import pytest
+from pydantic import Field
 
 # Skip the entire module if otel extras aren't installed.
 pytest.importorskip("opentelemetry.sdk.trace")
@@ -271,6 +272,9 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None:
         span_processor=SimpleSpanProcessor(exporter),
         disable_llm_spans=True,
     )
+    # ``step=-1`` mirrors the synthetic value ``OpenAIProvider._llm_event``
+    # mints (openai.py:643) — LLM-provider events aren't tied to graph step
+    # sequencing.
     started = NodeEvent(
         node_name="openarmature.llm.complete",
         namespace=("openarmature.llm.complete",),
@@ -556,8 +560,8 @@ async def test_concurrent_fan_out_no_lifo_violation() -> None:
     import warnings
 
     class _ParentState(State):
-        items: list[int] = []
-        results: list[int] = []
+        items: list[int] = Field(default_factory=list[int])
+        results: list[int] = Field(default_factory=list[int])
 
     class _ChildState(State):
         item: int = 0