feat(rrsi): wire the regularized RRSI mechanism into evolve() through the Policies seams

agentdescent/__init__.py

@@ -118,6 +118,7 @@
     Sandbox,
     SandboxProvider,
     SandboxSpec,
+    ScreenPolicy,
     VerifierProtocol,
 )
 from .evalcache import CacheProtocol, FileCache, MemoryCache

agentdescent/aggregator.py

@@ -54,7 +54,7 @@
 from .metrics import Meter
 from .policies import (
     AcceptancePolicy, ConflictPolicy, FusionPolicy, MergeContext,
-    PromotionPolicy,
+    PromotionPolicy, ScreenPolicy,
 )
 from .scheduler import AuditScheduler
 from .staleness import StaleAction, StalenessPolicy, get_policy
@@ -229,6 +229,18 @@ class MergeOutcome(str, Enum):
     #: so a policy-violating proposal never buys a held-out sweep. The
     #: reflector (or the contract) is the thing to look at.
     VIOLATES_INVARIANT = "violates-invariant"
+    #: A pre-evaluation screen (see
+    #: :class:`~agentdescent.policies.ScreenPolicy`) rejected the card before
+    #: it was measured -- leakage, an over-budget edit bundle, inert machinery.
+    #: The reason is on the report; the card was settled without a held-out
+    #: sweep, so no invented measurement ever justified it.
+    CRITIC_REJECTED = "critic-rejected"
+    #: An RRSI-style selector refused the candidate at its historical-best
+    #: noise floor (``score < S* - delta``).
+    NOISE_FLOOR = "noise-floor"
+    #: An RRSI-style selector refused the candidate on its cost rule (either
+    #: branch): the improvement costs more policy tokens than it is worth.
+    COST_RULE = "cost-rule"
 
     def __str__(self) -> str:            # so f-strings render the value, not the name
         return self.value
@@ -665,6 +677,7 @@ def __init__(
         fusion: Optional["FusionPolicy"] = None,
         acceptance: Optional["AcceptancePolicy"] = None,
         promotion: Optional["PromotionPolicy"] = None,
+        screen: Optional["ScreenPolicy"] = None,
     ) -> None:
         self.ledger = ledger
         self.verifier = verifier
@@ -689,6 +702,11 @@ def __init__(
             verifier, tournament=cfg.fusion_tournament)
         self.acceptance_policy = acceptance or DefaultAcceptance.from_config(cfg)
         self.promotion_policy = promotion or DefaultPromotion.from_config(cfg)
+        #: Pre-evaluation screen on evidence cards (see
+        #: :class:`~agentdescent.policies.ScreenPolicy`). ``None`` keeps the
+        #: old behaviour exactly -- every card that survives the trust region
+        #: and staleness filter proceeds to the gate.
+        self.screen_policy = screen
         # Two things a policy may need are built in here and nowhere a caller
         # can reach: the verifier (for anything that ranks) and the config (for
         # anything that reads a threshold). So every installed policy is offered
@@ -1056,10 +1074,21 @@ def checkpoint(self) -> Optional[dict]:
             }
         except AttributeError:
             return None
+        # An acceptance policy that keeps its own state (an RRSI-style selector
+        # holds the historical-best floor, the per-edit ledger and the annealed
+        # round counter) opts into checkpointing through an optional hook. A
+        # policy without one is simply absent from the payload -- the same way
+        # the posterior is absent from an aggregator that never had one.
+        state = None
+        hook = getattr(getattr(self, "acceptance_policy", None),
+                       "checkpoint_state", None)
+        if callable(hook):
+            state = hook()
         return {
             "posteriors": posteriors,
             "promoted_at": dict(self._promoted_at),
             "seen": sorted(self._seen),
+            "acceptance_state": state,
         }
 
     def restore(self, state: dict) -> None:
@@ -1092,6 +1121,15 @@ def restore(self, state: dict) -> None:
         seen = state.get("seen")
         if isinstance(seen, list):
             self._seen = set(seen)
+        acc = state.get("acceptance_state")
+        if acc is not None:
+            hook = getattr(getattr(self, "acceptance_policy", None),
+                           "restore_state", None)
+            if callable(hook):
+                try:
+                    hook(acc)
+                except Exception:  # noqa: BLE001 - a bad payload must not hide the artifact
+                    pass
 
     @property
     def _survival(self) -> Dict[str, int]:
@@ -1359,6 +1397,47 @@ def _prepare(self, artifact_id: str, branch: str = "", cards=None) -> "Union[_Ca
         region = self._trust_region()
         cards, oversized = self._apply_trust_region(cards, region)
 
+        # Pre-evaluation screen (RRSI critic and friends): runs here, before
+        # staleness and certainly before the expensive measurement, so a card
+        # the screen rejects is settled without buying a held-out sweep and
+        # without an invented measurement. The screen is a *card* gate -- it
+        # names which card failed and why -- and a batch that is only rejected
+        # cards reports ``critic-rejected`` with the reasons.
+        if self.screen_policy is not None:
+            kept: List[EvidenceCard] = []
+            rejected_screen: List[Tuple[EvidenceCard, str]] = []
+            for card in cards:
+                if self.meter is not None:
+                    self.meter.add("critic_checked")
+                reason = self.screen_policy.screen(card)
+                if reason is None:
+                    kept.append(card)
+                else:
+                    rejected_screen.append((card, reason))
+            if rejected_screen:
+                if self.meter is not None:
+                    self.meter.add("critic_rejected", len(rejected_screen))
+                self.buffer.settle([c for c, _ in rejected_screen])
+                if not kept:
+                    # A screen rejection is still a merge decision: an annealed
+                    # budget and a stall detector must advance on it, or a search
+                    # that only ever produced screen-rejected bundles would stay
+                    # pinned at b_max and never enter structured exploration
+                    # (RRSI review). The acceptance policy opts in through the
+                    # same per-merge accounting it uses on an accept/reject.
+                    on_reject = getattr(self.acceptance_policy,
+                                        "on_screen_reject", None)
+                    if callable(on_reject):
+                        on_reject()
+                    detail = "; ".join(
+                        f"{reason} (task {card.trajectory_refs[0].id if card.trajectory_refs else '?'})"
+                        for card, reason in rejected_screen)
+                    return MergeReport(artifact_id, None, False, n_considered, 0,
+                                       len(rejected_screen), 0, 0.0, None,
+                                       detail, MergeOutcome.CRITIC_REJECTED,
+                                       oversized=len(oversized))
+            cards = kept
+
         head = snap.version
         survivors, discarded = self._staleness_filter(artifact, head, cards)
         self.buffer.settle(discarded)
@@ -1403,6 +1482,14 @@ def _prepare(self, artifact_id: str, branch: str = "", cards=None) -> "Union[_Ca
             if self.meter is not None:
                 self.meter.add("invariant_rejections", len(violations))
             self.buffer.settle(survivors)
+            # Same per-merge accounting as the all-screen-rejected path: an
+            # annealed budget / stall detector must advance on a governance
+            # refusal too, or a streak of violates-invariant decisions freezes
+            # the RRSI schedule at b_max (review #254 P2).
+            on_reject = getattr(self.acceptance_policy,
+                                "on_screen_reject", None)
+            if callable(on_reject):
+                on_reject()
             return MergeReport(artifact_id, None, False, n_considered,
                                len(survivors), len(discarded), conflicts, 0.0,
                                None, "; ".join(violations),

agentdescent/evolution.py

@@ -833,7 +833,7 @@ def _publish_stable(aggregator) -> None:
 _WIRED_POLICIES = ("task_sampler", "selection", "proposal", "conflict", "fusion",
                    "acceptance", "promotion", "staleness", "verifier", "ledger",
                    "aggregator_factory", "eval_cache", "sandbox_spec", "evaluator",
-                   "executor")
+                   "executor", "screen")
 
 #: The same bundle, minus the one field the barrier-free loop does not read.
 #: `async_evolve`'s worker calls `eng.run` directly; there is no executor seam in
@@ -1110,6 +1110,8 @@ def _cost_fields(meter: Meter) -> Dict[str, Any]:
         "cas_conflicts": m.cas_conflicts,
         "invariant_checked": m.invariant_checked,
         "invariant_rejections": m.invariant_rejections,
+        "critic_checked": m.critic_checked,
+        "critic_rejected": m.critic_rejected,
         "cache_hits": m.cache_hits,
         "cache_misses": m.cache_misses,
         "trajectories_recorded": m.trajectories_recorded,
@@ -1467,6 +1469,11 @@ class EvolutionResult:
     #: Commits that lost a compare-and-swap race and rebased. Zero with one
     #: writer by construction.
     cas_conflicts: int = 0
+    #: Pre-evaluation screens: cards passed through a screen policy, and cards
+    #: it rejected (settled without a held-out sweep). A run with no screen
+    #: reports zeros -- the honest reading "no screen installed".
+    critic_checked: int = 0
+    critic_rejected: int = 0
     #: Evaluation cache. ``misses`` is the number of evaluations actually
     #: performed; the ratio is the duplicate-computation figure a multi-process
     #: run has to report.
@@ -2060,8 +2067,12 @@ def _build_engine(tasks, reward, *, agent, run, propose, strategy, initial_state
     if agent is not None:
         run = run or agent.solve
         propose = propose or agent.propose
-    if run is None or propose is None:
-        raise ValueError("provide agent=, or both run= and propose=")
+    if run is None or (propose is None and
+                       (policies_bundle is None or policies_bundle.proposal is None)):
+        raise ValueError(
+            "provide agent=, or both run= and propose=, or a proposal policy "
+            "through Policies(proposal=...): the engine needs a way to turn a "
+            "rollout into a candidate change.")
     if policies_bundle is not None and policies_bundle.proposal is not None:
         # 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
@@ -2365,7 +2376,8 @@ def deserialize(aid: str, version: int, state: dict) -> EvolvingArtifact:
     # behaviour. The task/proposal/eval-cache halves of `Policies` are read above
     # and are unaffected, so only the merge side is named.
     if aggregator_factory is not None:
-        dropped = [name for name in ("conflict", "fusion", "acceptance", "promotion")
+        dropped = [name for name in ("conflict", "fusion", "acceptance", "promotion",
+                                     "screen")
                    if getattr(pol, name, None) is not None]
         if dropped:
             warnings.warn(
@@ -2378,7 +2390,8 @@ def deserialize(aid: str, version: int, state: dict) -> EvolvingArtifact:
     def _default_aggregator(ledger, verifier, audit, config, policy):
         return Aggregator(ledger, verifier, audit, config, staleness_policy=policy,
                           meter=meter, conflict=pol.conflict, fusion=pol.fusion,
-                          acceptance=pol.acceptance, promotion=pol.promotion)
+                          acceptance=pol.acceptance, promotion=pol.promotion,
+                          screen=pol.screen)
 
     # A declared selection policy is what asks for a population layer -- the rule
     # the port runner already used, moved here so it is the engine's rule rather
@@ -2388,7 +2401,8 @@ def _default_aggregator(ledger, verifier, audit, config, policy):
         from .population import population_factory
         aggregator_factory = population_factory(
             selection, artifact_id, meter=meter, conflict=pol.conflict,
-            fusion=pol.fusion, acceptance=pol.acceptance, promotion=pol.promotion)
+            fusion=pol.fusion, acceptance=pol.acceptance, promotion=pol.promotion,
+            screen=pol.screen)
 
     # `None` defers to whatever the config already says, so `agg_config=` keeps
     # working and the two cannot silently disagree; an explicit True or False

agentdescent/metrics.py

@@ -85,6 +85,8 @@ class MeterSnapshot:
     trajectory_write_failures: int = 0
     invariant_checked: int = 0
     invariant_rejections: int = 0
+    critic_checked: int = 0
+    critic_rejected: int = 0
     sandbox_wait_s: float = 0.0
     sandbox_setup_s: float = 0.0
     sandboxes_created: int = 0
@@ -107,6 +109,7 @@ class MeterSnapshot:
     "trajectories_recorded", "trajectories_dedup",
     "trajectory_write_seconds", "trajectory_write_failures",
     "invariant_checked", "invariant_rejections",
+    "critic_checked", "critic_rejected",
     "sandbox_wait_s", "sandbox_setup_s", "sandboxes_created",
     "sandboxes_reused", "sandbox_failures", "env_mismatch",
 })
@@ -232,6 +235,11 @@ class Meter:
     #: reading "no rules to enforce", not "the gate is broken".
     invariant_checked: int = 0
     invariant_rejections: int = 0
+    #: Pre-evaluation screens: cards passed through the screen policy, and
+    #: cards it rejected (settled without a held-out sweep). A run with no
+    #: screen reports zeros, which is the honest reading "no screen installed".
+    critic_checked: int = 0
+    critic_rejected: int = 0
 
     #: Sandbox accounting. Zero on the default single-workspace path; filled once
     #: the sandbox pool exists. Kept here from the start so the result schema
@@ -327,6 +335,8 @@ def snapshot(self) -> MeterSnapshot:
                 trajectory_write_failures=self.trajectory_write_failures,
                 invariant_checked=self.invariant_checked,
                 invariant_rejections=self.invariant_rejections,
+                critic_checked=self.critic_checked,
+                critic_rejected=self.critic_rejected,
                 sandbox_wait_s=self.sandbox_wait_s,
                 sandbox_setup_s=self.sandbox_setup_s,
                 sandboxes_created=self.sandboxes_created,

agentdescent/policies.py

@@ -278,6 +278,25 @@ class ProposalPolicy(Protocol):
     def propose(self, ctx: ProposalContext) -> Sequence[str]: ...
 
 
+@runtime_checkable
+class ScreenPolicy(Protocol):
+    """A pre-evaluation gate on evidence cards.
+
+    Called by the aggregator **before** the expensive measurement, so a card
+    the screen rejects is settled and reported without ever buying a held-out
+    sweep. ``screen(card)`` returns ``None`` for a card that may proceed, or a
+    reason string for one that must not. The reason becomes the merge report's
+    detail and the outcome is ``critic-rejected``.
+
+    The seam exists for the RRSI pre-evaluation critic (leakage and edit-budget
+    screening, see ``examples/rrsi/``), and it is deliberately a *card* gate:
+    screening after a candidate is built is too late, and screening the whole
+    batch loses which card failed and why.
+    """
+
+    def screen(self, card: "EvidenceCard") -> Optional[str]: ...
+
+
 @runtime_checkable
 class ConflictPolicy(Protocol):
     """Which of a batch of mutually contradictory changes survive.
@@ -482,6 +501,12 @@ class Policies:
     #: worker, which is what the engine has always done.
     selection: Optional["SelectionPolicy"] = None
     proposal: Optional[ProposalPolicy] = None
+    #: A pre-evaluation gate on evidence cards (see
+    #: :class:`ScreenPolicy`). ``None`` keeps the old behaviour exactly --
+    #: every card that passes the trust region and staleness filter is merged.
+    #: Wired by the aggregator's ``_prepare``, so it runs before any
+    #: measurement.
+    screen: Optional[ScreenPolicy] = None
     conflict: Optional[ConflictPolicy] = None
     fusion: Optional[FusionPolicy] = None
     acceptance: Optional[AcceptancePolicy] = None

agentdescent/population.py

@@ -85,11 +85,12 @@ class PopulationAggregator(Aggregator):
 
     def __init__(self, ledger, verifier, audit, config, staleness_policy=None,
                  *, selection: SelectionPolicy, artifact_id: str, meter=None,
-                 conflict=None, fusion=None, acceptance=None, promotion=None):
+                 conflict=None, fusion=None, acceptance=None, promotion=None,
+                 screen=None):
         super().__init__(ledger, verifier, audit, config,
                          staleness_policy=staleness_policy, meter=meter,
                          conflict=conflict, fusion=fusion, acceptance=acceptance,
-                         promotion=promotion)
+                         promotion=promotion, screen=screen)
         self.selection = selection
         self.population_artifact = artifact_id
         self._archive: List[Dict[str, object]] = []
@@ -526,7 +527,7 @@ def restore(self, state: dict) -> None:
 
 def population_factory(selection: SelectionPolicy, artifact_id: str, *,
                        meter=None, conflict=None, fusion=None, acceptance=None,
-                       promotion=None):
+                       promotion=None, screen=None):
     """The ``aggregator_factory=`` adapter for one run.
 
     The factory path bypasses the engine's default-aggregator construction, so
@@ -541,6 +542,6 @@ def build(ledger, verifier, audit, config, staleness_policy=None):
             ledger, verifier, audit, config, staleness_policy,
             selection=selection, artifact_id=artifact_id, meter=meter,
             conflict=conflict, fusion=fusion, acceptance=acceptance,
-            promotion=promotion)
+            promotion=promotion, screen=screen)
 
     return build

changelog.d/254-rrsi-integration.md

@@ -0,0 +1,19 @@
+Added
+RRSI wired into `evolve()` (`examples/rrsi/rrsi_policies.py`, #254 / #221):
+`build_rrsi_bundle()` assembles the three regularizers -- `RRsiCritic`
+(pre-evaluation screen: annealed edit-budget gate + deterministic leakage
+screening + optional model reviewer), `RRsiProposalPolicy` (injects budget,
+measured per-edit history, pruning targets and stall exploration into the
+reflection), and `RRsiAcceptance` (the pinned Algorithm-2 floor + both
+cost-rule branches + novelty, with monotone `S*`) -- into one `Policies`
+bundle sharing a checkpointable `RRsiState`. New engine seam
+`Policies(screen=...)` (`ScreenPolicy`) runs **before measurement**, so a
+leaking or over-budget card is settled and reported as `critic-rejected`
+without a held-out sweep; new outcomes `noise-floor` / `cost-rule`; new
+counters `critic_checked` / `critic_rejected`; the aggregator's checkpoint
+now carries an acceptance policy's optional state, so a resumed run does not
+lose `S*`. The selection core and schedule fixtures are reused, not
+rewritten; declared deviations (one-candidate-per-round judging, per-merge
+anneal on the barrier-free path, configured `noise_band`) are recorded in
+`docs/algo-rrsi.md`. Default behaviour unchanged -- a run without the bundle
+reports zeros.