From e5fdb32eead3771e2417358e4222d0d03c926f3b Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Sat, 7 Nov 2026 06:33:27 +0800 Subject: [PATCH 1/2] fix(evolve): treat sync and async as two contracts (#809) Deprecate evolve(asynchronous=True) as a bridge to async_evolve: require explicit max_seconds and max_rollouts (no silent 20s default, no rounds reinterpretation), share one _POLICY_DRIVER_SUPPORT table, and document the honest dual entry. Spec derives max_rollouts at plan time when missing. --- agentdescent/evolution.py | 104 +++++++++--------- agentdescent/evolvespec.py | 17 ++- changelog.d/809-sync-async-contracts.md | 5 + docs/api.md | 4 +- docs/async.md | 57 +++++----- docs/evolution.md | 42 +++---- examples/evoskill/evoskill_skill_discovery.py | 6 +- tests/test_async_early_stop.py | 2 + tests/test_async_evolve.py | 1 + tests/test_bring_your_own_agent.py | 2 +- tests/test_budget.py | 13 +-- tests/test_checkpoint.py | 2 +- tests/test_dataset_to_skill.py | 3 +- tests/test_merge_diagnostics.py | 2 +- tests/test_progress_hooks.py | 2 +- tests/test_provider_and_split_contracts.py | 29 +++-- tests/test_sync_async_contracts.py | 79 +++++++++++++ tests/test_token_budget.py | 3 + tests/test_worker_resilience.py | 6 +- 19 files changed, 250 insertions(+), 129 deletions(-) create mode 100644 changelog.d/809-sync-async-contracts.md create mode 100644 tests/test_sync_async_contracts.py diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py index e68bd712..35fbe920 100644 --- a/agentdescent/evolution.py +++ b/agentdescent/evolution.py @@ -865,22 +865,27 @@ def _publish_stable(aggregator) -> None: pass -#: What `evolve` can honour from a `Policies` bundle today. Everything else +#: What each driver can honour from a `Policies` bundle today. Everything else #: raises rather than being accepted and ignored -- see -#: `Policies.require_supported`. The set grows as the implementations land. -_WIRED_POLICIES = ("task_sampler", "selection", "proposal", "conflict", "fusion", - "acceptance", "promotion", "staleness", "verifier", "ledger", - "aggregator_factory", "eval_cache", "sandbox_spec", "evaluator", - "executor", "screen") - -#: The same bundle, minus the one field the barrier-free loop does not read. -#: `async_evolve`'s worker calls `eng.run` directly; there is no executor seam in -#: it yet, so a supplied one was accepted and then dropped -- the exact outcome -#: `require_supported` exists to make impossible, arriving through a *shared* -#: constant rather than a missing check. It matters more since a supplied -#: executor started working on the synchronous path: flipping -#: `asynchronous=True` would otherwise silently stop honouring it. -_ASYNC_WIRED_POLICIES = tuple(p for p in _WIRED_POLICIES if p != "executor") +#: `Policies.require_supported`. One table so sync and async cannot drift +#: apart without a test noticing (#809). ``executor`` is sync-only: the +#: barrier-free worker calls ``eng.run`` directly and has no executor seam yet. +_POLICY_DRIVER_SUPPORT = { + "evolve": ( + "task_sampler", "selection", "proposal", "conflict", "fusion", + "acceptance", "promotion", "staleness", "verifier", "ledger", + "aggregator_factory", "eval_cache", "sandbox_spec", "evaluator", + "executor", "screen", + ), + "async_evolve": ( + "task_sampler", "selection", "proposal", "conflict", "fusion", + "acceptance", "promotion", "staleness", "verifier", "ledger", + "aggregator_factory", "eval_cache", "sandbox_spec", "evaluator", + "screen", + ), +} +_WIRED_POLICIES = _POLICY_DRIVER_SUPPORT["evolve"] +_ASYNC_WIRED_POLICIES = _POLICY_DRIVER_SUPPORT["async_evolve"] def _propose_via_policy(policy, settled_reader=None): @@ -3165,8 +3170,9 @@ def evolve( Name of the evolving artifact; becomes a filename, so it must match ``[A-Za-z0-9_.-]+``. rounds: - Number of round barriers to run. Under ``asynchronous=True`` this becomes - a worker-rollout budget of ``rounds * n_workers`` instead. + Number of round barriers to run. Sync-only. ``evolve(asynchronous=True)`` + no longer reinterprets this as a rollout budget (#809) — pass + ``max_rollouts=`` or call :func:`~agentdescent.async_evolve.async_evolve`. n_workers: Workers per round (``>= 1``). max_concurrency: @@ -3231,8 +3237,12 @@ def evolve( documents what it does and does not fix -- a commit landing mid-rollout still produces a stale card. asynchronous, async_ratio: - Delegate to :func:`~agentdescent.async_evolve.async_evolve` -- no round - barrier, with ``async_ratio`` as the staleness lag budget. + **Deprecated bridge** to :func:`~agentdescent.async_evolve.async_evolve` + (#809). Sync and async are two contracts, not a mode flag: prefer + calling ``async_evolve`` directly. When ``asynchronous=True``, both + ``max_seconds`` and ``max_rollouts`` are required (no silent + reinterpretation of ``rounds``, no invented 20-second budget). + ``async_ratio`` is the async lag budget. max_seconds: Wall-clock budget. ``None`` (default) means unbounded; the async path uses ``20.0`` when unset. @@ -3568,12 +3578,10 @@ def evolve( trajectory_store = _obs.trajectory_store if asynchronous: - # barrier-free mode: hand the same plug-ins to the async runtime. `rounds` - # becomes a worker-rollout budget (rounds x n_workers) alongside max_seconds. - # Two sync-only knobs have no meaning here and were previously dropped in - # silence, so `parallel=TensorParallel(...)` looked honoured while the run - # was plain DP. Say so instead: the async runtime shards round-robin itself - # and its concurrency *is* `n_workers`. + # Deprecated bridge (#809): sync and async are two contracts, not a mode + # flag. Prefer ``async_evolve`` directly. Validate budgets before the + # deprecation warning so a misconfigured call fails closed without the + # extra noise. for name, value, why in ( ("parallel", parallel, f"the async runtime shards data-parallel across its own {n_workers} " @@ -3593,35 +3601,34 @@ def evolve( f"evolve(asynchronous=True) ignores {name}=: {why}. Use the " f"synchronous path if you need it.", RuntimeWarning, stacklevel=2) - # The two above that are *not* ignored but silently REDEFINED were the - # sharper edge, because nothing said so. Flipping one boolean turned an - # unbounded run into a 20-second one, and a partial artifact with - # `error=None` and a populated `history` is indistinguishable from a - # converged one. if max_seconds is None: - warnings.warn( - "evolve(asynchronous=True) has no unbounded mode: max_seconds=None " - "becomes 20.0 seconds here, where it means 'no limit' on the " - "synchronous path. Pass max_seconds= explicitly, and check " - "result.stop_reason -- a budget expiry otherwise looks exactly " - "like convergence.", RuntimeWarning, stacklevel=2) - if rounds != 15 and max_rollouts is None: # i.e. the caller chose a value - warnings.warn( - f"evolve(asynchronous=True) has no round barrier, so rounds={rounds} " - f"is reinterpreted as a budget of {rounds * max(1, n_workers)} worker " - "rollouts, and RoundInfo.round becomes a merger-sweep index -- " - "len(result.history) is not comparable with the synchronous path. " - "Pass max_rollouts= to say the budget outright.", - RuntimeWarning, stacklevel=2) + raise ValueError( + "evolve(asynchronous=True) no longer invents max_seconds=20.0 " + "(that made an unbounded sync run look like a 20-second async " + "one). Pass max_seconds= explicitly, or call " + "async_evolve(max_seconds=...).") + if max_rollouts is None: + raise ValueError( + "evolve(asynchronous=True) no longer reinterprets " + f"rounds={rounds} as a budget of " + f"{rounds * max(1, n_workers)} worker rollouts. Pass " + "max_rollouts= for an equal-budget stop (or " + "budget=BudgetSettings(max_rollouts=...)), or call " + "async_evolve(max_iters=...).") + warnings.warn( + "evolve(asynchronous=True) is a deprecated bridge to async_evolve(): " + "sync and async are two contracts (different budgets, different " + "Policies wiring). Call async_evolve(...) directly; see docs/async.md " + "and #809.", + FutureWarning, stacklevel=2) from .async_evolve import async_evolve return async_evolve( tasks, reward, agent=agent, run=run, propose=propose, strategy=strategy, initial_state=initial_state, blast_radius=blast_radius, artifact_id=artifact_id, n_workers=n_workers, async_ratio=async_ratio, resync_on_commit=resync_on_commit, - max_seconds=20.0 if max_seconds is None else max_seconds, - max_iters=(max_rollouts if max_rollouts is not None - else rounds * max(1, n_workers)), + max_seconds=max_seconds, + max_iters=max_rollouts, max_calls=max_calls, max_tokens=max_tokens, held_out_frac=held_out_frac, repo_path=repo_path, agg_config=agg_config, staleness_policy=staleness_policy, aggregator_factory=aggregator_factory, oracle_budget=oracle_budget, @@ -3643,9 +3650,6 @@ def evolve( trajectory_store=trajectory_store, contract=contract, extra_artifacts=extra_artifacts, population_islands=population_islands, - # Bags already folded into the flat locals above; forward them so - # the async surface does not look like it dropped the new knobs, - # and a bag-only caller still works if resolve is re-run (#804). budget=budget, runtime=runtime, observability=observability) if pipelined_gate: diff --git a/agentdescent/evolvespec.py b/agentdescent/evolvespec.py index 84e4c9d7..37803ec3 100644 --- a/agentdescent/evolvespec.py +++ b/agentdescent/evolvespec.py @@ -1305,9 +1305,20 @@ def run(skill: str, task: Task) -> str: knobs.setdefault("n_workers", max(1, min(default_workers, train))) if not knobs.get("asynchronous"): knobs.setdefault("max_concurrency", knobs["n_workers"]) - elif "max_seconds" not in knobs: - knobs["max_seconds"] = 600.0 - notes.append("asynchronous=True needs max_seconds; defaulted to 600.") + else: + # Async is a second contract (#809). Plan must name budgets in async + # units — never leave rounds to be reinterpreted at evolve() time. + if "max_seconds" not in knobs: + knobs["max_seconds"] = 600.0 + notes.append("asynchronous=True needs max_seconds; defaulted to 600.") + if "max_rollouts" not in knobs: + rounds = int(knobs.get("rounds", 15)) + workers = int(knobs["n_workers"]) + knobs["max_rollouts"] = rounds * max(1, workers) + notes.append( + f"asynchronous=True needs max_rollouts; derived " + f"{knobs['max_rollouts']} from rounds*n_workers at plan time " + f"(rounds is sync-only — pass max_rollouts= to be explicit).") if "agg_config" in knobs and isinstance(knobs["agg_config"], Mapping): knobs["agg_config"] = build_agg_config(spec, **knobs["agg_config"]) elif spec.kind == "text" and (spec.agg_config or (ov and ov.agg_config)): diff --git a/changelog.d/809-sync-async-contracts.md b/changelog.d/809-sync-async-contracts.md new file mode 100644 index 00000000..a70d6d06 --- /dev/null +++ b/changelog.d/809-sync-async-contracts.md @@ -0,0 +1,5 @@ +Changed +`evolve(asynchronous=True)` is a deprecated bridge to `async_evolve`: it now +requires explicit `max_seconds` and `max_rollouts` (no silent 20s default, no +`rounds` reinterpretation) and emits `FutureWarning`. Policies wiring is one +shared `_POLICY_DRIVER_SUPPORT` table for both drivers (#809). diff --git a/docs/api.md b/docs/api.md index a4ea6b38..c8ebbb68 100644 --- a/docs/api.md +++ b/docs/api.md @@ -289,7 +289,7 @@ evolve( | `initial_state` | `Optional[Dict[str, str]]` | `None` | Seed the artifact instead of starting from `strategy.initial()`. Ignored when resuming an existing `repo_path`. | | `blast_radius` | `float` | `0.2` | Governance layer, in `[0, 1]` (see above). | | `artifact_id` | `str` | `'artifact'` | Name of the evolving artifact; becomes a filename, so it must match `[A-Za-z0-9_.-]+`. | -| `rounds` | `int` | `15` | Number of round barriers to run. Under `asynchronous=True` this becomes a worker-rollout budget of `rounds * n_workers` instead. | +| `rounds` | `int` | `15` | Number of round barriers to run. Sync-only. `evolve(asynchronous=True)` no longer reinterprets this as a rollout budget (#809) — pass `max_rollouts=` or call `async_evolve`. | | `n_workers` | `int` | `4` | Workers per round (`>= 1`). | | `max_concurrency` | `int` | `1` | How many of them actually run at once (see above). | | `refresh_interval` | `int` | `1` | How many rounds a worker keeps its ledger snapshot before taking the round's fresh one. `1` (default) is what this loop always did: every worker proposes against the current head, so a diff's staleness `eta` is **0 by construction** -- and that made `staleness_policy=` a knob with nothing to decide on this path (measured over an 8-round run: all 15 staleness decisions saw `eta=0` and returned ACCEPT, so Full, Guarded and Reflective were indistinguishable). Above `1`, workers hold a spread of versions -- the refresh is staggered by worker id -- so their diffs arrive with a spread of `eta` and the staleness policy, the `alpha` tolerances in `agg_config` and the `all-stale` outcome all become reachable synchronously. Costs no extra ledger read: a worker either adopts the snapshot the round already took, or keeps the older one it has. Ignored under `asynchronous=True`, where the lag budget is `async_ratio`. | @@ -298,7 +298,7 @@ evolve( | `patience` | `Optional[int]` | `None` | Stop after this many consecutive rounds with no improvement in held-out reward. `None` disables it. Cheap insurance for a run that plateaus below `target_reward`. | | `max_worker_errors` | `int` | `3` | How much total failure to tolerate before giving up -- and only while *no* worker has ever completed a rollout, which reads as a misconfiguration (wrong key, dead endpoint). Once any worker has succeeded the backend demonstrably works, so failures are treated as transient and the run continues on whatever evidence it did gather. Counts consecutive failed rollouts per worker on the async path (see `result.retired_workers`) and consecutive rounds in which *every* worker failed on the sync path. | | `eval_concurrency` | `int` | `8` | How many held-out tasks to score at once. Every gate goes through this -- each round's measurement and, far more often, the aggregator's per-candidate comparisons -- so it is the merge half of the run's parallelism, independent of `n_workers`. `1` restores the old sequential behaviour. | -| `asynchronous` | `bool` | `False` | Delegate to `async_evolve` -- no round barrier, with `async_ratio` as the staleness lag budget. | +| `asynchronous` | `bool` | `False` | **Deprecated bridge** to `async_evolve` (#809). Sync and async are two contracts, not a mode flag: prefer calling `async_evolve` directly. When `asynchronous=True`, both `max_seconds` and `max_rollouts` are required (no silent reinterpretation of `rounds`, no invented 20-second budget). `async_ratio` is the async lag budget. | | `async_ratio` | `int` | `3` | As `asynchronous`. | | `resync_on_commit` | `bool` | `True` | Asynchronous path only. Refresh every worker's snapshot as soon as a sweep commits, so no one *starts* a rollout against a superseded artifact. See `async_evolve`, which documents what it does and does not fix -- a commit landing mid-rollout still produces a stale card. | | `pipelined_gate` | `bool` | `False` | Under `asynchronous=True`, run a merge's **measurement** phase on its own threads instead of on the merger, so the merger goes back to draining while the gate runs. Off by default; documented in full on `async_evolve`, which implements it. Warns and does nothing on the synchronous path, where the round barrier idles every worker for the whole merge regardless. | diff --git a/docs/async.md b/docs/async.md index a0b4124a..bf432287 100644 --- a/docs/async.md +++ b/docs/async.md @@ -10,8 +10,8 @@ a run reproducible and easy to reason about — and it is also what makes the wh round wait for its slowest rollout. ```python -result = evolve(tasks, reward, agent=agent, - asynchronous=True, async_ratio=3, max_seconds=120) +result = async_evolve(tasks, reward, agent=agent, + async_ratio=3, max_seconds=120, max_iters=200) ``` Two stages — *rollout + propose* (the workers) and *aggregate + commit* (the @@ -44,41 +44,46 @@ budget discards everything (`outcomes()` fills with `all-stale` and nothing commits); a tight lag budget re-introduces the waiting you removed the barrier to avoid. `result.forced_refreshes` is the mismatch showing itself. -## `evolve(asynchronous=True)` vs `async_evolve()` +## `evolve` vs `async_evolve` — two contracts (#809) -They are the same engine; the first delegates. The wrapper exists so that -switching costs one argument, but three of `evolve()`'s parameters have no -meaning without a barrier and it **says so** rather than dropping them silently: +They share a loop shape, but they are **not** one function with a mode flag. +`async_evolve` is the barrier-free entry. `evolve(asynchronous=True)` remains +only as a **deprecated bridge** (emits `FutureWarning`): it forwards to +`async_evolve` and refuses the old silent redefinitions. -| argument | what happens under `asynchronous=True` | +| argument | under `evolve(asynchronous=True)` | |---|---| -| `parallel=` | ignored — the async runtime shards data-parallel across its own workers | -| `max_concurrency=` | ignored — concurrency *is* `n_workers` | -| `round_timeout=` | ignored — there is no barrier to bound; use the backend's own timeout | -| `rounds=` | **reinterpreted** as a budget of `rounds × n_workers` worker rollouts (silent, and exact, once `max_rollouts=` says the budget outright) | -| `max_seconds=None` | **becomes 20.0 seconds**, where it means "no limit" on the sync path | - -Each of those emits a `RuntimeWarning`. The last two are the sharp ones: flipping -one boolean turns an unbounded run into a 20-second one, and a partial artifact -with `error=None` and a populated `history` is indistinguishable from a converged -one. **Check `result.stop_reason`** — `"target_reward"` is convergence, -`"max_seconds"` / `"max_iters"` / `"max_rollouts"` / `"max_calls"` is a budget -expiry. +| `parallel=` / `max_concurrency=` / `round_timeout=` / `refresh_interval=` | ignored with `RuntimeWarning` — sync-only | +| `rounds=` | **not** an async budget; omitting `max_rollouts=` raises `ValueError` | +| `max_seconds=None` | **raises** — no invented 20-second budget | +| `max_rollouts=` / `budget=BudgetSettings(...)` | required equal-budget stop (maps to async `max_iters`) | +| Policies `executor=` | refused on async (`_POLICY_DRIVER_SUPPORT`) — sync-only seam | -Say the rollout budget outright rather than deriving it from `rounds`. Either -spelling works and both silence the reinterpretation warning: +Prefer the honest call: ```python -from agentdescent import async_evolve, evolve - -evolve(tasks, reward, agent=agent, asynchronous=True, - n_workers=6, max_seconds=120, max_rollouts=200, max_calls=400) +from agentdescent import async_evolve, BudgetSettings async_evolve(tasks, reward, agent=agent, n_workers=6, async_ratio=3, max_seconds=120, max_iters=200, max_calls=400) +# or +async_evolve(tasks, reward, agent=agent, + budget=BudgetSettings(max_seconds=120, max_rollouts=200, max_calls=400), + n_workers=6, async_ratio=3) ``` +The deprecated bridge still works when both budgets are explicit: + +```python +evolve(tasks, reward, agent=agent, asynchronous=True, + n_workers=6, max_seconds=120, max_rollouts=200, max_calls=400) +``` + +**Check `result.stop_reason`** — `"target_reward"` is convergence; +`"max_seconds"` / `"max_iters"` / `"max_rollouts"` / `"max_calls"` is a budget +expiry. Cross-link: settings bags in #804; population on both drivers in #806. + `max_calls` is the second half of an equal-budget comparison: two configurations matched on rollouts still differ in model spend whenever one asks for more proposals per rollout, and a rollout that solves its task never proposes at all — @@ -187,7 +192,7 @@ print(stats.rollouts, stats.commits, stats.discarded_stale, | you want | use | |---|---| | reproducibility, a clean per-round trace, a paper table | synchronous (`max_concurrency=n_workers` for the speedup) | -| maximum throughput, long or uneven rollouts | `asynchronous=True` | +| maximum throughput, long or uneven rollouts | `async_evolve(...)` | | both, to compare | run each and read `result.history` — but see the reinterpretation table above before comparing lengths | `len(result.history)` is **not** comparable across the two: on the async path a diff --git a/docs/evolution.md b/docs/evolution.md index c4e143ef..e079faf9 100644 --- a/docs/evolution.md +++ b/docs/evolution.md @@ -68,7 +68,7 @@ result = evolve( propose=reflector(claude(model="claude-haiku-4-5")), # who reflects strategy=SingleSlot(initial_value="Answer concisely."), # what evolves n_workers=4, max_concurrency=4, # ...in parallel - # asynchronous=True, # ...or barrier-free + # async_evolve(...), # ...or barrier-free ) print(result.rendered) # the evolved system prompt ``` @@ -649,7 +649,8 @@ limit, credit exhaustion) — progress isn't lost. # | "max_rollouts" | "max_calls" | "error" ``` - This matters most under `asynchronous=True`, where **`max_seconds=None` means + This matters most under the deprecated `asynchronous=True` bridge, where + **`max_seconds` is now required** (#809); historically `max_seconds=None` meant 20 seconds**, not "unbounded" as it does on the synchronous path — flipping one boolean could truncate a run and the result looked converged. Both that and `rounds` changing meaning (it becomes a `rounds × n_workers` rollout budget, @@ -836,13 +837,13 @@ evolve(tasks, reward, agent=agent, n_workers=4, max_concurrency=4) # 4 workers need a rollout contract exposing its turns (see [duration-aware scheduling](duration-scheduling.md)). -* **Across rounds — `asynchronous=True` (barrier-free).** Removing the round - barrier entirely is [`async_evolve()`](#the-barrier-free-runtime-async_evolve), - reachable as `evolve(asynchronous=True, async_ratio=…)`. It takes the **same** - plug-ins, so every example runs async with a `--async` flag. +* **Across rounds — barrier-free `async_evolve`.** Removing the round barrier + entirely is [`async_evolve()`](#the-barrier-free-runtime-async_evolve). It takes + the **same** plug-ins, so every example runs async with a `--async` flag. + `evolve(asynchronous=True)` remains only as a deprecated bridge (#809). ```python -evolve(tasks, reward, agent=agent, asynchronous=True, async_ratio=3, max_seconds=30) +async_evolve(tasks, reward, agent=agent, async_ratio=3, max_seconds=30, max_iters=120) ``` ## The barrier-free runtime: `async_evolve()` @@ -861,17 +862,15 @@ result = async_evolve(tasks, reward, agent=agent, staleness_policy=get_policy("reflective")) ``` -Reach it via `evolve(asynchronous=True)` or directly. Small `async_ratio` → -near-synchronous, few stale diffs; large → highly asynchronous, many stale diffs -for the policy to rebase or discard. - -!!! warning "Five arguments change meaning under `asynchronous=True`" - Three are ignored (`parallel`, `max_concurrency`, `round_timeout`) and two are - **redefined**: `rounds` becomes a budget of `rounds × n_workers` rollouts, and - `max_seconds=None` becomes 20 seconds where it meant "no limit". Each warns, - but check `result.stop_reason` — a budget expiry otherwise looks exactly like - convergence. The full table, the pipeline's cold-start and backpressure - behaviour, and what only the async path can report are on +Call `async_evolve` directly. Small `async_ratio` → near-synchronous, few stale +diffs; large → highly asynchronous, many stale diffs for the policy to rebase or +discard. + +!!! warning "`evolve(asynchronous=True)` is a deprecated bridge (#809)" + Sync and async are two contracts. The bool still forwards to `async_evolve` + for one migration window, but it **requires** explicit `max_seconds` and + `max_rollouts` (no invented 20-second budget, no `rounds` reinterpretation) + and emits `FutureWarning`. Prefer `async_evolve(...)`. Full table on [the async page](async.md). ### More async / parallel recipes @@ -883,7 +882,7 @@ The two levers compose; pick per workload: evolve(tasks, reward, agent=agent, n_workers=8, max_concurrency=8, rounds=10) # 2. Barrier-free async, time-bounded: run for 20 min, keep the best head so far. -evolve(tasks, reward, agent=agent, asynchronous=True, async_ratio=3, max_seconds=1200) +async_evolve(tasks, reward, agent=agent, async_ratio=3, max_seconds=1200) # 3. Async to a target: stop as soon as held-out reward crosses a bar. async_evolve(tasks, reward, agent=agent, n_workers=6, target_reward=0.85) @@ -893,8 +892,9 @@ async_evolve(tasks, reward, agent=agent, n_workers=4, max_iters=200) # 5. Faithful port on the async path: skip the per-trajectory re-run, score the # candidate on held-out only (see EvoSkill), with concurrent held-out eval. -evolve(tasks, reward, run=run, propose=propose, strategy=strat, - aggregator_factory=factory, asynchronous=True, self_verify=False) +async_evolve(tasks, reward, run=run, propose=propose, strategy=strat, + aggregator_factory=factory, self_verify=False, max_seconds=600, + max_iters=200) # 6. Highly-async, staleness-heavy: large lag budget + reflective rebase-and-verify. async_evolve(tasks, reward, agent=agent, n_workers=8, async_ratio=8, diff --git a/examples/evoskill/evoskill_skill_discovery.py b/examples/evoskill/evoskill_skill_discovery.py index 16427394..fb4e311a 100644 --- a/examples/evoskill/evoskill_skill_discovery.py +++ b/examples/evoskill/evoskill_skill_discovery.py @@ -446,6 +446,10 @@ def factory(ledger, verifier, audit, config, policy): # still bounded by the data, because a worker with an empty shard is not a # worker. workers = min(max_workers, len(train)) + # #809: async no longer reinterprets rounds; name the rollout budget. + async_rollouts = max_rollouts + if asynchronous and async_rollouts is None: + async_rollouts = iterations * max(1, workers) result = evolve(tasks, reward, run=run, propose=make_propose(ctx, complete, propose_and_generate), strategy=SkillLibraryTree(), blast_radius=0.2, artifact_id="skill_library", rounds=iterations, @@ -465,7 +469,7 @@ def factory(ledger, verifier, audit, config, policy): # committed meanwhile and the rebase holds. staleness_policy=get_policy(staleness), aggregator_factory=factory, verbose=verbose, - max_rollouts=max_rollouts) + max_rollouts=async_rollouts) if verbose: report_engine(result) diff --git a/tests/test_async_early_stop.py b/tests/test_async_early_stop.py index 965cb449..6a61d2e3 100644 --- a/tests/test_async_early_stop.py +++ b/tests/test_async_early_stop.py @@ -16,6 +16,8 @@ def _flat_run(**kw): """A run that can never improve -- so only a stop rule ends it early.""" + kw.setdefault("max_rollouts", 400) + kw.setdefault("max_seconds", 20.0) return evolve(TASKS, lambda t, o: 0.5, run=lambda a, t: "x", propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), asynchronous=True, n_workers=2, rounds=200, **kw) diff --git a/tests/test_async_evolve.py b/tests/test_async_evolve.py index 32e0df18..070ab7fc 100644 --- a/tests/test_async_evolve.py +++ b/tests/test_async_evolve.py @@ -62,6 +62,7 @@ def test_async_evolve_stops_at_max_iters(): def test_evolve_asynchronous_flag_delegates_to_async(): r = evolve(_tasks(), REWARD, agent=_Composer(), strategy=AppendRules(), asynchronous=True, n_workers=4, rounds=12, async_ratio=3, max_seconds=8.0, + max_rollouts=48, held_out_frac=0.5) assert r.final_reward >= r.history[0].held_out_reward diff --git a/tests/test_bring_your_own_agent.py b/tests/test_bring_your_own_agent.py index f6ee1b95..ad839fb2 100644 --- a/tests/test_bring_your_own_agent.py +++ b/tests/test_bring_your_own_agent.py @@ -42,7 +42,7 @@ def test_the_same_agent_evolves_asynchronously(): def test_evolve_switches_to_async_with_one_flag(): res = evolve(TASKS, REWARD, run=RUN, propose=REFLECT(), strategy=SingleSlot(), rounds=4, n_workers=3, - asynchronous=True, async_ratio=3, max_seconds=4.0) + asynchronous=True, async_ratio=3, max_seconds=4.0, max_rollouts=24) assert res.final_reward == 1.0 diff --git a/tests/test_budget.py b/tests/test_budget.py index 24005349..d6ae4cec 100644 --- a/tests/test_budget.py +++ b/tests/test_budget.py @@ -132,19 +132,16 @@ def test_async_takes_the_same_two_budgets(): assert res.error is None -def test_async_max_rollouts_silences_the_rounds_reinterpretation_warning(): - """`rounds` on the async path is a rollout budget in disguise; saying the - budget outright is the way to stop guessing at it.""" +def test_async_bridge_requires_max_rollouts_instead_of_reinterpreting_rounds(): + """#809: rounds is no longer an async budget behind the bridge.""" + import pytest import warnings - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") + with pytest.raises(ValueError, match="max_rollouts"): _evolve(asynchronous=True, max_seconds=5.0, rounds=3, max_concurrency=1) - assert any("reinterpreted as a budget" in str(w.message) for w in caught) with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") _evolve(asynchronous=True, max_seconds=5.0, rounds=3, max_rollouts=12, max_concurrency=1) - assert not any("reinterpreted as a budget" in str(w.message) for w in caught), \ - "max_rollouts= makes the budget explicit, so there is nothing to warn about" + assert not any("reinterpreted as a budget" in str(w.message) for w in caught) diff --git a/tests/test_checkpoint.py b/tests/test_checkpoint.py index fae57963..a236055e 100644 --- a/tests/test_checkpoint.py +++ b/tests/test_checkpoint.py @@ -564,7 +564,7 @@ def test_async_path_checkpoints_too(tmp_path): evolve(tasks, lambda t, o: 0.5, run=lambda r, t: t.id, checkpointing=True, propose=lambda r, t, o, rew: f"change-{t.id}", rounds=2, n_workers=2, repo_path=repo, - max_seconds=30.0, asynchronous=True) + max_seconds=30.0, asynchronous=True, max_rollouts=8) payload = load_checkpoint(repo) assert payload is not None, "the async path did not write a checkpoint" diff --git a/tests/test_dataset_to_skill.py b/tests/test_dataset_to_skill.py index cc57b4fa..0f66d101 100644 --- a/tests/test_dataset_to_skill.py +++ b/tests/test_dataset_to_skill.py @@ -68,7 +68,8 @@ def test_the_async_path_takes_the_same_pieces(): import warnings with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") - res = _evolve(asynchronous=True, max_seconds=5.0, max_concurrency=1) + res = _evolve(asynchronous=True, max_seconds=5.0, max_concurrency=1, + max_rollouts=20) assert isinstance(res, EvolutionResult) assert not [x for x in w if "ignores max_concurrency" in str(x.message)] diff --git a/tests/test_merge_diagnostics.py b/tests/test_merge_diagnostics.py index df2c3c7e..29d00c7d 100644 --- a/tests/test_merge_diagnostics.py +++ b/tests/test_merge_diagnostics.py @@ -50,7 +50,7 @@ def test_outcomes_survive_save_and_load(): def test_the_async_path_reports_them_too(): - res = _run(n_workers=2, asynchronous=True, max_seconds=4.0) + res = _run(n_workers=2, asynchronous=True, max_seconds=4.0, max_rollouts=8) assert res.outcomes() diff --git a/tests/test_progress_hooks.py b/tests/test_progress_hooks.py index 5a5a6185..daabaeb4 100644 --- a/tests/test_progress_hooks.py +++ b/tests/test_progress_hooks.py @@ -67,7 +67,7 @@ def test_async_on_round_fires_per_sweep(): def test_evolve_forwards_on_round_to_the_async_path(): seen = [] evolve(_tasks(), REWARD, agent=_Agent(), strategy=AppendRules(), - asynchronous=True, n_workers=2, max_seconds=3.0, rounds=3, + asynchronous=True, n_workers=2, max_seconds=3.0, rounds=3, max_rollouts=6, on_round=seen.append) assert seen, "on_round must survive the delegation to async_evolve" diff --git a/tests/test_provider_and_split_contracts.py b/tests/test_provider_and_split_contracts.py index 5dfe42d6..3a9bb95a 100644 --- a/tests/test_provider_and_split_contracts.py +++ b/tests/test_provider_and_split_contracts.py @@ -3,7 +3,7 @@ * a provider that answered `content: null` broke the one contract in the package (`Completion` is prompt -> str) and the failure was then *retried as a backend transient*; -* flipping `asynchronous=True` turned an unbounded run into a 20-second one, and +* flipping `asynchronous=True` used to turn an unbounded run into a 20-second one, and the truncated result was indistinguishable from a converged one; * the train/held-out split is positional, so grouped data holds out one end of the file and every gate in the run is then measured against a different @@ -125,21 +125,26 @@ def _quick(**kw): n_workers=2, held_out_frac=0.5, **kw) -def test_async_says_that_an_unset_max_seconds_becomes_twenty(): - """`None` means "no limit" on one path and "20 seconds" on the other.""" - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - _quick(rounds=1, asynchronous=True) - assert any("20.0 seconds" in str(w.message) for w in caught), \ - "flipping one boolean silently bounded the run" +def test_async_bridge_refuses_an_unset_max_seconds(): + """#809: no invented 20-second budget behind asynchronous=True.""" + import pytest + with pytest.raises(ValueError, match="max_seconds"): + _quick(rounds=1, asynchronous=True, max_rollouts=4) + + +def test_async_bridge_refuses_rounds_as_a_rollout_budget(): + """#809: rounds is sync-only; max_rollouts must be named.""" + import pytest + with pytest.raises(ValueError, match="max_rollouts"): + _quick(rounds=3, asynchronous=True, max_seconds=1.0) -def test_async_says_that_rounds_changes_meaning(): +def test_async_bridge_warns_that_it_is_deprecated(): with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") - _quick(rounds=3, asynchronous=True, max_seconds=1.0) - assert any("rollouts" in str(w.message) and "rounds=3" in str(w.message) - for w in caught) + _quick(rounds=1, asynchronous=True, max_seconds=2.0, max_rollouts=4) + assert any(issubclass(w.category, FutureWarning) + and "async_evolve" in str(w.message) for w in caught) def test_stop_reason_tells_convergence_from_a_spent_budget(): diff --git a/tests/test_sync_async_contracts.py b/tests/test_sync_async_contracts.py new file mode 100644 index 00000000..3e4021e2 --- /dev/null +++ b/tests/test_sync_async_contracts.py @@ -0,0 +1,79 @@ +"""Sync and async are two contracts behind one deprecated bridge (#809).""" +from __future__ import annotations + +import inspect +import warnings + +import pytest + +from agentdescent.async_evolve import async_evolve +from agentdescent.evolution import ( + SingleSlot, + Task, + _ASYNC_WIRED_POLICIES, + _POLICY_DRIVER_SUPPORT, + _WIRED_POLICIES, + evolve, +) + + +TASKS = [Task(id=str(i), prompt="p", meta={"gold": "x"}) for i in range(8)] + + +def test_policy_driver_support_table_is_the_single_source(): + assert _WIRED_POLICIES == _POLICY_DRIVER_SUPPORT["evolve"] + assert _ASYNC_WIRED_POLICIES == _POLICY_DRIVER_SUPPORT["async_evolve"] + assert "executor" in _WIRED_POLICIES + assert "executor" not in _ASYNC_WIRED_POLICIES + # async is sync minus executor only — any further drift fails here. + assert set(_WIRED_POLICIES) - set(_ASYNC_WIRED_POLICIES) == {"executor"} + assert set(_ASYNC_WIRED_POLICIES) <= set(_WIRED_POLICIES) + + +def test_async_evolve_imports_the_shared_async_wired_set(): + import importlib + # ``import agentdescent.async_evolve`` can resolve to the *function* + # re-exported on the package; load the module object explicitly. + mod = importlib.import_module("agentdescent.async_evolve") + assert tuple(mod._ASYNC_WIRED_POLICIES) == tuple(_ASYNC_WIRED_POLICIES) + + +def test_bridge_requires_explicit_budgets_and_warns(): + with pytest.raises(ValueError, match="max_seconds"): + evolve(TASKS, lambda t, o: 0.5, run=lambda a, t: "x", + propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), + asynchronous=True, max_rollouts=4) + with pytest.raises(ValueError, match="max_rollouts"): + evolve(TASKS, lambda t, o: 0.5, run=lambda a, t: "x", + propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), + asynchronous=True, max_seconds=2.0) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + res = evolve( + TASKS, lambda t, o: 0.5, run=lambda a, t: "x", + propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), + asynchronous=True, n_workers=1, max_seconds=5.0, max_rollouts=4, + self_verify=False, + ) + assert res.error is None + assert any(issubclass(w.category, FutureWarning) + and "async_evolve" in str(w.message) for w in caught) + + +def test_budget_settings_work_on_both_drivers(): + from agentdescent import BudgetSettings + bag = BudgetSettings(max_seconds=5.0, max_rollouts=4) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + sync = evolve( + TASKS, lambda t, o: 0.5, run=lambda a, t: "x", + propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), + budget=bag, rounds=2, n_workers=1, self_verify=False, + ) + async_res = async_evolve( + TASKS, lambda t, o: 0.5, run=lambda a, t: "x", + propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), + budget=bag, n_workers=1, self_verify=False, + ) + assert sync.error is None and async_res.error is None + assert "max_iters" in inspect.signature(async_evolve).parameters diff --git a/tests/test_token_budget.py b/tests/test_token_budget.py index cfa69270..795fe7db 100644 --- a/tests/test_token_budget.py +++ b/tests/test_token_budget.py @@ -72,6 +72,7 @@ def _run_evolve(*, max_tokens=None, rounds=3, asynchronous=False, **kw): ) if asynchronous: defaults["max_seconds"] = 30 + defaults.setdefault("max_rollouts", max(8, rounds * 4)) defaults.update(kw) with warnings.catch_warnings(): warnings.simplefilter("ignore") @@ -294,6 +295,7 @@ def test_async_governor_degrades_self_verify(): max_tokens=600, asynchronous=True, max_seconds=30, + max_rollouts=40, usage=agent.usage, ) assert result.stop_reason == "max_tokens" @@ -319,6 +321,7 @@ def test_async_governor_fusion_degrades(): fusion_tournament=True, asynchronous=True, max_seconds=30, + max_rollouts=40, usage=agent.usage, ) # The run should have stopped due to the token budget. diff --git a/tests/test_worker_resilience.py b/tests/test_worker_resilience.py index 65c92add..3ff5be27 100644 --- a/tests/test_worker_resilience.py +++ b/tests/test_worker_resilience.py @@ -20,6 +20,8 @@ def _evolve(run, **kw): + kw.setdefault("max_seconds", 20.0) + kw.setdefault("max_rollouts", 120) return evolve(TASKS, lambda t, o: 1.0 if o == t.meta["gold"] else 0.0, run=run, propose=lambda *a: None, strategy=SingleSlot(initial_value="v"), asynchronous=True, @@ -126,7 +128,8 @@ def reward(task, output): warnings.simplefilter("ignore") res = evolve(TASKS, reward, run=lambda rendered, t: "wrong", propose=propose, strategy=SingleSlot(initial_value="v"), asynchronous=True, - n_workers=2, rounds=30, held_out_frac=0.5, max_seconds=15.0) + n_workers=2, rounds=30, held_out_frac=0.5, max_seconds=15.0, + max_rollouts=60) # Non-vacuity: the merger must actually have hit the outage and got past it, # otherwise this passes without ever exercising the recovery path. assert next(merger_calls) > 4, "the merger never reached the outage" @@ -149,6 +152,7 @@ def step(self): raise ContractError("aggregator is broken") propose=lambda *a: "tweak", strategy=SingleSlot(initial_value="v"), asynchronous=True, n_workers=2, rounds=20, held_out_frac=0.5, max_seconds=10.0, + max_rollouts=40, aggregator_factory=lambda *a, **kw: BadAggregator()) From b840cc1e8e6b0e93ca5e2793ef89f47da2a72a98 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Sat, 7 Nov 2026 07:02:07 +0800 Subject: [PATCH 2/2] fix(tests): name max_rollouts for evolve(asynchronous=True) callers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #809/#817 bridge fails closed without explicit max_rollouts. Point the bench equal-budget matrix and fault_matrix async engine at rounds×workers so CI matches the documented contract. --- bench/run.py | 10 ++++++++-- tests/test_fault_matrix.py | 4 +++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/bench/run.py b/bench/run.py index f822ea25..8f37a594 100644 --- a/bench/run.py +++ b/bench/run.py @@ -102,15 +102,21 @@ def _propose(rendered: str, task: Task, output: str, reward: float) -> Optional[ def workload(config: Config, seed: int) -> EvolutionResult: """One run. Only `config` varies; data, actors and semantics do not.""" train, _test = _dataset(seed) + rounds = 30 + # #809: evolve(asynchronous=True) no longer reinterprets rounds as a + # rollout budget. Name the equal-budget stop explicitly (rounds × workers). + async_kw = {} + if config.asynchronous: + async_kw["max_rollouts"] = rounds * max(1, config.n_workers) return evolve( train, _reward, run=_run, propose=_propose, strategy=AppendRules(), - rounds=30, n_workers=config.n_workers, + rounds=rounds, n_workers=config.n_workers, max_concurrency=config.n_workers, eval_concurrency=config.eval_concurrency, asynchronous=config.asynchronous, async_ratio=config.async_ratio, resync_on_commit=config.resync_on_commit, staleness_policy=get_policy(config.staleness), - held_out_frac=0.4, seed=seed, max_seconds=60.0) + held_out_frac=0.4, seed=seed, max_seconds=60.0, **async_kw) def test_quality(seed: int) -> Callable[[EvolutionResult], float]: diff --git a/tests/test_fault_matrix.py b/tests/test_fault_matrix.py index 4131193a..42623b7f 100644 --- a/tests/test_fault_matrix.py +++ b/tests/test_fault_matrix.py @@ -22,7 +22,9 @@ ENGINES = [ pytest.param({"n_workers": 3, "max_concurrency": 3}, id="sync"), - pytest.param({"n_workers": 3, "asynchronous": True, "max_seconds": 8.0}, + # #809: bridge requires explicit max_rollouts (equal-budget: rounds×workers). + pytest.param({"n_workers": 3, "asynchronous": True, "max_seconds": 8.0, + "max_rollouts": 36}, id="async"), ]