test(conformance): drive 0011 parallel-branches

Extends the conformance harness so the eight spec fixtures introduced
by proposal 0011 parse and run end-to-end:

- pipeline-utilities/032-038 (parallel-branches basic, fail-fast,
  collect, different-state-schemas, with-branch-middleware-retry,
  determinism, compose-with-fan-out).
- graph-engine/021-observer-branch-name (NodeEvent.branch_name).

Harness changes:

- ParallelBranchSpec / ParallelBranchesSpec models on NodeSpec;
  sleep_ms node-companion modifier; recoverable_state expected key
  on the pipeline-utilities expected discriminator.
- adapter.build_graph dispatches the parallel_branches directive to
  builder.add_parallel_branches_node, wrapping the result in a
  tracing variant so the execution-order trace records the
  dispatcher as one engine step.
- test_pipeline_utilities driver: lifts the fixture-number gate from
  23 to 38; loads top-level plural subgraphs blocks; translates
  per-branch middleware lists; wires graph-attached observers per
  run; asserts parallel-branches observer_event_invariants
  (branch_started_event_order, alpha_inner_attempt_indices_seen,
  fan-out-inside-branch invariants, plain-branch invariants);
  routes fail_fast assertions to surface branch_name + cause_message
  + recoverable_state alongside category.
- test_conformance driver: drops the 021 skip and adds an
  invariants checker for outermost-events-have-no-branch-name and
  inner-events-carry-correct-branch-name.

The checkpoint-resume fixtures (024-031) move into the deferred set
since their cases-shape with first_run_expected_error / resume:
blocks is driven by test_checkpoint.py.

Author: chris-colinsky

tests/conformance/adapter.py

@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import copy
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass, field
@@ -19,12 +20,14 @@
 
 from openarmature.graph import (
     END,
+    BranchSpec,
     CompiledGraph,
     EndSentinel,
     ExplicitMapping,
     FanOutNode,
     FieldNameMatching,
     GraphBuilder,
+    ParallelBranchesNode,
     ProjectionStrategy,
     Reducer,
     State,
@@ -56,6 +59,14 @@ def _parse_type(s: str) -> Any:
         return float
     if s == "bool":
         return bool
+    # Unparameterized container types — parallel-branches fixtures
+    # 034/035/037 use ``dict`` and ``list<dict>`` as state-field types
+    # for accumulator slots (branch_errors, merged_dict, collected_labels)
+    # where the element shape is heterogeneous across branches.
+    if s == "dict":
+        return dict[str, Any]
+    if s == "list<dict>":
+        return list[dict[str, Any]]
     if s.startswith("list<") and s.endswith(">"):
         return list[_parse_type(s[5:-1])]
     if s.startswith("dict<") and s.endswith(">"):
@@ -357,6 +368,23 @@ async def fn(_state: Any) -> Mapping[str, Any]:
     return fn
 
 
+def _wrap_with_sleep(
+    fn: Callable[[Any], Awaitable[Mapping[str, Any]]],
+    sleep_ms: int,
+) -> Callable[[Any], Awaitable[Mapping[str, Any]]]:
+    # ``sleep_ms`` companion modifier on a NodeSpec — sleep that many
+    # milliseconds before the wrapped body fires. Used by parallel-branches
+    # fixtures 033 (slow third branch for fail-fast cancellation) and 037
+    # (randomized completion timing to verify insertion-order determinism).
+    delay = sleep_ms / 1000.0
+
+    async def fn_with_sleep(state: Any) -> Mapping[str, Any]:
+        await asyncio.sleep(delay)
+        return await fn(state)
+
+    return fn_with_sleep
+
+
 @dataclass(frozen=True)
 class _TracingFanOutNode(FanOutNode[State, State]):
     """Conformance helper: a FanOutNode that appends its name to a shared
@@ -383,6 +411,24 @@ async def run_with_context(
         )
 
 
+@dataclass(frozen=True)
+class _TracingParallelBranchesNode(ParallelBranchesNode[State]):
+    """Conformance helper: a ParallelBranchesNode that appends its name
+    to the shared trace list once when the engine runs it. The
+    parallel-branches dispatcher itself counts as one engine step from
+    the parent's POV per §11.6, mirroring the fan-out tracing wrapper."""
+
+    trace_list: list[str] = field(default_factory=list[str])
+
+    async def run_with_context(
+        self,
+        state: State,
+        context: _InvocationContext,
+    ) -> Mapping[str, Any]:
+        self.trace_list.append(self.name)
+        return await super().run_with_context(state, context)
+
+
 @dataclass(frozen=True)
 class _TracingSubgraphNode(SubgraphNode[State, State]):
     """Conformance helper: a SubgraphNode that appends its name to a shared
@@ -457,6 +503,7 @@ def build_graph(
     node_middleware: Mapping[str, Sequence[Any]] | None = None,
     graph_middleware: Sequence[Any] | None = None,
     fan_out_instance_middleware: Mapping[str, Sequence[Any]] | None = None,
+    parallel_branches_branch_middleware: Mapping[str, Mapping[str, Sequence[Any]]] | None = None,
 ) -> BuiltGraph:
     """Translate a graph-shaped fixture block into a `BuiltGraph`.
 
@@ -486,6 +533,7 @@ def build_graph(
     subgraphs = subgraphs or {}
     node_middleware = node_middleware or {}
     fan_out_instance_middleware = fan_out_instance_middleware or {}
+    parallel_branches_branch_middleware = parallel_branches_branch_middleware or {}
 
     for mw in graph_middleware or ():
         builder.add_middleware(mw)
@@ -505,7 +553,8 @@ def build_graph(
                 trace_list=trace,
                 middleware=per_node_mw,
             )
-        elif "fan_out" in node_spec:
+            continue
+        if "fan_out" in node_spec:
             _add_fan_out_node(
                 builder,
                 node_name,
@@ -514,55 +563,47 @@ def build_graph(
                 trace,
                 instance_middleware=fan_out_instance_middleware.get(node_name, ()),
             )
-        elif "raises" in node_spec:
-            builder.add_node(
+            continue
+        if "parallel_branches" in node_spec:
+            _add_parallel_branches_node(
+                builder,
                 node_name,
-                _make_raising_fn(node_name, node_spec["raises"], trace),
-                middleware=per_node_mw,
+                node_spec["parallel_branches"],
+                subgraphs,
+                trace,
+                branch_middleware=parallel_branches_branch_middleware.get(node_name, {}),
             )
+            continue
+
+        body: Callable[[Any], Awaitable[Mapping[str, Any]]]
+        if "raises" in node_spec:
+            body = _make_raising_fn(node_name, node_spec["raises"], trace)
         elif "flaky" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_flaky_fn(node_name, node_spec["flaky"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_flaky_fn(node_name, node_spec["flaky"], trace)
         elif "flaky_by_index" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_flaky_by_index_fn(node_name, node_spec["flaky_by_index"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_flaky_by_index_fn(node_name, node_spec["flaky_by_index"], trace)
         elif "flaky_instance_only" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_flaky_instance_only_fn(node_name, node_spec["flaky_instance_only"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_flaky_instance_only_fn(node_name, node_spec["flaky_instance_only"], trace)
         elif "update" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_update_fn(node_name, node_spec["update"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_update_fn(node_name, node_spec["update"], trace)
         elif "update_pure" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_pure_update_fn(node_name, node_spec["update_pure"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_pure_update_fn(node_name, node_spec["update_pure"], trace)
         elif "update_from_field" in node_spec:
-            builder.add_node(
-                node_name,
-                _make_update_from_field_fn(node_name, node_spec["update_from_field"], trace),
-                middleware=per_node_mw,
-            )
+            body = _make_update_from_field_fn(node_name, node_spec["update_from_field"], trace)
         else:
             raise ValueError(
                 f"node {node_name!r} has no recognized directive "
                 "(update / update_pure / update_from_field / raises / flaky / "
-                "flaky_by_index / flaky_instance_only / fan_out / subgraph)"
+                "flaky_by_index / flaky_instance_only / fan_out / parallel_branches / "
+                "subgraph)"
             )
 
+        sleep_ms = node_spec.get("sleep_ms")
+        if sleep_ms is not None:
+            body = _wrap_with_sleep(body, int(sleep_ms))
+
+        builder.add_node(node_name, body, middleware=per_node_mw)
+
     for edge_spec in spec.get("edges", []):
         source = edge_spec["from"]
         if "to" in edge_spec:
@@ -623,6 +664,8 @@ def _record_event(event: NodeEvent) -> dict[str, Any]:
         rec["error"] = event.error.category
     if event.fan_out_index is not None:
         rec["fan_out_index"] = event.fan_out_index
+    if event.branch_name is not None:
+        rec["branch_name"] = event.branch_name
     return rec
 
 
@@ -734,6 +777,57 @@ def _add_fan_out_node(
     )
 
 
+def _add_parallel_branches_node(
+    builder: GraphBuilder[Any],
+    node_name: str,
+    cfg: Mapping[str, Any],
+    subgraphs: Mapping[str, CompiledGraph[State]],
+    trace: list[str],
+    *,
+    branch_middleware: Mapping[str, Sequence[Any]],
+) -> None:
+    """Translate a fixture's ``parallel_branches:`` block into a
+    ``builder.add_parallel_branches_node`` call.
+
+    Each branch's ``subgraph`` name resolves against the shared
+    ``subgraphs`` registry (built from the fixture's top-level
+    ``subgraphs:`` block). ``branch_middleware`` maps branch-name to a
+    pre-translated middleware list; the test driver populates it from
+    each branch's ``middleware:`` block.
+    """
+    branches_cfg = cast("dict[str, dict[str, Any]]", cfg["branches"])
+    branches: dict[str, BranchSpec[Any]] = {}
+    for branch_name, branch_cfg in branches_cfg.items():
+        sub_compiled = subgraphs[branch_cfg["subgraph"]]
+        branches[branch_name] = BranchSpec(
+            subgraph=sub_compiled,
+            inputs=dict(branch_cfg.get("inputs") or {}),
+            outputs=dict(branch_cfg.get("outputs") or {}),
+            middleware=tuple(branch_middleware.get(branch_name, ())),
+        )
+
+    builder.add_parallel_branches_node(
+        node_name,
+        branches=branches,
+        error_policy=cfg.get("error_policy", "fail_fast"),
+        errors_field=cfg.get("errors_field"),
+    )
+
+    # Swap the registered node for a tracing variant so the
+    # conformance trace records the dispatcher as one engine step. The
+    # builder's validation has already run; we only replace the stored
+    # Node instance.
+    original = cast("ParallelBranchesNode[State]", builder._nodes[node_name])
+    builder._nodes[node_name] = _TracingParallelBranchesNode(
+        name=original.name,
+        branches=original.branches,
+        error_policy=original.error_policy,
+        errors_field=original.errors_field,
+        middleware=original.middleware,
+        trace_list=trace,
+    )
+
+
 def _resolve_callable_int_resolver(cfg: Mapping[str, Any]) -> Callable[[Any], int]:
     """Build a state-reader callable from a fixture's callable config.
 

tests/conformance/harness/directives.py

@@ -235,6 +235,33 @@ class FanOutSpec(_AllowExtras):
     instance_middleware: list[MiddlewareSpec] | None = None
 
 
+class ParallelBranchSpec(_AllowExtras):
+    """One entry inside a ``parallel_branches.branches`` mapping.
+
+    Permissive on extras because fixtures may carry extra knobs
+    (e.g., per-branch annotations the harness ignores).
+    """
+
+    subgraph: str
+    inputs: dict[str, str] | None = None
+    outputs: dict[str, str] | None = None
+    middleware: list[MiddlewareSpec] | None = None
+
+
+class ParallelBranchesSpec(_AllowExtras):
+    """``parallel_branches:`` block on a NodeSpec (pipeline-utilities §11).
+
+    Mirrors :class:`FanOutSpec` but topology-driven: M heterogeneous
+    branches, each referencing a different compiled subgraph by name
+    against the case's top-level ``subgraphs:`` block. Branch insertion
+    order is preserved per §11.8.
+    """
+
+    branches: dict[str, ParallelBranchSpec]
+    error_policy: Literal["fail_fast", "collect"] | None = None
+    errors_field: str | None = None
+
+
 class CallsLlmSpec(_AllowExtras):
     """LLM-using node: sends ``messages`` to the harness's mock provider
     and stores the response (assistant content) in ``stores_response_in``.
@@ -294,6 +321,7 @@ class NodeSpec(_ForbidExtras):
     raises: str | None = None
     subgraph: str | None = None
     fan_out: FanOutSpec | None = None
+    parallel_branches: ParallelBranchesSpec | None = None
     flaky: FlakySpec | None = None
     flaky_by_index: FlakyByIndexSpec | None = None
     flaky_per_index: FlakyPerIndexSpec | None = None
@@ -309,6 +337,13 @@ class NodeSpec(_ForbidExtras):
     also_emits_via_global_tracer: GlobalTracerSpec | None = None
     # Pair with ``raises`` to specify the error category (graph-engine §4).
     error_category: str | None = None
+    # Parallel-branches fixtures (033, 037): the node sleeps this many
+    # milliseconds before its update fires. Used to force deterministic
+    # branch-completion ordering (037 — different branches finish at
+    # different wall-clock times yet final state must be insertion-order
+    # deterministic per §11.8) and to slow a third branch so fail-fast
+    # cancellation has time to land before it finishes (033).
+    sleep_ms: int | None = None
 
     _PRIMARY_FIELDS = (
         "update",
@@ -318,6 +353,7 @@ class NodeSpec(_ForbidExtras):
         "raises",
         "subgraph",
         "fan_out",
+        "parallel_branches",
         "flaky",
         "flaky_by_index",
         "flaky_per_index",

tests/conformance/harness/expectations.py

@@ -141,6 +141,12 @@ class PipelineUtilitiesExpected(_ForbidExtras):
     # - dict[recorder_name, list[record]] when multiple recorders (001).
     # - list[record] flat when a single recorder.
     trace_records: Any = None
+    # Parallel-branches fixtures (032-038). On fail_fast,
+    # ``recoverable_state`` carries the pre-entry parent state
+    # snapshot per spec §11.5; the harness asserts it equals the
+    # ``recoverable_state`` attached to the raised
+    # ``ParallelBranchesBranchFailed``.
+    recoverable_state: dict[str, Any] | None = None
 
 
 # ---------------------------------------------------------------------------
@@ -201,6 +207,7 @@ class ObservabilityExpected(_ForbidExtras):
         "timing_records",
         "trace_records",
         "expected_observer_event",
+        "recoverable_state",
     }
 )
 _OBSERVABILITY_KEYS = frozenset(