diff --git a/CHANGELOG.md b/CHANGELOG.md index 70e9e4be..c3280fb7 100644 --- a/CHANGELOG.md +++ b/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/` 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. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 2ac7b3f3..f872020e 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: @@ -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,9 +1299,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 +1320,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 +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: 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..9820e8c8 --- /dev/null +++ b/tests/test_multi_head.py @@ -0,0 +1,403 @@ +"""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_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 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 _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.ledger import Ledger + from agentdescent.population import PopulationAggregator + from agentdescent.selection import Beam + + ae = importlib.import_module("agentdescent.async_evolve") + + 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 + + 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): + 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") + + 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, ( + "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): + """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)