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

[chris-colinsky]

Summary

Phase 3 of the implementation plan — pipeline-utilities §9 fan-out runtime: parallel per-item / per-count subgraph dispatch with bounded concurrency, two error policies, configurable empty-input behavior, and per-instance middleware composition. Lands all 8 target fixtures (pipeline-utilities 017–023 + graph-engine 017-observer-fan-out-index) and adds 19 unit tests for the spec-corner cases the fixtures only exercise implicitly.

What's in the PR

Engine

• src/openarmature/graph/fan_out.py — new module:
  • FanOutNode: third sibling to FunctionNode and SubgraphNode. Recognized in _invoke as a distinct dispatch type; its run_with_context wraps the parent's middleware chain around a single fan-out dispatch.
  • FanOutConfig: frozen dataclass with all §9 config fields (subgraph, items_field, item_field, count, concurrency, error_policy, on_empty, count_field, inputs, extra_outputs, instance_middleware, errors_field).
  • Per-instance projection (items_field mode + count mode), concurrency-bounded execution via asyncio.gather + asyncio.Semaphore, fan-in merge for fail_fast and collect policies, on_empty raise/noop handling.
• errors.py — five new categories:
  • Compile: FanOutCountModeAmbiguous, FanOutFieldNotList.
  • Runtime: FanOutEmpty, FanOutInvalidCount, FanOutInvalidConcurrency. Each subclasses NodeException so they surface as category: "node_exception" with an additional fan_out_category attribute (matching fixture 023's expected shape).
• observer.py — _InvocationContext gains fan_out_index; descend_into_fan_out_instance stamps the index onto the child context. Inherited unchanged through any subgraph descents inside an instance.
• compiled.py — _step_fan_out_node wraps the whole fan-out as one parent dispatch per §9.6. Per-instance events fire with fan_out_index populated.
• builder.py — GraphBuilder.add_fan_out_node with full §9 compile-time validation.

Harness

• adapter.py — fan_out: translation; new test 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 tracing.
• test_pipeline_utilities.py — fan_out_instance_middleware threading; cases-fixture merge of shared subgraph blocks; state_field_read and queue_chunk callable resolvers for count/concurrency.
• test_conformance.py — fan_out removed from _UNSUPPORTED_NODE_DIRECTIVES; graph-engine fixture 017 now passes.

Unit tests (test_fan_out.py, 19 tests)

items_field projection · count modes (literal + state-reading callable) · count and concurrency callables resolved exactly once at entry · inputs mapping · concurrency limit · fail_fast recoverable_state contract · collect errors_field shape · on_empty raise/noop · count_field write · extra_outputs merge · instance_middleware retry · fan-in determinism under nondeterministic timing · four compile errors (count_mode_ambiguous both/neither, field_not_list, inputs/extra_outputs undeclared field references).

Conformance impact

• pipeline-utilities 017–023 now pass.
• graph-engine 017-observer-fan-out-index now passes.
• 024–031 (checkpointing) still skip; Phase 5.

Test plan

• uv run pytest -q — 276 passed, 0 skipped, 3 expected warnings.
• uv run pyright src/ tests/ — 0 errors.
• uv run ruff check src/ tests/ — clean.

Notes

• errors_field records use stringified fan_out_index per fixture 019's choice — see comment in _fan_in_collect.
• Fan-out chain composition is built fresh per instance; for N instances × M instance_middlewares = N×M closures per dispatch. Documented in compose_chain (Phase 2 finding) and fan_out.py module docstring; worth measuring at large N once that path is exercised in real workloads.
• Spec version pin intentionally not bumped (per phased-rollout strategy).

[Copilot]

Pull request overview

Adds spec v0.4 pipeline-utilities §9 fan-out support to the graph engine, including bounded-concurrency per-instance subgraph dispatch, new fan-out error types, observer tagging via fan_out_index, and conformance/unit-test coverage.

Changes:

• Introduces FanOutNode/FanOutConfig runtime with fail-fast vs collect policies, empty-input handling, and per-instance middleware.
• Threads fan_out_index through invocation contexts and node observer events; adds compiled-engine dispatch path for fan-out nodes.
• Extends conformance adapter/harness to translate fan_out and related directives; adds a new test_fan_out.py unit test suite.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/openarmature/graph/fan_out.py	Implements fan-out runtime node, projection, bounded concurrency, and fan-in behavior.
src/openarmature/graph/builder.py	Adds GraphBuilder.add_fan_out_node with compile-time validation and config wiring.
src/openarmature/graph/compiled.py	Dispatches FanOutNode as a single engine step and emits events with fan_out_index.
src/openarmature/graph/observer.py	Adds fan_out_index to invocation context; supports descending into fan-out instances.
src/openarmature/graph/errors.py	Adds fan-out compile/runtime error categories.
src/openarmature/graph/__init__.py	Exports fan-out node/config and new fan-out errors.
tests/conformance/adapter.py	Adds fan-out fixture translation and new directive seams (update_pure, update_from_field, flaky_by_index, flaky_instance_only).
tests/conformance/test_pipeline_utilities.py	Enables fan-out fixtures up to phase 3; supports per-fan-out instance middleware translation and shared subgraph blocks.
tests/conformance/test_conformance.py	Removes fan-out-related directives from unsupported set.
tests/unit/test_fan_out.py	Adds comprehensive unit tests for fan-out spec corner cases.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.