feat(replay): settled-evidence pool gets a consumer - ReplaySampler + ProposalContext.rejected

CHANGELOG.md

@@ -8,6 +8,15 @@ 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.
 - **`Investigator` — a read-only delegation whose findings persist into the record.**
   The Genesis reference implementation has an `:read` investigator role
   (`agents/investigator.ex`) that a manager hands a question and that may write only

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,

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,13 +412,44 @@ 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
             ):
                 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.
 

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

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,25 @@ 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 :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`.
+    """
+    from .sampling import ReplaySampler
+    if not isinstance(sampler, ReplaySampler):
+        return None
+    return {
+        "settled": sum(sampler.settled_counts().values()),
+        "picks": sampler.picks,
+        "replay_affected": sampler.replay_affected,
+    }
+
+
 def _cost_fields(meter: Meter) -> Dict[str, Any]:
     """The meter's counters, keyed as :class:`EvolutionResult` fields.
 
@@ -1251,6 +1280,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 +1884,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 +2153,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 +2810,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 +3306,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

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",

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"] = ()
 
 
 # ---------------------------------------------------------------------------

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)