feat(ledger): multiple live heads — fork, head_for_worker, branch-aware commit

CHANGELOG.md

@@ -4402,3 +4402,19 @@ First public release on PyPI as **`agentdescent`**.
 [0.3.0]: https://github.com/Birfy/agentdescent/compare/v0.2.0...v0.3.0
 [0.2.0]: https://github.com/Birfy/agentdescent/compare/v0.1.0...v0.2.0
 [0.1.0]: https://github.com/Birfy/agentdescent/releases/tag/v0.1.0
+
+- **`Ledger.fork` / `live_heads` / `discard_head` — multiple live heads (#75).**
+  The ledger held one `dev` branch; `PopulationAggregator` serialised candidates
+  by rewriting it, so no two workers could expand different candidates at once.
+  `fork("head/0")` creates an independent branch with its own version vector;
+  `live_heads()` lists `dev` plus all `head/` forks; `discard_head` cleans up.
+  `Candidate.branch` and `EvidenceCard.branch` carry the fork name.
+  `PopulationAggregator.step(n_workers)` forks a `head/<slot>` per selected
+  candidate and `head_for_worker(wid)` distributes workers across them.
+  `_commit_with_retry` commits to `EvidenceCard.branch`; the sync and async
+  worker loops snapshot from `head_for_worker` instead of hardcoded `dev`.
+  `_decide` passes the candidate's own branch through to the commit (and the
+  CAS-rebase path rebases against that branch, not `dev`), so a fork candidate
+  updates only its fork. `finalize` discards all forks and commits the best
+  candidate to `dev`.
+  Default: `SingleHead` + no forks = old behaviour exactly. 12 tests.

agentdescent/aggregator.py

@@ -80,7 +80,7 @@ class AggregatorProtocol(Protocol):
 
     def ingest(self, card: EvidenceCard) -> None: ...
 
-    def step(self) -> List["MergeReport"]: ...
+    def step(self, n_workers: int = 1) -> List["MergeReport"]: ...
 
     #: **Checkpointing** (optional, duck-typed — not part of the Protocol
     #: contract). An aggregator that holds search state beyond what the
@@ -337,6 +337,8 @@ class _Candidate:
     conflicts: int
     #: Diffs the trust region refused in this batch; carried onto the report.
     oversized: int = 0
+    #: The ledger branch this candidate was proposed against.
+    branch: Optional[str] = None
 
     #: Filled by `Aggregator._measure`. `None` until then, which is what makes a
     #: candidate that skipped the phase fail loudly rather than commit on zeros.
@@ -415,11 +417,18 @@ def _attributable_card(c: "_Candidate"):
 
 
 class EvidenceBuffer:
-    """Cards bucketed by target artifact (design doc, section 4.1).
+    """Cards bucketed by (target artifact, branch) (design doc, section 4.1).
 
     Thread-safe: in the asynchronous runtime many worker threads call
     :meth:`add` concurrently while the aggregator thread calls :meth:`ready` /
-    :meth:`drain`.  All bucket mutations are guarded by an internal lock."""
+    :meth:`drain`.  All bucket mutations are guarded by an internal lock.
+
+    The bucket key is a ``(artifact_id, branch)`` pair, not just ``artifact_id``:
+    a multi-head run has several live branches (``dev``, ``head/0``, ...), and a
+    diff proposed against ``head/1`` must be merged against ``head/1``'s head,
+    not against ``dev``'s. The branch half of the key is the card's
+    :attr:`~agentdescent.evolvable.EvidenceCard.branch`, empty for single-head
+    runs so a caller that only knows artifact ids keeps working."""
 
     #: Cap on the settled pool: at most this many cards, and at most this many
     #: characters of diff payload across them. It is a *diagnostic* ring, so old
@@ -428,8 +437,8 @@ class EvidenceBuffer:
     SETTLED_MAX_CHARS = 2_000_000
 
     def __init__(self) -> None:
-        self._buckets: Dict[str, List[EvidenceCard]] = defaultdict(list)
-        self._waited: Dict[str, int] = defaultdict(int)
+        self._buckets: Dict[Tuple[str, str], List[EvidenceCard]] = defaultdict(list)
+        self._waited: Dict[Tuple[str, str], int] = defaultdict(int)
         #: Optional consumer of settled evidence, wired by the engine when a
         #: sampler (e.g. :class:`~agentdescent.sampling.ReplaySampler`) wants to
         #: learn from the pool. ``None`` keeps the old behaviour exactly.
@@ -439,30 +448,48 @@ def __init__(self) -> None:
         self._settled_chars = 0
         self._lock = threading.Lock()
 
+    def _key(self, card: EvidenceCard) -> Tuple[str, str]:
+        return (card.diff.target, card.branch or "")
+
     def add(self, card: EvidenceCard) -> None:
         with self._lock:
-            self._buckets[card.diff.target].append(card)
+            self._buckets[self._key(card)].append(card)
 
     def tick(self) -> None:
         with self._lock:
             for aid in list(self._buckets):
                 self._waited[aid] += 1
 
     def ready(self, config: AggregatorConfig) -> List[str]:
+        """Artifact ids with a ready bucket. Each id is listed once, whatever
+        branches it has: the caller drains all of them and merges per branch."""
         with self._lock:
-            out = []
-            for aid, cards in self._buckets.items():
-                if not cards:
+            ready_aids: List[str] = []
+            for (aid, _), cards in self._buckets.items():
+                if aid in ready_aids or not cards:
                     continue
-                if len(cards) >= config.batch_trigger or self._waited[aid] >= config.max_wait_rounds:
-                    out.append(aid)
-            return out
-
-    def drain(self, artifact_id: str) -> List[EvidenceCard]:
+                batch = any(len(self._buckets[(a, b)])
+                            >= config.batch_trigger
+                            for a, b in self._buckets if a == aid)
+                waited = any(self._waited[(a, b)]
+                             >= config.max_wait_rounds
+                             for a, b in self._buckets if a == aid)
+                if batch or waited:
+                    ready_aids.append(aid)
+            return ready_aids
+
+    def drain(self, aid: str) -> Dict[str, List[EvidenceCard]]:
+        """Pop every bucket for ``aid``, grouped by branch (``""`` = dev).
+
+        Returns ``{branch: [cards]}`` so the aggregator merges each branch
+        against its own head. Single-head runs have one key (``""``), which
+        is why this call site looks like the old one."""
         with self._lock:
-            cards = self._buckets.pop(artifact_id, [])
-            self._waited.pop(artifact_id, None)
-            return cards
+            grouped: Dict[str, List[EvidenceCard]] = {}
+            for key in [k for k in self._buckets if k[0] == aid]:
+                grouped[key[1]] = self._buckets.pop(key, [])
+                self._waited.pop(key, None)
+            return grouped
 
     def pending(self) -> int:
         with self._lock:
@@ -541,6 +568,14 @@ def settled_chars(self) -> int:
             return self._settled_chars
 
 
+def _branch_of(cards: Sequence["EvidenceCard"]) -> Optional[str]:
+    """The common branch across ``cards``, or ``None`` if all carry ``None``."""
+    branches = {c.branch for c in cards if c.branch is not None}
+    if len(branches) == 1:
+        return branches.pop()
+    return None
+
+
 def diffs_conflict(a: Diff, b: Diff) -> bool:
     """Syntactic overlap: do two diffs edit an overlapping set of keys?
 
@@ -752,7 +787,14 @@ def recent_settled(self, task: Any, n: int = 3) -> List[EvidenceCard]:
         See ``Buffer.recent_settled``."""
         return self.buffer.recent_settled(task, n)
 
-    def step(self) -> List[MergeReport]:
+    def head_for_worker(self, worker_id: int = 0) -> str:
+        """The ledger branch a worker should snapshot and propose against.
+
+        ``dev`` by default -- every single-head caller. A multi-head aggregator
+        overrides this to distribute workers across population forks."""
+        return Ledger.DEV
+
+    def step(self, n_workers: int = 1) -> List[MergeReport]:
         """Fire every artifact bucket that is ready and return per-artifact reports.
 
         The published contract, unchanged: it measures on the calling thread and
@@ -788,10 +830,18 @@ def begin_step(self, *, skip_in_flight: bool = False) -> List[Union["_Candidate"
         for aid in self.buffer.ready(self.config):
             if skip_in_flight and aid in self._in_flight:
                 continue
-            prepared = self._prepare(aid)
-            if isinstance(prepared, _Candidate):
-                self._in_flight.add(aid)
-            out.append(prepared)
+            # Drain every branch of this artifact and merge each against its
+            # own head. In a multi-head run a fork's cards must not be judged
+            # against dev's version vector. `_in_flight` stays keyed by the
+            # **artifact id**, not (aid, branch): the pipelined gate's contract
+            # is at most one candidate per artifact in measurement at once, and
+            # that contract predates multi-head -- an artifact id is what its
+            # tests and callers assert on.
+            for branch, cards in self.buffer.drain(aid).items():
+                prepared = self._prepare(aid, branch=branch, cards=cards)
+                if isinstance(prepared, _Candidate):
+                    self._in_flight.add(aid)
+                out.append(prepared)
         return out
 
     def measure(self, items: List[Union["_Candidate", MergeReport]]) -> List[Union["_Candidate", MergeReport]]:
@@ -1172,7 +1222,8 @@ def _audit(self, artifact, artifact_id, best_state, best_diff,
                 return True
         return False
 
-    def _commit_with_retry(self, artifact_id, candidate, diff, head):
+    def _commit_with_retry(self, artifact_id, candidate, diff, head,
+                           branch: str = Ledger.DEV):
         """CAS, rebasing onto whatever landed first. ``None`` when it kept losing.
 
         With one writer a conflict cannot happen: every commit goes through a
@@ -1199,7 +1250,7 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head):
             base_vv = {artifact_id: head.get(artifact_id, 0)}
             try:
                 _, new_version = self.ledger.commit(
-                    candidate, base_vv, branch=Ledger.DEV,
+                    candidate, base_vv, branch=branch,
                     message=f"merge {diff.diff_id} -> {artifact_id}")
                 return new_version
             except CASConflict:
@@ -1209,7 +1260,10 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head):
                     return None
                 time.sleep(random.uniform(0.0, self.config.cas_backoff)
                            * (2 ** attempt))
-                snap = self.ledger.snapshot(Ledger.DEV)
+                # Rebase against the *same branch* the original commit targeted
+                # -- a fork candidate must not rebase onto `dev`'s head, which
+                # is a different version vector entirely (PR #196 review).
+                snap = self.ledger.snapshot(branch)
                 fresh = snap.get(artifact_id)
                 if fresh is None:
                     return None
@@ -1245,18 +1299,30 @@ def _process(self, artifact_id: str) -> MergeReport:
         self._measure(prepared)
         return self._decide(prepared)
 
-    def _prepare(self, artifact_id: str) -> "Union[_Candidate, MergeReport]":
+    def _prepare(self, artifact_id: str, branch: str = "", cards=None) -> "Union[_Candidate, MergeReport]":
         """Drain, filter, and choose what to put forward -- no measurement.
 
+        ``branch`` is the ledger branch the cards were proposed against
+        (empty = ``dev``, the single-head default). Each branch is merged
+        against its **own** head: a multi-head run's ``head/1`` workers
+        propose against ``head/1``'s version vector, and their diffs must be
+        judged against that vector, not against ``dev``'s -- otherwise every
+        cross-branch diff looks stale by construction.
+
+        ``cards``, when given, is the drained evidence for this branch (the
+        caller got it from :meth:`EvidenceBuffer.drain`). When omitted, the
+        default single-head path drains ``dev``'s bucket itself.
+
         Returns a finished :class:`MergeReport` for the merges that end here: an
         artifact that is not in the ledger, and a batch with no survivors. Those
         are not failures of the phase, they are merges with nothing to measure,
         and giving them a report rather than a candidate is what lets the caller
         skip the expensive phase entirely.
         """
         self._seen.add(artifact_id)
-        cards = self.buffer.drain(artifact_id)
-        snap = self.ledger.snapshot(Ledger.DEV)
+        if cards is None:
+            cards = self.buffer.drain(artifact_id).get(branch or "", [])
+        snap = self.ledger.snapshot(branch or Ledger.DEV)
         artifact = snap.get(artifact_id)
         if artifact is None:
             return MergeReport(artifact_id, None, False, len(cards), 0, 0, 0, 0.0, None,
@@ -1301,7 +1367,8 @@ def _prepare(self, artifact_id: str) -> "Union[_Candidate, MergeReport]":
             head=head, fused=fused,
             considered=n_considered, survived=len(survivors),
             discarded=len(discarded), conflicts=conflicts,
-            oversized=len(oversized))
+            oversized=len(oversized),
+            branch=branch or _branch_of(kept_cards))
 
     def _measure(self, c: "_Candidate") -> "_Candidate":
         """Score the base and the candidate. **Touches no aggregator state.**
@@ -1420,8 +1487,13 @@ def _observe_outcome(committed: bool) -> None:
                           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)
+        # The candidate was prepared against its branch's own head (`c.head`),
+        # so it must commit to *that* branch -- a fork candidate going to `dev`
+        # would overwrite the confirmed branch and leave the fork unchanged
+        # (PR #196 review). Empty branch means `dev`, the single-head default.
+        new_version = self._commit_with_retry(
+            c.artifact_id, c.candidate, c.diff, c.head,
+            branch=c.branch or Ledger.DEV)
         committed = new_version is not None
         _observe_outcome(committed)
         if not committed:

agentdescent/async_evolve.py

@@ -47,6 +47,7 @@
     _ASYNC_WIRED_POLICIES, _cost_fields, _fusion_trials, _replay_stats,
     _resolve_policies,
     SOLVED, _build_engine, _checked_proposal, _checked_reward,
+    _head_for_worker, _step,
 )
 from .aggregator import Aggregator, AggregatorConfig, check_reports
 from .evolvable import ContractError, EvidenceCard, vv_staleness
@@ -598,7 +599,7 @@ def _collect_gate(block: bool = False) -> List[Any]:
     def _gated_step() -> List[Any]:
         """`aggregator.step()`, with phase 2 off this thread when asked."""
         if not _pipelined:
-            return eng.aggregator.step()
+            return _step(eng.aggregator, n_workers)
         reports = _collect_gate()
         items = eng.aggregator.begin_step(skip_in_flight=True)
         if any(not hasattr(i, "committed_version") for i in items):
@@ -609,7 +610,8 @@ def _gated_step() -> List[Any]:
         return reports
 
     def _worker(wid: int, shard: List[Task]) -> None:
-        snap = eng.ledger.snapshot(Ledger.DEV)
+        worker_branch = _head_for_worker(eng.aggregator, wid)
+        snap = eng.ledger.snapshot(worker_branch)
         base_v = snap.version.get(eng.artifact_id, 0)
         artifact = snap.get(eng.artifact_id)
         shard_ids = [t.id for t in shard]          # the sampler works on ids
@@ -656,7 +658,13 @@ def _worker(wid: int, shard: List[Task]) -> None:
             forced = epoch[0] != local_epoch
             committed_since = commit_epoch[0] != local_commit
             if head_v - base_v > async_ratio or forced or committed_since:
-                snap = eng.ledger.snapshot(Ledger.DEV)
+                # Re-select the branch on every refresh. `PopulationAggregator
+                # .step()` creates forks when a batch is selected, so a worker
+                # that started before the forks existed would otherwise snapshot
+                # and submit against `dev` for the whole run -- the advertised
+                # parallel heads never explored.
+                worker_branch = _head_for_worker(eng.aggregator, wid)
+                snap = eng.ledger.snapshot(worker_branch)
                 base_v = snap.version.get(eng.artifact_id, 0)
                 artifact = snap.get(eng.artifact_id)
                 local_epoch = epoch[0]
@@ -722,6 +730,7 @@ def _worker(wid: int, shard: List[Task]) -> None:
                                 diff=diff, base_version={eng.artifact_id: base_v},
                                 touched=[eng.artifact_id], before_after_delta=delta,
                                 trajectory_refs=[task],
+                                branch=worker_branch,
                                 # Same signal as the synchronous path, and the
                                 # reason `GroupAdvantage` accumulates rather than
                                 # batching at a barrier: there is no barrier here,