feat(fusion): LearnableFusionPolicy — the merge learns when merging wins

agentdescent/__init__.py

@@ -50,7 +50,7 @@
     check_dependency_graph,
     check_invariants,
 )
-from .fusion import KeepContradictions, ReflectiveFusion, reflective_merge
+from .fusion import KeepContradictions, ReflectiveFusion, LearnableFusionPolicy, reflective_merge
 from .sampling import DifficultyWeighted, ReplaySampler, RoundRobin, TaskSampler
 from .replay import ReplayAwareProposal
 from .selection import (
@@ -267,6 +267,7 @@
     "TaskSampler",
     "ReflectiveFusion",
     "KeepContradictions",
+    "LearnableFusionPolicy",
     "DefaultConflict",
     "DefaultFusion",
     "reflective_merge",

agentdescent/fusion.py

@@ -56,13 +56,16 @@
 
 from __future__ import annotations
 
+import math
+import random
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
 
+from .defaults import DefaultFusion
 from .evolvable import Diff, Evolvable, combined_rationale
 from .policies import FusionTrial
 
 __all__ = ["MERGE_PROMPT", "KeepContradictions", "ReflectiveFusion",
-           "reflective_merge"]
+           "LearnableFusionPolicy", "reflective_merge"]
 
 
 #: What the model is asked, and it is asked for a **union of deltas** rather than
@@ -335,3 +338,242 @@ def reflective_merge(complete, **kwargs) -> Dict[str, Any]:
     """
     return {"fusion": ReflectiveFusion(complete, **kwargs),
             "conflict": KeepContradictions()}
+
+
+class LearnableFusionPolicy:
+    """A fusion policy that learns when merging wins -- and stops paying for
+    the ranking it cannot change.
+
+    The tournament (``DefaultFusion(tournament=True)``) answers "does merging
+    beat the best single?" by scoring every candidate on the cheap layer, every
+    merge. The answer lands in a :class:`~agentdescent.policies.FusionTrial`
+    and is then thrown away. This policy learns the answer instead: an online
+    logistic over **diff-only features** (zero verifier calls) is trained on
+    the tournament's own trials, and once it is confident that the union is the
+    right answer for a merge shape, it returns the union **without buying the
+    cheap sweep**. When the prediction is uncertain -- or says the tournament
+    would change the outcome -- it runs the tournament and observes a fresh
+    label.
+
+    This is the allocator's idea (#204) applied to the fusion decision: value-
+    directed spending of the ranking cost, learned from the run's own history,
+    with the same safety. Three guarantees:
+
+    * **The default is the default.** With no trials the policy is
+      :class:`DefaultFusion(tournament=True)` verbatim -- every merge is
+      ranked, every trial recorded. Learning changes nothing until it has data.
+    * **Counterfactual verification.** ``epsilon`` of the confident skips are
+      still run through the tournament, so a wrong prediction is observed
+      rather than reinforced (the allocator's counterfactual exploration).
+    * **A rollback guard.** A counterfactual merge the model was *confident*
+      about skipping but the tournament ranked a single instead breaks the
+      trust, and after ``max_wrong_skips`` such events (default one) skipping
+      is retired permanently -- a rule that learned to be wrong is disabled,
+      not patched. The cost of a false alarm is merely losing the saving; the
+      cost of a wrong skip is committing a worse merge.
+
+    Commits are unaffected: the skip returns the same union the non-tournament
+    default would have committed, and the acceptance gate still scores it on
+    the full held-out set before anything commits.
+    """
+
+    def __init__(
+        self,
+        inner: Optional[Any] = None,
+        *,
+        tournament: bool = True,
+        seed: int = 0,
+        min_trials: int = 20,
+        uncertainty_threshold: float = 0.15,
+        epsilon: float = 0.1,
+        max_wrong_skips: int = 1,
+        lr: float = 0.2,
+        reg: float = 0.05,
+        max_logit: float = 3.0,
+    ) -> None:
+        if not (0.0 < uncertainty_threshold < 1.0):
+            raise ValueError("uncertainty_threshold must be in (0, 1)")
+        if not (0.0 <= epsilon <= 1.0):
+            raise ValueError("epsilon must be in [0, 1]")
+        if max_wrong_skips < 1:
+            raise ValueError("max_wrong_skips must be >= 1")
+        if not (isinstance(min_trials, int) and min_trials >= 1):
+            raise ValueError("min_trials must be an int >= 1")
+        for name, value in (("lr", lr), ("reg", reg), ("max_logit", max_logit)):
+            if not (isinstance(value, (int, float)) and math.isfinite(value)
+                    and value > 0.0):
+                raise ValueError(f"{name} must be a finite number > 0")
+        self.inner = inner or DefaultFusion(tournament=tournament)
+        #: The engine's budget governor flips this when tokens run low
+        #: (``aggregator.fusion_policy.tournament = should``); turning it off
+        #: degrades to the union path, no labels collected.
+        self.tournament = tournament
+        #: The engine reads ``aggregator.fusion_policy.trials`` for
+        #: ``EvolutionResult.fusion_stats``. These are the inner tournament's
+        #: real measured trials; a skipped merge has none by construction.
+        self.trials = self.inner.trials
+        self.min_trials = int(min_trials)
+        self.threshold = float(uncertainty_threshold)
+        self.epsilon = float(epsilon)
+        self.max_wrong_skips = int(max_wrong_skips)
+        self._rng = random.Random(seed)
+        self._model = _FusionModel(n_features=5, lr=lr, reg=reg,
+                                   max_logit=max_logit)
+        self._union_policy = DefaultFusion(tournament=False)
+        #: The budget-degraded union path records unranked trials, and those
+        #: must land on the **same** list the engine reads -- otherwise
+        #: ``fusion_stats`` goes silent exactly when the governor flipped the
+        #: tournament off, which is the one time it should still speak.
+        self._union_policy.trials = self.inner.trials
+        self._skip_enabled = True
+        #: Consecutive counterfactual merges the model would have skipped but
+        #: the tournament ranked a single instead (a wrong skip in a row).
+        self._wrong_skips = 0
+        #: Confident skips that avoided the cheap sweep.
+        self.skipped = 0
+        #: Merges actually ranked (the labels).
+        self.tournament_runs = 0
+
+    # -- engine hooks ---------------------------------------------------------
+
+    def bind(self, verifier: Any) -> None:
+        """Forward the verifier to the inner tournament and the union path."""
+        self.inner.bind(verifier)
+        self._union_policy.bind(verifier)
+
+    def configure(self, config: Any) -> None:
+        pass
+
+    # -- the fusion decision --------------------------------------------------
+
+    def select(self, artifact: Evolvable,
+               diffs: List[Diff]) -> Tuple[Diff, Evolvable, bool]:
+        """Learn or predict: skip the ranking sweep when the union is
+        confidently predicted to win (a learned logistic, not a proof)."""
+        if not self.tournament:
+            return self._union_policy.select(artifact, diffs)
+        union, _reason = self._union_of(diffs)
+        if (self._skip_enabled and self._model.n_obs >= self.min_trials
+                and union is not None
+                and self._rng.random() >= self.epsilon):
+            p = self._model.predict(self._features(diffs, union))
+            if p >= 1.0 - self.threshold:
+                self.skipped += 1
+                return self._union_policy.select(artifact, diffs)
+        return self._run_tournament(artifact, diffs)
+
+    def _run_tournament(self, artifact: Evolvable,
+                        diffs: List[Diff]) -> Tuple[Diff, Evolvable, bool]:
+        """Rank the candidates (the label-producing path), then observe."""
+        self.tournament_runs += 1
+        result = self.inner.select(artifact, diffs)
+        fused = bool(result[2])
+        union, _reason = self._union_of(diffs)
+        self._observe(self._features(diffs, union), fused)
+        return result
+
+    def _observe(self, features: Tuple[float, ...], fused: bool) -> None:
+        # Snapshot the **pre-update** prediction first. The rollback question
+        # is "would the rule that was in force have skipped this merge?" -- and
+        # a negative label immediately drags the logistic down, often below the
+        # confidence cut in one step, so scoring after the update misses exactly
+        # the confident-and-wrong counterfactual the guard exists for.
+        predicted_skip = self._model.predict(features) >= 1.0 - self.threshold
+        self._model.update(features, fused)
+        if self._model.n_obs < self.min_trials:
+            return
+        # Skipping is right iff the tournament also picked the union
+        # (``fused``). The rollback guard is deliberately unforgiving: a
+        # counterfactual merge the model was *confident* about skipping but the
+        # tournament ranked a single instead breaks the trust, and after
+        # ``max_wrong_skips`` such events (default one) skipping is retired
+        # permanently. The cost of a false alarm is merely losing the saving;
+        # the cost of a wrong skip is committing a worse merge. Conservative is
+        # correct here.
+        if predicted_skip and not fused:
+            self._wrong_skips += 1
+            if self._wrong_skips >= self.max_wrong_skips:
+                self._skip_enabled = False
+        else:
+            self._wrong_skips = 0
+
+    # -- features (diff-only, zero verifier calls) -----------------------------
+
+    def _union_of(self, diffs: List[Diff]) -> Tuple[Optional[Diff], str]:
+        """The fused candidate, via the inner policy's own rule.
+
+        Deliberately delegated rather than copied: ``DefaultFusion._union_of``
+        is the single source of the three-way "nothing-to-fuse" rule, and a
+        byte-for-byte copy here would silently drift if that rule ever changed
+        again. The inner is the tournament policy the learner wraps; it must
+        have a ``_union_of`` (``DefaultFusion`` does)."""
+        return self.inner._union_of(diffs)
+
+    @staticmethod
+    def _features(diffs: List[Diff], union: Optional[Diff]) -> Tuple[float, ...]:
+        """The pre-fusion signal the learner decides on: merge shape, size and
+        key overlap, all computable from the diffs alone."""
+        import math as _math
+
+        keys = [set(d.ops) for d in diffs]
+        union_keys = set().union(*keys) if keys else set()
+        inter = set.intersection(*keys) if keys else set()
+        overlap = len(inter) / len(union_keys) if union_keys else 0.0
+        total_ops = sum(len(d.ops) for d in diffs)
+        max_ops = max((len(d.ops) for d in diffs), default=0)
+        return (
+            _math.log1p(len(diffs)),
+            overlap,
+            _math.log1p(total_ops),
+            _math.log1p(max_ops),
+            1.0 if union is not None else 0.0,
+        )
+
+    def status(self) -> dict:
+        # The skip fraction's denominator is *all* merges decided (skipped or
+        # ranked) -- "skips / tournament_runs" would climb past 1.0 on a
+        # warm skip-heavy run and read as an occupancy instead of a fraction.
+        decided = self.skipped + self.tournament_runs
+        return {
+            "skipped": self.skipped,
+            "tournament_runs": self.tournament_runs,
+            "model_observations": self._model.n_obs,
+            "skip_enabled": self._skip_enabled,
+            "skips_per_merge": (self.skipped / decided
+                                if decided else 0.0),
+        }
+
+
+class _FusionModel:
+    """A tiny online logistic over the fusion features.
+
+    Deliberately a private twin of the allocator's
+    :class:`~agentdescent.allocator.LearningValueModel` rather than a reuse:
+    same idea (momentum learning rate, L2, logit clip), different feature count
+    and weight layout (bias at ``[-1]`` here), and sharing the allocator's
+    hard-coded three-feature class would force this module's five features into
+    a layout that does not fit. None of this is a research contribution -- it
+    is the smallest thing that learns without diverging on one noisy run."""
+
+    def __init__(self, n_features: int, lr: float = 0.2, reg: float = 0.05,
+                 max_logit: float = 3.0) -> None:
+        self.n_features = int(n_features)
+        self.lr = lr
+        self.reg = reg
+        self.max_logit = max_logit
+        self._w = [0.0] * (n_features + 1)
+        self.n_obs = 0
+
+    def predict(self, features: Sequence[float]) -> float:
+        logit = self._w[-1] + sum(
+            w * x for w, 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 update(self, features: Sequence[float], positive: bool) -> None:
+        p = self.predict(features)
+        err = (1.0 if positive else 0.0) - p
+        self._w[-1] += self.lr * err - self.lr * self.reg * self._w[-1]
+        for i, x in enumerate(features):
+            self._w[i] += self.lr * err * x - self.lr * self.reg * self._w[i]
+        self.n_obs += 1

changelog.d/298-learnable-fusion.md

@@ -0,0 +1,15 @@
+Added
+LearnableFusionPolicy (#298): a `FusionPolicy` that learns when merging wins.
+An online logistic over diff-only features (candidate count, key overlap, diff
+sizes) is trained on the tournament's own `FusionTrial`s, and once confident
+that the union is the right answer for a merge shape it returns the union
+**without buying the cheap ranking sweep**; uncertain merges still run the
+tournament and observe a fresh label. The rollback guard scores the **pre-
+update** prediction -- the rule actually in force -- and retires skipping
+permanently after a confident-and-wrong counterfactual (default
+`max_wrong_skips=1`). The budget-governor's `tournament=False` degrades to the
+plain union and keeps recording unranked trials on the list `fusion_stats`
+reads. With no trials the policy is `DefaultFusion(tournament=True)` verbatim.
+Commits are unaffected: a skip returns the same union the non-tournament
+default would have committed, and the acceptance gate still scores it on the
+full held-out set.

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*.
 
-342 public names across 56 modules.
+343 public names across 56 modules.
 
 ---
 
@@ -2877,6 +2877,31 @@ Combine competing values for the same key, when a dict update cannot.  ·&n
 
 A conflict policy that leaves contradicting diffs for fusion to resolve.
 
+### `LearnableFusionPolicy(...)`
+
+A fusion policy that learns when merging wins -- and stops paying for the ranking it cannot change.
+
+```python
+LearnableFusionPolicy(
+    inner: Optional[Any] = None,
+    *,
+    tournament: bool = True,
+    seed: int = 0,
+    min_trials: int = 20,
+    uncertainty_threshold: float = 0.15,
+    epsilon: float = 0.1,
+    max_wrong_skips: int = 1,
+    lr: float = 0.2,
+    reg: float = 0.05,
+    max_logit: float = 3.0
+) -> None
+```
+
+| method | what it does |
+|---|---|
+| `bind(verifier: Any) -> None` | Forward the verifier to the inner tournament and the union path. |
+| `select(artifact: Evolvable, diffs: List[Diff]) -> Tuple[Diff, Evolvable, bool]` | Learn or predict: skip the ranking sweep when the union is confidently predicted to win (a learned logistic, not a proof). |
+
 ### `ReflectiveFusion(...)`
 
 Combine contradicting diffs by asking a model to synthesise their values.