graph: spec v0.4 fan-out runtime (proposal 0005 PU side) — phase 3 (#16)

* graph: spec v0.4 fan-out runtime (proposal 0005 PU side) — phase 3

Implements pipeline-utilities §9: parallel fan-out of compiled
subgraphs with bounded concurrency, two error policies, configurable
empty-input behavior, and per-instance middleware composition. Lands
all 8 target conformance fixtures (pipeline-utilities 017-023 +
graph-engine 017-observer-fan-out-index).

Engine:
  - fan_out.py: FanOutNode (a third Node sibling alongside FunctionNode
    and SubgraphNode), FanOutConfig dataclass, per-instance projection
    helpers, concurrency-bounded execution via asyncio.gather +
    Semaphore, fan-in merge for fail_fast and collect policies,
    on_empty raise/noop handling.
  - errors.py: FanOutCountModeAmbiguous, FanOutFieldNotList compile
    errors; FanOutEmpty, FanOutInvalidCount, FanOutInvalidConcurrency
    runtime errors. The runtime trio subclasses NodeException so they
    surface as graph-engine §4 node_exception with an additional
    fan_out_category attribute (matching fixture 023's expected
    shape).
  - observer.py: _InvocationContext gains fan_out_index; new
    descend_into_fan_out_instance helper stamps the index onto the
    child context so inner-node events fire with it populated.
  - compiled.py: _step_fan_out_node — wraps the whole fan-out as one
    parent dispatch (per §9.6) with started/completed events around
    the chain; _dispatch_started/_completed propagate fan_out_index
    from context onto every NodeEvent.
  - builder.py: GraphBuilder.add_fan_out_node with full §9 compile-
    time validation (mode mutual-exclusion, items_field list-typed
    check, declared-field references for collect_field, target_field,
    count_field, errors_field, inputs, extra_outputs).

Conformance harness:
  - adapter.py: fan_out node directive translation; new test seam
    directives update_pure, update_from_field, flaky_by_index (both
    fail_count_per_idx and fail_when_idx shapes), flaky_instance_only;
    _TracingFanOutNode for execution-order trace recording.
  - test_pipeline_utilities.py: instance_middleware threading via a
    new fan_out_instance_middleware kwarg on build_graph; cases-
    fixture handling now merges shared subgraph blocks into each
    case so 018-019, 021-023 see them; supports state_field_read and
    queue_chunk callable resolvers for count + concurrency.
  - test_conformance.py: removed fan_out from
    _UNSUPPORTED_NODE_DIRECTIVES so graph-engine fixture 017 runs.

Unit tests (test_fan_out.py): 19 tests covering items_field
projection, count modes (literal int + state-reading callable),
count + concurrency callables resolved exactly once at entry, inputs
mapping projection, concurrency limit enforcement, fail_fast
recoverable_state contract, collect errors_field shape, on_empty
raise/noop, count_field write, extra_outputs merge,
instance_middleware retry composition, fan-in determinism under
nondeterministic completion timing, four compile-error checks
(count_mode_ambiguous both/neither, field_not_list,
inputs/extra_outputs undeclared field references).

Total tests: 276 passing, 0 skipped.

* test: document fixture-global counter limitation in flaky test seams

The PR review thread suggested keying flaky_by_index and
flaky_instance_only by id(state) for per-instance counters. Tried
that; broke fixture 021. Root cause: instance-level retry (021)
constructs a fresh subgraph state on each retry, so id(state) resets
per retry attempt — the per-instance counter starts over and retry
exhausts.

True per-instance semantics need an identifier stable across both
node-level retry (state stable, id works) AND instance-level retry
(state changes, id doesn't work). A state-field key would work but
the field name is fixture-specific (item for 020, input for 021).
Reverting to a fixture-global counter and documenting the limitation
in the docstring + an inline comment, so a future fixture exercising
the gap surfaces a real failure rather than silently miscounting.

The existing fixtures (019 collect, 020 node-level retry, 021
instance-level retry) align with the global counter at runtime —
not because the semantics are correct, but because the timing
happens to land the failure on the right call.

Author: chris-colinsky

src/openarmature/graph/__init__.py

@@ -14,6 +14,11 @@
     ConflictingReducers,
     DanglingEdge,
     EdgeException,
+    FanOutCountModeAmbiguous,
+    FanOutEmpty,
+    FanOutFieldNotList,
+    FanOutInvalidConcurrency,
+    FanOutInvalidCount,
     GraphError,
     MappingReferencesUndeclaredField,
     MultipleOutgoingEdges,
@@ -26,6 +31,7 @@
     UnreachableNode,
 )
 from .events import NodeEvent
+from .fan_out import FanOutConfig, FanOutNode
 from .middleware import (
     Middleware,
     RetryMiddleware,
@@ -51,6 +57,13 @@
     "EdgeException",
     "EndSentinel",
     "ExplicitMapping",
+    "FanOutConfig",
+    "FanOutCountModeAmbiguous",
+    "FanOutEmpty",
+    "FanOutFieldNotList",
+    "FanOutInvalidConcurrency",
+    "FanOutInvalidCount",
+    "FanOutNode",
     "FieldNameMatching",
     "FunctionNode",
     "GraphBuilder",

src/openarmature/graph/builder.py

@@ -11,17 +11,22 @@
 """
 
 from collections.abc import Awaitable, Callable, Iterable, Mapping
-from typing import Any, Self, cast
+from types import GenericAlias, UnionType
+from typing import Any, Self, cast, get_args, get_origin
 
 from .compiled import CompiledGraph
 from .edges import ConditionalEdge, EndSentinel, StaticEdge
 from .errors import (
     ConflictingReducers,
     DanglingEdge,
+    FanOutCountModeAmbiguous,
+    FanOutFieldNotList,
+    MappingReferencesUndeclaredField,
     MultipleOutgoingEdges,
     NoDeclaredEntry,
     UnreachableNode,
 )
+from .fan_out import ConcurrencyResolver, CountResolver, FanOutConfig, FanOutNode
 from .middleware import Middleware
 from .nodes import FunctionNode, Node
 from .projection import FieldNameMatching, ProjectionStrategy
@@ -79,6 +84,156 @@ def add_subgraph_node[ChildT: State](
         )
         return self
 
+    def add_fan_out_node[ChildT: State](
+        self,
+        name: str,
+        *,
+        subgraph: CompiledGraph[ChildT],
+        collect_field: str,
+        target_field: str,
+        items_field: str | None = None,
+        item_field: str | None = None,
+        count: int | CountResolver | None = None,
+        concurrency: int | ConcurrencyResolver | None = 10,
+        error_policy: str = "fail_fast",
+        on_empty: str = "raise",
+        count_field: str | None = None,
+        inputs: Mapping[str, str] | None = None,
+        extra_outputs: Mapping[str, str] | None = None,
+        instance_middleware: Iterable[Middleware] | None = None,
+        errors_field: str | None = None,
+        middleware: Iterable[Middleware] | None = None,
+    ) -> Self:
+        """Register a fan-out node per pipeline-utilities §9.
+
+        Validates configuration at registration time:
+
+        - Exactly one of ``items_field`` or ``count`` MUST be specified
+          (``fan_out_count_mode_ambiguous`` otherwise).
+        - ``items_field`` MUST refer to a list-typed field on the parent
+          state schema (``fan_out_field_not_list`` otherwise).
+        - ``items_field`` mode requires ``item_field``; ``count`` mode
+          forbids ``item_field``.
+        - ``on_empty`` and ``error_policy`` MUST be one of the
+          spec-defined string literals.
+        - ``inputs`` / ``extra_outputs`` / ``count_field`` field
+          references go through the existing
+          ``mapping_references_undeclared_field`` rule.
+
+        See spec §9 for full field semantics.
+        """
+        if name in self._nodes:
+            raise ValueError(f"node {name!r} already declared")
+
+        # Mode validation: exactly one of items_field / count.
+        if (items_field is None) == (count is None):
+            raise FanOutCountModeAmbiguous(
+                node_name=name,
+                message=(
+                    "must specify exactly one of items_field or count "
+                    f"(got items_field={items_field!r}, count={count!r})"
+                ),
+            )
+        if items_field is not None and item_field is None:
+            raise FanOutCountModeAmbiguous(node_name=name, message="items_field mode requires item_field")
+        if count is not None and item_field is not None:
+            raise FanOutCountModeAmbiguous(node_name=name, message="count mode forbids item_field")
+
+        # items_field must be a list-typed parent field.
+        if items_field is not None:
+            parent_fields = self.state_cls.model_fields
+            if items_field not in parent_fields:
+                raise MappingReferencesUndeclaredField(
+                    direction="fan_out.items_field", side="parent", field_name=items_field
+                )
+            ann = parent_fields[items_field].annotation
+            if not _is_list_typed(ann):
+                raise FanOutFieldNotList(node_name=name, field_name=items_field)
+
+        # error_policy + on_empty literal validation.
+        if error_policy not in {"fail_fast", "collect"}:
+            raise ValueError(
+                f"fan-out node {name!r}: error_policy must be 'fail_fast' or 'collect', got {error_policy!r}"
+            )
+        if on_empty not in {"raise", "noop"}:
+            raise ValueError(f"fan-out node {name!r}: on_empty must be 'raise' or 'noop', got {on_empty!r}")
+
+        # *_field references must match declared fields.
+        parent_fields = self.state_cls.model_fields
+        sub_fields = subgraph.state_cls.model_fields
+        if target_field not in parent_fields:
+            raise MappingReferencesUndeclaredField(
+                direction="fan_out.target_field", side="parent", field_name=target_field
+            )
+        if collect_field not in sub_fields:
+            raise MappingReferencesUndeclaredField(
+                direction="fan_out.collect_field", side="subgraph", field_name=collect_field
+            )
+        # NOTE: item_field is intentionally NOT validated against declared
+        # subgraph fields. Per fixture 023, the spec allows item_field to
+        # name a field the subgraph doesn't declare (treated as a
+        # placeholder when the subgraph doesn't read the item). The
+        # runtime projection in fan_out._build_instance_states skips the
+        # assignment if the field isn't declared, so non-declared
+        # item_field values are effectively no-ops.
+        if count_field is not None and count_field not in parent_fields:
+            raise MappingReferencesUndeclaredField(
+                direction="fan_out.count_field", side="parent", field_name=count_field
+            )
+        if errors_field is not None and errors_field not in parent_fields:
+            raise MappingReferencesUndeclaredField(
+                direction="fan_out.errors_field", side="parent", field_name=errors_field
+            )
+        for sub_f, parent_f in (inputs or {}).items():
+            if sub_f not in sub_fields:
+                raise MappingReferencesUndeclaredField(
+                    direction="fan_out.inputs", side="subgraph", field_name=sub_f
+                )
+            if parent_f not in parent_fields:
+                raise MappingReferencesUndeclaredField(
+                    direction="fan_out.inputs", side="parent", field_name=parent_f
+                )
+        for parent_f, sub_f in (extra_outputs or {}).items():
+            if parent_f not in parent_fields:
+                raise MappingReferencesUndeclaredField(
+                    direction="fan_out.extra_outputs", side="parent", field_name=parent_f
+                )
+            if sub_f not in sub_fields:
+                raise MappingReferencesUndeclaredField(
+                    direction="fan_out.extra_outputs", side="subgraph", field_name=sub_f
+                )
+
+        cfg = FanOutConfig(
+            subgraph=subgraph,
+            collect_field=collect_field,
+            target_field=target_field,
+            items_field=items_field,
+            item_field=item_field,
+            count=count,
+            concurrency=concurrency,
+            error_policy=cast(Any, error_policy),
+            on_empty=cast(Any, on_empty),
+            count_field=count_field,
+            inputs=dict(inputs or {}),
+            extra_outputs=dict(extra_outputs or {}),
+            instance_middleware=tuple(instance_middleware or ()),
+            errors_field=errors_field,
+        )
+        # FanOutNode satisfies the Node[StateT] structural protocol (run
+        # returns a partial update; name and middleware are present),
+        # but pyright loses the StateT correspondence through the second
+        # type parameter — cast restores it for the dict assignment.
+        fan_out: Node[StateT] = cast(
+            "Node[StateT]",
+            FanOutNode[StateT, ChildT](
+                name=name,
+                config=cfg,
+                middleware=tuple(middleware) if middleware is not None else (),
+            ),
+        )
+        self._nodes[name] = fan_out
+        return self
+
     def add_middleware(self, middleware: Middleware) -> Self:
         """Register a per-graph middleware applied to every node in this graph.
 
@@ -195,3 +350,28 @@ def _reachable_nodes(
                     reachable.add(name)
                     frontier.append(name)
         return reachable
+
+
+def _is_list_typed(annotation: Any) -> bool:
+    """True if ``annotation`` resolves to a ``list[...]`` shape.
+
+    Used by ``add_fan_out_node`` to validate that ``items_field`` refers
+    to a list-typed parent field. Handles both bare ``list[X]`` and
+    ``Annotated[list[X], reducer]`` forms (the latter is how state
+    fields commonly attach an `append` reducer).
+    """
+    if annotation is list:
+        return True
+    origin = get_origin(annotation)
+    if origin is list:
+        return True
+    # Annotated[T, ...] — peel the metadata, recurse on the type.
+    if isinstance(annotation, GenericAlias):
+        return False
+    args = get_args(annotation)
+    if args and origin is None:
+        # Likely Annotated; first arg is the underlying type.
+        return _is_list_typed(args[0])
+    if isinstance(annotation, UnionType):
+        return any(_is_list_typed(a) for a in args)
+    return False

src/openarmature/graph/compiled.py

@@ -38,6 +38,7 @@
     StateValidationError,
 )
 from .events import NodeEvent
+from .fan_out import FanOutNode
 from .middleware import ChainCall, Middleware, compose_chain
 from .nodes import Node
 from .observer import (
@@ -257,7 +258,15 @@ async def _invoke(
         while True:
             node = self.nodes[current]
 
-            if isinstance(node, SubgraphNode):
+            if isinstance(node, FanOutNode):
+                # Fan-out nodes are recognized as a distinct node type
+                # per pipeline-utilities §9. Dispatched through
+                # ``_step_fan_out_node`` which wraps the whole fan-out
+                # as one parent dispatch (per §9.6) — instance-level
+                # concurrency lives inside the FanOutNode itself.
+                fn_node = cast("FanOutNode[StateT, State]", node)
+                state = await self._step_fan_out_node(fn_node, current, state, context)
+            elif isinstance(node, SubgraphNode):
                 # Subgraph wrappers are transparent to the observer protocol
                 # (per fixture 013): no event is dispatched for the wrapper
                 # itself, the step counter does not advance for it, and any
@@ -457,6 +466,62 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
         return _merge_partial(state, final_partial, self.reducers, current)
 
+    async def _step_fan_out_node(
+        self,
+        node: FanOutNode[StateT, State],
+        current: str,
+        state: StateT,
+        context: _InvocationContext,
+    ) -> StateT:
+        """Run one fan-out-as-node step through the parent's middleware chain.
+
+        Per pipeline-utilities §9.6: the parent's per-graph + per-node
+        middleware wraps the fan-out as a SINGLE dispatch — one started
+        event before the fan-out begins, one completed event after all
+        instances complete and fan-in is done. Per-instance events
+        come from the inner subgraph executions; their pre_state /
+        post_state shape is the inner subgraph's state, and they carry
+        ``fan_out_index`` populated.
+
+        Raw exceptions escaping the chain become NodeException per §4.
+        """
+        step = context.take_step()
+        namespace = context.namespace_prefix + (current,)
+
+        async def innermost(s: Any) -> Mapping[str, Any]:
+            self._dispatch_started(context, current, namespace, step, s)
+            try:
+                partial = await node.run_with_context(s, context)
+            except RuntimeGraphError as e:
+                self._dispatch_completed(context, current, namespace, step, s, error=e)
+                raise
+            except Exception as e:
+                wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                self._dispatch_completed(context, current, namespace, step, s, error=wrapped)
+                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)
+                raise
+
+            self._dispatch_completed(context, current, namespace, step, s, post_state=merged)
+            return partial
+
+        chain: ChainCall = compose_chain(
+            list(self.middleware) + list(node.middleware),
+            innermost,
+        )
+
+        try:
+            final_partial = await chain(state)
+        except RuntimeGraphError:
+            raise
+        except Exception as e:
+            raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
+        return _merge_partial(state, final_partial, self.reducers, current)
+
     @staticmethod
     def _dispatch_started(
         context: _InvocationContext,
@@ -479,6 +544,7 @@ def _dispatch_started(
                 error=None,
                 parent_states=context.parent_states_prefix,
                 attempt_index=attempt_index,
+                fan_out_index=context.fan_out_index,
             ),
         )
 
@@ -506,5 +572,6 @@ def _dispatch_completed(
                 error=error,
                 parent_states=context.parent_states_prefix,
                 attempt_index=attempt_index,
+                fan_out_index=context.fan_out_index,
             ),
         )