feat(steps): add reflection ensemble step

ace/__init__.py

@@ -53,6 +53,7 @@
         ObservabilityStep,
         PersistStep,
         ReflectStep,
+        ReflectionEnsembleStep,
         UpdateStep,
         learning_tail,
     )
@@ -105,6 +106,7 @@
     "AgentStep": ("ace.steps", "AgentStep"),
     "EvaluateStep": ("ace.steps", "EvaluateStep"),
     "ReflectStep": ("ace.steps", "ReflectStep"),
+    "ReflectionEnsembleStep": ("ace.steps", "ReflectionEnsembleStep"),
     "UpdateStep": ("ace.steps", "UpdateStep"),
     "DeduplicateStep": ("ace.steps", "DeduplicateStep"),
     "CheckpointStep": ("ace.steps", "CheckpointStep"),
@@ -175,6 +177,7 @@ def __getattr__(name: str) -> object:
     "AgentStep",
     "EvaluateStep",
     "ReflectStep",
+    "ReflectionEnsembleStep",
     "UpdateStep",
     "DeduplicateStep",
     "CheckpointStep",

ace/steps/__init__.py

@@ -23,6 +23,7 @@
 from .observability import ObservabilityStep
 from .persist import PersistStep
 from .reflect import ReflectStep
+from .reflection_ensemble import ReflectionEnsembleStep
 from .update import UpdateStep
 
 __all__ = [
@@ -35,12 +36,25 @@
     "ObservabilityStep",
     "PersistStep",
     "ReflectStep",
+    "ReflectionEnsembleStep",
     "UpdateStep",
     "learning_tail",
 ]
 
 
-def _reflect_step(reflector: ReflectorLike) -> StepProtocol[ACEStepContext]:
+def _reflect_step(
+    reflector: ReflectorLike,
+    *,
+    reflection_ensemble_size: int = 1,
+    reflection_ensemble_workers: int | None = None,
+) -> StepProtocol[ACEStepContext]:
+    if reflection_ensemble_size != 1:
+        return ReflectionEnsembleStep(
+            reflector,
+            ensemble_size=reflection_ensemble_size,
+            workers=reflection_ensemble_workers,
+        )
+
     provides = getattr(reflector, "provides", ())
     if callable(reflector) and "reflections" in provides:
         return reflector  # type: ignore[return-value]
@@ -56,6 +70,8 @@ def learning_tail(
     dedup_interval: int = 10,
     checkpoint_dir: str | Path | None = None,
     checkpoint_interval: int = 10,
+    reflection_ensemble_size: int = 1,
+    reflection_ensemble_workers: int | None = None,
 ) -> list[StepProtocol[ACEStepContext]]:
     """Return the standard ACE learning steps.
 
@@ -71,11 +87,17 @@ def learning_tail(
     reflector itself when it already satisfies the step protocol and exposes
     ``provides = {'reflections'}``, followed by ``UpdateStep``. The agentic
     SkillManager mutates the skillbook directly through its tools, so no
-    ``ApplyStep`` follows. Optional ``DeduplicateStep`` and ``CheckpointStep``
-    are appended when configured.
+    ``ApplyStep`` follows. Set ``reflection_ensemble_size > 1`` to run the
+    same reflector multiple times on each trace and pass all resulting
+    reflections to one ``UpdateStep``. Optional ``DeduplicateStep`` and
+    ``CheckpointStep`` are appended when configured.
     """
     steps: list[StepProtocol[ACEStepContext]] = [
-        _reflect_step(reflector),
+        _reflect_step(
+            reflector,
+            reflection_ensemble_size=reflection_ensemble_size,
+            reflection_ensemble_workers=reflection_ensemble_workers,
+        ),
         UpdateStep(skill_manager, skillbook),
     ]
     if dedup_manager:

ace/steps/reflection_ensemble.py

@@ -0,0 +1,109 @@
+"""ReflectionEnsembleStep — run N independent reflections for one trace."""
+
+from __future__ import annotations
+
+from types import MappingProxyType
+
+from pipeline import Pipeline
+from pipeline.protocol import StepProtocol
+
+from ..core.context import ACEStepContext
+from ..core.outputs import ReflectorOutput
+from ..protocols import ReflectorLike
+from .reflect import ReflectStep
+
+
+class ReflectionEnsembleStep:
+    """Run the same trace through a reflector multiple times.
+
+    This is a map-reduce step: one input context is expanded into
+    ``ensemble_size`` sub-contexts, each sub-context runs the normal
+    reflection step through the pipeline engine, and the resulting
+    ``ReflectorOutput`` objects are flattened back onto the original
+    context as ``ctx.reflections``.
+
+    This intentionally sits above the generic ``Branch`` primitive. Branch is
+    for peer child pipelines whose contexts can be merged by a general merge
+    strategy; reflection ensembles need N repeated writes to ``reflections`` to
+    be concatenated and then consumed by one ``UpdateStep``.
+
+    It is also distinct from runner ``epochs``. Epochs replay samples or traces
+    as later passes after the skillbook has changed; an ensemble repeats only
+    reflection on the current trace before a single SkillManager update.
+    """
+
+    requires = frozenset({"trace", "skillbook"})
+    provides = frozenset({"reflections"})
+
+    async_boundary = True
+    max_workers = 1
+
+    def __init__(
+        self,
+        reflector: ReflectorLike | StepProtocol[ACEStepContext],
+        *,
+        ensemble_size: int = 2,
+        workers: int | None = None,
+    ) -> None:
+        if ensemble_size < 1:
+            raise ValueError("ensemble_size must be >= 1.")
+        if workers is not None and workers < 1:
+            raise ValueError("workers must be >= 1.")
+
+        self.reflector = reflector
+        self.ensemble_size = ensemble_size
+        self.workers = workers or min(ensemble_size, ReflectStep.max_workers)
+        self._reflect_step = self._coerce_reflect_step(reflector)
+
+    @staticmethod
+    def _coerce_reflect_step(
+        reflector: ReflectorLike | StepProtocol[ACEStepContext],
+    ) -> StepProtocol[ACEStepContext]:
+        provides = getattr(reflector, "provides", ())
+        if callable(reflector) and "reflections" in provides:
+            return reflector  # type: ignore[return-value]
+        return ReflectStep(reflector)  # type: ignore[arg-type]
+
+    def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
+        subcontexts = [
+            ctx.replace(
+                metadata=MappingProxyType(
+                    {
+                        **ctx.metadata,
+                        "reflection_ensemble_index": index,
+                        "reflection_ensemble_size": self.ensemble_size,
+                    }
+                )
+            )
+            for index in range(self.ensemble_size)
+        ]
+
+        pipe = Pipeline().then(self._reflect_step)
+        results = pipe.run(subcontexts, workers=self.workers)
+        pipe.wait_for_background()
+
+        reflections: list[ReflectorOutput] = []
+        for index, result in enumerate(results):
+            if result.error is not None:
+                raise RuntimeError(
+                    f"Reflection ensemble member {index} failed."
+                ) from result.error
+            if result.output is None:
+                raise RuntimeError(
+                    f"Reflection ensemble member {index} produced no output."
+                )
+            reflections.extend(result.output.reflections)
+
+        if not reflections:
+            raise RuntimeError("Reflection ensemble produced no reflections.")
+
+        return ctx.replace(
+            reflections=tuple(reflections),
+            metadata=MappingProxyType(
+                {
+                    **ctx.metadata,
+                    "reflection_ensemble_size": self.ensemble_size,
+                    "reflection_ensemble_completed": len(reflections),
+                }
+            ),
+        )

docs/design/ACE_ARCHITECTURE.md

@@ -103,7 +103,7 @@ Key fields:
 | `skillbook` | `SkillbookView \| None` | Read-only projection of the real Skillbook |
 | `trace` | `object \| None` | Raw execution record — any type, no enforced schema |
 | `agent_output` | `AgentOutput \| None` | Produced by `AgentStep` |
-| `reflections` | `tuple[ReflectorOutput, ...]` | Produced by `ReflectStep` / `RRStep` |
+| `reflections` | `tuple[ReflectorOutput, ...]` | Produced by `ReflectStep` / `ReflectionEnsembleStep` / `RRStep` |
 | `skill_manager_output` | `UpdateBatch \| None` | Produced by `UpdateStep` (audit log of mutations the SM already applied) |
 | `injected_skill_ids` | `tuple[str, ...]` | Produced by `AgentStep` — skill IDs rendered into the agent prompt; downstream attribution scope |
 | `epoch`, `total_epochs` | `int` | Runner bookkeeping |
@@ -112,7 +112,7 @@ Key fields:
 
 The `trace` field holds the raw execution record from any external system — a browser-use `AgentHistoryList`, a LangChain result dict, a Claude Code transcript, or any arbitrary Python object. The Reflector receives the raw trace and is responsible for making sense of it.
 
-The `reflections` field is a tuple. In single-trace mode, it's a 1-tuple. In batch mode, it holds one `ReflectorOutput` per trace. Downstream steps iterate uniformly — no special-casing.
+The `reflections` field is a tuple. In single-reflection mode, it's a 1-tuple. In ensemble mode, it holds multiple independent `ReflectorOutput` objects for the same trace. Downstream steps iterate uniformly — no special-casing.
 
 ### Context vs constructor injection
 
@@ -173,6 +173,7 @@ Reusable step implementations in `ace/steps/`. Each satisfies `StepProtocol[ACES
 | **AgentStep** | `sample`, `skillbook` | `agent_output` | None | 1 |
 | **EvaluateStep** | `sample`, `agent_output` | `trace` | None | 1 |
 | **ReflectStep** | `trace`, `skillbook` | `reflections` | None | 3; `async_boundary = True` |
+| **ReflectionEnsembleStep** | `trace`, `skillbook` | `reflections` | None | 1; map-reduce fan-out to repeated `ReflectStep` workers |
 | **UpdateStep** | `reflections`, `skillbook` | `skill_manager_output` | Agentic SkillManager mutates skillbook directly via ADD / UPDATE / REMOVE / TAG tools; output is an audit log | 1 |
 | **DeduplicateStep** | `global_sample_index` | — | Consolidates similar skills | 1 |
 | **CheckpointStep** | `global_sample_index` | — | Saves skillbook to disk | 1 |
@@ -299,6 +300,8 @@ All runners provide a `from_roles` factory that takes pre-built role instances.
 
 Every integration assembles the same `[Reflect → Update → Apply]` suffix. `learning_tail()` returns this standard step list, with optional dedup and checkpoint steps. If the provided reflector already exposes `provides = {'reflections'}` (e.g. `RRStep`), it's inserted directly instead of being wrapped in `ReflectStep`.
 
+When `reflection_ensemble_size > 1`, the tail uses `ReflectionEnsembleStep`: a specialized map-reduce step that runs repeated reflections over the same trace and concatenates them into one `ctx.reflections` tuple for a single `UpdateStep`. This complements the generic pipeline `Branch` primitive rather than replacing it; `Branch` is for peer child pipelines with general context merge strategies, while reflection ensembles need repeated writes to the same learning field to be collected, not conflict-resolved. It is also distinct from runner `epochs`: epochs replay samples or traces as later passes after the skillbook has changed, while an ensemble repeats only the reflection step on the current trace before one SkillManager update.
+
 ---
 
 ## Integration Pattern

docs/design/ACE_REFERENCE.md

@@ -23,7 +23,7 @@ from ace import ACERunner
 
 # Core steps
 from ace import (
-    AgentStep, EvaluateStep, ReflectStep, UpdateStep,
+    AgentStep, EvaluateStep, ReflectStep, ReflectionEnsembleStep, UpdateStep,
     DeduplicateStep, CheckpointStep, LoadTracesStep, ExportSkillbookMarkdownStep,
     ObservabilityStep, PersistStep, learning_tail,
 )
@@ -259,6 +259,46 @@ class ReflectStep:
         return ctx.replace(reflections=(reflection,))
 ```
 
+### ReflectionEnsembleStep
+
+Runs the same trace through a reflector multiple times and places all
+resulting `ReflectorOutput` objects on `ctx.reflections`. This is a
+map-reduce step: it expands one context into `ensemble_size` sub-contexts,
+runs the normal reflection step through the pipeline engine, waits for those
+reflections, and merges the outputs back onto the original context.
+
+This is deliberately separate from generic `Branch`. `Branch` runs peer child
+pipelines and merges their returned contexts with a general merge strategy;
+`ReflectionEnsembleStep` repeats the same reflection step and concatenates the
+resulting `reflections` tuples so one downstream `UpdateStep` can reason over
+the full ensemble.
+
+It is also not the same as `epochs`. Epochs replay samples or traces as later
+passes after earlier updates have changed the skillbook. A reflection ensemble
+does not advance the runner loop; it repeats reflection on the current trace
+before one SkillManager update.
+
+```python
+class ReflectionEnsembleStep:
+    requires = frozenset({"trace", "skillbook"})
+    provides = frozenset({"reflections"})
+
+    async_boundary = True
+    max_workers = 1
+
+    def __init__(
+        self,
+        reflector: ReflectorLike | StepProtocol[ACEStepContext],
+        *,
+        ensemble_size: int = 2,
+        workers: int | None = None,
+    ) -> None: ...
+
+    def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
+        ...
+        return ctx.replace(reflections=(reflection_1, reflection_2, ...))
+```
+
 ### UpdateStep
 
 Runs the agentic `SkillManager`. The SM's tools mutate the real `Skillbook`
@@ -417,10 +457,20 @@ def learning_tail(
     dedup_interval: int = 10,
     checkpoint_dir: str | Path | None = None,
     checkpoint_interval: int = 10,
+    reflection_ensemble_size: int = 1,
+    reflection_ensemble_workers: int | None = None,
 ) -> list[StepProtocol[ACEStepContext]]:
     """Return the standard ACE learning steps."""
     steps: list[StepProtocol[ACEStepContext]] = [
-        ReflectStep(reflector),
+        (
+            ReflectStep(reflector)
+            if reflection_ensemble_size == 1
+            else ReflectionEnsembleStep(
+                reflector,
+                ensemble_size=reflection_ensemble_size,
+                workers=reflection_ensemble_workers,
+            )
+        ),
         UpdateStep(skill_manager, skillbook),
     ]
     if dedup_manager:
@@ -430,6 +480,13 @@ def learning_tail(
     return steps
 ```
 
+`reflection_ensemble_size` is the high-level API for repeated reflection over
+the same trace. Use generic `Branch` when branches perform different pipeline
+work or write disjoint fields; use `ReflectionEnsembleStep` / this helper when
+the goal is N independent reflector opinions followed by one SkillManager
+update. Use runner `epochs` when you want additional passes over samples or
+traces after the skillbook has evolved.
+
 ### TraceAnalyser `from_roles`
 
 ```python

docs/guides/composing-pipelines.md

@@ -205,6 +205,17 @@ pipe = Pipeline([
 See the [Pipeline Engine docs](../pipeline/branching.md) for full branching
 and merge strategy details.
 
+!!! note "Reflection ensembles vs Branch and epochs"
+    Use `Branch` for parallel child pipelines that do different work or write
+    disjoint fields. For N independent reflections over the same trace followed
+    by one SkillManager update, use `learning_tail(...,
+    reflection_ensemble_size=N)` instead. It is a specialized map-reduce step
+    that collects repeated `reflections` outputs rather than asking a generic
+    branch merge strategy to resolve them. This is also different from
+    `epochs`, which replay samples or traces as later passes after the
+    skillbook has changed; ensemble reflection happens before the current
+    SkillManager update.
+
 ## Using RRStep (Recursive Reflector)
 
 `RRStep` satisfies both `StepProtocol` and `ReflectorLike`, so it can be used