diff --git a/agentdescent/__init__.py b/agentdescent/__init__.py index f2b0862b..fdb67ca2 100644 --- a/agentdescent/__init__.py +++ b/agentdescent/__init__.py @@ -118,6 +118,7 @@ Sandbox, SandboxProvider, SandboxSpec, + ScreenPolicy, VerifierProtocol, ) from .evalcache import CacheProtocol, FileCache, MemoryCache diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index aab14812..87b08561 100644 --- a/agentdescent/aggregator.py +++ b/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), diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index ed38f387..2bedfe6c 100644 --- a/agentdescent/evolution.py +++ b/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 diff --git a/agentdescent/metrics.py b/agentdescent/metrics.py index 22f6ed0a..07b7d258 100644 --- a/agentdescent/metrics.py +++ b/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, diff --git a/agentdescent/policies.py b/agentdescent/policies.py index ef5626a1..1221e784 100644 --- a/agentdescent/policies.py +++ b/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 diff --git a/agentdescent/population.py b/agentdescent/population.py index 78a13025..0e510035 100644 --- a/agentdescent/population.py +++ b/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 diff --git a/changelog.d/254-rrsi-integration.md b/changelog.d/254-rrsi-integration.md new file mode 100644 index 00000000..2e8419a8 --- /dev/null +++ b/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. diff --git a/docs/algo-rrsi.md b/docs/algo-rrsi.md index 91ba4c0c..f999f044 100644 --- a/docs/algo-rrsi.md +++ b/docs/algo-rrsi.md @@ -1,9 +1,9 @@ -# RRSI — regularized harness mechanisms (microport increments) +# RRSI — regularized harness self-improvement (microport + evolve() wiring) -This is an **offline mechanism microport** of selected pieces from Google -Research's *Regularized Recursive Self-Improvement of Agent Harnesses* (RRSI). -It is not a full port and does not claim to reproduce the paper's benchmark -results. +This is a **mechanism microport** of Google Research's *Regularized Recursive +Self-Improvement of Agent Harnesses* (RRSI), wired into `evolve()` through the +engine's existing `Policies` seams. It does not claim to reproduce the paper's +benchmark results; the reported numbers are upstream's, not measured here. ## Sources and pin @@ -93,18 +93,57 @@ out of the tried set \(T_t\) and the yield \(g_t\). From it: | numpy scalar measurements (`np.float32`) | Written as-is | Accepted as any `numbers.Real` except `bool` | | Text-only diff (string literals / comments) declared structural, prose mentions a signal word | Declared tag kept | Recorded as `prompt` | -None of these mechanisms is wired into `evolve()`, `aggregator_factory`, a -proposal policy, or a resumable run record yet. There is no RRSI CLI, -harness-directory domain, leakage critic, proposer, trial aggregation, pruning -loop (the history computes \(B_t\) but nothing removes machinery), or -evaluation/accounting adapter. AgentDescent's default evolution behavior is -unchanged. Real-model and benchmark measurements are **unmeasured** here; -upstream reported values are not AgentDescent results. + +## Wired into `evolve()` (`examples/rrsi/rrsi_policies.py`) + +`build_rrsi_bundle(complete, protected=..., reviewer=..., config=..., +base_score=...)` assembles the three slots into one `Policies` bundle and +returns `(bundle, state)`. The three share one :class:`RRsiState` — `S*`, the +measured per-edit ledger, the component bookkeeping and the annealed round +counter — which is what makes them one algorithm rather than three knobs. + +| Slot | Policy | What it regularizes | +|---|---|---| +| `Policies(screen=...)` | `RRsiCritic` | Pre-evaluation screen: annealed edit-budget gate (`len(ops) > b_t` rejected) and deterministic leakage screening against the evolve set (`protected` strings), plus an optional model `reviewer`. Runs **before measurement** (issue #221: "screening before evaluation is important"), so a leaking or over-budget card is settled and reported as `critic-rejected` without a held-out sweep. | +| `Policies(proposal=...)` | `RRsiProposalPolicy` | The regularized proposer: injects the annealed budget, the measured per-edit history, pruning targets and the stall-exploration note into the reflection prompt. | +| `Policies(acceptance=...)` | `RRsiAcceptance` | The pinned `selection.judge` decision — historical-best floor, both cost-rule branches, novelty — with `S*` advanced through the pinned `update_best_score`. `S*`, the ledger and the round counter checkpoint/restore through the aggregator's optional acceptance hook, so a resumed run does not lose the floor. | + +### Declared deviations (recorded, not silent) + +* **One candidate per round.** The engine merges to one candidate per round, + so the multi-candidate `select_round` "choose highest admissible argmax" + step is not exercised; each merge's candidate is judged against the + incumbent and `S*`. This is the "one winner per round" semantics issue #221 + asks to establish first. +* **The round counter advances once per merge.** On the synchronous path that + is the round barrier; on the barrier-free path there is no round barrier, so + the annealed schedule anneals per merge instead of per RRSI round. Callers + who need the paper's round semantics run the synchronous path. +* **`noise_band` is a configured parameter** (default `0.0`, the conservative + floor `score >= S*`). The paper estimates it by repeatedly evaluating the + unchanged base harness; set `RRsiConfig.noise_band` from that measurement. +* **Cost is `EvidenceCard.cost_tokens`, averaged per rollout** (the proposing + rollout's usage, when the caller shares a `Usage` with its adapters), **not a + batch sum** -- summing across a merge's cards would make the cost rule depend + on how many cards happened to land in the bucket. This is still a proxy: + Algorithm 2's `dC` is the *candidate* harness's cost relative to the + incumbent's on the same evaluation, and the proposing rollout runs the + incumbent, so the cost rule is approximate until candidate-side held-out + telemetry exists. Missing or zero telemetry yields `dC = 0`, matching the + pinned upstream behaviour. +* **Protected strings come from the evolve set only** — never the final + holdout. The integration tests enforce that discipline by construction. + +What is still not wired: a full harness-directory domain adapter, multi-round +`select_round` trial aggregation, and a RRSI CLI. AgentDescent's default +evolution behavior is unchanged — a run without the bundle reports zero +`critic_checked` / `critic_rejected` and no `critic-rejected` / +`noise-floor` / `cost-rule` outcomes. ## Local verification ```bash -pytest -q tests/test_rrsi_selection.py tests/test_rrsi_schedule.py tests/test_rrsi_history.py +pytest -q tests/test_rrsi_selection.py tests/test_rrsi_schedule.py tests/test_rrsi_history.py tests/test_rrsi_integration.py ``` Selection goldens cover the historical floor, both cost branches, domain veto, @@ -113,7 +152,9 @@ numeric inputs. Schedule fixtures cover endpoints, monotonic anneal, the float guard, clamping, degenerate horizons and invalid bounds. History fixtures cover bundle fan-out, unmeasured records, the prune window and \(B_t\), `render`'s cap, torn-line recovery, atomic rewrites, stall and exploration, and -`normalize`'s evidence check. Completing the full -RRSI integration (issue #221) still needs pre-evaluation leakage screening, -the proposer and pruning loop that consume this history, and the -`evolve()`/CLI fixture. +`normalize`'s evidence check. Integration fixtures +cover the budget/leakage screen before evaluation, the model reviewer being +asked only when deterministic checks pass, history attribution, stall +exploration, pruning targets, `S*` monotonicity and checkpoint/restore, and +offline end-to-end runs (leak → `critic-rejected`, clean → commit, untouched → +zeros). diff --git a/docs/api.md b/docs/api.md index 15f0e6ab..e8f84dac 100644 --- a/docs/api.md +++ b/docs/api.md @@ -57,6 +57,8 @@ EvolutionResult( redispatched: int = 0, duplicates_dropped: int = 0, cas_conflicts: int = 0, + critic_checked: int = 0, + critic_rejected: int = 0, cache_hits: int = 0, cache_misses: int = 0, trajectories_recorded: int = 0, @@ -1056,7 +1058,8 @@ Aggregator( conflict: Optional['ConflictPolicy'] = None, fusion: Optional['FusionPolicy'] = None, acceptance: Optional['AcceptancePolicy'] = None, - promotion: Optional['PromotionPolicy'] = None + promotion: Optional['PromotionPolicy'] = None, + screen: Optional['ScreenPolicy'] = None ) -> None ``` @@ -1130,6 +1133,9 @@ The vocabulary of `category`. | `CAS_CONFLICT` | `'cas-conflict'` | | `UNKNOWN_ARTIFACT` | `'unknown-artifact'` | | `VIOLATES_INVARIANT` | `'violates-invariant'` | +| `CRITIC_REJECTED` | `'critic-rejected'` | +| `NOISE_FLOOR` | `'noise-floor'` | +| `COST_RULE` | `'cost-rule'` | ### `MergeReport(...)` @@ -2838,7 +2844,8 @@ PopulationAggregator( conflict = None, fusion = None, acceptance = None, - promotion = None + promotion = None, + screen = None ) ``` @@ -2863,7 +2870,8 @@ population_factory( conflict = None, fusion = None, acceptance = None, - promotion = None + promotion = None, + screen = None ) ``` diff --git a/examples/rrsi/__init__.py b/examples/rrsi/__init__.py index 051e9824..f5643c5d 100644 --- a/examples/rrsi/__init__.py +++ b/examples/rrsi/__init__.py @@ -6,4 +6,6 @@ * ``schedule`` — cosine-annealed L0 edit budget * ``components`` — the component vocabulary K / K_str and evidence-checked tags * ``history`` — the per-edit history L_t: yield, prune set, stall exploration +* ``rrsi_policies`` — the same mechanisms wired into ``evolve()`` through the + engine's ``Policies`` seams (screen / proposal / acceptance) """ diff --git a/examples/rrsi/rrsi_policies.py b/examples/rrsi/rrsi_policies.py new file mode 100644 index 00000000..fdfd4e5c --- /dev/null +++ b/examples/rrsi/rrsi_policies.py @@ -0,0 +1,614 @@ +"""RRSI policies wired into ``evolve()`` through the engine's seams. + +This is the integration the upstream-faithful fixtures were waiting for. The +selection core (``selection.py``, pinned to Algorithm 2) and the cosine-annealed +edit budget (``schedule.py``) are **reused here, not rewritten**; this module +only adapts them to the engine's ``Policies`` slots: + +* :class:`RRsiCritic` -- the pre-evaluation screen (``Policies(screen=...)``): + leakage screening against the evolve set plus the annealed edit-budget gate. + It runs before any measurement (issue #221: "screening before evaluation is + important"), so a leaking or over-budget card is settled without ever buying + a held-out sweep. The pinned selection core already has a home for its + verdict: ``Candidate.gate_failure``. +* :class:`RRsiProposalPolicy` -- the regularized proposer + (``Policies(proposal=...)``): injects the annealed budget, the measured + per-edit history, pruning targets and stall-exploration guidance into the + reflection prompt, so the proposal distribution is constrained without + restricting which harness components may be edited. +* :class:`RRsiAcceptance` -- the regularized selector + (``Policies(acceptance=...)``): the pinned ``judge`` decision (historical- + best floor, both cost-rule branches, novelty) with ``S*`` maintained through + the pinned ``update_best_score`` and a checkpointable state, so a resumed + run does not lose the floor. + +The three share one :class:`RRsiState` -- ``S*``, the measured per-edit +ledger, the component bookkeeping, the annealed round counter -- which is what +makes them one algorithm rather than three knobs. Build them with +:func:`build_rrsi_bundle`. + +**Fidelity and deviations** (recorded, per the port conventions): + +* The selection decision is the pinned ``judge`` verbatim. The engine merges + to **one candidate per round**, so the multi-candidate ``select_round`` + "choose highest admissible" step is not exercised here; each merge's + candidate is judged against the incumbent and ``S*``. This is the "one + winner per round" semantics issue #221 asks to establish first. +* The annealed round counter advances once per **merge** (per round barrier on + the synchronous path). On the barrier-free path there is no round barrier, + so the schedule anneals per merge instead of per RRSI round -- a declared + semantic deviation, not a silent one. +* ``delta`` (the noise band) is a configured :class:`RRsiConfig.noise_band`, + default ``0.0`` (the conservative floor ``score >= S*``). The paper estimates + it by repeatedly evaluating the unchanged base harness; this module treats it + as a parameter and documents how to set it. +* Token cost is read from ``EvidenceCard.cost_tokens`` (the proposing rollout's + usage, when the caller shares a ``Usage`` with its adapters). Missing or zero + telemetry yields ``dC = 0``, matching the pinned upstream behaviour. + +Privacy and data isolation follow the port conventions: the critic's +``protected`` strings must be derived from the **evolve set only** -- never the +final holdout -- and the integration tests enforce that discipline by +construction. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, Sequence, Set, Tuple + +from agentdescent.aggregator import MergeOutcome +from agentdescent.policies import ( + AcceptDecision, + AcceptancePolicy, + MergeContext, + Policies, + ProposalContext, + ProposalPolicy, + ScreenPolicy, +) + +from .schedule import edit_budget +from .selection import ( + Candidate, + Measurement, + SelectionConfig, + judge, + update_best_score, +) + +__all__ = [ + "RRsiAcceptance", + "RRsiConfig", + "RRsiCritic", + "RRsiProposalPolicy", + "RRsiState", + "build_rrsi_bundle", +] + + +# -- configuration and shared state ------------------------------------------- + + +@dataclass(frozen=True) +class RRsiConfig: + """The RRSI knobs: Algorithm 2's selection rule plus the annealed budget. + + ``beta0`` / ``beta1`` are the large-gain cost rule's tolerance and slope; + ``w_s`` / ``w_c`` / ``w_n`` weight the within-band score/cost/novelty rule; + ``noise_band`` is the historical-best floor tolerance (the paper's + empirically estimated ``delta``). ``b_min`` / ``b_max`` / ``T`` are the + annealed edit budget's schedule. + """ + + # -- selection (Algorithm 2) --------------------------------------------- + beta0: float = 0.0 + beta1: float = 0.0 + w_s: float = 1.0 + w_c: float = 1.0 + w_n: float = 0.0 + #: The historical-best noise floor: ``score >= S* - noise_band``. + noise_band: float = 0.0 + + # -- proposal (annealed edit budget) ------------------------------------- + b_min: int = 1 + b_max: int = 5 + #: Number of RRSI rounds the budget anneals over. + T: int = 20 + #: Rounds without S* movement (beyond ``noise_band``) that count as a stall. + stall_window: int = 5 + #: Rounds of component yield a component must show positive gain over to + #: keep its place (the L1-style pruner's window). + prune_window: int = 5 + #: How much of the measured per-edit history fits in the reflection prompt. + max_history_chars: int = 1500 + + +@dataclass +class EditRecord: + """One measured candidate, as the proposer conditions on in later rounds.""" + + round: int + components: Tuple[str, ...] + diff_size: int + delta_score: float + delta_cost: float + accepted: bool + + +#: A diff -> component-set extractor. The default takes the first path segment +#: of each op key, so a FileTree harness ("skills/x/SKILL.md") attributes an +#: edit to its top-level component ("skills"), which is what RRSI's novelty and +#: pruning rules count. +def default_components(diff: Any) -> Tuple[str, ...]: + out: Set[str] = set() + for key in diff.ops: + head = str(key).split("/", 1)[0] + if head: + out.add(head) + return tuple(sorted(out)) + + +class RRsiState: + """S*, the per-edit ledger, component bookkeeping and the annealed round. + + One instance is shared by the critic, the proposer and the selector -- that + sharing is what makes the three slots one regularized algorithm. It is + checkpointable (see :meth:`checkpoint`) so a resumed run does not lose the + historical-best floor or the measured history. + """ + + def __init__(self, config: Optional[RRsiConfig] = None, *, + base_score: float = 0.0, + components: Optional[Callable[[Any], Tuple[str, ...]]] = None): + self.config = config or RRsiConfig() + self.components = components or default_components + #: The annealed round counter. Advanced once per merge (see the module + #: docstring for the barrier-free deviation). Never reset on resume. + self.round = 0 + #: The historical-best evolve-set score ``S*``, monotone via the pinned + #: ``update_best_score``. + self.best_score = float(base_score) + #: The incumbent's measured policy-token cost, ``None`` until a first + #: accepted candidate carried cost telemetry. ``None`` is "unknown", + #: which the pinned cost rule reads as ``dC = 0``. + self.incumbent_cost: Optional[float] = None + #: component -> how often it appeared in a proposed diff (novelty's + #: denominator, and the pruner's "has it been exercised" test). + self.attempts: Counter = Counter() + #: component -> rolling measured gain per round, for the pruner. + self._round_gains: List[Dict[str, float]] = [] + #: The measured per-edit history the proposer conditions on. + self.ledger: List[EditRecord] = [] + #: ``S*`` after each merge, for stall detection. + self.progress: List[float] = [] + + def record_merge(self) -> None: + """Advance the schedule by one merge, whatever its outcome. + + ``round`` and ``progress`` are the anneal and the stall detector, and + both must move on **every** merge decision -- a rejection is a round the + search spent stuck, and the sparsity anneal is exactly what makes a + stuck search try smaller bundles, while a flat ``progress`` is what + makes :meth:`stalled` trip into structured exploration. Advancing only + on accepts left the budget pinned at ``b_max`` and ``stalled()`` False + through a streak of rejections (review).""" + self.round += 1 + self.progress.append(self.best_score) + + def budget(self) -> int: + """The annealed edit budget for the current round (``b_t``).""" + return edit_budget(self.round, self.config.T, + self.config.b_min, self.config.b_max) + + def stalled(self) -> bool: + """True when ``S*`` has not moved beyond the noise band recently. + + Structured exploration fires here: the proposer reserves part of its + budget for components never yet exercised.""" + window = (self.progress[-self.config.stall_window:] + if self.config.stall_window > 0 else []) + if len(window) < 2: + return False + return (window[-1] - window[0]) <= self.config.noise_band + + def pruning_targets(self) -> List[str]: + """Components whose recent measured gain is non-positive (L1-style). + + A component that has been exercised in the last ``prune_window`` rounds + but whose net gain over that window is not strictly positive has stopped + earning its place; the proposer is told it is a deletion candidate. A + component with no gain record is not a target (it may simply be new).""" + window = self._round_gains[-self.config.prune_window:] + if not window: + return [] + totals: Dict[str, float] = defaultdict(float) + exercised: Set[str] = set() + for gains in window: + for comp, gain in gains.items(): + totals[comp] += gain + exercised.add(comp) + return sorted(comp for comp in exercised + if totals.get(comp, 0.0) <= 0.0 + and self.attempts.get(comp, 0) > 0) + + def components_of(self, diff: Any) -> Tuple[str, ...]: + return self.components(diff) + + def record_measurement(self, diff: Any, *, delta_score: float, + delta_cost: float, accepted: bool) -> None: + """Record one measured candidate into the ledger and the pruner's books. + + Called for accepted *and* rejected candidates: rejected mechanisms are + negative evidence the proposer conditions on (the paper's evidence-aware + credit assignment), and a rejection is exactly what the pruner should + count against a component.""" + comps = self.components_of(diff) + for c in comps: + self.attempts[c] += 1 + gains = {c: delta_score for c in comps} + self._round_gains.append(gains) + self.ledger.append(EditRecord( + round=self.round, components=comps, diff_size=len(diff.ops), + delta_score=delta_score, delta_cost=delta_cost, accepted=accepted)) + + def checkpoint(self) -> dict: + """Serialise the state the ledger does not persist.""" + return { + "round": self.round, + "best_score": self.best_score, + "incumbent_cost": self.incumbent_cost, + "attempts": dict(self.attempts), + "round_gains": self._round_gains, + "ledger": [{"round": r.round, "components": list(r.components), + "diff_size": r.diff_size, "delta_score": r.delta_score, + "delta_cost": r.delta_cost, "accepted": r.accepted} + for r in self.ledger], + "progress": self.progress, + } + + def restore(self, data: dict) -> None: + """Restore :meth:`checkpoint`'s form. Unknown keys are ignored.""" + try: + if isinstance(data.get("round"), int): + self.round = data["round"] + if isinstance(data.get("best_score"), (int, float)): + self.best_score = float(data["best_score"]) + if data.get("incumbent_cost") is not None: + self.incumbent_cost = float(data["incumbent_cost"]) + if isinstance(data.get("attempts"), dict): + self.attempts = Counter( + {k: int(v) for k, v in data["attempts"].items() + if isinstance(v, (int, float))}) + if isinstance(data.get("round_gains"), list): + self._round_gains = [ + {k: float(v) for k, v in gains.items() + if isinstance(v, (int, float))} + for gains in data["round_gains"] if isinstance(gains, dict)] + if isinstance(data.get("ledger"), list): + self.ledger = [] + for r in data["ledger"]: + if not isinstance(r, dict): + continue + try: + self.ledger.append(EditRecord( + round=int(r["round"]), + components=tuple(r.get("components", ())), + diff_size=int(r.get("diff_size", 0)), + delta_score=float(r.get("delta_score", 0.0)), + delta_cost=float(r.get("delta_cost", 0.0)), + accepted=bool(r.get("accepted", False)))) + except (TypeError, ValueError, KeyError): + continue + if isinstance(data.get("progress"), list): + self.progress = [float(v) for v in data["progress"] + if isinstance(v, (int, float))] + except Exception: # noqa: BLE001 - a bad payload must not take the run down + return + + +# -- the critic: pre-evaluation screen ------------------------------------------ + + +def _diff_text(diff: Any) -> str: + """The card's editable content, for the deterministic leakage scan.""" + return "\n".join([str(k) for k in diff.ops] + + [str(v) for v in diff.ops.values()]) + + +class RRsiCritic: + """The pre-evaluation screen: leakage + annealed edit budget. + + Wired as ``Policies(screen=...)`` and run by the aggregator **before** + measurement (issue #221: "screening before evaluation is important"). Each + card is checked deterministically for + + * **edit budget** -- ``len(diff.ops) > b_t`` is rejected: the proposer was + told to bundle at most ``b_t`` independent edits, and an over-budget + bundle is exactly the "many unrelated modifications in one candidate" + the annealed sparsity constraint exists to stop; + * **leakage** -- any ``protected`` string appearing in the diff's keys or + values is rejected: task names, entity names, task-specific values or + answers encoded into the harness are benchmark-specific content, and a + leaking candidate must never receive the inflated evolve-set score that + would make it attractive to later rounds. + + An optional model ``reviewer`` (a ``Completion``) performs the paper's + model critic: it reads the candidate and returns ``"leak"``, ``"inert"`` + or ``"ok"``. The deterministic checks run first; the model is only asked + when they pass, so a leaking card never pays a model call either. + + ``protected`` must come from the **evolve set only** -- never the final + holdout. The integration tests enforce that discipline by construction. + """ + + def __init__(self, state: RRsiState, *, protected: Sequence[str] = (), + reviewer: Optional[Any] = None): + self.state = state + self._protected = tuple(s for s in protected if s) + self._reviewer = reviewer + + def screen(self, card: Any) -> Optional[str]: + """Return ``None`` to let the card through, or a reason to reject it.""" + diff = getattr(card, "diff", None) + if diff is None: + return "card carries no diff" + budget = self.state.budget() + if len(diff.ops) > budget: + return (f"edit bundle has {len(diff.ops)} edit(s); annealed budget " + f"at round {self.state.round} is {budget}") + text = _diff_text(diff) + for token in self._protected: + if token in text: + return f"leaks evolve-set content ({token!r})" + if self._reviewer is not None: + verdict = self._reviewer_verdict(diff, text) + if verdict != "ok": + return f"critic: {verdict}" + return None + + def _reviewer_verdict(self, diff: Any, text: str) -> str: + """Ask the model critic for its verdict. Best-effort: an unusable reply + lets the card through (the deterministic checks remain the gate).""" + prompt = ( + "You are screening an edit to an agent harness before it is " + "evaluated.\n\nEdit:\n{text}\n\nReply exactly one word: 'leak' if " + "the edit encodes benchmark-specific content (task names, entity " + "names, values, answers), 'inert' if it adds machinery that cannot " + "change behavior, or 'ok'.") + try: + reply = (self._reviewer(prompt.format(text=text[:4000])).strip().lower()) + except Exception: # noqa: BLE001 - a broken reviewer must not block a run + return "ok" + return "ok" if reply not in ("leak", "inert") else reply + + +# -- the regularized proposer --------------------------------------------------- + + +def _expected_block(task: Any, show_meta: bool, limit: int) -> str: + meta = getattr(task, "meta", None) + if not show_meta or not meta: + return "" + text = ", ".join(f"{k}={v!r}" for k, v in meta.items()) + if len(text) > limit: + text = text[:limit] + " ..." + return f"\nWhat the scorer expected (task metadata):\n{text}\n" + + +def _history_summary(ledger: Sequence[EditRecord], max_chars: int) -> str: + """A compact recent-history summary for the reflection prompt.""" + if not ledger: + return "" + recent = ledger[-8:] + lines = [] + for r in recent: + tag = "accepted" if r.accepted else "rejected" + comps = ",".join(r.components) or "?" + lines.append( + f"- r{r.round} [{tag}] {comps}: dS={r.delta_score:+.2f} " + f"dC={r.delta_cost:+.2f} ({r.diff_size} edit(s))") + out = "\n".join(lines) + if len(out) > max_chars: + out = out[:max_chars] + " ..." + return out + + +class RRsiProposalPolicy: + """The regularized proposer: ``Policies(proposal=...)``. + + Wraps a ``Completion`` and builds the reflection prompt itself, so the + RRSI guidance -- the annealed budget, the measured per-edit history, the + pruning targets and the stall-exploration note -- reaches the reflector + exactly like the ordinary failure prompt does. The strategy still turns the + reply into a diff; the critic then *enforces* the budget deterministically, + so guidance and enforcement agree. + + ``complete`` has the repo's ``Completion`` shape (``prompt -> str``); the + same model can drive solve and reflection, or a cheap model can reflect on + an expensive agent. + """ + + def __init__(self, complete: Callable[[str], str], state: RRsiState, *, + show_meta: bool = True, meta_chars: int = 600): + self.complete = complete + self.state = state + self.show_meta = show_meta + self.meta_chars = meta_chars + + def propose(self, ctx: ProposalContext) -> Sequence[str]: + config = self.state.config + budget = self.state.budget() + lines = [ + f"The artifact just failed a task (score {ctx.reward:.2f} out of 1.0).", + "", + "Artifact so far:", + ctx.rendered, + "", + "Task input:", + str(ctx.task.prompt), + "", + "It produced:", + ctx.output, + _expected_block(ctx.task, self.show_meta, self.meta_chars), + "", + f"[RRSI budget] Round {self.state.round} of {config.T} allows at most " + f"{budget} independent edit(s) in one candidate; bundle fewer, more " + "attributable edits later in the run.", + ] + history = _history_summary(self.state.ledger, config.max_history_chars) + if history: + lines.append("") + lines.append("[RRSI history] Measured so far:") + lines.append(history) + targets = self.state.pruning_targets() + if targets: + lines.append("") + lines.append("[RRSI pruning] These components have not earned their " + f"place recently; consider deleting them: {', '.join(targets)}") + if self.state.stalled(): + lines.append("") + lines.append("[RRSI exploration] The search has stalled within the " + "noise band; spend part of the budget on components never " + "yet exercised.") + lines.extend([ + "", + "Propose exactly ONE concise candidate (a single imperative rule, or " + "an edit script), or NONE if no change would help.", + ]) + prompt = "\n".join(lines) + reply = (self.complete(prompt) or "").strip() + if not reply or reply.upper().startswith("NONE"): + return [] + return [reply] + + +# -- the regularized selector --------------------------------------------------- + + +def _card_cost(cards: Sequence[Any]) -> Optional[float]: + """The candidate's measured policy-token cost, or ``None`` when unknown. + + A **per-rollout mean**, not a batch sum: `EvidenceCard.cost_tokens` is one + proposing rollout's usage, so summing across a merge's cards would scale + with however many cards happened to land in the bucket and reject a + candidate purely for batch size (review). The mean is still a proxy -- + Algorithm 2's ``dC`` is the *candidate* harness's cost relative to the + incumbent's on the same evaluation, and the proposing rollout runs the + incumbent -- so this is a documented deviation; callers who share a + ``Usage`` get an approximate, batch-size-independent cost signal, and + missing/zero telemetry yields ``dC = 0`` (the pinned upstream behaviour). + """ + costs = [int(getattr(card, "cost_tokens", 0) or 0) for card in cards] + if not costs: + return None + return sum(costs) / len(costs) if any(costs) else None + + +class RRsiAcceptance: + """Algorithm 2's selector, adapted to the engine's per-merge decision. + + Wired as ``Policies(acceptance=...)``. Each merge's candidate is judged by + the **pinned** ``selection.judge`` -- historical-best floor, both cost-rule + branches, novelty -- against the incumbent and ``S*``; ``S*`` is advanced + through the pinned ``update_best_score``. The state (floor, ledger, round) + checkpoint/restore through the aggregator's optional acceptance hook, so a + resumed run does not lose the floor. + """ + + def __init__(self, state: RRsiState): + self.state = state + + def accept(self, ctx: MergeContext) -> AcceptDecision: + config = self.state.config + cand_rate = MergeContext.rate(ctx.cand_counts) + base_rate = MergeContext.rate(ctx.base_counts) + cand_cost = _card_cost(ctx.cards) + base_cost = self.state.incumbent_cost + observed = cand_rate - base_rate + + candidate = Candidate( + variant=ctx.diff.diff_id if ctx.diff is not None else "?", + components=self.state.components_of(ctx.diff), + measurement=Measurement(score=cand_rate, cost=cand_cost)) + incumbent = Measurement(score=base_rate, cost=base_cost) + selection_config = SelectionConfig( + beta0=config.beta0, beta1=config.beta1, + w_s=config.w_s, w_c=config.w_c, w_n=config.w_n) + decision = judge( + candidate, incumbent, self.state.best_score, config.noise_band, + selection_config, dict(self.state.attempts)) + + # The schedule advances on *every* decision this policy makes -- a + # rejection is a round the search spent stuck, not a free one (see + # RRsiState.record_merge). best_score is advanced only on an accept, + # and the accept records the **post-decision** S*: appending before the + # update would make an improve-after-rejection-streak look stalled even + # though S* just moved (review). + if not decision.admissible: + self.state.record_merge() + self.state.record_measurement(ctx.diff, delta_score=observed, + delta_cost=decision.delta_cost or 0.0, + accepted=False) + category = (MergeOutcome.NOISE_FLOOR.value + if "floor" in decision.reason + else MergeOutcome.COST_RULE.value) + return AcceptDecision(False, category, decision.reason, + p_improve=0.0, observed_delta=observed) + + self.state.best_score = update_best_score(self.state.best_score, candidate) + self.state.incumbent_cost = cand_cost + self.state.record_merge() + self.state.record_measurement(ctx.diff, delta_score=observed, + delta_cost=decision.delta_cost or 0.0, + accepted=True) + return AcceptDecision(True, "", decision.reason, + p_improve=max(0.0, observed), observed_delta=observed) + + # -- checkpoint hooks (aggregator optional acceptance state) --------------- + def checkpoint_state(self) -> dict: + return self.state.checkpoint() + + def restore_state(self, data: dict) -> None: + self.state.restore(data) + + def on_screen_reject(self) -> None: + """Advance the schedule on a pre-measurement merge refusal. + + The screen (critic) and the governance invariant gate both run in + ``Aggregator._prepare``, before this policy's ``accept`` is ever reached. + Without this hook a streak of screen-rejected or violates-invariant + batches would leave the anneal and the stall detector frozen even though + a merge was decided. The aggregator calls this once per such refusal; + ``record_merge`` advances ``round``/``progress`` exactly as an acceptance + decision does (review: the round must advance on every merge).""" + self.state.record_merge() + + +# -- the bundle ----------------------------------------------------------------- + + +def build_rrsi_bundle( + complete: Callable[[str], str], + *, + protected: Sequence[str] = (), + reviewer: Optional[Any] = None, + config: Optional[RRsiConfig] = None, + base_score: float = 0.0, +) -> Tuple[Policies, RRsiState]: + """Assemble the three RRSI policies into one ``Policies`` bundle. + + Returns ``(bundle, state)``; keep ``state`` to read (or checkpoint) the + historical-best floor and the measured history. ``protected`` is the evolve + set's distinctive strings (task ids, gold answers, names) -- never the + final holdout. ``complete`` drives the proposer; an optional ``reviewer`` + drives the model critic. ``base_score`` seeds ``S*`` (the base harness's + evolve-set score when you have it).""" + state = RRsiState(config, base_score=base_score) + bundle = Policies( + screen=RRsiCritic(state, protected=protected, reviewer=reviewer), + proposal=RRsiProposalPolicy(complete, state), + acceptance=RRsiAcceptance(state), + ) + return bundle, state diff --git a/tests/test_rrsi_integration.py b/tests/test_rrsi_integration.py new file mode 100644 index 00000000..f0823119 --- /dev/null +++ b/tests/test_rrsi_integration.py @@ -0,0 +1,432 @@ +"""The RRSI integration has to be right in the exact places issue #221 calls out: + +* leakage and edit-budget screening must run **before** evaluation -- a leaking + or over-budget card must never buy a held-out sweep, never mind commit; +* the selector must reuse the pinned Algorithm-2 rule (historical-best floor, + both cost-rule branches, novelty) and keep ``S*`` monotone and resumable; +* the proposer must actually condition on the measured history, the pruning + targets and the stall-exploration note -- not just carry the knobs; +* the untouched path stays untouched (a run without the bundle reports zeros). + +Every fixture here is offline and deterministic. +""" + +import pytest + +from agentdescent import Contract, Invariant +from agentdescent.aggregator import MergeOutcome +from agentdescent.evolution import SingleSlot, Task, evolve +from agentdescent.evolvable import Diff, EvidenceCard +from agentdescent.policies import MergeContext, Policies + +from examples.rrsi.rrsi_policies import ( + RRsiAcceptance, + RRsiConfig, + RRsiCritic, + RRsiProposalPolicy, + RRsiState, + build_rrsi_bundle, +) + + +def _task(n): + return Task(id=f"t{n}", prompt=f"item {n}", meta={"gold": "42"}) + + +def _diff(ops, diff_id="d"): + return Diff(diff_id=diff_id, target="artifact", ops=ops) + + +def _card(ops, cost_tokens=0): + return EvidenceCard(diff=_diff(ops), base_version={}, touched=["artifact"], + trajectory_refs=[_task(0)], cost_tokens=cost_tokens) + + +def _context(state, *, base=(5, 5), cand=(5, 5), ops=("value",), costs=0): + from agentdescent.evolution import EvolvingArtifact + art = EvolvingArtifact("artifact", {"value": "seed"}, strategy=SingleSlot()) + cand_art = EvolvingArtifact("artifact", {"value": "new"}, strategy=SingleSlot()) + cards = [_card({k: "x" for k in ops}, cost_tokens=costs)] + return MergeContext( + artifact=art, candidate=cand_art, cards=cards, + base_counts=base, cand_counts=cand, diff=cards[0].diff) + + +# -- the critic: budget + leakage before evaluation ----------------------------- + + +def test_the_budget_anneals_and_the_critic_enforces_it(): + state = RRsiState(RRsiConfig(b_min=1, b_max=5, T=4)) + critic = RRsiCritic(state) + assert state.budget() == 5 # round 0: early, wide + assert critic.screen(_card({"a": "x", "b": "y", "c": "z"})) is None + # round 3 of T=4 has annealed to 2: a 3-op diff now exceeds it + state.round = 3 + assert state.budget() == 2 + assert critic.screen(_card({"a": "x", "b": "y"})) is None + reason = critic.screen(_card({"a": "x", "b": "y", "c": "z"})) + assert reason is not None and "budget" in reason + + +def test_the_critic_rejects_evolve_set_leakage(): + state = RRsiState() + critic = RRsiCritic(state, protected=("t3", "42")) + assert critic.screen(_card({"value": "generic rule"})) is None + reason = critic.screen(_card({"value": "always answer t3 with 42"})) + assert reason is not None and "leaks evolve-set content" in reason + + +def test_the_model_reviewer_is_only_asked_when_deterministic_checks_pass(): + seen = {"n": 0} + + def reviewer(prompt): + seen["n"] += 1 + return "leak" + + state = RRsiState() + critic = RRsiCritic(state, protected=("t3",), reviewer=reviewer) + # deterministic leak -> the reviewer is never asked + critic.screen(_card({"value": "mention t3"})) + assert seen["n"] == 0 + # clean diff -> the reviewer is asked and can veto + reason = critic.screen(_card({"value": "generic rule"})) + assert reason == "critic: leak" + assert seen["n"] == 1 + + +def test_protected_strings_never_come_from_the_holdout(): + # The discipline issue #221 demands: protected strings are derived from the + # evolve set only. A holdout answer must not be in the critic's set. + train_gold = {"t0": "42", "t1": "17"} + holdout_gold = "SECRET-FINAL-HOLDOUT-ANSWER" + protected = [f"{k}={v}" for k, v in train_gold.items()] + assert all(holdout_gold not in p for p in protected) + assert "SECRET-FINAL-HOLDOUT-ANSWER" not in " ".join(protected) + + +# -- the state: history, stall, pruning ---------------------------------------- + + +def test_record_measurement_builds_the_history_ledger(): + state = RRsiState() + state.record_measurement(_diff({"skills/x.md": "a"}), delta_score=0.1, + delta_cost=0.0, accepted=True) + state.record_measurement(_diff({"memory/y.md": "b"}), delta_score=-0.1, + delta_cost=0.0, accepted=False) + assert state.attempts["skills"] == 1 + assert state.attempts["memory"] == 1 + assert len(state.ledger) == 2 + assert state.ledger[0].accepted is True + assert state.ledger[1].accepted is False + + +def test_stall_detection_and_pruning_targets(): + state = RRsiState(RRsiConfig(noise_band=0.05, prune_window=2, + stall_window=3)) + # a component that never yields positive gain is a pruning target + state.record_measurement(_diff({"tools/a.md": "x"}), delta_score=0.0, + delta_cost=0.0, accepted=False) + state.record_measurement(_diff({"tools/b.md": "x"}), delta_score=-0.05, + delta_cost=0.0, accepted=False) + state._round_gains = state._round_gains[-2:] + assert "tools" in state.pruning_targets() + # a healthy component is not + state2 = RRsiState(RRsiConfig(prune_window=2)) + state2.record_measurement(_diff({"skill/a.md": "x"}), delta_score=0.3, + delta_cost=0.0, accepted=True) + assert state2.pruning_targets() == [] + # a flat S* trajectory beyond the noise band is a stall + state3 = RRsiState(RRsiConfig(noise_band=0.05, stall_window=3)) + state3.progress = [0.5, 0.51, 0.5] + assert state3.stalled() is True + state3.progress = [0.5, 0.6, 0.7] + assert state3.stalled() is False + + +def test_the_proposer_prompt_carries_budget_history_pruning_and_stall(): + from agentdescent.policies import ProposalContext + + calls = {"prompts": []} + + def complete(prompt): + calls["prompts"].append(prompt) + return "always state the year" + + state = RRsiState(RRsiConfig(T=10, prune_window=2)) + state.record_measurement(_diff({"tools/a.md": "x"}), delta_score=0.0, + delta_cost=0.0, accepted=False) + state.progress = [0.5, 0.5, 0.5, 0.5] # stalled + policy = RRsiProposalPolicy(complete, state) + out = policy.propose(ProposalContext( + rendered="seed", task=_task(0), output="answer", reward=0.0)) + prompt = calls["prompts"][0] + assert out == ["always state the year"] + assert f"Round {state.round}" in prompt + assert "[RRSI history]" in prompt + assert "[RRSI pruning]" in prompt and "tools" in prompt + assert "[RRSI exploration]" in prompt + + +def test_state_checkpoint_round_trip(): + state = RRsiState() + state.best_score = 0.9 + state.round = 3 + state.record_measurement(_diff({"skill/a.md": "x"}), delta_score=0.4, + delta_cost=0.0, accepted=True) + fresh = RRsiState() + fresh.restore(state.checkpoint()) + assert fresh.best_score == 0.9 + assert fresh.round == 3 + assert len(fresh.ledger) == 1 + assert fresh.ledger[0].components == ("skill",) + assert fresh.attempts["skill"] == 1 + + +# -- the selector: pinned Algorithm 2 via the engine's acceptance slot ---------- + + +def test_acceptance_applies_the_historical_best_floor(): + state = RRsiState(base_score=0.9) + policy = RRsiAcceptance(state) + # candidate 0.8 < S* (0.9) with delta 0 -> noise floor + d = policy.accept(_context(state, base=(5, 5), cand=(8, 2))) + assert d.accept is False + assert d.category == MergeOutcome.NOISE_FLOOR.value + assert state.best_score == 0.9 # S* unchanged + + +def test_acceptance_large_gain_cost_rule(): + state = RRsiState(base_score=0.5) + state.incumbent_cost = 100 + policy = RRsiAcceptance(state) + # deltaS = 0.7 - 0.5 = 0.2 > noise_band(0); deltaC = (300-100)/100 = 2.0 + # beta0=0.1, beta1=1.0 -> budget 0.1 + 0.2 = 0.3 < 2.0 -> refuse + d = policy.accept(_context(state, base=(5, 5), cand=(7, 3), costs=300)) + assert d.accept is False + assert d.category == MergeOutcome.COST_RULE.value + + +def test_a_streak_of_rejections_anneals_the_budget_and_trips_stall(): + """The annealed round must advance on *every* merge decision, not just + accepts: a search that keeps being rejected is spending rounds stuck, and + the sparsity anneal plus the stall note are exactly the mechanisms that + exist to get it unstuck. Before the fix, ten straight rejections left + `round` at 0, the budget at b_max, and `stalled()` False (review).""" + st = RRsiState(RRsiConfig(b_min=1, b_max=5, T=4, stall_window=3), + base_score=0.5) + acc = RRsiAcceptance(st) + for _ in range(10): + d = acc.accept(_context(st, base=(5, 5), cand=(3, 7))) + assert d.accept is False + assert st.round == 10, ( + f"the round must advance per merge, not per accept; round={st.round}") + assert st.budget() <= 2, ( + "a streak of rejections must anneal the edit budget, not pin it at " + f"b_max; budget={st.budget()}") + assert st.stalled() is True, ( + "a flat progress must trip the stall note while the search is stuck") + + +def test_an_improvement_after_a_rejection_streak_is_not_a_stall(): + """The accept must append the *post-decision* S*, so an improvement that + breaks a rejection streak is not reported as a stall. Before the fix, + `record_merge()` ran before `best_score` was updated, so an accepted jump + left progress flat (`[0.5, 0.5, 0.5]`) and `stalled()` True even though S* + just moved (review).""" + st = RRsiState(RRsiConfig(stall_window=3), base_score=0.5) + acc = RRsiAcceptance(st) + # two rejections (base 0.5 -> cand 0.3): progress stays at 0.5 + assert acc.accept(_context(st, base=(5, 5), cand=(3, 7))).accept is False + assert acc.accept(_context(st, base=(5, 5), cand=(3, 7))).accept is False + # one acceptance (base 0.5 -> cand 0.8): S* moves to 0.8 + assert acc.accept(_context(st, base=(5, 5), cand=(8, 2))).accept is True + assert st.best_score == pytest.approx(0.8) + assert st.progress == [0.5, 0.5, 0.8], ( + "the accept must append the post-decision S*, not the pre-decision one; " + f"got {st.progress}") + assert st.stalled() is False, ( + "an improvement that broke the streak must not read as a stall") + assert st.round == 3 + + +def test_an_improvement_after_a_rejection_streak_survives_checkpoint(): + """The corrected progress must round-trip through the acceptance state's + checkpoint: a resumed run must not inherit a stale flat history that reads + as a stall when S* actually moved.""" + st = RRsiState(RRsiConfig(stall_window=3), base_score=0.5) + acc = RRsiAcceptance(st) + acc.accept(_context(st, base=(5, 5), cand=(3, 7))) # reject + acc.accept(_context(st, base=(5, 5), cand=(3, 7))) # reject + acc.accept(_context(st, base=(5, 5), cand=(8, 2))) # accept -> 0.8 + fresh = RRsiState(RRsiConfig(stall_window=3), base_score=0.5) + fresh.restore(st.checkpoint()) + assert fresh.best_score == pytest.approx(0.8) + assert fresh.progress == [0.5, 0.5, 0.8] + assert fresh.stalled() is False, ( + "restored history must not report a stall for an S* that moved") + + +def test_screen_rejections_advance_the_schedule(): + """The aggregator's screen-rejected path (critic) must advance the anneal + too -- it is a merge decision, and the RRSI acceptance policy opts in via + on_screen_reject (review: the round advanced only on accepts).""" + state = RRsiState(RRsiConfig(b_min=1, b_max=5, T=4)) + policy = RRsiAcceptance(state) + for _ in range(5): + policy.on_screen_reject() + assert state.round == 5 + assert state.budget() <= 4 # annealed below b_max after rejections + + +def test_card_cost_is_a_per_rollout_mean_not_a_batch_sum(): + """The cost rule must not depend on how many cards happened to land in the + bucket: `EvidenceCard.cost_tokens` is one proposing rollout's usage, so + `_card_cost` is a per-rollout mean, not a sum. Three cards at 100 tokens + each must read the same as one card at 100 (review: the sum gave dC=+2.0 + and rejected on batch size).""" + from examples.rrsi.rrsi_policies import _card_cost + + single = _card_cost([_card({"a": "x"}, cost_tokens=100)]) + batch = _card_cost([_card({"a": "x"}, cost_tokens=100), + _card({"b": "y"}, cost_tokens=100), + _card({"c": "z"}, cost_tokens=100)]) + assert single == batch == 100.0, ( + f"per-rollout mean must be batch-size independent; " + f"single={single}, batch={batch}") + # missing/zero telemetry -> None (dC = 0, the documented behaviour) + assert _card_cost([_card({"a": "x"})]) is None + assert _card_cost([]) is None + + +def test_acceptance_accepts_an_improvement_and_advances_s_star(): + state = RRsiState(base_score=0.5) + policy = RRsiAcceptance(state) + d = policy.accept(_context(state, base=(5, 5), cand=(7, 3))) + assert d.accept is True + assert state.best_score == pytest.approx(0.7) # monotone advance + assert state.round == 1 + # a later candidate below the new S* is refused by the floor + d2 = policy.accept(_context(state, base=(7, 3), cand=(5, 5))) + assert d2.accept is False + assert d2.category == MergeOutcome.NOISE_FLOOR.value + + +def test_acceptance_within_band_rule_counts_novelty(): + # within the noise band, novelty can carry an otherwise-flat candidate + cfg = RRsiConfig(noise_band=0.05, w_s=1.0, w_c=1.0, w_n=1.0) + state = RRsiState(cfg, base_score=0.5) + policy = RRsiAcceptance(state) + # deltaS = 0.1 <= band(0.05)? no -> 0.1 > 0.05 is large-gain. Use cand=0.52 + # deltaS=0.02 <= band: within-band; novel component skill (unseen) -> w_n*1 + d = policy.accept(_context(state, base=(50, 50), cand=(52, 48), + ops=("skill/x.md",))) + assert d.accept is True + assert state.best_score == pytest.approx(0.52) + + +# -- end to end ---------------------------------------------------------------- + + +def _offline_actors(): + def run(rendered, task): + return "answer" + (" 2026" if "year" in rendered else "") + + def reward(task, output): + return 1.0 if "2026" in output else 0.0 + + return run, reward + + +def _tasks(n=8): + return [Task(id=f"t{i}", prompt=f"item {i}") for i in range(n)] + + +def _complete_factory(leak=False): + def complete(prompt): + return "always mention t3" if leak else "always state the year" + return complete + + +def test_end_to_end_leaking_candidate_is_screened_before_measurement(tmp_path): + run, reward = _offline_actors() + bundle, state = build_rrsi_bundle(_complete_factory(leak=True), + protected=("t3",)) + result = evolve(_tasks(), reward, run=run, + strategy=SingleSlot(initial_value="seed"), rounds=2, + n_workers=2, repo_path=str(tmp_path / "repo"), + policies=bundle) + assert result.critic_rejected >= 1 + assert result.critic_checked >= 1 + assert result.outcomes().get("critic-rejected", 0) >= 1 + assert result.outcomes().get("committed", 0) == 0 + + +def test_end_to_end_a_clean_run_commits_under_the_bundle(tmp_path): + run, reward = _offline_actors() + bundle, state = build_rrsi_bundle(_complete_factory(), + protected=("t3",)) + result = evolve(_tasks(), reward, run=run, + strategy=SingleSlot(initial_value="seed"), rounds=2, + n_workers=2, repo_path=str(tmp_path / "repo"), + policies=bundle) + assert result.outcomes().get("committed", 0) >= 1 + assert result.outcomes().get("critic-rejected", 0) == 0 + assert result.critic_rejected == 0 + assert state.best_score > 0.0 # S* advanced + + +def test_untouched_path_reports_zeros(tmp_path): + run, reward = _offline_actors() + result = evolve(_tasks(), reward, run=run, + strategy=SingleSlot(initial_value="seed"), rounds=1, + n_workers=2, repo_path=str(tmp_path / "repo"), + propose=lambda r, t, o, re: "always state the year") + assert result.critic_checked == 0 + assert result.critic_rejected == 0 + assert result.outcomes().get("critic-rejected", 0) == 0 + + +def test_resume_preserves_the_historical_best_floor(tmp_path): + run, reward = _offline_actors() + repo = str(tmp_path / "repo") + bundle1, state1 = build_rrsi_bundle(_complete_factory()) + evolve(_tasks(), reward, run=run, strategy=SingleSlot(initial_value="seed"), + rounds=1, n_workers=2, repo_path=repo, policies=bundle1, + checkpointing=True) + assert state1.best_score > 0.0 + # a fresh bundle on the same repo: the aggregator restores the acceptance + # state, so the floor is preserved, not reset + bundle2, state2 = build_rrsi_bundle(_complete_factory()) + evolve(_tasks(), reward, run=run, strategy=SingleSlot(initial_value="seed"), + rounds=1, n_workers=2, repo_path=repo, policies=bundle2, + checkpointing=True) + assert state2.best_score == pytest.approx(state1.best_score) + + +def test_invariant_rejections_advance_the_rrsi_schedule(tmp_path): + """A violates-invariant refusal is still a merge decision: the RRSI anneal + and stall detector must advance once per rejection (review #254 P2). A + one-op SingleSlot proposal passes the critic but fails mutable_keys that + only permit an unrelated key; after a streak, round/progress move and + stall trips, without inventing reward/cost measurements.""" + run, reward = _offline_actors() + cfg = RRsiConfig(b_min=1, b_max=5, T=4, stall_window=3) + bundle, state = build_rrsi_bundle(_complete_factory(), protected=("t3",), + config=cfg) + contract = Contract(invariants=(Invariant("mutable_keys", ("other",)),)) + result = evolve(_tasks(), reward, run=run, + strategy=SingleSlot(initial_value="seed"), rounds=3, + n_workers=2, repo_path=str(tmp_path / "repo"), + policies=bundle, contract=contract) + assert result.outcomes().get("violates-invariant", 0) >= 1 + assert result.outcomes().get("committed", 0) == 0 + n_rej = result.outcomes().get("violates-invariant", 0) + assert state.round == n_rej, ( + f"exactly one schedule advance per invariant refusal; " + f"round={state.round} rejections={n_rej}") + assert state.best_score == 0.0, "S* must stay at base (no invented measurement)" + assert len(state.progress) == n_rej + assert all(p == 0.0 for p in state.progress) + if n_rej >= 3: + assert state.stalled() is True + assert state.budget() < 5, "rejection streak must anneal the edit budget"