shared/evolution.py

"""Governed self-evolution loop — the Round 4 flagship (pure core).

Darwin Godel Machine (arXiv:2505.22954, ICLR 2026) showed an agent can improve
itself by editing its own scaffold and keeping an archive of discovered variants;
Group-Evolving Agents (arXiv:2602.04837) showed a *shared* experience pool turns
early exploratory diversity into sustained progress. This module assembles both
out of the repo's own primitives:

- the **genome** is a :class:`shared.harness.HarnessConfig` (Chunk 1), mutated by
  flipping primitives, swapping the model, or pointing at a GEPA-evolved prompt;
- **fitness** is a holdout score vector (Chunk 2's ``score_config`` on the live
  path; a deterministic synthetic surface on the dry-run path);
- the **selection gate** accepts a child only when its paired holdout improvement
  over its parent is CI-significant (``bootstrap_paired_diff_ci``); ties and
  regressions are recorded honestly, never silently kept;
- the **lineage archive** is DGM-style: every candidate carries a parent pointer,
  and parents for new candidates are sampled from the archive of accepted genomes
  (open-ended search, not hill-climbing a single point);
- the **shared-experience archive** is GEA-style: which (failure-mode -> mutation)
  pairs have paid off is written to :class:`shared.memory.LongTermMemory` (the
  namespaced key/value store) and read back to steer future proposals across the
  whole population.

This module is pure and deterministic given a seed: fitness is injected, mutation
choices come from a seeded RNG, and nothing here calls an LLM or touches the
network. Governance (CaMeL secure execution, FormalGuard pre-execution proofs,
filesystem denylist, hard cost-abort, kill-switch, per-generation provenance) is
layered on by Chunk 4 via the hooks this module exposes (``GovernanceHooks``).
The CLI, the live/dry-run fitness functions, and the report live in
``agents/_meta/evolve.py``.
"""

from __future__ import annotations

import random
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from typing import Protocol

from shared.harness import GENOME_VERSION, HarnessConfig
from shared.stats import CI, bootstrap_paired_diff_ci, is_significant

# Namespace under which the GEA shared-experience archive is stored in LongTermMemory.
EXPERIENCE_NAMESPACE = "evolve/experience"


@dataclass(frozen=True)
class Candidate:
    """One member of the evolving population.

    ``cid`` is the requested-genome identity (so precedence-shadowed variants stay
    distinct and the lineage DAG never self-loops). ``mutation`` is a short human
    description of the edit that produced this candidate from its parent.
    """

    cid: str
    config: HarnessConfig
    generation: int
    parent_id: str | None = None
    mutation: str = "baseline"
    # Reserved seams for future mutation operators, unused on the genome-only path:
    prompt_text: str | None = None  # a GEPA-synthesized instruction (not yet emitted; see docs)
    patch: str | None = None  # a swe-agent scaffold patch (Chunk 4 governance territory)

    @staticmethod
    def of(
        config: HarnessConfig,
        *,
        generation: int,
        parent_id: str | None,
        mutation: str,
        **extra: object,
    ) -> Candidate:
        return Candidate(
            cid=config.requested_fingerprint(),
            config=config,
            generation=generation,
            parent_id=parent_id,
            mutation=mutation,
            prompt_text=extra.get("prompt_text"),  # type: ignore[arg-type]
            patch=extra.get("patch"),  # type: ignore[arg-type]
        )


@dataclass(frozen=True)
class Fitness:
    """A candidate's holdout fitness: per-example scores aligned by ``example_ids``."""

    scores: tuple[float, ...]
    example_ids: tuple[str, ...]
    cost_usd: float = 0.0
    failure_modes: tuple[tuple[str, int], ...] = ()

    @property
    def mean(self) -> float:
        return sum(self.scores) / len(self.scores) if self.scores else 0.0

    def dominant_failure_mode(self) -> str | None:
        """The most frequent MAST mode in this candidate's runs, if any."""
        if not self.failure_modes:
            return None
        return max(self.failure_modes, key=lambda kv: kv[1])[0]


# A fitness function scores a candidate's genome over the holdout. Injected so the
# loop is testable and dry-run-deterministic without any LLM call.
FitnessFn = Callable[[Candidate], Fitness]


@dataclass
class LineageRecord:
    """One node in the DGM-style lineage archive."""

    candidate: Candidate
    fitness: Fitness
    parent_id: str | None
    parent_mean: float | None
    gate_ci: CI | None  # paired diff (candidate - parent); None for the baseline
    accepted: bool
    reason: str
    genome_version: int = GENOME_VERSION


class GovernanceHooks(Protocol):
    """Pre-execution governance gates, implemented by Chunk 4.

    The loop calls ``vet(candidate)`` before any candidate is scored; a falsey
    return means the candidate is rejected un-run (e.g. a FormalGuard closure
    proof failed, or a scaffold patch touched a denylisted path). The default
    :class:`AllowAllGovernance` permits everything so the loop runs standalone.
    """

    def vet(self, candidate: Candidate) -> tuple[bool, str]:  # (allowed, reason)
        ...


class AllowAllGovernance:
    """Default no-op governance: every candidate is allowed (Chunk 3 standalone)."""

    def vet(self, candidate: Candidate) -> tuple[bool, str]:
        return True, "allow-all (no governance configured)"


# ---------------------------------------------------------------------------
# Mutation — MAST-steered, GEA-shared
# ---------------------------------------------------------------------------

# How a MAST failure category steers the next mutation. Grounded in the taxonomy:
# FC1 (system design / poor planning) -> add deliberation (reflexion) or a better
# prompt; FC2 (inter-agent misalignment) -> a stronger model or v2 prompt; FC3
# (task verification) -> add the CaMeL verification path (secure). These are
# priors, not guarantees; the experience archive overrides them as evidence accrues.
_MODE_PRIOR: dict[str, tuple[str, object]] = {
    "FC1": ("reflexion", True),
    "FC2": ("prompt_version", "v2"),
    "FC3": ("secure", True),
}

# The fields the genome-only mutation operator may toggle, with their candidate
# values. Deterministic and small so dry-run evolution is reproducible.
_MUTABLE: tuple[tuple[str, tuple[object, ...]], ...] = (
    ("reflexion", (True, False)),
    ("prompt_version", ("v2", None)),
    ("secure", (True, False)),
    ("model", ("gpt-4o-mini", "deepseek")),
    # The learned-router axis: evolve the routing policy itself, not just which
    # primitives are on. None = static cost-mode table; "llm_judge" = Conductor.
    # (Shadowed when a model pin is also set; resolve() collapses that case so the
    # search does not waste a generation on a behaviourally-identical genome.)
    ("router_policy", (None, "llm_judge")),
)


def _category_of(mode_id: str) -> str:
    """Map a MAST mode id (e.g. 'FM-3.2') to its category 'FC3'."""
    # FM-<cat>.<n> -> FC<cat>
    try:
        cat_num = mode_id.split("-", 1)[1].split(".", 1)[0]
        return f"FC{cat_num}"
    except (IndexError, ValueError):
        return "FC1"


@dataclass(frozen=True)
class Mutation:
    """A proposed child plus the exact (field, value) edit that produced it."""

    candidate: Candidate
    field: str
    value: object


def propose_mutation(
    parent: Candidate,
    parent_fitness: Fitness,
    generation: int,
    *,
    experience: ExperienceArchive,
    rng: random.Random,
) -> Mutation:
    """Propose a child genome from a parent, steered by MAST + shared experience.

    Strategy, in priority order:
    1. If the parent has a dominant failure mode and the experience archive records
       a mutation that previously improved that mode, apply it (GEA reuse).
    2. Else, if the parent has a dominant failure mode, apply the taxonomy prior
       for its category (MAST steering).
    3. Else, pick a random mutable field/value (exploration).
    The chosen change is only kept if it actually alters the resolved genome;
    otherwise we fall back to a random different field so no generation is wasted
    on a no-op mutation. Returns the child *and* the applied (field, value) so the
    caller can record the exact edit in the shared-experience archive.
    """
    mode = parent_fitness.dominant_failure_mode()
    field_name: str | None = None
    value: object = None
    rationale = "random exploration"

    if mode is not None:
        learned = experience.best_mutation_for(mode)
        if learned is not None:
            field_name, value = learned
            rationale = f"GEA reuse: {mode} -> {field_name}={value}"
        else:
            cat = _category_of(mode)
            if cat in _MODE_PRIOR:
                field_name, value = _MODE_PRIOR[cat]
                rationale = f"MAST prior: {cat} -> {field_name}={value}"

    if field_name is None:
        field_name, choices = rng.choice(_MUTABLE)
        value = rng.choice(choices)

    child_config = parent.config.evolve(**{field_name: value})
    # Avoid a no-op (resolves to the parent's behaviour): try other fields.
    tries = 0
    while child_config.fingerprint() == parent.config.fingerprint() and tries < len(_MUTABLE) * 2:
        field_name, choices = rng.choice(_MUTABLE)
        value = rng.choice(choices)
        child_config = parent.config.evolve(**{field_name: value})
        rationale = f"random exploration ({field_name}={value})"
        tries += 1

    child = Candidate.of(child_config, generation=generation, parent_id=parent.cid, mutation=rationale)
    return Mutation(candidate=child, field=field_name, value=value)


def _candidate_mutations(parent: Candidate, generation: int, seen: set[str]) -> list[Mutation]:
    """All single-field mutations of the parent that are novel and not no-ops."""
    out: list[Mutation] = []
    for field_name, choices in _MUTABLE:
        for value in choices:
            child_config = parent.config.evolve(**{field_name: value})
            if child_config.fingerprint() == parent.config.fingerprint():
                continue  # no behavioural change
            child = Candidate.of(
                child_config,
                generation=generation,
                parent_id=parent.cid,
                mutation=f"surrogate-EI: {field_name}={value}",
            )
            if child.cid in seen:
                continue
            out.append(Mutation(candidate=child, field=field_name, value=value))
    return out


def propose_surrogate_ei(
    parent: Candidate,
    generation: int,
    *,
    observations: Sequence[tuple[dict, float]],
    best: float,
    seen: set[str],
    rng: random.Random,
    lam: float = 0.5,
    xi: float = 0.01,
) -> Mutation | None:
    """Surrogate-guided proposal: pick the candidate mutation with the highest
    Expected Improvement under a Bayesian surrogate fit on the run's observations.

    This is online, model-based (Bayesian-optimization-style) search over the
    harness genome: the surrogate is fit on the (genome, observed-fitness) pairs
    seen *so far this run* (not the true surface), so it is not circular — it
    generalizes from evaluated genomes to un-evaluated neighbours. Returns ``None``
    on cold start (too few observations) or when no novel candidate remains, so the
    caller can fall back to the MAST/GEA/random proposer.
    """
    candidates = _candidate_mutations(parent, generation, seen)
    if not candidates or len(observations) < len(_MUTABLE):
        return None
    # Lazy import: numpy/surrogate are only needed on this opt-in path.
    from shared.surrogate import BayesianHarnessSurrogate

    genomes = [obs[0] for obs in observations]
    targets = [obs[1] for obs in observations]
    surrogate = BayesianHarnessSurrogate.fit(genomes, targets, lam=lam)

    scored = [(surrogate.expected_improvement(m.candidate.config.to_dict(), best, xi=xi), m) for m in candidates]
    max_ei = max(ei for ei, _ in scored)
    # Break ties deterministically via the seeded rng (keeps the loop reproducible).
    best_choices = [m for ei, m in scored if ei >= max_ei - 1e-12]
    return rng.choice(best_choices)


# ---------------------------------------------------------------------------
# Shared-experience archive (GEA), backed by EpisodicMemory
# ---------------------------------------------------------------------------


class ExperienceArchive:
    """GEA-style shared experience: which (failure-mode -> mutation) pairs paid off.

    Persisted in the repo's long-term memory store
    (:class:`shared.memory.LongTermMemory`, the namespaced key/value backend) under
    ``evolve/experience`` so the knowledge survives across generations and
    (optionally) across runs, and so the whole population draws on one pool rather
    than each lineage re-discovering the same fixes. An in-memory dict backend is
    used when no store is supplied (tests, dry-run determinism). The backend must
    expose ``get(namespace, key, default)`` and ``set(namespace, key, value)``.
    """

    def __init__(self, memory: object | None = None) -> None:
        self._memory = memory
        self._local: dict[str, dict[str, float]] = {}

    def _load(self, mode: str) -> dict[str, float]:
        if self._memory is not None:
            return dict(self._memory.get(EXPERIENCE_NAMESPACE, mode, default={}) or {})
        return dict(self._local.get(mode, {}))

    def _store(self, mode: str, table: dict[str, float]) -> None:
        if self._memory is not None:
            self._memory.set(EXPERIENCE_NAMESPACE, mode, table)
        else:
            self._local[mode] = table

    def record(self, mode: str | None, field_name: str, value: object, delta: float) -> None:
        """Accumulate the observed fitness delta for applying a mutation to a mode.

        ``value`` is the actual value the mutation set (e.g. ``True``, ``"v2"``),
        stored so :meth:`best_mutation_for` can later re-propose the exact edit.
        """
        if mode is None:
            return
        key = f"{field_name}={value}"
        table = self._load(mode)
        # Running sum of deltas; positive means the mutation has helped this mode.
        table[key] = table.get(key, 0.0) + delta
        self._store(mode, table)

    def best_mutation_for(self, mode: str) -> tuple[str, object] | None:
        """Return the (field, value) with the highest positive cumulative delta for a mode."""
        table = self._load(mode)
        if not table:
            return None
        best_key, best_delta = max(table.items(), key=lambda kv: kv[1])
        if best_delta <= 0:
            return None
        field_name, _, value_str = best_key.partition("=")
        return field_name, _parse_value(value_str)


def _parse_value(text: str) -> object:
    if text == "True":
        return True
    if text == "False":
        return False
    if text in ("None", ""):
        return None
    return text


# ---------------------------------------------------------------------------
# Selection gate
# ---------------------------------------------------------------------------


def evaluate_gate(
    candidate: Fitness,
    parent: Fitness,
    *,
    min_delta: float = 0.0,
    seed: int = 1234,
) -> tuple[bool, CI | None, str]:
    """Decide whether to accept ``candidate`` over ``parent``.

    Accept ONLY when the paired holdout improvement is CI-significant (the CI
    excludes zero) and positive beyond ``min_delta``. The comparison is paired by
    example, so the candidate and parent must have been scored on the same
    holdout; misalignment is a hard error, not a silent garbage comparison.
    Ties (CI spans 0) and regressions (significant but negative) are reported
    honestly and rejected.
    """
    if candidate.example_ids != parent.example_ids:
        return False, None, "rejected: holdout misaligned (candidate vs parent example ids differ)"
    ci = bootstrap_paired_diff_ci(list(candidate.scores), list(parent.scores), seed=seed)
    if not is_significant(ci):
        return False, ci, f"rejected: tie (paired delta {ci.point:+.3f}, 95% CI spans 0)"
    if ci.point < 0:
        return False, ci, f"rejected: regression (paired delta {ci.point:+.3f}, CI [{ci.low:+.3f}, {ci.high:+.3f}])"
    if ci.point <= min_delta:
        return False, ci, f"rejected: below min_delta {min_delta:+.3f} (paired delta {ci.point:+.3f})"
    return True, ci, f"accepted: CI-significant improvement {ci.point:+.3f} (CI [{ci.low:+.3f}, {ci.high:+.3f}])"


# ---------------------------------------------------------------------------
# The loop
# ---------------------------------------------------------------------------


@dataclass
class EvolutionConfig:
    generations: int = 5
    population: int = 3  # candidates proposed per generation
    seed: int = 1234
    min_delta: float = 0.0
    max_cost_usd: float | None = None  # hard-abort ceiling on cumulative fitness cost (whole run)
    per_generation_max_cost_usd: float | None = None  # hard-abort ceiling on a single generation
    kill_switch: Callable[[], bool] | None = None  # returns True to stop the loop
    # Proposal strategy: "default" = MAST-steered + GEA-reuse + random; "surrogate-ei"
    # = online Bayesian optimization (fit a surrogate on this run's observations,
    # propose the max-Expected-Improvement mutation), falling back to "default" on
    # cold start. Default keeps pre-existing behaviour byte-for-byte.
    proposal_strategy: str = "default"


@dataclass
class EvolutionResult:
    base: Candidate
    base_fitness: Fitness
    records: list[LineageRecord] = field(default_factory=list)
    archive: list[LineageRecord] = field(default_factory=list)  # accepted genomes (incl. baseline)
    stopped_reason: str = "completed"
    generations_completed: int = 0  # generation-loop iterations actually executed
    generations_productive: int = 0  # generations that evaluated >=1 novel candidate
    duplicates_skipped: int = 0  # proposals skipped because the genome was already seen
    gate_tests_run: int = 0  # total paired gate comparisons (the multiple-comparison family size)
    # best-vs-baseline on the SELECTION set (the same data used to choose best;
    # upward-biased by selection — the winner's curse).
    baseline_gate_ci: CI | None = None
    # best-vs-baseline on a HELD-OUT VALIDATION set (fresh data, never used for
    # selection) — the honest headline when a validate_fn is supplied.
    base_val_fitness: Fitness | None = None
    best_val_fitness: Fitness | None = None
    val_gate_ci: CI | None = None

    @property
    def best(self) -> LineageRecord:
        """The accepted record with the highest mean SELECTION fitness (baseline if none beat it)."""
        return max(self.archive, key=lambda r: r.fitness.mean)

    @property
    def headline_ci(self) -> CI | None:
        """The CI the report should headline: validation if available, else selection.

        Validation is the honest one — it is computed on data not used to choose
        ``best``, so it is free of the selection/winner's-curse inflation that
        contaminates the selection-set comparison.
        """
        return self.val_gate_ci if self.val_gate_ci is not None else self.baseline_gate_ci

    @property
    def improved(self) -> bool:
        """True only when best beats baseline by a CI-significant positive margin.

        Uses the validation comparison when present (free of selection bias), else
        the selection-set comparison. NOT a raw mean comparison: a tie or a
        non-significant difference is not an improvement.
        """
        ci = self.headline_ci
        return ci is not None and is_significant(ci) and ci.point > 0


def evolve(
    base_config: HarnessConfig,
    fitness_fn: FitnessFn,
    *,
    config: EvolutionConfig | None = None,
    experience: ExperienceArchive | None = None,
    governance: GovernanceHooks | None = None,
    validate_fn: FitnessFn | None = None,
) -> EvolutionResult:
    """Run the governed evolution loop and return the full lineage.

    Open-ended (DGM-style): each generation samples a parent from the archive of
    accepted genomes and proposes ``population`` children; each child is vetted by
    governance, scored on the SELECTION set, and gated against its parent.
    Accepted children join the archive and can themselves be sampled as parents
    later. The loop is deterministic given ``config.seed`` when ``fitness_fn`` is
    deterministic.

    ``validate_fn``, if supplied, scores a candidate on a HELD-OUT validation set
    disjoint from the selection set. After the loop, the baseline and the winning
    genome are re-scored with it and a paired baseline-vs-best CI is computed on
    that fresh data — the honest headline, free of the selection (winner's-curse)
    inflation that contaminates the selection-set comparison. Without it, the
    report falls back to the selection-set baseline comparison and says so.
    """
    cfg = config or EvolutionConfig()
    exp = experience or ExperienceArchive()
    gov = governance or AllowAllGovernance()
    rng = random.Random(cfg.seed)

    base = Candidate.of(base_config, generation=0, parent_id=None, mutation="baseline")
    base_fit = fitness_fn(base)
    base_record = LineageRecord(
        candidate=base,
        fitness=base_fit,
        parent_id=None,
        parent_mean=None,
        gate_ci=None,
        accepted=True,
        reason="baseline",
    )

    result = EvolutionResult(base=base, base_fitness=base_fit, records=[base_record], archive=[base_record])
    fitness_by_id: dict[str, Fitness] = {base.cid: base_fit}
    seen: set[str] = {base.cid}
    cumulative_cost = base_fit.cost_usd
    # (genome, observed mean) pairs for the optional surrogate-EI proposer.
    observations: list[tuple[dict, float]] = [(base.config.to_dict(), base_fit.mean)]

    for gen in range(1, cfg.generations + 1):
        if cfg.kill_switch is not None and cfg.kill_switch():
            result.stopped_reason = f"kill-switch tripped before generation {gen}"
            break
        if cfg.max_cost_usd is not None and cumulative_cost > cfg.max_cost_usd:
            result.stopped_reason = f"cost cap ${cfg.max_cost_usd:.2f} reached (spent ${cumulative_cost:.4f})"
            break
        result.generations_completed = gen
        produced_novel = False
        gen_cost = 0.0

        # DGM-style parent sampling: bias toward the best, but allow any archived
        # genome (deterministic via the seeded rng).
        parent_record = _sample_parent(result.archive, rng)
        parent = parent_record.candidate
        parent_fit = fitness_by_id[parent.cid]

        for _ in range(cfg.population):
            if cfg.per_generation_max_cost_usd is not None and gen_cost > cfg.per_generation_max_cost_usd:
                result.stopped_reason = (
                    f"per-generation cost cap ${cfg.per_generation_max_cost_usd:.2f} reached in generation {gen}"
                )
                break
            proposal = None
            if cfg.proposal_strategy == "surrogate-ei":
                proposal = propose_surrogate_ei(
                    parent,
                    gen,
                    observations=observations,
                    best=result.best.fitness.mean,
                    seen=seen,
                    rng=rng,
                )
            if proposal is None:  # default strategy, or surrogate cold-start fallback
                proposal = propose_mutation(parent, parent_fit, gen, experience=exp, rng=rng)
            child = proposal.candidate
            if child.cid in seen:
                result.duplicates_skipped += 1
                continue  # never re-evaluate or self-loop a genome already in the DAG
            seen.add(child.cid)

            # A child whose RESOLVED genome equals its parent's would always tie;
            # don't spend a (possibly paid) fitness evaluation to confirm it.
            if child.config.fingerprint() == parent.config.fingerprint():
                noop_reason = "rejected: no-op (resolves to parent)"
                noop = LineageRecord(child, Fitness((), ()), parent.cid, parent_fit.mean, None, False, noop_reason)
                result.records.append(noop)
                continue

            allowed, gov_reason = gov.vet(child)
            if not allowed:
                vetoed = LineageRecord(
                    child, Fitness((), ()), parent.cid, parent_fit.mean, None, False, f"vetoed: {gov_reason}"
                )
                result.records.append(vetoed)
                continue

            child_fit = fitness_fn(child)
            produced_novel = True
            cumulative_cost += child_fit.cost_usd
            gen_cost += child_fit.cost_usd
            fitness_by_id[child.cid] = child_fit
            observations.append((child.config.to_dict(), child_fit.mean))

            # Per-comparison bootstrap seed: deterministic but independent across
            # gate tests (avoids correlated Monte-Carlo error across the family).
            gate_seed = (cfg.seed * 1_000_003 + int(child.cid, 16)) % (2**31)
            accepted, ci, reason = evaluate_gate(child_fit, parent_fit, min_delta=cfg.min_delta, seed=gate_seed)
            result.gate_tests_run += 1
            # Feed the exact applied edit + observed delta into the shared archive.
            exp.record(
                parent_fit.dominant_failure_mode(),
                proposal.field,
                proposal.value,
                child_fit.mean - parent_fit.mean,
            )

            record = LineageRecord(child, child_fit, parent.cid, parent_fit.mean, ci, accepted, reason)
            result.records.append(record)
            if accepted:
                result.archive.append(record)

            if cfg.max_cost_usd is not None and cumulative_cost > cfg.max_cost_usd:
                result.stopped_reason = f"cost cap ${cfg.max_cost_usd:.2f} reached (spent ${cumulative_cost:.4f})"
                _finalize(result, base_fit, cfg, validate_fn)
                return result

            # Per-generation cap: also checked AFTER accrual so a single candidate
            # cannot silently overrun the generation's budget (the top-of-loop
            # check alone would let the cap-crossing candidate complete first).
            if cfg.per_generation_max_cost_usd is not None and gen_cost > cfg.per_generation_max_cost_usd:
                result.stopped_reason = (
                    f"per-generation cost cap ${cfg.per_generation_max_cost_usd:.2f} reached in generation {gen}"
                )
                break

        if produced_novel:
            result.generations_productive += 1

    _finalize(result, base_fit, cfg, validate_fn)
    return result


def _finalize(
    result: EvolutionResult,
    base_fit: Fitness,
    cfg: EvolutionConfig,
    validate_fn: FitnessFn | None,
) -> None:
    """Compute the honest baseline-vs-best comparisons after the search ends.

    Always computes the selection-set baseline CI (so the headline delta and its
    CI describe the *same* comparison — best vs baseline, not best vs its parent).
    When a validation fn is supplied, re-scores baseline and best on the held-out
    set and computes the unbiased validation CI.
    """
    best = result.best
    if best.candidate.cid != result.base.cid and best.fitness.scores and base_fit.scores:
        _, ci, _ = evaluate_gate(best.fitness, base_fit, min_delta=cfg.min_delta, seed=cfg.seed)
        result.baseline_gate_ci = ci

    if validate_fn is not None:
        base_val = validate_fn(result.base)
        best_val = validate_fn(best.candidate)
        result.base_val_fitness = base_val
        result.best_val_fitness = best_val
        if best.candidate.cid != result.base.cid:
            _, vci, _ = evaluate_gate(best_val, base_val, min_delta=cfg.min_delta, seed=cfg.seed)
            result.val_gate_ci = vci


def _sample_parent(archive: list[LineageRecord], rng: random.Random) -> LineageRecord:
    """Sample a parent from the archive, biased toward higher fitness (DGM-style).

    Deterministic given the rng. Half the time take the current best; otherwise
    take a uniformly random archived genome, so the search stays open-ended.
    """
    if len(archive) == 1:
        return archive[0]
    if rng.random() < 0.5:
        return max(archive, key=lambda r: r.fitness.mean)
    return rng.choice(archive)