feat(budget): value-directed allocation — learn which candidate is worth the expensive evaluation, with counterfactual exploration

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.

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

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,20 +1362,38 @@ 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))
 
         # -- 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)