Engine

class AsyncGEPAEngine(Generic[DataInst, Trajectory, RolloutOutput]):  # noqa: UP046  # TypeVars are exported names shared across ports and engine
    """Async evolution engine orchestrating the GEPA loop.

    This engine executes the core evolution algorithm:
    1. Evaluate baseline candidate
    2. For each iteration until max_iterations or convergence:
       a. Generate reflective dataset from traces
       b. Propose new candidate text (an empty reflection after one retry,
          or a reflection that times out, records a skipped iteration and
          moves on)
       c. Evaluate proposal
       d. Accept if improves above threshold
       e. Record iteration
    3. Return frozen EvolutionResult

    The engine is generic over the adapter's ``DataInst``, ``Trajectory``
    and ``RolloutOutput`` types.

    Attributes:
        adapter (AsyncGEPAAdapter): Implementation of AsyncGEPAAdapter protocol.
        config (EvolutionConfig): Evolution parameters.

    Examples:
        Basic usage:

        ```python
        from gepa_adk.engine import AsyncGEPAEngine
        from gepa_adk.domain.models import EvolutionConfig, Candidate

        engine = AsyncGEPAEngine(
            adapter=my_adapter,
            config=EvolutionConfig(max_iterations=50),
            initial_candidate=Candidate(components={"instruction": "Be helpful"}),
            batch=training_data,
        )
        result = await engine.run()
        print(f"Final score: {result.final_score}")
        ```

    Notes:
        Avoid reusing engine instances after run() completes; a repeated
        run() still starts fresh through ``_reset_run_state``.
        When a seeded ``rng`` is provided, it is used for the auto-created
        merge proposer. The API layer passes the same ``rng`` to candidate
        selectors for full determinism across stochastic components.
        Evaluation, acceptance, reuse, Pareto and merge log events carry
        ``candidate_id`` (``Candidate.id``) as the per-candidate correlation
        key. Rows named in ``EvaluationBatch.failed_indices`` are counted per
        iteration and reported on the result, and so is the token usage the
        evaluated rows' trajectories report and the reflection calls the
        adapter's proposer reports (``TokenRollup`` with ``evaluation`` and
        ``reflection`` splits). A proposal whose
        ``Candidate.id`` was already scored in the run is not evaluated
        again; its iteration is recorded with ``skip_reason="duplicate"``.
        A proposal whose ``output_schema`` text fails validation is not
        evaluated; its iteration is recorded with
        ``skip_reason="schema_validation_failed"`` and counts toward
        stagnation, so ``max_iterations``, ``patience`` and stoppers still
        end the run. A retryable ``ReflectionError`` (the proposer already
        retried once) is recorded the same way with
        ``skip_reason="reflection_error"``; a non-retryable one aborts the
        run with the partial result attached to the error, and so do
        ``config.max_consecutive_reflection_errors`` consecutive ones with
        the same cause type and text. An
        ``IncompleteProposalError`` (truncated reflection output) is recorded
        with ``skip_reason="incomplete_proposal"`` and not evaluated.
        A proposal that ``config.proposal_validator`` rejects is recorded
        with ``skip_reason="proposal_rejected"`` and the validator's reason,
        counts toward stagnation and is not evaluated.
        A merge candidate that is already scored or has an invalid schema is
        not evaluated, and the iteration that scheduled it is still recorded.
        An evaluation policy takes effect only through the Pareto state that
        a candidate selector creates. No policy and no selector is full
        evaluation; a selector alone is full evaluation over the Pareto
        state; a selector with a policy uses the policy; a policy without a
        selector raises ``ConfigurationError`` at construction.
        With ``config.reflection_minibatch_size`` smaller than the trainset,
        a proposal runs first on a fresh seeded sample of the trainset rows
        its parent has scores for, and goes on only when its mean there
        beats the parent's cached scores on the same rows; otherwise its
        iteration is recorded with ``skip_reason="minibatch_rejected"``.
        When the valset is the trainset, a passing proposal is evaluated on
        the full trainset and that batch is reused for scoring. When the
        valset is separate, no full trainset pass runs: the minibatch batch
        is the proposal's reflection batch, the valset is scored, and on
        acceptance that batch and its trainset rows become the candidate's
        cached batch, so ``trainset_score`` is the mean over those rows.
        With ``config.checkpoint_path`` set, state is checkpointed after the
        baseline and after every recorded iteration, and ``config.resume``
        continues a run from that file (see ``_resume_from_checkpoint``).
    """

    def __init__(  # noqa: PLR0913, PLR0915  # public constructor; decomposition tracked in #507
        self,
        adapter: AsyncGEPAAdapter[DataInst, Trajectory, RolloutOutput],
        config: EvolutionConfig,
        initial_candidate: Candidate,
        batch: list[DataInst],
        valset: list[DataInst] | None = None,
        candidate_selector: CandidateSelectorProtocol | None = None,
        component_selector: ComponentSelectorProtocol | None = None,
        evaluation_policy: EvaluationPolicyProtocol | None = None,
        merge_proposer: ProposerProtocol | None = None,
        rng: random.Random | None = None,
    ) -> None:
        """Initialize the evolution engine.

        Args:
            adapter (AsyncGEPAAdapter[DataInst, Trajectory, RolloutOutput]):
                Implementation of AsyncGEPAAdapter protocol for evaluation and proposal
                generation.
            config (EvolutionConfig): Evolution parameters controlling iterations, thresholds,
                and early stopping.
            initial_candidate (Candidate): Starting candidate with at least one component.
            batch (list[DataInst]): Trainset data instances for reflection and mutation.
            valset (list[DataInst] | None): Optional validation data for scoring
                candidates. Defaults to trainset when omitted. Must be non-empty if
                provided.
            candidate_selector (CandidateSelectorProtocol | None): Optional selector
                strategy for Pareto-aware candidate sampling. When provided, initializes
                ParetoState for multi-objective tracking.
            component_selector (ComponentSelectorProtocol | None): Optional selector
                strategy for choosing which components to update. Defaults to
                RoundRobinComponentSelector.
            evaluation_policy (EvaluationPolicyProtocol | None): Optional policy for
                selecting which validation examples to evaluate per iteration. Defaults
                to FullEvaluationPolicy. Requires ``candidate_selector``, because the
                policy acts only through the Pareto state the selector creates.
            merge_proposer (ProposerProtocol | None): Optional proposer for merge
                operations. If provided and config.use_merge is True, merge proposals
                will be attempted after successful mutations.
            rng (random.Random | None): Optional seeded random.Random instance for
                deterministic engine decisions. When provided, used for the auto-created
                merge proposer and for drawing reflection minibatch rows. None preserves
                current random behavior, and minibatch rows are then drawn from a
                ``random.Random(config.seed)``.

        Raises:
            ValueError: If batch is empty, valset is provided but empty,
                or initial_candidate has no components.
            ConfigurationError: If evaluation_policy is provided without a
                candidate_selector, including an explicit FullEvaluationPolicy,
                or if ``config.checkpoint_path`` is set together with a
                candidate_selector (Pareto state is not checkpointed).

        Examples:
            Creating an engine:

            ```python
            engine = AsyncGEPAEngine(
                adapter=my_adapter,
                config=EvolutionConfig(max_iterations=50),
                initial_candidate=Candidate(components={"instruction": "Be helpful"}),
                batch=training_data,
                candidate_selector=selector,
            )
            ```

        Notes:
            Configures trainset and valset routing for reflection and scoring.
            A valset that is the same object as the batch (or omitted) marks
            the engine to reuse reflection batches for scoring.
            Initializes stopper lifecycle tracking for custom stop callbacks.
            Initializes the evaluation and pending failure counters, the
            pending and run token rollups, the pending and run reflection
            rollups (None until the proposer reports one), the last proposer
            rollup consumed (so an unchanged one is not recounted), the map
            from each scored
            candidate's id to its acceptance score, the random source for
            reflection minibatch rows, the slots holding the last mutation
            parent's trainset batch and the trainset rows it covers, the
            slots for the batch and rows the last minibatch gate evaluated,
            and the per-candidate map of cached batch rows (None for the
            full trainset).
            Initializes an in-memory streak of identical consecutive
            reflection errors (the cause's type name and text, the count, and
            the history index of the latest one), which ``run()`` resets and
            ``_record_reflection_error`` uses to enforce
            ``config.max_consecutive_reflection_errors``.
            Valid selector and policy combinations: neither is full
            evaluation; a selector alone is full evaluation over the Pareto
            state; a selector with a policy uses that policy; a policy alone
            is rejected. A checkpoint path with a candidate selector is
            rejected too, and the restored-state flag starts False.
        """
        # Validation
        if len(batch) == 0:
            raise ValueError("batch must contain at least one data instance")
        if valset is not None and len(valset) == 0:
            raise ValueError(
                "valset must contain at least one validation data instance"
            )

        if not initial_candidate.components:
            raise ValueError("initial_candidate must have at least one component")

        if evaluation_policy is not None and candidate_selector is None:
            raise ConfigurationError(
                "evaluation_policy requires a candidate_selector: a policy "
                "takes effect only through the Pareto state a candidate "
                "selector creates",
                field="evaluation_policy",
                value=type(evaluation_policy).__name__,
                constraint="candidate_selector is not None",
            )

        if config.checkpoint_path is not None and candidate_selector is not None:
            raise ConfigurationError(
                "checkpoint_path cannot be combined with a candidate_selector: "
                "Pareto state is not checkpointed yet",
                field="candidate_selector",
                value=type(candidate_selector).__name__,
                constraint="candidate_selector is None when checkpoint_path is set",
            )

        # Store dependencies
        self.adapter = adapter
        self.config = config
        self._initial_candidate = initial_candidate
        self._trainset = batch
        self._valset = valset if valset is not None else batch
        # Identity, not equality: an equal-but-separate valset is scored apart.
        self._valset_is_trainset = self._valset is self._trainset
        self._state: _EngineState | None = None
        self._rng = rng
        # Draws each iteration's reflection minibatch rows
        self._minibatch_rng = rng if rng is not None else random.Random(config.seed)  # noqa: S311  # evolution sampling, not security
        # Trainset batch of the parent the last mutation reflected on, and
        # the trainset rows it covers (None: the full trainset in order)
        self._mutation_parent_batch: EvaluationBatch | None = None
        self._mutation_parent_rows: list[int] | None = None
        # The batch and rows the last minibatch gate evaluated the proposal on
        self._gate_batch: EvaluationBatch | None = None
        self._gate_rows: list[int] | None = None
        self._candidate_selector = candidate_selector
        self._component_selector = component_selector or RoundRobinComponentSelector()
        self._pareto_state: ParetoState | None = None
        self._candidate_eval_batches: dict[int, EvaluationBatch] = {}
        # Trainset rows each cached batch covers (None: the full trainset)
        self._candidate_eval_rows: dict[int, list[int] | None] = {}
        # Acceptance score of every candidate scored this run, keyed by id
        self._scored: dict[str, float] = {}
        if merge_proposer is not None:
            self._merge_proposer = merge_proposer
        elif config.use_merge:
            self._merge_proposer = MergeProposer(rng=rng or random.Random())  # noqa: S311  # evolution sampling, not security
        else:
            self._merge_proposer = None
        self._merges_due: int = 0
        self._merge_invocations: int = 0
        # Stopper state tracking (T001, T002)
        self._start_time: float | None = None
        self._total_evaluations: int = 0
        # Failed rows evaluated since the last record (see _count_batch)
        self._pending_failed_evaluations: int = 0
        # Token usage since the last record and over the run (see _count_batch)
        self._pending_token_usage: TokenRollup = _ZERO_TOKENS
        self._run_token_usage: TokenRollup = _ZERO_TOKENS
        # Reflection usage of this iteration and of the run; None = not observed
        self._pending_reflection_usage: TokenRollup | None = None
        self._run_reflection_usage: TokenRollup | None = None
        # Last proposer rollup consumed, so an unchanged one is not recounted
        self._consumed_proposer_usage: TokenRollup | None = None
        self._active_stoppers: list[object] = []
        # Run of identical consecutive reflection errors (in memory only):
        # the cause's (type name, text), how many in a row, and the history
        # index of the latest one
        self._reflection_error_signature: tuple[str, str] | None = None
        self._reflection_error_streak: int = 0
        self._reflection_error_index: int = -1
        # True once run() restored state from a checkpoint (skips the baseline)
        self._restored: bool = False
        if evaluation_policy is None:
            self._evaluation_policy: EvaluationPolicyProtocol = FullEvaluationPolicy()
        else:
            self._evaluation_policy = evaluation_policy

    def _take_pending_failed_evaluations(self) -> int:
        """Return the pending failure count and reset it to zero.

        Returns:
            Failed rows counted by ``_count_batch`` since the last call.
        """
        count = self._pending_failed_evaluations
        self._pending_failed_evaluations = 0
        return count

    def _take_pending_token_usage(self) -> TokenRollup:
        """Return the pending token rollup and reset it to zero.

        Returns:
            Token usage folded in by ``_count_batch`` since the last call.
        """
        usage = self._pending_token_usage
        self._pending_token_usage = _ZERO_TOKENS
        return usage

    def _take_proposer_usage(self) -> None:
        """Store the reflection usage the adapter's proposer reported.

        Reads ``last_token_usage`` from the adapter's ``_proposer`` into
        ``_pending_reflection_usage``, so the next record carries it. An
        adapter without a proposer, or a proposer without the attribute,
        leaves it None (not observed). The proposer builds a new rollup
        object on every ``propose()`` and every reflection call, so a rollup
        that is the same object (``is``) as the last one consumed means
        ``propose()`` did not run this iteration; it is treated as not
        observed, and the previous iteration's reflection is not counted
        again. A rollup with no counted and no unknown row means
        ``propose()`` ran but made no reflection call (an empty reflective
        dataset), which is not observed reflection either.
        """
        usage = getattr(
            getattr(self.adapter, "_proposer", None), "last_token_usage", None
        )
        if not isinstance(usage, TokenRollup) or usage is self._consumed_proposer_usage:
            self._pending_reflection_usage = None
            return
        if usage.rows_counted == 0 and usage.rows_unknown == 0:
            self._consumed_proposer_usage = usage
            self._pending_reflection_usage = None
            return
        self._consumed_proposer_usage = usage
        self._pending_reflection_usage = usage

    def _take_record_token_usage(self) -> TokenRollup:
        """Build the next record's token rollup and reset the pending usage.

        Returns:
            A split rollup whose ``evaluation`` is the pending evaluation
            rollup and whose ``reflection`` is the pending reflection usage,
            or None when none was observed. The reflection usage is also
            folded into the run's reflection rollup.
        """
        evaluation = self._take_pending_token_usage()
        reflection = self._pending_reflection_usage
        self._pending_reflection_usage = None
        if reflection is not None:
            self._run_reflection_usage = (
                reflection
                if self._run_reflection_usage is None
                else self._run_reflection_usage.combine(reflection)
            )
        return TokenRollup.split(evaluation=evaluation, reflection=reflection)

    def _run_token_rollup(self) -> TokenRollup:
        """Return the run's token rollup with its evaluation and reflection splits.

        Returns:
            ``TokenRollup.split`` of the run's evaluation rollup and its
            reflection rollup, which is None when no reflection usage was
            observed.
        """
        return TokenRollup.split(
            evaluation=self._run_token_usage, reflection=self._run_reflection_usage
        )

    def _aggregate_acceptance_score(self, scores: list[float]) -> float:
        """Aggregate scores for acceptance decisions based on acceptance_metric.

        Args:
            scores: List of per-example scores from evaluation batch.

        Returns:
            Aggregated acceptance score (sum or mean based on config).

        Raises:
            InvalidScoreListError: If scores list is empty or contains
                non-finite values.

        Notes:
            Sums or averages acceptance scores after validating they are
            non-empty and finite. Uses sum or mean based on config.acceptance_metric.
        """
        # Validate scores are non-empty
        if not scores:
            raise InvalidScoreListError(
                "Cannot aggregate acceptance score from empty score list",
                scores=scores,
                reason="empty",
            )

        # Validate scores are finite
        if not all(math.isfinite(score) for score in scores):
            raise InvalidScoreListError(
                "Cannot aggregate acceptance score from non-finite values (NaN/inf)",
                scores=scores,
                reason="non-finite",
            )

        # Aggregate based on acceptance_metric
        if self.config.acceptance_metric == "sum":
            return sum(scores)
        else:  # acceptance_metric == "mean"
            return sum(scores) / len(scores)

    def _setup_stoppers(self) -> list[object]:
        """Call setup() on stoppers that have lifecycle methods.

        Returns:
            List of stoppers that had setup() called (for cleanup in reverse order).

        Notes:
            Only invokes setup() on stoppers implementing the lifecycle method.
            Stoppers that fail setup() are excluded from stop_callbacks for the
            remainder of execution to prevent inconsistent state.
        """
        setup_stoppers: list[object] = []
        active_stoppers: list[object] = []
        stop_callbacks = self.config.stop_callbacks
        if stop_callbacks:
            for stopper in stop_callbacks:
                setup_method = getattr(stopper, "setup", None)
                if setup_method is not None and callable(setup_method):
                    try:
                        setup_method()
                        setup_stoppers.append(stopper)
                        active_stoppers.append(stopper)
                    except Exception:
                        logger.exception(
                            "stopper.setup_error",
                            stopper=type(stopper).__name__,
                        )
                        # Stopper excluded from active list due to setup failure
                else:
                    # Stopper has no setup() method, still active
                    active_stoppers.append(stopper)
        # Store active stoppers for _should_stop() to use
        self._active_stoppers = active_stoppers
        return setup_stoppers

    def _cleanup_stoppers(self, setup_stoppers: list[object]) -> None:
        """Call cleanup() on stoppers in reverse order of setup.

        Args:
            setup_stoppers: List of stoppers that had setup() called.

        Notes:
            Follows the reverse-order cleanup contract (T025).
            When cleanup() raises, logs the error and continues (T026).
        """
        for stopper in reversed(setup_stoppers):
            cleanup_method = getattr(stopper, "cleanup", None)
            if cleanup_method is not None and callable(cleanup_method):
                try:
                    cleanup_method()
                except Exception:
                    logger.exception(
                        "stopper.cleanup_error",
                        stopper=type(stopper).__name__,
                    )

    def _build_stopper_state(self) -> StopperState:
        """Build a StopperState snapshot from current engine state.

        Constructs an immutable snapshot of evolution state for stopper
        callbacks to evaluate. Captures all metrics needed by stoppers
        including elapsed time, total evaluations and the latest score.

        Returns:
            Frozen StopperState containing current iteration, best score,
            stagnation counter, total evaluations, candidates count, elapsed
            time, and the latest iteration's score.

        Notes:
            Obtains elapsed_seconds from monotonic time since run() started.
            Uses zero if _start_time has not yet been set. Takes latest_score
            from the last ``iteration_history`` record, accepted or not, and
            leaves it ``None`` while the history is empty. Must run after
            run() created the engine state, which the method asserts.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        start, history = self._start_time, self._state.iteration_history
        elapsed = time.monotonic() - start if start is not None else 0.0
        candidates_count = (
            len(self._pareto_state.candidates) if self._pareto_state is not None else 0
        )
        return StopperState(
            iteration=self._state.iteration,
            best_score=self._state.best_score,
            stagnation_counter=self._state.stagnation_counter,
            total_evaluations=self._total_evaluations,
            candidates_count=candidates_count,
            elapsed_seconds=elapsed,
            latest_score=history[-1].score if history else None,
        )

    @property
    def pareto_state(self) -> ParetoState | None:
        """Return the current Pareto state, if initialized."""
        return self._pareto_state

    def _build_component_list(self, candidate: Candidate) -> list[str]:
        """Build list of available component keys from candidate.

        Excludes generic 'instruction' alias if agent-specific keys exist
        (e.g., 'agent1_instruction').

        Args:
            candidate: Candidate to extract component keys from.

        Returns:
            List of component keys to consider for update.

        Notes:
            Selects component keys, filtering out the default component name when
            more specific per-agent component keys are present.
        """
        keys = list(candidate.components.keys())
        if len(keys) > 1 and DEFAULT_COMPONENT_NAME in keys:
            # If multiple keys exist, assume default component might be an alias/proxy
            # or simply one of many.
            # For now, simplistic rule: if other keys exist, exclude default.
            return [k for k in keys if k != DEFAULT_COMPONENT_NAME]
        return keys

    def _count_batch(
        self, candidate: Candidate, batch: EvaluationBatch, phase: str
    ) -> None:
        """Count one adapter evaluation and log it under the candidate's id.

        Adds the batch's rows to the evaluation counter that stoppers read
        and its ``failed_indices`` to the failures pending for the next
        record (the baseline or the current iteration), folds the batch's
        ``TokenRollup`` into the pending and the run token rollups, then logs
        ``evaluation.completed`` with the candidate's id, the phase, the row
        count and that same failure count.

        Args:
            candidate: Candidate whose components were evaluated.
            batch: Evaluation batch the adapter just returned.
            phase: ``"reflection"`` for a trainset evaluation with traces,
                ``"scoring"`` for a valset evaluation.

        Notes:
            Called once per adapter ``evaluate()`` call, so a reused batch
            is neither counted nor logged, because no row was evaluated
            again. An adapter that leaves ``failed_indices`` as None reports
            zero failures.
        """
        failed = len(batch.failed_indices or [])
        self._total_evaluations += len(batch.scores)
        self._pending_failed_evaluations += failed
        usage = TokenRollup.from_batch(batch)
        self._pending_token_usage = self._pending_token_usage.combine(usage)
        self._run_token_usage = self._run_token_usage.combine(usage)
        logger.debug(
            "evaluation.completed",
            candidate_id=candidate.id,
            phase=phase,
            n=len(batch.scores),
            failed=failed,
            iteration=self._state.iteration if self._state is not None else 0,
        )

    async def _initialize_baseline(self) -> None:
        """Initialize baseline evaluation.

        Evaluates the initial candidate on trainset for reflection and
        on valset for scoring. Caches the reflection batch for use in
        the first mutation proposal. When the valset is the trainset, the
        reflection batch also serves scoring, so the baseline is evaluated
        once.

        Notes:
            Sets up both reflection and scoring baselines up front. With a
            candidate selector and objective scores, the baseline's Pareto
            entry gets per-objective means (``statistics.fmean``) for the
            OBJECTIVE, HYBRID and CARTESIAN frontiers, and per-example
            objective scores as well for CARTESIAN. The
            baseline evaluation is counted and logged via ``_count_batch``.
            Rows that failed during these evaluations are stored on the state
            as ``baseline_failed_evaluations``. The baseline's token usage
            stays in the run rollup and is cleared from the pending one, so
            the first iteration's record does not include it. The baseline's
            cached batch covers the full trainset, so its rows are None on
            the state and in the Pareto row map. Ends by writing
            a checkpoint through ``_write_checkpoint`` when
            ``config.checkpoint_path`` is set.
        """
        # Create pareto_state before evaluation if candidate_selector exists
        # so that _evaluate_scoring can use evaluation_policy
        if self._candidate_selector is not None:
            self._pareto_state = ParetoState(frontier_type=self.config.frontier_type)

        reflection_batch = await self.adapter.evaluate(
            self._trainset,
            self._initial_candidate.components,
            capture_traces=True,
        )
        self._count_batch(self._initial_candidate, reflection_batch, "reflection")
        # Use _evaluate_scoring for baseline to get eval_indices
        (
            baseline_score,
            scoring_batch,
            baseline_eval_indices,
        ) = await self._evaluate_scoring(
            self._initial_candidate, reflection_batch=reflection_batch
        )
        baseline_reflection_score = sum(reflection_batch.scores) / len(
            reflection_batch.scores
        )
        baseline_valset_mean = (
            sum(scoring_batch.scores) / len(scoring_batch.scores)
            if scoring_batch.scores
            else 0.0
        )
        self._state = _EngineState(
            best_candidate=self._initial_candidate,
            best_score=baseline_score,
            original_score=baseline_score,
            iteration=0,
            stagnation_counter=0,
            iteration_history=[],
            last_eval_batch=reflection_batch,
            best_reflection_score=baseline_reflection_score,
            best_valset_mean=baseline_valset_mean,
            best_objective_scores=scoring_batch.objective_scores,
            baseline_failed_evaluations=self._take_pending_failed_evaluations(),
        )
        self._take_pending_token_usage()
        if self._candidate_selector is not None:
            # Prepare objective scores for baseline if needed
            objective_scores: dict[str, float] | None = None
            per_example_objective_scores: dict[int, dict[str, float]] | None = None

            if scoring_batch.objective_scores is not None:
                if self.config.frontier_type in (
                    FrontierType.OBJECTIVE,
                    FrontierType.HYBRID,
                ):
                    objective_scores_by_name: dict[str, list[float]] = {}
                    for obj_scores in scoring_batch.objective_scores:
                        for obj_name, obj_score in obj_scores.items():
                            objective_scores_by_name.setdefault(obj_name, []).append(
                                obj_score
                            )
                    objective_scores = {
                        obj_name: fmean(scores)
                        for obj_name, scores in objective_scores_by_name.items()
                    }

                if self.config.frontier_type == FrontierType.CARTESIAN:
                    per_example_objective_scores = {
                        baseline_eval_indices[i]: scoring_batch.objective_scores[i]
                        for i in range(len(baseline_eval_indices))
                    }
                    objective_scores_by_name: dict[str, list[float]] = {}
                    for obj_scores in scoring_batch.objective_scores:
                        for obj_name, obj_score in obj_scores.items():
                            objective_scores_by_name.setdefault(obj_name, []).append(
                                obj_score
                            )
                    objective_scores = {
                        obj_name: fmean(scores)
                        for obj_name, scores in objective_scores_by_name.items()
                    }

            assert self._pareto_state is not None, "Pareto state not initialized"  # noqa: S101  # narrows state that run() sets before this call
            candidate_idx = self._pareto_state.add_candidate(
                self._initial_candidate,
                scoring_batch.scores,
                score_indices=baseline_eval_indices,
                objective_scores=objective_scores,
                per_example_objective_scores=per_example_objective_scores,
                logger=logger,
            )
            self._candidate_eval_batches[candidate_idx] = reflection_batch
            self._candidate_eval_rows[candidate_idx] = None
        self._write_checkpoint()

    async def _evaluate_reflection(
        self, candidate: Candidate
    ) -> tuple[float, EvaluationBatch]:
        """Evaluate a candidate on the trainset for reflection.

        Args:
            candidate: Candidate to evaluate.

        Returns:
            Tuple of (mean score across trainset examples, evaluation batch).

        Notes:
            Supplies trajectories for reflective dataset construction.
            The evaluation is counted and logged via ``_count_batch``.
        """
        eval_batch = await self.adapter.evaluate(
            self._trainset,
            candidate.components,
            capture_traces=True,
        )
        self._count_batch(candidate, eval_batch, "reflection")
        score = sum(eval_batch.scores) / len(eval_batch.scores)
        return score, eval_batch

    async def _evaluate_scoring(
        self,
        candidate: Candidate,
        reflection_batch: EvaluationBatch | None = None,
    ) -> tuple[float, EvaluationBatch, list[int]]:
        """Evaluate a candidate on the valset for scoring decisions.

        Args:
            candidate: Candidate to evaluate on the validation set.
            reflection_batch: The candidate's trainset reflection batch. When
                the valset is the trainset, scoring reuses this batch (or its
                entries at the selected indices) instead of calling the
                adapter again.

        Returns:
            Tuple of (aggregated acceptance score, evaluation batch, eval_indices).
            Score is aggregated using acceptance_metric (sum or mean).
            eval_indices are the valset indices that were actually evaluated.

        Notes:
            Supplies scores without traces for acceptance decisions.
            Aggregation method (sum/mean) is determined by config.acceptance_metric.
            Uses evaluation_policy to determine which examples to evaluate.
            Only the canonical ordered index list counts as a full evaluation;
            any other selection, including a permutation, is built row by row
            in the policy's order.
            A reused batch adds nothing to the evaluation or failure
            counters, because no example is evaluated again; its reuse log
            names the ``candidate_id``. The acceptance score is stored under
            the candidate's id so a later duplicate proposal is not
            evaluated again.
        """
        # Get indices to evaluate from evaluation policy
        valset_ids = list(range(len(self._valset)))
        if self._pareto_state is not None:
            eval_indices = self._evaluation_policy.get_eval_batch(
                valset_ids, self._pareto_state
            )
        else:
            # Fallback to all indices if no pareto state yet
            eval_indices = valset_ids

        # Filter valset to only include selected indices
        # Only the canonical ordered list counts as a full evaluation, so a
        # policy that permutes the indices gets a batch built in its order.
        is_full_eval = list(eval_indices) == valset_ids

        if reflection_batch is not None and self._valset_is_trainset:
            logger.debug(
                "evaluation.reuse_trainset_batch",
                candidate_id=candidate.id,
                components=sorted(candidate.components),
                reason="valset_is_trainset",
                full=is_full_eval,
                n=len(eval_indices),
            )
            eval_batch = (
                reflection_batch
                if is_full_eval
                else _select_batch_rows(reflection_batch, eval_indices)
            )
        else:
            eval_valset = (
                self._valset
                if is_full_eval
                else [self._valset[i] for i in eval_indices]
            )
            eval_batch = await self.adapter.evaluate(
                eval_valset,
                candidate.components,
                capture_traces=False,
            )
            self._count_batch(candidate, eval_batch, "scoring")
        score = self._aggregate_acceptance_score(eval_batch.scores)
        self._scored[candidate.id] = score
        return score, eval_batch, eval_indices

    async def _propose_mutation(
        self,
    ) -> tuple[Candidate, list[str], int | None]:
        """Propose a new candidate via reflective mutation.

        Uses the cached evaluation batch from the most recent best candidate
        evaluation to generate the reflective dataset, avoiding redundant
        adapter calls.

        Returns:
            Tuple of (new candidate with proposed component updates,
            list of component names that were updated, Pareto index of the
            parent the reflector rewrote). The index is the selector's draw;
            without a selector (or when the selector found no candidate) it
            is the index of the current best candidate in the Pareto state,
            searched from the end, and None when there is no Pareto state or
            the parent is absent from it.

        Raises:
            ConfigurationError: If the candidate selector returns an index
                with no cached reflection batch, such as a negative index.
                An empty frontier (``NoCandidateAvailableError``) is not an
                error: the most recent cached batch is used instead.

        Notes:
            Spawns a new candidate with updated components based on reflective
            dataset analysis and component selector strategy. The selected
            parent is logged with its ``candidate_id``. Every candidate's
            reflection batch is cached when it is added to the Pareto state,
            so the selected parent is never re-evaluated. The parent's
            trainset batch it reflected on, and the trainset rows that batch
            covers (None for the full trainset), are kept on the engine as
            ``_mutation_parent_batch`` and ``_mutation_parent_rows`` for the
            reflection minibatch gate. The reflective dataset is built from
            whatever batch the parent has, which is its sampled minibatch
            rows when it was accepted with a separate valset. The parent is drawn
            once per iteration: the proposal carries ``parent_id`` set to the
            parent's ``Candidate.id`` and ``generation`` set to the parent's
            plus one, and the returned index is the one the loop passes to
            ``ParetoState.add_candidate`` as ``parent_indices``.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        assert self._state.last_eval_batch is not None, "No eval batch cached"  # noqa: S101  # narrows state that run() sets before this call

        selected_candidate = self._state.best_candidate
        selected_idx: int | None = None
        eval_batch = self._state.last_eval_batch
        eval_rows = self._state.last_eval_rows

        if self._candidate_selector is not None and self._pareto_state is not None:
            try:
                selected_idx = await self._candidate_selector.select_candidate(
                    self._pareto_state
                )
                selected_candidate = self._pareto_state.candidates[selected_idx]
                eval_batch = self._candidate_eval_batches.get(selected_idx)
                eval_rows = self._candidate_eval_rows.get(selected_idx)
                logger.info(
                    "pareto_selection.mutation_parent_selected",
                    candidate_idx=selected_idx,
                    candidate_id=selected_candidate.id,
                    iteration=self._state.iteration,
                    selector_type=type(self._candidate_selector).__name__,
                )
            except NoCandidateAvailableError as exc:
                logger.info(
                    "pareto_selection.empty_frontier_fallback",
                    iteration=self._state.iteration,
                    selector_type=type(self._candidate_selector).__name__,
                    error=str(exc),
                )
                eval_batch = self._state.last_eval_batch
                eval_rows = self._state.last_eval_rows

        if eval_batch is None:
            # Every added candidate is cached at once; a miss is a bad index
            raise ConfigurationError(
                f"candidate selector returned index {selected_idx}, which has "
                "no cached reflection batch",
                field="candidate_selector",
                value=selected_idx,
                constraint="0 <= index < len(pareto_state.candidates)",
            )
        # The minibatch gate compares the proposal with this parent batch
        self._mutation_parent_batch = eval_batch
        self._mutation_parent_rows = eval_rows

        # Build component list
        available_components = self._build_component_list(selected_candidate)

        # Select components to update
        components_to_update = await self._component_selector.select_components(
            components=available_components,
            iteration=self._state.iteration,
            candidate_idx=selected_idx if selected_idx is not None else 0,
        )

        logger.info(
            "mutation.components_selected",
            iteration=self._state.iteration,
            components=components_to_update,
            selector=type(self._component_selector).__name__,
        )

        # Build reflective dataset
        reflective_dataset = await self.adapter.make_reflective_dataset(
            selected_candidate.components,
            eval_batch,
            components_to_update,
        )

        # Propose new texts
        proposed_components = await self.adapter.propose_new_texts(
            selected_candidate.components,
            reflective_dataset,
            components_to_update,
        )

        # The parent's Pareto index, for the genealogy edge
        parent_idx = selected_idx
        if parent_idx is None and self._pareto_state is not None:
            parent_idx = next(
                (
                    idx
                    for idx in range(len(self._pareto_state.candidates) - 1, -1, -1)
                    if self._pareto_state.candidates[idx].id == selected_candidate.id
                ),
                None,
            )

        # Create new candidate with proposed components and its lineage
        new_components = dict(selected_candidate.components)
        new_components.update(proposed_components)
        return (
            Candidate(
                components=new_components,
                generation=selected_candidate.generation + 1,
                parent_id=selected_candidate.id,
            ),
            components_to_update,
            parent_idx,
        )

    async def _record_iteration(  # noqa: PLR0913  # one argument per recorded field; private helper
        self,
        score: float,
        component_text: str,
        evolved_component: str,
        accepted: bool,
        objective_scores: list[dict[str, float]] | None = None,
        reflection_reasoning: str | None = None,
        skip_reason: str | None = None,
        candidate_id: str | None = None,
        parent_ids: list[str] | None = None,
        rejection_reason: str | None = None,
        *,
        rows_evaluated: int | None = None,
    ) -> None:
        """Record iteration outcome and notify the ``on_iteration`` callback.

        Args:
            score (float): Score achieved in this iteration.
            component_text (str): The text of the component that was evaluated.
            evolved_component (str): The name of the component that was evolved
                (e.g., "instruction", "output_schema").
            accepted (bool): Whether proposal was accepted.
            objective_scores (list[dict[str, float]] | None): Optional objective scores
                from this iteration's evaluation. None when adapter does not provide
                objective scores.
            reflection_reasoning (str | None): Optional natural language reasoning from
                the reflection agent explaining the mutation. None when
                reasoning is not available.
            skip_reason (str | None): Why the iteration produced no evaluated proposal
                (e.g., ``"empty_proposal"``). None for ordinary iterations.
            candidate_id (str | None): Id of the candidate the record concerns, stored on
                the record and passed to the callback. None when nothing was
                proposed.
            parent_ids (list[str] | None): Ids of the candidates the proposal was made from,
                stored on the record. None when nothing was proposed.
            rejection_reason (str | None): The reason ``config.proposal_validator``
                returned for a ``"proposal_rejected"`` skip, stored on the
                record. None for every other iteration.
            rows_evaluated (int | None): Number of rows ``score`` aggregates over, stored
                on the record: the scoring batch's row count on an evaluated
                record, the minibatch's on a ``"minibatch_rejected"`` skip,
                0 on a skip that evaluated nothing and None on a
                ``"duplicate"`` skip, whose score is not this iteration's.
                Defaults to None (not recorded); every engine path passes it.

        Notes:
            Appends an IterationRecord to ``state.iteration_history`` so the
            chronological evolution trace is preserved for analysis. The
            record's ``failed_evaluations`` takes the failures counted since
            the iteration began, including any merge evaluation, and its
            ``token_usage`` the token rollup of the same evaluations (zeros
            when the iteration evaluated nothing), split into ``evaluation``
            and ``reflection``; ``reflection`` is None unless the proposer
            reported the iteration's reflection usage. Every record
            path goes through here, so ``config.on_iteration``, when set, is
            called with the record and ``candidate_id`` right after the
            append; an awaitable return value is awaited. Exceptions from the
            callback are not caught. Ends by writing a checkpoint through
            ``_write_checkpoint`` when ``config.checkpoint_path`` is set.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        record = IterationRecord(
            iteration_number=self._state.iteration,
            score=score,
            component_text=component_text,
            evolved_component=evolved_component,
            accepted=accepted,
            objective_scores=objective_scores,
            reflection_reasoning=reflection_reasoning,
            skip_reason=skip_reason,
            failed_evaluations=self._take_pending_failed_evaluations(),
            token_usage=self._take_record_token_usage(),
            candidate_id=candidate_id,
            parent_ids=parent_ids,
            rejection_reason=rejection_reason,
            rows_evaluated=rows_evaluated,
        )
        self._state.iteration_history.append(record)
        callback = self.config.on_iteration
        if callback is not None:
            outcome = callback(record, candidate_id)
            if inspect.isawaitable(outcome):
                await outcome
        self._write_checkpoint()

    async def _record_empty_proposal(self, error: EmptyProposalError) -> None:
        """Record an iteration whose reflection returned an empty proposal.

        Args:
            error (EmptyProposalError): The error raised while proposing; its ``component`` names
                the component the reflection was working on.

        Notes:
            Logs ``evolution.proposal_skipped`` with
            ``reason="empty_proposal"``, counts the iteration toward
            stagnation and appends a not-accepted IterationRecord with
            ``score=0.0``, empty ``component_text`` and
            ``skip_reason="empty_proposal"``. Nothing is evaluated, so the
            record's ``rows_evaluated`` is 0. The
            ``on_iteration`` callback receives ``None`` as the candidate id.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        logger.debug(
            "evolution.proposal_skipped",
            iteration=self._state.iteration,
            reason="empty_proposal",
            component=error.component,
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=0.0,
            component_text="",
            evolved_component=error.component,
            accepted=False,
            skip_reason="empty_proposal",
            rows_evaluated=0,
        )

    async def _record_reflection_timeout(self, error: ReflectionTimeoutError) -> None:
        """Record an iteration whose reflection agent timed out.

        Args:
            error (ReflectionTimeoutError): The error raised while proposing; its
                ``component`` names the component the reflection was working on and its
                ``timeout_seconds`` the timeout it ran under.

        Notes:
            Logs ``evolution.proposal_skipped`` with
            ``reason="reflection_timeout"`` and ``timeout_seconds``, counts
            the iteration toward stagnation and appends a not-accepted
            IterationRecord with ``score=0.0``, empty ``component_text`` and
            ``skip_reason="reflection_timeout"``. Nothing is evaluated
            (``rows_evaluated=0``), and accepted candidates and the Pareto state are untouched.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        logger.debug(
            "evolution.proposal_skipped",
            iteration=self._state.iteration,
            reason="reflection_timeout",
            component=error.component,
            timeout_seconds=error.timeout_seconds,
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=0.0,
            component_text="",
            evolved_component=error.component,
            accepted=False,
            skip_reason="reflection_timeout",
            rows_evaluated=0,
        )

    async def _record_reflection_error(self, error: ReflectionError) -> None:
        """Record an iteration whose reflection function kept raising.

        Args:
            error (ReflectionError): The retryable error the proposer raised after its retry;
                its ``component`` names the component the reflection was
                working on, ``cause`` the provider exception and
                ``attempts`` the number of reflection calls made.

        Notes:
            Logs ``evolution.proposal_skipped`` at warning level with
            ``reason="reflection_error"``, the cause's ``error_type`` and
            text and ``attempts``, counts the iteration toward stagnation
            and appends a not-accepted IterationRecord with ``score=0.0``,
            empty ``component_text`` and ``skip_reason="reflection_error"``.
            Nothing is evaluated (``rows_evaluated=0``), and accepted candidates and the Pareto
            state are untouched.

            When the record just appended is the
            ``config.max_consecutive_reflection_errors``-th in a row whose
            cause has the same exception type name and text, raises a
            non-retryable ``ReflectionError``; ``run()`` attaches the
            partial result. Any other record in between, from a successful
            reflection or another skip, or a different cause, restarts the
            count.

        Raises:
            ReflectionError: With ``retryable=False``, the last cause and its
                attempts, once the identical errors reach the limit.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        error_type = type(error.cause).__name__
        logger.warning(
            "evolution.proposal_skipped",
            iteration=self._state.iteration,
            reason="reflection_error",
            component=error.component,
            error_type=error_type,
            error=str(error.cause),
            attempts=error.attempts,
        )
        self._state.stagnation_counter += 1
        signature = (error_type, str(error.cause))
        index = len(self._state.iteration_history)
        if (
            signature == self._reflection_error_signature
            and self._reflection_error_index == index - 1
        ):
            self._reflection_error_streak += 1
        else:
            self._reflection_error_signature = signature
            self._reflection_error_streak = 1
        self._reflection_error_index = index
        await self._record_iteration(
            score=0.0,
            component_text="",
            evolved_component=error.component,
            accepted=False,
            skip_reason="reflection_error",
            rows_evaluated=0,
        )
        limit = self.config.max_consecutive_reflection_errors
        if self._reflection_error_streak >= limit:
            logger.error(
                "evolution.reflection_errors_exhausted",
                component=error.component,
                error_type=error_type,
                error=str(error.cause),
                consecutive=self._reflection_error_streak,
            )
            raise ReflectionError(
                error.component,
                cause=error.cause,
                retryable=False,
                attempts=error.attempts,
            ) from error

    async def _record_incomplete_proposal(self, error: IncompleteProposalError) -> None:
        """Record an iteration whose reflection output was incomplete.

        Args:
            error (IncompleteProposalError): The error the reflection function raised; its
                ``component`` names the component the reflection was working
                on, ``finish_reason`` why the output is incomplete and
                ``raw_text`` the truncated text.

        Notes:
            Logs ``evolution.proposal_skipped`` at warning level with
            ``reason="incomplete_proposal"``, ``finish_reason`` and
            ``response_length``, counts the iteration toward stagnation and
            appends a not-accepted IterationRecord with ``score=0.0``, the
            truncated text as ``component_text`` and
            ``skip_reason="incomplete_proposal"``. Nothing is evaluated
            (``rows_evaluated=0``), and accepted candidates and the Pareto state are untouched.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        logger.warning(
            "evolution.proposal_skipped",
            iteration=self._state.iteration,
            reason="incomplete_proposal",
            component=error.component,
            finish_reason=error.finish_reason,
            response_length=error.output_length,
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=0.0,
            component_text=error.raw_text,
            evolved_component=error.component,
            accepted=False,
            skip_reason="incomplete_proposal",
            rows_evaluated=0,
        )

    async def _record_schema_validation_skip(
        self, proposal: Candidate, evolved_components: list[str]
    ) -> None:
        """Record an iteration whose proposed output schema failed validation.

        Args:
            proposal (Candidate): The proposal whose ``output_schema`` text is invalid.
            evolved_components (list[str]): Names of the components the proposal evolved,
                logged with the skip.

        Notes:
            Logs ``evolution.proposal_skipped`` with
            ``reason="schema_validation_failed"``, counts the iteration toward
            stagnation and appends a not-accepted IterationRecord with
            ``score=0.0``, the invalid schema text as ``component_text``,
            ``evolved_component="output_schema"`` and
            ``skip_reason="schema_validation_failed"``. Nothing is evaluated
            (``rows_evaluated=0``).
            The ``on_iteration`` callback receives the proposal's id, and the
            record carries it as ``candidate_id`` with ``parent_ids`` naming
            the proposal's parent.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        logger.debug(
            "evolution.proposal_skipped",
            iteration=self._state.iteration,
            reason="schema_validation_failed",
            candidate_id=proposal.id,
            components=evolved_components,
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=0.0,
            component_text=proposal.components["output_schema"],
            evolved_component="output_schema",
            accepted=False,
            skip_reason="schema_validation_failed",
            rows_evaluated=0,
            candidate_id=proposal.id,
            parent_ids=_parent_ids(proposal),
        )

    async def _record_if_rejected(
        self, proposal: Candidate, evolved_components: list[str]
    ) -> bool:
        """Record a skipped iteration when the proposal validator rejects it.

        Args:
            proposal (Candidate): The candidate proposed this iteration.
            evolved_components (list[str]): Names of the components evolved this
                iteration; each is passed to the validator in order. When
                empty, every component of the proposal is checked.

        Returns:
            True when ``config.proposal_validator`` returned a reason for a
            component and the iteration was recorded as rejected; False when
            no validator is set or it accepted every component.

        Notes:
            The validator is called with the component name and the proposed
            text; the first non-None return is the reason. A rejection logs
            ``evolution.proposal_skipped`` at warning level with
            ``reason="proposal_rejected"`` and the ``rejection_reason``,
            counts toward stagnation and appends a not-accepted
            IterationRecord with ``score=0.0``, the rejected text, the
            proposal's ``candidate_id`` and ``parent_ids`` and the reason.
            No adapter call is made, so the record's ``rows_evaluated`` is
            0. Exceptions from the validator are not caught and propagate
            out of ``run()``.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        validator = self.config.proposal_validator
        if validator is None:
            return False
        names = evolved_components or list(proposal.components)
        for name in names:
            text = proposal.components[name]
            reason = validator(name, text)
            if reason is None:
                continue
            logger.warning(
                "evolution.proposal_skipped",
                iteration=self._state.iteration,
                reason="proposal_rejected",
                rejection_reason=reason,
                component=name,
                candidate_id=proposal.id,
            )
            self._state.stagnation_counter += 1
            await self._record_iteration(
                score=0.0,
                component_text=text,
                evolved_component=name,
                accepted=False,
                skip_reason="proposal_rejected",
                rows_evaluated=0,
                candidate_id=proposal.id,
                parent_ids=_parent_ids(proposal),
                rejection_reason=reason,
            )
            return True
        return False

    async def _record_if_duplicate(
        self, proposal: Candidate, evolved_components: list[str]
    ) -> bool:
        """Record a skipped iteration when the proposal was already scored.

        Args:
            proposal (Candidate): The candidate proposed this iteration.
            evolved_components (list[str]): Names of the components evolved this
                iteration; the first one names the record's component.

        Returns:
            True when the proposal's id was already scored and the iteration
            was recorded as a duplicate; False when it must be evaluated.

        Notes:
            A duplicate logs ``proposal.duplicate`` with the stored
            acceptance score, counts toward stagnation and appends a
            not-accepted IterationRecord carrying that score, the proposal's
            text for the evolved component and ``skip_reason="duplicate"``.
            No adapter call is made. The record's ``rows_evaluated`` is None
            because its score comes from an earlier evaluation. The
            ``on_iteration`` callback receives the duplicate proposal's id,
            and the record carries it as
            ``candidate_id`` with ``parent_ids`` naming the proposal's parent.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        score = self._scored.get(proposal.id)
        if score is None:
            return False
        component = (
            evolved_components[0]
            if evolved_components
            else next(iter(proposal.components))
        )
        logger.info(
            "proposal.duplicate",
            iteration=self._state.iteration,
            candidate_id=proposal.id,
            score=score,
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=score,
            component_text=proposal.components.get(component, ""),
            evolved_component=component,
            accepted=False,
            skip_reason="duplicate",
            candidate_id=proposal.id,
            parent_ids=_parent_ids(proposal),
            rows_evaluated=None,
        )
        return True

    def _effective_minibatch_size(self) -> int | None:
        """Return the reflection minibatch size in effect for this run.

        Returns:
            ``min(reflection_minibatch_size, len(trainset))`` when that is
            smaller than the trainset, otherwise None, meaning every proposal
            is evaluated on the full trainset.
        """
        size = self.config.reflection_minibatch_size
        if size is None:
            return None
        k = min(size, len(self._trainset))
        return k if k < len(self._trainset) else None

    async def _gate_on_minibatch(
        self, proposal: Candidate, evolved_components: list[str]
    ) -> bool:
        """Decide whether a proposal earns its full evaluation.

        Samples trainset rows from the rows the mutation parent has scores
        for, evaluates the proposal on them with traces, and compares the
        mean of its scores with the mean of the parent's cached scores on the
        same rows.

        Args:
            proposal (Candidate): The candidate proposed this iteration.
            evolved_components (list[str]): Names of the components evolved this
                iteration; the first one names the record's component.

        Returns:
            True when the minibatch is disabled or the proposal's mean is
            strictly greater than the parent's; False when the proposal was
            rejected and its iteration recorded.

        Notes:
            The parent's known rows are ``_mutation_parent_rows``, or every
            trainset row when that is None. When the parent has at most
            ``k`` known rows, all of them are used (sorted) and the seeded
            random source is not drawn from; otherwise ``k`` of them are
            drawn with it and sorted. Each sampled trainset index is mapped
            to its position in the parent's batch, so the comparison is
            always on rows both sides scored. The evaluated batch and its
            trainset indices are stored as ``_gate_batch`` and
            ``_gate_rows`` (both None when the minibatch is disabled), so the
            loop can reuse them as the proposal's reflection batch.
            The minibatch evaluation is counted through ``_count_batch`` as a
            ``"reflection"`` evaluation. A rejection logs
            ``proposal.minibatch_rejected``, counts toward stagnation and
            appends a not-accepted IterationRecord whose score is the
            acceptance aggregate over the minibatch rows, whose
            ``rows_evaluated`` is the number of minibatch rows and whose
            ``skip_reason`` is ``"minibatch_rejected"``, with the proposal's
            ``candidate_id`` and its parent in ``parent_ids``. A rejected proposal
            is not stored as scored, so an identical later proposal draws a
            fresh sample. A pass logs ``proposal.minibatch_passed``. Both
            log events carry ``parent_rows``, the number of rows the parent
            has scores for. A full parent batch is sampled from ``range(n)``
            and read by trainset row, so no per-iteration row list or
            position map is built for it; only a sampled parent batch maps
            row to position.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        self._gate_batch = None
        self._gate_rows = None
        k = self._effective_minibatch_size()
        if k is None:
            return True
        parent_batch = self._mutation_parent_batch
        assert parent_batch is not None, "No parent batch cached"  # noqa: S101  # set by the mutation step that precedes the gate
        # A full batch is indexed by trainset row, so no position map or
        # row list is built for it; a sampled batch maps row to position.
        known: list[int] | range = (
            self._mutation_parent_rows
            if self._mutation_parent_rows is not None
            else range(len(self._trainset))
        )
        position = (
            {row: pos for pos, row in enumerate(known)}
            if self._mutation_parent_rows is not None
            else None
        )
        indices = (
            sorted(known)
            if len(known) <= k
            else sorted(self._minibatch_rng.sample(known, k))
        )
        minibatch = await self.adapter.evaluate(
            [self._trainset[i] for i in indices],
            proposal.components,
            capture_traces=True,
        )
        self._count_batch(proposal, minibatch, "reflection")
        self._gate_batch = minibatch
        self._gate_rows = indices
        proposal_mean = sum(minibatch.scores) / len(minibatch.scores)
        parent_mean = sum(
            parent_batch.scores[i if position is None else position[i]] for i in indices
        ) / len(indices)
        fields = {
            "iteration": self._state.iteration,
            "candidate_id": proposal.id,
            "indices": indices,
            "parent_rows": len(known),
            "proposal_mean": proposal_mean,
            "parent_mean": parent_mean,
        }
        if proposal_mean > parent_mean:
            logger.info("proposal.minibatch_passed", **fields)
            return True
        logger.info("proposal.minibatch_rejected", **fields)
        component = (
            evolved_components[0]
            if evolved_components
            else next(iter(proposal.components))
        )
        self._state.stagnation_counter += 1
        await self._record_iteration(
            score=self._aggregate_acceptance_score(minibatch.scores),
            component_text=proposal.components.get(component, ""),
            evolved_component=component,
            accepted=False,
            skip_reason="minibatch_rejected",
            candidate_id=proposal.id,
            parent_ids=_parent_ids(proposal),
            rows_evaluated=len(minibatch.scores),
        )
        return False

    async def _evaluate_passed_proposal(
        self, proposal: Candidate
    ) -> tuple[float, EvaluationBatch, list[int] | None]:
        """Build the reflection batch of a proposal that passed the gate.

        Args:
            proposal: The candidate that passed the minibatch gate (or ran
                with the minibatch disabled).

        Returns:
            Tuple of (mean score over the batch, the batch, the trainset
            indices its rows cover or None for the full trainset in order).

        Notes:
            With a reflection minibatch in effect and a valset separate from
            the trainset, the gate's batch (``_gate_batch``, evaluated with
            traces) is the reflection batch and no full trainset pass runs;
            its rows are ``_gate_rows``. Otherwise the proposal is evaluated
            on the full trainset through ``_evaluate_reflection``, so a
            valset that is the trainset can reuse that batch for scoring.
        """
        if (
            self._effective_minibatch_size() is not None
            and not self._valset_is_trainset
            and self._gate_batch is not None
        ):
            batch = self._gate_batch
            return sum(batch.scores) / len(batch.scores), batch, self._gate_rows
        score, batch = await self._evaluate_reflection(proposal)
        return score, batch, None

    def _start_merge(self, merge_result: ProposalResult) -> bool:
        """Count a merge attempt and decide whether to evaluate it.

        Args:
            merge_result (ProposalResult): The merge proposer's result for this iteration.

        Returns:
            True when the merge candidate should be evaluated; False when it
            is skipped for an invalid schema or an already scored id.

        Notes:
            Decrements ``merges_due``, increments the invocation count and
            logs ``merge_scheduling.merge_attempted`` for every attempt,
            skipped or not.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        self._merges_due -= 1
        self._merge_invocations += 1
        logger.info(
            "merge_scheduling.merge_attempted",
            iteration=self._state.iteration,
            candidate_id=merge_result.candidate.id,
            parent_indices=merge_result.parent_indices,
            ancestor_idx=merge_result.metadata.get("ancestor_idx"),
            merges_due=self._merges_due,
            total_invocations=self._merge_invocations,
        )
        return not self._skip_merge_candidate(merge_result.candidate)

    def _skip_merge_candidate(self, candidate: Candidate) -> bool:
        """Decide whether a merge candidate must not be evaluated.

        Args:
            candidate (Candidate): The candidate produced by the merge proposer.

        Returns:
            True when the candidate has an invalid output schema or its id
            was already scored; False when it should be evaluated.

        Notes:
            Logs ``merge.proposal_skipped`` with
            ``reason="schema_validation_failed"`` or ``reason="duplicate"``.
            No iteration record is written for a skipped merge; the
            iteration's own proposal record is unaffected.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        if not self._validate_schema_component(candidate):
            reason = "schema_validation_failed"
        elif candidate.id in self._scored:
            reason = "duplicate"
        else:
            return False
        logger.debug(
            "merge.proposal_skipped",
            iteration=self._state.iteration,
            candidate_id=candidate.id,
            reason=reason,
        )
        return True

    def _should_stop(self) -> StopReason | None:
        """Check if evolution should terminate.

        Returns:
            The ``StopReason`` that triggered termination, or ``None`` if
            evolution should continue. Conditions are checked in priority
            order: max iterations, early stopping patience, custom stoppers.

        Notes:
            Only active stoppers (those that passed setup or have no setup
            method) are invoked. Patience-based early stopping returns
            ``PATIENCE`` and is off when ``config.patience`` is 0;
            ``MAX_ITERATIONS`` means only the iteration cap.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        # Condition 1: Max iterations reached (built-in, fast path)
        if self._state.iteration >= self.config.max_iterations:
            return StopReason.MAX_ITERATIONS

        # Condition 2: Early stopping (patience exhausted, built-in)
        if (
            self.config.patience > 0
            and self._state.stagnation_counter >= self.config.patience
        ):
            return StopReason.PATIENCE

        # Condition 3: Custom stoppers (T010-T013)
        # Use _active_stoppers which excludes stoppers that failed setup
        active_stoppers = getattr(self, "_active_stoppers", None)
        if active_stoppers:
            stopper_state = self._build_stopper_state()
            for stopper in active_stoppers:
                try:
                    if stopper(stopper_state):
                        # Log stopper trigger (T013)
                        logger.info(
                            "stopper.triggered",
                            stopper=type(stopper).__name__,
                            iteration=self._state.iteration,
                        )
                        return StopReason.STOPPER_TRIGGERED
                except Exception:
                    # T032: Handle stopper exception gracefully
                    logger.exception(
                        "stopper.error",
                        stopper=type(stopper).__name__,
                        iteration=self._state.iteration,
                    )
                    # Continue checking other stoppers

        return None

    def _should_accept(self, proposal_score: float, best_score: float) -> bool:
        """Check if proposal should be accepted.

        Args:
            proposal_score: Score of the proposed candidate.
            best_score: Current best score.

        Returns:
            True if proposal_score > best_score + min_improvement_threshold.

        Notes:
            Signals True when proposal exceeds best score by the configured
            improvement threshold, enabling configurable acceptance sensitivity.
        """
        threshold = self.config.min_improvement_threshold
        return proposal_score > best_score + threshold

    def _validate_schema_component(self, proposal: Candidate) -> bool:
        """Validate output_schema component if present.

        Validates that proposed schema text is syntactically correct and
        structurally valid (inherits from BaseModel, no imports/functions).
        Invalid schemas are rejected to prevent evolution from accepting
        non-functional schema proposals.

        Args:
            proposal (Candidate): Candidate containing components to validate.

        Returns:
            True if valid or no output_schema component present.
            False if output_schema validation fails.

        Notes:
            This validation runs before expensive evaluation to reject
            invalid schemas early. Security checks (no imports, no functions)
            are enforced to prevent code injection.
        """
        if "output_schema" not in proposal.components:
            return True

        schema_text = proposal.components["output_schema"]

        try:
            validate_schema_text(schema_text)
        except SchemaValidationError as e:
            logger.warning(
                "schema_validation.rejected",
                iteration=self._state.iteration if self._state else None,
                validation_stage=e.validation_stage,
                line_number=e.line_number,
                error=e.validation_error,
            )
            return False
        else:
            logger.debug(
                "schema_validation.passed",
                iteration=self._state.iteration if self._state else None,
            )
            return True

    def _accept_proposal(  # noqa: PLR0913  # optional fields are keyword-only; private helper
        self,
        proposal: Candidate,
        score: float,
        eval_batch: EvaluationBatch,
        *,
        candidate_idx: int | None = None,
        eval_rows: list[int] | None = None,
        reflection_score: float | None = None,
        valset_mean: float | None = None,
        objective_scores: list[dict[str, float]] | None = None,
    ) -> None:
        """Accept a proposal and update state.

        Args:
            proposal (Candidate): Proposed candidate to accept.
            score (float): Acceptance score of the proposed candidate (sum or mean).
            eval_batch (EvaluationBatch): Reflection batch from proposal evaluation (cached for
                next iteration's reflective dataset generation).
            candidate_idx (int | None): Optional ParetoState candidate index to update with
                lineage metadata.
            eval_rows (list[int] | None): Trainset indices the rows of ``eval_batch`` cover, in
                batch order; None means the full trainset in order.
            reflection_score (float | None): Optional trainset score to store with best
                candidate metadata: the mean over ``eval_batch``.
            valset_mean (float | None): Optional valset mean score to track separately from
                acceptance score.
            objective_scores (list[dict[str, float]] | None): Optional objective scores
                from scoring batch. None when adapter does not provide objective scores.

        Notes:
            Replaces the cached reflection batch and its trainset rows for
            the next proposal iteration and tracks acceptance score and valset mean separately.
            The proposal is stored as it is: a mutation already carries its
            parent's id and generation from ``_propose_mutation``, and a
            merge candidate keeps the lineage the merge proposer set.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        if candidate_idx is not None and self._pareto_state is not None:
            self._pareto_state.candidates[candidate_idx] = proposal
        self._state.best_candidate = proposal
        self._state.best_score = score
        self._state.stagnation_counter = 0
        self._state.last_eval_batch = eval_batch
        self._state.last_eval_rows = eval_rows
        if reflection_score is not None:
            self._state.best_reflection_score = reflection_score
        if valset_mean is not None:
            self._state.best_valset_mean = valset_mean
        self._state.best_objective_scores = objective_scores

    def _build_result(
        self, stop_reason: StopReason = StopReason.COMPLETED
    ) -> EvolutionResult:
        """Build final result from current state.

        Args:
            stop_reason (StopReason): Why the evolution run terminated.

        Returns:
            Frozen EvolutionResult with all metrics and original_components.

        Notes:
            Synthesizes a frozen EvolutionResult containing all evolution metrics,
            history, and original_components snapshot, suitable for immutable
            result reporting. The evolved_components dict contains all component
            values from the best candidate. ``total_failed_evaluations`` is
            the baseline count plus the sum over ``iteration_history``, and
            ``token_usage`` is the run token rollup, baseline included,
            with its ``evaluation`` split and a ``reflection`` split that is
            None when no reflection usage was observed.
        """
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call
        baseline_failed = self._state.baseline_failed_evaluations
        return EvolutionResult(
            stop_reason=stop_reason,
            original_score=self._state.original_score,
            final_score=self._state.best_score,
            evolved_components=dict(self._state.best_candidate.components),
            iteration_history=self._state.iteration_history,
            total_iterations=self._state.iteration,
            valset_score=self._state.best_valset_mean,
            trainset_score=self._state.best_reflection_score,
            objective_scores=self._state.best_objective_scores,
            original_components=dict(self._initial_candidate.components),
            baseline_failed_evaluations=baseline_failed,
            total_failed_evaluations=baseline_failed
            + sum(r.failed_evaluations for r in self._state.iteration_history),
            token_usage=self._run_token_rollup(),
        )

    def _write_checkpoint(self) -> None:
        """Write the engine's state to ``config.checkpoint_path``.

        Does nothing when no checkpoint path is configured. Otherwise
        builds the checkpoint dict (version, iteration, stagnation counter,
        best candidate and scores, history, scored map, evaluation counter,
        best reflection batch, random states and run identity) and writes it
        atomically through ``write_checkpoint``.

        Raises:
            TypeError: If a batch field holds a value JSON cannot encode.
            OSError: If the file cannot be written. Nothing is swallowed.

        Notes:
            ``rng_state`` is the engine ``rng``'s state when one was given,
            else null; ``minibatch_rng_state`` is always written.
            ``valset_size`` is null when the valset is the trainset.
            ``last_eval_rows`` holds the trainset indices the cached batch
            covers, null for the full trainset.
            ``run_token_usage`` is the run's token rollup so far, with its
            ``evaluation`` and ``reflection`` splits, so a resumed run
            reports the whole run's usage.
        """
        if self.config.checkpoint_path is None or self._state is None:
            return
        path = Path(self.config.checkpoint_path)
        state = self._state
        data = {
            "checkpoint_version": CHECKPOINT_VERSION,
            "iteration": state.iteration,
            "stagnation_counter": state.stagnation_counter,
            "best_candidate": state.best_candidate.to_dict(),
            "best_score": state.best_score,
            "original_score": state.original_score,
            "best_reflection_score": state.best_reflection_score,
            "best_valset_mean": state.best_valset_mean,
            "best_objective_scores": state.best_objective_scores,
            "baseline_failed_evaluations": state.baseline_failed_evaluations,
            "iteration_history": [r.to_dict() for r in state.iteration_history],
            "scored": dict(self._scored),
            "total_evaluations": self._total_evaluations,
            "last_eval_batch": batch_to_dict(state.last_eval_batch),
            "last_eval_rows": state.last_eval_rows,
            "rng_state": (
                None if self._rng is None else rng_state_to_json(self._rng.getstate())
            ),
            "minibatch_rng_state": rng_state_to_json(self._minibatch_rng.getstate()),
            "initial_candidate_id": self._initial_candidate.id,
            "trainset_size": len(self._trainset),
            "valset_size": None if self._valset_is_trainset else len(self._valset),
            "seed": self.config.seed,
            "run_token_usage": self._run_token_rollup().to_dict(),
            "written_at": datetime.now(UTC).isoformat(),
        }
        write_checkpoint(path, data)
        logger.debug("checkpoint.written", iteration=state.iteration, path=str(path))

    def _load_checkpoint(self) -> dict[str, Any]:
        """Read the checkpoint and refuse one from a different run.

        Returns:
            The checkpoint dict.

        Raises:
            ConfigurationError: If ``config.checkpoint_path`` is unset, the
                file is missing, its version is not ``CHECKPOINT_VERSION``,
                or its initial candidate, trainset size or valset size
                differs from this engine's.
        """
        path = self.config.checkpoint_path
        if path is None:
            raise ConfigurationError(
                "resume=True requires checkpoint_path",
                field="checkpoint_path",
                value=None,
                constraint="not None when resume=True",
            )
        data = read_checkpoint(Path(path))
        check_run_matches(
            data,
            initial_candidate_id=self._initial_candidate.id,
            trainset_size=len(self._trainset),
            valset_size=None if self._valset_is_trainset else len(self._valset),
        )
        return data

    def _resume_from_checkpoint(self) -> None:
        """Restore engine state from ``config.checkpoint_path``.

        Rebuilds ``_EngineState``, the scored map, the evaluation counter,
        the run token rollup, the mutation parent batch and the random
        states from the file, and marks the engine restored so the loop
        skips the baseline.

        Raises:
            ConfigurationError: If ``_load_checkpoint`` refuses the file.

        Notes:
            The engine ``rng`` is restored only when one was given and a
            state was stored; the minibatch random source always is. A
            checkpoint without ``run_token_usage`` restores an unknown
            rollup whose ``rows_unknown`` is the checkpointed evaluation
            count, so the pre-resume rows read as unknown, not free. A
            checkpoint without ``last_eval_rows`` restores the cached batch
            as covering the full trainset; ``_propose_mutation`` reads the
            parent's rows from the restored state. A
            stored rollup without splits restores as evaluation usage with
            reflection not observed; a split one restores its ``evaluation``
            split (zeros when absent) as evaluation and its ``reflection``
            split as reflection, so reflection usage never enters the
            evaluation total. The
            ``on_iteration`` callback is not called for restored history.
            Logs ``checkpoint.resumed`` with the iteration, the stagnation
            counter, the evaluation counter and the path.
        """
        data = self._load_checkpoint()
        last_eval_batch = batch_from_dict(data["last_eval_batch"])
        # Absent in older files: the cached batch is the full trainset
        last_eval_rows = data.get("last_eval_rows")
        self._state = _EngineState(
            best_candidate=Candidate.from_dict(data["best_candidate"]),
            best_score=data["best_score"],
            original_score=data["original_score"],
            iteration=data["iteration"],
            stagnation_counter=data["stagnation_counter"],
            iteration_history=[
                IterationRecord.from_dict(r) for r in data["iteration_history"]
            ],
            last_eval_batch=last_eval_batch,
            last_eval_rows=last_eval_rows,
            best_reflection_score=data["best_reflection_score"],
            best_valset_mean=data["best_valset_mean"],
            best_objective_scores=data["best_objective_scores"],
            baseline_failed_evaluations=data["baseline_failed_evaluations"],
        )
        self._scored = dict(data["scored"])
        self._total_evaluations = data["total_evaluations"]
        stored_usage = data.get("run_token_usage")
        loaded = (
            TokenRollup.from_dict(stored_usage)
            if stored_usage is not None
            else _unknown_tokens(data["total_evaluations"])
        )
        has_split = loaded.evaluation is not None or loaded.reflection is not None
        self._run_token_usage = (
            (loaded.evaluation or _ZERO_TOKENS) if has_split else loaded
        )
        self._run_reflection_usage = loaded.reflection
        self._mutation_parent_batch = last_eval_batch
        if self._rng is not None and data["rng_state"] is not None:
            self._rng.setstate(rng_state_from_json(data["rng_state"]))
        self._minibatch_rng.setstate(rng_state_from_json(data["minibatch_rng_state"]))
        self._restored = True
        logger.info(
            "checkpoint.resumed",
            iteration=self._state.iteration,
            stagnation_counter=self._state.stagnation_counter,
            total_evaluations=self._total_evaluations,
            path=str(self.config.checkpoint_path),
        )

    def _reset_run_state(self) -> None:
        """Clear per-run state so a repeated ``run()`` starts fresh.

        Resets the stopper timer, the evaluation and pending failure
        counters, the pending and run token and reflection rollups, the
        reflection error streak, the scored-candidate map, the engine state
        and the restored flag.
        """
        # Initialize stopper state tracking (T004)
        self._start_time = time.monotonic()
        self._total_evaluations = 0
        self._pending_failed_evaluations = 0
        self._pending_token_usage = self._run_token_usage = _ZERO_TOKENS
        self._pending_reflection_usage = self._run_reflection_usage = None
        self._reflection_error_signature = None
        self._reflection_error_streak = 0
        self._reflection_error_index = -1
        # A fresh run scores every candidate again; stale ids would read
        # first-iteration proposals as duplicates.
        self._scored.clear()
        # State from an earlier run must not leak into this run's partial
        # result; the baseline or the checkpoint rebuilds it.
        self._state = None
        self._restored = False

    async def run(self) -> EvolutionResult:
        """Execute the evolution loop.

        Runs the core evolution loop:
        1. Evaluate baseline candidate
        2. For each iteration until max_iterations or convergence:
           a. Generate reflective dataset from traces
           b. Propose new candidate text
           c. Evaluate proposal
           d. Accept if improves above threshold
           e. Record iteration
        3. Return frozen EvolutionResult
        4. On ``KeyboardInterrupt`` or ``asyncio.CancelledError``, return
           partial result with best-so-far components
        5. On an ``EvolutionError`` after the baseline, attach a partial
           result to the error and re-raise it

        Returns:
            EvolutionResult containing evolution metrics and components.
            On normal completion, ``stop_reason`` reflects the termination
            condition. On interrupt, a partial result is returned with
            ``StopReason.KEYBOARD_INTERRUPT`` or ``StopReason.CANCELLED``.

        Raises:
            KeyboardInterrupt: Re-raised if interrupt occurs before baseline
                evaluation completes (no meaningful partial result possible).
            asyncio.CancelledError: Re-raised if cancellation occurs before
                baseline evaluation completes.
            EvolutionError: Re-raised when it aborts the run, for example a
                non-retryable ``ReflectionError`` (also raised after
                ``config.max_consecutive_reflection_errors`` identical
                reflection errors in a row). When the baseline was
                scored, its ``partial_result`` holds an ``EvolutionResult``
                with ``StopReason.ERROR``, the recorded iterations and the
                best candidate so far; otherwise it stays ``None``.
            Exception: Adapter ``Exception`` subclasses propagate unchanged.

        Examples:
            Running evolution:

            ```python
            result = await engine.run()
            print(f"Improved: {result.improved}")
            print(f"Best score: {result.final_score}")
            ```

        Notes:
            Outputs a frozen EvolutionResult after completing the evolution
            loop. Engine instance should not be reused after run() completes.
            Method is idempotent if called multiple times (restarts fresh):
            the evaluation counters and the scored-candidate map used for
            duplicate detection are cleared before the baseline runs.
            Fail-fast behavior: adapter ``Exception`` subclasses propagate
            unchanged. ``KeyboardInterrupt`` and ``asyncio.CancelledError``
            (``BaseException`` subclasses) are caught and converted to partial
            results with appropriate ``StopReason``. An ``EvolutionError`` is
            logged as ``evolution.aborted`` at error level, without a
            traceback, and re-raised; after the
            baseline it carries ``partial_result`` whose
            ``total_iterations`` counts recorded iterations only, since the
            aborted iteration was never recorded. Logs seed value at start
            for reproducibility tracking, and logs
            ``reflection.minibatch.enabled`` when a reflection minibatch
            smaller than the trainset is in effect; with a separate valset a
            proposal that passes that gate is scored on the valset without a
            full trainset pass, and ``trainset_score`` is then the mean over
            the accepted candidate's sampled rows. ``_reset_run_state``
            resets the evaluation counter, the pending failure counter, the
            engine state, the pending and run token rollups and the pending
            and run reflection rollups at start, so a reused engine
            never attaches an earlier run's state to ``partial_result``.
            With ``config.resume`` it then restores
            the checkpoint through ``_resume_from_checkpoint``, which sets
            the counters and scored map from the file, and the loop skips
            the baseline. An exception raised by
            ``config.proposal_validator`` is not an ``EvolutionError``, so it
            propagates unchanged without a partial result; a rejection it
            returns is recorded with ``skip_reason="proposal_rejected"``.
        """
        logger.info("engine.start", seed=self.config.seed)
        if self._effective_minibatch_size() is not None:
            logger.info(
                "reflection.minibatch.enabled",
                size=self.config.reflection_minibatch_size,
                trainset_size=len(self._trainset),
            )

        self._reset_run_state()
        if self.config.resume:
            self._resume_from_checkpoint()

        # Setup stopper lifecycle (T023)
        setup_stoppers = self._setup_stoppers()

        try:
            return await self._run_evolution_loop()
        except EvolutionError as error:
            logger.error(  # noqa: TRY400  # re-raised below with partial_result attached
                "evolution.aborted",
                iteration=self._state.iteration if self._state else 0,
                error_type=type(error).__name__,
                error=str(error),
            )
            if self._state is not None:
                # The aborted iteration was never recorded
                self._state.iteration = len(self._state.iteration_history)
                error.partial_result = self._build_result(stop_reason=StopReason.ERROR)
            raise
        except KeyboardInterrupt:
            logger.info(
                "evolution.interrupted",
                iteration=self._state.iteration if self._state else 0,
            )
            if self._state is None:
                raise
            return self._build_result(stop_reason=StopReason.KEYBOARD_INTERRUPT)
        except asyncio.CancelledError:
            logger.info(
                "evolution.cancelled",
                iteration=self._state.iteration if self._state else 0,
            )
            if self._state is None:
                raise
            return self._build_result(stop_reason=StopReason.CANCELLED)
        finally:
            # Cleanup stopper lifecycle (T024)
            self._cleanup_stoppers(setup_stoppers)

    async def _run_evolution_loop(self) -> EvolutionResult:  # noqa: C901, PLR0912, PLR0915  # decomposition tracked in #507
        """Run propose-evaluate-accept iterations until a stop condition holds.

        This method contains the actual evolution loop logic, separated
        from lifecycle management for clean try/finally handling.

        Returns:
            EvolutionResult with evolution outcomes.

        Raises:
            ReflectionError: If proposing raises a non-retryable
                ``ReflectionError``, or after
                ``config.max_consecutive_reflection_errors`` identical
                retryable ones in a row; ``run()`` attaches the partial result.

        Notes:
            Only called from run(). Handles the evolution loop body
            while run() manages stopper lifecycle. The loop tracks
            ``StopReason`` to report why evolution terminated.
            The parent index ``_propose_mutation`` returns is the proposal's
            only genealogy edge in the Pareto state; the selector is not
            asked a second time.
            Each iteration records ``reflection_reasoning`` from the
            adapter's proposer via a ``getattr`` chain when available.
            Each proposal's reflection batch is handed to scoring so a
            valset that is the trainset is not evaluated twice.
            An ``EmptyProposalError`` from proposing is recorded as a
            skipped iteration (``skip_reason="empty_proposal"``) that counts
            toward stagnation; the loop then checks stop conditions and
            continues. A ``ReflectionTimeoutError`` is handled the same way
            with ``skip_reason="reflection_timeout"``, and a retryable
            ``ReflectionError`` with ``skip_reason="reflection_error"``; a
            non-retryable ``ReflectionError`` propagates to ``run()``,
            which attaches the partial result. An
            ``IncompleteProposalError`` is recorded with
            ``skip_reason="incomplete_proposal"``, counts toward stagnation
            and is followed by the stop check. A proposal that
            ``config.proposal_validator`` rejects is recorded, before schema
            validation, with ``skip_reason="proposal_rejected"`` and the
            validator's reason, counts toward stagnation and is followed by
            the stop check; an exception from the validator propagates. A
            proposal whose
            ``output_schema`` text fails validation is recorded the same way
            with ``skip_reason="schema_validation_failed"``, counts toward
            stagnation and is followed by the stop check. A proposal whose
            ``Candidate.id`` was already scored (baseline, proposal or merge)
            is recorded with
            ``skip_reason="duplicate"`` and its stored score instead of being
            evaluated. With a reflection minibatch in effect, a proposal
            that does not beat its parent on the iteration's sampled rows is
            recorded with ``skip_reason="minibatch_rejected"`` and not
            evaluated further. An evaluated proposal's record carries the
            scoring batch's row count as ``rows_evaluated``. A proposal that
            passes is evaluated on the
            full trainset when the valset is the trainset (that batch is
            reused for scoring); with a separate valset the gate's batch is
            its reflection batch and only the valset is evaluated (see
            ``_evaluate_passed_proposal``). The trainset rows of each cached
            batch are stored beside it for the next gate and reflection. A
            merge candidate is always evaluated on the full trainset, so
            its rows are None. A merge candidate that is a duplicate or has an
            invalid schema is not evaluated, and the iteration's own record
            and stop check still follow. Each iteration starts with no
            pending failures, a zero pending token rollup and no pending
            reflection usage, so an unrecorded iteration's failures and tokens
            do not carry into the next record. After the proposal returns,
            or raises one of the skip errors, the adapter proposer's
            ``last_token_usage`` is read through ``_take_proposer_usage`` so
            the iteration's record carries its reflection cost.
            Each accept decision is logged as ``proposal.accepted`` or
            ``proposal.rejected`` with the proposal's ``candidate_id``.
            Every recorded iteration, skipped or not, reaches
            ``config.on_iteration`` before the stop check.
            State restored from a checkpoint replaces the baseline
            evaluation.
        """
        # A restored checkpoint already holds the baseline
        if not self._restored:
            await self._initialize_baseline()
        assert self._state is not None, "Engine state not initialized"  # noqa: S101  # narrows state that run() sets before this call

        # Evolution loop
        stop_reason = self._should_stop()
        while stop_reason is None:
            self._state.iteration += 1
            # Failures from an unrecorded previous iteration do not carry over
            self._pending_failed_evaluations = 0
            self._pending_token_usage = _ZERO_TOKENS
            self._pending_reflection_usage = None

            # Propose mutation (returns candidate and list of components evolved)
            try:
                (
                    proposal,
                    evolved_components_list,
                    parent_idx,
                ) = await self._propose_mutation()
            except EmptyProposalError as error:
                # Empty reflection after retry: failed iteration, not fatal
                self._take_proposer_usage()
                await self._record_empty_proposal(error)
                stop_reason = self._should_stop()
                continue
            except ReflectionTimeoutError as error:
                # Reflection timed out: skipped iteration, not fatal
                self._take_proposer_usage()
                await self._record_reflection_timeout(error)
                stop_reason = self._should_stop()
                continue
            except ReflectionError as error:
                # A programming or client error will not clear on its own
                if not error.retryable:
                    raise
                # Provider error survived the proposer's retry: skip it
                self._take_proposer_usage()
                await self._record_reflection_error(error)
                stop_reason = self._should_stop()
                continue
            except IncompleteProposalError as error:
                # Truncated reflection output: skipped iteration, not fatal
                self._take_proposer_usage()
                await self._record_incomplete_proposal(error)
                stop_reason = self._should_stop()
                continue
            self._take_proposer_usage()

            # Caller's validator rejected the text: record it, no evaluation
            if await self._record_if_rejected(proposal, evolved_components_list):
                stop_reason = self._should_stop()
                continue

            # Validate schema component if present (reject invalid early)
            if not self._validate_schema_component(proposal):
                # Invalid schema: record the skip, no evaluation
                await self._record_schema_validation_skip(
                    proposal, evolved_components_list
                )
                stop_reason = self._should_stop()
                continue

            # Already scored this run: reuse its score, no evaluation
            if await self._record_if_duplicate(proposal, evolved_components_list):
                stop_reason = self._should_stop()
                continue

            # Beat the parent on a trainset minibatch before the full pass
            if not await self._gate_on_minibatch(proposal, evolved_components_list):
                stop_reason = self._should_stop()
                continue

            # Reflection batch: the gate's rows with a separate valset,
            # otherwise the full trainset pass
            (
                reflection_score,
                reflection_batch,
                reflection_rows,
            ) = await self._evaluate_passed_proposal(proposal)
            proposal_score, scoring_batch, eval_indices = await self._evaluate_scoring(
                proposal,
                reflection_batch=reflection_batch if reflection_rows is None else None,
            )

            candidate_idx = None
            if self._pareto_state is not None:
                # Use eval_indices returned from _evaluate_scoring (T066)
                # Prepare objective scores if available (T068)
                objective_scores: dict[str, float] | None = None
                per_example_objective_scores: dict[int, dict[str, float]] | None = None

                if scoring_batch.objective_scores is not None:
                    if self.config.frontier_type in (
                        FrontierType.OBJECTIVE,
                        FrontierType.HYBRID,
                    ):
                        # Aggregate objective scores across evaluated examples
                        objective_scores_accum: dict[str, list[float]] = {}
                        for obj_scores in scoring_batch.objective_scores:
                            for obj_name, obj_score in obj_scores.items():
                                objective_scores_accum.setdefault(obj_name, []).append(
                                    obj_score
                                )
                        # Take mean per objective
                        objective_scores = {
                            obj_name: fmean(scores)
                            for obj_name, scores in objective_scores_accum.items()
                        }

                    if self.config.frontier_type == FrontierType.CARTESIAN:
                        # For CARTESIAN, need per-example objective scores mapped to valset indices
                        per_example_objective_scores = {
                            eval_indices[i]: scoring_batch.objective_scores[i]
                            for i in range(len(eval_indices))
                        }
                        # Also need aggregated for validation
                        objective_scores_by_name: dict[str, list[float]] = {}
                        for obj_scores in scoring_batch.objective_scores:
                            for obj_name, obj_score in obj_scores.items():
                                objective_scores_by_name.setdefault(
                                    obj_name, []
                                ).append(obj_score)
                        objective_scores = {
                            obj_name: fmean(scores)
                            for obj_name, scores in objective_scores_by_name.items()
                        }

                # The genealogy edge is the parent the reflector rewrote
                parent_indices = [parent_idx] if parent_idx is not None else None

                # Pass scores with correct index mapping (T066)
                candidate_idx = self._pareto_state.add_candidate(
                    proposal,
                    scoring_batch.scores,
                    score_indices=eval_indices,
                    objective_scores=objective_scores,
                    per_example_objective_scores=per_example_objective_scores,
                    parent_indices=parent_indices,
                    logger=logger,
                )
                self._candidate_eval_batches[candidate_idx] = reflection_batch
                self._candidate_eval_rows[candidate_idx] = reflection_rows
                logger.info(
                    "pareto_frontier.candidate_added",
                    candidate_idx=candidate_idx,
                    candidate_id=proposal.id,
                    iteration=self._state.iteration,
                )

            # Calculate valset mean using only evaluated scores (T067)
            valset_mean = (
                sum(scoring_batch.scores) / len(scoring_batch.scores)
                if scoring_batch.scores
                else 0.0
            )

            # Accept if improves above threshold
            accepted = self._should_accept(proposal_score, self._state.best_score)
            if accepted:
                logger.info(
                    "proposal.accepted",
                    iteration=self._state.iteration,
                    candidate_id=proposal.id,
                    score=proposal_score,
                    best_score=self._state.best_score,
                )
                self._accept_proposal(
                    proposal,
                    proposal_score,
                    reflection_batch,
                    candidate_idx=candidate_idx,
                    eval_rows=reflection_rows,
                    reflection_score=reflection_score,
                    valset_mean=valset_mean,
                    objective_scores=scoring_batch.objective_scores,
                )
                # Schedule merge if enabled
                if (
                    self.config.use_merge
                    and self._merge_proposer is not None
                    and self._merge_invocations < self.config.max_merge_invocations
                ):
                    self._merges_due += 1
                    logger.debug(
                        "merge_scheduling.merge_scheduled",
                        iteration=self._state.iteration,
                        candidate_id=proposal.id,
                        merges_due=self._merges_due,
                    )
            else:
                logger.debug(
                    "proposal.rejected",
                    iteration=self._state.iteration,
                    candidate_id=proposal.id,
                    score=proposal_score,
                    best_score=self._state.best_score,
                )
                # Increment stagnation counter on rejection
                self._state.stagnation_counter += 1

            # Attempt merge if scheduled
            if (
                self._merges_due > 0
                and self._merge_proposer is not None
                and self._pareto_state is not None
                and self._merge_invocations < self.config.max_merge_invocations
            ):
                merge_result = await self._merge_proposer.propose(self._pareto_state)
                # A skipped merge (invalid schema or already scored) is not
                # evaluated; the iteration is still recorded below
                if merge_result is not None and self._start_merge(merge_result):
                    # Evaluate merge proposal
                    (
                        merge_reflection_score,
                        merge_reflection_batch,
                    ) = await self._evaluate_reflection(merge_result.candidate)
                    (
                        merge_proposal_score,
                        merge_scoring_batch,
                        merge_eval_indices,
                    ) = await self._evaluate_scoring(
                        merge_result.candidate,
                        reflection_batch=merge_reflection_batch,
                    )

                    # Add merge candidate to ParetoState
                    merge_candidate_idx = None
                    assert self._pareto_state is not None, (  # noqa: S101  # narrows state that run() sets before this call
                        "Pareto state not initialized"
                    )
                    merge_objective_scores: dict[str, float] | None = None
                    merge_per_example_objective_scores: (
                        dict[int, dict[str, float]] | None
                    ) = None

                    if merge_scoring_batch.objective_scores is not None:
                        if self.config.frontier_type in (
                            FrontierType.OBJECTIVE,
                            FrontierType.HYBRID,
                        ):
                            merge_objective_scores_accum: dict[str, list[float]] = {}
                            for obj_scores in merge_scoring_batch.objective_scores:
                                for obj_name, obj_score in obj_scores.items():
                                    merge_objective_scores_accum.setdefault(
                                        obj_name, []
                                    ).append(obj_score)
                            merge_objective_scores = {
                                obj_name: fmean(scores)
                                for obj_name, scores in merge_objective_scores_accum.items()
                            }

                        if self.config.frontier_type == FrontierType.CARTESIAN:
                            merge_per_example_objective_scores = {
                                merge_eval_indices[
                                    i
                                ]: merge_scoring_batch.objective_scores[i]
                                for i in range(len(merge_eval_indices))
                            }
                            merge_objective_scores_by_name: dict[str, list[float]] = {}
                            for obj_scores in merge_scoring_batch.objective_scores:
                                for obj_name, obj_score in obj_scores.items():
                                    merge_objective_scores_by_name.setdefault(
                                        obj_name, []
                                    ).append(obj_score)
                            merge_objective_scores = {
                                obj_name: fmean(scores)
                                for obj_name, scores in merge_objective_scores_by_name.items()
                            }

                    merge_candidate_idx = self._pareto_state.add_candidate(
                        merge_result.candidate,
                        merge_scoring_batch.scores,
                        score_indices=merge_eval_indices,
                        objective_scores=merge_objective_scores,
                        per_example_objective_scores=merge_per_example_objective_scores,
                        parent_indices=merge_result.parent_indices,
                        logger=logger,
                    )
                    self._candidate_eval_batches[merge_candidate_idx] = (
                        merge_reflection_batch
                    )
                    self._candidate_eval_rows[merge_candidate_idx] = None

                    merge_valset_mean = (
                        sum(merge_scoring_batch.scores)
                        / len(merge_scoring_batch.scores)
                        if merge_scoring_batch.scores
                        else 0.0
                    )

                    # Accept merge if improves
                    merge_accepted = self._should_accept(
                        merge_proposal_score, self._state.best_score
                    )
                    if merge_accepted:
                        self._accept_proposal(
                            merge_result.candidate,
                            merge_proposal_score,
                            merge_reflection_batch,
                            candidate_idx=merge_candidate_idx,
                            reflection_score=merge_reflection_score,
                            valset_mean=merge_valset_mean,
                            objective_scores=merge_scoring_batch.objective_scores,
                        )
                        logger.info(
                            "merge_scheduling.merge_accepted",
                            iteration=self._state.iteration,
                            candidate_id=merge_result.candidate.id,
                            merge_score=merge_proposal_score,
                        )
                    else:
                        logger.debug(
                            "merge_scheduling.merge_rejected",
                            iteration=self._state.iteration,
                            candidate_id=merge_result.candidate.id,
                            merge_score=merge_proposal_score,
                            best_score=self._state.best_score,
                        )
                elif merge_result is None:
                    # Merge not possible, decrement counter
                    if self._merges_due > 0:
                        self._merges_due -= 1

            # Record iteration with actual evolved component name (T033)
            # For single-component evolution, use the first (and only) component.
            # For multi-component round-robin, this tracks which component was
            # evolved in this iteration.
            if evolved_components_list:
                evolved_component_name = evolved_components_list[0]
            else:
                # Empty list indicates logic error - use first key from proposal
                logger.warning(
                    "engine.empty_evolved_components_list",
                    iteration=self._state.iteration,
                    proposal_keys=list(proposal.components.keys()),
                )
                evolved_component_name = next(iter(proposal.components.keys()))
            await self._record_iteration(
                score=proposal_score,
                component_text=proposal.components.get(evolved_component_name, ""),
                evolved_component=evolved_component_name,
                accepted=accepted,
                objective_scores=scoring_batch.objective_scores,
                # TODO: Surface last_reasoning through adapter protocol instead
                # of reaching into private _proposer attribute. Safe (defaults
                # to None) but couples engine to adapter internals.
                reflection_reasoning=getattr(
                    getattr(self.adapter, "_proposer", None),
                    "last_reasoning",
                    None,
                ),
                candidate_id=proposal.id,
                parent_ids=_parent_ids(proposal),
                rows_evaluated=len(scoring_batch.scores),
            )

            stop_reason = self._should_stop()

        # Build and return result
        return self._build_result(stop_reason=stop_reason)