[mvallebr]

Summary

Full-audit follow-up. The sync engine runs in production and is the behavioral reference; wherever the engines diverged, this PR changes the async engine to match sync — sync behavior is untouched except where the same decision was hoisted from runtime introspection into compiled DAG metadata (provably identical values, computed earlier).

The audit also surfaced concrete bugs (a validation that could never fire, a vacuous duplicate-name check, a broken async test helper that silently disconnected its producer, and real corpus drift) plus a hidden deadlock discovered by registering the orphaned nested_fanout corpus packs (#128).

Changes

Async → sync parity (the engine drifts are user-visible contracts):

• publish() routes on the compiled publish_plan.strategy and reads completion_policy from output_contract instead of re-deriving from node.mode + runtime output type — completion events now fire at the same point in both engines
• Output contracts are validated at publish time (TypeError on mismatch), like sync; value-compiled async steps can no longer silently return streams
• Materializer crashes propagate and fail the step (previously swallowed into an empty published list under CONTINUE); mid-stream item failures keep the sync prefix semantics (valid prefix + had_error) while preserving lazy, out-of-core materializer access
• Producer failures are delivered in-band to every consumer branch under any on_error policy — no more silent stream truncation under CONTINUE (mirrors SyncFanout's ExceptionMarker delivery); queued PipelineStopExceptions carry their cause
• AsyncPipelineExecutor.execute() awaits cleanup() in finally on failure (no orphaned pump tasks); cleanup() logs instead of bare except: pass
• Terminal lazy streams are stored in executor.outputs like sync

Build/run separation:

• The builder now compiles fn_kind (sync/sync_generator/async/async_generator) and async_stream_deps into the DAG; runtimes consult that metadata instead of inspect.is*function / annotation-__origin__ re-derivations
• The builder compiles publish_async_fanout for single-consumer async streams too — EACH-mode consumers unroll through queue branches by contract, so the single-consumer case was always queue-delivered in practice; the DAG JSON now tells the truth about it

Core fixes:

• Deleted the unreachable "requires a custom materializer" validation (its factory identity check could never be true after resolution) + _is_builtin_type; regression test locks in that the default memory factory handles custom item types
• Collapsed ~130 near-duplicate lines (_validate_no_async_handlers/_validate_no_sync_handlers) into one parameterized check; error messages preserved
• _validate_declared_step_names passed an empty dag to the duplicate check (vacuous) — replaced with an explicit reserved-name check; real duplicate detection stays in the expanded pass
• get_execution_levels() raises a descriptive ValueError on cycles instead of silently returning a partial topological order
• Removed dead _check_scalar_producer_to_iterable_consumer and the __import__("typing") hack; composite materializers reuse the core is_async_callable (superset: also detects partials/async callables)

Test infrastructure:

• Registered the orphaned nested_fanout corpus packs (both engines) with full build-time specs; discovered the topology deadlocks the sync engine — tracked in fix(execution): nested fan-out/fan-in topology deadlocks the sync engine #128, runtime assertions stay unregistered until then
• Fixed real corpus drift: the async linear twin had silently dropped its step-level observer (and the frozen observers JSON entry)
• New corpus-symmetry meta-test: pack sets must match across engines, so an orphaned pack can never recur silently
• test_runner_step_results (both engines): a failing iterator propagates instead of being swallowed into a truncated "passing" list
• Documented why the shared run_pipeline fixture is sync-only (pipelines are engine-specific by contract — test: engine-agnostic dual-engine corpus execution (unlock shared run_pipeline fixture) #129 tracks the dual-engine design)
• Direct unit tests for ExecutionState; module docstrings for the ~25 flagged modules; English state.py docs with fail-loud seed(); typed _identity/async_adapter

Deferred items are filed as issues #128–#132.

Tests

• Full suite: 816 passed (was 802), including new regression/parity/symmetry/state tests
• Corpus spec tests (sync+async) and sync/async parity meta-tests green; corpus JSON stays frozen — new compiled fields are covered by dedicated unit tests and excluded from the corpus normalization like output_contract/publish_plan before them
• ruff format --check + ruff check --select F401 clean; pre-commit patch coverage gate passes
• Note for reviewers: several async tests were updated because they encoded the old divergent semantics (under-annotated async generators sync rejects loudly; the async threshold helper passed fn directly where the sync twin wraps it, silently disconnecting producer → consumer; the handoff failure test asserted truncation). The sync side was not weakened anywhere.