diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 87b08561..ad644921 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -38,8 +38,8 @@ from collections import defaultdict, deque from dataclasses import dataclass, field from enum import Enum -from typing import (Any, Callable, Deque, Dict, List, Optional, Protocol, Tuple, - Union, runtime_checkable) +from typing import (Any, Callable, Deque, Dict, List, Optional, Protocol, Sequence, + Tuple, Union, runtime_checkable) from .evolvable import ( ContractError, Diff, EvidenceCard, Evolvable, VersionVector, @@ -229,6 +229,13 @@ class MergeOutcome(str, Enum): #: so a policy-violating proposal never buys a held-out sweep. The #: reflector (or the contract) is the thing to look at. VIOLATES_INVARIANT = "violates-invariant" + #: A contract-breaking diff landed on an artifact with declared dependents + #: (``Contract.depends_on``), but the worker supplied no adapted dependent + #: states to land in the same ``commit_atomic``. The atomic adaptation + #: transaction would otherwise silently commit only the breaking artifact, + #: leaving dependents registered against a superseded contract. Explicitly + #: rejected rather than half-applied. + MISSING_ADAPTERS = "missing-adapters" #: A pre-evaluation screen (see #: :class:`~agentdescent.policies.ScreenPolicy`) rejected the card before #: it was measured -- leakage, an over-budget edit bundle, inert machinery. @@ -358,6 +365,11 @@ class _Candidate: oversized: int = 0 #: The ledger branch this candidate was proposed against. branch: Optional[str] = None + #: Adapted states of the dependents that must land in the *same* atomic + #: commit as a contract-breaking change (the design's atomic adaptation + #: transaction). ``None``/empty when the diff is not contract-breaking, or + #: when the artifact has no declared dependents. + adapters: Optional[Sequence[Evolvable]] = 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. @@ -719,12 +731,26 @@ def __init__( self.acceptance_policy, self.promotion_policy): install_policy(policy, verifier, cfg) self.buffer = EvidenceBuffer() - #: Optional value allocator (see :mod:`agentdescent.allocator`), wired - #: by the engine so the merge path can feed commit outcomes back into - #: it. ``None`` keeps the old behaviour; the aggregator never consults - #: the allocator for a decision (the worker does, through the budget - #: governor), it only reports what happened. + #: Optional value allocator (see :mod:`agentdescent/allocator.py`), + #: wired by the engine so the merge path can feed commit outcomes back + #: into it. ``None`` keeps the old behaviour; the aggregator never + #: consults the allocator for a decision (the worker does, through the + #: budget governor), it only reports what happened. self.allocator: Optional[Any] = None + #: Called after a **contract-breaking** commit lands, with the changed + #: artifact id and the branch it committed on. The engine wires it to + #: re-measure artifacts whose ``Contract.depends_on`` names the changed + #: artifact (their cached evaluations were taken under the superseded + #: contract). Optional -- ``None`` keeps the old behaviour of not + #: noticing. + self.on_contract_change: Optional[Callable[[str, str], None]] = None + #: Reverse contract-dependency graph: ``changed_artifact_id -> + #: [dependents]``, wired by the engine from the registered artifacts' + #: ``Contract.depends_on``. Used to refuse a contract-breaking commit + #: that would land without adapted dependents (see + #: :data:`MergeOutcome.MISSING_ADAPTERS`). Empty (single-artifact runs) + #: keeps the old behaviour. + self.contract_dependents: Dict[str, List[str]] = {} self._posteriors: Dict[str, BetaPosterior] = defaultdict(BetaPosterior) #: How many queued audits the L-value consumer drained and ran against #: the oracle. Zero when ``audit_drain_per_step`` is 0 (the default). @@ -1284,7 +1310,7 @@ def _audit(self, artifact, artifact_id, best_state, best_diff, return False def _commit_with_retry(self, artifact_id, candidate, diff, head, - branch: str = Ledger.DEV): + branch: str = Ledger.DEV, adapters=None): """CAS, rebasing onto whatever landed first. ``None`` when it kept losing. With one writer a conflict cannot happen: every commit goes through a @@ -1305,11 +1331,35 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head, measurement is what the retry is trying to preserve. A diff whose value depends on which of two commits landed first is a diff the staleness policy should have caught. + + A **contract-breaking** diff is committed atomically: ``commit`` refuses + a contract change via ``_assert_contract`` (it has to be re-registered + deliberately, not merged), while ``commit_atomic`` skips that guard -- + which is the point of the atomic path (design doc section 6, the "atomic + adaptation transaction"). So the sanctioned route for a breaking change + is exactly the one the ledger offers but no engine path used. ``adapters`` + are the dependent artifacts that must land in the same git commit (their + states already adapted to the new contract); ``None`` is a single-artifact + breaking change, which is still atomic by being one commit. """ attempts = max(1, self.config.cas_attempts) for attempt in range(attempts): - base_vv = {artifact_id: head.get(artifact_id, 0)} + if diff.contract_breaking: + # `commit_atomic` CAS-checks *every* state it writes, so the + # base vector must name each one (candidate + adapters) with the + # version it was measured against. `head` is the merge's full + # version vector (from the branch snapshot), so every id is + # there; a missing entry means "new artifact, base 0". + new_states = [candidate] + list(adapters or ()) + base_vv = {s.id: head.get(s.id, 0) for s in new_states} + else: + base_vv = {artifact_id: head.get(artifact_id, 0)} try: + if diff.contract_breaking: + _, vv = self.ledger.commit_atomic( + new_states, base_vv, branch=branch, + message=f"merge {diff.diff_id} -> {artifact_id} (contract)") + return vv[artifact_id] _, new_version = self.ledger.commit( candidate, base_vv, branch=branch, message=merge_commit_message(diff, artifact_id)) @@ -1334,6 +1384,16 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head, # and reporting a conflict would be truer than reporting a # commit that did not happen. return None + # The primary rebases via `fresh.apply(diff)`, but an adapter is + # a *final state*, not a diff -- it cannot be rebased here, only + # re-validated. If a concurrent commit advanced one of the + # artifacts an adapter writes, committing the old adapter state + # as its next version would delete whatever that commit added. + # Reject the stale transaction; it re-files and is re-prepared + # by the worker. + for ad in (adapters or ()): + if snap.version.get(ad.id, 0) != ad.version - 1: + return None return None def _known_artifacts(self): @@ -1623,11 +1683,44 @@ def _observe_outcome(committed: bool) -> None: # -- commit (section 4.1): CAS on dev -------------------------------- # 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. + # would overwrite the confirmed branch and leave the fork unchanged. + # Empty branch means `dev`, the single-head default. + # A contract-breaking change must land **with** its adapted dependents + # in the same atomic commit -- the design's "atomic adaptation + # transaction". The worker produces the breaking diff; the adapted + # dependent states are what make it safe. The adapters must cover + # *exactly* the declared dependents: every dependent that relies on + # this artifact's contract has to land adapted in the same commit, and + # a stray adapter for an unrelated artifact would leave a real + # dependent registered against the superseded contract while pretending + # the transaction is complete. The guard runs whenever *either* side is + # nonempty: an artifact with no declared dependents must not carry + # adapters either -- a breaking change to it has nothing to land with, + # and committing an unrelated adapter would silently change an artifact + # no gate asked it to touch. + required = set(self.contract_dependents.get(c.artifact_id, ())) + if c.diff.contract_breaking: + adapters = list(getattr(c, "adapters", None) or ()) + adapter_ids = [a.id for a in adapters] + missing = sorted(required - set(adapter_ids)) + extra = sorted(set(adapter_ids) - required) + duplicated = len(adapter_ids) != len(set(adapter_ids)) + if missing or extra or duplicated: + _observe_outcome(False) + return report(None, None, + f"contract-breaking change to {c.artifact_id} has " + f"declared dependents {sorted(required)}; its " + f"adapters (ids {sorted(set(adapter_ids))}) must " + f"cover exactly those, but they are missing " + f"{missing}, extra {extra} and " + f"duplicated={duplicated}. Refusing rather than " + f"half-apply: every dependent must land adapted " + f"in the same atomic commit.", + MergeOutcome.MISSING_ADAPTERS) new_version = self._commit_with_retry( c.artifact_id, c.candidate, c.diff, c.head, - branch=c.branch or Ledger.DEV) + branch=c.branch or Ledger.DEV, + adapters=getattr(c, "adapters", None)) committed = new_version is not None _observe_outcome(committed) if not committed: @@ -1636,4 +1729,7 @@ def _observe_outcome(committed: bool) -> None: prior.update(True, weight=2.0) # a committed improvement is strong evidence. + if c.diff.contract_breaking and self.on_contract_change is not None: + self.on_contract_change(c.artifact_id, c.branch or Ledger.DEV) + return report(c.diff, new_version, "committed", MergeOutcome.COMMITTED) diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py index dce16c2b..1d5919bb 100644 --- a/agentdescent/async_evolve.py +++ b/agentdescent/async_evolve.py @@ -166,6 +166,12 @@ def async_evolve( #: the reference skill contract, which declares nothing and enforces #: nothing. contract: Optional[Contract] = None, + #: Additional artifacts to register alongside the primary. Each value is an + #: :class:`EvolvingArtifact` with its own ``id``, ``blast_radius`` and + #: ``strategy``. The worker loop proposes against each of them too, and the + #: aggregator buckets their diffs by target. Empty (the default) is the old + #: single-artifact behaviour exactly. + extra_artifacts: Optional[Dict[str, "EvolvingArtifact"]] = None, ) -> EvolutionResult: """Evolve an artifact **without a round barrier**. @@ -421,6 +427,13 @@ def async_evolve( contract, where governance invariants are declared. ``None`` (the default) is the reference skill contract, which declares nothing and enforces nothing. + extra_artifacts: + Exactly as in :func:`~agentdescent.evolution.evolve`: additional + artifacts to register alongside the primary, as ``{aid: EvolvingArtifact}``. + The barrier-free worker proposes against each of them too (no parallel + strategy, so it is the primary plus every extra), and the aggregator + merges their diffs by target. Empty (the default) is the old + single-artifact behaviour exactly. Returns ------- @@ -467,7 +480,8 @@ def async_evolve( usage=usage, verifier=_pol.verifier, ledger_impl=_pol.ledger, policies_bundle=_pol, checkpointing=checkpointing, allocator=allocator, record_trajectories=record_trajectories, - trajectory_store=trajectory_store, contract=contract) + trajectory_store=trajectory_store, contract=contract, + extra_artifacts=extra_artifacts) eng.meter.start() # The cost-aware governor for the async path. The merger loop has no round # barrier, so the governor checks per sweep (after each merge, before the @@ -756,85 +770,98 @@ def _worker(wid: int, shard: List[Task]) -> None: predicted = estimator.estimate(cost) if estimator is not None else 0.0 t_start = time.time() try: - # Render once: the same string feeds the provider, the sampler - # and the trajectory record, so all three agree by construction. - rendered = artifact.render() - output = eng.run(rendered, task) - score = _checked_reward(eng.reward(task, output), task) - eng.meter.add("rollouts") - eng.meter.add("rollout_seconds", time.time() - t_start) - # Record the rollout before anything downstream decides it does - # not matter: the trajectory is the run's evidence. None when - # recording is off / deduped / the write failed -- that empty is - # what the card's trace_ids carry. - trace = eng._record_rollout( - artifact_id=eng.artifact_id, rendered=rendered, - task=task, base_version={eng.artifact_id: base_v}, - purpose="rollout", output=output, reward=score, - wallclock=time.time() - t_start, worker_id=f"w{wid}") - sampler.record(task.id, score) # learn which tasks carry signal - # Before the solved-task branch, for the same reason as on the - # synchronous path: a group that only saw the failures has no - # variance to standardise against. - adv = advantage.observe( - advantage.key(base_v, str(task.meta.get("cluster", ""))), score) - if score < solved_threshold: - # o1-style test-time scaling: see evolve() for the reasoning. - if call_budget is not None: - call_budget.allocate(score, governor.remaining_fraction()) - proposal = _checked_proposal( - eng.propose(artifact.render(), task, output, score), task) - if proposal: - diff = eng.strategy.to_diff(artifact.state, proposal, - f"w{wid}", base_v, eng.artifact_id) - if diff is not None: - # Optional local self-verify: re-run the trajectory with the - # diff applied for a before/after signal. Faithful repos that - # only score the candidate on held-out (e.g. EvoSkill) pass - # self_verify=False to skip this extra rollout. - # With an allocator installed the decision becomes - # per-candidate once the budget is tight (see - # :mod:`agentdescent.allocator`). - if self_verify and governor.candidate_worth( - _async_allocator_context( - diff, adv, - getattr(artifact, "blast_radius", 0.2))): - t_self = time.time() - # The candidate's render is computed once for - # both the rollout and the record, so the - # trajectory and the measurement agree. - rendered_after = artifact.apply(diff).render() - after_output = eng.run(rendered_after, task) - after = _checked_reward( - eng.reward(task, after_output), task) - delta = after - score - self_trace = eng._record_rollout( - artifact_id=eng.artifact_id, - rendered=rendered_after, - task=task, base_version={eng.artifact_id: base_v}, - purpose="self_verify", output=after_output, - reward=after, wallclock=time.time() - t_self, - worker_id=f"w{wid}") - else: - delta = 0.0 - self_trace = None - card = EvidenceCard( - diff=diff, base_version={eng.artifact_id: base_v}, - touched=[eng.artifact_id], before_after_delta=delta, - trajectory_refs=[task], - branch=worker_branch, - trace_ids=tuple( - t for t in (trace, self_trace) - if t is not None), - # Same signal as the synchronous path, and the - # reason `GroupAdvantage` accumulates rather than - # batching at a barrier: there is no barrier here, - # so a batched version would silently record - # nothing on exactly the runtime the project is - # making claims about. - advantage=adv) - with intake_lock: - intake.append(card) + # Which artifacts this worker proposes for. PP stages are not a + # thing on the barrier-free path (no parallel strategy), so it is + # the primary plus every registered extra -- one rollout each, + # the same cross-product the synchronous worker makes. + targets = [(eng.artifact_id, artifact, base_v)] + [ + (aid, snap.get(aid), snap.version.get(aid, 0)) + for aid in eng.extra_artifact_ids] + for aid, art, art_v in targets: + if art is None: + continue + # Render once: the same string feeds the provider, the + # sampler and the trajectory record, so all three agree by + # construction. + rendered = art.render() + output = eng.run(rendered, task) + score = _checked_reward(eng.reward(task, output), task) + eng.meter.add("rollouts") + eng.meter.add("rollout_seconds", time.time() - t_start) + # Record the rollout before anything downstream decides it + # does not matter: the trajectory is the run's evidence. + # None when recording is off / deduped / the write failed -- + # that empty is what the card's trace_ids carry. + trace = eng._record_rollout( + artifact_id=aid, rendered=rendered, + task=task, base_version={aid: art_v}, + purpose="rollout", output=output, reward=score, + wallclock=time.time() - t_start, worker_id=f"w{wid}") + sampler.record(task.id, score) # learn which tasks carry signal + # Before the solved-task branch, for the same reason as on the + # synchronous path: a group that only saw the failures has no + # variance to standardise against. + adv = advantage.observe( + advantage.key(art_v, str(task.meta.get("cluster", ""))), score) + if score < solved_threshold: + # o1-style test-time scaling: see evolve() for the reasoning. + if call_budget is not None: + call_budget.allocate(score, governor.remaining_fraction()) + proposal = _checked_proposal( + eng.propose(art.render(), task, output, score), task) + if proposal: + strat = eng.artifact_strategies.get(aid, eng.strategy) + diff = strat.to_diff(art.state, proposal, + f"w{wid}", art_v, aid) + if diff is not None: + # Optional local self-verify: re-run the trajectory with the + # diff applied for a before/after signal. Faithful repos that + # only score the candidate on held-out (e.g. EvoSkill) pass + # self_verify=False to skip this extra rollout. + # With an allocator installed the decision becomes + # per-candidate once the budget is tight (see + # :mod:`agentdescent.allocator`). + if self_verify and governor.candidate_worth( + _async_allocator_context( + diff, adv, + getattr(art, "blast_radius", 0.2))): + t_self = time.time() + # The candidate's render is computed once + # for both the rollout and the record, so + # the trajectory and the measurement agree. + rendered_after = art.apply(diff).render() + after_output = eng.run(rendered_after, task) + after = _checked_reward( + eng.reward(task, after_output), task) + delta = after - score + self_trace = eng._record_rollout( + artifact_id=aid, + rendered=rendered_after, + task=task, base_version={aid: art_v}, + purpose="self_verify", + output=after_output, reward=after, + wallclock=time.time() - t_self, + worker_id=f"w{wid}") + else: + delta = 0.0 + self_trace = None + card = EvidenceCard( + diff=diff, base_version={aid: art_v}, + touched=[aid], before_after_delta=delta, + trajectory_refs=[task], + branch=worker_branch, + trace_ids=tuple( + t for t in (trace, self_trace) + if t is not None), + # Same signal as the synchronous path, and the + # reason `GroupAdvantage` accumulates rather than + # batching at a barrier: there is no barrier here, + # so a batched version would silently record + # nothing on exactly the runtime the project is + # making claims about. + advantage=adv) + with intake_lock: + intake.append(card) except ContractError as e: # a caller-contract violation, not a flaky backend: stop at once. with counter_lock: @@ -970,7 +997,12 @@ def _merge_batch(batch: List[EvidenceCard]) -> None: # and the collect-only sweeps above would otherwise take one every 5ms. if batch: snap = eng.ledger.snapshot(Ledger.DEV) - head_vv, head_art = snap.version, snap.get(eng.artifact_id) + head_vv = snap.version + # Each card targets its own artifact; the rebase branch re-applies + # against *that* artifact's head, not the primary's (they can differ + # in a multi-artifact run). + def _head_art(card): + return snap.get(card.diff.target) def _discarded() -> None: """Record a card this gate is dropping, denominator included. @@ -997,6 +1029,10 @@ def _discarded() -> None: if action is StaleAction.ACCEPT: eng.aggregator.ingest(card if eta == 0 else card.rebased_onto(head_vv)) elif action is StaleAction.REBASE: + head_art = _head_art(card) + if head_art is None: + _discarded() + continue cand = head_art.apply(card.diff) # cheap re-verify on current head # Gate work, and easy to forget it is: `evidence_eval` runs the # agent on the card's trajectories, so the "cheap" re-verify is diff --git a/agentdescent/evalcache.py b/agentdescent/evalcache.py index e705b442..f4c775c9 100644 --- a/agentdescent/evalcache.py +++ b/agentdescent/evalcache.py @@ -136,6 +136,19 @@ def _count(self, counter: str) -> None: if self._meter is not None: self._meter.add(counter) + def invalidate(self, rendered: str, fingerprint: str = "") -> None: + """Drop every cached evaluation whose rendered prompt is ``rendered``. + + Coarse on purpose: the key is ``(rendered, task_id, fingerprint)``, so + all tasks measured on this render are invalidated at once. Used by the + engine when an artifact's dependencies' contracts change -- the cached + scores were taken under the old contract and must be re-measured. + """ + with self._lock: + for key in [k for k in self._values if k[0] == rendered + and (not fingerprint or k[2] == fingerprint)]: + self._values.pop(key, None) + class FileCache: """A directory of evaluations, so separate processes can share them. @@ -157,6 +170,32 @@ def __init__(self, directory: str, meter: Optional[Any] = None) -> None: def attach_meter(self, meter: Any) -> None: self._memory.attach_meter(meter) + def invalidate(self, rendered: str, fingerprint: str = "") -> None: + """Drop evaluations cached for ``rendered`` in this process. + + File entries are keyed by content, so a re-evaluation would read the + stale file back from disk; delete them too, so the re-measure is real.""" + self._memory.invalidate(rendered, fingerprint) + for name in os.listdir(self.directory): + path = os.path.join(self.directory, name) + if not os.path.isfile(path): + continue + try: + with open(path, encoding="utf-8") as fh: + payload = json.load(fh) + key = payload.get("key") + # Keys are (rendered, task_id, fingerprint); invalidate every + # task measured on this render, optionally narrowed by the + # fingerprint -- the same match MemoryCache uses. Prefix + # comparison against [rendered, fingerprint] would compare the + # *task id* to the fingerprint and delete nothing (review). + if (key and isinstance(key, list) + and key[0] == rendered + and (not fingerprint or key[2] == fingerprint)): + os.remove(path) + except (OSError, ValueError, KeyError, TypeError, IndexError): + continue + def _path(self, key: Any) -> str: import hashlib diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index 2bedfe6c..405c6468 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -770,22 +770,39 @@ def _resolve_sections(parallel, strategy) -> Dict[str, int]: return assign_key_sections(keys, n_sections) -def _reject_pipeline_parallel(parallel) -> None: - """PP is a multi-artifact paradigm; ``evolve()`` evolves exactly one artifact. +def _reject_pipeline_parallel(parallel, extra_artifacts: Optional[Dict[str, "EvolvingArtifact"]], + artifact_id: str = "artifact") -> None: + """PP is a multi-artifact paradigm; ``evolve()`` evolves one artifact. ``WorkUnit.stage`` -- the only thing distinguishing PP's units, since it hands every worker the whole task list -- was never read by the driver, so passing ``parallel=PipelineParallel(...)`` silently degraded to n_workers all rolling - out the same tasks: strictly worse than the default, with no signal. Say so.""" - if type(parallel).__name__ == "PipelineParallel" or getattr(parallel, "name", "") == "PP": + out the same tasks: strictly worse than the default, with no signal. Say so. + With ``extra_artifacts`` registered the stages have somewhere to live, so the + refusal lifts -- and each stage must name a registered artifact, or the worker + loop would propose against an id the ledger has never heard of.""" + if type(parallel).__name__ != "PipelineParallel" and getattr(parallel, "name", "") != "PP": + return + registered = set(extra_artifacts or {}) | {artifact_id} + stages = getattr(parallel, "stages", None) or () + unknown = [s for s in stages if s not in registered] + if not extra_artifacts: + raise ValueError( + "evolve() cannot run PipelineParallel: it evolves a single " + "artifact_id, while PP needs one artifact per stage. " + "Pass extra_artifacts={...} to register the stages' artifacts. " + "Passing PP without extra_artifacts used to be accepted " + "and quietly ignored (every worker got the whole task list and " + "the stage was never read), which is worse than the " + "DataParallel default. The PP primitives are still usable " + "directly -- see agentdescent.parallel.PipelineChain for stage " + "ordering and upstream blame attribution.") + if unknown: raise ValueError( - "evolve() cannot run PipelineParallel: it evolves a single artifact_id, " - "while PP needs one artifact per stage. Passing it used to be accepted " - "and quietly ignored (every worker got the whole task list and the " - "stage was never read), which is worse than the DataParallel default. " - "The PP primitives are still usable directly -- see " - "agentdescent.parallel.PipelineChain for stage ordering and upstream " - "blame attribution.") + f"PipelineParallel stages must name registered artifacts; unknown " + f"stage(s) {sorted(unknown)}. Registered: " + f"{sorted(registered)}. Each stage is an artifact the worker loop " + "proposes against.") def _safe_log(ledger: Ledger, limit: int = 40) -> List[str]: @@ -1861,6 +1878,49 @@ class _Engine: train_ids: List[str] artifact_id: str blast_radius: float + #: The extra artifacts registered alongside the primary, in declaration + #: order. Empty for a single-artifact run. The worker loop proposes against + #: each of these too, and the aggregator buckets their diffs by target. + extra_artifact_ids: Tuple[str, ...] = () + #: The strategies registered per artifact id -- extras may carry their own + #: (an :class:`EvolvingArtifact` holds one), so ``to_diff`` must use the + #: target's strategy, not the primary's. ``engine.strategy`` remains the + #: primary's, for the callers that already read it. + artifact_strategies: Dict[str, Strategy] = field(default_factory=dict) + #: Reverse contract dependency graph: ``changed_artifact_id -> [dependents]``. + #: When artifact A commits a contract-breaking diff, every B in this list + #: (whose ``Contract.depends_on`` names A) has cached evaluations taken under + #: the old contract, so the engine re-measures them. + contract_dependents: Dict[str, List[str]] = field(default_factory=dict) + + def invalidate_dependents(self, changed_artifact_id: str, + branch: str = Ledger.DEV) -> None: + """Drop cached evaluations of artifacts that depend on ``changed``. + + Called when a contract-breaking diff commits on ``changed_artifact_id`` + (on ``branch`` -- a fork's dependents are measured against the fork's + head, so evicting the *fork's* renders is what forces their re-measure): + every artifact whose ``Contract.depends_on`` names it was measured under + the superseded contract, so its cached scores no longer mean what they + said. Re-measuring is forced by evicting those cache entries -- the + dependent's render may not have changed, which is exactly why a plain + ``(rendered, task)`` cache cannot see the staleness on its own. + + ``None`` when no cache invalidation hook exists (a custom cache that + predates it simply re-measures nothing and stays correct, only slower to + notice the change).""" + dependents = self.contract_dependents.get(changed_artifact_id) or () + if not dependents: + return + snap = self.ledger.snapshot(branch) + invalidate = getattr(self.runtime.cache, "invalidate", None) + if not callable(invalidate): + return + for aid in dependents: + art = snap.get(aid) + if art is None: + continue + invalidate(art.render()) #: Where rollouts run. Always present, defaulting to this process, so the #: round body has one path rather than one per substrate. executor: Any = None @@ -2058,7 +2118,8 @@ def _build_engine(tasks, reward, *, agent, run, propose, strategy, initial_state allocator: Optional[Any] = None, record_trajectories: bool = False, trajectory_store: Optional[Any] = None, - contract: Optional[Contract] = None) -> _Engine: + contract: Optional[Contract] = None, + extra_artifacts: Optional[Dict[str, "EvolvingArtifact"]] = None) -> _Engine: """Wire the ledger, runtime, verifier and aggregator (shared by :func:`evolve` and :func:`~agentdescent.async_evolve.async_evolve`).""" import tempfile @@ -2204,6 +2265,18 @@ def serialize(a: EvolvingArtifact) -> dict: out["contract"] = contract_to_dict(contract) return out + #: Strategies per registered artifact id. Extras carry their own (an + #: :class:`EvolvingArtifact` holds one); the primary uses ``strategy``. A + #: snapshot deserialises each artifact through its OWN strategy, so a diff + #: targeting an extra is rendered/applied by the rules that artifact was + #: registered under -- not the primary's, which is what ``apply`` would + #: otherwise do. + artifact_strategies: Dict[str, Strategy] = {artifact_id: strategy} + for _aid, _art in (extra_artifacts or {}).items(): + if _aid == artifact_id: + continue + artifact_strategies[_aid] = getattr(_art, "_strategy", None) or strategy + def deserialize(aid: str, version: int, state: dict) -> EvolvingArtifact: # The gate and _assert_contract must read the *operator-supplied* # contract, not whatever a previous run serialized. On a fresh run the @@ -2221,7 +2294,8 @@ def deserialize(aid: str, version: int, state: dict) -> EvolvingArtifact: resolved = contract_from_dict(state["contract"]) return EvolvingArtifact(aid, state.get("state", {}), version, state.get("blast_radius", blast_radius), runtime, - strategy, contract=resolved) + artifact_strategies.get(aid, strategy), + contract=resolved) scratch: Optional[str] = None if repo_path: @@ -2300,6 +2374,32 @@ def deserialize(aid: str, version: int, state: dict) -> EvolvingArtifact: raise GovernanceError( f"artifact {artifact_id!r} declares an invalid contract: " + "; ".join(violations)) + # Register any extra artifacts the caller declared. The primary is the one + # the worker loop runs on; extras exist so a proposal can name them as a + # diff target and the aggregator can merge into them. ``commit_atomic`` is + # the path a contract-breaking diff that touches both takes. + for aid, art in (extra_artifacts or {}).items(): + if aid == artifact_id: + continue # the primary, already registered + ledger.register(art) + + # Build the reverse contract-dependency graph from the declared contracts. + # ``Contract.depends_on`` names the artifacts whose contracts a dependent + # relies on; when one of those commits a contract-breaking diff, the + # dependents' cached evaluations are stale (measured under the superseded + # contract), so the engine re-measures them. This is the seam the design + # calls the atomic adaptation transaction's consumer side: the breaking diff + # itself already lands atomically via ``commit_atomic``. + contract_dependents: Dict[str, List[str]] = {} + for aid, art in (extra_artifacts or {}).items(): + if aid == artifact_id: + continue + deps = tuple(getattr(getattr(art, "contract", None), "depends_on", ()) or ()) + for dep in deps: + if dep in contract_dependents: + contract_dependents[dep].append(aid) + else: + contract_dependents[dep] = [aid] # The cheap layer must actually be cheap. It used to be pinned to the full # held-out set (`rule_subset=len(held_out)`) with zero noise, on the reasoning @@ -2506,12 +2606,28 @@ def _default_aggregator(ledger, verifier, audit, config, policy): attach = getattr(executor, "attach_meter", None) if callable(attach): attach(meter) - return _Engine(ledger, runtime, verifier, aggregator, strategy, run, reward, - propose, train, held_out, {t.id: t for t in train}, - [t.id for t in train], artifact_id, blast_radius, - executor=executor, meter=meter, scratch_repo=scratch, - checkpoint_payload=_cp_payload, checkpointing=checkpointing, - checkpoint_round_base=_cp_round_base) + eng = _Engine(ledger, runtime, verifier, aggregator, strategy, run, reward, + propose, train, held_out, {t.id: t for t in train}, + [t.id for t in train], artifact_id, blast_radius, + extra_artifact_ids=tuple( + aid for aid in (extra_artifacts or {}) if aid != artifact_id), + artifact_strategies=artifact_strategies, + contract_dependents=contract_dependents, + executor=executor, meter=meter, scratch_repo=scratch, + checkpoint_payload=_cp_payload, checkpointing=checkpointing, + checkpoint_round_base=_cp_round_base) + # A contract-breaking commit lands atomically (commit_atomic); this is the + # consumer side -- re-measure the artifacts that depended on the changed + # contract. Wired here, after the engine exists, so the callback has + # somewhere to read the ledger from. The reverse dependency graph is also + # handed to the aggregator so it can refuse a breaking commit that has + # declared dependents but no adapted states to land with it. + try: + aggregator.on_contract_change = eng.invalidate_dependents + aggregator.contract_dependents = contract_dependents + except AttributeError: # a custom aggregator without the slot + pass + return eng def evolve( @@ -2643,6 +2759,13 @@ def evolve( #: (the default) is the reference skill contract, which declares nothing and #: enforces nothing -- the untouched path exactly. contract: Optional[Contract] = None, + #: Additional artifacts to register alongside the primary. Each value is an + #: :class:`EvolvingArtifact` with its own ``id``, ``blast_radius`` and + #: ``strategy``. The primary (``artifact_id``) is the one the worker loop + #: runs and proposes on; these are registered so ``propose`` can return a + #: diff whose ``target`` names one of them, and the aggregator merges it. + #: Empty (the default) is the old single-artifact behaviour exactly. + extra_artifacts: Optional[Dict[str, "EvolvingArtifact"]] = None, ) -> EvolutionResult: """Evolve an artifact. Provide either ``agent`` (with ``solve``/``propose``) or the ``run`` / ``propose`` callables directly. @@ -3007,6 +3130,18 @@ def evolve( ``violates-invariant`` outcome and never buys a held-out sweep. ``None`` (the default) is the reference skill contract, which declares nothing and enforces nothing -- the untouched path exactly. + extra_artifacts: + Additional artifacts to register alongside the primary + (``artifact_id``), as ``{aid: EvolvingArtifact}``. The primary is the + one the worker loop rolls out and proposes on; each extra is registered + in the ledger and the worker proposes against it too, so a proposal can + name one of them as a diff target and the aggregator merges it there + (the evidence buffer already buckets by ``diff.target``). ``depends_on`` + on an extra's ``Contract`` declares which artifacts it relies on: a + contract-breaking change to one re-measures its dependents (their cached + scores were taken under the superseded contract) and lands atomically via + ``Ledger.commit_atomic``. Empty (the default) is the old single-artifact + behaviour exactly. Returns ------- @@ -3093,7 +3228,8 @@ def evolve( efficiency_floor=efficiency_floor, call_budget=call_budget, allocator=allocator, record_trajectories=record_trajectories, - trajectory_store=trajectory_store, contract=contract) + trajectory_store=trajectory_store, contract=contract, + extra_artifacts=extra_artifacts) if pipelined_gate: # The mirror of the block above, and the same reasoning: a knob accepted @@ -3134,7 +3270,7 @@ def evolve( # rollout: an incompatible pairing used to be discovered one diff at a time, by # silently discarding it. section_map = _resolve_sections(parallel, strategy) - _reject_pipeline_parallel(parallel) + _reject_pipeline_parallel(parallel, extra_artifacts, artifact_id) # Resolved once: the round body is the hot path, and a strategy either has # the hook for the whole run or does not. observe_plan = getattr(parallel, "observe", None) @@ -3154,7 +3290,8 @@ def evolve( usage=usage, verifier=_pol.verifier, ledger_impl=_pol.ledger, policies_bundle=_pol, checkpointing=checkpointing, allocator=allocator, record_trajectories=record_trajectories, - trajectory_store=trajectory_store, contract=contract) + trajectory_store=trajectory_store, contract=contract, + extra_artifacts=extra_artifacts) # Start the clock after the wiring, before the first unit of work: setup # is not what a time-to-quality number is asking about. eng.meter.start() @@ -3378,13 +3515,35 @@ def _run_unit(unit) -> None: print(f"round {r:>3} worker {unit.worker} failed: " f"{type(e).__name__}: {str(e)[:100]}") - def _run_unit_inner(unit) -> None: - if not unit.keys: + def _artifact_targets(unit) -> List[str]: + """The artifacts this worker proposes for in one unit. + + Under PP each worker owns one stage, and a stage *is* an artifact id + (``PipelineParallel(stages=["skill-a", "skill-b"])`` -- the reason + PP was refused while ``evolve()`` could register only one artifact). + Otherwise the worker rolls out the primary and also proposes for every + registered extra, so one rollout's lesson reaches the whole library. + """ + if getattr(parallel, "name", "") == "PP" or type(parallel).__name__ == "PipelineParallel": + stages = getattr(parallel, "stages", None) + if stages: + return [stages[unit.stage % len(stages)]] + return [artifact_id] + return [artifact_id] + list(eng.extra_artifact_ids) + + def _propose_one(unit, task, aid: str) -> None: + """Roll out artifact ``aid``, propose against it, file its card.""" + if aid == artifact_id: + mine, mine_v, mine_branch = _snapshot_for(unit.worker) + else: + # Extras are read from the round snapshot; they have no per-worker + # staleness mechanism of their own yet (Phase 1), so a worker sees + # the head it was given this round, on `dev`. + mine = snap.get(aid) + mine_v = snap.version.get(aid, 0) + mine_branch = Ledger.DEV + if mine is 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, 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 -- # reads or writes state that has to stay in this process. @@ -3401,8 +3560,8 @@ def _run_unit_inner(unit) -> None: # is off (or the write deduped/failed) -- that empty is what the # card's trace_ids below will carry. trace = eng._record_rollout( - artifact_id=artifact_id, rendered=rendered, task=task, - base_version={artifact_id: mine_v}, purpose="rollout", + artifact_id=aid, rendered=rendered, task=task, + base_version={aid: mine_v}, purpose="rollout", output=output, reward=score, wallclock=outcome.seconds, worker_id=f"w{unit.worker}", lease_id=outcome.lease_id) sampler.record(task.id, score) # learn which tasks carry signal @@ -3414,7 +3573,7 @@ def _run_unit_inner(unit) -> None: # `None`, which is worse than not having it. adv = advantage.observe( advantage.key(mine_v, str(task.meta.get("cluster", ""))), score) - if observe_plan is not None: + if aid == artifact_id and observe_plan is not None: # ...and let the parallel strategy learn too, if it wants to. # `plan` alone is a pure function of its arguments, which is # enough to shard and not enough to schedule: UCB over task @@ -3438,15 +3597,17 @@ def _run_unit_inner(unit) -> None: propose(mine.render(), task, output, score), task) if not proposal: return - diff = strategy.to_diff(mine.state, proposal, f"w{unit.worker}", mine_v, artifact_id) + strat = eng.artifact_strategies.get(aid, strategy) + diff = strat.to_diff(mine.state, proposal, f"w{unit.worker}", mine_v, aid) if diff is None: return # Tensor parallelism means each worker owns a disjoint *section* of the # artifact, which is what makes the merge a conflict-free union. The # plan assigns the section; enforce it here, or the guarantee is only a # comment: without this every worker could edit the same hot key and TP - # degenerated into differently-sharded DP. - if unit.section is not None: + # degenerated into differently-sharded DP. Extras are whole-artifact + # proposals, so the check applies to the primary only. + if unit.section is not None and aid == artifact_id: outside = [k for k in diff.ops if section_map.get(k) != unit.section] if outside: @@ -3487,15 +3648,15 @@ def _run_unit_inner(unit) -> None: # provider's output, not just the score -- the same contract as # the rollout path. self_trace = eng._record_rollout( - artifact_id=artifact_id, rendered=rendered_after, task=task, - base_version={artifact_id: mine_v}, purpose="self_verify", + artifact_id=aid, rendered=rendered_after, task=task, + base_version={aid: mine_v}, purpose="self_verify", output=after_output, reward=after, wallclock=time.time() - t_self, worker_id=f"w{unit.worker}") else: delta = 0.0 self_trace = None aggregator.ingest(EvidenceCard( - diff=diff, base_version={artifact_id: mine_v}, touched=[artifact_id], + diff=diff, base_version={aid: mine_v}, touched=[aid], before_after_delta=delta, trajectory_refs=[task], branch=mine_branch, # The records that justify this card, when recording is on: the @@ -3510,6 +3671,13 @@ def _run_unit_inner(unit) -> None: # decide whether anything should. advantage=adv)) + def _run_unit_inner(unit) -> None: + if not unit.keys: + return + task = by_id[sampler.pick(unit.keys, r)] # a task from this worker's shard + for aid in _artifact_targets(unit): + _propose_one(unit, task, aid) + try: # the parallel strategy assigns this round's tasks to workers; they run # concurrently (rollout+propose overlap) then the aggregator is the barrier. @@ -3680,8 +3848,8 @@ def _bounded(u=None): f"one successfully read: {type(e).__name__}: {str(e)[:160]}") if final is None: # nothing was ever read: hand back the seed final = EvolvingArtifact(artifact_id, dict(initial_state or strategy.initial()), - blast_radius=blast_radius, runtime=eng.runtime, - strategy=strategy) + blast_radius=blast_radius, runtime=eng.runtime, + strategy=strategy) # Scoring runs the agent, so a dead backend must not raise out of the driver # and discard everything already committed. try: diff --git a/agentdescent/evolvable.py b/agentdescent/evolvable.py index 18156464..22c1d190 100644 --- a/agentdescent/evolvable.py +++ b/agentdescent/evolvable.py @@ -88,6 +88,14 @@ class Contract: (the default) is the untouched path exactly -- no rule, no check, no overhead. Because the check is part of the merge pipeline, a contract that declares a rule can never silently stop enforcing it. + + ``depends_on`` names the artifact ids whose **contracts** this artifact + relies on. It is what turns a single-artifact commit into an atomic + adaptation transaction: when one of those artifacts commits a contract- + breaking change, this artifact must land (re-registered) in the *same* git + commit, via :meth:`~agentdescent.ledger.Ledger.commit_atomic`. Empty means + the artifact depends on nothing and a breaking change elsewhere never + touches it. """ input_schema: str = "any" @@ -95,6 +103,7 @@ class Contract: side_effects: Tuple[str, ...] = () major: int = 1 invariants: Tuple[Any, ...] = () + depends_on: Tuple[str, ...] = () def is_compatible_with(self, other: "Contract") -> bool: return self.major == other.major diff --git a/agentdescent/evolvespec.py b/agentdescent/evolvespec.py index 2eb54f5a..7bdbede6 100644 --- a/agentdescent/evolvespec.py +++ b/agentdescent/evolvespec.py @@ -226,6 +226,12 @@ class EvolveSpec: agg_config: Dict[str, Any] = field(default_factory=dict) #: Passed straight to ``evolve()``. evolve: Dict[str, Any] = field(default_factory=dict) + #: Additional artifacts to evolve alongside the primary, as + #: ``{aid: EvolvingArtifact}``. Composed by :func:`compose` into + #: ``extra_artifacts=`` for ``evolve()`` -- the worker loop proposes against + #: each, and the aggregator merges their diffs by target. Empty (default) is + #: a single-artifact run. + extra_artifacts: Dict[str, Any] = field(default_factory=dict) #: Extra import prefixes a ``module:attribute`` ref may resolve into. The #: package's own modules are always allowed; widening this is the moment to #: think about who can write the spec. @@ -963,6 +969,15 @@ def run(skill: str, task: Task) -> str: # -- knobs: kind defaults < spec.evolve < caller overrides ------------------ knobs: Dict[str, Any] = {**defaults, **spec.evolve, **overrides} + # The dedicated field and the pass-through agree; an explicit `spec.evolve` + # entry wins, exactly as it would for any other knob. + if spec.extra_artifacts: + if "extra_artifacts" in knobs: + merged = dict(knobs["extra_artifacts"]) + merged.update(spec.extra_artifacts) + knobs["extra_artifacts"] = merged + else: + knobs["extra_artifacts"] = dict(spec.extra_artifacts) held_out = float(knobs.get("held_out_frac", 0.3)) train = _train_count(len(tasks), held_out) knobs.setdefault("n_workers", max(1, min(default_workers, train))) diff --git a/agentdescent/policies.py b/agentdescent/policies.py index 1221e784..6723002b 100644 --- a/agentdescent/policies.py +++ b/agentdescent/policies.py @@ -369,12 +369,14 @@ def full_eval(self, artifact: "Evolvable") -> float: ... @runtime_checkable class LedgerProtocol(Protocol): - """Seven methods: four the aggregator calls, three more the engine calls. + """Eight methods: four the aggregator calls, four more the engine calls. Deriving this from the aggregator alone -- ``snapshot``, ``head_version``, ``commit``, ``promote_to_stable`` -- yields a contract that type-checks and then dies in ``_build_engine``, which calls ``register`` before any merge happens, in ``_safe_log`` at the end of every run, and in ``_Engine.cleanup``. + ``commit_atomic`` is the aggregator's path for a contract-breaking diff, which + must land as a single git commit with its adapters. ``repo_path`` is an attribute, not a method: the checkpoint module reads it to know where to write ``checkpoints/``. Every :class:`~agentdescent.ledger.Ledger` @@ -394,6 +396,9 @@ def snapshot(self, branch: str = ...) -> Any: ... def head_version(self, branch: str = ...) -> Dict[str, int]: ... def commit(self, new_state: "Evolvable", base_version: Dict[str, int], branch: str = ..., message: str = ...) -> Tuple[str, int]: ... + def commit_atomic(self, new_states: List["Evolvable"], + base_version: Dict[str, int], + branch: str = ..., message: str = ...) -> Tuple[str, Dict[str, int]]: ... def promote_to_stable(self, artifact_id: str) -> Optional[int]: ... def log(self, branch: str = ..., limit: int = ...) -> List[str]: ... def close(self) -> None: ... diff --git a/changelog.d/197-multi-artifact.md b/changelog.d/197-multi-artifact.md new file mode 100644 index 00000000..145c8125 --- /dev/null +++ b/changelog.d/197-multi-artifact.md @@ -0,0 +1,10 @@ +Added +Multi-artifact evolution — `extra_artifacts=` on `evolve()` and `async_evolve()` +(#197): register any number of artifacts beside the primary; the worker loop +proposes against each (PP stages name them; `PipelineParallel` lifts once +extras exist), the aggregator merges by `diff.target`, and a contract-breaking +diff lands atomically via `Ledger.commit_atomic` with exact dependent-adapter +coverage (`missing-adapters` when incomplete) and CAS retry that refuses a +stale adapter. `Contract.depends_on` drives dependent cache eviction +(`MemoryCache` / `FileCache.invalidate`). Sync/async rollout records attribute +extras to the extra's artifact id. Default path without extras is unchanged. diff --git a/docs/aggregator.md b/docs/aggregator.md index 39762eed..2d55bb7e 100644 --- a/docs/aggregator.md +++ b/docs/aggregator.md @@ -53,7 +53,7 @@ Evidence cards are bucketed by artifact; a bucket fires on batch size `B` or a 3. **Fusion** — complementary diffs are fused (model-soup style) and the union goes to the gate. With `fusion_tournament=True` it is first run against the singles on held-out and the best wins; see [below](#is-ranking-the-fusion-worth-a-sweep-per-candidate). 4. **Audit gate** — the candidate is submitted to the `AuditScheduler`; a high-blast-radius / low-trust merge is forced through the oracle, which can **veto it outright** (`oracle-rejected`) before the acceptance test runs. *The optimizer audits itself.* This is a blocking gate on the accept path, not a post-commit spot-check. 5. **Statistical acceptance** — commit only if `P(Δ > 0) > 1 − δ` under a Beta posterior comparison (not a point threshold); `δ` anneals with version. -6. **Commit** — compare-and-swap on `dev`, one artifact per merge. The `Ledger` *also* offers `commit_atomic` (2PC across several artifacts, for a contract-breaking diff that must land with its adapters), but the reference aggregator buckets by artifact and never needs it — no engine path calls it today. +6. **Commit** — compare-and-swap on `dev`, one artifact per merge. A **contract-breaking** diff is the exception: it lands through the `Ledger`'s `commit_atomic` (2PC across several artifacts, the design's "atomic adaptation transaction" — the sanctioned route for a deliberate contract change, since `commit()` refuses one), and the engine re-measures any artifact whose `Contract.depends_on` names the changed artifact. In a single-artifact run that path still never fires, because there is nothing to commit together. 7. **Dual-branch promotion** — `dev → stable` after *K* **regression-free rounds** on dev. One round is one `step()`. A commit restarts the clock (the new version has survived nothing yet) and so does an oracle rejection. So a *converged* artifact — one that stopped committing because nothing beats it — is the one most likely to be promoted, which is the point. A run that ends cleanly also publishes its head via `finalize()`, so stopping on `target_reward` does not leave `stable` a confirmation short. Deep dive on the *why*: [concepts §4](concepts.md#4-the-aggregator-a-discrete-space-optimizer). diff --git a/docs/api.md b/docs/api.md index e8f84dac..ee1293b9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -260,7 +260,8 @@ evolve( allocator: Optional[Any] = None, record_trajectories: bool = False, trajectory_store: Optional['TrajectoryStore'] = None, - contract: Optional[Contract] = None + contract: Optional[Contract] = None, + extra_artifacts: Optional[Dict[str, 'EvolvingArtifact']] = None ) -> EvolutionResult ``` @@ -320,6 +321,7 @@ evolve( | `record_trajectories` | `bool` | `False` | Persist every rollout and gate evaluation as a `Trajectory` under `/trajectories/` -- what the run provider saw, what it returned, what that earned, against which artifact version and environment. Off by default: a throwaway repo's records die with it, so the write cost is pure overhead there, and recording is a deliberate choice about keeping a run's evidence. The store is the durable base every learning-type increment reads -- process-supervised reflection, offline re-weighting, audit replay -- and it is what an `EvidenceCard.trace_ids` entry points at. Pass `trajectory_store=` with a store you built to control where records land instead. | | `trajectory_store` | `Optional['TrajectoryStore']` | `None` | An explicit `TrajectoryStore` to record into. Pass one to control where records land (a shared store, a custom root, a custom privacy configuration) without going through the `/trajectories/` default. The explicit store wins over `record_trajectories`; `None` + `record_trajectories` builds the default store. | | `contract` | `Optional[Contract]` | `None` | The artifact's contract -- the externally-visible interface, and where governance invariants are declared. Pass `Contract(invariants=(Invariant("mutable_keys", ("a", "b")),))` and the aggregator enforces each rule **before** a candidate is measured (see `invariants`); a diff that breaks one reports a `violates-invariant` outcome and never buys a held-out sweep. `None` (the default) is the reference skill contract, which declares nothing and enforces nothing -- the untouched path exactly. | +| `extra_artifacts` | `Optional[Dict[str, 'EvolvingArtifact']]` | `None` | Additional artifacts to register alongside the primary (`artifact_id`), as `{aid: EvolvingArtifact}`. The primary is the one the worker loop rolls out and proposes on; each extra is registered in the ledger and the worker proposes against it too, so a proposal can name one of them as a diff target and the aggregator merges it there (the evidence buffer already buckets by `diff.target`). `depends_on` on an extra's `Contract` declares which artifacts it relies on: a contract-breaking change to one re-measures its dependents (their cached scores were taken under the superseded contract) and lands atomically via `Ledger.commit_atomic`. Empty (the default) is the old single-artifact behaviour exactly. | ### `reflector(...)` @@ -971,7 +973,8 @@ Contract( output_schema: str = 'any', side_effects: Tuple[str, ...] = (), major: int = 1, - invariants: Tuple[Any, ...] = () + invariants: Tuple[Any, ...] = (), + depends_on: Tuple[str, ...] = () ) -> None ``` @@ -1133,6 +1136,7 @@ The vocabulary of `category`. | `CAS_CONFLICT` | `'cas-conflict'` | | `UNKNOWN_ARTIFACT` | `'unknown-artifact'` | | `VIOLATES_INVARIANT` | `'violates-invariant'` | +| `MISSING_ADAPTERS` | `'missing-adapters'` | | `CRITIC_REJECTED` | `'critic-rejected'` | | `NOISE_FLOOR` | `'noise-floor'` | | `COST_RULE` | `'cost-rule'` | @@ -3138,6 +3142,7 @@ EvolveSpec( policies: Dict[str, Any] = , agg_config: Dict[str, Any] = , evolve: Dict[str, Any] = , + extra_artifacts: Dict[str, Any] = , allow: Sequence[str] = (), version: int = 1 ) -> None @@ -3241,7 +3246,8 @@ async_evolve( allocator: Optional[Any] = None, record_trajectories: bool = False, trajectory_store: Optional[Any] = None, - contract: Optional[Contract] = None + contract: Optional[Contract] = None, + extra_artifacts: Optional[Dict[str, 'EvolvingArtifact']] = None ) -> EvolutionResult ``` @@ -3300,6 +3306,7 @@ async_evolve( | `record_trajectories` | `bool` | `False` | Exactly as in `evolve`: persist every rollout and gate evaluation as a `Trajectory` under `/trajectories/`. Off by default; pass `trajectory_store=` to control where records land. | | `trajectory_store` | `Optional[Any]` | `None` | An explicit `TrajectoryStore` to record into. `None` + `record_trajectories` builds the default store at `/trajectories`. | | `contract` | `Optional[Contract]` | `None` | Exactly as in `evolve`: the artifact's contract, where governance invariants are declared. `None` (the default) is the reference skill contract, which declares nothing and enforces nothing. | +| `extra_artifacts` | `Optional[Dict[str, 'EvolvingArtifact']]` | `None` | Exactly as in `evolve`: additional artifacts to register alongside the primary, as `{aid: EvolvingArtifact}`. The barrier-free worker proposes against each of them too (no parallel strategy, so it is the primary plus every extra), and the aggregator merges their diffs by target. Empty (the default) is the old single-artifact behaviour exactly. | --- @@ -3795,7 +3802,7 @@ The exception tuple a caller catches to treat any ledger problem as recoverable. ### `LedgerProtocol` -Seven methods: four the aggregator calls, three more the engine calls. +Eight methods: four the aggregator calls, four more the engine calls. ### `LocalWorkspaceSandbox` diff --git a/docs/concepts.md b/docs/concepts.md index 3cff7d80..b635d5a4 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -261,10 +261,12 @@ artifacts don't starve). Then, in order: 5. **Statistical acceptance** — commit only if `P(Δ > 0) > 1 − δ` under a Beta posterior comparison of candidate vs base, **not** a point threshold. `δ` anneals with version (LR decay); a trust-region caps diff size. -6. **Commit** — compare-and-swap on `dev`, one artifact per merge. (`Ledger` - also implements `commit_atomic`, a 2PC across artifacts for a - contract-breaking diff that must land with its adapters, but the reference - aggregator buckets per artifact and no engine path uses it.) +6. **Commit** — compare-and-swap on `dev`, one artifact per merge. A + contract-breaking diff is routed through `Ledger.commit_atomic` — the 2PC + across artifacts the design calls the *atomic adaptation transaction*, and the + only commit path that sanctions a deliberate contract change — and artifacts + that `depends_on` the changed one are re-measured. Single-artifact runs never + touch it; `evolve(extra_artifacts=...)` is what makes it reachable.) 7. **Dual-branch promotion** — `dev → stable` after *K* **regression-free rounds** on dev (EMA-style confirmation). A commit restarts the clock, so the artifact most likely to be promoted is the one that has *stopped* changing @@ -314,10 +316,12 @@ mechanism: groups) — both implemented, the latter also reachable from `evolve()` as [`DifficultyWeighted`](sampling.md) task sampling. The design's cross-product **(task-cluster × artifact)** is - **not implemented**: `TaskCluster` has no artifact dimension, and both reference - runtimes register exactly one artifact, as does `evolve()` — so the second axis - has nowhere to live yet, and the mechanism operates on clusters while the - problem statement is about artifacts. The design also calls for a **tail canary + **partially implemented**: `TaskCluster` has no artifact dimension yet, but + `evolve(extra_artifacts=...)` now gives the artifact axis a place to live — + the worker loop proposes against every registered artifact and the aggregator + merges their diffs by target, so the *mechanism* operates on multiple + artifacts while the *triggering* (which artifact a task's failure implicates) + still runs on clusters. The design also calls for a **tail canary set** inside held-out eval; that is **not implemented** — held-out is one undifferentiated split. - **L-value (signal layer)** — most diffs are marginal, a few are high-value diff --git a/docs/data-model.md b/docs/data-model.md index 86424d96..d7159ab2 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -159,7 +159,8 @@ world moved underneath it, and what happens next is the ## `Contract` — the interface that must not silently change ```python -Contract(input_schema="task", output_schema="text", side_effects=(), major=1) +Contract(input_schema="task", output_schema="text", side_effects=(), major=1, + depends_on=()) ``` Artifacts depend on each other's *interfaces*, not their contents. A change that @@ -168,8 +169,14 @@ contract an artifact was **registered** with and refuses any later commit whose major disagrees — a breaking change has to be re-registered deliberately rather than merged like an ordinary diff. -In a single-artifact `evolve()` run this never fires; it exists for the -multi-artifact library the design targets. Until recently it never fired at all: +`depends_on` names the artifact ids whose **contracts** this artifact relies on. +It is what turns a breaking change into an atomic adaptation transaction: when a +registered artifact commits a contract-breaking diff, the engine routes it +through `Ledger.commit_atomic` and re-measures every artifact whose `depends_on` +names it (their cached scores were taken under the superseded contract). + +In a single-artifact `evolve()` run none of this fires; it exists for +`evolve(extra_artifacts=...)`. Until recently it never fired at all: `Contract`, `is_compatible_with` and `ContractRejected` all existed and nothing called any of them, while the docstrings described the enforcement as if it were there. diff --git a/docs/modules.md b/docs/modules.md index eba82eb9..fd35af2e 100644 --- a/docs/modules.md +++ b/docs/modules.md @@ -131,13 +131,13 @@ class at a time. They are all in one table instead: | name | why it exists | what would reach it | |---|---|---| -| [`Ledger.commit_atomic`](ledger.md) | 2PC across several artifacts, for a contract-breaking diff that must land with its adapters | a multi-artifact library; `evolve()` registers exactly one | +| [`Ledger.commit_atomic`](ledger.md) | 2PC across several artifacts, for a contract-breaking diff that must land with its adapters | `evolve(extra_artifacts=...)` — a contract-breaking diff is routed through it (a single-artifact run still has nothing to commit together) | | [`L1SerialGate`](governance.md) | "at most one L1 diff in evaluation anywhere" | concurrent merging; every shipped runtime merges on one thread, so the guarantee already holds by construction | | [`ResumeQueue`](duration-scheduling.md) | turn-level checkpoints of a timed-out rollout | task-level re-run: `async_evolve(resume_queue=...)` pushes a straggler's task + version and the next idle worker re-runs it against the latest head (`result.resumed`). Turn-level continuation still needs `run(rendered, task) -> output` to expose its turns | | [`AuditScheduler.pop`](duration-scheduling.md) | draining the Ĝ-ordered audit queue out of band | `AuditScheduler(collect=True)`; the default computes priorities without queuing, because nothing drains it | | [`EvidenceBuffer.settled`](aggregator.md) | discarded evidence stays addressable — the structural advantage of artifacts over gradients | re-filing settled cards into the trajectory pool; today it is a bounded diagnostic ring | -| `TaskScheduler` × artifact axis | the design's L-task is `(task cluster × artifact)` | more than one artifact; `TaskCluster` has no artifact dimension. The *cluster* axis is reachable from `evolve()` via [`ClusterParallel`](parallelism.md) | -| [`PipelineParallel`](parallelism.md) | one artifact per stage, with upstream blame | a multi-artifact run; `evolve()` **refuses** it rather than degrading to DP in silence | +| `TaskScheduler` × artifact axis | the design's L-task is `(task cluster × artifact)` | more than one artifact; `TaskCluster` has no artifact dimension. The *cluster* axis is reachable from `evolve()` via [`ClusterParallel`](parallelism.md) — the *artifact* axis is now reachable via `extra_artifacts=` | +| [`PipelineParallel`](parallelism.md) | one artifact per stage, with upstream blame | `evolve(extra_artifacts=..., parallel=PipelineParallel(...))` — the stages name registered artifacts; without extras it is **refused** rather than degrading to DP in silence | The rule they share: a primitive that is implemented and unreachable is honest; one that is *reachable and silently does nothing* is not, which is why diff --git a/tests/test_evalcache.py b/tests/test_evalcache.py index 52f94b17..e0a69326 100644 --- a/tests/test_evalcache.py +++ b/tests/test_evalcache.py @@ -11,6 +11,7 @@ """ import json +import os from pathlib import Path import threading import time @@ -162,6 +163,36 @@ def test_a_file_cache_separates_environments_too(tmp_path): assert cache.get_or_eval(cache_key("r", "t", "image-b"), lambda: 0.0) == 0.0 +def test_file_cache_invalidate_deletes_the_targeted_entries(tmp_path): + """`FileCache.invalidate` must remove the on-disk entries it names, so a + fresh process re-evaluates rather than reading the stale score back. The + keys are (rendered, task_id, fingerprint); with a non-empty fingerprint the + old prefix comparison compared the *task id* to the fingerprint and deleted + nothing (review).""" + cache = FileCache(str(tmp_path)) + cache.get_or_eval(cache_key("R", "t1", "fp"), lambda: 0.3) + cache.get_or_eval(cache_key("R", "t2", "fp"), lambda: 0.4) + cache.invalidate("R", "fp") + assert len(os.listdir(str(tmp_path))) == 0, ( + "both fingerprint-matched files must be deleted") + # a fresh reader re-evaluates instead of reading the stale score back + fresh = FileCache(str(tmp_path)) + assert fresh.get_or_eval(cache_key("R", "t1", "fp"), lambda: 0.9) == 0.9 + + +def test_file_cache_invalidate_without_fingerprint_drops_all_tasks_on_render(tmp_path): + """No fingerprint -> every task measured on this render is invalidated, + matching MemoryCache's coarse semantics.""" + import os + cache = FileCache(str(tmp_path)) + cache.get_or_eval(cache_key("R", "t1", "fp"), lambda: 0.3) + cache.get_or_eval(cache_key("R", "t2", "fp"), lambda: 0.4) + cache.get_or_eval(cache_key("other", "t1", "fp"), lambda: 0.5) + cache.invalidate("R") + assert len(os.listdir(str(tmp_path))) == 1, ( + "only the other-render entry survives") + + def test_an_unwritable_cache_does_not_fail_the_run(tmp_path): """The answer has already been paid for. Losing the chance to reuse it is a smaller problem than losing the run.""" diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py new file mode 100644 index 00000000..163f4fb2 --- /dev/null +++ b/tests/test_multi_artifact.py @@ -0,0 +1,603 @@ +"""Multi-artifact evolution: ``extra_artifacts=``, PP stages, and atomic contracts. + +The library's single-artifact heritage is the constraint these tests pin down. +Before this feature three things were impossible at once -- ``evolve()`` +registered exactly one artifact, ``PipelineParallel`` was refused (it needs one +artifact per stage and a single-artifact run has nowhere for the stages to live), +and a contract-breaking change could not land through the engine at all (the +ledger's ``_assert_contract`` refuses it, and nothing routed through +``commit_atomic``). The tests here assert all three seams are now real: +registration, cross-artifact diff production by the worker loop, and the atomic +adaptation transaction. +""" + +import warnings + +import pytest + +from agentdescent.evolution import ( + AppendRules, + EvolvingArtifact, + Task, + evolve, +) +from agentdescent.evolvable import Contract, Diff +from agentdescent.ledger import Ledger +from agentdescent.parallel import PipelineParallel + + +def _reward(task, output): + return 1.0 if output == "correct" else 0.0 + + +def _ledger(tmp_path, extra_ids=()): + """A scratch ledger with an 'artifact' plus the given extra artifact ids.""" + def serialize(a): + return {"state": a.state, "blast_radius": a.blast_radius} + + def deserialize(aid, version, state): + return EvolvingArtifact(aid, state.get("state", {}), version, + state.get("blast_radius", 0.2)) + + lg = Ledger(str(tmp_path / "repo"), serialize, deserialize) + lg.register(EvolvingArtifact("artifact", {"k": "v0"}, blast_radius=0.2)) + for aid in extra_ids: + if aid != "artifact": + lg.register(EvolvingArtifact(aid, {}, blast_radius=0.2)) + return lg + + +def _tasks(n=12): + return [Task(id=f"t{i}", prompt=f"p{i}") for i in range(n)] + + +class _Reflector: + """Fails t0 until a clue is in the artifact, then proposes that clue.""" + + def solve(self, rendered, task): + return "correct" if "clue" in rendered or task.id != "t0" else "wrong" + + def propose(self, rendered, task, output, reward): + return "t0 clue" if task.id == "t0" else None + + +def _run(*, extra_artifacts, parallel=None, rounds=4, n_workers=2, + agent=None, **kw): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + return evolve(_tasks(), _reward, agent=agent or _Reflector(), rounds=rounds, + n_workers=n_workers, max_concurrency=n_workers, + strategy=AppendRules(), extra_artifacts=extra_artifacts, + parallel=parallel, **kw) + + +class _BreakingRules(AppendRules): + """AppendRules, but every proposal is a semver-major (contract-breaking) diff. + + Shipped strategies never mark a diff ``contract_breaking`` -- the flag is + the caller's declaration that a change is a deliberate interface break, and + only a custom strategy knows when to make it. This is that declaration, so + the tests can reach the atomic adaptation transaction end to end.""" + + def to_diff(self, state, proposal, author, base_version, target): + d = super().to_diff(state, proposal, author, base_version, target) + if d is not None: + d = Diff(d.diff_id, d.target, dict(d.ops), contract_breaking=True, + author=d.author) + return d + + +# -- registration ------------------------------------------------------------- + + +def test_extra_artifacts_are_registered_in_the_ledger(): + extra = {"skill-b": EvolvingArtifact("skill-b", {}, blast_radius=0.2)} + res = _run(extra_artifacts=extra) + commits = [line for line in res.ledger_log if "register skill-b" in line] + assert commits, "the extra artifact must be registered into the ledger" + + +def test_single_artifact_default_behaviour_is_unchanged(): + res = _run(extra_artifacts=None) + assert res.state, "the primary artifact should still evolve" + assert res.final_reward == 1.0 + + +# -- cross-artifact diffs ----------------------------------------------------- + + +def test_worker_loop_proposes_against_extra_artifacts(): + extra = {"skill-b": EvolvingArtifact("skill-b", {}, blast_radius=0.2)} + res = _run(extra_artifacts=extra) + merged = [line for line in res.ledger_log if "-> skill-b" in line] + assert merged, "a proposal must target the extra artifact and be merged" + assert any("-> artifact" in line for line in res.ledger_log), ( + "the primary must still receive proposals in a multi-artifact run") + + +def test_extra_artifacts_carry_their_own_strategy(): + from agentdescent.strategies import SingleSlot + + class _SlotReflector(_Reflector): + def propose(self, rendered, task, output, reward): + return "the one true clue" if task.id == "t0" else None + + extra = {"slot-b": EvolvingArtifact( + "slot-b", {}, blast_radius=0.2, strategy=SingleSlot())} + res = _run(extra_artifacts=extra, agent=_SlotReflector()) + merged = [line for line in res.ledger_log if "-> slot-b" in line] + assert merged + + +# -- PipelineParallel with extra_artifacts ------------------------------------ + + +def test_pipeline_parallel_is_allowed_with_extra_artifacts(): + extra = {"stage-b": EvolvingArtifact("stage-b", {}, blast_radius=0.2)} + res = _run(extra_artifacts=extra, + parallel=PipelineParallel(stages=["artifact", "stage-b"])) + assert any("-> stage-b" in line for line in res.ledger_log), ( + "PP's second stage must have an artifact to propose against") + + +def test_pipeline_parallel_without_extra_artifacts_is_still_refused(): + with pytest.raises(ValueError) as excinfo: + _run(extra_artifacts=None, + parallel=PipelineParallel(stages=["artifact"])) + assert "PipelineParallel" in str(excinfo.value) + + +def test_pipeline_parallel_stages_must_name_registered_artifacts(): + extra = {"stage-b": EvolvingArtifact("stage-b", {}, blast_radius=0.2)} + with pytest.raises(ValueError) as excinfo: + _run(extra_artifacts=extra, + parallel=PipelineParallel(stages=["artifact", "ghost-stage"])) + assert "ghost-stage" in str(excinfo.value) + + +# -- the atomic adaptation transaction ---------------------------------------- + + +def test_contract_breaking_diff_lands_atomically(): + """A contract-breaking diff with no declared dependents lands atomically + through `commit_atomic` -- the sanctioned route for a deliberate contract + change, bypassing `_assert_contract`'s refusal.""" + # No `depends_on`: this breaking change has nothing it must land with. + extra = {"b": EvolvingArtifact("b", {}, blast_radius=0.2, + contract=Contract( + input_schema="text", output_schema="text", + major=1))} + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + res = evolve(_tasks(), _reward, agent=_Reflector(), rounds=3, + n_workers=2, max_concurrency=2, strategy=_BreakingRules(), + extra_artifacts=extra) + log = res.ledger_log + assert any("-> artifact" in line and "contract" in line for line in log), ( + "a contract-breaking diff must be committed atomically, with a message " + "that says so") + + +def test_contract_breaking_with_dependents_but_no_adapters_is_refused(tmp_path): + """A contract-breaking change to an artifact with declared dependents must + land *with* its adapted dependents in the same atomic commit. If none are + supplied, the merge is refused (`missing-adapters`) rather than half-applied + -- committing just the breaking artifact would leave the dependents + registered against a superseded contract.""" + from agentdescent.aggregator import Aggregator, AggregatorConfig, _Candidate + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + + lg = _ledger(tmp_path, extra_ids=("artifact", "b")) + agg = Aggregator(lg, ThreeLayerVerifier( + eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()), AuditScheduler(), + AggregatorConfig(batch_trigger=1)) + # The engine wires the reverse dependency graph: "artifact" depends on "b". + agg.contract_dependents = {"artifact": ["b"]} + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff( + diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.category == "missing-adapters", ( + f"breaking change with dependents but no adapters must be refused, " + f"got {report.category}") + assert report.committed_version is None, "nothing may commit" + assert lg.head_version(Ledger.DEV).get("artifact") == 1, ( + "the breaking change must not land without its adapters") + + +def test_contract_breaking_with_adapters_commits_both_atomically(tmp_path): + """Supplying the adapted dependents makes the atomic adaptation transaction + go through: candidate + adapters land in one `commit_atomic`, dependents + are re-registered against the new contract.""" + from agentdescent.aggregator import Aggregator, AggregatorConfig, _Candidate + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + lg = _ledger(tmp_path, extra_ids=("artifact", "b")) + agg = Aggregator(lg, ThreeLayerVerifier( + eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()), AuditScheduler(), + AggregatorConfig(batch_trigger=1)) + agg.contract_dependents = {"artifact": ["b"]} + invalidated = [] + agg.on_contract_change = lambda aid, branch: invalidated.append( + (aid, branch)) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + b_old = snap.get("b") + cand = base.apply(Diff( + diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + b_adapted = b_old.apply(Diff( + diff_id="adapter", target="b", ops={"k2": "v2"}, author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + adapters=[b_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.committed_version is not None, "with adapters, it commits" + after = lg.snapshot(Ledger.DEV) + assert after.version["artifact"] == 2 and after.version["b"] == 2, ( + "candidate and adapter must land in the same atomic commit") + assert invalidated == [("artifact", "dev")], ( + "dependents must be re-measured after the breaking change, on the " + f"branch that changed (got {invalidated})") + + +def _agg_with_dependents(tmp_path, extra_ids=("artifact", "b"), dependents=None): + from agentdescent.aggregator import Aggregator, AggregatorConfig, _Candidate + from agentdescent.scheduler import AuditScheduler + from agentdescent.verifier import ThreeLayerVerifier, VerifierBudget + lg = _ledger(tmp_path, extra_ids=extra_ids) + agg = Aggregator(lg, ThreeLayerVerifier( + eval_fn=lambda a, t: 0.5, held_out=[1, 2, 3], + budget=VerifierBudget()), AuditScheduler(), + AggregatorConfig(batch_trigger=1)) + agg.contract_dependents = dependents or {} + return lg, agg + + +def test_contract_breaking_with_partial_adapters_is_refused(tmp_path): + """The adapters must cover *every* declared dependent: with dependents + ['b', 'c'], an adapter for only 'b' is still `missing-adapters` -- 'c' + would stay registered against the superseded contract.""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "b", "c"), + dependents={"artifact": ["b", "c"]}) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + b_adapted = snap.get("b").apply(Diff(diff_id="adapter", target="b", + ops={"k2": "v2"}, author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + adapters=[b_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.category == "missing-adapters", ( + f"a partial adapter set must be refused, got {report.category}") + assert report.committed_version is None + assert "missing ['c']" in report.reason, report.reason + + +def test_contract_breaking_with_unrelated_adapter_is_refused(tmp_path): + """An adapter for an artifact that is not a declared dependent must be + refused too: the real dependent would stay against the superseded contract + while the transaction pretends to be complete.""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "b", "x"), + dependents={"artifact": ["b"]}) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + x_adapted = snap.get("x").apply(Diff(diff_id="adapter", target="x", + ops={"k3": "v3"}, author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + adapters=[x_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.category == "missing-adapters", ( + f"an adapter for an unrelated artifact must be refused, " + f"got {report.category}") + assert report.committed_version is None + assert "extra ['x']" in report.reason, report.reason + + +def test_contract_breaking_with_adapters_but_no_dependents_is_refused(tmp_path): + """An artifact with *no* declared dependents must not carry adapters in a + breaking commit either: the exact-set guard is the invariant, and an empty + required set with a nonempty adapter set is as exact a violation as a + partial one. Committing the stray adapter would change an artifact no gate + asked this diff to touch (PR #197 review: the guard only ran when the + required set was nonempty).""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "x"), + dependents={}) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + x_adapted = snap.get("x").apply(Diff(diff_id="adapter", target="x", + ops={"unexpected": "committed"}, + author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + adapters=[x_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.category == "missing-adapters", ( + f"adapters with no declared dependents must be refused, " + f"got {report.category}") + assert report.committed_version is None + assert "extra ['x']" in report.reason, report.reason + assert lg.head_version(Ledger.DEV).get("x") == 1, ( + "the unrelated adapter must not land") + + +def test_contract_breaking_with_duplicate_adapters_is_refused(tmp_path): + """Duplicating one dependent's adapter while another stays missing must be + refused, not treated as coverage.""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "b", "c"), + dependents={"artifact": ["b", "c"]}) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + b_adapted = snap.get("b").apply(Diff(diff_id="adapter", target="b", + ops={"k2": "v2"}, author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + adapters=[b_adapted, b_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.category == "missing-adapters", ( + f"duplicated adapters must be refused, got {report.category}") + assert report.committed_version is None + assert "duplicated=True" in report.reason, report.reason + + +def test_commit_atomic_retry_rejects_a_stale_adapter(tmp_path): + """A CAS conflict that advanced one of the adapter artifacts must not retry + the *old* adapter state over it: the adapter is a final state, not a diff, + so it cannot be rebased, and the retry rejects the stale transaction.""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "b"), + dependents={"artifact": ["b"]}) + snap = lg.snapshot(Ledger.DEV) + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + b_old = snap.get("b") + b_adapted = b_old.apply(Diff(diff_id="adapter", target="b", + ops={"k2": "v2"}, author="w")) + + # A concurrent writer commits to 'b' after the candidate was prepared but + # before our atomic commit lands -- the CAS on 'b' then fails, and the + # retry must refuse to commit the adapter written against the old head. + lg.commit(b_old.apply(Diff(diff_id="concurrent", target="b", + ops={"concurrent": "field"})), + {"b": b_old.version}, branch=Ledger.DEV) + + new_version = agg._commit_with_retry( + "artifact", cand, + Diff(diff_id="d", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + snap.version, adapters=[b_adapted]) + assert new_version is None, ( + "a stale adapter must be rejected, not committed over the fresh head") + after = lg.snapshot(Ledger.DEV) + assert after.version.get("b") == 2, ( + "the concurrent b commit must survive (no lost update)") + assert after.get("b").state.get("concurrent") == "field", ( + "the concurrent adapter field must not be deleted by the retry") + + +def test_contract_breaking_fork_candidate_commits_to_its_own_fork(tmp_path): + """A contract-breaking candidate prepared against a fork must land on that + fork, not on `dev` -- the same branch routing the non-breaking path got in + the multi-head fix. `_commit_with_retry` hard-coded `branch=Ledger.DEV` on + the `commit_atomic` call, so a breaking fork candidate overwrote `dev` + (deleting its independent state) while the fork stayed behind (review).""" + from agentdescent.aggregator import _Candidate + lg, agg = _agg_with_dependents(tmp_path, extra_ids=("artifact", "b"), + dependents={"artifact": ["b"]}) + # Give dev and the fork divergent states at the same artifact version: the + # wrong-branch CAS would otherwise be the one that *succeeds*. + lg.fork("head/0") + dev_art = lg.snapshot(Ledger.DEV).get("artifact") + lg.commit(dev_art.apply(Diff(diff_id="dev_only", target="artifact", + ops={"dev_only": "preserve"}, author="w")), + {"artifact": dev_art.version}, branch=Ledger.DEV) + fork_art = lg.snapshot("head/0").get("artifact") + lg.commit(fork_art.apply(Diff(diff_id="fork_only", target="artifact", + ops={"fork_only": "preserve"}, author="w")), + {"artifact": fork_art.version}, branch="head/0") + + snap = lg.snapshot("head/0") + base = snap.get("artifact") + cand = base.apply(Diff(diff_id="break", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True)) + b_old = snap.get("b") + b_adapted = b_old.apply(Diff(diff_id="adapter", target="b", + ops={"k2": "v2"}, author="w")) + c = _Candidate( + artifact_id="artifact", artifact=base, candidate=cand, + diff=Diff(diff_id="break", target="artifact", ops={"k": "v"}, + author="w", contract_breaking=True), + cards=[], survivor_cards=[], head=snap.version, + fused=False, considered=1, survived=1, discarded=0, conflicts=0, + branch="head/0", adapters=[b_adapted], + base_counts=(2.0, 0.0), cand_counts=(3.0, 0.0), + base_cheap=1.0, cand_cheap=1.0) + report = agg._decide(c) + assert report.committed_version is not None, "with adapters, it commits" + + after_fork = lg.snapshot("head/0") + assert after_fork.get("artifact").state.get("fork_only") == "preserve", ( + "the fork's own independent state must survive") + assert after_fork.get("artifact").state.get("k") == "v", ( + "the breaking diff must land on the fork") + after_dev = lg.snapshot(Ledger.DEV) + assert after_dev.get("artifact").state.get("dev_only") == "preserve", ( + "dev must be untouched by a fork candidate") + assert after_dev.version.get("artifact") == 2, ( + "dev's version must not advance from a fork candidate " + f"(got {after_dev.version.get('artifact')})") + + +def test_contract_dependents_are_re_measured_after_a_breaking_change(tmp_path): + """After a breaking commit on the primary, dependent 'b' loses its cached + scores (they were measured under the superseded contract). The engine's + `invalidate_dependents` (wired as `aggregator.on_contract_change`) evicts + them by the dependent's render.""" + from agentdescent.evalcache import MemoryCache + from agentdescent.evolution import _Engine + + cache = MemoryCache() + lg = _ledger(tmp_path, extra_ids=("artifact", "b")) + b_art = lg.snapshot(Ledger.DEV).get("b") + key = (b_art.render(), "t0", "") + cache.get_or_eval(key, lambda: 0.7) + + runtime = type("Runtime", (), {})() # a stand-in _Runtime with a cache + runtime.cache = cache + eng = _Engine.__new__(_Engine) + eng.ledger = lg + eng.runtime = runtime + eng.contract_dependents = {"artifact": ["b"]} + eng.invalidate_dependents("artifact") + + assert key not in cache._values, ( + "the dependent artifact's cached evaluation must be evicted after a " + "contract-breaking commit on what it depends on") + + +def test_commit_atomic_is_all_or_nothing(tmp_path): + def serialize(a): return {"state": a.state, "blast_radius": a.blast_radius} + + def deserialize(aid, version, state): + return EvolvingArtifact(aid, state.get("state", {}), version, + state.get("blast_radius", 0.2)) + + led = Ledger(str(tmp_path / "repo"), serialize, deserialize) + a = EvolvingArtifact("a", {"v": "1"}, blast_radius=0.2) + b = EvolvingArtifact("b", {"v": "2"}, blast_radius=0.2, + contract=Contract(depends_on=("a",))) + led.register(a) + led.register(b) + base = led.head_version(Ledger.DEV) + a_new = EvolvingArtifact("a", {"v": "1", "extra": "x"}, blast_radius=0.2, + contract=Contract(major=2)) + b_new = EvolvingArtifact("b", {"v": "2"}, blast_radius=0.2, + contract=Contract(depends_on=("a",))) + _, vv = led.commit_atomic([a_new, b_new], base) + assert vv["a"] == 2 and vv["b"] == 2 + + +def _run_trajectory_probe(tmp_path, *, asynchronous=False): + """Run evolve/async_evolve with an extra artifact and a trajectory store, + returning the store. skill-b carries a distinctive state marker so its + records can be told apart from the primary's by render.""" + from agentdescent.trajectory import TrajectoryStore + + extra = {"skill-b": EvolvingArtifact( + "skill-b", {"seed": "SKILLB-MARKER"}, blast_radius=0.2)} + store = TrajectoryStore(str(tmp_path / "traj")) + kw = dict(extra_artifacts=extra, trajectory_store=store, + repo_path=str(tmp_path / "repo")) + if asynchronous: + from agentdescent.async_evolve import async_evolve + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + async_evolve(_tasks(), _reward, agent=_Reflector(), + n_workers=2, max_iters=8, max_seconds=2.0, + held_out_frac=0.5, **kw) + else: + _run(extra_artifacts=extra, trajectory_store=store, + repo_path=str(tmp_path / "repo")) + return store + + +def _extra_records(store): + """Rollout/self-verify records whose rendered text is the extra's.""" + from agentdescent.trajectory import (PURPOSE_ROLLOUT, + PURPOSE_SELF_VERIFY) + out = [] + for r in store.iter_records(): + if r.purpose in (PURPOSE_ROLLOUT, PURPOSE_SELF_VERIFY) \ + and "SKILLB-MARKER" in r.rendered: + out.append(r) + return out + + +def test_sync_rollout_records_attribute_extras_to_the_extra(tmp_path): + """The sync worker loop records an extra's rollout under the *extra's* id, + not the primary's: `_record_rollout` is called once per target `aid`, so + an extra's records must carry `artifact_id=aid` and a version vector keyed + by it, matching the async path (review: both calls hard-coded the primary + id, so skill-b's evidence was attributed to 'artifact').""" + store = _run_trajectory_probe(tmp_path) + extra_recs = _extra_records(store) + assert extra_recs, "premise: skill-b must be rolled out and recorded" + assert all(r.artifact_id == "skill-b" for r in extra_recs), ( + "every record whose render is the extra's must carry the extra's id, " + f"got {sorted({r.artifact_id for r in extra_recs})}") + # the primary's records stay on the primary + from agentdescent.trajectory import PURPOSE_ROLLOUT + primary_recs = [r for r in store.iter_records(purpose=PURPOSE_ROLLOUT) + if "SKILLB-MARKER" not in r.rendered] + assert primary_recs and all(r.artifact_id == "artifact" + for r in primary_recs), ( + "the primary's rollout records must stay attributed to the primary") + + +def test_async_rollout_records_attribute_extras_to_the_extra(tmp_path): + """Async twin: the barrier-free worker already attributes extras correctly; + pin it so the two engines cannot drift again (review).""" + store = _run_trajectory_probe(tmp_path, asynchronous=True) + extra_recs = _extra_records(store) + assert extra_recs, "premise: skill-b must be rolled out and recorded" + assert all(r.artifact_id == "skill-b" for r in extra_recs), ( + "every record whose render is the extra's must carry the extra's id, " + f"got {sorted({r.artifact_id for r in extra_recs})}")