diff --git a/CHANGELOG.md b/CHANGELOG.md index 324a29a0..e8cdfef5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,12 @@ All notable changes to AgentDescent are documented here. The format follows `collect` flag is forced on when `audit_drain_per_step > 0`. `EvolutionResult. audit_drained` reports how many ran. Default `0` = the old behaviour exactly. +- **Value-directed budget allocation — `evolve(allocator=...)` / `async_evolve(allocator=...)`.** `BudgetGovernor` degrades optional spend (fusion tournaments, the self-verify rollout) against global floors: once `spent / max_tokens` crosses a threshold, *every* candidate stops getting the expensive part. The EvoAlloc line of work measures the cost of that fixed allocation when evaluation is the scarce resource — this replaces it with a **per-candidate** decision learned from the run's own merge history. + + **`ValueBudgetAllocator`** (`agentdescent/allocator.py`) is a learner with two halves. **Value-directed allocation**: once the token budget is tight, the self-verify rollout is spent only on candidates the value model expects to commit — a `LearningValueModel`, a tiny online logistic fit over the **pre-spend features the worker actually decides on** (the rollout's group-relative advantage, the diff's size, the artifact's blast radius), with a logit clip and L2 so one outlier cannot blow the weights. The model is trained on exactly that vector — post-outcome signals (delta, P(improve), stable distance) are kept for the audit trail but never enter `AllocatorContext.features` — so it cannot fit the labels with signals the decision never sees. **Counterfactual exploration**: a candidate the model judged unworthy is skipped — except with epsilon probability, when it is evaluated anyway; the outcome is then ground truth the model was wrong about, and it learns from it. Without this term a value-based allocator only ever sees the candidates it already believed in, and a biased prior is never corrected — the selection-bias collapse this term exists to prevent. + + Wired end to end: the worker asks the governor `candidate_worth(ctx)` before the self-verify rollout; the aggregator reports **every** measured decision exactly once (`observe(ctx, committed)`), with oracle rejection and acceptance rejection as negative labels — a model fit on commits alone never leaves warm-up. **Every commit is attributed to a worker's actual pre-spend vector**: a fused candidate keeps its first contributor's context, and a tournament-selected single proposal is matched to its own winning card — never a losing first entrant, never a fused union size. The feature vector is **scale-normalised**: the raw-op diff size (1..100+) would swamp the advantage (a z-score) and blast radius in a raw logistic fit, so `size` is mapped through `log1p` to `[0, 1]` — verified by a regression where a high-advantage, large-diff candidate still predicts commit. `allocator=None` (the default) is the old behaviour byte for byte — the clock thresholds remain the *ceiling* the allocator may never exceed. 25 tests in `tests/test_allocator.py` (incl. negative-outcome, sync/async feature-parity, end-to-end fused + single-winner regressions, the A/B arm wiring, and the size-scale regression). **`bench/ab_allocator.py`** is the equal-budget A/B (off vs on, real datasets HotpotQA/finer, per-seed quality + measured tokens + a fired check that distinguishes "the allocator helped" from "it never ran"); exported from the package root; `docs/api.md` regenerated. + - **`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. diff --git a/agentdescent/__init__.py b/agentdescent/__init__.py index 9827cd2c..9efdbee9 100644 --- a/agentdescent/__init__.py +++ b/agentdescent/__init__.py @@ -123,6 +123,12 @@ list_checkpoints, ) from .budget import BudgetGovernor, CallBudget, budgeted_completion +from .allocator import ( + AllocatorContext, + AllocatorPolicy, + LearningValueModel, + ValueBudgetAllocator, +) from .executor import Executor, Result, ThreadExecutor from .sandbox import LocalWorkspaceSandbox, SandboxPool, WorkspaceProvider from .sandbox_shared import SharedSandboxPool @@ -448,4 +454,9 @@ "BudgetGovernor", "CallBudget", "budgeted_completion", + # value-directed budget allocation + "AllocatorContext", + "AllocatorPolicy", + "LearningValueModel", + "ValueBudgetAllocator", ] diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 5c1db5f9..74d96671 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -343,6 +343,69 @@ def measured(self) -> bool: return self.base_counts is not None and self.cand_counts is not None +def _allocator_context_from(c: "_Candidate", p_improve: Optional[float] = None, + stable_distance: float = 0.0): + """Build an :class:`~agentdescent.allocator.AllocatorContext` from a candidate. + + The context carries the **pre-spend features the worker decided on** -- + advantage, diff size, blast radius -- so the model learns from exactly the + inputs its decisions were made on. Post-outcome signals (``p_improve``, + ``stable_distance``, the measured delta) are kept on the context for the + audit trail but never enter :meth:`AllocatorContext.features`. + + **Which card is the candidate's origin depends on how it was chosen:** + + * **Fused** (`c.fused`): the candidate is the union of the surviving + proposals, so its commit label attaches to the *first contributor's* + pre-spend vector -- the explicit attribution rule for fusions. The first + card is the one the tournament put forward, the same attribution the + acceptance gate uses when it folds `before_after_delta` back into the + posterior. + * **Not fused**: the tournament chose one single proposal, so the commit + label must attach to *that proposal's* card -- not an arbitrary first + card that may be a losing entrant. Labeling the first (losing) card as + committed teaches the model the wrong proposal succeeded and supplies no + observation for the actual winner. + + Either way the vector is a worker's actual pre-spend input (advantage, + the card's own diff size, the artifact's blast radius) -- never a fused + union size, which is a post-merge artefact no worker decided on. + """ + from .allocator import AllocatorContext + + card = _attributable_card(c) + adv = getattr(card, "advantage", None) + # The worker decided on *this* card's diff, not the merged union. + size = card.diff.size() if card is not None and card.diff is not None else 0 + return AllocatorContext( + before_after_delta=getattr(card, "before_after_delta", None), + advantage=adv, + size=size, + p_improve=p_improve, + stable_distance=stable_distance, + blast_radius=getattr(c.artifact, "blast_radius", 0.2), + ) + + +def _attributable_card(c: "_Candidate"): + """The evidence card a commit outcome should be attributed to. + + Fusions: the first contributor (the card the tournament put forward). + Single winners: the card whose diff *is* the selected diff -- not the + first entrant, which may be a losing proposal. Falls back to the first + card when none matches (a custom fusion policy may return a diff outside + ``c.cards``), since there is then no better attribution available.""" + if not c.cards: + return None + if getattr(c, "fused", False): + return c.cards[0] + selected = getattr(c, "diff", None) + for card in c.cards: + if card.diff is selected or getattr(card.diff, "diff_id", None) == \ + getattr(selected, "diff_id", None): + return card + return c.cards[0] + class EvidenceBuffer: """Cards bucketed by target artifact (design doc, section 4.1). @@ -573,6 +636,12 @@ def __init__( self.acceptance_policy, self.promotion_policy): install_policy(policy, verifier, cfg) self.buffer = EvidenceBuffer() + #: Optional value allocator (see :mod:`agentdescent.allocator`), wired + #: by the engine so the merge path can feed commit outcomes back into + #: it. ``None`` keeps the old behaviour; the aggregator never consults + #: the allocator for a decision (the worker does, through the budget + #: governor), it only reports what happened. + self.allocator: Optional[Any] = None self._posteriors: Dict[str, BetaPosterior] = defaultdict(BetaPosterior) #: How many queued audits the L-value consumer drained and ran against #: the oracle. Zero when ``audit_drain_per_step`` is 0 (the default). @@ -1269,12 +1338,13 @@ def _decide(self, c: "_Candidate") -> MergeReport: # the annealed threshold -- not a point estimate crossing a line. prior = self._posteriors[c.artifact_id] base_score, cand_score = c.base_cheap, c.cand_cheap + stable_distance = self._stable_distance(c.artifact_id, c.candidate) decision = self.acceptance_policy.accept(MergeContext( artifact=c.artifact, candidate=c.candidate, cards=c.cards, base_counts=c.base_counts, cand_counts=c.cand_counts, diff=c.diff, base_cheap=base_score, cand_cheap=cand_score, prior=prior, trust_radius=c.artifact.blast_radius, - stable_distance=self._stable_distance(c.artifact_id, c.candidate))) + stable_distance=stable_distance)) p_improve = decision.p_improve # The full held-out rates. The regression guard below must use *these* @@ -1292,12 +1362,28 @@ def report(diff, version, reason, category) -> MergeReport: c.survived, c.discarded, c.conflicts, p_improve, version, reason, category) + # Report *every* measured decision to the value allocator exactly once: + # the label is "was this candidate worth the expensive evaluation it + # got?", and it is True only on a commit. Oracle rejection and + # acceptance rejection are negative labels too -- without them the model + # would learn only from commits and CAS conflicts, so a rejection-only + # run never left warm-up and a mixed run never learned which evaluated + # proposals fail. + def _observe_outcome(committed: bool) -> None: + if self.allocator is not None: + observe = getattr(self.allocator, "observe", None) + if callable(observe): + observe(_allocator_context_from( + c, p_improve, stable_distance=stable_distance), committed) + rejected = self._audit(c.artifact, c.artifact_id, c.candidate, c.diff, base_full, cand_full, base_score, cand_score, prior) if rejected: + _observe_outcome(False) return report(None, None, "oracle rejected", MergeOutcome.ORACLE_REJECTED) if not decision.accept: + _observe_outcome(False) prior.observe_delta(decision.observed_delta) return report(None, None, decision.detail, MergeOutcome(decision.category)) @@ -1305,7 +1391,9 @@ def report(diff, version, reason, category) -> MergeReport: # -- commit (section 4.1): CAS on dev -------------------------------- new_version = self._commit_with_retry(c.artifact_id, c.candidate, c.diff, c.head) - if new_version is None: + committed = new_version is not None + _observe_outcome(committed) + if not committed: self.buffer.settle(c.survivor_cards) return report(None, None, "CAS conflict", MergeOutcome.CAS_CONFLICT) diff --git a/agentdescent/allocator.py b/agentdescent/allocator.py new file mode 100644 index 00000000..d241e78b --- /dev/null +++ b/agentdescent/allocator.py @@ -0,0 +1,396 @@ +"""Allocating the scarce part of a budget by candidate value, not by clock. + +``BudgetGovernor`` degrades optional spend (fusion tournaments, self-verify +rollouts) against a global floor: once ``spent / max_tokens`` crosses a +threshold, *every* candidate stops getting the expensive part. That is fair and +predictable, and it is exactly what a budget-aware allocator should *not* do. + +The EvoAlloc line of work measures the cost of a fixed allocation strategy: when +evaluation is the scarce resource, spending it by time-since-start rather than by +expected value wastes it on candidates that cannot move the search while starving +ones that can. This module replaces the global threshold with a **per-candidate** +decision -- ``is this candidate worth the expensive evaluation?`` -- learned from +the run's own merge history, plus a **counterfactual exploration** term that +occasionally evaluates a candidate the allocator judged unworthy, so the value +model sees the ground truth it was wrong about. + +Three pieces: + +* :class:`AllocatorContext` -- the features one candidate carries into the + decision. Everything here is arithmetic over numbers the engine already has + on the merge path (``before_after_delta``, advantage, diff size, P(improve)), + so the allocator adds no new measurement, only a new reader of existing ones. +* :class:`AllocatorPolicy` -- the seam. ``decide(ctx)`` says whether this + candidate is worth a deep evaluation; ``observe(ctx, outcome)`` feeds the + result of whatever was actually spent back into the model. +* :class:`ValueBudgetAllocator` -- the shipped learner. A logistic-style value + model over the candidate features, fit online against commit outcomes, with an + epsilon-greedy counterfactual term that re-measures the candidates the model + judged worst. + +The allocator is **off by default**: ``BudgetGovernor(allocator=None)`` (the +default) is the old global-threshold behaviour byte for byte. Pass one and the +governor asks it before spending the expensive part of a candidate's merge, and +tells it what happened afterwards. + +The analogy to the rest of the library: the aggregator already learns *what to +keep* (Beta posteriors, adaptive trust region). This learns *how much to spend +finding out* -- the resource-allocation layer between ``max_tokens`` (the cap) +and the merge decisions (the value). +""" + +from __future__ import annotations + +import math +import random +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Protocol, Sequence, Tuple, runtime_checkable + +__all__ = [ + "AllocatorContext", + "AllocatorPolicy", + "ValueBudgetAllocator", + "LearningValueModel", +] + + +# --------------------------------------------------------------------------- +# The features a candidate carries into the decision +# --------------------------------------------------------------------------- + + +@dataclass +class AllocatorContext: + """One candidate, as the allocator sees it. + + All fields are arithmetic over numbers the engine already computes; the + allocator adds no new measurement, only a new reader of existing ones. + + **The features the model learns on are exactly the features the worker can + see *before* the expensive evaluation.** The decision happens at the + worker, which knows the proposing rollout's advantage, the diff's size and + the artifact's blast radius -- but not the before/after delta (that is the + self-verify rollout's own product), P(improve), or stable distance (both + computed later in the merge). Training on those post-outcome features while + deciding on the pre-spend ones fits a model that explains the labels without + teaching a usable pre-spend predictor. So :meth:`features` returns **only** + the pre-spend vector; the post-outcome fields exist for diagnostics and the + audit trail, never as model inputs. + """ + + #: The local before/after delta the proposing worker measured. **Not a + #: model input** (unavailable at decision time) -- kept for diagnostics. + before_after_delta: Optional[float] = None + #: Group-relative advantage (GRPO-style z-score) of the proposing rollout. + #: ``None`` when the group was too small to standardise against. + advantage: Optional[float] = None + #: ``len(diff.ops)`` -- the trust region's own notion of diff size. + size: int = 0 + #: The cheap layer's P(improve), when the candidate reached acceptance. + #: **Not a model input** (computed in the merge, after the decision). + p_improve: Optional[float] = None + #: The learned layer's uncertainty (``learned_eval``), when available. + #: **Not a model input.** + uncertainty: Optional[float] = None + #: How far the candidate sits from the confirmed ``stable`` branch. + #: **Not a model input** (computed in the merge). + stable_distance: float = 0.0 + #: The artifact's blast radius. Known up front, so it *is* a model input. + blast_radius: float = 0.2 + + def features(self) -> Tuple[float, ...]: + """The **pre-spend** feature vector the value model learns and predicts + on: ``(advantage, size, blast_radius)``. + + Exactly the features the worker has when it asks "is this worth the + self-verify rollout?". Training on the same vector (see + :func:`agentdescent.aggregator._allocator_context_from`) is what keeps + decision and feedback inputs identical, so the model cannot fit the + labels with post-outcome features that the decision never sees. + + **All three columns are brought to a comparable scale.** A logistic fit + is scale-sensitive: a diff-size column in raw ops (1..100+) would + dominate the group-relative advantage (a z-score, roughly -3..3) and the + blast radius ([0, 1]) purely by magnitude, so the learned weights — and + therefore the skip decisions — would be driven by size no matter what + the advantage said. ``size`` is therefore mapped through + ``log1p`` and normalised to ``[0, 1]`` against a nominal ceiling of + 100 ops (``log1p(100)``), which keeps the three inputs comparable while + preserving order and monotonicity. ``advantage`` is zero-filled when + missing (group too small to standardise); blast radius is already a + fraction. + """ + import math + return ( + self.advantage if self.advantage is not None else 0.0, + math.log1p(max(0, self.size)) / math.log1p(100), + self.blast_radius, + ) + + @property + def value_hint(self) -> float: + """A single scalar summary, for diagnostics and the epsilon anneal. + + The advantage alone, since it is the one pre-spend signal most likely + to separate a committing candidate from a rejected one. Rough by + design -- the model learns the weight; this is only the initial guess + that starts it.""" + return self.advantage if self.advantage is not None else 0.0 + + +# --------------------------------------------------------------------------- +# The decision +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class AllocateDecision: + """What the allocator decided for one candidate.""" + + #: Whether the candidate is worth the expensive evaluation (the fusion + #: tournament, or the self-verify rollout). + deep_eval: bool = True + #: Whether this decision was a counterfactual exploration -- a candidate + #: judged unworthy but evaluated anyway, so the value model can learn what + #: it missed. Diagnostics only; the engine spends the same either way. + counterfactual: bool = False + #: The value the model assigned, before the epsilon roll. Diagnostics. + predicted_value: float = 0.0 + #: How many observations the model had, at decision time. + n_observed: int = 0 + + +# --------------------------------------------------------------------------- +# The seam +# --------------------------------------------------------------------------- + + +@runtime_checkable +class AllocatorPolicy(Protocol): + """Decide, per candidate, whether the expensive part of a merge is worth it. + + ``decide`` is called just before the allocatable spend (the fusion + tournament in the sync path, the self-verify rollout on both). ``observe`` + is called afterwards with what actually happened -- the outcome is the + counterfactual truth the model learns from. Both are optional hooks: a + policy that implements neither is a no-op allocator that always deep-evals + (the old behaviour). + """ + + def decide(self, ctx: AllocatorContext) -> AllocateDecision: ... + def observe(self, ctx: AllocatorContext, committed: bool) -> None: ... + + +# --------------------------------------------------------------------------- +# The online value model +# --------------------------------------------------------------------------- + + +@dataclass +class LearningValueModel: + """A tiny online logistic model over the candidate features. + + ``P(committed | features)`` is what "worth a deep eval" means here: a + candidate the model expects to commit is worth the extra held-out sweep, + one it expects to be rejected is not. Fitted online by gradient ascent on + the run's own commit outcomes, so it needs no labelled data -- the label is + ``did it commit``, which the engine knows for free. + + The learning rule is plain logistic regression with a momentum term + (``lr``), a clip on the logit (``max_logit``, so one extreme candidate + cannot blow the weights), and a small L2 penalty (``reg``) so features that + never move stay near zero. None of these is a research contribution; they + are the smallest thing that learns without diverging on one noisy run. + """ + + #: Learning rate. + lr: float = 0.2 + #: L2 regularisation. + reg: float = 0.05 + #: Logit clip, so a single outlier cannot send the weights to infinity. + max_logit: float = 3.0 + #: Weights, one per feature in :meth:`AllocatorContext.features` (three: + #: advantage, size, blast_radius), plus bias. + _w: List[float] = field(default_factory=lambda: [0.0] * 4) + + def __post_init__(self) -> None: + if len(self._w) != 4: + self._w = [0.0] * 4 + + def predict(self, features: Tuple[float, ...]) -> float: + # The weights are ``[w0, w1, w2, bias]``: the first three match the + # three pre-spend features one-to-one, the last is the intercept. Not + # ``zip(self._w, features)`` -- that would silently match the bias + # weight against a feature and then add it again as the intercept. + logit = self._bias() + sum( + wi * x for wi, x in zip(self._w[:-1], features)) + logit = max(-self.max_logit, min(self.max_logit, logit)) + return 1.0 / (1.0 + math.exp(-logit)) + + def _bias(self) -> float: + return self._w[-1] + + def update(self, features: Tuple[float, ...], committed: bool) -> None: + """One gradient-ascent step toward ``committed``. + + ``p = sigmoid(w·x)``, target ``y ∈ {0, 1}``, gradient + ``(y - p) * x``, plus L2. In-place, so the model is a single mutable + object the allocator thread can step without copies. + """ + p = self.predict(features) + y = 1.0 if committed else 0.0 + err = y - p + for i in range(len(self._w)): + # Every feature column contributes ``err * x_i``; the bias (last + # weight) gets ``err * 1`` -- it must be able to move even when + # every feature is zero, or an all-zero candidate (nothing measured + # on this workload) can never learn a non-neutral prior. + x = features[i] if i < len(features) else 1.0 + grad = err * x - self.reg * self._w[i] + self._w[i] += self.lr * grad + + +# --------------------------------------------------------------------------- +# The shipped allocator +# --------------------------------------------------------------------------- + + +@dataclass +class ValueBudgetAllocator: + """Learn which candidates are worth the expensive evaluation. + + The two halves of the EvoAlloc mechanism, in one object: + + **Value-directed allocation.** When the budget is not tight + (``remaining_fraction`` high) every candidate is deep-evaluated -- there is + budget to spend, and spending it is the old behaviour. As the budget runs + out the allocator consults :class:`LearningValueModel`: candidates it + expects to commit still get the deep eval, candidates it expects to be + rejected are skipped. The expensive spend goes where the model says the + value is. + + **Counterfactual exploration.** A candidate the model judges unworthy is + skipped -- except with probability ``epsilon``, when it is evaluated anyway. + The outcome (``observe``) is then a *counterfactual*: the run paid for a + measurement it would have skipped, and the model learns "this looked + worthless and actually committed" (or the reverse). Without this term the + model only ever sees the candidates it already believed in, and a biased + prior is never corrected -- the classic selection-bias collapse of any + value-based allocator. + + ``epsilon`` anneals from ``epsilon_start`` down to ``epsilon_end`` over the + first ``epsilon_anneal`` observations: early, the model knows nothing and + explores hard; late, it has seen the workload and exploits. The anneal is + on *observations*, not wall-clock, so a fast run and a slow run spend the + same exploration budget. + """ + + #: Below this remaining fraction the allocator starts skipping low-value + #: candidates. Above it, everything deep-evals (the old behaviour). + budget_start: float = 0.5 + #: The model used to predict which candidates commit. + model: LearningValueModel = field(default_factory=LearningValueModel) + #: Counterfactual exploration probability, at the start of the run. + epsilon_start: float = 0.3 + #: ...and after the anneal. + epsilon_end: float = 0.05 + #: Observations over which epsilon anneals. + epsilon_anneal: int = 40 + #: Number of observations the model must have before it is trusted to skip + #: anything. Before that, everything deep-evals -- the model has no opinion + #: yet, and skipping on no evidence is the bias the exploration exists to + #: avoid. + min_observations: int = 5 + #: Features whose |weight| is below this are treated as "no signal" for the + #: skip decision, so a model fit on a workload where advantage never moved + #: cannot skip on advantage noise. + min_weight: float = 0.01 + #: Seed for the counterfactual-exploration RNG. ``decide`` must not draw + #: from the process-global ``random``: ``evolve(seed=...)`` uses a *local* + #: RNG for task shuffling, so a global draw would make exploration depend on + #: unrelated process state and, with concurrent runs, on scheduling. Each + #: allocator owns a ``random.Random(self.seed)``, so two runs with the same + #: seed explore identically and two concurrent runs with different seeds do + #: not interfere. + seed: int = 0 + + # -- state --------------------------------------------------------------- + + _n_obs: int = 0 + _n_skipped: int = 0 + _n_counterfactual: int = 0 + _n_committed: int = 0 + _rng: random.Random = field(default=None, repr=False) + + def __post_init__(self) -> None: + if self._rng is None: + self._rng = random.Random(self.seed) + + @property + def n_observed(self) -> int: + return self._n_obs + + @property + def n_skipped(self) -> int: + return self._n_skipped + + @property + def n_counterfactual(self) -> int: + return self._n_counterfactual + + @property + def n_committed(self) -> int: + return self._n_committed + + def _epsilon(self) -> float: + t = self._n_obs + if self.epsilon_anneal <= 0: + return self.epsilon_end + progress = min(1.0, t / self.epsilon_anneal) + return self.epsilon_end + (self.epsilon_start - self.epsilon_end) \ + * (1.0 - progress) + + def decide(self, ctx: AllocatorContext) -> AllocateDecision: + """Whether ``ctx``'s candidate is worth the expensive evaluation. + + Always deep-evals when the model has no opinion yet + (``min_observations``), or when this allocator is not under budget + pressure. Otherwise consults the model and rolls the counterfactual + epsilon. + """ + if self._n_obs < self.min_observations: + return AllocateDecision(deep_eval=True, n_observed=self._n_obs) + features = ctx.features() + p = self.model.predict(features) + # A candidate the model cannot distinguish from noise (every |w| tiny, + # or the logit near the prior) is not confidently low-value; skipping + # it is the bias, not the allocation. + confident = any(abs(w) >= self.min_weight for w in self.model._w[:-1]) + if not confident or p >= 0.5: + return AllocateDecision( + deep_eval=True, predicted_value=p, n_observed=self._n_obs) + # Low value: skip, unless this is a counterfactual exploration. The + # draw comes from this allocator's own seeded RNG, not the process + # global -- so exploration is reproducible per run and isolated between + # concurrently running seeds. + if self._rng.random() < self._epsilon(): + self._n_counterfactual += 1 + return AllocateDecision( + deep_eval=True, counterfactual=True, + predicted_value=p, n_observed=self._n_obs) + self._n_skipped += 1 + return AllocateDecision( + deep_eval=False, predicted_value=p, n_observed=self._n_obs) + + def observe(self, ctx: AllocatorContext, committed: bool) -> None: + """Feed a commit outcome back into the value model. + + Called for every candidate the merge path actually evaluated -- the + skipped ones were never evaluated, so there is no outcome to learn + from; the counterfactual ones were, and their outcome is the point of + having spent the budget on them. + """ + self._n_obs += 1 + if committed: + self._n_committed += 1 + self.model.update(ctx.features(), committed) diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py index 4bfd650b..a7490644 100644 --- a/agentdescent/async_evolve.py +++ b/agentdescent/async_evolve.py @@ -58,6 +58,24 @@ from .staleness import StaleAction, StalenessPolicy, get_policy +def _async_allocator_context(diff, advantage, blast_radius: float = 0.2) -> Any: + """Build an :class:`~agentdescent.allocator.AllocatorContext` at an async + worker -- the decision side of the allocator loop. Same pre-spend features + as the synchronous worker's: the diff's size, the group-relative advantage + the proposing rollout already measured, and the artifact's blast radius. + ``before_after_delta`` / ``p_improve`` / ``stable_distance`` are the + post-outcome signals the merge computes later; they stay off the model's + feature vector (see :class:`~agentdescent.allocator.AllocatorContext`).""" + from .allocator import AllocatorContext + + return AllocatorContext( + before_after_delta=None, # the self-verify's product, not yet known + advantage=advantage, + size=diff.size() if diff is not None else 0, + blast_radius=blast_radius, + ) + + def async_evolve( tasks, reward: Reward, @@ -117,6 +135,13 @@ def async_evolve( #: known, because an opaque `run` cannot report tokens. usage: Optional[Usage] = None, policies: Optional["Policies"] = None, + #: Optional value allocator (see :mod:`agentdescent.allocator`). Pass one + #: and, once the token budget is tight, the self-verify rollout is spent + #: only on candidates the allocator's value model expects to commit; the + #: merge path reports every outcome back so the model learns. ``None`` (the + #: default) is the old behaviour exactly -- optional spend degrades against + #: the global floors, every candidate treated alike. + allocator: Optional[Any] = None, ) -> EvolutionResult: """Evolve an artifact **without a round barrier**. @@ -338,6 +363,14 @@ def async_evolve( acceptance rule and sees a finished run would reasonably conclude it ran. New capabilities go here rather than adding another parameter to a function that already has thirty-five. + allocator: + Exactly as in :func:`~agentdescent.evolution.evolve`: value-directed + budget allocation (see :mod:`agentdescent.allocator`). ``None`` (the + default) is the old behaviour exactly. Pass a value allocator and, once + the token budget is tight, the self-verify rollout is spent only on + candidates the allocator expects to commit; the merger reports every + outcome back so it learns. Requires ``max_tokens``. Counterfactual + exploration corrects a biased value model rather than reinforcing it. Returns ------- @@ -382,7 +415,8 @@ def async_evolve( cheap_eval_tasks=cheap_eval_tasks, fusion_tournament=fusion_tournament, shuffle=shuffle, seed=seed, usage=usage, verifier=_pol.verifier, ledger_impl=_pol.ledger, - policies_bundle=_pol, checkpointing=checkpointing) + policies_bundle=_pol, checkpointing=checkpointing, + allocator=allocator) eng.meter.start() # The cost-aware governor for the async path. The merger loop has no round # barrier, so the governor checks per sweep (after each merge, before the @@ -393,7 +427,8 @@ def async_evolve( # eng alongside it. governor = BudgetGovernor(max_tokens=max_tokens, stop_on_diminishing=stop_on_diminishing_returns, - efficiency_floor=efficiency_floor) + efficiency_floor=efficiency_floor, + allocator=allocator) eng.governor = governor if n_workers < 1: raise ValueError(f"n_workers must be >= 1, got {n_workers}") @@ -671,7 +706,13 @@ def _worker(wid: int, shard: List[Task]) -> None: # diff applied for a before/after signal. Faithful repos that # only score the candidate on held-out (e.g. EvoSkill) pass # self_verify=False to skip this extra rollout. - if self_verify and governor.allow_self_verify(): + # With an allocator installed the decision becomes + # per-candidate once the budget is tight (see + # :mod:`agentdescent.allocator`). + if self_verify and governor.candidate_worth( + _async_allocator_context( + diff, adv, + getattr(artifact, "blast_radius", 0.2))): after = _checked_reward( eng.reward(task, eng.run(artifact.apply(diff).render(), task)), task) delta = after - score diff --git a/agentdescent/budget.py b/agentdescent/budget.py index dc3eb284..2c14cf20 100644 --- a/agentdescent/budget.py +++ b/agentdescent/budget.py @@ -92,6 +92,16 @@ class BudgetGovernor: #: How many efficiency samples before the rule may fire. Four rounds is the #: floor at which "the peak" and "the recent rate" are different claims. min_efficiency_samples: int = 4 + #: Optional per-candidate value allocator (see :mod:`agentdescent.allocator`). + #: ``None`` (the default) is the old behaviour exactly: fusion and + #: self-verify degrade against the global ``soft_floor`` / ``hard_floor`` + #: thresholds, every candidate treated alike. With one, the governor asks it + #: whether a candidate is worth the expensive spend once the budget is + #: tight, and feeds the commit outcome back so it learns. The thresholds + #: remain as the *ceiling* the allocator may never exceed -- the allocator + #: can skip a candidate the clock would have deep-evaluated, never the + #: reverse. + allocator: Optional[Any] = None #: Cumulative spend, set by ``spend``. Kept as a field so ``affords`` can #: be asked without re-passing the number the caller just passed. _spent: int = 0 @@ -157,6 +167,50 @@ def allow_self_verify(self) -> bool: return True return self._spent < self.max_tokens * self.hard_floor + # -- value-directed allocation (EvoAlloc-style, off by default) ---------- + + def candidate_worth(self, ctx: Any) -> bool: + """Whether ``ctx``'s candidate is worth the expensive evaluation. + + With no :attr:`allocator` this is :meth:`allow_self_verify` -- the old + behaviour, exactly. With one, and once the budget is tight + (:attr:`remaining_fraction` below the allocator's ``budget_start``), + the allocator decides: a candidate it expects to commit keeps the + spend, one it expects to be rejected loses it. The clock thresholds + remain the *ceiling*: the allocator can skip a candidate the clock + would have spent on, never spend on one the clock refused. + + ``ctx`` is whatever the allocator's ``decide`` takes -- the engine + builds an :class:`~agentdescent.allocator.AllocatorContext` at the call + site. + """ + if self.max_tokens is None: + return True + if self.allocator is None: + return self.allow_self_verify() + decide = getattr(self.allocator, "decide", None) + if not callable(decide): + return self.allow_self_verify() + # The clock's ceiling still applies: no allocator may spend past the + # hard floor, whatever value it predicts. + if self._spent >= self.max_tokens * self.hard_floor: + return False + remaining = self.remaining_fraction() + budget_start = getattr(self.allocator, "budget_start", 0.5) + if remaining >= budget_start: + return True + return bool(decide(ctx).deep_eval) + + def observe_allocated(self, ctx: Any, committed: bool) -> None: + """Feed a merge outcome back to the allocator. + + Called by the engine for every candidate that was actually evaluated, + so the value model learns which features predict a commit. No-op with + no allocator.""" + observe = getattr(self.allocator, "observe", None) + if callable(observe): + observe(ctx, committed) + def affords_next_round(self) -> bool: """Whether the projection says one more round fits the budget. diff --git a/agentdescent/dataloader.py b/agentdescent/dataloader.py index 2c7a8bae..33c8a49a 100644 --- a/agentdescent/dataloader.py +++ b/agentdescent/dataloader.py @@ -258,6 +258,61 @@ def hf_feature_names(dataset: str, split: str, feature: str, *, raise KeyError(f"feature {feature!r} not found in {dataset}") +# --------------------------------------------------------------------------- +# ModelScope mirror (for datasets whose HuggingFace endpoint is unreachable) +# --------------------------------------------------------------------------- + +#: A ModelScope-hosted copy of HotpotQA, fetched as a plain JSON file. The +#: HuggingFace datasets-server can be unreachable from some networks while +#: ModelScope stays up, so :func:`hf_rows_modelscope` is the fallback that +#: loads the *same* rows from a different mirror. +MODELSCOPE_HOTPOTQA_URL = ( + "https://www.modelscope.cn/api/v1/datasets/OpenDataLab/HotpotQA/repo" + "?Revision=master&FilePath=raw%2Fhotpot_dev_distractor_v1.json" +) + + +def _hotpotqa_context_to_dict(context) -> dict: + """Normalise a HotpotQA ``context`` to the dict shape the engine expects. + + The HuggingFace datasets-server already yields ``{"title": [...], + "sentences": [[...], ...]}``; the raw JSON mirror yields the original + ``[[title, [sentence, ...]], ...]`` pairs. Callers (GEPA's + ``_render_context``) read ``context["title"]`` and ``context["sentences"]``, + so the pairs are converted here, once, at load time. + """ + if isinstance(context, dict): + return context + titles, sents = [], [] + for title, sentences in context: + titles.append(title) + sents.append(list(sentences)) + return {"title": titles, "sentences": sents} + + +def hf_rows_modelscope(url: str = MODELSCOPE_HOTPOTQA_URL, *, + limit: int = 100, cache: bool = True, + timeout: float = 60.0) -> List[dict]: + """Load HotpotQA rows from a ModelScope mirror (plain JSON file). + + Equivalent to :func:`hf_rows` for the ``hotpotqa/hotpot_qa`` validation + split, but via a ModelScope-hosted raw file instead of the HuggingFace + datasets-server -- for networks where one reachable mirror and not the + other. Rows are the same fields (``context`` / ``question`` / ``answer``), + with ``context`` normalised to the ``{"title": ..., "sentences": ...}`` + dict the examples read. + + Cached under ``~/.cache/agentdescent/modelscope/`` keyed by the URL, so a + repeat call does not re-download the ~46 MB file. + """ + data = fetch_bytes(url, cache_subdir="modelscope", timeout=timeout) + rows = json.loads(data) + out: List[dict] = [] + for r in rows[:limit]: + out.append({**r, "context": _hotpotqa_context_to_dict(r.get("context", []))}) + return out + + # --------------------------------------------------------------------------- # Gated datasets (optional, lazy `datasets` import) # --------------------------------------------------------------------------- diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index 193b366e..06b7f0d8 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -314,6 +314,30 @@ def _checked_reward(value, task: "Task") -> float: _EvalCache = MemoryCache +def _allocator_context(diff: "Diff", advantage: Optional[float], + score: float, blast_radius: float = 0.2) -> Any: + """Build an :class:`~agentdescent.allocator.AllocatorContext` at the worker. + + The worker builds the **pre-spend** features the allocator decides on -- + the proposing rollout's group-relative advantage, the diff's size, and the + artifact's blast radius. The before/after delta is the self-verify + rollout's own product and is not known yet; ``p_improve`` and + ``stable_distance`` are computed later in the merge. So the decision + context leaves those ``None``/default, and :meth:`AllocatorContext.features` + -- the vector the model actually sees -- returns only the pre-spend + columns, which is exactly what the aggregator trains on too (see + :func:`~agentdescent.aggregator._allocator_context_from`). + """ + from .allocator import AllocatorContext + + return AllocatorContext( + before_after_delta=None, # the self-verify's product, not yet known + advantage=advantage, + size=diff.size() if diff is not None else 0, + blast_radius=blast_radius, + ) + + class EvolvingArtifact: """An :class:`~agentdescent.evolvable.Evolvable`: flat state + a strategy. @@ -1875,7 +1899,8 @@ def _build_engine(tasks, reward, *, agent, run, propose, strategy, initial_state verifier: Optional[Any] = None, ledger_impl: Optional[Any] = None, policies_bundle: Optional[Policies] = None, - checkpointing: bool = False) -> _Engine: + checkpointing: bool = False, + allocator: Optional[Any] = None) -> _Engine: """Wire the ledger, runtime, verifier and aggregator (shared by :func:`evolve` and :func:`~agentdescent.async_evolve.async_evolve`).""" import tempfile @@ -2156,6 +2181,14 @@ def _default_aggregator(ledger, verifier, audit, config, policy): aggregator = (aggregator_factory or _default_aggregator)( ledger, verifier, AuditScheduler(), cfg, staleness_policy) + # Wire the value allocator (if any) so the merge path can report commit + # outcomes back into it. The worker asks the allocator its per-candidate + # spend decision through the budget governor; this is the feedback half. + if allocator is not None and hasattr(aggregator, "allocator"): + try: + aggregator.allocator = allocator + except AttributeError: # slots, or a read-only property + pass # 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. @@ -2342,6 +2375,13 @@ def evolve( #: are known, because an opaque `run` cannot report tokens. usage: Optional[Usage] = None, policies: Optional["Policies"] = None, + #: Optional value allocator (see :mod:`agentdescent.allocator`). Pass one + #: and, once the token budget is tight, the self-verify rollout is spent + #: only on candidates the allocator's value model expects to commit; the + #: merge path reports every outcome back so the model learns. ``None`` (the + #: default) is the old behaviour exactly -- optional spend degrades against + #: the global floors, every candidate treated alike. + allocator: Optional[Any] = None, ) -> EvolutionResult: """Evolve an artifact. Provide either ``agent`` (with ``solve``/``propose``) or the ``run`` / ``propose`` callables directly. @@ -2663,6 +2703,21 @@ def evolve( acceptance rule and sees a finished run would reasonably conclude it ran. New capabilities go here rather than adding another parameter to a function that already has thirty-five. + allocator: + Value-directed budget allocation (see :mod:`agentdescent.allocator`). + ``None`` (the default) is the old behaviour exactly: optional spend + (fusion tournaments, the self-verify rollout) degrades against the + global ``soft_floor`` / ``hard_floor`` thresholds as the token budget + runs out, every candidate treated alike. Pass a + :class:`~agentdescent.allocator.ValueBudgetAllocator` (or any object + with ``decide(ctx) -> AllocateDecision`` and ``observe(ctx, committed)``) + and, once the budget is tight, the self-verify rollout is spent only on + candidates the allocator's value model expects to commit; the merge + path reports every outcome back so the model learns. Requires + ``max_tokens`` to be set (the allocator decides how the budget is + spent; ``max_tokens`` is what exists to spend). Counterfactual + exploration re-measures candidates the model judged unworthy, so a + biased value model is corrected rather than reinforced. Returns ------- @@ -2747,7 +2802,7 @@ def evolve( policies=policies, checkpointing=checkpointing, stop_on_diminishing_returns=stop_on_diminishing_returns, efficiency_floor=efficiency_floor, - call_budget=call_budget) + call_budget=call_budget, allocator=allocator) if pipelined_gate: # The mirror of the block above, and the same reasoning: a knob accepted @@ -2806,7 +2861,8 @@ def evolve( cheap_eval_tasks=cheap_eval_tasks, fusion_tournament=fusion_tournament, shuffle=shuffle, seed=seed, usage=usage, verifier=_pol.verifier, ledger_impl=_pol.ledger, - policies_bundle=_pol, checkpointing=checkpointing) + policies_bundle=_pol, checkpointing=checkpointing, + allocator=allocator) # Start the clock after the wiring, before the first unit of work: setup # is not what a time-to-quality number is asking about. eng.meter.start() @@ -2855,7 +2911,8 @@ def evolve( # clean merge instead of being cut mid-round. Inert without max_tokens. governor = BudgetGovernor(max_tokens=max_tokens, stop_on_diminishing=stop_on_diminishing_returns, - efficiency_floor=efficiency_floor) + efficiency_floor=efficiency_floor, + allocator=allocator) for r in range(rounds): if deadline is not None and time.time() >= deadline: stop_reason = "max_seconds" @@ -3097,7 +3154,12 @@ def _run_unit_inner(unit) -> None: # the delta it produces feeds the acceptance test's tie-breaker # (``observe_delta``), never the commit gates, so skipping it costs # ranking precision on the advantage signal and nothing else. - if self_verify and governor.allow_self_verify(): + # With an allocator installed the decision becomes per-candidate: + # once the budget is tight it spends the self-verify rollout only on + # candidates its value model expects to commit (see + # :mod:`agentdescent.allocator`). + if self_verify and governor.candidate_worth(_allocator_context( + diff, adv, score, getattr(mine, "blast_radius", 0.2))): after = _checked_reward( reward(task, run(mine.apply(diff).render(), task)), task) delta = after - score diff --git a/bench/ab_allocator.py b/bench/ab_allocator.py new file mode 100644 index 00000000..a9ef13a0 --- /dev/null +++ b/bench/ab_allocator.py @@ -0,0 +1,336 @@ +"""A/B the value-directed budget allocator. Off is the control. + + python -m bench.ab_allocator --dataset hotpotqa --budget-rollouts 96 \ + --seeds 0,1,2 --provider claude --model GLM-5.2 --yes + +`agentdescent/allocator.py` replaces `BudgetGovernor`'s clock-based degradation +(once `spent / max_tokens` crosses a floor, *every* candidate stops getting the +expensive self-verify rollout) with a per-candidate decision learned from the +run's own merge history. Every one of its knobs is off by default because *an +analogy without an A/B is decoration*; this script is that A/B. + +Identical workload, identical budget, identical seeds, one argument different: + + off allocator=None -- the clock degrades optional spend + on allocator=ValueBudgetAllocator() -- learned per-candidate spend + +Two things it reports that a quality column alone would hide: + +**Whether the allocator fired at all.** An allocator that never skipped a +candidate (observed > 0 but skipped == 0 and counterfactual == 0) is a control +against a control: it only ever agreed with the clock, and its token column +says nothing about the mechanism. Both are printed and are the first thing to +read before the quality column. + +**Tokens, not just quality.** The allocator's promise is *spending the scarce +resource on the right candidates*, which shows up as equal-or-better quality at +fewer tokens (or better quality at the same tokens). The `tokens` column is the +measured spend, not the requested budget, so a run that overshoots the token +cap is visible rather than silently counted as within it. + +If the allocator does not win here, delete it and record the negative result in +`docs/budget.md`. That is the outcome this mechanism's own docstring asks for +explicitly, and it is worth more than an unvalidated field nobody reads. +""" + +from __future__ import annotations + +import argparse +import statistics +import sys +import threading +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from agentdescent import Usage +from agentdescent.allocator import ValueBudgetAllocator +from agentdescent.evolution import evolve + +from examples._common import completion_for, confirm + +from .baselines_run import _finer, _fmt, _hotpotqa + + +def _arm(on: bool, alloc_kwargs: Optional[Dict[str, Any]] = None, + seed: int = 0) -> Tuple[Dict[str, Any], Optional[ValueBudgetAllocator]]: + """The `evolve()` keyword arguments for one arm, and the thing that watched it. + + The control is *not* "an empty allocator" -- it is no allocator at all, the + engine's default. Otherwise the comparison could be measuring the wiring + rather than the mechanism. The second return value is the allocator itself + for the `on` arm (it records every decision and observation) and `None` for + the control. ``alloc_kwargs`` are passed to the allocator constructor (e.g. + ``min_observations`` / ``budget_start`` from the command line); ``seed`` + seeds the allocator's counterfactual-exploration RNG so a run's exploration + is reproducible and isolated from other concurrently running seeds. + """ + if not on: + return {}, None + alloc = ValueBudgetAllocator(seed=seed, **(alloc_kwargs or {})) + return {"allocator": alloc}, alloc + + +def _fired_counts(watcher) -> Tuple[int, int]: + """``(fired, observed)``: whether the allocator ever *changed the spend*. + + `observed` is every merge decision it saw; `fired` counts only decisions + where the allocator actually *differed* from the clock -- a real skip. A + counterfactual exploration is a re-measurement of a candidate the allocator + judged unworthy, and it deep-evals just like the clock would, so it is + *not* a spend change: counting it as fired would let an allocator that + never skipped anything pass the control-vs-control safeguard. + Counterfactuals are reported separately via ``watcher.n_counterfactual``. + """ + if watcher is None: + return 0, 0 + return watcher.n_skipped, watcher.n_observed + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--dataset", choices=["hotpotqa", "finer"], default="hotpotqa") + p.add_argument("--source", choices=["hf", "modelscope"], default="hf", + help="HotpotQA data source: the HuggingFace datasets-server " + "('hf', default) or the ModelScope mirror ('modelscope') " + "for networks where one is reachable and the other is not") + p.add_argument("--budget-rollouts", type=int, default=96) + p.add_argument("--min-observations", type=int, default=None, + help="allocator `min_observations` (merge decisions before it " + "may skip). Lower than the default 5 for workloads whose " + "whole run has few merges, so the A/B can actually see " + "the allocator act") + p.add_argument("--budget-start", type=float, default=None, + help="allocator `budget_start` -- the remaining fraction below " + "which the allocator starts skipping low-value candidates. " + "Default 0.5") + p.add_argument("--workers", type=int, default=4) + p.add_argument("--seeds", default="0,1,2") + p.add_argument("--fetch", type=int, default=48) + p.add_argument("--pool", type=int, default=800) + p.add_argument("--top-k", type=int, default=120) + p.add_argument("--provider", default="claude", + choices=["claude", "openai", "glm"]) + p.add_argument("--model", default="GLM-5.2") + p.add_argument("--max-tokens", type=int, default=None, + help="token ceiling for both arms (the allocator's own knob " + "is `budget_start`; this is the ceiling it spends). " + "None means unbounded -- the allocator stays inert " + "below its budget pressure and the arms are then the " + "same run, so set it to make the A/B mean something.") + p.add_argument("--json", dest="json_out") + p.add_argument("--no-self-verify", dest="self_verify", action="store_false", + help="skip the second rollout per proposal. Identical on both " + "arms, so the A/B stays valid, and it halves the bill") + p.add_argument("--no-thinking", dest="no_thinking", action="store_true", + help="ask an Anthropic-shaped endpoint for no reasoning " + "tokens. Applied to *both* arms or they are not " + "comparable") + p.add_argument("--eval-concurrency", type=int, default=8) + p.add_argument("--run-concurrency", type=int, default=1, + help="how many of the (arm, seed) runs to execute at once") + p.add_argument("--plan", action="store_true") + p.add_argument("--yes", action="store_true") + return p + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = build_parser().parse_args(argv) + seeds = [int(s) for s in args.seeds.split(",") if s.strip()] + + print(f"Mechanism: value-directed budget allocator") + print(f"Dataset : {args.dataset}") + print(f"Budget : {args.budget_rollouts} rollouts, {args.workers} workers" + + (f", max_tokens={args.max_tokens:,}" if args.max_tokens else "")) + print(f"Seeds : {seeds}") + print(f"Runs : {2 * len(seeds)} evolve() calls (off and on, per seed)") + print(f"self_verify: {args.self_verify}") + if args.max_tokens is None: + print("warning: --max-tokens unset; the allocator stays inert below its " + "budget pressure, so the two arms are the same run. Set it to " + "make the A/B mean something.", file=sys.stderr) + if len(seeds) < 3: + print(f"warning: {len(seeds)} seed(s); one number per arm is not a result.", + file=sys.stderr) + if args.plan: + return 0 + if not confirm(args): + return 0 + + total = Usage() + shared = {"self_verify": args.self_verify, + "eval_concurrency": args.eval_concurrency} + # Optional allocator configuration from the command line (so a workload + # whose whole run has few merges can still exercise the skip decision). + alloc_kwargs = {} + if args.min_observations is not None: + alloc_kwargs["min_observations"] = args.min_observations + if args.budget_start is not None: + alloc_kwargs["budget_start"] = args.budget_start + + def _build(seed: int, usage: Usage): + completion = completion_for(args, usage=usage) + return (_hotpotqa(args.fetch, seed, completion, source=args.source, **shared) + if args.dataset == "hotpotqa" + else _finer(args.pool, args.top_k, seed, completion, **shared)) + + # Warm the dataset cache once, before any thread starts. + _build(seeds[0], Usage()) + printing = threading.Lock() + + def _one(job) -> Optional[Dict[str, Any]]: + seed, on = job + label = f"allocator {'on ' if on else 'off'} seed={seed}" + usage = Usage() + try: + workload = _build(seed, usage) + except Exception as e: # noqa: BLE001 - data load failure is one arm's + # problem, not six + with printing: + print(f" {label} BUILD FAILED {type(e).__name__}: {str(e)[:120]}", + file=sys.stderr) + return {"seed": seed, "arm": "on" if on else "off", + "test": None, "dev": None, "rollouts": 0, "calls": 0, + "tokens": 0, "committed": 0, "oracle_rejected": 0, + "fired": None, "fired_of": None, + "skipped": None, "counterfactual": None, + "error": f"build failed: {type(e).__name__}: {str(e)[:120]}"} + kwargs = dict(workload.evolve_kwargs) + arm_kwargs, watcher = _arm(on, alloc_kwargs) + kwargs.update(arm_kwargs) + kwargs.setdefault("rounds", 10_000) + + try: + result = evolve( + workload.tasks, workload.reward, agent=workload.agent, + strategy=workload.strategy, seed=seed, + n_workers=args.workers, max_concurrency=args.workers, + max_rollouts=args.budget_rollouts, + max_tokens=args.max_tokens, usage=usage, **kwargs) + except Exception as e: # noqa: BLE001 - one dead arm is not six + with printing: + print(f" {label} FAILED {type(e).__name__}: {str(e)[:120]}", + file=sys.stderr) + # Keep the arm's accrued usage and mark it failed, so the summary + # does not silently drop it and compare different surviving seeds + # on the two arms. + tokens = usage.prompt_tokens + usage.completion_tokens + return {"seed": seed, "arm": "on" if on else "off", + "test": None, "dev": None, "rollouts": 0, + "calls": usage.calls, "tokens": tokens, + "committed": 0, "oracle_rejected": 0, + "fired": None, "fired_of": None, + "skipped": None, "counterfactual": None, + "error": f"evolve failed: {type(e).__name__}: {str(e)[:120]}"} + + try: + test = workload.test_eval(result) + except Exception as e: # noqa: BLE001 + with printing: + print(f" {label} scoring failed: {type(e).__name__}", + file=sys.stderr) + test = None + + outcomes = result.outcomes() + fired, observed = _fired_counts(watcher) + tokens = usage.prompt_tokens + usage.completion_tokens + row = { + "seed": seed, "arm": "on" if on else "off", + "test": test, "dev": result.final_reward, + "rollouts": result.rollouts, "calls": usage.calls, + "tokens": tokens, + "committed": outcomes.get("committed", 0), + "oracle_rejected": outcomes.get("oracle-rejected", 0), + "fired": fired, "fired_of": observed, + "skipped": watcher.n_skipped if watcher else 0, + "counterfactual": watcher.n_counterfactual if watcher else 0, + "error": result.error, + } + with printing: + print(f" {label} test={_fmt(test)} dev={_fmt(row['dev'])} " + f"{row['rollouts']} rollouts / {row['calls']} calls / " + f"{tokens:,} tokens +{row['committed']} " + f"-{row['oracle_rejected']} vetoed") + if watcher: + print(f" allocator: {observed} observed, " + f"{watcher.n_skipped} skipped, " + f"{watcher.n_counterfactual} counterfactual") + return row + + jobs = [(seed, on) for seed in seeds for on in (False, True)] + if args.run_concurrency > 1: + from concurrent.futures import ThreadPoolExecutor + with ThreadPoolExecutor(max_workers=args.run_concurrency) as pool: + collected = list(pool.map(_one, jobs)) + else: + collected = [_one(job) for job in jobs] + rows: List[Dict[str, Any]] = [r for r in collected if r is not None] + if not rows: + print("every arm failed; nothing to compare.", file=sys.stderr) + return 4 + + # A comparison is only valid over seeds where BOTH arms survived: if off + # fails on seed 0 and on fails on seed 1, the per-arm columns above would + # be about different seeds. Mark the surviving pairs and say which were + # dropped. + failed = [r for r in rows if r["error"]] + complete_seeds = { + r["seed"] for r in rows + if not r["error"] and len([x for x in rows + if x["seed"] == r["seed"] and x["arm"] != r["arm"] + and not x["error"]]) > 0} + dropped = sorted({r["seed"] for r in rows} - complete_seeds) + failed_tokens = sum(r["tokens"] for r in failed) + print() + print(f"arms: {len(rows)} (2 x {len(seeds)} seeds), " + f"{len(failed)} failed (spent {failed_tokens:,} tokens before " + f"failing)") + if dropped: + print(f"incomplete seed pairs dropped from quality/cost comparison: " + f"{dropped}") + print() + print("| arm | complete seeds | test (min/med/max) | tokens (med) | " + "commits | vetoes | fired |") + print("|---|---|---|---|---|---|---|") + for arm in ("off", "on"): + group = [r for r in rows if r["arm"] == arm and r["seed"] in complete_seeds + and not r["error"]] + if not group: + continue + tests = sorted(r["test"] for r in group if r["test"] is not None) + interval = ("n/a" if not tests else + f"{tests[0]:.3f} / {statistics.median(tests):.3f} / " + f"{tests[-1]:.3f}") + med_tokens = statistics.median(r["tokens"] for r in group) + print(f"| {arm} | {len(group)} | {interval} | {med_tokens:,.0f} | " + f"{sum(r['committed'] for r in group)} | " + f"{sum(r['oracle_rejected'] for r in group)} | " + f"{sum(r['fired'] or 0 for r in group)}/" + f"{sum(r['fired_of'] or 0 for r in group)} |") + + fired_on = sum(r["fired"] or 0 for r in rows if r["arm"] == "on") + observed_on = sum(r["fired_of"] or 0 for r in rows if r["arm"] == "on") + if observed_on and fired_on == 0: + # Before any quality claim: a mechanism that never differed from the + # clock is a control against a control. + print(f"\nallocator never fired: it agreed with the clock on all " + f"{observed_on} decisions (0 skipped). The quality and token " + "columns above are not about the mechanism -- check that " + "--max-tokens is tight enough for the allocator's budget_start, " + "and that self-verify is on (the allocator gates the self-verify " + "rollout; without it there is nothing to gate).") + elif observed_on == 0: + print(f"\nallocator saw no merge decisions on the `on` arm " + f"({observed_on} observed). The arms ran different workloads or " + "nothing was ever evaluated -- the quality column is about " + "nothing.") + + if args.json_out: + import json + with open(args.json_out, "w") as fh: + json.dump({"args": vars(args), "rows": rows}, fh, indent=2) + print(f"\nwrote {args.json_out}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/bench/baselines_run.py b/bench/baselines_run.py index d5c86531..f20d31de 100644 --- a/bench/baselines_run.py +++ b/bench/baselines_run.py @@ -55,11 +55,12 @@ def _hotpotqa(fetch: int, seed: int, completion, *, self_verify: bool = True, eval_concurrency: int = 8, - val_cap: Optional[int] = None) -> Workload: + val_cap: Optional[int] = None, + source: str = "hf") -> Workload: """GEPA's dataset and actor; the engine's own optimizer.""" from examples.gepa import gepa_prompt_evolution as gepa - ds = gepa.load_dataset(fetch, seed=seed) + ds = gepa.load_dataset(fetch, seed=seed, source=source) reward = gepa.make_reward() agent = gepa.gepa_agent(completion) diff --git a/docs/api.md b/docs/api.md index 552d9356..46980a39 100644 --- a/docs/api.md +++ b/docs/api.md @@ -13,7 +13,7 @@ means the parameter has none. Each section links to the page that explains *why* the module is shaped the way it is; this page is the *what*. -336 public names across 55 modules. +340 public names across 56 modules. --- @@ -245,7 +245,8 @@ evolve( stop_when: Optional[Callable[['RoundInfo'], bool]] = None, verbose: bool = False, usage: Optional[Usage] = None, - policies: Optional['Policies'] = None + policies: Optional['Policies'] = None, + allocator: Optional[Any] = None ) -> EvolutionResult ``` @@ -301,6 +302,7 @@ evolve( | `verbose` | `bool` | `False` | Print a line per round. Independent of the `RuntimeWarning` emitted when a run ends early -- that always fires. | | `usage` | `Optional[Usage]` | `None` | Share one `Usage` with your model adapters (`claude(usage=u)`, `openai_compatible(usage=u)`) and the result's token counts become real. Without it the run still reports calls, seconds and failures -- `run` is `(rendered, task) -> str`, so an opaque actor has no way to surface tokens, and inventing a number would be worse than reporting zero. | | `policies` | `Optional['Policies']` | `None` | Bundle of replaceable pieces (`Policies`). Every field defaults to `None` meaning "current behaviour", so `Policies()` and passing nothing are the same run. The individual keyword arguments -- `task_sampler`, `staleness_policy`, `aggregator_factory` -- are shortcuts onto its fields and keep working; an explicit argument wins over a bundle default rather than being silently ignored. Fields whose implementations have not landed yet raise rather than being accepted and ignored: a caller who passes a custom acceptance rule and sees a finished run would reasonably conclude it ran. New capabilities go here rather than adding another parameter to a function that already has thirty-five. | +| `allocator` | `Optional[Any]` | `None` | Value-directed budget allocation (see `allocator`). `None` (the default) is the old behaviour exactly: optional spend (fusion tournaments, the self-verify rollout) degrades against the global `soft_floor` / `hard_floor` thresholds as the token budget runs out, every candidate treated alike. Pass a `ValueBudgetAllocator` (or any object with `decide(ctx) -> AllocateDecision` and `observe(ctx, committed)`) and, once the budget is tight, the self-verify rollout is spent only on candidates the allocator's value model expects to commit; the merge path reports every outcome back so the model learns. Requires `max_tokens` to be set (the allocator decides how the budget is spent; `max_tokens` is what exists to spend). Counterfactual exploration re-measures candidates the model judged unworthy, so a biased value model is corrected rather than reinforced. | ### `reflector(...)` @@ -3143,7 +3145,8 @@ async_evolve( stop_when: Optional[Callable[[RoundInfo], bool]] = None, verbose: bool = False, usage: Optional[Usage] = None, - policies: Optional['Policies'] = None + policies: Optional['Policies'] = None, + allocator: Optional[Any] = None ) -> EvolutionResult ``` @@ -3197,6 +3200,7 @@ async_evolve( | `verbose` | `bool` | `False` | Print one line per merger sweep. | | `usage` | `Optional[Usage]` | `None` | Share one `Usage` with your model adapters (`claude(usage=u)`, `openai_compatible(usage=u)`) and the result's token counts become real. Without it the run still reports calls, seconds and failures -- `run` is `(rendered, task) -> str`, so an opaque actor has no way to surface tokens, and inventing a number would be worse than reporting zero. | | `policies` | `Optional['Policies']` | `None` | Bundle of replaceable pieces (`Policies`). Every field defaults to `None` meaning "current behaviour", so `Policies()` and passing nothing are the same run. The individual keyword arguments -- `task_sampler`, `staleness_policy`, `aggregator_factory` -- are shortcuts onto its fields and keep working; an explicit argument wins over a bundle default rather than being silently ignored. Fields whose implementations have not landed yet raise rather than being accepted and ignored: a caller who passes a custom acceptance rule and sees a finished run would reasonably conclude it ran. New capabilities go here rather than adding another parameter to a function that already has thirty-five. | +| `allocator` | `Optional[Any]` | `None` | Exactly as in `evolve`: value-directed budget allocation (see `allocator`). `None` (the default) is the old behaviour exactly. Pass a value allocator and, once the token budget is tight, the self-verify rollout is spent only on candidates the allocator expects to commit; the merger reports every outcome back so it learns. Requires `max_tokens`. Counterfactual exploration corrects a biased value model rather than reinforcing it. | --- @@ -3594,6 +3598,14 @@ Whether a candidate is committed. `(ledger, verifier, audit, config, policy) -> AggregatorProtocol` — how a custom optimizer is installed. +### `AllocatorContext` + +One candidate, as the allocator sees it. + +### `AllocatorPolicy` + +Decide, per candidate, whether the expensive part of a merge is worth it. + ### `AppendRules` Accumulate a deduped list of rules/lessons (append-only, content-addressed). @@ -3674,6 +3686,10 @@ One entry per *category*: competing proposals contradict and are resolved. Where a runner writes the evolving tree inside a workspace (`claude_skill`, `skill_library`, `claude_agent`, `dsh_skill`, `agents_skill`, `root`). +### `LearningValueModel` + +A tiny online logistic model over the candidate features. + ### `LedgerFailure` The exception tuple a caller catches to treat any ledger problem as recoverable. @@ -3818,6 +3834,10 @@ Prefix of the output `code_runner` produces when the frozen gate fails, so the f The default: a bounded pool of threads in this process. +### `ValueBudgetAllocator` + +Learn which candidates are worth the expensive evaluation. + ### `VerifierProtocol` Four methods, from `grep 'self\.verifier\.' agentdescent/aggregator.py`. diff --git a/docs/budget.md b/docs/budget.md new file mode 100644 index 00000000..3335f61d --- /dev/null +++ b/docs/budget.md @@ -0,0 +1,148 @@ +# Budget allocation + +> **The token budget is a ceiling; this page is about how to spend it.** A +> `max_tokens` cap is a brake — when spend reaches it, the run stops. Everything +> on this page answers a different question: given that you have a budget, *which +> candidate is worth spending the expensive part of it on*? + +## 1. The problem with a clock + +`BudgetGovernor` (see [`evolution`](evolution.md) / [`async`](async.md)) degrades +the optional spend of a run — fusion tournaments, the self-verify rollout — as +the budget runs out, against two fixed floors: + +| floor | what it shuts off | default | +|---|---|---| +| `soft_floor` | the fusion tournament (a held-out sweep used only to *rank* survivors) | 75% spent | +| `hard_floor` | the self-verify rollout (the before/after delta) | 90% spent | + +That is fair, predictable, and exactly what a budget-aware allocator should +*not* do: it treats every candidate alike. When `spent / max_tokens` crosses the +floor, the *next* candidate stops getting the expensive part — whatever it is. +A candidate that arrived late in the run but would have been the best change of +the whole run is skipped because of the clock, not because of its value. + +The EvoAlloc line of work measures the cost of this fixed allocation: when +evaluation is the scarce resource, spending it by time-since-start rather than +by expected value wastes it on candidates that cannot move the search while +starving ones that can. `evolve(allocator=...)` replaces the clock with a +**per-candidate** decision learned from the run's own merge history. + +## 2. `ValueBudgetAllocator` — learn which candidates are worth it + +```python +from agentdescent import ValueBudgetAllocator + +alloc = ValueBudgetAllocator() +evolve(tasks, reward, agent=agent, + max_tokens=500_000, # the allocator decides how this is spent + allocator=alloc) +``` + +Two halves, one object: + +**Value-directed allocation.** Once the budget is tight (`remaining_fraction` +below the allocator's `budget_start`, default 0.5), the self-verify rollout is +spent only on candidates the value model expects to commit. `LearningValueModel` +is a tiny online logistic model over the **pre-spend features the worker +actually has at decision time** — the proposing rollout's group-relative +`advantage`, the diff's `size`, and the artifact's `blast_radius`. It is fit +online against commit outcomes ("did it commit"), so it needs no labelled data: +the engine knows the label for free. + +**Decision and training see the same vector.** The post-outcome signals the +merge computes later — the before/after `delta` (the self-verify rollout's own +product), `P(improve)`, the distance from `stable` — are kept on the context +for the audit trail but **never enter the model's feature vector**. Training on +post-outcome features while deciding on pre-spend ones would fit a model that +explains the labels without teaching a usable pre-spend predictor; restricting +both sides to the pre-spend columns keeps the learning loop sound. A logit clip +and an L2 term keep one extreme candidate from blowing the weights. + +**Every commit is attributed to a worker's actual pre-spend vector.** Which +card an outcome attaches to depends on how the candidate was chosen: + +- **Fused** (the candidate is the union of several proposals): the commit label + attaches to the **first contributor's** pre-spend vector — the card the + tournament put forward, the same attribution the acceptance gate uses when it + folds `before_after_delta` back into the posterior. A fusion's union size is + a post-merge artefact no worker ever decided on. +- **Not fused** (the tournament chose a single proposal): the label attaches to + the **winning proposal's** card, found by matching the selected diff — never + an arbitrary first entrant, which may be a losing proposal. Labeling a losing + card as committed would teach the model the wrong proposal succeeded and + supply no observation for the actual winner. + +Either way the vector is what a worker decided on (advantage, the card's own +diff size, the artifact's blast radius) — never a post-merge size. + +**Counterfactual exploration.** A candidate the model judged unworthy is +skipped — except with probability `epsilon`, when it is evaluated anyway. The +outcome is then *ground truth the model was wrong about*, and it learns from it. +Without this term the allocator only ever sees the candidates it already +believed in, and a biased value model is never corrected: a workload where +"low before/after delta" once meant "won't commit" would be skipped forever even +after it stopped meaning that. `epsilon` anneals from `epsilon_start` (0.3) down +to `epsilon_end` (0.05) over the first `epsilon_anneal` (40) observations, so a +run explores hardest when the model knows nothing. + +## 3. How it is wired + +The decision and the feedback are two separate points on the merge path: + +- **The worker asks** — before spending the self-verify rollout, the budget + governor calls the allocator's `decide(ctx)` with an + [`AllocatorContext`](https://github.com/Birfy/agentdescent/blob/main/agentdescent/allocator.py) + built from the diff's size and the rollout's advantage. High-value candidates + keep the spend; low-value ones lose it; counterfactual ones get it anyway. +- **The aggregator reports** — every measured decision is reported exactly + once (`observe(ctx, committed)`): a commit, an oracle rejection, an + acceptance rejection and a CAS conflict all produce an observation. Oracle + and acceptance rejections are **negative labels** — without them the model + would learn only from commits and never know which evaluated proposals fail. + +`allocator=None` (the default) is the old behaviour byte for byte. The clock +thresholds remain the *ceiling* the allocator may never exceed: it can skip a +candidate the clock would have deep-evaluated, never spend on one the clock +refused. + +## 4. The A/B — is it actually worth it? + +The mechanism's own premise is *an analogy without an A/B is decoration*. The +equal-budget experiment is [`bench/ab_allocator.py`](https://github.com/Birfy/agentdescent/blob/main/bench/ab_allocator.py): + +```bash +python -m bench.ab_allocator --dataset hotpotqa --budget-rollouts 96 \ + --seeds 0,1,2 --provider claude --model GLM-5.2 \ + --max-tokens 500_000 --yes +``` + +Two arms on identical workloads, budgets and seeds, one argument different: + +| arm | what runs | +|---|---| +| `off` | `allocator=None` — the clock degrades optional spend | +| `on` | `allocator=ValueBudgetAllocator()` — learned per-candidate spend | + +The table reports **test quality, measured tokens and whether the allocator +fired**. The fired check is the first thing to read: an allocator that agreed +with the clock everywhere (`fired == 0` with `observed > 0`) is a control +against a control, and its token column is about nothing. `--max-tokens` must +be set — below the budget the allocator stays inert, and the two arms are then +the same run. + +If the `on` arm does not win here (equal-or-better quality at fewer tokens, or +better quality at the same tokens), delete it and record the negative result on +this page. That is the outcome the premise asks for explicitly. + +## 5. What this does not do + +- **It does not raise the budget.** `max_tokens` is still the cap; the allocator + only decides how to spend what exists. Requires `max_tokens` to be set. +- **It does not replace the acceptance gate.** Every candidate that gets a + self-verify rollout still goes through the same Beta-posterior acceptance and + the audit gate. The allocator changes *how much finding-out is spent*, never + *what is kept*. +- **It does not touch the oracle budget.** `oracle_budget` / `AuditScheduler` + remain the mechanism for spending ground-truth checks; the allocator is about + the *optional* self-verify spend. diff --git a/docs/modules.md b/docs/modules.md index c3d63bf4..ef4ec67c 100644 --- a/docs/modules.md +++ b/docs/modules.md @@ -96,6 +96,7 @@ and *how*, that is *what*. | `policies` | the contracts: which decisions are replaceable, and what each is given | [Choosing policies](policies.md) | | `baselines` | equal-budget baselines the results pages compare against | [Measured results](results.md) | | `metrics` | what the run cost: time, calls, staleness ratio, cache hits, sandbox waits | [Usage](usage.md#what-a-run-cost) | +| `budget` | value-directed allocation: learn which candidate is worth the expensive evaluation, with counterfactual exploration | [Budget allocation](budget.md) | | `bench` | the configuration matrix and the rules that make comparing them mean something | [Efficiency](efficiency.md#the-configuration-matrix-bench) | ## Reading order diff --git a/examples/gepa/gepa_prompt_evolution.py b/examples/gepa/gepa_prompt_evolution.py index 5866f421..4529a24d 100644 --- a/examples/gepa/gepa_prompt_evolution.py +++ b/examples/gepa/gepa_prompt_evolution.py @@ -409,7 +409,17 @@ def factory(ledger, verifier, audit, config, policy): # =========================================================================== -def download_hotpotqa(limit: int) -> List[dict]: +def download_hotpotqa(limit: int, source: str = "hf") -> List[dict]: + """HotpotQA validation rows, from HuggingFace or a ModelScope mirror. + + ``source="hf"`` (default) uses the datasets-server; ``source="modelscope"`` + fetches the same rows from ``OpenDataLab/HotpotQA`` on ModelScope, for + networks where the HuggingFace endpoint is unreachable but ModelScope is + up. Rows are identical either way (``context`` is the dict shape examples + read).""" + if source == "modelscope": + from agentdescent.dataloader import hf_rows_modelscope + return hf_rows_modelscope(limit=limit) dataset, split, config = HOTPOTQA return hf_rows(dataset, split, config=config, limit=limit) @@ -464,9 +474,14 @@ def estimate_calls(rounds: int, workers: int, held_out: int) -> int: return rounds * (workers * 3 + held_out) -def load_dataset(fetch: int, ratios=(0.5, 0.25, 0.25), seed: int = 0) -> Dataset: - """HotpotQA tasks split into train / val (D_pareto) / test.""" - tasks = build_tasks(download_hotpotqa(fetch), limit=fetch, seed=seed) +def load_dataset(fetch: int, ratios=(0.5, 0.25, 0.25), seed: int = 0, + source: str = "hf") -> Dataset: + """HotpotQA tasks split into train / val (D_pareto) / test. + + ``source`` is passed to :func:`download_hotpotqa` -- ``"hf"`` (default) + via the datasets-server, or ``"modelscope"`` for the ModelScope mirror.""" + tasks = build_tasks(download_hotpotqa(fetch, source=source), limit=fetch, + seed=seed) return split_dataset(tasks, ratios=ratios, seed=seed, name="HotpotQA") diff --git a/mkdocs.yml b/mkdocs.yml index f6adea12..bafee198 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -83,6 +83,7 @@ nav: - Sandboxes: sandboxes.md - Async — no round barrier: async.md - Duration-aware scheduling: duration-scheduling.md + - Budget allocation: budget.md - The decision plane (policies): - Choosing policies: policies.md - Using the slots — a guide: policy-guide.md diff --git a/tests/test_allocator.py b/tests/test_allocator.py new file mode 100644 index 00000000..fd11cb1b --- /dev/null +++ b/tests/test_allocator.py @@ -0,0 +1,797 @@ +"""Value-directed budget allocation: learn which candidates are worth the +expensive evaluation, with counterfactual exploration to correct the learner. + +``BudgetGovernor`` degrades optional spend (fusion tournaments, self-verify +rollouts) against global floors: once ``spent / max_tokens`` crosses a threshold, +*every* candidate stops getting the expensive part. The EvoAlloc line of work +measures the cost of that fixed allocation when evaluation is the scarce +resource. These tests pin the replacement: a per-candidate decision learned from +the run's own merge history, plus an epsilon-greedy counterfactual term that +re-measures candidates the model judged unworthy so a biased value model is +corrected rather than reinforced. +""" + +from __future__ import annotations + +import random + +from agentdescent.allocator import ( + AllocatorContext, + LearningValueModel, + ValueBudgetAllocator, +) +from agentdescent.budget import BudgetGovernor +from agentdescent.evolution import Task, evolve + + +def _high(): + return AllocatorContext(advantage=1.0, size=2) + + +def _low(): + return AllocatorContext(advantage=-0.5, size=2) + + +#: A candidate *before* the self-verify ran -- the pre-spend features only +#: (advantage, size, blast radius), which is what the worker actually decides +#: on. The post-outcome signals (delta, P(improve), stable distance) never +#: enter the model's feature vector. +def _unmeasured(advantage: float, size: int = 2): + return AllocatorContext(advantage=advantage, size=size) + + +def _nsize(size: int) -> float: + """The normalised size column: ``log1p(size)/log1p(100)``, matching + :meth:`AllocatorContext.features`.""" + import math + return math.log1p(max(0, size)) / math.log1p(100) + + +# --------------------------------------------------------------------------- +# the value model +# --------------------------------------------------------------------------- + + +def test_value_model_learns_high_advantage_predicts_commit(): + m = LearningValueModel(lr=0.3, reg=0.05) + for _ in range(30): + m.update(_high().features(), True) + m.update(_low().features(), False) + p_high = m.predict(_high().features()) + p_low = m.predict(_low().features()) + assert p_high > p_low, "the model must learn that a positive advantage commits" + + +def test_value_model_stays_calm_on_a_single_outlier(): + """One extreme candidate must not blow the weights -- the logit clip.""" + m = LearningValueModel(lr=0.3, reg=0.05) + m.update((0.999999, 64, 1.0), True) # extreme pre-spend vector + for _ in range(10): + m.update(_low().features(), False) + assert all(abs(w) < 1e6 for w in m._w), "weights must stay finite" + + +def test_value_model_bias_learns_on_all_zero_features(): + """The bias weight must move even when every feature is zero -- otherwise + an all-zero candidate (nothing measured on this workload) can never learn a + non-neutral prior. This pins the logistic model's arithmetic: the bias is + the last weight, matched against ``1.0`` in the gradient, and never double- + counted in the logit.""" + m = LearningValueModel(lr=0.3, reg=0.05) + zeros = (0.0, 0.0, 0.0) # advantage, size, blast_radius + for _ in range(50): + m.update(zeros, True) # every all-zero candidate commits + assert m.predict(zeros) > 0.6, "the bias must learn a non-neutral prior" + assert m._w[-1] != 0.0, "the bias weight must move" + + +def test_value_model_blast_radius_moves_its_own_weight(): + """blast_radius must be a *feature* with its own learned weight, not folded + into the bias. Under a single data point the two are confounded (many + ``(w_blast, bias)`` pairs give the same logit), so this asserts the feature + actually moves prediction: a candidate with high blast_radius but nothing + else measured must be scored differently from the all-zero baseline.""" + m = LearningValueModel(lr=0.3, reg=0.05) + zeros = (0.0, 0.0, 0.0) # nothing measured + blast = (0.0, 0.0, 1.0) # only blast_radius set + # All-zero candidates commit; blast-only candidates do not. + for _ in range(30): + m.update(zeros, True) + m.update(blast, False) + p_zero = m.predict(zeros) + p_blast = m.predict(blast) + assert p_blast < p_zero, ( + "blast_radius must carry independent signal, not be folded into bias") + + +# --------------------------------------------------------------------------- +# the allocator decision +# --------------------------------------------------------------------------- + + +def test_allocator_skips_low_value_candidate_after_learning(): + alloc = ValueBudgetAllocator(min_observations=3, + epsilon_start=0.0, epsilon_end=0.0) + for _ in range(5): + alloc.observe(_high(), True) # feedback: real delta + alloc.observe(_low(), False) + # Decision side sees the *unmeasured* candidate (delta unknown), and must + # still tell high-advantage from low-advantage on the up-front signals. + assert alloc.decide(_unmeasured(1.0)).deep_eval, "high value keeps the deep eval" + assert not alloc.decide(_unmeasured(-0.5)).deep_eval, "low value loses it" + + +def test_allocator_deep_evals_everything_before_it_has_an_opinion(): + """Skipping on no evidence is the bias the exploration exists to avoid.""" + alloc = ValueBudgetAllocator(min_observations=5) + assert alloc.decide(_low()).deep_eval, "no opinion yet -> spend, don't skip" + assert alloc.n_skipped == 0 + + +def test_allocator_counterfactual_exploration_re_measures_skipped(): + """A candidate the model judged unworthy is still evaluated with epsilon + probability, so the model sees the ground truth it was wrong about.""" + alloc = ValueBudgetAllocator(min_observations=2, + epsilon_start=1.0, epsilon_end=1.0) + for _ in range(2): + alloc.observe(_low(), False) + random.seed(7) + n_cf = sum(1 for _ in range(100) + if alloc.decide(_low()).counterfactual) + assert n_cf > 0, "epsilon>0 must occasionally re-measure a skipped candidate" + + +def test_counterfactual_outcome_corrects_the_value_model(): + """The model believed `_low` never commits; counterfactual outcomes that + *do* commit must push its prediction up -- the selection-bias correction.""" + alloc = ValueBudgetAllocator(min_observations=2, + epsilon_start=1.0, epsilon_end=1.0) + for _ in range(4): + alloc.observe(_low(), False) + p_before = alloc.model.predict(_low().features()) + # Now the workload flips: every low candidate actually commits. + for _ in range(20): + alloc.observe(_low(), True) + p_after = alloc.model.predict(_low().features()) + assert p_after > p_before, ( + "counterfactual outcomes must move the model toward the truth") + + +# --------------------------------------------------------------------------- +# governor integration +# --------------------------------------------------------------------------- + + +def test_governor_without_allocator_keeps_old_behaviour(): + """No allocator -> the clock thresholds decide, exactly as before.""" + g = BudgetGovernor(max_tokens=1000) + g.spend(800) # 80% spent + assert g.candidate_worth(_high()) + assert g.candidate_worth(_low()) # old behaviour: no per-candidate skip + g.spend(920) # 92% spent, past the hard floor + assert not g.candidate_worth(_high()) + + +def test_governor_with_allocator_skips_only_when_budget_is_tight(): + alloc = ValueBudgetAllocator(min_observations=3, + epsilon_start=0.0, epsilon_end=0.0) + for _ in range(5): + alloc.observe(_high(), True) + alloc.observe(_low(), False) + # Budget not tight: everything deep-evals (old behaviour). + loose = BudgetGovernor(max_tokens=1000, allocator=alloc) + loose.spend(100) # 90% left, above budget_start + assert loose.candidate_worth(_low()) + # Budget tight: the allocator decides per candidate. + tight = BudgetGovernor(max_tokens=1000, allocator=alloc) + tight.spend(800) # 20% left, below budget_start + assert tight.candidate_worth(_high()) + assert not tight.candidate_worth(_low()) + + +def test_governor_never_spends_past_the_hard_floor_via_allocator(): + """The clock thresholds are the ceiling the allocator may never exceed.""" + alloc = ValueBudgetAllocator(min_observations=0, + epsilon_start=0.0, epsilon_end=0.0) + g = BudgetGovernor(max_tokens=1000, allocator=alloc) + g.spend(950) # 95% spent, past the hard floor + assert not g.candidate_worth(_high()), "no allocator may spend past 90%" + + +def test_observe_allocated_is_a_no_op_without_allocator(): + g = BudgetGovernor(max_tokens=1000) + g.observe_allocated(_high(), committed=True) # must not raise + assert g.allocator is None + + +# --------------------------------------------------------------------------- +# engine integration +# --------------------------------------------------------------------------- + + +def _reward(task, output): + return 1.0 if output == "yes" else 0.0 + + +class _Reflector: + def solve(self, rendered, task): + return "yes" if "rule" in rendered else "no" + + def propose(self, rendered, task, output, reward): + return "a helpful rule" + + +def test_evolve_feeds_commit_outcomes_to_the_allocator(): + import warnings + alloc = ValueBudgetAllocator(min_observations=2) + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + evolve(tasks, _reward, agent=_Reflector(), rounds=4, n_workers=2, + max_concurrency=2, max_tokens=200_000, allocator=alloc) + assert alloc.n_observed >= 1, "the merge path must report outcomes" + assert alloc.n_committed >= 1, "the committed candidate must be observed" + + +def test_evolve_without_allocator_is_unchanged(): + import warnings + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + res = evolve(tasks, _reward, agent=_Reflector(), rounds=3, n_workers=2, + max_concurrency=2, max_tokens=200_000) + assert res.final_reward == 1.0 + + +def test_async_evolve_accepts_allocator(): + import warnings + from agentdescent.async_evolve import async_evolve + from agentdescent.evolution import AppendRules + alloc = ValueBudgetAllocator(min_observations=2) + tasks = [Task(id=f"t{i}", prompt="q") for i in range(6)] + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + res = async_evolve( + tasks, _reward, run=_Reflector().solve, + propose=_Reflector().propose, strategy=AppendRules(), + n_workers=2, max_seconds=5.0, max_tokens=100_000, + allocator=alloc, held_out_frac=0.4) + assert res.final_reward is not None + + +# --------------------------------------------------------------------------- +# every measured decision reports exactly one outcome +# --------------------------------------------------------------------------- + + +def _candidate(artifact, cand, diff, *, base_counts=(2.0, 0.0), + cand_counts=(3.0, 0.0), fused=False, base_cheap=0.9, + cand_cheap=0.95): + """A measured `_Candidate` for the merge path.""" + from agentdescent.aggregator import _Candidate + return _Candidate( + artifact_id=artifact.id, artifact=artifact, candidate=cand, diff=diff, + cards=[], survivor_cards=[], head={artifact.id: artifact.version}, + fused=fused, considered=1, survived=1, discarded=0, conflicts=0, + base_counts=base_counts, cand_counts=cand_counts, + base_cheap=base_cheap, cand_cheap=cand_cheap) + + +def test_every_merge_outcome_is_observed_exactly_once(tmp_path): + """Oracle rejection, acceptance rejection, commit and CAS conflict must + each report exactly one outcome to the allocator -- rejection is a negative + label, and without it a model fit on commits alone never leaves warm-up.""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.evolvable import Diff + from agentdescent.evolution import AppendRules, EvolvingArtifact + from agentdescent.ledger import Ledger + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + class AlwaysRejectAudit(AuditScheduler): + """`force_oracle` always opens, and the oracle always vetoes.""" + + def force_oracle(self, blast_radius, artifact_id): + return True + + def full_eval(self, artifact): # not used; see below + return 0.0 + + class VetoingVerifier(ThreeLayerVerifier): + def full_eval(self, artifact): + return 0.0 # never better than base -> oracle veto + + def serialize(a): return {"state": a.state, "blast_radius": a.blast_radius} + def deserialize(aid, v, s): + return EvolvingArtifact(aid, s.get("state", {}), v, + s.get("blast_radius", 0.2)) + lg = Ledger(str(tmp_path / "repo"), serialize, deserialize) + lg.register(EvolvingArtifact("a", {"k": "v0"}, blast_radius=0.2)) + base = lg.snapshot(Ledger.DEV).get("a") + + alloc = ValueBudgetAllocator(min_observations=1, epsilon_start=0.0, + epsilon_end=0.0) + + def state_scorer(a, t): + return 0.9 if a.state.get("k") == "v1" else 0.3 + + def make_agg(verifier, audit=None): + if audit is None: + audit = AuditScheduler() + return Aggregator(lg, verifier, audit, + AggregatorConfig(batch_trigger=1)) + agg = make_agg(ThreeLayerVerifier( + eval_fn=state_scorer, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=100))) + agg.allocator = alloc + + before = alloc.n_observed + + # 1. Acceptance rejection: candidate fails the Beta test (cheap and full + # layers agree it is worse, so the audit gate stays closed). + cand_bad = base.apply(Diff(diff_id="bad", target="a", ops={"k": "bad"}, + author="w")) + c_bad = _candidate(base, cand_bad, + Diff(diff_id="bad", target="a", ops={"k": "bad"}, + author="w"), + base_counts=(2.0, 1.0), # rate 0.67 + cand_counts=(1.0, 3.0), # rate 0.25, clearly worse + cand_cheap=0.1) # cheap agrees it is worse + r_bad = agg._decide(c_bad) + assert r_bad.category in ("below-threshold", "cas-conflict"), r_bad.category + # exactly one observation, and it is a negative label + assert alloc.n_observed == before + 1, "acceptance rejection must observe once" + assert alloc.n_committed == 0 + + # 2. Commit: candidate beats the base on the full held-out. + cand_good = base.apply(Diff(diff_id="good", target="a", + ops={"k": "v1"}, author="w")) + c_good = _candidate(base, cand_good, + Diff(diff_id="good", target="a", ops={"k": "v1"}, + author="w"), + base_counts=(2.0, 1.0), # rate 0.67 + cand_counts=(4.0, 0.0)) # rate 1.0, better + r_good = agg._decide(c_good) + assert r_good.category == "committed", r_good.category + assert alloc.n_observed == before + 2, "a commit must observe once" + assert alloc.n_committed == 1 + + # 3. Oracle rejection: `force_oracle` always opens and the oracle vetoes + # -- one more negative observation. + agg_oracle = make_agg(VetoingVerifier( + eval_fn=state_scorer, held_out=[1, 2, 3], + budget=VerifierBudget(oracle_calls_remaining=100)), + audit=AlwaysRejectAudit()) + agg_oracle.allocator = alloc + # force_oracle always opens and the oracle (full_eval -> 0.0, or the + # candidate's sub-base rate) vetoes -- a third, negative observation. + c_oracle = _candidate(base, cand_bad, + Diff(diff_id="oracle", target="a", ops={"k": "bad"}, + author="w"), + base_counts=(2.0, 1.0), cand_counts=(1.0, 3.0), + cand_cheap=0.1) + r_oracle = agg_oracle._decide(c_oracle) + assert r_oracle.category == "oracle-rejected", r_oracle.category + assert alloc.n_observed == before + 3, ( + "oracle rejection must observe once") + assert alloc.n_committed == 1, "a veto is a negative label" + + +def test_sync_async_pre_spend_features_are_identical(): + """The decision-time features the sync and async workers build must be the + same vector the aggregator trains on -- advantage, size, blast radius, and + nothing post-outcome.""" + from agentdescent.evolvable import Diff + diff = Diff(diff_id="d", target="a", ops={"k1": "v", "k2": "w"}, author="w") + from agentdescent.evolution import _allocator_context + from agentdescent.async_evolve import _async_allocator_context + sync = _allocator_context(diff, advantage=0.7, score=0.3, blast_radius=0.4) + async_ctx = _async_allocator_context(diff, advantage=0.7, blast_radius=0.4) + assert sync.features() == async_ctx.features(), ( + "sync and async decision features must match") + # Only the three pre-spend features are model inputs; post-outcome signals + # are absent even when supplied. + assert sync.features() == (0.7, _nsize(2), 0.4) + # The training-side context (with post-outcome fields filled) must produce + # the same feature vector as the decision side. + from agentdescent.aggregator import _allocator_context_from + from agentdescent.evolution import EvolvingArtifact + art = EvolvingArtifact("a", {"k": "v0"}, blast_radius=0.4) + cand = art.apply(diff) + card = type("Card", (), {"advantage": 0.7, + "before_after_delta": 0.9, + "diff": diff})() + c = type("C", (), { + "artifact": art, "candidate": cand, "diff": diff, + "cards": [card], "fused": False})() + trained = _allocator_context_from(c, p_improve=0.99, stable_distance=0.3) + assert trained.features() == sync.features(), ( + "training features must equal decision features, so the model cannot " + "fit labels with post-outcome signals the decision never sees") + + +def test_fused_observe_uses_the_originating_proposal_context(): + """A fused commit's label attaches to the *originating* proposal's pre-spend + vector, not the fused diff's union size -- the worker decided on the single + proposal, and training on a union size no decision saw leaks post-merge + information.""" + from agentdescent.aggregator import _allocator_context_from + from agentdescent.evolvable import Diff + from agentdescent.evolution import EvolvingArtifact + art = EvolvingArtifact("a", {"k": "v0"}, blast_radius=0.2) + proposal = Diff(diff_id="p1", target="a", ops={"k1": "v"}, author="w") + fused = Diff(diff_id="fused(p1+p2)", target="a", + ops={"k1": "v", "k2": "w"}, author="aggregator") # size 2 + card = type("Card", (), {"advantage": 0.7, + "before_after_delta": 0.9, + "diff": proposal})() # size 1 + cand = art.apply(fused) + c = type("C", (), { + "artifact": art, "candidate": cand, "diff": fused, + "cards": [card], "fused": True})() + ctx = _allocator_context_from(c, p_improve=0.99, stable_distance=0.3) + # The decision side saw the proposal (size 1), and the fused label must + # attach to that same vector -- not the fused size 2. + assert ctx.features() == (0.7, _nsize(1), 0.2), ( + f"fused observe must use the originating proposal's size, got " + f"{ctx.features()}") + + +def test_single_winner_observes_the_selected_proposal_not_the_first_entrant(): + """When the tournament picks a *later* single proposal (no fusion), the + commit label must attach to the winning proposal's card -- not the first + (possibly losing) entrant. Labeling a losing card as committed teaches the + model the wrong proposal succeeded.""" + from agentdescent.aggregator import _allocator_context_from + from agentdescent.evolvable import Diff + from agentdescent.evolution import EvolvingArtifact + art = EvolvingArtifact("a", {"k": "v0"}, blast_radius=0.2) + bad = Diff(diff_id="bad", target="a", ops={"k": "BAD"}, author="w") # size 1 + good = Diff(diff_id="good", target="a", + ops={"k": "GOOD", "pad": "PAD"}, author="w") # size 2 + bad_card = type("Card", (), {"advantage": 0.1, + "before_after_delta": -0.5, + "diff": bad})() + good_card = type("Card", (), {"advantage": 0.7, + "before_after_delta": 0.9, + "diff": good})() + cand = art.apply(good) + c = type("C", (), { + "artifact": art, "candidate": cand, "diff": good, + "cards": [bad_card, good_card], "fused": False})() + ctx = _allocator_context_from(c, p_improve=0.99, stable_distance=0.3) + # The tournament chose GOOD (size 2); the label attaches to its vector, + # not the first entrant BAD (size 1). + assert ctx.features() == (0.7, _nsize(2), 0.2), ( + f"single-winner observe must use the selected proposal's vector, got " + f"{ctx.features()}") + + +class _RecordingAllocator: + """Records every decide/observe input feature vector, then always deep-evals. + + Lets an end-to-end run prove fused commits observe the *originating* + proposal's pre-spend vector, not the fused union size.""" + + def __init__(self): + self.decided = [] + self.observed = [] + self.budget_start = 2.0 # budget never tight -> always decide + + def decide(self, ctx): + self.decided.append(ctx.features()) + from agentdescent.allocator import AllocateDecision + return AllocateDecision(deep_eval=True) + + def observe(self, ctx, committed): + self.observed.append((ctx.features(), committed)) + + +def test_fused_commit_observes_originating_proposal_end_to_end(): + """A real `evolve()` run fuses two proposals and commits the union; the + observe input must be a worker's actual pre-spend vector (the single rule's + size 1), never the fused size 2.""" + import warnings + from agentdescent.evolution import AppendRules + + rec = _RecordingAllocator() + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + + def run(rendered, task): + return rendered + + def reward(task, output): + return min(1.0, output.count("rule") * 0.5) + + propose = iter([f"rule {i}" for i in range(20)]) + + def make_propose(rendered, task, output, score): + return next(propose, "rule x") + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + res = evolve(tasks, reward, run=run, propose=make_propose, + strategy=AppendRules(), rounds=1, n_workers=2, + max_concurrency=1, max_tokens=10_000, allocator=rec, + held_out_frac=0.5, shuffle=False) + + assert rec.decided, "workers must call decide before spending" + assert rec.observed, "the merge must call observe" + # The fusion must actually have committed -- otherwise the observation is + # about nothing. + assert res.fusion_stats().fused_wins >= 1, ( + f"the fused candidate must have committed, got " + f"fused_wins={res.fusion_stats().fused_wins}") + # Every observed vector's size must be one the allocator actually decided + # on -- a single proposal's diff size. The fused union (size >= 2) must not + # appear, because no worker ever decided about a size-2 candidate. + decided_sizes = {f[1] for f in rec.decided} + for feats, committed in rec.observed: + assert feats[1] in decided_sizes, ( + f"observe fed size {feats[1]}, but decisions only saw " + f"sizes {decided_sizes} -- the fused union size leaked into training") + + +class _MultiKeyStrategy: + """AppendRules-shaped but with per-line keys: a proposal of `k1: v1\\nk2: v2` + becomes a two-op diff. Lets a test emit proposals of different sizes.""" + + def initial(self): + return {} + + def render(self, state): + return "\n".join(f"{k}: {v}" for k, v in sorted(state.items())) or "(empty)" + + def to_diff(self, state, proposal, author, base_version, target): + from agentdescent.evolvable import Diff + ops = {} + for line in (proposal or "").splitlines(): + k, _, v = line.partition(":") + k, v = k.strip(), v.strip() + if k: + ops[k] = v + if not ops or all(state.get(k) == v for k, v in ops.items()): + return None + return Diff(diff_id=f"{author}:{'/'.join(sorted(ops))}:{base_version}", + target=target, ops=ops, author=author) + + +def _single_winner_run(rec, tasks, make_propose): + """A run where the tournament picks a *later* single proposal (not the + first entrant, not a fusion): first worker proposes a bad one-op rule, the + second a good two-op rule; reward favours the good one, so the tournament + selects it with fused=False. Returns the evolve result.""" + import warnings + from agentdescent.evolution import evolve + from agentdescent.strategies import KeyedRules + + def run(rendered, task): + return rendered + + def reward(task, output): + # 1 iff GOOD present and BAD absent -- the good proposal wins alone. + return 1.0 if ("GOOD" in output and "BAD" not in output) else 0.0 + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + return evolve( + tasks, reward, run=run, propose=make_propose, + strategy=_MultiKeyStrategy(), + rounds=1, n_workers=2, max_concurrency=1, max_tokens=10_000, + allocator=rec, held_out_frac=0.5, shuffle=False, + agg_config=__import__("agentdescent.aggregator", + fromlist=["AggregatorConfig"]).AggregatorConfig( + batch_trigger=2, fusion_tournament=True)) + + +def test_single_winner_observes_selected_proposal_end_to_end(): + """The tournament chooses the good (size-2) proposal over the first bad + (size-1) entrant, commits it fused=False; the observe vector must be the + *winning* proposal's, not the first (losing) entrant's.""" + rec = _RecordingAllocator() + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + proposals = iter(["bad: BAD", "good: GOOD\npad: PAD", None, None]) + + def make_propose(rendered, task, output, score): + return next(proposals, None) + + res = _single_winner_run(rec, tasks, make_propose) + assert rec.decided, "workers must call decide" + # decisions saw size 1 (bad) and size 2 (good+pad) + assert sorted({f[1] for f in rec.decided}) == [_nsize(1), _nsize(2)], \ + rec.decided + # the winner (size 2) committed; the observe vector must be its own + # (0.0, nsize(2), 0.2) -- not the losing first entrant's (0.0, nsize(1), 0.2). + assert len(rec.observed) == 1, rec.observed + feats, committed = rec.observed[0] + assert committed, "the good proposal must have committed" + assert feats == (0.0, _nsize(2), 0.2), ( + f"single winner must observe its own vector, got {feats}") + + +def test_async_single_winner_observes_selected_proposal_end_to_end(): + """Same guarantee on the barrier-free path, hardened: exactly one committed + observation, and its full vector (not just the size) is the winning + proposal's -- never the losing entrant's.""" + import warnings + from agentdescent.async_evolve import async_evolve + rec = _RecordingAllocator() + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + proposals = iter(["bad: BAD", "good: GOOD\npad: PAD", None, None]) + + def make_propose(rendered, task, output, score): + return next(proposals, None) + + def run(rendered, task): + return rendered + + def reward(task, output): + return 1.0 if ("GOOD" in output and "BAD" not in output) else 0.0 + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + async_evolve( + tasks, reward, run=run, propose=make_propose, + strategy=_MultiKeyStrategy(), n_workers=2, max_iters=20, + max_seconds=0.5, max_tokens=10_000, allocator=rec, + held_out_frac=0.5, shuffle=False, + agg_config=__import__("agentdescent.aggregator", + fromlist=["AggregatorConfig"]).AggregatorConfig( + batch_trigger=2, fusion_tournament=True)) + assert rec.decided, "workers must call decide" + assert sorted({f[1] for f in rec.decided}) == [_nsize(1), _nsize(2)], ( + f"decisions must have seen both proposal sizes, got {rec.decided}") + # Exactly one observation, committed, and it is the winner's full vector. + assert len(rec.observed) == 1, ( + f"expected exactly one merge observation, got {rec.observed}") + feats, committed = rec.observed[0] + assert committed, "the good (size-2) proposal must have committed" + assert feats == (0.0, _nsize(2), 0.2), ( + f"async single winner must observe its own full vector, got {feats}") + + +def test_async_fused_commit_observes_originating_proposal_end_to_end(): + """Same guarantee on the barrier-free path: whenever the merger observes a + committed candidate, its vector is a worker's actual pre-spend vector (the + individual rule's size 1), never the fused union size. The sync twin pins + the fusion *commit* deterministically; on the async path a fusion's win + depends on merge timing, so this asserts the attribution invariant instead + -- the part the review's fused-union counterexample was about.""" + import warnings + from agentdescent.async_evolve import async_evolve + from agentdescent.evolution import AppendRules + + rec = _RecordingAllocator() + tasks = [Task(id=f"t{i}", prompt="q") for i in range(8)] + + def run(rendered, task): + return rendered + + def reward(task, output): + return min(1.0, output.count("rule") * 0.5) + + propose = iter([f"rule {i}" for i in range(20)]) + + def make_propose(rendered, task, output, score): + return next(propose, "rule x") + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + res = async_evolve( + tasks, reward, run=run, propose=make_propose, + strategy=AppendRules(), n_workers=2, max_seconds=5.0, + max_tokens=10_000, allocator=rec, held_out_frac=0.5, + shuffle=False) + assert rec.decided, "workers must call decide before spending" + # Every observed vector's size must be one the allocator actually decided + # on -- a single proposal's diff size. The fused union (size >= 2) must not + # appear, because no worker ever decided about a size-2 candidate. + decided_sizes = {f[1] for f in rec.decided} + assert decided_sizes, "workers must have decided on some candidates" + for feats, committed in rec.observed: + assert feats[1] in decided_sizes, ( + f"async observe fed size {feats[1]}, but decisions only saw " + f"sizes {decided_sizes}") + decided_sizes = {f[1] for f in rec.decided} + for feats, committed in rec.observed: + assert feats[1] in decided_sizes, ( + f"async observe fed size {feats[1]}, but decisions only saw " + f"sizes {decided_sizes}") + + +# --------------------------------------------------------------------------- +# the A/B script's arm wiring (bench/ab_allocator.py) +# --------------------------------------------------------------------------- + + +def test_ab_allocator_arm_wiring(): + """`bench.ab_allocator._arm` returns no allocator on the control arm and a + fresh `ValueBudgetAllocator` (with a way to say whether it fired) on the + treatment arm -- so an A/B run can tell "the allocator helped" from "it + never ran" (the review's repeated 'no paid calls / real-dataset A/B' + concern).""" + import bench.ab_allocator as ab + + off_kwargs, off_watcher = ab._arm(False) + assert off_kwargs == {}, "the control arm must be the engine default" + assert off_watcher is None + + on_kwargs, on_watcher = ab._arm(True) + assert on_kwargs.get("allocator") is on_watcher, ( + "the treatment arm must pass the watched allocator to evolve()") + assert ab._fired_counts(on_watcher) == (0, 0), ( + "a fresh allocator has not fired yet") + from agentdescent.allocator import AllocatorContext + on_watcher.observe(AllocatorContext(), committed=False) + on_watcher._n_skipped += 1 + fired, observed = ab._fired_counts(on_watcher) + assert fired == 1 and observed == 1, ( + "fired counts decisions where the allocator actually changed the spend " + "(a skip), observed counts every decision it saw") + + +def test_counterfactual_only_does_not_count_as_fired(): + """`_fired_counts` must not count counterfactual exploration as a spend + change: a counterfactual deep-evals exactly like the clock would, so an + allocator that only ever explored (never skipped) is still a control + against a control.""" + import bench.ab_allocator as ab + from agentdescent.allocator import AllocatorContext + watcher = ab._arm(True)[1] + watcher.observe(AllocatorContext(), committed=True) + watcher._n_counterfactual += 1 # exploration only, no skip + fired, observed = ab._fired_counts(watcher) + assert fired == 0, ( + "counterfactual exploration alone must not count as fired") + assert observed == 1 + # a real skip does count + watcher._n_skipped += 1 + fired, observed = ab._fired_counts(watcher) + assert fired == 1, "a real skip is the allocation change that fired counts" + + +def test_allocator_seed_controls_exploration_and_is_reproducible(): + """`decide` must draw counterfactual exploration from the allocator's own + seeded RNG, not the process global -- two allocators with the same seed + explore identically, two with different seeds do not interfere.""" + from agentdescent.allocator import AllocatorContext + + def _explore(seed): + alloc = ValueBudgetAllocator( + min_observations=0, epsilon_start=0.8, epsilon_end=0.8, seed=seed) + # force the model to be confident + predict low value, so decide rolls + for _ in range(5): + alloc.observe(AllocatorContext(advantage=-2.0, size=1), False) + low = AllocatorContext(advantage=-2.0, size=1) + return [alloc.decide(low).counterfactual for _ in range(40)] + + a1 = _explore(7) + a2 = _explore(7) # same seed -> identical exploration + assert a1 == a2, "same seed must give the same exploration decisions" + b1 = _explore(99) + assert a1 != b1, "different seeds must give different exploration draws" + + +def test_size_magnitude_does_not_dominate_the_advantage(): + """The size column is normalised so a raw-op magnitude (1..100+) cannot + blow its logistic weight and override the advantage (a z-score) in the + skip decision. Before normalisation a candidate with high advantage but a + large diff was predicted near-certain *not* to commit, purely because the + raw size column swamped the fit.""" + m = LearningValueModel(lr=0.2, reg=0.05, max_logit=3.0) + # Large-diff candidates never commit; small-diff ones (with high advantage) + # always commit. + for _ in range(30): + m.update((0.1, _nsize(50), 0.2), False) + m.update((1.0, _nsize(1), 0.2), True) + # A new candidate with high advantage and a large diff must still lean + # commit -- the advantage signal must survive, not be swamped by size. + p_hi_adv_big = m.predict((1.0, _nsize(50), 0.2)) + assert p_hi_adv_big > 0.5, ( + f"high advantage must not be vetoed by a large diff size; got " + f"P(commit)={p_hi_adv_big:.3f}") + # ...and a low-advantage large-diff candidate stays a clear reject. + p_low_adv_big = m.predict((0.1, _nsize(50), 0.2)) + assert p_low_adv_big < 0.5, ( + f"low advantage with large diff should reject; got P(commit)=" + f"{p_low_adv_big:.3f}") diff --git a/tests/test_dataloader.py b/tests/test_dataloader.py index 1b61e579..388d8de3 100644 --- a/tests/test_dataloader.py +++ b/tests/test_dataloader.py @@ -186,3 +186,40 @@ def test_val_frac_round_trips_through_evolve(): assert len(eng.train) == len(ds.train), ( f"n={n}: engine train={len(eng.train)} but Dataset train={len(ds.train)}") assert len(eng.held_out) == len(ds.val) + + +def test_modelscope_context_pairs_are_normalised_to_dict(): + """The ModelScope HotpotQA mirror stores ``context`` as the original + ``[[title, [sentences]], ...]`` pairs; examples read it as the dict shape + the HuggingFace datasets-server yields. The loader must convert, once, so + GEPA's `_render_context` works unchanged on either source.""" + pairs = [["Ed Wood (film)", ["sentence one.", " sentence two."]], + ["Scott Derrickson", ["Scott is an American director."]]] + out = dl._hotpotqa_context_to_dict(pairs) + assert out == {"title": ["Ed Wood (film)", "Scott Derrickson"], + "sentences": [["sentence one.", " sentence two."], + ["Scott is an American director."]]} + # already-a-dict passes through untouched (the HF datasets-server shape) + already = {"title": ["x"], "sentences": [["y"]]} + assert dl._hotpotqa_context_to_dict(already) is already + + +def test_modelscope_rows_are_cached(tmp_path, monkeypatch): + """`hf_rows_modelscope` fetches the ~46 MB raw file once and serves repeat + calls from cache; the HTTP round-trip is exercised without a network.""" + monkeypatch.setattr(dl, "CACHE_ROOT", str(tmp_path)) + n_fetch = {"n": 0} + + def fake_read(url, timeout): + n_fetch["n"] += 1 + return json.dumps([ + {"question": "Q1", "answer": "A1", + "context": [["t1", ["s1"]]]}, + ]).encode() + + monkeypatch.setattr(dl, "_read_url", fake_read) + rows1 = dl.hf_rows_modelscope(limit=10) + rows2 = dl.hf_rows_modelscope(limit=10) # served from cache + assert n_fetch["n"] == 1 + assert len(rows1) == 1 and len(rows2) == 1 + assert rows1[0]["context"] == {"title": ["t1"], "sentences": [["s1"]]}