From c18c5ddb2bcc2820836019139e9b6376b0ab9df6 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:00:03 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat(ledger):=20multiple=20live=20heads=20?= =?UTF-8?q?=E2=80=94=20fork,=20head=5Ffor=5Fworker,=20branch-aware=20commi?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ledger.fork(name, from_branch=DEV): creates a head/ branch with its own version vector. A commit against it does not touch dev. live_heads() lists dev plus all head/ forks. discard_head cleans up. Fork names must start with head/ (git cannot have both dev as a ref file and dev/0 as a ref directory). Candidate.branch and EvidenceCard.branch carry the fork name through the selection and merge pipeline so the aggregator knows where to commit back. Aggregator.head_for_worker(worker_id) returns dev by default. _Candidate carries branch, _branch_of(cards) reads it from evidence, and _commit_with_retry commits to the candidate's branch. step(n_workers=1) accepts a worker count. _in_flight stays keyed by artifact id, not (aid, branch) — the pipelined gate's contract (at most one candidate per artifact in measurement) predates multi-head and its tests assert on ids. PopulationAggregator.step(n_workers) forks a head/ per selected candidate and distributes workers across them via head_for_worker. finalize discards all forks and commits the best candidate to dev. Worker loops (sync evolution.py + async async_evolve.py): _snapshot_for returns (artifact, base_v, branch); each worker snapshots from head_for_worker(wid) via _head_for_worker (compat helper for custom aggregators without the method); async re-selects the branch on every refresh so forks created mid-run are explored. _step(aggregator, n_workers) passes the worker count only to aggregators whose step() accepts it, so a pre-multi-head zero-arg step() keeps working. 16 tests in tests/test_multi_head.py. async/pipelined/genesis regression exit 0. --- CHANGELOG.md | 13 ++ agentdescent/aggregator.py | 120 ++++++++++++---- agentdescent/async_evolve.py | 15 +- agentdescent/evolution.py | 46 +++++- agentdescent/evolvable.py | 3 + agentdescent/ledger.py | 57 ++++++++ agentdescent/population.py | 127 +++++++++++++---- agentdescent/selection.py | 4 + docs/api.md | 16 ++- tests/test_multi_head.py | 265 +++++++++++++++++++++++++++++++++++ 10 files changed, 598 insertions(+), 68 deletions(-) create mode 100644 tests/test_multi_head.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 70e9e4be..7a8418c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4402,3 +4402,16 @@ 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/` 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`. + `finalize` discards all forks and commits the best candidate to `dev`. + Default: `SingleHead` + no forks = old behaviour exactly. 12 tests. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 2ac7b3f3..5d1dd19e 100644 --- a/agentdescent/aggregator.py +++ b/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,9 +448,12 @@ 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: @@ -449,20 +461,35 @@ def tick(self) -> None: 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: @@ -1245,9 +1296,20 @@ 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, @@ -1255,8 +1317,9 @@ def _prepare(self, artifact_id: str) -> "Union[_Candidate, MergeReport]": 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 +1364,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.** diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py index a7490644..7e245968 100644 --- a/agentdescent/async_evolve.py +++ b/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, diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index 06b7f0d8..e63618e3 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -1000,6 +1000,35 @@ def _replay_stats(sampler) -> Optional[Dict[str, Any]]: } +def _head_for_worker(aggregator: Any, worker_id: int) -> str: + """The branch a worker should snapshot, if the aggregator has one. + + ``head_for_worker`` is a multi-head extension. An existing custom + aggregator (e.g. ``PromptBreederPopulation``) predates it and has no such + method -- for those, every worker starts from ``dev``, which is the + behaviour every single-head run has always had.""" + from .ledger import Ledger + fn = getattr(aggregator, "head_for_worker", None) + if fn is None: + return Ledger.DEV + return fn(worker_id) + + +def _step(aggregator: Any, n_workers: int): + """Run one merge, passing the worker count only to aggregators that want it. + + ``step(n_workers)`` is a multi-head extension. An existing custom + aggregator may implement the published zero-argument ``step()``; passing an + argument it does not accept raises ``TypeError`` mid-run. Inspect the + signature once and honour whichever shape it has.""" + import inspect + try: + accepts_n = "n_workers" in inspect.signature(aggregator.step).parameters + except (TypeError, ValueError): # a builtin or unreadable signature + accepts_n = False + return aggregator.step(n_workers) if accepts_n else aggregator.step() + + def _cost_fields(meter: Meter) -> Dict[str, Any]: """The meter's counters, keyed as :class:`EvolutionResult` fields. @@ -3046,13 +3075,21 @@ def _snapshot_for(worker: int): a worker either adopts the snapshot this round already took, or keeps the older one it has. """ + branch = _head_for_worker(aggregator, worker) + if branch != Ledger.DEV: + fork_snap = ledger.snapshot(branch) + fork_art = fork_snap.get(artifact_id) + if fork_art is not None: + fork_v = fork_snap.version.get(artifact_id, 0) + return fork_art, fork_v, branch if refresh_interval <= 1: - return artifact, base_v + return artifact, base_v, Ledger.DEV with snap_lock: due = (r % refresh_interval) == (stable_hash(worker) % refresh_interval) if worker not in worker_snaps or due: worker_snaps[worker] = (artifact, base_v) - return worker_snaps[worker] + art, ver = worker_snaps[worker] + return art, ver, Ledger.DEV def _run_unit(unit) -> None: """One worker: rollout -> propose -> ingest evidence (against `snap`). @@ -3083,7 +3120,7 @@ def _run_unit_inner(unit) -> None: return # This worker's own view of the artifact. Identical to the round's # under the default `refresh_interval=1`; older, by design, above it. - mine, mine_v = _snapshot_for(unit.worker) + mine, mine_v, mine_branch = _snapshot_for(unit.worker) task = by_id[sampler.pick(unit.keys, r)] # a task from this worker's shard # The rollout is the part that can move elsewhere. Everything around # it -- which task, what the output implies, who is told about it -- @@ -3168,6 +3205,7 @@ def _run_unit_inner(unit) -> None: aggregator.ingest(EvidenceCard( diff=diff, base_version={artifact_id: mine_v}, touched=[artifact_id], before_after_delta=delta, trajectory_refs=[task], + branch=mine_branch, # Recorded always, acted on by nobody unless a policy from # `agentdescent.advantage` is installed. It is arithmetic over # two numbers the round already has, and a signal that is only @@ -3243,7 +3281,7 @@ def _bounded(u=None): # needs no `worker_starved_seconds` of its own: here it would be # `merge_seconds x n_workers` by construction. with eng.meter.timed("merge_seconds"), eng.meter.timed("merge_gate_seconds"): - reports = check_reports(aggregator.step(), aggregator) + reports = check_reports(_step(aggregator, n_workers), aggregator) except ContractError: raise # a caller-contract violation: the run is meaningless except Exception as e: # noqa: BLE001 - a rollout backend failure (e.g. an diff --git a/agentdescent/evolvable.py b/agentdescent/evolvable.py index 7106f544..da44677b 100644 --- a/agentdescent/evolvable.py +++ b/agentdescent/evolvable.py @@ -163,6 +163,9 @@ class EvidenceCard: advantage: Optional[float] = None cost_tokens: int = 0 cost_wallclock: float = 0.0 + #: The ledger branch this diff was proposed against. ``None`` means ``dev``. + #: A multi-head run sets this so the aggregator commits back to the same branch. + branch: Optional[str] = None def rebased_onto(self, head: VersionVector) -> "EvidenceCard": """Return a copy whose base is advanced to ``head`` for touched keys. diff --git a/agentdescent/ledger.py b/agentdescent/ledger.py index 6db710df..cf39820e 100644 --- a/agentdescent/ledger.py +++ b/agentdescent/ledger.py @@ -590,3 +590,60 @@ def log(self, branch: str = DEV, limit: int = 20) -> List[str]: self.repo_path, "log", f"-{limit}", "--pretty=format:%h %s" ) return out.splitlines() if out else [] + + # -- multiple live heads (issue #75) -------------------------------------- + # + # `DEV` is the primary head and the one every existing caller reads. A + # multi-head run forks additional branches under `HEAD_PREFIX` so that a + # caller reading only `dev` still gets the primary head, and `live_heads()` + # can find the forks from git itself. + + #: The fork namespace. ``head/0``, ``head/1``, ... are live heads beside + #: ``dev``. Not ``dev/0``: git cannot have both a branch ``dev`` (a ref file) + #: and ``dev/0`` (which needs ``dev`` to be a ref *directory*). + HEAD_PREFIX = "head/" + + def fork(self, name: str, from_branch: str = DEV) -> str: + """Create or reset ``name`` to hold ``from_branch``'s current state. + + The branch is a *live head*: workers may be started from it, it has its + own version vector, and a commit against it does not touch ``dev``. + ``name`` must be a live-head name (``head/``); forking over + ``stable`` or another caller's branch is refused rather than silently + repointing it.""" + if not name.startswith(self.HEAD_PREFIX): + raise ValueError( + f"fork name must start with {self.HEAD_PREFIX!r}, got {name!r}; " + "`dev` and `stable` are not forks and resetting them here would " + "move a head other callers are reading.") + with self._exclusive(): + self._ensure_open() + self._checkout(from_branch) + _git(self.repo_path, "branch", "-f", name) + return name + + def live_heads(self) -> List[str]: + """Every live head branch: ``dev`` first, then the ``head/`` forks.""" + with self._exclusive(): + self._ensure_open() + out = _git(self.repo_path, "branch", "--list", + "--format=%(refname:short)") + names = [n.strip() for n in (out or "").splitlines() if n.strip()] + forks = sorted(n for n in names if n.startswith(self.HEAD_PREFIX)) + return [self.DEV] + forks + + def discard_head(self, name: str) -> None: + """Delete a forked head branch. + + ``dev`` and ``stable`` are refused: deleting them is not a population + operation, and a caller who means to reset them should use ``fork`` + from an explicit source.""" + if not name.startswith(self.HEAD_PREFIX): + raise ValueError( + f"discard_head refuses {name!r}: only {self.HEAD_PREFIX}* " + "branches are forks.") + with self._exclusive(): + self._ensure_open() + if self._current_branch == name: + self._checkout(self.DEV) + _git(self.repo_path, "branch", "-D", name) diff --git a/agentdescent/population.py b/agentdescent/population.py index a0890cb4..6c299bdb 100644 --- a/agentdescent/population.py +++ b/agentdescent/population.py @@ -1,20 +1,31 @@ -"""Population search on a single-head ledger: how a `SelectionPolicy` runs today. +"""Population search: how a `SelectionPolicy` runs, single head or many. :mod:`agentdescent.selection` defines *where the next batch starts*; this module -is what makes the answer take effect before the ledger can hold several live -heads. The trick is the one GEPA and DGM each found on their own: keep the -candidate pool **in the aggregator**, and make "selection" a ledger commit that -rewrites the head. The heads are serialised rather than concurrent, so one -`dev` branch is enough. +is what makes the answer take effect. The candidate pool lives **in the +aggregator**, and "selection" is a ledger commit that moves the head to the +chosen candidate. :class:`PopulationAggregator` subclasses the shipped :class:`~agentdescent.aggregator.Aggregator` -- staleness, conflict, fusion, statistical acceptance and promotion all run unchanged -- and adds three things around it: it archives every distinct committed head with its held-out score, asks the selection policy which archived candidate the next batch should mutate, -and commits that candidate back to ``dev``. ``finalize`` commits the archive's -best scorer, so a run's final artifact is its best candidate rather than -whichever one it was exploring when the budget ran out. +and commits that candidate back to the right branch. + +**Single head (the default).** One ``dev`` branch, all workers start from it, +the selection policy picks one candidate and ``dev`` is rewritten to it. This is +the serialised population that GEPA and DGM each found on their own, and it is +what a caller with no ``n_workers`` gets. + +**Multiple live heads (``Ledger.fork``).** When the run has several workers, +``step(n_workers)`` asks the policy for as many candidates as there are workers. +``chosen[0]`` becomes the primary ``dev`` head; each further candidate is +committed to its own ``head/`` fork, and workers are distributed across +them via :meth:`head_for_worker`. Every branch's head is admitted to the archive +each round, so a worker's improvement on any fork is visible to the next +selection. This is a **beam-style** multi-head: each round re-commits the chosen +candidates to their forks, so a fork's head is reset to its candidate between +rounds and the archive is what carries exploration forward. ## Why this lives in the engine @@ -104,6 +115,21 @@ def __init__(self, ledger, verifier, audit, config, staleness_policy=None, #: which is what a run without a token ceiling should report -- it never #: runs short of one. self.budget_remaining: float = 1.0 + #: The fork names this aggregator has created, in creation order. + #: ``head_for_worker`` cycles through them. + self._forks: List[str] = [] + + def head_for_worker(self, worker_id: int = 0) -> str: + """The branch a worker should snapshot. + + Worker 0 always gets ``dev`` (the primary head, chosen[0]). The rest + cycle through the forks (``head/0``, ``head/1``, ...), so a batch of N + workers explores up to ``1 + len(self._forks)`` candidates in parallel. + With no forks created yet, every worker gets ``dev`` (the old + behaviour).""" + if worker_id == 0 or not self._forks: + return Ledger.DEV + return self._forks[(worker_id - 1) % len(self._forks)] # -- the archive --------------------------------------------------------- @@ -152,7 +178,8 @@ def _best_state(self) -> Optional[Dict[str, str]]: best = max(self._archive, key=lambda entry: entry["score"]) return dict(best["state"]) - def _commit_state(self, state: Dict[str, str], message: str) -> Optional[int]: + def _commit_state(self, state: Dict[str, str], message: str, + branch: str = Ledger.DEV) -> Optional[int]: """Replace the head's state outright, deletions included. ``ops`` carries ``None`` for every key the head has and the target does @@ -161,7 +188,7 @@ def _commit_state(self, state: Dict[str, str], message: str) -> Optional[int]: it had ever accumulated and "start from candidate C" quietly meant "start from C plus the incumbent". See the module docstring. """ - snap = self.ledger.snapshot(Ledger.DEV) + snap = self.ledger.snapshot(branch) head = snap.get(self.population_artifact) if head is None or dict(head.state) == state: return None @@ -174,7 +201,7 @@ def _commit_state(self, state: Dict[str, str], message: str) -> Optional[int]: ops=ops, author="population")) try: _, version = self.ledger.commit(candidate, base_vv, - branch=Ledger.DEV, message=message) + branch=branch, message=message) return version except CASConflict: return None @@ -201,7 +228,7 @@ def _offered(self, chosen: Candidate, candidates: List[Candidate]) -> None: # -- AggregatorProtocol -------------------------------------------------- - def step(self) -> List[MergeReport]: + def step(self, n_workers: int = 1) -> List[MergeReport]: # Admit the pre-merge head first: on the first step that is the seed, # which a post-merge-only admit would lose the moment anything commits # over it -- and a population that forgot its seed cannot fall back. @@ -211,15 +238,29 @@ def step(self) -> List[MergeReport]: self._admit(pre_head, before.version.get(self.population_artifact, 0)) reports = super().step() - snap = self.ledger.snapshot(Ledger.DEV) - head = snap.get(self.population_artifact) - if head is None: + # Admit the head of **every** live branch, not just dev. In a multi-head + # run, workers commit their proposals back to their own fork, so the + # fork heads are where new candidates appear. Admitting only dev would + # throw every fork's exploration away -- the archive would never see a + # candidate a worker improved, and selection could never choose it. + # ``dev`` is in ``live_heads()``, so this covers it too; the per-branch + # admit below is gone. + any_head = False + for branch_name in self.ledger.live_heads(): + snap = self.ledger.snapshot(branch_name) + branch_head = snap.get(self.population_artifact) + if branch_head is not None: + self._admit(branch_head, + snap.version.get(self.population_artifact, 0)) + any_head = True + if not any_head: return reports - head_version = snap.version.get(self.population_artifact, 0) - self._admit(head, head_version) candidates = self._candidates() if len(candidates) < 2: return reports + head = self.ledger.snapshot(Ledger.DEV).get(self.population_artifact) + if head is None: + return reports head_candidate = next( (c for c in candidates if dict(c.state) == dict(head.state)), candidates[0]) @@ -242,32 +283,60 @@ def step(self) -> List[MergeReport]: # the same entry forever. `Beam(4)` was `Beam(1)`, and `ParetoFrontier` # sat on whichever front member was admitted first, usually the seed. ctx = SelectionContext(head=head_candidate, candidates=tuple(candidates), - round=self._selections, n_workers=1, + round=self._selections, n_workers=n_workers, budget_remaining=self.budget_remaining) self._selections += 1 - chosen = list(self.selection.select(ctx, 1)) + # Ask for as many starting points as there are workers, so a multi-head + # run can explore several candidates in parallel. The ledger's fork + # mechanism (``head/``) holds each one as its own branch. + n_heads = min(len(candidates), n_workers if n_workers else 1) + chosen = list(self.selection.select(ctx, n_heads)) if not chosen: return reports + # chosen[0] is the primary head -> dev. The rest become forks, and each + # fork is **created before** the candidate is committed to it, so a + # worker that starts from head/1 mid-step never sees a half-written fork. + self._forks = [] self._offered(chosen[0], candidates) - target = dict(chosen[0].state) - with self._archive_lock: - for entry in self._archive: - if dict(entry["state"]) == target: - entry["selected"] = int(entry["selected"]) + 1 - break - version = self._commit_state(target, "population: select parent") + for slot, cand in enumerate(chosen): + target = dict(cand.state) + with self._archive_lock: + for entry in self._archive: + if dict(entry["state"]) == target: + entry["selected"] = int(entry["selected"]) + 1 + break + if slot == 0: + version = self._commit_state(target, "population: select parent") + else: + fork_name = f"head/{slot - 1}" + self.ledger.fork(fork_name, from_branch=Ledger.DEV) + self._commit_state(target, f"population: head/{slot - 1}", + branch=fork_name) + self._forks.append(fork_name) if version is not None: reports.append(MergeReport( self.population_artifact, None, False, 0, 0, 0, 0, 0.0, version, - reason=f"population: parent switched (archive={len(candidates)})", + reason=f"population: parent switched (archive={len(candidates)}, " + f"heads={len(self._forks)})", category="population-select")) return reports def finalize(self) -> None: - """Leave the best-scoring candidate on the head, then promote.""" + """Leave the best-scoring candidate on the head, then promote. + + Multi-head cleanup: discard all forks so the ledger ends with one + ``dev`` branch holding the best candidate. A fork that scored higher + than ``dev`` is committed to ``dev`` first.""" best = self._best_state() if best is not None: self._commit_state(best, "population: final best") + # Discard forks + for fork_name in list(self._forks): + try: + self.ledger.discard_head(fork_name) + except Exception: # noqa: BLE001 + pass + self._forks = [] super().finalize() # -- checkpointing ------------------------------------------------------- diff --git a/agentdescent/selection.py b/agentdescent/selection.py index f3edafce..70d5a237 100644 --- a/agentdescent/selection.py +++ b/agentdescent/selection.py @@ -121,6 +121,10 @@ class Candidate: selected: int = 0 #: Version this candidate was derived from, for tree-shaped policies. parent: Optional[int] = None + #: The ledger branch this candidate lives on. ``None`` means ``dev`` (the + #: primary head). A multi-head policy sets it to a ``head/`` fork so + #: the engine knows which branch to snapshot for this worker. + branch: Optional[str] = None #: A policy prior -- AlphaZero's ``P(s,a)``, the slot :class:`FlatPuct` #: otherwise fills with a uniform ``1/N``. ``None`` means "no opinion" and #: falls back to uniform, which is what an unrated candidate must get: zero diff --git a/docs/api.md b/docs/api.md index b47694d9..9c0ea1a4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -986,7 +986,8 @@ EvidenceCard( trajectory_refs: List[Any] = , advantage: Optional[float] = None, cost_tokens: int = 0, - cost_wallclock: float = 0.0 + cost_wallclock: float = 0.0, + branch: Optional[str] = None ) -> None ``` @@ -1041,11 +1042,12 @@ Aggregator( | `checkpoint() -> Optional[dict]` | Serialise the search state the ledger does not persist. | | `finalize() -> None` | Publish the current dev head to stable at the end of a clean run. | | `finish_step(items: List[Union['_Candidate', MergeReport]]) -> List[MergeReport]` | Phase 3: decide the measured candidates, then age and promote. | +| `head_for_worker(worker_id: int = 0) -> str` | The ledger branch a worker should snapshot and propose against. | | `measure(items: List[Union['_Candidate', MergeReport]]) -> List[Union['_Candidate', MergeReport]]` | Phase 2 for a batch from `begin_step`. **Off-thread safe.** | | `recent_settled(task: Any, n: int = 3) -> List[EvidenceCard]` | The most recent N settled cards whose trajectory_refs mention task. | | `restore(state: dict) -> None` | Restore state written by `checkpoint`. | | `set_settled_consumer(consumer: Optional[Callable[[EvidenceCard], None]]) -> None` | Subscribe a consumer to the settled-evidence pool. | -| `step() -> List[MergeReport]` | Fire every artifact bucket that is ready and return per-artifact reports. | +| `step(n_workers: int = 1) -> List[MergeReport]` | Fire every artifact bucket that is ready and return per-artifact reports. | ### `AggregatorConfig(...)` @@ -1080,10 +1082,12 @@ The contract a custom aggregator must satisfy to plug into `evolve`. ### `EvidenceBuffer() -> None` -Cards bucketed by target artifact (design doc, section 4.1). +Cards bucketed by (target artifact, branch) (design doc, section 4.1). | method | what it does | |---|---| +| `drain(aid: str) -> Dict[str, List[EvidenceCard]]` | Pop every bucket for `aid`, grouped by branch (`""` = dev). | +| `ready(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. | | `recent_settled(task: Any, n: int = 3) -> List[EvidenceCard]` | The most recent N settled cards whose `trajectory_refs` mention `task`. | | `set_settled_consumer(consumer: Optional[Callable[[EvidenceCard], None]]) -> None` | Subscribe a consumer to every card that is settled. | | `settle(cards: List[EvidenceCard]) -> None` | Keep discarded-diff evidence addressable, under a hard bound. | @@ -1169,6 +1173,9 @@ Ledger( | `close() -> None` | Refuse further use of this ledger. Idempotent. | | `commit(...)` | Compare-and-swap commit of a single artifact. | | `commit_atomic(...)` | Two-phase, all-or-nothing commit of several artifacts. | +| `discard_head(name: str) -> None` | Delete a forked head branch. | +| `fork(name: str, from_branch: str = 'dev') -> str` | Create or reset `name` to hold `from_branch`'s current state. | +| `live_heads() -> List[str]` | Every live head branch: `dev` first, then the `head/` forks. | | `promote_to_stable(artifact_id: str) -> Optional[int]` | EMA-style confirmation: copy dev's current artifact onto stable. | | `register(artifact: Evolvable, branch: str = 'dev') -> None` | Add a brand-new artifact at version 1 on both branches. | | `snapshot(branch: str = 'dev') -> Snapshot` | Materialize every artifact on `branch` into live Evolvables. | @@ -2797,8 +2804,9 @@ PopulationAggregator( |---|---| | `checkpoint() -> Optional[dict]` | Serialise the archive and the selection counter. | | `finalize() -> None` | Leave the best-scoring candidate on the head, then promote. | +| `head_for_worker(worker_id: int = 0) -> str` | The branch a worker should snapshot. | | `restore(state: dict) -> None` | Restore the archive written by `checkpoint`. | -| `step() -> List[MergeReport]` | Fire every artifact bucket that is ready and return per-artifact reports. | +| `step(n_workers: int = 1) -> List[MergeReport]` | Fire every artifact bucket that is ready and return per-artifact reports. | ### `population_factory(...)` diff --git a/tests/test_multi_head.py b/tests/test_multi_head.py new file mode 100644 index 00000000..0e3f85c3 --- /dev/null +++ b/tests/test_multi_head.py @@ -0,0 +1,265 @@ +"""Multiple live heads: Ledger.fork / live_heads / discard_head, PopulationAggregator +multi-head selection, head_for_worker, EvidenceCard.branch, Candidate.branch.""" + +import tempfile +from agentdescent.evolvable import Diff, EvidenceCard +from agentdescent.evolution import AppendRules, EvolvingArtifact, Task +from agentdescent.ledger import Ledger + + +def _ledger(tmp_path): + lg = Ledger(str(tmp_path), lambda a: {"state": dict(a.state)}, + lambda aid, v, s: EvolvingArtifact(aid, s.get("state", {}), v, + 0.2, None, AppendRules())) + lg.register(EvolvingArtifact("a", {"k": "v0"}, 1, 0.2, None, AppendRules())) + return lg + + +def test_fork_creates_independent_branch(tmp_path): + lg = _ledger(tmp_path) + assert lg.live_heads() == ["dev"] + lg.fork("head/0") + assert lg.live_heads() == ["dev", "head/0"] + + +def test_commit_to_fork_does_not_touch_dev(tmp_path): + lg = _ledger(tmp_path) + lg.fork("head/0") + snap = lg.snapshot("head/0") + head = snap.get("a") + cand = head.apply(Diff(diff_id="d", target="a", ops={"k2": "v2"}, author="w")) + lg.commit(cand, {"a": snap.version["a"]}, branch="head/0") + assert lg.head_version("dev").get("a") == 1 + assert lg.head_version("head/0").get("a") == 2 + + +def test_fork_refuses_non_head_names(tmp_path): + lg = _ledger(tmp_path) + import pytest + with pytest.raises(ValueError): + lg.fork("stable") + with pytest.raises(ValueError): + lg.fork("dev") + + +def test_discard_head(tmp_path): + lg = _ledger(tmp_path) + lg.fork("head/0") + assert "head/0" in lg.live_heads() + lg.discard_head("head/0") + assert lg.live_heads() == ["dev"] + + +def test_discard_head_refuses_non_fork(tmp_path): + lg = _ledger(tmp_path) + import pytest + with pytest.raises(ValueError): + lg.discard_head("dev") + + +def test_live_heads_after_multiple_forks(tmp_path): + lg = _ledger(tmp_path) + lg.fork("head/0") + lg.fork("head/1") + lg.fork("head/2") + assert lg.live_heads() == ["dev", "head/0", "head/1", "head/2"] + + +def test_candidate_has_branch_field(): + from agentdescent.selection import Candidate + c = Candidate(artifact_id="a", version=1) + assert c.branch is None + c2 = Candidate(artifact_id="a", version=1, branch="head/0") + assert c2.branch == "head/0" + + +def test_evidence_card_has_branch_field(): + card = EvidenceCard( + diff=Diff(diff_id="d", target="a", ops={"x": "v"}, author="w"), + base_version={"a": 1}, touched=["a"]) + assert card.branch is None + card2 = EvidenceCard( + diff=Diff(diff_id="d", target="a", ops={"x": "v"}, author="w"), + base_version={"a": 1}, touched=["a"], branch="head/0") + assert card2.branch == "head/0" + + +def test_head_for_worker_returns_dev_by_default(tmp_path): + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + lg = _ledger(tmp_path) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + agg = Aggregator(lg, v, AuditScheduler(), AggregatorConfig()) + assert agg.head_for_worker(0) == "dev" + assert agg.head_for_worker(99) == "dev" + + +def test_population_head_for_worker_cycles_forks(tmp_path): + from agentdescent.aggregator import AggregatorConfig + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + from agentdescent.population import PopulationAggregator + from agentdescent.selection import SingleHead + lg = _ledger(tmp_path) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + agg = PopulationAggregator(lg, v, AuditScheduler(), AggregatorConfig(batch_trigger=1), + selection=SingleHead(), artifact_id="a") + # Before any forks: dev + assert agg.head_for_worker(0) == "dev" + # Simulate forks + agg._forks = ["head/0", "head/1"] + assert agg.head_for_worker(0) == "dev" # worker 0 = primary head + assert agg.head_for_worker(1) == "head/0" + assert agg.head_for_worker(2) == "head/1" + assert agg.head_for_worker(3) == "head/0" # wraps + + +def test_population_commit_state_to_fork(tmp_path): + from agentdescent.aggregator import AggregatorConfig + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + from agentdescent.population import PopulationAggregator + from agentdescent.selection import SingleHead + lg = _ledger(tmp_path) + lg.fork("head/0") + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + agg = PopulationAggregator(lg, v, AuditScheduler(), AggregatorConfig(batch_trigger=1), + selection=SingleHead(), artifact_id="a") + ver = agg._commit_state({"k": "v1", "k3": "v3"}, "test", branch="head/0") + assert ver is not None + assert lg.head_version("dev").get("a") == 1 + assert lg.head_version("head/0").get("a") == 2 + + +def test_default_evolve_unchanged(tmp_path): + """Default path (no population, no audit drain) must be unchanged.""" + from agentdescent.evolution import evolve + from agentdescent.strategies import AppendRules + tasks = [Task(id=f"t{i}", prompt=f"p{i}") for i in range(6)] + r = evolve(tasks, lambda t, o: 0.0, + run=lambda rd, t: "a", propose=lambda rd, t, o, r: None, + strategy=AppendRules(), + max_rollouts=4, n_workers=1, rounds=2) + + +def test_evidence_buffer_buckets_by_branch(): + """Cards from different branches must land in separate merge buckets.""" + from agentdescent.aggregator import EvidenceBuffer, AggregatorConfig + base = {"k": "v"} + d0 = Diff(diff_id="d0", target="a", ops={"k0": "v0"}, author="w0") + d1 = Diff(diff_id="d1", target="a", ops={"k1": "v1"}, author="w1") + c0 = EvidenceCard(diff=d0, base_version={"a": 1}, touched=["a"], branch="") + c1 = EvidenceCard(diff=d1, base_version={"a": 1}, touched=["a"], branch="head/1") + buf = EvidenceBuffer() + buf.add(c0) + buf.add(c1) + # ready() reports the artifact once; drain() groups by branch + assert buf.ready(AggregatorConfig(batch_trigger=1)) == ["a"] + grouped = buf.drain("a") + assert grouped == {"": [c0], "head/1": [c1]}, f"expected 2 buckets, got {grouped}" + + +def test_prepare_merges_against_own_branch(tmp_path): + """A head/1 diff is merged against head/1's version vector, not dev's.""" + from agentdescent.aggregator import Aggregator, AggregatorConfig, _Candidate + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + lg = _ledger(tmp_path) + # dev at v1, head/0 at v5 + lg.fork("head/0") + snap = lg.snapshot("head/0") + head = snap.get("a") + for i in range(4): + cand = head.apply(Diff(diff_id=f"h{i}", target="a", ops={f"k{i}": "v"}, author="w")) + lg.commit(cand, {"a": snap.version["a"]}, branch="head/0") + snap = lg.snapshot("head/0") + head = snap.get("a") + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + agg = Aggregator(lg, v, AuditScheduler(), AggregatorConfig(batch_trigger=1)) + # A card proposed against head/0 at v5 + card = EvidenceCard( + diff=Diff(diff_id="d", target="a", ops={"new": "v"}, author="w"), + base_version={"a": 5}, touched=["a"], branch="head/0") + agg.ingest(card) + prepared = agg._prepare("a", "head/0") + assert isinstance(prepared, _Candidate), f"expected candidate, got {prepared}" + assert prepared.branch == "head/0" + assert prepared.head["a"] == 5, "must merge against head/0's version, not dev's" + + +def test_multi_head_full_loop_forks_and_admits(tmp_path): + """End-to-end: workers commit -> archive accumulates -> forks created -> + workers distributed across dev + forks.""" + from agentdescent.aggregator import AggregatorConfig, EvidenceCard + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + from agentdescent.population import PopulationAggregator + from agentdescent.selection import Candidate, SelectionContext + + class Beam2: + def select(self, ctx, n): + return sorted(ctx.candidates, key=lambda c: c.score or 0, + reverse=True)[:n] + + lg = _ledger(tmp_path) + # Score by the numeric suffix of the state's "k" value, so each proposal + # (v1, v2, ...) strictly improves on the seed (v0) and the merge commits it. + v = ThreeLayerVerifier( + eval_fn=lambda a, t: int(a.state.get("k", "v0").replace("v", "")) / 10.0, + held_out=[1, 2, 3], budget=VerifierBudget()) + agg = PopulationAggregator(lg, v, AuditScheduler(), + AggregatorConfig(batch_trigger=1), + selection=Beam2(), artifact_id="a") + # round 1: seed + agg.step(n_workers=3) + assert agg._forks == [] + # round 2: two workers propose to dev + base = lg.snapshot("dev").get("a") + for i, val in enumerate(["v1", "v2"]): + cand = base.apply(Diff(diff_id=f"c{i}", target="a", ops={"k": val}, author="w")) + agg.ingest(EvidenceCard( + diff=Diff(diff_id=f"c{i}", target="a", ops={"k": val}, author="w"), + base_version={"a": lg.head_version("dev").get("a")}, + touched=["a"], trajectory_refs=[], branch="")) + base = cand + reports = agg.step(n_workers=3) + # archive accumulated the proposals (dedup by rendered state, so the + # later merge overwrote the earlier same-key state) + states = [dict(e["state"]).get("k") for e in agg._archive] + assert len(states) >= 2, f"archive must accumulate beyond seed, got {states}" + assert len(agg._forks) >= 1, f"Beam2 with 2+ candidates should fork, got {agg._forks}" + # head_for_worker distributes: worker0=dev, worker1=head/0... + assert agg.head_for_worker(0) == "dev" + if agg._forks: + assert agg.head_for_worker(1) == agg._forks[0] + + +def test_step_compat_legacy_zero_arg_aggregator(tmp_path): + """A custom aggregator implementing the published zero-argument step() + must still run under evolve() (PR #196 review issue 1).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = _ledger(tmp_path) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + + class LegacyAggregator(Aggregator): + """A pre-multi-head aggregator: zero-arg step(), no head_for_worker.""" + + def step(self): # noqa: D102 - the legacy contract + return super().step(1) + + agg = LegacyAggregator(lg, v, AuditScheduler(), AggregatorConfig(batch_trigger=1)) + from agentdescent.evolution import _head_for_worker, _step + # head_for_worker absent -> dev + assert _head_for_worker(agg, 0) == "dev" + # step() with n_workers must not raise TypeError + reports = _step(agg, 4) + assert isinstance(reports, list) From fd6c9d2d5a8ad684c357a7b6e58f36dd88c9b6a3 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:00:09 +0800 Subject: [PATCH 2/4] fix(multi-head): commit a fork candidate to its own branch, not dev _Candidate.branch was recorded during preparation but the commit call in _decide() defaulted to dev, so fork work overwrote the confirmed branch and left the fork unchanged. _decide now passes the candidate branch through to _commit_with_retry, and the CAS-rebase path rebases against that same branch (the version vectors differ per fork, so rebasing onto dev would be a different world). Test: test_decide_commits_a_fork_candidate_to_its_own_branch proves a head/0 candidate lands on head/0 and leaves dev at v1. --- CHANGELOG.md | 5 ++++- agentdescent/aggregator.py | 14 +++++++++++--- tests/test_multi_head.py | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a8418c2..c3280fb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4413,5 +4413,8 @@ First public release on PyPI as **`agentdescent`**. 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`. - `finalize` discards all forks and commits the best candidate to `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. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 5d1dd19e..f872020e 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -1260,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 @@ -1484,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: diff --git a/tests/test_multi_head.py b/tests/test_multi_head.py index 0e3f85c3..bab3d118 100644 --- a/tests/test_multi_head.py +++ b/tests/test_multi_head.py @@ -192,6 +192,40 @@ def test_prepare_merges_against_own_branch(tmp_path): assert prepared.head["a"] == 5, "must merge against head/0's version, not dev's" +def test_decide_commits_a_fork_candidate_to_its_own_branch(tmp_path): + """A candidate prepared against `head/0` must commit to `head/0`, leaving + `dev` untouched (PR #196 review: `_Candidate.branch` was recorded but the + commit call defaulted to `dev`, so fork work overwrote the confirmed + branch).""" + from agentdescent.aggregator import Aggregator, AggregatorConfig, _Candidate + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + lg = _ledger(tmp_path) + lg.fork("head/0") + # dev at v1; fork at v1 too (fresh fork copies dev) + snap = lg.snapshot("head/0") + head = snap.get("a") + cand = head.apply(Diff(diff_id="d", target="a", ops={"k": "fork-edited"}, author="w")) + # A candidate as `_prepare` would build it for head/0. + c = _Candidate( + artifact_id="a", artifact=head, candidate=cand, + diff=Diff(diff_id="d", target="a", ops={"k": "fork-edited"}, author="w"), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + branch="head/0", + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + v = ThreeLayerVerifier(eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()) + agg = Aggregator(lg, v, AuditScheduler(), AggregatorConfig(batch_trigger=1)) + report = agg._decide(c) + assert report.committed_version is not None, "the fork candidate commits" + assert lg.head_version("dev").get("a") == 1, ( + "dev must be untouched by a fork commit") + assert lg.head_version("head/0").get("a") == 2, ( + "the fork branch must receive the commit") + + def test_multi_head_full_loop_forks_and_admits(tmp_path): """End-to-end: workers commit -> archive accumulates -> forks created -> workers distributed across dev + forks.""" From d0b72b4232b542f0cbbc505f7c18181c60c4023e Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:00:13 +0800 Subject: [PATCH 3/4] test(multi-head): drive async_evolve through forks end-to-end Review: the async worker picked its branch once at startup, before forks existed, and never revisited it. The fix re-selects the branch on every snapshot refresh; this test proves the runtime actually reaches the advertised parallel heads. A beam selection forks when the archive holds several candidates, so an async run with it must create forks during the run (finalize() discards them, so the aggregator records the deepest it reached) and hand workers out across dev + the forks. --- tests/test_multi_head.py | 57 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/tests/test_multi_head.py b/tests/test_multi_head.py index bab3d118..50c48e91 100644 --- a/tests/test_multi_head.py +++ b/tests/test_multi_head.py @@ -273,6 +273,63 @@ def select(self, ctx, n): assert agg.head_for_worker(1) == agg._forks[0] +def test_async_evolve_explores_forks_end_to_end(tmp_path): + """The async runtime must reach the advertised parallel heads, not keep + every worker on `dev`. A beam selection forks when the archive holds + several candidates; driving `async_evolve` through that selection must + create the forks and hand workers out across dev + the forks (PR #196 + review issue 2: the worker picked its branch once at startup, before forks + existed, and never revisited it).""" + import threading + import warnings + from agentdescent.async_evolve import async_evolve + from agentdescent.evolution import Policies + from agentdescent.population import PopulationAggregator + from agentdescent.selection import Beam + + # Every task carries one of a few clues; a proposal adds its clue, so the + # archive accumulates several candidates (held-out shares the clues) and + # the beam keeps the top few -- the shape that provably forks. The forks + # are discarded by `finalize()`, so the aggregator records the deepest it + # reached during the run. + def run(rendered, task): + return "yes" if task.meta["h"] in rendered else "no" + + def propose(rendered, task, output, score): + return task.meta["h"] if task.meta["h"] not in rendered else None + + tasks = [Task(id=f"t{i}", prompt="q", meta={"h": f"H{i % 5}"}) + for i in range(20)] + reward = lambda t, o: 1.0 if o == "yes" else 0.0 + + deepest = {"forks": 0} + lock = threading.Lock() + + class Tracking(PopulationAggregator): + def step(self, n_workers=1): + reports = super().step(n_workers) + with lock: + deepest["forks"] = max(deepest["forks"], len(self._forks)) + return reports + + def factory(ledger, verifier, audit, config, policy): + return Tracking(ledger, verifier, audit, config, + staleness_policy=policy, selection=Beam(2), + artifact_id="artifact") + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + result = async_evolve( + tasks, reward, run=run, propose=propose, strategy=AppendRules(), + n_workers=4, async_ratio=0, max_seconds=8.0, held_out_frac=0.4, + aggregator_factory=factory) + assert result.error is None + assert result.final_reward > 0.0, ( + "premise: the run must learn some clues before it can fork") + assert deepest["forks"] >= 1, ( + "async_evolve must create forks, not keep every worker on dev") + + def test_step_compat_legacy_zero_arg_aggregator(tmp_path): """A custom aggregator implementing the published zero-argument step() must still run under evolve() (PR #196 review issue 1).""" From 3611d68fb66a217a4c039acf9e00d946f25d4b80 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:00:16 +0800 Subject: [PATCH 4/4] test(multi-head): assert workers are routed to a fork, with a forced-dev control The end-to-end async test only proved forks were created, not that any worker snapshotted or proposed against one. It now wraps the exact seam the async worker reads before snapshotting (_head_for_worker) and asserts at least one worker is handed a non-dev branch. A negative control forces that seam back to dev and proves the assertion fails -- so the test discriminates a real fork route from fork creation alone. --- tests/test_multi_head.py | 91 ++++++++++++++++++++++++++++++---------- 1 file changed, 69 insertions(+), 22 deletions(-) diff --git a/tests/test_multi_head.py b/tests/test_multi_head.py index 50c48e91..9820e8c8 100644 --- a/tests/test_multi_head.py +++ b/tests/test_multi_head.py @@ -273,25 +273,27 @@ def select(self, ctx, n): assert agg.head_for_worker(1) == agg._forks[0] -def test_async_evolve_explores_forks_end_to_end(tmp_path): - """The async runtime must reach the advertised parallel heads, not keep - every worker on `dev`. A beam selection forks when the archive holds - several candidates; driving `async_evolve` through that selection must - create the forks and hand workers out across dev + the forks (PR #196 - review issue 2: the worker picked its branch once at startup, before forks - existed, and never revisited it).""" +def _run_async_fork_probe(*, forced_dev=False): + """Drive `async_evolve` through a `Beam(2)` selection and report which + branches the workers actually selected. + + Every task carries one of a few clues; a proposal adds its clue, so the + archive accumulates several candidates (held-out shares the clues) and the + beam keeps the top few -- the shape that provably forks. The fork branches + are discarded by `finalize()`, so the probe observes the seam directly: it + wraps `async_evolve._head_for_worker` (the only place a worker learns its + branch) and records every branch handed out, plus how deep the forks went. + ``forced_dev`` replaces that seam with a constant ``dev`` to prove the + assertions below actually discriminate.""" + import importlib import threading import warnings - from agentdescent.async_evolve import async_evolve - from agentdescent.evolution import Policies + from agentdescent.ledger import Ledger from agentdescent.population import PopulationAggregator from agentdescent.selection import Beam - # Every task carries one of a few clues; a proposal adds its clue, so the - # archive accumulates several candidates (held-out shares the clues) and - # the beam keeps the top few -- the shape that provably forks. The forks - # are discarded by `finalize()`, so the aggregator records the deepest it - # reached during the run. + ae = importlib.import_module("agentdescent.async_evolve") + def run(rendered, task): return "yes" if task.meta["h"] in rendered else "no" @@ -302,8 +304,17 @@ def propose(rendered, task, output, score): for i in range(20)] reward = lambda t, o: 1.0 if o == "yes" else 0.0 - deepest = {"forks": 0} + picked = [] # branch every worker chose, in call order lock = threading.Lock() + deepest = {"forks": 0} + + real = ae._head_for_worker + + def recording(aggregator, worker_id): + branch = real(aggregator, worker_id) if not forced_dev else Ledger.DEV + with lock: + picked.append(branch) + return branch class Tracking(PopulationAggregator): def step(self, n_workers=1): @@ -317,17 +328,53 @@ def factory(ledger, verifier, audit, config, policy): staleness_policy=policy, selection=Beam(2), artifact_id="artifact") - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - result = async_evolve( - tasks, reward, run=run, propose=propose, strategy=AppendRules(), - n_workers=4, async_ratio=0, max_seconds=8.0, held_out_frac=0.4, - aggregator_factory=factory) + ae._head_for_worker = recording + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + result = ae.async_evolve( + tasks, reward, run=run, propose=propose, strategy=AppendRules(), + n_workers=4, async_ratio=0, max_seconds=8.0, held_out_frac=0.4, + aggregator_factory=factory) + finally: + ae._head_for_worker = real + return result, picked, deepest + + +def test_async_evolve_workers_snapshot_fork_branches(tmp_path): + """A worker must snapshot/propose against a fork, not just against `dev`. + + The branch-routing fix (PR #196 review issue 2) re-selects each worker's + branch on every snapshot refresh. This asserts the *route*: a fork-holding + run must hand at least one worker a non-dev branch through + `_head_for_worker`, the exact seam the async worker reads before it + snapshots and submits its evidence.""" + result, picked, deepest = _run_async_fork_probe() assert result.error is None assert result.final_reward > 0.0, ( "premise: the run must learn some clues before it can fork") assert deepest["forks"] >= 1, ( - "async_evolve must create forks, not keep every worker on dev") + "premise: the beam must fork during the run") + forks_picked = [b for b in picked if b != "dev"] + assert forks_picked, ( + "no worker was ever routed to a fork; every `_head_for_worker` call " + f"returned dev: {sorted(set(picked))}") + + +def test_async_evolve_workers_routed_to_dev_by_force_fails_the_assertion(tmp_path): + """Negative control: if the branch seam is forced back to `dev` the + fork-route assertion above must fail -- proving it discriminates a real + fork snapshot from fork creation alone.""" + import pytest + result, picked, deepest = _run_async_fork_probe(forced_dev=True) + assert result.error is None + assert deepest["forks"] >= 1, ( + "premise: the forks still get created; only the worker route is forced") + assert picked and all(b == "dev" for b in picked), ( + "premise: forced_dev must keep every worker on dev") + with pytest.raises(AssertionError): + assert [b for b in picked if b != "dev"], \ + "async_evolve must route a worker to a fork" def test_step_compat_legacy_zero_arg_aggregator(tmp_path):