From bc3ef9072f0f3e5bbf3878f6fa4c96477f94c2c3 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:40:29 +0800 Subject: [PATCH 1/7] =?UTF-8?q?feat(multi-artifact):=20evolve=20multiple?= =?UTF-8?q?=20artifacts=20via=20extra=5Fartifacts=3D=20=E2=80=94=20worker?= =?UTF-8?q?=20loop,=20PP=20stages,=20atomic=20contract=20changes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit evolve() and async_evolve() register any number of artifacts beside the primary. The worker loop proposes against each (PP stages name them; the stages must resolve to registered artifact ids), the aggregator merges their diffs by target (the evidence buffer already bucketed by diff.target), and a contract-breaking diff lands atomically through Ledger.commit_atomic — the one path the ledger sanctions for a deliberate contract change, previously unreachable from any engine path. Contract.depends_on declares the reverse dependency graph: when a registered artifact commits a contract-breaking change, dependents' cached evaluations are evicted so they re-measure under the new contract (MemoryCache/FileCache gain invalidate(rendered)). EvolveSpec carries extra_artifacts; docs/api.md regenerated; 10 tests in tests/test_multi_artifact.py. --- CHANGELOG.md | 30 +++++ agentdescent/aggregator.py | 37 +++++- agentdescent/async_evolve.py | 198 ++++++++++++++++++------------ agentdescent/evalcache.py | 36 ++++++ agentdescent/evolution.py | 232 +++++++++++++++++++++++++++++------ agentdescent/evolvable.py | 9 ++ agentdescent/evolvespec.py | 15 +++ agentdescent/policies.py | 7 +- docs/aggregator.md | 2 +- docs/api.md | 14 ++- docs/concepts.md | 20 +-- docs/data-model.md | 13 +- docs/modules.md | 6 +- tests/test_multi_artifact.py | 208 +++++++++++++++++++++++++++++++ 14 files changed, 685 insertions(+), 142 deletions(-) create mode 100644 tests/test_multi_artifact.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ce998af..e354a155 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -314,6 +314,36 @@ value-directed evaluation or Genesis blind-mode proposals. task past the cap — so a chronically slow task gets exactly `max_attempts` re-runs instead of an infinite resume loop starving the worker's fresh tasks. +- **Multi-artifact evolution — `extra_artifacts=` on `evolve()` and `async_evolve()`.** + The design's cross-product (task-cluster × artifact) has a place to live: both + engines register any number of artifacts, the worker loop proposes against each + (PP stages name them; `PipelineParallel` is no longer refused when they exist), + the aggregator merges their diffs by target (the evidence buffer already bucketed + by `diff.target`), and a contract-breaking diff lands **atomically** via the + ledger's `commit_atomic` — which had no engine path before. + + **`extra_artifacts=`** — a `{aid: EvolvingArtifact}` dict alongside the primary. + Each extra carries its own `blast_radius`, `strategy` and `Contract`; the worker + loop rolls each out and proposes against it (synchronously and on the barrier-free + path), and a proposal can name any registered artifact as its diff target. + + **`PipelineParallel`** — `_reject_pipeline_parallel` lifts once `extra_artifacts` + are registered, and each stage must name a registered artifact (unknown stages + are refused, not silently dropped). `evolve(parallel=PipelineParallel(...))` + without extras still raises, as before. + + **Contract dependency graph + atomic adaptation.** `Contract` gains + `depends_on` (artifact ids whose contracts it relies on). When a registered + artifact commits a contract-breaking diff, the commit goes through + `Ledger.commit_atomic` (the only path the ledger sanctions for a deliberate + contract change), and the dependents' cached evaluations are evicted so they are + re-measured under the superseded-then-replaced contract. `Contract.is_compatible_with` + is now enforced by the ledger for every commit. + + `EvolveSpec` supports `extra_artifacts` too. `MemoryCache` / `FileCache` gain + `invalidate(rendered)`. 10 tests in `tests/test_multi_artifact.py`; full suite + exit 0. + - **`ReplaySampler` + `ProposalContext.rejected` — make the settled-evidence pool a consumer.** The aggregator's settled pool (stale / oversized / CAS-conflict evidence cards) was diagnostic-only: nothing read it back. Two consumers now close that loop. **`ReplaySampler`** (``agentdescent/sampling.py``) is a ``TaskSampler`` that up-weights tasks whose recent proposals were discarded — a signal that is different from pass rate (a task whose proposals keep going stale is not "easy", it is being out-competed by the parallel scheduler). Additive replay bonus, saturating at ``capped_at`` (default 10), scaled by ``temperature`` (default 0.0 = ``DifficultyWeighted`` identically). Wired via ``aggregator.set_settled_consumer(sampler.settle)`` on both the sync and async paths. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index aab14812..2204a2a9 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -701,12 +701,18 @@ 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. 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], None]] = None 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). @@ -1246,7 +1252,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 @@ -1267,11 +1273,27 @@ 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)} try: + if diff.contract_breaking: + new_states = [candidate] + list(adapters or ()) + _, vv = self.ledger.commit_atomic( + new_states, base_vv, branch=Ledger.DEV, + 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)) @@ -1549,4 +1571,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) + 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..d2940f4e 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,29 @@ 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) + if fingerprint: + prefix = [rendered, fingerprint] + else: + prefix = [rendered] + 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") + if key and isinstance(key, list) and key[:len(prefix)] == prefix: + os.remove(path) + except (OSError, ValueError, KeyError, TypeError): + continue + def _path(self, key: Any) -> str: import hashlib diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index ed38f387..abc57692 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]: @@ -1854,6 +1871,46 @@ 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) -> None: + """Drop cached evaluations of artifacts that depend on ``changed``. + + Called when a contract-breaking diff commits on ``changed_artifact_id``: + 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(Ledger.DEV) + 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 @@ -2051,7 +2108,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 @@ -2193,6 +2251,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 @@ -2210,7 +2280,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: @@ -2289,6 +2360,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 @@ -2492,12 +2589,25 @@ 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. + try: + aggregator.on_contract_change = eng.invalidate_dependents + except AttributeError: # a custom aggregator without the slot + pass + return eng def evolve( @@ -2629,6 +2739,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. @@ -2993,6 +3110,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 ------- @@ -3079,7 +3208,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 @@ -3120,7 +3250,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) @@ -3140,7 +3270,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() @@ -3364,13 +3495,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. @@ -3400,7 +3553,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 @@ -3424,15 +3577,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: @@ -3481,7 +3636,7 @@ def _run_unit_inner(unit) -> None: 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 @@ -3496,6 +3651,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. @@ -3666,8 +3828,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 ef5626a1..311cb841 100644 --- a/agentdescent/policies.py +++ b/agentdescent/policies.py @@ -350,12 +350,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` @@ -375,6 +377,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/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 b99e04d1..d068e9fe 100644 --- a/docs/api.md +++ b/docs/api.md @@ -258,7 +258,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 ``` @@ -318,6 +319,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(...)` @@ -969,7 +971,8 @@ Contract( output_schema: str = 'any', side_effects: Tuple[str, ...] = (), major: int = 1, - invariants: Tuple[Any, ...] = () + invariants: Tuple[Any, ...] = (), + depends_on: Tuple[str, ...] = () ) -> None ``` @@ -3105,6 +3108,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 @@ -3208,7 +3212,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 ``` @@ -3267,6 +3272,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. | --- @@ -3762,7 +3768,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_multi_artifact.py b/tests/test_multi_artifact.py new file mode 100644 index 00000000..3fe4af2b --- /dev/null +++ b/tests/test_multi_artifact.py @@ -0,0 +1,208 @@ +"""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 _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(): + extra = {"b": EvolvingArtifact( + "b", {}, blast_radius=0.2, + contract=Contract(input_schema="text", output_schema="text", + major=1, depends_on=("artifact",)))} + + 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_dependents_are_re_measured_after_a_breaking_change(): + """After a breaking commit on the primary, dependent 'b' loses its cached + scores (they were measured under the superseded contract).""" + from agentdescent.evalcache import MemoryCache + + cache = MemoryCache() + b = EvolvingArtifact("b", {"k": "v"}, blast_radius=0.2, + contract=Contract(depends_on=("artifact",))) + key = (b.render(), "t0", "") + cache.get_or_eval(key, lambda: 0.7) + + extra = {"b": EvolvingArtifact( + "b", {"k": "v"}, blast_radius=0.2, + contract=Contract(depends_on=("artifact",)))} + + from agentdescent.policies import Policies + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + evolve(_tasks(), _reward, agent=_Reflector(), rounds=3, + n_workers=2, max_concurrency=2, strategy=_BreakingRules(), + extra_artifacts=extra, policies=Policies(eval_cache=cache)) + + # the dependent b was re-measured: a hit would return the stale 0.7 + 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 From f61473b6f319d12db5949eb80d50375e414ccaf1 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:40:38 +0800 Subject: [PATCH 2/7] =?UTF-8?q?fix(multi-artifact):=20enforce=20the=20atom?= =?UTF-8?q?ic=20adaptation=20transaction=20=E2=80=94=20refuse=20a=20breaki?= =?UTF-8?q?ng=20change=20without=20adapted=20dependents?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _commit_with_retry accepted `adapters` but _decide never supplied dependent states, so a contract-breaking commit silently landed as just the candidate — leaving artifacts whose `Contract.depends_on` names it registered against a superseded contract (PR #197 review). - _decide refuses a contract-breaking diff on an artifact with declared dependents when no adapted states are provided (new `missing-adapters` MergeOutcome), rather than half-applying the break. - When adapters are supplied, the atomic commit now CAS-checks every state it writes: the base version vector names the candidate and each adapter, so the whole transaction is genuinely all-or-nothing. Tests: refusing a breaking change with dependents but no adapters; committing candidate + adapters in one atomic commit and re-measuring dependents; direct Engine.invalidate_dependents eviction. --- CHANGELOG.md | 8 ++ agentdescent/aggregator.py | 61 +++++++++++++-- agentdescent/evolution.py | 5 +- docs/api.md | 1 + tests/test_multi_artifact.py | 140 ++++++++++++++++++++++++++++++----- 5 files changed, 187 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e354a155..ba43ced2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -340,6 +340,14 @@ value-directed evaluation or Genesis blind-mode proposals. re-measured under the superseded-then-replaced contract. `Contract.is_compatible_with` is now enforced by the ledger for every commit. + **The atomic adaptation transaction is enforced, not assumed.** A + contract-breaking change to an artifact with declared dependents must land + *with* the adapted dependent states in the same `commit_atomic` — if none are + supplied, the merge is refused (`missing-adapters`) rather than silently + committing just the breaking artifact and leaving dependents registered + against a superseded contract. The base version vector for an atomic commit + now names every state it writes (candidate + adapters). + `EvolveSpec` supports `extra_artifacts` too. `MemoryCache` / `FileCache` gain `invalidate(rendered)`. 10 tests in `tests/test_multi_artifact.py`; full suite exit 0. diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 2204a2a9..dc07d1b2 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" def __str__(self) -> str: # so f-strings render the value, not the name return self.value @@ -346,6 +353,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. @@ -713,6 +725,13 @@ def __init__( #: evaluations were taken under the superseded contract). Optional -- #: ``None`` keeps the old behaviour of not noticing. self.on_contract_change: Optional[Callable[[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). @@ -1286,10 +1305,18 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head, """ 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: - new_states = [candidate] + list(adapters or ()) _, vv = self.ledger.commit_atomic( new_states, base_vv, branch=Ledger.DEV, message=f"merge {diff.diff_id} -> {artifact_id} (contract)") @@ -1558,11 +1585,31 @@ 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. If this artifact has declared + # dependents (`Contract.depends_on`) and the worker supplied none, + # committing just the breaking artifact would leave them registered + # against a superseded contract. Refuse rather than half-apply: the + # worker that wants the break must also produce the adapters. + if (c.diff.contract_breaking + and self.contract_dependents.get(c.artifact_id) + and not getattr(c, "adapters", None)): + _observe_outcome(False) + return report(None, None, + f"contract-breaking change to {c.artifact_id} has " + f"declared dependents " + f"{sorted(self.contract_dependents[c.artifact_id])} " + "but no adapted states to land with it; supply them " + "as the same atomic commit's adapters", + 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: diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index abc57692..c6ac698f 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -2602,9 +2602,12 @@ def _default_aggregator(ledger, verifier, audit, config, policy): # 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. + # 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 diff --git a/docs/api.md b/docs/api.md index d068e9fe..3b7d5a5f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1133,6 +1133,7 @@ The vocabulary of `category`. | `CAS_CONFLICT` | `'cas-conflict'` | | `UNKNOWN_ARTIFACT` | `'unknown-artifact'` | | `VIOLATES_INVARIANT` | `'violates-invariant'` | +| `MISSING_ADAPTERS` | `'missing-adapters'` | ### `MergeReport(...)` diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py index 3fe4af2b..165ca483 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -30,6 +30,23 @@ 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)] @@ -142,10 +159,14 @@ def test_pipeline_parallel_stages_must_name_registered_artifacts(): def test_contract_breaking_diff_lands_atomically(): - extra = {"b": EvolvingArtifact( - "b", {}, blast_radius=0.2, - contract=Contract(input_schema="text", output_schema="text", - major=1, depends_on=("artifact",)))} + """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") @@ -158,29 +179,108 @@ def test_contract_breaking_diff_lands_atomically(): "that says so") -def test_contract_dependents_are_re_measured_after_a_breaking_change(): +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: invalidated.append(aid) + 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"], ( + "dependents must be re-measured after the breaking change") + + +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).""" + 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() - b = EvolvingArtifact("b", {"k": "v"}, blast_radius=0.2, - contract=Contract(depends_on=("artifact",))) - key = (b.render(), "t0", "") + 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) - extra = {"b": EvolvingArtifact( - "b", {"k": "v"}, blast_radius=0.2, - contract=Contract(depends_on=("artifact",)))} - - from agentdescent.policies import Policies - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - evolve(_tasks(), _reward, agent=_Reflector(), rounds=3, - n_workers=2, max_concurrency=2, strategy=_BreakingRules(), - extra_artifacts=extra, policies=Policies(eval_cache=cache)) + 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") - # the dependent b was re-measured: a hit would return the stale 0.7 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") From 694818747c0266b14b346558d20a2b32d5b126dd Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:40:43 +0800 Subject: [PATCH 3/7] fix(multi-artifact): require exact dependent-ID coverage and reject stale adapters Two safety gaps in the atomic adaptation transaction. 1. _decide() only refused an *empty* adapter list, so a breaking change to an artifact with dependents [b, c] could commit with just b adapted (c stayed against the superseded contract), or with an adapter for an unrelated artifact (b stayed). The adapter ids must now cover exactly the declared dependents -- no missing id, no extra id, no duplicate -- or the merge is refused as missing-adapters. 2. _commit_with_retry()'s CAS-retry rebased only the primary candidate and kept the old adapter states, so a concurrent commit to an adapter's artifact would be overwritten by the stale adapter written as the next version. An adapter is a final state, not a diff, so it cannot be rebased; the retry now re-validates each adapter against the fresh head and rejects the transaction when one is stale. Tests: partial-set, wrong-target, duplicate adapters all refused; a concurrent adapter update survives (no lost update) when the atomic retry is rejected. --- agentdescent/aggregator.py | 51 ++++++++----- tests/test_multi_artifact.py | 136 +++++++++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 16 deletions(-) diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index dc07d1b2..d9ab56bc 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -1345,6 +1345,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): @@ -1590,22 +1600,31 @@ def _observe_outcome(committed: bool) -> None: # 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. If this artifact has declared - # dependents (`Contract.depends_on`) and the worker supplied none, - # committing just the breaking artifact would leave them registered - # against a superseded contract. Refuse rather than half-apply: the - # worker that wants the break must also produce the adapters. - if (c.diff.contract_breaking - and self.contract_dependents.get(c.artifact_id) - and not getattr(c, "adapters", None)): - _observe_outcome(False) - return report(None, None, - f"contract-breaking change to {c.artifact_id} has " - f"declared dependents " - f"{sorted(self.contract_dependents[c.artifact_id])} " - "but no adapted states to land with it; supply them " - "as the same atomic commit's adapters", - MergeOutcome.MISSING_ADAPTERS) + # 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. + required = set(self.contract_dependents.get(c.artifact_id, ())) + if c.diff.contract_breaking and required: + 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, diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py index 165ca483..eada500d 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -259,6 +259,142 @@ def test_contract_breaking_with_adapters_commits_both_atomically(tmp_path): "dependents must be re-measured after the breaking change") +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_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_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 From c8639f8e80c1ad3145015fe1fc3d17ec3c6de830 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:40:47 +0800 Subject: [PATCH 4/7] fix(multi-artifact): reject adapters when the artifact has no declared dependents The exact-set guard only ran when the required set was nonempty, so a contract-breaking diff with no dependents could carry an adapter for an unrelated artifact and commit it -- changing an artifact no gate asked this diff to touch. Run the guard whenever either side is nonempty: an empty required set with a nonempty adapter set is as exact a violation as a partial one. --- agentdescent/aggregator.py | 8 ++++++-- tests/test_multi_artifact.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index d9ab56bc..67eaf949 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -1605,9 +1605,13 @@ def _observe_outcome(committed: bool) -> None: # 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 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 and required: + 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)) diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py index eada500d..56fb8faf 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -331,6 +331,42 @@ def test_contract_breaking_with_unrelated_adapter_is_refused(tmp_path): 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 cac6394b36465163820c93fd89e65ef7d73f5196 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:40:52 +0800 Subject: [PATCH 5/7] fix(multi-artifact): route contract-breaking commits to the candidate's branch _commit_with_retry passed branch through to the non-breaking commit path but hard-coded Ledger.DEV on the commit_atomic path, so a contract-breaking candidate prepared against a fork overwrote dev (deleting its independent state) while the fork stayed behind. commit_atomic now targets the same branch the candidate was prepared on, matching the multi-head branch routing. Regression: with dev and head/0 at the same artifact version and divergent states, a breaking fork candidate lands on head/0 and dev's state and version are untouched. --- agentdescent/aggregator.py | 2 +- tests/test_multi_artifact.py | 53 ++++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 67eaf949..27c2355c 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -1318,7 +1318,7 @@ def _commit_with_retry(self, artifact_id, candidate, diff, head, try: if diff.contract_breaking: _, vv = self.ledger.commit_atomic( - new_states, base_vv, branch=Ledger.DEV, + new_states, base_vv, branch=branch, message=f"merge {diff.diff_id} -> {artifact_id} (contract)") return vv[artifact_id] _, new_version = self.ledger.commit( diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py index 56fb8faf..13295c5d 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -431,6 +431,59 @@ def test_commit_atomic_retry_rejects_a_stale_adapter(tmp_path): "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 From 3a10ce1104261e4882b578cc79758f041b73330f Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:41:01 +0800 Subject: [PATCH 6/7] fix(evalcache): FileCache.invalidate matches on the real key triple FileCache.invalidate built prefix=[rendered, fingerprint] and compared it with key[:2], but cache_key is (rendered, task_id, fingerprint) -- so with a non-empty fingerprint it compared the task id to the fingerprint and deleted nothing, silently keeping stale on-disk scores while the in-memory layer reported the invalidation as successful. It now matches key[0] == rendered and (not fingerprint or key[2] == fingerprint), the same semantics MemoryCache uses. Regressions cover fingerprint-targeted and render-wide invalidation, and a fresh FileCache re-evaluating. Also thread the commit branch through on_contract_change so invalidate_dependents evicts the *fork's* dependent renders, not always dev's: the hook is now (artifact_id, branch) and the aggregator passes c.branch or Ledger.DEV. The direct engine call defaults to dev, unchanged. --- agentdescent/aggregator.py | 13 +++++++------ agentdescent/evalcache.py | 15 +++++++++------ agentdescent/evolution.py | 9 ++++++--- tests/test_evalcache.py | 31 +++++++++++++++++++++++++++++++ tests/test_multi_artifact.py | 8 +++++--- 5 files changed, 58 insertions(+), 18 deletions(-) diff --git a/agentdescent/aggregator.py b/agentdescent/aggregator.py index 27c2355c..ddc6a649 100644 --- a/agentdescent/aggregator.py +++ b/agentdescent/aggregator.py @@ -720,11 +720,12 @@ def __init__( #: budget governor), it only reports what happened. self.allocator: Optional[Any] = None #: Called after a **contract-breaking** commit lands, with the changed - #: artifact id. 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], None]] = None + #: 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 @@ -1642,6 +1643,6 @@ 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) + 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/evalcache.py b/agentdescent/evalcache.py index d2940f4e..f4c775c9 100644 --- a/agentdescent/evalcache.py +++ b/agentdescent/evalcache.py @@ -176,10 +176,6 @@ def invalidate(self, rendered: str, fingerprint: str = "") -> None: 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) - if fingerprint: - prefix = [rendered, fingerprint] - else: - prefix = [rendered] for name in os.listdir(self.directory): path = os.path.join(self.directory, name) if not os.path.isfile(path): @@ -188,9 +184,16 @@ def invalidate(self, rendered: str, fingerprint: str = "") -> None: with open(path, encoding="utf-8") as fh: payload = json.load(fh) key = payload.get("key") - if key and isinstance(key, list) and key[:len(prefix)] == prefix: + # 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): + except (OSError, ValueError, KeyError, TypeError, IndexError): continue def _path(self, key: Any) -> str: diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index c6ac698f..8cd62308 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -1886,10 +1886,13 @@ class _Engine: #: 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) -> None: + 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``: + 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 @@ -1902,7 +1905,7 @@ def invalidate_dependents(self, changed_artifact_id: str) -> None: dependents = self.contract_dependents.get(changed_artifact_id) or () if not dependents: return - snap = self.ledger.snapshot(Ledger.DEV) + snap = self.ledger.snapshot(branch) invalidate = getattr(self.runtime.cache, "invalidate", None) if not callable(invalidate): return 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 index 13295c5d..7ab10f47 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -232,7 +232,8 @@ def test_contract_breaking_with_adapters_commits_both_atomically(tmp_path): AggregatorConfig(batch_trigger=1)) agg.contract_dependents = {"artifact": ["b"]} invalidated = [] - agg.on_contract_change = lambda aid: invalidated.append(aid) + 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") @@ -255,8 +256,9 @@ def test_contract_breaking_with_adapters_commits_both_atomically(tmp_path): 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"], ( - "dependents must be re-measured after the breaking change") + 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 a3c0bc94b92dac98629327fa445ccb003b2141ea Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:41:06 +0800 Subject: [PATCH 7/7] fix(multi-artifact): attribute extra-artifact trajectories to the extra The sync worker loop's _record_rollout calls hard-coded artifact_id (the primary) and a version vector keyed by it, so an extra artifact's rollout and self-verify records were filed under the primary while their rendered text was the extra's -- the EvidenceCard's touched=[aid] pointed at trace_ids owned by the wrong artifact, and a primary/extra with identical renders could collide on the trace id. Both calls now use the target aid and {aid: mine_v}, matching the async path. Sync and async regressions run evolve/async_evolve with an extra artifact + TrajectoryStore and assert every rollout/self-verify record whose render is the extra's carries the extra's id. --- agentdescent/evolution.py | 8 ++--- tests/test_multi_artifact.py | 68 ++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index 8cd62308..33dcb067 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -3546,8 +3546,8 @@ def _propose_one(unit, task, aid: str) -> 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 @@ -3634,8 +3634,8 @@ def _propose_one(unit, task, aid: str) -> 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: diff --git a/tests/test_multi_artifact.py b/tests/test_multi_artifact.py index 7ab10f47..163f4fb2 100644 --- a/tests/test_multi_artifact.py +++ b/tests/test_multi_artifact.py @@ -533,3 +533,71 @@ def deserialize(aid, version, state): 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})}")