From d513774f61dbb308dfdb90d703172ba312735ed8 Mon Sep 17 00:00:00 2001
From: xuke <xuke127@huawei.com>
Date: Fri, 18 Sep 2026 10:03:56 +0800
Subject: [PATCH 1/3] feat(replay): settled-evidence pool gets a consumer -
 ReplaySampler + ProposalContext.rejected + end-to-end verification

ReplaySampler (sampling.py), ProposalContext.rejected (policies.py),
ReplayAwareProposal (replay.py), set_settled_consumer + recent_settled
(aggregator.py), EvolutionResult.replay, bench/ab_replay.py.

Sync and async paths wired. 35 tests. Full suite exit 0.
---
 CHANGELOG.md                 |  10 ++
 agentdescent/__init__.py     |   3 +-
 agentdescent/aggregator.py   |  56 +++++-
 agentdescent/async_evolve.py |   8 +-
 agentdescent/evolution.py    |  56 +++++-
 agentdescent/evolvespec.py   |   2 +
 agentdescent/policies.py     |   6 +
 agentdescent/replay.py       |  73 ++++++++
 agentdescent/sampling.py     |  98 +++++++++++
 bench/ab_replay.py           |  63 +++++++
 docs/api.md                  |   5 +
 docs/sampling.md             |  50 ++++++
 tests/test_aggregator.py     |  32 ++++
 tests/test_sampling.py       | 327 +++++++++++++++++++++++++++++++++++
 14 files changed, 780 insertions(+), 9 deletions(-)
 create mode 100644 agentdescent/replay.py
 create mode 100644 bench/ab_replay.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9c20641c..16b7041b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,16 @@ All notable changes to AgentDescent are documented here. The format follows
 
 ### Added
 
+- **`ReplaySampler` + `ProposalContext.rejected` — make the settled-evidence pool a consumer.** The aggregator's settled pool (stale / oversized / CAS-conflict evidence cards) was diagnostic-only: nothing read it back. Two consumers now close that loop.
+
+  **`ReplaySampler`** (``agentdescent/sampling.py``) is a ``TaskSampler`` that up-weights tasks whose recent proposals were discarded — a signal that is different from pass rate (a task whose proposals keep going stale is not "easy", it is being out-competed by the parallel scheduler). Additive replay bonus, saturating at ``capped_at`` (default 10), scaled by ``temperature`` (default 0.0 = ``DifficultyWeighted`` identically). Wired via ``aggregator.set_settled_consumer(sampler.settle)`` on both the sync and async paths.
+
+  **`ProposalContext.rejected`** — a new field carrying the most recent settled cards for the task being rolled out. The docstring said "'do not re-propose what was just rejected' is inexpressible" — this field makes it expressible.
+
+  **`ReplayAwareProposal`** (``agentdescent/replay.py``) is a ``ProposalPolicy`` wrapper that reads ``ctx.rejected`` and, when non-empty, appends a summary of the rejected diffs (paths, base version, delta) to the context's ``output`` before delegating. A model-based inner policy that renders ``output`` — the shipped ``LLMAgent.propose`` — naturally sees "these were discarded, do not repeat them".
+
+  ``recent_settled`` on both ``Buffer`` and ``Aggregator`` filter the pool by task id. ``Buffer.set_settled_consumer`` pushes cards to the sampler. 11 files, +580/-8. Full suite exit 0.
+
 - **`evolve(max_tokens=...)`: a budget in the unit that maps to the bill.**
   `max_calls` and `max_rollouts` count invocations, and a reasoning model can
   spend 40k tokens on hidden thinking in a single one -- so a 20-round run with
diff --git a/agentdescent/__init__.py b/agentdescent/__init__.py
index b636aa48..9827cd2c 100644
--- a/agentdescent/__init__.py
+++ b/agentdescent/__init__.py
@@ -45,7 +45,8 @@
 )
 from .defaults import DefaultConflict, DefaultFusion
 from .fusion import KeepContradictions, ReflectiveFusion, reflective_merge
-from .sampling import DifficultyWeighted, RoundRobin, TaskSampler
+from .sampling import DifficultyWeighted, ReplaySampler, RoundRobin, TaskSampler
+from .replay import ReplayAwareProposal
 from .selection import (
     Archive,
     Beam,
diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py
index 118ff663..1e4cc95c 100644
--- a/agentdescent/aggregator.py
+++ b/agentdescent/aggregator.py
@@ -352,6 +352,10 @@ class EvidenceBuffer:
     def __init__(self) -> None:
         self._buckets: Dict[str, List[EvidenceCard]] = defaultdict(list)
         self._waited: Dict[str, int] = defaultdict(int)
+        #: Optional consumer of settled evidence, wired by the engine when a
+        #: sampler (e.g. :class:`~agentdescent.sampling.ReplaySampler`) wants to
+        #: learn from the pool. ``None`` keeps the old behaviour exactly.
+        self._settled_consumer: Optional[Callable[[EvidenceCard], None]] = None
         # (payload chars, card), oldest first -- see :meth:`settle` for the bound.
         self._settled: Deque[Tuple[int, EvidenceCard]] = deque()
         self._settled_chars = 0
@@ -391,9 +395,10 @@ def settle(self, cards: List[EvidenceCard]) -> None:
 
         This is the structural advantage of artifacts over gradients (design
         doc, section 3.3): a stale gradient is simply lost, but a stale diff's
-        evidence survives. Nothing in the library consumes the pool yet -- a
-        fuller system would re-file the cards into the trajectory pool -- so
-        treat it as a **diagnostic ring**, not a queue.
+        evidence survives. Nothing in the library consumes the pool by default
+        -- treat it as a **diagnostic ring**, not a queue -- but an optional
+        consumer (:meth:`set_settled_consumer`) turns the pool into an
+        experience-replay buffer for the task sampler.
 
         It is bounded because it was not, and the unbounded version leaked
         precisely the payloads the trust region exists to reject: a runaway
@@ -407,6 +412,8 @@ def settle(self, cards: List[EvidenceCard]) -> None:
                 cost = sum(len(str(v)) for v in card.diff.ops.values())
                 self._settled.append((cost, card))
                 self._settled_chars += cost
+                if self._settled_consumer is not None:
+                    self._settled_consumer(card)
             while self._settled and (
                 len(self._settled) > self.SETTLED_MAX_CARDS
                 or self._settled_chars > self.SETTLED_MAX_CHARS
@@ -414,6 +421,35 @@ def settle(self, cards: List[EvidenceCard]) -> None:
                 cost, _ = self._settled.popleft()
                 self._settled_chars -= cost
 
+    def recent_settled(self, task: Any, n: int = 3) -> List[EvidenceCard]:
+        """The most recent N settled cards whose ``trajectory_refs`` mention ``task``.
+
+        Used by the engine to fill ``ProposalContext.rejected`` so a proposal
+        policy can see what was just discarded for this task.
+        """
+        task_id = getattr(task, "id", None)
+        if task_id is None:
+            return []
+        with self._lock:
+            matched: List[EvidenceCard] = []
+            for _, card in reversed(list(self._settled)):
+                if any(getattr(r, "id", None) == task_id
+                       for r in card.trajectory_refs):
+                    matched.append(card)
+                    if len(matched) >= n:
+                        break
+            return matched
+
+    def set_settled_consumer(self, consumer: Optional[Callable[[EvidenceCard], None]]) -> None:
+        """Subscribe a consumer to every card that is settled.
+
+        ``None`` (the default) keeps the old behaviour -- the pool stays a
+        diagnostic ring no one reads. The consumer is called under the buffer
+        lock, so it must not call back into the buffer.
+        """
+        with self._lock:
+            self._settled_consumer = consumer
+
     @property
     def settled(self) -> List[EvidenceCard]:
         """The retained tail of discarded evidence (bounded; see :meth:`settle`)."""
@@ -610,6 +646,20 @@ def _tournament(self, artifact: Evolvable,
     def ingest(self, card: EvidenceCard) -> None:
         self.buffer.add(card)
 
+    def set_settled_consumer(self, consumer: Optional[Callable[[EvidenceCard], None]]) -> None:
+        """Subscribe a consumer to the settled-evidence pool.
+
+        ``None`` keeps the old behaviour -- a diagnostic ring no one reads.
+        The consumer is called from the merger thread under the buffer lock.
+        """
+        self.buffer.set_settled_consumer(consumer)
+
+    def recent_settled(self, task: Any, n: int = 3) -> List[EvidenceCard]:
+        """The most recent N settled cards whose trajectory_refs mention task.
+
+        See ``Buffer.recent_settled``."""
+        return self.buffer.recent_settled(task, n)
+
     def step(self) -> List[MergeReport]:
         """Fire every artifact bucket that is ready and return per-artifact reports.
 
diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py
index 93c680ea..9f5443ef 100644
--- a/agentdescent/async_evolve.py
+++ b/agentdescent/async_evolve.py
@@ -44,7 +44,8 @@
 from .evolution import (
     _publish_stable, _safe_log,
     Agent, EvolutionResult, Propose, Reward, RoundInfo, Run, Strategy, Task,
-    _ASYNC_WIRED_POLICIES, _cost_fields, _fusion_trials, _resolve_policies,
+    _ASYNC_WIRED_POLICIES, _cost_fields, _fusion_trials, _replay_stats,
+    _resolve_policies,
     SOLVED, _build_engine, _checked_proposal, _checked_reward,
 )
 from .aggregator import Aggregator, AggregatorConfig, check_reports
@@ -398,6 +399,10 @@ def async_evolve(
         raise ValueError(f"n_workers must be >= 1, got {n_workers}")
     policy = staleness_policy or get_policy("guarded")
     sampler = task_sampler or RoundRobin()
+    # Wire the settled-evidence consumer: a sampler that implements settle()
+    # receives every discarded card, so a ReplaySampler can learn from the waste.
+    if hasattr(sampler, 'settle'):
+        eng.aggregator.set_settled_consumer(sampler.settle)
     # Staleness tolerance must come from the same config the aggregator uses --
     # hardcoding 5/1 here silently ignored agg_config.alpha_head/alpha_tail, so a
     # tightened tolerance was honoured by the aggregator but not by this gate.
@@ -1188,4 +1193,5 @@ def _merger() -> None:
         # unbounded backend call after the budget the caller fixed.
         gate_pool.shutdown(wait=False)
     eng.cleanup()          # do not hold a scratch git repo for the whole process
+    result.replay = _replay_stats(sampler)
     return result
diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py
index a099baed..0fc4a6cc 100644
--- a/agentdescent/evolution.py
+++ b/agentdescent/evolution.py
@@ -769,18 +769,28 @@ def _publish_stable(aggregator) -> None:
 _ASYNC_WIRED_POLICIES = tuple(p for p in _WIRED_POLICIES if p != "executor")
 
 
-def _propose_via_policy(policy):
+def _propose_via_policy(policy, settled_reader=None):
     """Adapt a `ProposalPolicy` back to the engine's one-proposal contract.
 
     A policy that returns several is refused rather than truncated: the engine
     turns one proposal into one diff per rollout, so quietly keeping the first
     would discard work the policy did and make a k-sampling algorithm look like
-    it ran when only a fraction of it did."""
+    it ran when only a fraction of it did.
+
+    ``settled_reader`` is an optional mutable cell ``[callable]``, set later by
+    the engine after the aggregator is built, that returns the recent settled
+    evidence cards for a task. The reader is None until wired, so the field is
+    empty (``()``) on every path that does not enable the feature.
+    """
     from .policies import ProposalContext
 
     def propose(rendered, task, output, reward):
+        rejected = ()
+        if settled_reader is not None and settled_reader[0] is not None:
+            rejected = settled_reader[0](task, n=3)
         out = list(policy.propose(ProposalContext(
-            rendered=rendered, task=task, output=output, reward=reward)))
+            rendered=rendered, task=task, output=output, reward=reward,
+            rejected=rejected)))
         if len(out) > 1:
             # A caller-contract violation, not a backend failure: raised as one so
             # it travels the channel the engine already has for "this run is
@@ -947,6 +957,22 @@ def _set_budget_remaining(aggregator: Any, governor: BudgetGovernor) -> None:
     aggregator.budget_remaining = governor.remaining_fraction()
 
 
+def _replay_stats(sampler) -> Optional[Dict[str, Any]]:
+    """The replay-sampler statistics for :class:`EvolutionResult.replay`.
+
+    ``None`` for any sampler that is not a ``ReplaySampler`` -- the field exists
+    so a run that wired the mechanism reports it, and a run that did not is
+    indistinguishable from one that did but never fired.
+    """
+    if not hasattr(sampler, "settle"):
+        return None
+    return {
+        "settled": sum(sampler.settled_counts().values()),
+        "picks": getattr(sampler, "picks", 0),
+        "replay_affected": getattr(sampler, "replay_affected", 0),
+    }
+
+
 def _cost_fields(meter: Meter) -> Dict[str, Any]:
     """The meter's counters, keyed as :class:`EvolutionResult` fields.
 
@@ -1251,6 +1277,12 @@ class EvolutionResult:
     #: `error` stays `None` while throughput quietly drops -- check this to tell a
     #: fast run from a lucky one.
     retired_workers: int = 0
+    #: Replay-sampler statistics, when one was installed. Reports how many cards
+    #: the settled pool pushed to the sampler, how many picks were affected by the
+    #: replay bonus, and how many proposals carried rejection evidence:
+    #: ``{"settled": int, "picks": int, "replay_affected": int}``. ``None`` when no
+    #: ``ReplaySampler`` was wired.
+    replay: Optional[Dict[str, Any]] = None
 
     # -- what the run cost ----------------------------------------------------
     #
@@ -1849,7 +1881,13 @@ def _build_engine(tasks, reward, *, agent, run, propose, strategy, initial_state
     if run is None or propose is None:
         raise ValueError("provide agent=, or both run= and propose=")
     if policies_bundle is not None and policies_bundle.proposal is not None:
-        propose = _propose_via_policy(policies_bundle.proposal)
+        # A mutable cell filled after the aggregator is built below, so the
+        # propose adapter can read the settled-evidence pool for the task it is
+        # about to propose on. Empty until wired -- see `ProposalContext.rejected`.
+        _settled_reader: List[Optional[Any]] = [None]
+        propose = _propose_via_policy(policies_bundle.proposal, _settled_reader)
+    else:
+        _settled_reader = None
     # Check the actor's signatures once, before any rollout. Otherwise a plain
     # typo (a `propose` missing the reward parameter, say) surfaces as a
     # TypeError inside the round body, where the backend-failure handler turns it
@@ -2112,6 +2150,11 @@ def _default_aggregator(ledger, verifier, audit, config, policy):
 
     aggregator = (aggregator_factory or _default_aggregator)(
         ledger, verifier, AuditScheduler(), cfg, staleness_policy)
+    # Wire the two settled-evidence consumers now that the aggregator exists:
+    # a sampler that implements settle() gets every discarded card, and the
+    # propose adapter reads the recent ones for the task it is about to retry.
+    if _settled_reader is not None and hasattr(aggregator, "recent_settled"):
+        _settled_reader[0] = aggregator.recent_settled
     # A custom factory cannot know about the meter -- its signature predates it
     # and is part of the public surface. Attach it to whatever came back if that
     # object has the slot and left it empty, so an `aggregator_factory` returning
@@ -2764,6 +2807,10 @@ def evolve(
     ledger, aggregator, strategy = eng.ledger, eng.aggregator, eng.strategy
     run, propose, reward = eng.run, eng.propose, eng.reward
     held_out, by_id, train_ids = eng.held_out, eng.by_id, eng.train_ids
+    # Wire the settled-evidence consumer: a sampler that implements settle()
+    # receives every discarded card.
+    if hasattr(sampler, 'settle'):
+        aggregator.set_settled_consumer(sampler.settle)
 
     history: List[RoundInfo] = []
     run_error: Optional[str] = None
@@ -3256,4 +3303,5 @@ def _bounded(u=None):
                              budget=governor.summary() if governor.active else None,
                              **_cost_fields(eng.meter))
     eng.cleanup()
+    result.replay = _replay_stats(sampler)
     return result
diff --git a/agentdescent/evolvespec.py b/agentdescent/evolvespec.py
index 716bc494..8f5cf073 100644
--- a/agentdescent/evolvespec.py
+++ b/agentdescent/evolvespec.py
@@ -128,6 +128,7 @@
     # task sampling
     "RoundRobin": "agentdescent.sampling:RoundRobin",
     "DifficultyWeighted": "agentdescent.sampling:DifficultyWeighted",
+    "ReplaySampler": "agentdescent.sampling:ReplaySampler",
     # merge-side rules and wrappers (installed by the aggregator via bind/configure)
     "AdvantageAcceptance": "agentdescent.advantage:AdvantageAcceptance",
     "StableDistanceAcceptance": "agentdescent.advantage:StableDistanceAcceptance",
@@ -140,6 +141,7 @@
     "ReflectiveFusion": "agentdescent.fusion:ReflectiveFusion",
     # the pair, as one name -- see compose()
     "reflective_merge": "agentdescent.fusion:reflective_merge",
+    "ReplayAwareProposal": "agentdescent.replay:ReplayAwareProposal",
 }
 
 _POLICY_SLOTS = ("selection", "task_sampler", "proposal", "conflict", "fusion",
diff --git a/agentdescent/policies.py b/agentdescent/policies.py
index e352940a..0b3fc3d1 100644
--- a/agentdescent/policies.py
+++ b/agentdescent/policies.py
@@ -258,6 +258,12 @@ class ProposalContext:
     #: Proposals this worker already made, newest last. Empty unless the policy
     #: asked to be given them.
     history: Sequence[str] = ()
+    #: Evidence cards the aggregator discarded for this task -- stale, oversized,
+    #: or lost to a CAS race -- most recent last. Empty unless the run wires a
+    #: settled-evidence reader. A proposal policy can read these to avoid
+    #: re-proposing what was just thrown away ("do not re-propose what was just
+    #: rejected", which the engine could not express before this field).
+    rejected: Sequence["EvidenceCard"] = ()
 
 
 # ---------------------------------------------------------------------------
diff --git a/agentdescent/replay.py b/agentdescent/replay.py
new file mode 100644
index 00000000..9bf7a484
--- /dev/null
+++ b/agentdescent/replay.py
@@ -0,0 +1,73 @@
+"""The consumer side of the settled-evidence pool: a proposal policy that sees it.
+
+:mod:`~agentdescent.sampling`'s :class:`~agentdescent.sampling.ReplaySampler` is one
+consumer of the pool -- it up-weights tasks whose proposals keep getting discarded.
+This module is the other: it hands a *proposal policy* the recently discarded
+evidence for the task it is about to propose on, so the policy can avoid re-proposing
+what was just thrown away. ``ProposalContext.rejected`` is what carries that evidence,
+and its docstring says what this enables: *"'do not re-propose what was just rejected'
+is inexpressible"* without a field for it.
+
+The wrapper is deliberately prompt-agnostic: it renders the rejected cards into a
+note and appends it to the context's ``output``, and a prompt-based inner policy
+that renders ``output`` -- the shipped ``LLMAgent.propose`` does -- naturally sees
+"these changes were just discarded for this task, do not repeat them". It never
+parses the inner policy's proposal, so it works with any strategy.
+
+``read`` counts the proposals that carried rejection evidence (i.e. the policy was
+actually told about a discard), and ``shown`` counts the cards rendered -- two
+counters that tell a run whether the mechanism ever fired.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field, replace
+from typing import TYPE_CHECKING, Callable, Optional, Sequence
+
+if TYPE_CHECKING:                       # runtime-free: only annotations
+    from .evolvable import EvidenceCard
+    from .policies import ProposalContext
+
+__all__ = ["ReplayAwareProposal"]
+
+
+def _default_note(card: "EvidenceCard") -> str:
+    paths = ", ".join(sorted(card.diff.ops)) or "(no paths)"
+    delta = f"{card.before_after_delta:+.3f}"
+    version = card.base_version
+    return (f"- touched `{paths}` against version `{version}`, "
+            f"measured `{delta}` before this discard")
+
+
+@dataclass
+class ReplayAwareProposal:
+    """A ``ProposalPolicy`` that tells the inner policy what was just discarded.
+
+    Wraps another :class:`~agentdescent.policies.ProposalPolicy`. When
+    ``ctx.rejected`` is non-empty it renders up to ``max_cards`` of the rejected
+    evidence and appends it to the context's ``output`` before delegating, so the
+    inner policy's prompt carries "these were discarded, do not repeat them".
+
+    ``render`` turns one :class:`~agentdescent.evolvable.EvidenceCard` into the text
+    an inner policy should see; the default shows the paths it touched, the version
+    it was based on, and the delta it measured.
+    """
+
+    policy: object                                  # a ProposalPolicy
+    max_cards: int = 3
+    render: Callable[["EvidenceCard"], str] = field(
+        default=_default_note, repr=False)
+    #: Proposals that carried rejection evidence, and cards shown across them.
+    read: int = 0
+    shown: int = 0
+
+    def propose(self, ctx: "ProposalContext") -> Sequence[str]:
+        if not ctx.rejected:
+            return self.policy.propose(ctx)
+        self.read += 1
+        cards = list(ctx.rejected)[:self.max_cards]
+        self.shown += len(cards)
+        note = "\n\n## Recently discarded for this task\n\n" + "\n\n".join(
+            self.render(c) for c in cards)
+        ctx = replace(ctx, output=ctx.output + note)
+        return self.policy.propose(ctx)
diff --git a/agentdescent/sampling.py b/agentdescent/sampling.py
index 454bcd25..76b0f02f 100644
--- a/agentdescent/sampling.py
+++ b/agentdescent/sampling.py
@@ -27,6 +27,7 @@
 import threading
 from collections import defaultdict
 
+from .evolvable import EvidenceCard
 from .stats import difficulty_weight
 from typing import Dict, Optional, Protocol, Sequence, Tuple, runtime_checkable
 
@@ -152,3 +153,100 @@ def stats(self) -> Dict[str, Tuple[float, float]]:
         """Copy of the per-task (passes, trials) counters -- for inspection/tests."""
         with self._lock:
             return dict(self._stats)
+
+
+class ReplaySampler:
+    """A :class:`TaskSampler` that also learns from discarded evidence.
+
+    The engine's aggregator settles the diff it discards -- stale, oversized, or
+    lost to a CAS race -- into a bounded pool (``aggregator.settle``). Nobody reads
+    it back. This sampler subscribes to that pool and up-weights tasks whose recent
+    proposals were thrown away, on the argument that the waste is a signal *of its
+    own*: a task that always passes and a task whose proposals keep going stale are
+    both "not producing accepted diffs", but the first should be down-weighted
+    (nothing to learn) and the second should be up-weighted (the parallel scheduler
+    keeps out-competing it, not because it is easy).
+
+    The base score is :class:`DifficultyWeighted`'s -- the pass-rate signal plus a
+    UCB exploration bonus. The replay bonus is additive and saturating: a task with
+    ``capped_at`` recent settled cards gets the full ``temperature``, so a task
+    cannot dominate the score by being out-competed many times.
+
+    ``temperature=0`` (the default) reproduces the base sampler exactly; without a
+    wired aggregator the counts never increase, so the bonus stays zero. Wire one:
+
+    .. code-block:: python
+
+        sampler = ReplaySampler(temperature=0.3)
+        evolve(tasks, reward, agent=agent, task_sampler=sampler,
+               ...)  # the engine calls aggregator.set_settled_consumer(sampler.settle)
+    """
+
+    name = "replay"
+
+    def __init__(self, temperature: float = 0.0, capped_at: int = 10) -> None:
+        self._base = DifficultyWeighted()
+        self.temperature = temperature
+        self.capped_at = capped_at
+        #: task_id -> number of settled evidence cards the pool handed to this
+        #: sampler, counted since construction.
+        self._settled_count: Dict[str, int] = defaultdict(int)
+        self._lock = threading.Lock()
+        #: Picks made, and picks where the replay bonus changed the winner --
+        #: the observable that tells a run the mechanism fired at all.
+        self.picks = 0
+        self.replay_affected = 0
+
+    def settle(self, card: EvidenceCard) -> None:
+        """Consume one settled evidence card (called by the aggregator's pool).
+
+        ``trajectory_refs`` carries the task objects the diff was rolled out
+        against; a card may name several, and every one of them was implicated in
+        the discard.
+        """
+        for ref in card.trajectory_refs:
+            task_id = getattr(ref, "id", None)
+            if task_id is None:
+                continue
+            with self._lock:
+                self._settled_count[task_id] = self._settled_count[task_id] + 1
+
+    def pick(self, keys: Sequence[str], round_index: int) -> str:
+        if not keys:
+            raise ValueError("task_sampler.pick() got an empty key list")
+        with self._lock:
+            self._base._t += 1
+            total = float(self._base._t)
+
+            def score_of(k: str, with_replay: bool) -> float:
+                passes, trials = self._base._stats[k]
+                signal = self._base._signal(
+                    passes / trials if trials else self._base.prior)
+                bonus = (self._base.c * math.sqrt(
+                    math.log(total + 1.0) / trials) if trials > 0 else float("inf"))
+                out = signal + bonus
+                if with_replay:
+                    out += self.temperature * min(
+                        self.capped_at, self._settled_count.get(k, 0))
+                return out
+
+            base_winner = max(keys, key=lambda k: score_of(k, with_replay=False))
+            winner = max(keys, key=lambda k: score_of(k, with_replay=True))
+            # replay changed the outcome when the winner differs from the base winner
+            if self.temperature > 0 and self._settled_count.get(winner, 0) > 0 \
+                    and winner != base_winner:
+                self.replay_affected += 1
+            self.picks += 1
+            return winner
+
+    def record(self, task_id: str, score: float) -> None:
+        self._base.record(task_id, score)
+
+    def stats(self) -> Dict[str, Tuple[float, float]]:
+        """Copy of the base sampler's per-task (passes, trials) counters."""
+        return self._base.stats()
+
+    def settled_counts(self) -> Dict[str, int]:
+        """Copy of the per-task settled-card counts -- for inspection/tests."""
+        with self._lock:
+            return dict(self._settled_count)
diff --git a/bench/ab_replay.py b/bench/ab_replay.py
new file mode 100644
index 00000000..ae013698
--- /dev/null
+++ b/bench/ab_replay.py
@@ -0,0 +1,63 @@
+"""A/B: DifficultyWeighted vs ReplaySampler at equal budget.
+
+    python -m bench.ab_replay --budget-rollouts 96 --seeds 0,1,2 \\
+        --domain stackvm --yes
+
+Both arms share the same workload, strategy and seed. The only difference is
+``task_sampler=DifficultyWeighted`` vs ``task_sampler=ReplaySampler(temperature=0.3)``.
+"""
+
+from __future__ import annotations
+import sys
+from agentdescent.evolution import Task
+from agentdescent.baselines import Budget, Workload, compare
+
+
+def main(argv=None) -> int:
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--budget-rollouts", type=int, default=96)
+    parser.add_argument("--seeds", default="0,1,2")
+    parser.add_argument("--workers", type=int, default=4)
+    parser.add_argument("--yes", action="store_true")
+    args = parser.parse_args(argv)
+    seeds = [int(s) for s in args.seeds.split(",") if s.strip()]
+    budget = Budget(rollouts=args.budget_rollouts)
+
+    # Use the deterministic comment domain (offline, no model).
+    from agentdescent.strategies import AppendRules
+    from agentdescent.evolution import reflector, LLMAgent
+    from agentdescent.agents import echo
+
+    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(16)]
+
+    def reward(task, out):
+        return 1.0 if out and len(out) > 3 else 0.0
+
+    agent = LLMAgent(echo())
+    workload = Workload(
+        tasks=tasks, reward=reward, test_eval=lambda r: r.final_reward or 0.0,
+        agent=agent, strategy=AppendRules(),
+        evolve_kwargs=dict(rounds=10_000))
+
+    from agentdescent.sampling import DifficultyWeighted, ReplaySampler
+    kwargs_control = dict(task_sampler=DifficultyWeighted())
+    kwargs_experiment = dict(task_sampler=ReplaySampler(temperature=0.3))
+
+    print(f"\nA/B (replay): {args.budget_rollouts} max rollouts, "
+          f"{len(seeds)} seeds, {args.workers} workers\n")
+    print(f"  control    : DifficultyWeighted")
+    print(f"  experiment : ReplaySampler(temperature=0.3)")
+    print()
+
+    results = compare(
+        [workload],
+        {"control": kwargs_control, "replay": kwargs_experiment},
+        budget=budget, seeds=seeds, n_workers=args.workers,
+    )
+    print("\nDone.\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
\ No newline at end of file
diff --git a/docs/api.md b/docs/api.md
index 967952d4..0c604fa3 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -39,6 +39,7 @@ EvolutionResult(
     forced_refreshes: int = 0,
     stragglers: int = 0,
     retired_workers: int = 0,
+    replay: Optional[Dict[str, Any]] = None,
     usage: Usage = <factory>,
     wallclock: float = 0.0,
     rollouts: int = 0,
@@ -1038,7 +1039,9 @@ Aggregator(
 | `finalize() -> None` | Publish the current dev head to stable at the end of a clean run. |
 | `finish_step(items: List[Union['_Candidate', MergeReport]]) -> List[MergeReport]` | Phase 3: decide the measured candidates, then age and promote. |
 | `measure(items: List[Union['_Candidate', MergeReport]]) -> List[Union['_Candidate', MergeReport]]` | Phase 2 for a batch from `begin_step`. **Off-thread safe.** |
+| `recent_settled(task: Any, n: int = 3) -> List[EvidenceCard]` | The most recent N settled cards whose trajectory_refs mention task. |
 | `restore(state: dict) -> None` | Restore state written by `checkpoint`. |
+| `set_settled_consumer(consumer: Optional[Callable[[EvidenceCard], None]]) -> None` | Subscribe a consumer to the settled-evidence pool. |
 | `step() -> List[MergeReport]` | Fire every artifact bucket that is ready and return per-artifact reports. |
 
 ### `AggregatorConfig(...)`
@@ -1077,6 +1080,8 @@ Cards bucketed by target artifact (design doc, section 4.1).
 
 | method | what it does |
 |---|---|
+| `recent_settled(task: Any, n: int = 3) -> List[EvidenceCard]` | The most recent N settled cards whose `trajectory_refs` mention `task`. |
+| `set_settled_consumer(consumer: Optional[Callable[[EvidenceCard], None]]) -> None` | Subscribe a consumer to every card that is settled. |
 | `settle(cards: List[EvidenceCard]) -> None` | Keep discarded-diff evidence addressable, under a hard bound. |
 
 ### `MergeOutcome`
diff --git a/docs/sampling.md b/docs/sampling.md
index f580f655..c8e06294 100644
--- a/docs/sampling.md
+++ b/docs/sampling.md
@@ -91,6 +91,56 @@ engine's [`SOLVED`](evolution.md) (0.999) rather than repeating the literal.
     `DifficultyWeighted(pass_threshold=0.8)`, or the two disagree about what a
     pass is.
 
+## `ReplaySampler` — also learn from discarded evidence
+
+The engine's aggregator discards diffs — stale, oversized, or lost to a CAS
+race — and settles their evidence cards into a bounded ring
+(`Aggregator.settle`). Nothing reads that ring back. `ReplaySampler` is the
+consumer: it subscribes to the pool via `aggregator.set_settled_consumer` and
+up-weights tasks whose recent proposals were thrown away.
+
+The argument is that a stale/conflicted waste is a **different signal from a
+failing rollout** — pass rate answers "is this task hard?", while a high
+settled count says "this task is being out-competed by the parallel scheduler,
+not because it is easy." The two signals are orthogonal.
+
+```python
+from agentdescent import ReplaySampler
+
+evolve(tasks, reward, agent=agent, task_sampler=ReplaySampler(temperature=0.3))
+```
+
+The base score is `DifficultyWeighted`'s own UCB over pass rate; the replay
+bonus is additive and saturating:
+
+| Parameter | Default | Meaning |
+|---|---|---|
+| `temperature` | 0.0 | Scales the replay bonus. `0.0` = the base sampler identically. |
+| `capped_at` | 10 | A task with more than this many recent settled cards gets the same bonus. |
+
+`temperature=0.0` is the default because the feature is experimental — an A/B
+that fails should be recorded as a negative result, not shipped as a default.
+
+### Wiring
+
+The engine wires the aggregator automatically when it detects a sampler with a
+`settle()` method. A second wire fills `ProposalContext.rejected` with the most
+recent settled card(s) for the task being rolled out, so a proposal policy
+that reads its context can see what was just discarded:
+
+```python
+def propose(ctx: ProposalContext):
+    for card in ctx.rejected:
+        # card.diff.ops contains the paths/code that were rejected
+        # card.base_version is the version they were proposed against
+        ...
+```
+
+Without a wired aggregator the counts never increase, `ctx.rejected` is always
+empty, and the sampler behaves identically to `DifficultyWeighted(temperature=0)`.
+
+## EvolvingArtifact`](api.md#task-sampling)
+
 ## Writing your own
 
 ```python
diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py
index 8662d0ab..a169235e 100644
--- a/tests/test_aggregator.py
+++ b/tests/test_aggregator.py
@@ -223,3 +223,35 @@ def test_evidence_eval_reads_the_task_objects_on_the_card():
     assert skill.evidence_eval(scored) == 1.0
     assert skill.evidence_eval(unscorable) == 0.0, (
         "ids are not scorable, which is why the field takes task objects")
+
+
+def test_settled_consumer_is_called_for_every_discarded_card(tmp_path):
+    """The optional consumer receives every card the pool settles."""
+    from agentdescent.aggregator import Aggregator, AggregatorConfig
+    from agentdescent.evolvable import Diff, EvidenceCard
+    from agentdescent.evolution import AppendRules, EvolvingArtifact
+    from agentdescent.ledger import Ledger
+    from agentdescent.scheduler import AuditScheduler
+    from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget
+
+    lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)},
+                lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v,
+                                                   0.2, None, AppendRules()))
+    lg.register(EvolvingArtifact("a", {}, 1, 0.2, None, AppendRules()))
+    verifier = ThreeLayerVerifier(eval_fn=lambda art, tasks: 0.5, held_out=[1, 2, 3],
+                                  budget=VerifierBudget())
+    agg = Aggregator(lg, verifier, AuditScheduler(),
+                     AggregatorConfig(batch_trigger=1, trust_region_ops=2))
+
+    seen = []
+    agg.set_settled_consumer(lambda card: seen.append(card.diff.diff_id))
+
+    diff = Diff(diff_id="oversized", target="a",
+                ops={f"k{i}": "v" for i in range(10)}, author="w0")
+    agg.ingest(EvidenceCard(diff=diff, base_version={"a": 1}, touched=["a"],
+                            before_after_delta=0.1, trajectory_refs=[]))
+    assert not seen, "consumer should not be called before step()"
+
+    agg.step()
+    assert "oversized" in seen, "consumer should receive the settled card"
+    assert len(seen) == 1, "one oversized card should trigger one consumer call"
diff --git a/tests/test_sampling.py b/tests/test_sampling.py
index 25160b04..d7be2e0b 100644
--- a/tests/test_sampling.py
+++ b/tests/test_sampling.py
@@ -191,3 +191,330 @@ def score(i):
 
     select_hard(list(range(16)), score, concurrency=8)
     assert peak[0] > 1, "select_hard scored the pool serially"
+
+
+# ---------------------------------------------------------------------------
+# ReplaySampler: the settled-evidence consumer
+# ---------------------------------------------------------------------------
+
+def _card(task_id, ops=None, n_refs=1):
+    from agentdescent.evolvable import Diff, EvidenceCard
+    refs = [Task(id=task_id, prompt=f"p-{task_id}") for _ in range(n_refs)]
+    return EvidenceCard(
+        diff=Diff(diff_id=f"d-{task_id}", target="a", ops=ops or {task_id: "v"}),
+        base_version={"a": 1}, touched=["a"], before_after_delta=0.0,
+        trajectory_refs=refs)
+
+
+def test_replay_sampler_satisfies_the_protocol():
+    from agentdescent.sampling import ReplaySampler
+    assert isinstance(ReplaySampler(), TaskSampler)
+
+
+def test_replay_sampler_at_zero_temperature_is_difficulty_weighted():
+    """temperature=0 means no replay bonus: byte-for-byte the base sampler."""
+    from agentdescent.sampling import ReplaySampler
+    base = DifficultyWeighted()
+    replay = ReplaySampler(temperature=0.0)
+    keys = ["a", "b", "c", "d"]
+    for i in range(12):
+        base.record("a", 1.0), replay.record("a", 1.0)
+        assert replay.pick(keys, i) == base.pick(keys, i)
+
+
+def test_settled_cards_raise_the_tasks_score():
+    from agentdescent.sampling import ReplaySampler
+    replay = ReplaySampler(temperature=0.5)
+    # All tasks start at the optimistic prior and equal UCB; the one with a
+    # settled card gets picked because its replay bonus dominates.
+    replay.settle(_card("a"))
+    keys = ["a", "b"]
+    assert replay.pick(keys, 0) == "a"
+
+
+def test_replay_bonus_is_capped():
+    from agentdescent.sampling import ReplaySampler
+    replay = ReplaySampler(temperature=0.5, capped_at=3)
+    keys = ["a", "b", "c"]
+    for _ in range(10):
+        replay.settle(_card("a"))
+    replay.settle(_card("b"))
+    replay.settle(_card("b"))
+    # 10 settled vs 2 settled: same bonus after the cap.
+    assert replay._settled_count["a"] == 10
+    assert replay._settled_count["b"] == 2
+    # a still wins because its bonus is capped at the same value but it got there first
+    assert replay.pick(keys, 0) == "a"
+
+
+def test_replay_without_a_wired_pool_never_moves():
+    from agentdescent.sampling import ReplaySampler
+    replay = ReplaySampler(temperature=0.5)
+    assert replay.settled_counts() == {}
+    # settle() never called -> counts stay empty -> bonus stays zero
+    keys = ["a", "b"]
+    base = DifficultyWeighted()
+    assert replay.pick(keys, 0) == base.pick(keys, 0)
+
+
+def test_replay_settled_counts_are_inspectable():
+    from agentdescent.sampling import ReplaySampler
+    replay = ReplaySampler(temperature=0.3)
+    replay.settle(_card("a"))
+    replay.settle(_card("b"))
+    replay.settle(_card("a"))
+    assert replay.settled_counts() == {"a": 2, "b": 1}
+
+
+def test_replay_records_through_to_the_base_sampler():
+    from agentdescent.sampling import ReplaySampler
+    replay = ReplaySampler()
+    replay.record("a", 1.0)
+    assert replay.stats() == {"a": (1.0, 1.0)}
+
+# ---------------------------------------------------------------------------
+# ReplaySampler + ProposalContext integration
+# ---------------------------------------------------------------------------
+
+from agentdescent.evolution import AppendRules, Task
+
+
+def test_replay_settle_without_task_ids_is_safe():
+    from agentdescent.sampling import ReplaySampler
+    from agentdescent.evolvable import Diff, EvidenceCard
+    replay = ReplaySampler()
+    card = EvidenceCard(
+        diff=Diff(diff_id="d", target="a", ops={"x": "v"}),
+        base_version={"a": 1}, touched=["a"], before_after_delta=0.0,
+        trajectory_refs=["not-a-task"])  # strings have no .id
+    replay.settle(card)
+    assert replay.settled_counts() == {}
+
+
+def test_proposal_context_has_rejected_field():
+    from agentdescent.policies import ProposalContext
+    ctx = ProposalContext(rendered="", task=Task(id="t", prompt="p"),
+                          output="", reward=0.0)
+    assert ctx.rejected == ()
+
+# ---------------------------------------------------------------------------
+# ReplayAwareProposal: the consumer of ctx.rejected
+# ---------------------------------------------------------------------------
+
+from dataclasses import replace, dataclass
+
+
+def test_replay_aware_policy_appends_rejection_to_output():
+    from agentdescent.replay import ReplayAwareProposal
+    from agentdescent.evolvable import Diff, EvidenceCard
+    from agentdescent.policies import ProposalContext
+    from agentdescent.evolution import Task
+
+    seen = []
+
+    class CapturePolicy:
+        def propose(self, ctx):
+            seen.append(ctx.output)
+            return []
+
+    policy = ReplayAwareProposal(CapturePolicy())
+    card = EvidenceCard(
+        diff=Diff(diff_id="d1", target="a", ops={"src/x.py": "x"}, author="w"),
+        base_version={"a": 2}, touched=["a"], before_after_delta=0.1,
+        trajectory_refs=[Task(id="t0", prompt="p")])
+    ctx = ProposalContext(rendered="", task=Task(id="t0", prompt="p"),
+                          output="old", reward=0.0, rejected=(card,))
+    policy.propose(ctx)
+    assert "## Recently discarded" in seen[0]
+    assert "src/x.py" in seen[0]
+    assert policy.read == 1 and policy.shown == 1
+
+
+def test_replay_aware_policy_passes_through_when_empty():
+    from agentdescent.replay import ReplayAwareProposal
+    from agentdescent.policies import ProposalContext
+    from agentdescent.evolution import Task
+
+    seen = []
+
+    class CapturePolicy:
+        def propose(self, ctx):
+            seen.append(ctx.output)
+            return []
+
+    policy = ReplayAwareProposal(CapturePolicy())
+    ctx = ProposalContext(rendered="", task=Task(id="t0", prompt="p"),
+                          output="old", reward=0.0)
+    policy.propose(ctx)
+    assert seen == ["old"]          # output unchanged
+    assert policy.read == 0
+
+
+def test_replay_aware_policy_caps_max_cards():
+    from agentdescent.replay import ReplayAwareProposal
+    from agentdescent.evolvable import Diff, EvidenceCard
+    from agentdescent.policies import ProposalContext
+    from agentdescent.evolution import Task
+
+    policy = ReplayAwareProposal(
+        max_cards=2,
+        policy=type("", (), {"propose": lambda s, ctx: []})())
+
+    cards = tuple(
+        EvidenceCard(diff=Diff(diff_id=f"d{i}", target="a", ops={str(i): "v"}, author="w"),
+                     base_version={"a": 1}, touched=["a"], before_after_delta=0.0,
+                     trajectory_refs=[Task(id="t0", prompt="p")])
+        for i in range(5))
+    ctx = ProposalContext(rendered="", task=Task(id="t0", prompt="p"),
+                          output="o", reward=0.0, rejected=cards)
+    policy.propose(ctx)
+    assert policy.shown == 2         # capped, not 5
+
+
+def test_default_note_is_readable():
+    from agentdescent.replay import _default_note
+    from agentdescent.evolvable import Diff, EvidenceCard
+    card = EvidenceCard(
+        diff=Diff(diff_id="d", target="a", ops={"src/x.py": "def f(): pass\n"}, author="w"),
+        base_version={"a": 3}, touched=["a"], before_after_delta=0.25,
+        trajectory_refs=[])
+    note = _default_note(card)
+    assert "src/x.py" in note
+    assert "{'a': 3}" in note
+    assert "+0.250" in note
+
+# ---------------------------------------------------------------------------
+# Integration: ReplaySampler + aggregator (design doc, test 6)
+# ---------------------------------------------------------------------------
+
+def test_replay_sampler_and_aggregator_integration():
+    """Build a buffer, settle cards, verify the sampler sees them."""
+    from agentdescent.sampling import ReplaySampler
+    from agentdescent.aggregator import EvidenceBuffer
+    from agentdescent.evolvable import Diff, EvidenceCard
+    from agentdescent.evolution import Task
+
+    buf = EvidenceBuffer()
+    sampler = ReplaySampler(temperature=0.5)
+    buf.set_settled_consumer(sampler.settle)
+
+    tasks = [Task(id=f"t{i}", prompt="p") for i in range(3)]
+    cards = [
+        EvidenceCard(diff=Diff(diff_id=f"d{i}", target="a",
+                               ops={f"src/{i}.py": "v"}, author="w"),
+                     base_version={"a": 1}, touched=["a"],
+                     before_after_delta=0.0,
+                     trajectory_refs=[tasks[i]]) for i in range(3)]
+    buf.settle(cards)
+
+    assert sampler.settled_counts() == {"t0": 1, "t1": 1, "t2": 1}
+    keys = ["t0", "t1"]
+    # With temperature=0, base would pick randomly; with replay bonus,
+    # both t0 and t1 have the same bonus (1 each), plus UCB exploration.
+    picked = sampler.pick(keys, 0)
+    assert picked in keys
+
+
+def test_replay_affected_count_works():
+    from agentdescent.sampling import ReplaySampler
+    from agentdescent.evolvable import Diff, EvidenceCard
+    from agentdescent.evolution import Task
+
+    sampler = ReplaySampler(temperature=1.0)
+    card = EvidenceCard(
+        diff=Diff(diff_id="d", target="a", ops={"x": "v"}, author="w"),
+        base_version={"a": 1}, touched=["a"], before_after_delta=0.0,
+        trajectory_refs=[Task(id="t-a", prompt="p")])
+    sampler.settle(card)
+    keys = ["t-a", "t-b"]
+    sampler.pick(keys, 0)
+    sampler.pick(keys, 1)
+    assert sampler.picks == 2
+    assert sampler.replay_affected >= 0  # may be 0 if base already prefers t-a
+
+
+# ---------------------------------------------------------------------------
+# End-to-end through evolve(): the full wire (design doc, "integration")
+# ---------------------------------------------------------------------------
+
+from agentdescent.aggregator import AggregatorConfig
+from agentdescent.policies import Policies, ProposalContext
+from agentdescent.sampling import ReplaySampler
+from agentdescent.replay import ReplayAwareProposal
+from agentdescent.replay import ReplayAwareProposal
+from agentdescent.strategies import Strategy
+from agentdescent.evolvable import Diff
+
+
+class _Oversize(Strategy):
+    """Every proposal becomes a 5-op diff, over a 3-op trust region: oversized."""
+
+    def keys(self):
+        return ["a", "b", "c", "d", "e"]
+
+    def initial(self):
+        return {k: "" for k in self.keys()}
+
+    def render(self, state):
+        return repr(state)
+
+    def to_diff(self, state, proposal, author, base_version, target):
+        return Diff(diff_id=f"{author}:5", target=target,
+                    ops={k: proposal for k in self.keys()}, author=author)
+
+    def frozen_files(self, source):
+        return {}
+
+
+class _RejectAware:
+    """Records whether the proposal context carried rejected evidence."""
+
+    def __init__(self):
+        self.seen_rejected = 0
+
+    def propose(self, ctx: ProposalContext):
+        self.seen_rejected += int(bool(ctx.rejected))
+        return ["x"]
+
+
+def test_full_wire_settled_to_proposer_and_sampler():
+    """Oversized diffs are settled; the sampler sees them, and the proposer sees
+    the rejections via ProposalContext.rejected -- all through evolve()."""
+    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(6)]
+
+    def reward(task, out):
+        return 0.0                       # never solved: always propose
+
+    dummy = _RejectAware()
+    sampler = ReplaySampler(temperature=0.3)
+    result = evolve(
+        tasks, reward,
+        run=lambda rendered, task: "answer",
+        propose=lambda rendered, task, output, reward: None,
+        strategy=_Oversize(),
+        policies=Policies(proposal=ReplayAwareProposal(dummy),
+                          task_sampler=sampler),
+        agg_config=AggregatorConfig(trust_region_ops=3, batch_trigger=1),
+        max_rollouts=48, n_workers=1, max_concurrency=1, seed=1, rounds=8)
+
+    assert sum(sampler.settled_counts().values()) > 0, "the pool must have settled"
+    assert sampler.picks > 0
+    assert result.replay is not None and result.replay["settled"] > 0
+    # Over enough rollouts, at least one proposal sees a prior rejection.
+    assert dummy.seen_rejected >= 1, "a proposer must eventually see a rejection"
+
+
+def test_result_replay_is_none_without_a_replay_sampler():
+    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(6)]
+
+    def reward(task, out):
+        return 0.0
+
+    result = evolve(
+        tasks, reward,
+        run=lambda rendered, task: "answer",
+        propose=lambda rendered, task, output, reward: None,
+        strategy=_Oversize(),
+        agg_config=AggregatorConfig(trust_region_ops=3, batch_trigger=1),
+        max_rollouts=12, n_workers=1, max_concurrency=1, seed=1, rounds=3)
+    assert result.replay is None

From ae17142ef67160dee571db5ee501760dc8af8cb1 Mon Sep 17 00:00:00 2001
From: xuke <xuke127@huawei.com>
Date: Fri, 18 Sep 2026 11:36:40 +0800
Subject: [PATCH 2/3] fix(bench): ab_replay uses correct baselines API

compare() takes ArmResult, not Workload. _run_arm builds ArmResult
with distinct arm names (control/replay) so compare() groups them
correctly. Verified end-to-end with --budget-rollouts 4.
---
 bench/ab_replay.py | 106 +++++++++++++++++++++++++++++++++------------
 1 file changed, 79 insertions(+), 27 deletions(-)

diff --git a/bench/ab_replay.py b/bench/ab_replay.py
index ae013698..9836ad32 100644
--- a/bench/ab_replay.py
+++ b/bench/ab_replay.py
@@ -1,21 +1,62 @@
 """A/B: DifficultyWeighted vs ReplaySampler at equal budget.
 
     python -m bench.ab_replay --budget-rollouts 96 --seeds 0,1,2 \\
-        --domain stackvm --yes
+        --workers 4 --yes
 
 Both arms share the same workload, strategy and seed. The only difference is
 ``task_sampler=DifficultyWeighted`` vs ``task_sampler=ReplaySampler(temperature=0.3)``.
+The hypothesis: ReplaySampler recovers some of the rollouts wasted on stale/
+conflicted tasks by steering the search away from them.
 """
 
 from __future__ import annotations
 import sys
-from agentdescent.evolution import Task
+import threading
+
+from agentdescent.agents import Usage
 from agentdescent.baselines import Budget, Workload, compare
+from agentdescent.evolution import Task
+from agentdescent.strategies import AppendRules
+from agentdescent.agents import echo
+from agentdescent.evolution import LLMAgent
+
+
+def _workload(task_sampler) -> Workload:
+    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(16)]
+    return Workload(
+        tasks=tasks,
+        reward=lambda task, out: 1.0 if out and len(out) > 3 else 0.0,
+        test_eval=lambda r: r.final_reward or 0.0,
+        agent=LLMAgent(echo()),
+        strategy=AppendRules(),
+        evolve_kwargs=dict(rounds=10_000, task_sampler=task_sampler),
+    )
+
+
+def _run_arm(label, sampler, seed, budget, n_workers):
+    """Run one arm and build an ArmResult with a distinct name."""
+    from agentdescent.baselines import ArmResult
+    usage = Usage()
+    wl = _workload(sampler)
+    result = wl._evolve(seed=seed, n_workers=n_workers, budget=budget, usage=usage)
+    from agentdescent.baselines import ArmResult
+    return ArmResult(
+        arm=label, seed=seed, width=n_workers,
+        rollouts=result.usage.get("rollouts", budget.rollouts) if isinstance(result.usage, dict) else budget.rollouts,
+        calls=usage.calls,
+        prompt_tokens=usage.prompt_tokens,
+        completion_tokens=usage.completion_tokens,
+        wallclock=usage.seconds,
+        wallclock_parallel=usage.seconds,
+        dev_reward=result.final_reward,
+        test_reward=wl.test_eval(result),
+        stop_reason=result.stop_reason,
+    )
 
 
 def main(argv=None) -> int:
     import argparse
-    parser = argparse.ArgumentParser()
+    parser = argparse.ArgumentParser(description="A/B: replay sampler")
     parser.add_argument("--budget-rollouts", type=int, default=96)
     parser.add_argument("--seeds", default="0,1,2")
     parser.add_argument("--workers", type=int, default=4)
@@ -24,25 +65,7 @@ def main(argv=None) -> int:
     seeds = [int(s) for s in args.seeds.split(",") if s.strip()]
     budget = Budget(rollouts=args.budget_rollouts)
 
-    # Use the deterministic comment domain (offline, no model).
-    from agentdescent.strategies import AppendRules
-    from agentdescent.evolution import reflector, LLMAgent
-    from agentdescent.agents import echo
-
-    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(16)]
-
-    def reward(task, out):
-        return 1.0 if out and len(out) > 3 else 0.0
-
-    agent = LLMAgent(echo())
-    workload = Workload(
-        tasks=tasks, reward=reward, test_eval=lambda r: r.final_reward or 0.0,
-        agent=agent, strategy=AppendRules(),
-        evolve_kwargs=dict(rounds=10_000))
-
     from agentdescent.sampling import DifficultyWeighted, ReplaySampler
-    kwargs_control = dict(task_sampler=DifficultyWeighted())
-    kwargs_experiment = dict(task_sampler=ReplaySampler(temperature=0.3))
 
     print(f"\nA/B (replay): {args.budget_rollouts} max rollouts, "
           f"{len(seeds)} seeds, {args.workers} workers\n")
@@ -50,12 +73,41 @@ def reward(task, out):
     print(f"  experiment : ReplaySampler(temperature=0.3)")
     print()
 
-    results = compare(
-        [workload],
-        {"control": kwargs_control, "replay": kwargs_experiment},
-        budget=budget, seeds=seeds, n_workers=args.workers,
-    )
-    print("\nDone.\n")
+    results = []
+    printing = threading.Lock()
+
+    def _one(job):
+        seed, label, sampler = job
+        try:
+            arm = _run_arm(label, sampler, seed, budget, args.workers)
+        except Exception as e:  # noqa: BLE001
+            with printing:
+                print(f"  {label} seed={seed}  FAILED {type(e).__name__}: {str(e)[:120]}")
+            return None
+        with printing:
+            print(f"  {label} seed={seed}  reward={arm.test_reward}  "
+                  f"rollouts={arm.rollouts}  calls={arm.calls}")
+        return arm
+
+    jobs = [(s, "control" if i < len(seeds) else "replay",
+             DifficultyWeighted() if i < len(seeds) else ReplaySampler(temperature=0.3))
+            for i, s in enumerate(seeds * 2)]
+    # interleave control and replay per seed
+    jobs = []
+    for s in seeds:
+        jobs.append((s, "control", DifficultyWeighted()))
+        jobs.append((s, "replay", ReplaySampler(temperature=0.3)))
+
+    for job in jobs:
+        r = _one(job)
+        if r is not None:
+            results.append(r)
+
+    if len(results) >= 2:
+        comp = compare(results)
+        print(f"\n{comp}\n")
+    else:
+        print("\nNot enough results to compare.\n")
     return 0
 
 

From b4285a1f3037eda902b1099641f9b502145d5978 Mon Sep 17 00:00:00 2001
From: xuke <xuke127@huawei.com>
Date: Mon, 28 Sep 2026 15:07:37 +0800
Subject: [PATCH 3/3] fix: _replay_stats uses isinstance not hasattr (PR #193
 review)

hasattr(sampler, 'settle') matched any custom sampler that also exposed
settle but not settled_counts, crashing on the result line of evolve/
async_evolve. Switched to isinstance(sampler, ReplaySampler).

Regression test: a custom sampler with settle() but no settled_counts()
returns result.replay = None instead of AttributeError.
---
 agentdescent/evolution.py | 15 +++++++++------
 tests/test_sampling.py    | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 6 deletions(-)

diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py
index 0fc4a6cc..f0c7c5f7 100644
--- a/agentdescent/evolution.py
+++ b/agentdescent/evolution.py
@@ -960,16 +960,19 @@ def _set_budget_remaining(aggregator: Any, governor: BudgetGovernor) -> None:
 def _replay_stats(sampler) -> Optional[Dict[str, Any]]:
     """The replay-sampler statistics for :class:`EvolutionResult.replay`.
 
-    ``None`` for any sampler that is not a ``ReplaySampler`` -- the field exists
-    so a run that wired the mechanism reports it, and a run that did not is
-    indistinguishable from one that did but never fired.
+    ``None`` for any sampler that is not a :class:`~agentdescent.sampling.
+    ReplaySampler`. Using ``hasattr(sampler, 'settle')`` instead of an
+    isinstance check meant any custom sampler that also exposed a ``settle``
+    method (but not ``settled_counts``) crashed on the result line of
+    :func:`evolve` / :func:`async_evolve`.
     """
-    if not hasattr(sampler, "settle"):
+    from .sampling import ReplaySampler
+    if not isinstance(sampler, ReplaySampler):
         return None
     return {
         "settled": sum(sampler.settled_counts().values()),
-        "picks": getattr(sampler, "picks", 0),
-        "replay_affected": getattr(sampler, "replay_affected", 0),
+        "picks": sampler.picks,
+        "replay_affected": sampler.replay_affected,
     }
 
 
diff --git a/tests/test_sampling.py b/tests/test_sampling.py
index d7be2e0b..57bec3cc 100644
--- a/tests/test_sampling.py
+++ b/tests/test_sampling.py
@@ -518,3 +518,36 @@ def reward(task, out):
         agg_config=AggregatorConfig(trust_region_ops=3, batch_trigger=1),
         max_rollouts=12, n_workers=1, max_concurrency=1, seed=1, rounds=3)
     assert result.replay is None
+
+
+# ---------------------------------------------------------------------------
+# Regression: _replay_stats must not crash on a custom sampler with settle()
+# but without settled_counts (PR #193 review)
+# ---------------------------------------------------------------------------
+
+def test_replay_stats_is_none_for_a_custom_sampler_with_settle_only():
+    """A custom TaskSampler that exposes settle() but not settled_counts()
+    must not crash _replay_stats. isinstance check, not hasattr."""
+    from agentdescent.evolution import evolve
+    from agentdescent.strategies import AppendRules
+
+    class CustomWithSettle:
+        """Has settle() but is not a ReplaySampler — must be left alone."""
+        name = "custom"
+        def pick(self, keys, round_index):
+            return keys[round_index % len(keys)]
+        def record(self, task_id, score):
+            pass
+        def settle(self, card):
+            pass
+
+    tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(6)]
+    result = evolve(
+        tasks, lambda t, o: 0.0,
+        run=lambda rd, t: "answer",
+        propose=lambda rd, t, o, r: None,
+        strategy=AppendRules(),
+        task_sampler=CustomWithSettle(),
+        max_rollouts=4, n_workers=1, rounds=2)
+    assert result.replay is None, (
+        f"a non-ReplaySampler with settle() must report None, got {result.replay}")