observability: phase 6.1 PR-C.2 — proposal 0013 + v0.10.0 (#26)

* 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.

* otel: address PR #26 review

- 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.

* 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.

* fan-out: surface resolver failures with event pair

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``.

* tests: drop redundant Protocol assertion

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: 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.

Author: chris-colinsky

.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

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

@@ -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
 
@@ -67,8 +81,7 @@
     RuntimeGraphError,
     StateValidationError,
 )
-from .events import NodeEvent
-from .fan_out import FanOutNode
+from .events import FanOutEventConfig, NodeEvent
 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
@@ -913,6 +932,100 @@ 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.
+        # 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
+
+        # 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
         # ``_step_function_node``.
@@ -923,12 +1036,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 +1074,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 +1162,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 +1173,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 +1187,7 @@ def _dispatch_started(
         pre_state: State,
         *,
         attempt_index: int = 0,
+        fan_out_config: FanOutEventConfig | None = None,
     ) -> None:
         _dispatch(
             context,
@@ -1058,6 +1202,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 +1217,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 +1232,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"]