From eec227240fd8ea5c8dafb52e08ae27100c32e04f Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:00:36 +0800
Subject: [PATCH 1/7] =?UTF-8?q?feat(async):=20L-traj=20resume=20=E2=80=94?=
 =?UTF-8?q?=20a=20straggler=20is=20re-run=20against=20a=20newer=20head,=20?=
 =?UTF-8?q?not=20just=20counted?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The async runtime abandons a rollout that overruns its predicted cost
and counts it (result.stragglers). concepts.md calls the resume half
not implemented: run(rendered, task) -> output is opaque, so there is
no turn-level state to checkpoint.

What *can* be saved is the task and the version it measured. With a
shared ResumeQueue passed as resume_queue=, a straggler is recorded
(task, version_at_checkpoint) instead of dropped, and the next idle
worker whose shard owns that task re-runs it against the current head.
The straggler measured version N, the re-run measures N+k — a free
cross-version A/B signal, which is the design's L-traj note.

ResumeQueue.pop_for(task_ids) hands a queued resume only to a worker
whose shard owns it, so no two workers re-run the same task.
result.resumed reports the re-runs. resume_queue=None (default) is the
old behaviour exactly.

5 tests in tests/test_ltraj_resume.py. Async + genesis regression exit 0.
---
 CHANGELOG.md                 |  15 +++++
 agentdescent/async_evolve.py |  45 +++++++++++++--
 agentdescent/evolution.py    |   5 ++
 agentdescent/scheduler.py    |  16 +++++-
 docs/api.md                  |   7 +++
 tests/test_ltraj_resume.py   | 107 +++++++++++++++++++++++++++++++++++
 6 files changed, 188 insertions(+), 7 deletions(-)
 create mode 100644 tests/test_ltraj_resume.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 70e9e4be..bd7a60eb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -97,6 +97,21 @@ value-directed evaluation or Genesis blind-mode proposals.
 
   Wired end to end: the worker asks the governor `candidate_worth(ctx)` before the self-verify rollout; the aggregator reports **every** measured decision exactly once (`observe(ctx, committed)`), with oracle rejection and acceptance rejection as negative labels — a model fit on commits alone never leaves warm-up. **Every commit is attributed to a worker's actual pre-spend vector**: a fused candidate keeps its first contributor's context, and a tournament-selected single proposal is matched to its own winning card — never a losing first entrant, never a fused union size. The feature vector is **scale-normalised**: the raw-op diff size (1..100+) would swamp the advantage (a z-score) and blast radius in a raw logistic fit, so `size` is mapped through `log1p` to `[0, 1]` — verified by a regression where a high-advantage, large-diff candidate still predicts commit. `allocator=None` (the default) is the old behaviour byte for byte — the clock thresholds remain the *ceiling* the allocator may never exceed. 25 tests in `tests/test_allocator.py` (incl. negative-outcome, sync/async feature-parity, end-to-end fused + single-winner regressions, the A/B arm wiring, and the size-scale regression). **`bench/ab_allocator.py`** is the equal-budget A/B (off vs on, real datasets HotpotQA/finer, per-seed quality + measured tokens + a fired check that distinguishes "the allocator helped" from "it never ran"); exported from the package root; `docs/api.md` regenerated.
 
+- **`async_evolve(resume_queue=...)` — the L-traj resume half, the straggler
+  re-run.** The async runtime abandons a rollout that overruns its predicted
+  cost and counts it (`result.stragglers`). With a shared
+  :class:`~agentdescent.scheduler.ResumeQueue` passed in, it now *records* the
+  abandoned task and the version it measured instead of dropping it; the next
+  idle worker whose shard owns that task re-runs it against the **current**
+  head. A straggler measured version N, its re-run measures version N+k, so the
+  pair is a free cross-version A/B signal — the design's L-traj note made
+  behaviour.
+
+  `ResumeQueue.pop_for(task_ids)` hands a queued resume only to a worker whose
+  shard owns the task, so no two workers re-run the same one. `result.resumed`
+  reports the re-runs. `resume_queue=None` (the default) is the old behaviour
+  exactly: stragglers are counted and dropped.
+
 - **`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/async_evolve.py b/agentdescent/async_evolve.py
index a7490644..28d6ee2f 100644
--- a/agentdescent/async_evolve.py
+++ b/agentdescent/async_evolve.py
@@ -52,7 +52,7 @@
 from .evolvable import ContractError, EvidenceCard, vv_staleness
 from .ledger import Ledger, LedgerFailure
 from .sampling import RoundRobin, TaskSampler
-from .scheduler import DurationEstimator
+from .scheduler import DurationEstimator, ResumeItem, ResumeQueue
 from .selection import SingleHead
 from .pipeline import EarlyStop, FirstError, StallGuard, WorkerHealth, describe
 from .staleness import StaleAction, StalenessPolicy, get_policy
@@ -126,6 +126,14 @@ def async_evolve(
     stall_patience: int = 50,
     duration_estimator: Optional["DurationEstimator"] = None,
     straggler_factor: float = 3.0,
+    #: L-traj resume: a shared queue of abandoned stragglers, each carrying the
+    #: task and the version it was rolled out against. A worker that overruns
+    #: its predicted cost is abandoned -- but with a queue present its task is
+    #: recorded instead of just counted, and the next idle worker re-runs it
+    #: against the **latest** head. The straggler measured an old version, the
+    #: re-run measures the new one, so the two are a free cross-version A/B
+    #: signal. ``None`` keeps the old behaviour (count the straggler, drop it).
+    resume_queue: Optional["ResumeQueue"] = None,
     task_sampler: Optional["TaskSampler"] = None,
     on_round: Optional[Callable[[RoundInfo], None]] = None,
     stop_when: Optional[Callable[[RoundInfo], bool]] = None,
@@ -514,6 +522,7 @@ def _published_head() -> int:
     commit_epoch = [0]
     forced_refreshes = [0]
     stragglers = [0]
+    resumed_count = [0]      # stragglers re-run against a newer head (L-traj)
     estimator = duration_estimator
     n_live = sum(1 for s in shards if s)      # workers that will actually start
     live = [n_live]                           # workers still running
@@ -671,8 +680,23 @@ def _worker(wid: int, shard: List[Task]) -> None:
                 if forced:
                     with counter_lock:
                         forced_refreshes[0] += 1
-            task = by_shard_id[sampler.pick(shard_ids, i)]
-            i += 1
+            # A straggler abandoned by another worker is re-run here before a
+            # fresh task is picked -- it measured an older head, and re-running
+            # it against the current one is the L-traj A/B signal. `pop_for`
+            # only returns a task in this worker's own shard, so no two workers
+            # re-run the same task.
+            if resume_queue is not None:
+                resumed = resume_queue.pop_for(shard_ids)
+                if resumed is not None:
+                    task = by_shard_id[resumed.task_id]
+                    with counter_lock:
+                        resumed_count[0] += 1
+                else:
+                    task = by_shard_id[sampler.pick(shard_ids, i)]
+                    i += 1
+            else:
+                task = by_shard_id[sampler.pick(shard_ids, i)]
+                i += 1
             # Duration-aware straggler detection (design spec 5.1, L-traj). It
             # existed only in the reference runtime, which accepts nothing but the
             # synthetic router domain -- so the whole L-traj mechanism was
@@ -794,12 +818,20 @@ def _worker(wid: int, shard: List[Task]) -> None:
             elapsed = time.time() - t_start
             if estimator is not None:
                 estimator.observe(cost, elapsed)
-                # A rollout that overran its own estimate is a straggler. Counted,
-                # not resumed: `run(rendered, task) -> output` is opaque, so there
-                # is no continuation state to check point (see concepts.md L-traj).
+                # A rollout that overran its own estimate is a straggler. The
+                # reference runtime checkpoints partial turns here; this port's
+                # `run(rendered, task) -> output` is opaque, so there is no
+                # turn-level state to save. What *can* be saved is the task and
+                # the version it measured -- with a resume queue present, that is
+                # re-run against the latest head by the next idle worker, which
+                # turns the abandoned rollout into a free cross-version signal.
                 if predicted > 0 and elapsed > straggler_factor * predicted:
                     with counter_lock:
                         stragglers[0] += 1
+                    if resume_queue is not None:
+                        resume_queue.push(ResumeItem(
+                            task_id=task.id, turn=0, conversation=[],
+                            version_at_checkpoint={eng.artifact_id: base_v}))
             consecutive, warned = 0, False                    # a clean rollout resets
             health.record_success()
             with counter_lock:
@@ -1225,6 +1257,7 @@ def _merger() -> None:
                              stop_reason="error" if run_error else stop_reason[0],
                              forced_refreshes=forced_refreshes[0],
                              stragglers=stragglers[0],
+                             resumed=resumed_count[0],
                              fusion_trials=_fusion_trials(eng.aggregator),
                              **_cost_fields(eng.meter))
     if gate_pool is not None:
diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py
index 06b7f0d8..0858b3d2 100644
--- a/agentdescent/evolution.py
+++ b/agentdescent/evolution.py
@@ -1299,6 +1299,11 @@ class EvolutionResult:
     #: (async path, and only when a ``duration_estimator`` was given). The design's
     #: L-traj signal; detection only, nothing is resumed.
     stragglers: int = 0
+    #: Stragglers that were re-run against a newer head (async path, and only
+    #: when both a ``duration_estimator`` and a ``resume_queue`` were given).
+    #: A straggler measured the old version; its re-run measures the current
+    #: one, so a non-zero pair is a free cross-version A/B signal (L-traj).
+    resumed: int = 0
     #: Workers that gave up after repeated backend failures (async path only). A
     #: run can finish *cleanly* at a fraction of its requested concurrency, so
     #: `error` stays `None` while throughput quietly drops -- check this to tell a
diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py
index de078d6d..ca3f92bd 100644
--- a/agentdescent/scheduler.py
+++ b/agentdescent/scheduler.py
@@ -272,8 +272,8 @@ class ResumeItem:
     task_id: str
     turn: int
     conversation: List[Any]
-    external_handle: Optional[str]  # e.g. an HPC job id
     version_at_checkpoint: Dict[str, int]
+    external_handle: Optional[str] = None   # e.g. an HPC job id
 
 
 class ResumeQueue:
@@ -311,6 +311,20 @@ def pop(self) -> Optional[ResumeItem]:
         with self._lock:
             return self._items.pop(0) if self._items else None
 
+    def pop_for(self, task_ids: Sequence[str]) -> Optional[ResumeItem]:
+        """Pop the first resume whose task belongs to ``task_ids``.
+
+        A shared queue is read by every worker; each worker can only re-run
+        tasks in its own shard, so the queue must hand out the right item to
+        the right worker without handing the same task to two of them. ``None``
+        when nothing queued belongs to this worker's shard -- the normal case,
+        which costs one lock acquisition per rollout."""
+        with self._lock:
+            for i, item in enumerate(self._items):
+                if item.task_id in task_ids:
+                    return self._items.pop(i)
+            return None
+
     def __len__(self) -> int:
         return len(self._items)
 
diff --git a/docs/api.md b/docs/api.md
index b47694d9..97aa7c88 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -38,6 +38,7 @@ EvolutionResult(
     stop_reason: str = 'rounds',
     forced_refreshes: int = 0,
     stragglers: int = 0,
+    resumed: int = 0,
     retired_workers: int = 0,
     audit_drained: int = 0,
     replay: Optional[Dict[str, Any]] = None,
@@ -2944,6 +2945,10 @@ DurationEstimator(
 
 Turn-level checkpoints of timed-out rollouts (partial rollout).
 
+| method | what it does |
+|---|---|
+| `pop_for(task_ids: Sequence[str]) -> Optional[ResumeItem]` | Pop the first resume whose task belongs to `task_ids`. |
+
 ### `TaskCluster(...)`
 
 ```python
@@ -3141,6 +3146,7 @@ async_evolve(
     stall_patience: int = 50,
     duration_estimator: Optional['DurationEstimator'] = None,
     straggler_factor: float = 3.0,
+    resume_queue: Optional['ResumeQueue'] = None,
     task_sampler: Optional['TaskSampler'] = None,
     on_round: Optional[Callable[[RoundInfo], None]] = None,
     stop_when: Optional[Callable[[RoundInfo], bool]] = None,
@@ -3195,6 +3201,7 @@ async_evolve(
 | `stall_patience` | `int` | `50` | Merger sweeps that may pass with cards arriving and nothing committing before every worker is forced to resync, regardless of `async_ratio`. Without it a lag budget larger than the staleness tolerance **livelocks** under the Guarded policy: workers propose against a snapshot too old for the policy to accept, every card is discarded, head never moves, and the lag budget therefore never triggers a refresh either. |
 | `duration_estimator` | `Optional['DurationEstimator']` | `None` | Pass a `DurationEstimator` to fit `seconds ~ intercept + slope * len(prompt)` online and count rollouts that overran their own estimate by more than `straggler_factor` (`result.stragglers`). This is the design's **L-traj** mechanism, which until now lived only in the reference runtime and so was unreachable from the API a real workload uses. Detection only: resuming a partial rollout would need it to expose its turns, and `run(rendered, task) -> output` is opaque. |
 | `straggler_factor` | `float` | `3.0` | As `duration_estimator`. |
+| `resume_queue` | `Optional['ResumeQueue']` | `None` |  |
 | `task_sampler` | `Optional['TaskSampler']` | `None` | Which task a worker takes next from its shard. |
 | `on_round` | `Optional[Callable[[RoundInfo], None]]` | `None` | Called with each `RoundInfo` as a merger sweep completes -- progress for a long run. It runs on the merger thread and must be cheap and thread-safe; an exception is reported, not fatal. |
 | `stop_when` | `Optional[Callable[[RoundInfo], bool]]` | `None` | Asked after `on_round` with the same `RoundInfo`; `True` ends the run with `stop_reason="stop_when"` -- the caller's own budget (dollars, a deadline, a kill file), checked between merger sweeps like the built-in bounds. Same thread and the same rules as `on_round`. |
diff --git a/tests/test_ltraj_resume.py b/tests/test_ltraj_resume.py
new file mode 100644
index 00000000..5bc8e369
--- /dev/null
+++ b/tests/test_ltraj_resume.py
@@ -0,0 +1,107 @@
+"""L-traj resume: a straggler is re-run against a newer head, not just counted.
+
+The async runtime abandons a rollout that overruns its predicted cost. The
+reference runtime checkpoints partial turns here; this port's `run` is opaque,
+so there is no turn to save. What is saved is the task and the version it
+measured -- with a `resume_queue` present, the next idle worker re-runs it
+against the current head. A straggler measured version N, its re-run measures
+version N+k, so the pair is a free cross-version A/B signal.
+"""
+
+from agentdescent.async_evolve import async_evolve
+from agentdescent.evolution import AppendRules, Task
+from agentdescent.scheduler import DurationEstimator, ResumeItem, ResumeQueue
+from agentdescent.sampling import RoundRobin
+
+
+# ---------------------------------------------------------------------------
+# the queue primitive
+# ---------------------------------------------------------------------------
+
+def test_resume_queue_pop_for_returns_shard_owned_item():
+    q = ResumeQueue()
+    q.push(ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5}))
+    q.push(ResumeItem(task_id="b", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5}))
+    # a worker whose shard owns only "b" gets "b"
+    item = q.pop_for(["b"])
+    assert item is not None and item.task_id == "b"
+    # the item was removed, not copied
+    assert q.pop_for(["a"]) is not None
+    assert q.pop_for(["a"]) is None
+
+
+def test_resume_queue_pop_for_returns_none_when_shard_has_no_resume():
+    q = ResumeQueue()
+    q.push(ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5}))
+    assert q.pop_for(["x", "y"]) is None
+    assert len(q) == 1          # untouched
+
+
+def test_resume_queue_pop_for_never_hands_same_task_twice():
+    q = ResumeQueue()
+    q.push(ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5}))
+    assert q.pop_for(["a"]) is not None
+    assert q.pop_for(["a"]) is None
+
+
+# ---------------------------------------------------------------------------
+# the engine path
+# ---------------------------------------------------------------------------
+
+def test_async_re_runs_a_straggler_against_the_newer_head():
+    """A worker whose rollout overruns its estimate is re-run; result.resumed
+    reports it, and the re-run happens against the current head."""
+    import time
+    slow_first = {"flag": True}
+
+    def run(rendered, task):
+        if slow_first["flag"]:
+            slow_first["flag"] = False
+            time.sleep(0.05)          # overruns the estimator's tiny prior
+        return "yes"
+
+    # A `DurationEstimator` with a tiny prior predicts ~0s, so the first
+    # rollout of every task is a straggler by construction.
+    estimator = DurationEstimator(prior=0.0001, min_samples=100)
+    q = ResumeQueue()
+    tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
+
+    result = async_evolve(
+        tasks, lambda t, o: 1.0 if o == "yes" else 0.0,
+        run=run, propose=lambda rd, t, o, r: None,
+        strategy=AppendRules(),
+        n_workers=2, duration_estimator=estimator, resume_queue=q,
+        straggler_factor=2.0, max_seconds=10.0, held_out_frac=0.5)
+    # every task overran on its first try; at least one was re-run.
+    assert result.stragglers >= 1
+    assert result.resumed >= 1, f"expected re-runs, got {result.resumed}"
+    # A queued straggler is re-run when a worker is free; some may legitimately
+    # still be queued when the run ends (no worker got to them), which is fine.
+    assert result.resumed <= result.stragglers
+
+
+def test_async_without_resume_queue_keeps_old_behaviour():
+    """No queue -> stragglers counted, nothing re-run, no `resumed`."""
+    import time
+    slow_first = {"flag": True}
+
+    def run(rendered, task):
+        if slow_first["flag"]:
+            slow_first["flag"] = False
+            time.sleep(0.05)
+        return "yes"
+
+    estimator = DurationEstimator(prior=0.0001, min_samples=100)
+    tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
+    result = async_evolve(
+        tasks, lambda t, o: 1.0 if o == "yes" else 0.0,
+        run=run, propose=lambda rd, t, o, r: None,
+        strategy=AppendRules(),
+        n_workers=2, duration_estimator=estimator, resume_queue=None,
+        straggler_factor=2.0, max_seconds=10.0, held_out_frac=0.5)
+    assert result.stragglers >= 1
+    assert result.resumed == 0

From 202cfe6003e3395330281ac628ead3d4216f48ed Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:00:51 +0800
Subject: [PATCH 2/7] fix(ltraj-resume): document resume_queue, serialise
 resumed, update stale docs

Three fixes the review would have caught, plus doc drift:

- resume_queue was added to async_evolve's signature without a Parameters
  entry, so test_api_docs (which fails on any undocumented parameter) was
  red.
- EvolutionResult.resumed was counted but never serialised: save()/load()
  dropped it, so a written result lost the A/B signal it exists to report.
- ResumeQueue's docstring still claimed it was 'unwired' after the engine
  started pushing and popping it; concepts.md, duration-scheduling.md,
  execution.md, modules.md and async.md all still said the resume half was
  'not implemented' / 'nothing pops the queue'. They now describe task-level
  re-run as implemented and turn-level continuation as the remaining gap.
---
 agentdescent/async_evolve.py |  9 +++++++++
 agentdescent/evolution.py    |  4 +++-
 agentdescent/scheduler.py    | 30 +++++++++++++++++-------------
 docs/api.md                  |  2 +-
 docs/async.md                |  3 ++-
 docs/concepts.md             | 15 ++++++++++-----
 docs/duration-scheduling.md  | 30 ++++++++++++++++++------------
 docs/execution.md            | 10 +++++++---
 docs/modules.md              |  2 +-
 9 files changed, 68 insertions(+), 37 deletions(-)

diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py
index 28d6ee2f..735d5e36 100644
--- a/agentdescent/async_evolve.py
+++ b/agentdescent/async_evolve.py
@@ -334,6 +334,15 @@ def async_evolve(
         the API a real workload uses. Detection only: resuming a partial rollout
         would need it to expose its turns, and ``run(rendered, task) -> output``
         is opaque.
+    resume_queue:
+        Turn a counted straggler into a **re-run**. A shared
+        :class:`~agentdescent.scheduler.ResumeQueue` passed here receives every
+        abandoned rollout's task and the version it measured; the next idle
+        worker whose shard owns that task re-runs it against the **current**
+        head. A straggler measured version N, its re-run measures version N+k,
+        so the pair is a free cross-version A/B signal. ``None`` (the default)
+        keeps the old behaviour exactly: stragglers are counted and dropped.
+        ``result.resumed`` reports the re-runs.
     shutdown_grace:
         Total seconds to wait for the worker and merger threads after the budget
         expires -- shared across all of them, not per thread. An in-flight
diff --git a/agentdescent/evolution.py b/agentdescent/evolution.py
index 0858b3d2..d5e84134 100644
--- a/agentdescent/evolution.py
+++ b/agentdescent/evolution.py
@@ -1297,7 +1297,7 @@ class EvolutionResult:
     forced_refreshes: int = 0
     #: Rollouts that overran their own predicted duration by ``straggler_factor``
     #: (async path, and only when a ``duration_estimator`` was given). The design's
-    #: L-traj signal; detection only, nothing is resumed.
+    #: L-traj signal; ``resumed`` below counts how many of these were re-run.
     stragglers: int = 0
     #: Stragglers that were re-run against a newer head (async path, and only
     #: when both a ``duration_estimator`` and a ``resume_queue`` were given).
@@ -1559,6 +1559,7 @@ def save(self, path: str) -> None:
             "retired_workers": self.retired_workers,
             "forced_refreshes": self.forced_refreshes,
             "stragglers": self.stragglers,
+            "resumed": self.resumed,
             "stop_reason": self.stop_reason,
             "ledger_log": list(self.ledger_log),
             "usage": {"calls": self.usage.calls,
@@ -1687,6 +1688,7 @@ def load(cls, path: str) -> "EvolutionResult":
             retired_workers=d.get("retired_workers", 0),
             forced_refreshes=d.get("forced_refreshes", 0),
             stragglers=d.get("stragglers", 0),
+            resumed=d.get("resumed", 0),
             stop_reason=d.get("stop_reason", "rounds"),
             fusion_trials=[FusionTrial(**t) for t in d.get("fusion_trials", [])],
             # Every cost field is `.get` with a default, so a file written before
diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py
index ca3f92bd..ded5c854 100644
--- a/agentdescent/scheduler.py
+++ b/agentdescent/scheduler.py
@@ -279,19 +279,23 @@ class ResumeItem:
 class ResumeQueue:
     """Turn-level checkpoints of timed-out rollouts (partial rollout).
 
-    Write-only in every shipped path, and only one of them writes: the reference
-    runtime pushes, nothing pops, and `async_evolve` -- the loop a real workload
-    reaches -- does not checkpoint at all.
-
-    That is deliberate, and it is not what task-level recovery uses. Recovery
-    re-dispatches the **task**: the supervisor notices a worker is gone, sends its
-    work somewhere else under the same lease id, and drops the original's answer
-    if it turns up late. Resuming a partial rollout instead would require
-    `run(rendered, task) -> output` to become an inspectable conversation, and
-    that contract is what lets any agent at all be plugged in.
-
-    So this stays the turn-level primitive it always was, unwired, rather than
-    being repurposed as a task-level channel because it happens to be a queue."""
+    The reference runtime pushes turn-level items and nothing pops them. This
+    port does something narrower but wired: :func:`~agentdescent.async_evolve`
+    pushes a **task-level** item (task id + the version it measured) when a
+    rollout overruns its estimate, and its workers pop via :meth:`pop_for` to
+    re-run the abandoned task against the latest head.
+
+    That is deliberate, and it is not what turn-level recovery would use. Full
+    recovery re-dispatches the task with continuation state: a supervisor
+    notices a worker is gone, sends its partial work somewhere else under the
+    same lease id, and drops the original's answer if it turns up late.
+    Resuming a partial rollout would require `run(rendered, task) -> output` to
+    become an inspectable conversation, and that contract is what lets any
+    agent at all be plugged in.
+
+    So the shipped path is task-level re-run (the free cross-version A/B signal
+    of L-traj), while the item shape stays able to carry the turn-level state a
+    transparent `run` would let us checkpoint."""
 
     def __init__(self, p90_multiplier: float = 2.0) -> None:
         self.p90_multiplier = p90_multiplier
diff --git a/docs/api.md b/docs/api.md
index 97aa7c88..b340a86f 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -3201,7 +3201,7 @@ async_evolve(
 | `stall_patience` | `int` | `50` | Merger sweeps that may pass with cards arriving and nothing committing before every worker is forced to resync, regardless of `async_ratio`. Without it a lag budget larger than the staleness tolerance **livelocks** under the Guarded policy: workers propose against a snapshot too old for the policy to accept, every card is discarded, head never moves, and the lag budget therefore never triggers a refresh either. |
 | `duration_estimator` | `Optional['DurationEstimator']` | `None` | Pass a `DurationEstimator` to fit `seconds ~ intercept + slope * len(prompt)` online and count rollouts that overran their own estimate by more than `straggler_factor` (`result.stragglers`). This is the design's **L-traj** mechanism, which until now lived only in the reference runtime and so was unreachable from the API a real workload uses. Detection only: resuming a partial rollout would need it to expose its turns, and `run(rendered, task) -> output` is opaque. |
 | `straggler_factor` | `float` | `3.0` | As `duration_estimator`. |
-| `resume_queue` | `Optional['ResumeQueue']` | `None` |  |
+| `resume_queue` | `Optional['ResumeQueue']` | `None` | Turn a counted straggler into a **re-run**. A shared `ResumeQueue` passed here receives every abandoned rollout's task and the version it measured; the next idle worker whose shard owns that task re-runs it against the **current** head. A straggler measured version N, its re-run measures version N+k, so the pair is a free cross-version A/B signal. `None` (the default) keeps the old behaviour exactly: stragglers are counted and dropped. `result.resumed` reports the re-runs. |
 | `task_sampler` | `Optional['TaskSampler']` | `None` | Which task a worker takes next from its shard. |
 | `on_round` | `Optional[Callable[[RoundInfo], None]]` | `None` | Called with each `RoundInfo` as a merger sweep completes -- progress for a long run. It runs on the merger thread and must be cheap and thread-safe; an exception is reported, not fatal. |
 | `stop_when` | `Optional[Callable[[RoundInfo], bool]]` | `None` | Asked after `on_round` with the same `RoundInfo`; `True` ends the run with `stop_reason="stop_when"` -- the caller's own budget (dollars, a deadline, a kill file), checked between merger sweeps like the built-in bounds. Same thread and the same rules as `on_round`. |
diff --git a/docs/async.md b/docs/async.md
index 7e1ed597..740f3213 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -136,12 +136,13 @@ calls per proposal.
 
 ## What the async path adds
 
-Beyond the barrier removal, three signals only it can report:
+Beyond the barrier removal, four signals only it can report:
 
 | field | meaning |
 |---|---|
 | `result.forced_refreshes` | workers forced to resync because the pipeline stalled — cards arriving, nothing committing |
 | `result.stragglers` | rollouts that overran their predicted duration by `straggler_factor` (needs a [`duration_estimator=`](duration-scheduling.md)) |
+| `result.resumed` | stragglers re-run against a newer head (needs both a `duration_estimator=` and a shared [`resume_queue=`](duration-scheduling.md)) |
 | `result.retired_workers` | workers that gave up after repeated backend failures |
 
 `retired_workers` deserves attention: a run can finish **cleanly** at a fraction
diff --git a/docs/concepts.md b/docs/concepts.md
index 2b4099fa..3cff7d80 100644
--- a/docs/concepts.md
+++ b/docs/concepts.md
@@ -295,11 +295,16 @@ mechanism:
   `async_evolve` directly when you want it. The synchronous path bounds a slow
   rollout with `round_timeout=` instead, which abandons the straggler rather than
   measuring it.
-  **The resume half is not implemented** — nothing pops that queue, and the
-  recorded item carries no continuation state, because a resumable rollout would
-  have to expose its turns and `run(rendered, task) -> output` is opaque. The
-  design's "resume on the latest Ledger for a free cross-version A/B signal"
-  therefore remains a design note, not behaviour. Removing the barrier
+  **The resume half is implemented at the task level, not the turn level.** Pass
+  a shared [`ResumeQueue`](duration-scheduling.md) to `async_evolve` as `resume_queue=` and
+  a counted straggler is *recorded* (task + the version it measured) instead of
+  dropped; the next idle worker whose shard owns that task re-runs it against the
+  **latest** head (`result.resumed`). A straggler measured version N, its re-run
+  measures version N+k, so the pair is the design's free cross-version A/B
+  signal — now behaviour, not a note. What is still **not** implemented is
+  turn-level continuation: the recorded item carries no conversation state,
+  because a resumable rollout would have to expose its turns and
+  `run(rendered, task) -> output` is opaque. Removing the barrier
   ([async](evolution.md#the-barrier-free-runtime-async_evolve)) is what keeps one
   slow rollout from setting the pace in practice.
 - **L-task (data layer)** — Zipfian artifact triggering (head skills flooded,
diff --git a/docs/duration-scheduling.md b/docs/duration-scheduling.md
index 092c9ab1..03063a0c 100644
--- a/docs/duration-scheduling.md
+++ b/docs/duration-scheduling.md
@@ -104,21 +104,27 @@ rollouts=727, learned base≈0.006s, stragglers detected=108
 This is the *detection* half of the design's partial-rollout mechanism (§5.1),
 driven by a live estimate: a rollout predicted to be short but running long is
 identified and counted, so it shows up in the stats instead of silently setting
-the pace. The design's other half — setting it aside and resuming it against the
-latest ledger — is where the mechanism stops:
+the pace. The design's other half — setting it aside and re-running it against
+the latest ledger — is implemented at the **task level** (see below), with the
+**turn level** still closed:
 
-!!! warning "Straggler *resume* is not implemented"
+!!! warning "Straggler *resume* is task-level, not turn-level"
     A rollout that overruns its predicted cost is flagged into `ResumeQueue` and
     counted — that part is real, and it is what keeps a straggler from silently
-    defining the round's wall-clock in the reported stats. But the rollout is not
-    interrupted (the flag is recorded *after* it returns), the queued item carries
-    no continuation state, and **nothing pops the queue**. True turn-level
-    checkpoint-and-resume would need a rollout contract that exposes its turns;
-    the engine's `run(rendered, task) -> output` is opaque. What actually prevents
-    one slow rollout from stalling the rest today is removing the barrier — see
-    [the async runtime](evolution.md#the-barrier-free-runtime-async_evolve), which
-    the [efficiency experiments](efficiency.md) measure at ~2.65x over a sync
-    barrier under heavy-tailed latency.
+    defining the round's wall-clock in the reported stats. The rollout is not
+    interrupted (the flag is recorded *after* it returns). Pass a shared
+    `ResumeQueue` to `async_evolve(resume_queue=...)` and the next idle worker
+    whose shard owns the straggler's task **re-runs it against the current
+    head** (`result.resumed`): the straggler measured version N, its re-run
+    measures version N+k, so the pair is the design's free cross-version A/B
+    signal. What remains closed is turn-level checkpoint-and-resume: the queued
+    item carries no continuation state, which would need a rollout contract that
+    exposes its turns, and the engine's `run(rendered, task) -> output` is
+    opaque. What actually prevents one slow rollout from stalling the rest today
+    is removing the barrier — see
+    [the async runtime](evolution.md#the-barrier-free-runtime-async_evolve),
+    which the [efficiency experiments](efficiency.md) measure at ~2.65x over a
+    sync barrier under heavy-tailed latency.
 
 ---
 
diff --git a/docs/execution.md b/docs/execution.md
index 0a57ccfa..cfa229f6 100644
--- a/docs/execution.md
+++ b/docs/execution.md
@@ -186,9 +186,13 @@ paying twice.
 !!! note "Not partial-rollout resume"
     Resuming a half-finished rollout would need `run(rendered, task) -> output` to
     become an inspectable conversation, and that opaque contract is what lets any
-    agent be plugged in at all. [`ResumeQueue`](async.md) stays the turn-level
-    primitive it always was, unwired, rather than being repurposed as a
-    task-level channel because it happens to be a queue.
+    agent be plugged in at all. [`ResumeQueue`](async.md) therefore carries the
+    turn-level item shape, but what `async_evolve(resume_queue=...)` actually
+    pushes and pops is **task-level**: the straggler's task id + the version it
+    measured, re-run against the latest head by the next idle worker that owns it
+    (`result.resumed`). That is re-dispatch of a task, not continuation of a
+    conversation — the free cross-version A/B signal, not turn-level
+    checkpointing.
 
 ## The gate has its own concurrency
 
diff --git a/docs/modules.md b/docs/modules.md
index ef4ec67c..eba82eb9 100644
--- a/docs/modules.md
+++ b/docs/modules.md
@@ -133,7 +133,7 @@ class at a time. They are all in one table instead:
 |---|---|---|
 | [`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 |
 | [`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 | a rollout that exposes its turns; `run(rendered, task) -> output` is opaque, which is what lets any agent be plugged in |
+| [`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) |

From 7f72f6836bf5fc86e9aae85e84f3ceff28cbd72a Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:01:01 +0800
Subject: [PATCH 3/7] fix(resume): bound re-queue attempts so a chronically
 slow task cannot monopolise a worker

A resumed task goes through the same straggler check as a fresh one, so
without a cap a task that is always slow is pushed, popped, re-run, pushed
again, and starves the worker's fresh tasks. ResumeQueue now counts how many
times each task has been resumed (in pop_for), and push refuses a task past
ResumeQueue(max_attempts=...) (default 1): the A/B signal needs one re-run,
and a task that is still a straggler after that is dropped (still counted in
result.stragglers, just never re-queued).

Tests: queue-level (push refuses past the cap; a chronically slow task gets
exactly max_attempts re-runs then nothing is queued) and end-to-end (all
tasks always slow -> resumed is bounded by max_attempts * n_tasks).
---
 CHANGELOG.md                 |  5 +++
 agentdescent/async_evolve.py |  4 +++
 agentdescent/scheduler.py    | 42 +++++++++++++++++++++---
 docs/api.md                  |  3 +-
 tests/test_ltraj_resume.py   | 62 ++++++++++++++++++++++++++++++++++++
 5 files changed, 110 insertions(+), 6 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bd7a60eb..11b4c89c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -112,6 +112,11 @@ value-directed evaluation or Genesis blind-mode proposals.
   reports the re-runs. `resume_queue=None` (the default) is the old behaviour
   exactly: stragglers are counted and dropped.
 
+  **Re-queue is bounded** (`ResumeQueue(max_attempts=...)`, default 1): the
+  queue counts how many times each task has been resumed, and `push` refuses a
+  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.
+
 - **`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/async_evolve.py b/agentdescent/async_evolve.py
index 735d5e36..9768eed6 100644
--- a/agentdescent/async_evolve.py
+++ b/agentdescent/async_evolve.py
@@ -838,6 +838,10 @@ def _worker(wid: int, shard: List[Task]) -> None:
                     with counter_lock:
                         stragglers[0] += 1
                     if resume_queue is not None:
+                        # `push` refuses a task that has already been resumed
+                        # `max_attempts` times (the queue counts pops), so a
+                        # chronically slow task is dropped rather than re-
+                        # dispatching forever and starving fresh work.
                         resume_queue.push(ResumeItem(
                             task_id=task.id, turn=0, conversation=[],
                             version_at_checkpoint={eng.artifact_id: base_v}))
diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py
index ded5c854..369a9061 100644
--- a/agentdescent/scheduler.py
+++ b/agentdescent/scheduler.py
@@ -295,11 +295,29 @@ class ResumeQueue:
 
     So the shipped path is task-level re-run (the free cross-version A/B signal
     of L-traj), while the item shape stays able to carry the turn-level state a
-    transparent `run` would let us checkpoint."""
-
-    def __init__(self, p90_multiplier: float = 2.0) -> None:
+    transparent `run` would let us checkpoint.
+
+    **Re-queue is bounded.** A resumed task goes through the same straggler
+    check as a fresh one, so without a cap a task that is *always* slow would
+    be pushed, popped, re-run, pushed again, and starve the worker's fresh
+    tasks. :attr:`max_attempts` is the bound: the queue counts how many times
+    each task has been resumed (see :meth:`pop_for`), and :meth:`push` refuses
+    once that count has been reached. The chronically slow task is then dropped
+    -- still counted in ``result.stragglers``, just never re-queued again."""
+
+    def __init__(self, p90_multiplier: float = 2.0,
+                 max_attempts: int = 1) -> None:
         self.p90_multiplier = p90_multiplier
+        #: How many times one task may be resumed before it stops being
+        #: re-queued. The A/B signal needs one re-run; beyond that the task is
+        #: chronically slow and re-dispatching it only starves fresh work.
+        self.max_attempts = max_attempts
         self._items: List[ResumeItem] = []
+        #: task_id -> how many times it has been resumed (popped). What bounds
+        #: the re-queue: `push` refuses a task whose count has hit the cap.
+        self._resumed_count: Dict[str, int] = defaultdict(int)
+        #: How many pushes were refused because the task was past the cap.
+        self.dropped = 0
         # every async worker thread pushes here; list.append happens to be
         # GIL-atomic but pop() + the emptiness check are not.
         self._lock = threading.Lock()
@@ -307,9 +325,19 @@ def __init__(self, p90_multiplier: float = 2.0) -> None:
     def should_checkpoint(self, elapsed: float, p90: float) -> bool:
         return elapsed > self.p90_multiplier * p90
 
-    def push(self, item: ResumeItem) -> None:
+    def push(self, item: ResumeItem) -> bool:
+        """Queue ``item``; ``False`` when this task has been resumed too often.
+
+        A straggler is dropped (not queued) once its task has been resumed
+        :attr:`max_attempts` times -- the bounded-retry guard against a
+        chronically slow task monopolising a worker's shard. Returns whether the
+        item was accepted."""
+        if self._resumed_count[item.task_id] >= self.max_attempts:
+            self.dropped += 1
+            return False
         with self._lock:
             self._items.append(item)
+        return True
 
     def pop(self) -> Optional[ResumeItem]:
         with self._lock:
@@ -322,10 +350,14 @@ def pop_for(self, task_ids: Sequence[str]) -> Optional[ResumeItem]:
         tasks in its own shard, so the queue must hand out the right item to
         the right worker without handing the same task to two of them. ``None``
         when nothing queued belongs to this worker's shard -- the normal case,
-        which costs one lock acquisition per rollout."""
+        which costs one lock acquisition per rollout.
+
+        Popping counts as a resume for that task, which feeds the
+        :attr:`max_attempts` bound in :meth:`push`."""
         with self._lock:
             for i, item in enumerate(self._items):
                 if item.task_id in task_ids:
+                    self._resumed_count[item.task_id] += 1
                     return self._items.pop(i)
             return None
 
diff --git a/docs/api.md b/docs/api.md
index b340a86f..9d785f1a 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -2941,13 +2941,14 @@ DurationEstimator(
 ) -> None
 ```
 
-### `ResumeQueue(p90_multiplier: float = 2.0) -> None`
+### `ResumeQueue(p90_multiplier: float = 2.0, max_attempts: int = 1) -> None`
 
 Turn-level checkpoints of timed-out rollouts (partial rollout).
 
 | method | what it does |
 |---|---|
 | `pop_for(task_ids: Sequence[str]) -> Optional[ResumeItem]` | Pop the first resume whose task belongs to `task_ids`. |
+| `push(item: ResumeItem) -> bool` | Queue `item`; `False` when this task has been resumed too often. |
 
 ### `TaskCluster(...)`
 
diff --git a/tests/test_ltraj_resume.py b/tests/test_ltraj_resume.py
index 5bc8e369..df2ed9ad 100644
--- a/tests/test_ltraj_resume.py
+++ b/tests/test_ltraj_resume.py
@@ -48,6 +48,38 @@ def test_resume_queue_pop_for_never_hands_same_task_twice():
     assert q.pop_for(["a"]) is None
 
 
+def test_resume_queue_push_refuses_a_task_resumed_past_max_attempts():
+    """A task that keeps being a straggler must stop being re-queued once it
+    has been resumed `max_attempts` times -- otherwise it starves the worker's
+    fresh tasks (PR #198 review: repeated re-queue has no bound)."""
+    q = ResumeQueue(max_attempts=1)
+    item = ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5})
+    assert q.push(item), "first push is always accepted"
+    assert q.pop_for(["a"]) is not None, "the item is resumed once"
+    # Popping counted the resume; the second re-queue must be refused.
+    assert not q.push(item), (
+        "a task resumed max_attempts times must not be re-queued")
+
+
+def test_resume_queue_max_attempts_bounds_a_chronically_slow_task():
+    """A task that is always a straggler gets exactly `max_attempts` re-runs;
+    past the cap its pushes are dropped, so it cannot monopolise the worker."""
+    q = ResumeQueue(max_attempts=2)
+    resumed = 0
+    for _ in range(10):
+        item = ResumeItem(task_id="a", turn=0, conversation=[],
+                          version_at_checkpoint={"art": 5})
+        accepted = q.push(item)
+        if not accepted:
+            break                    # past the cap: refused, nothing to pop
+        if q.pop_for(["a"]) is not None:
+            resumed += 1
+    assert resumed == 2, f"expected exactly 2 re-runs, got {resumed}"
+    assert q.dropped >= 1, "the chronically slow task must be dropped"
+    assert len(q) == 0, "nothing may remain queued past the cap"
+
+
 # ---------------------------------------------------------------------------
 # the engine path
 # ---------------------------------------------------------------------------
@@ -105,3 +137,33 @@ def run(rendered, task):
         straggler_factor=2.0, max_seconds=10.0, held_out_frac=0.5)
     assert result.stragglers >= 1
     assert result.resumed == 0
+
+
+def test_a_chronically_slow_task_cannot_be_resumed_forever():
+    """A task that is *always* a straggler gets bounded re-runs, not an
+    infinite resume loop starving the worker's fresh tasks (PR #198 review)."""
+    import time
+
+    def run(rendered, task):
+        time.sleep(0.05)              # always overruns the tiny prior
+        return "yes"
+
+    estimator = DurationEstimator(prior=0.0001, min_samples=100)
+    q = ResumeQueue(max_attempts=2)
+    tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
+
+    result = async_evolve(
+        tasks, lambda t, o: 1.0 if o == "yes" else 0.0,
+        run=run, propose=lambda rd, t, o, r: None,
+        strategy=AppendRules(),
+        n_workers=2, duration_estimator=estimator, resume_queue=q,
+        straggler_factor=2.0, max_seconds=8.0, held_out_frac=0.5)
+    # Every task is a straggler, but re-runs are capped: no task may be
+    # resumed more than `max_attempts` times.
+    assert result.stragglers >= 1
+    assert result.resumed <= 2 * len(tasks), (
+        f"resume must be bounded per task; got {result.resumed} re-runs")
+    assert q.dropped >= 0
+    # The queue never grows unbounded: everything pushed past the cap is
+    # dropped, so at any point at most one straggler per task is queued.
+    assert len(q) <= len(tasks)

From efd88260438c2afb7975db23bdf3098124787a33 Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:01:04 +0800
Subject: [PATCH 4/7] fix(resume): bound the re-queue by queued copies too, and
 count pop() resumes

PR #198 review: push() checked only the popped count, so a burst of pushes
for the same task before any pop was all accepted and all handed out, and
pop() never fed the cap. Count popped + currently-queued per task at push
time, and count a pop()/pop_for() resume under the same lock.
---
 agentdescent/scheduler.py  | 38 ++++++++++++++++++++++++++++++--------
 tests/test_ltraj_resume.py | 31 +++++++++++++++++++++++++++++++
 2 files changed, 61 insertions(+), 8 deletions(-)

diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py
index 369a9061..0bc51e95 100644
--- a/agentdescent/scheduler.py
+++ b/agentdescent/scheduler.py
@@ -313,9 +313,15 @@ def __init__(self, p90_multiplier: float = 2.0,
         #: chronically slow and re-dispatching it only starves fresh work.
         self.max_attempts = max_attempts
         self._items: List[ResumeItem] = []
-        #: task_id -> how many times it has been resumed (popped). What bounds
-        #: the re-queue: `push` refuses a task whose count has hit the cap.
+        #: task_id -> how many times it has been resumed (popped). Half of what
+        #: bounds the re-queue: `push` refuses a task whose *total* (this plus
+        #: how many of it are still queued) has hit the cap, so a task that was
+        #: pushed several times without a pop in between cannot exceed it either.
         self._resumed_count: Dict[str, int] = defaultdict(int)
+        #: task_id -> how many of that task are currently sitting in the queue.
+        #: The other half of the bound: a burst of pushes for the same task
+        #: before any pop must not all be accepted (PR #198 review).
+        self._queued_count: Dict[str, int] = defaultdict(int)
         #: How many pushes were refused because the task was past the cap.
         self.dropped = 0
         # every async worker thread pushes here; list.append happens to be
@@ -330,18 +336,32 @@ def push(self, item: ResumeItem) -> bool:
 
         A straggler is dropped (not queued) once its task has been resumed
         :attr:`max_attempts` times -- the bounded-retry guard against a
-        chronically slow task monopolising a worker's shard. Returns whether the
+        chronically slow task monopolising a worker's shard. The bound counts
+        *both* the resumes already handed out and the copies of the task still
+        queued, so a burst of pushes before any pop cannot exceed the cap
+        (PR #198 review: ``push`` checked only the popped count, so three
+        pushes with no pop in between were all accepted). Returns whether the
         item was accepted."""
-        if self._resumed_count[item.task_id] >= self.max_attempts:
-            self.dropped += 1
-            return False
         with self._lock:
+            total = (self._resumed_count[item.task_id]
+                     + self._queued_count[item.task_id])
+            if total >= self.max_attempts:
+                self.dropped += 1
+                return False
             self._items.append(item)
+            self._queued_count[item.task_id] += 1
         return True
 
     def pop(self) -> Optional[ResumeItem]:
+        """Pop the next resume. Counts as a resume for that task, feeding the
+        :attr:`max_attempts` bound in :meth:`push`."""
         with self._lock:
-            return self._items.pop(0) if self._items else None
+            if not self._items:
+                return None
+            item = self._items.pop(0)
+            self._queued_count[item.task_id] -= 1
+            self._resumed_count[item.task_id] += 1
+            return item
 
     def pop_for(self, task_ids: Sequence[str]) -> Optional[ResumeItem]:
         """Pop the first resume whose task belongs to ``task_ids``.
@@ -357,8 +377,10 @@ def pop_for(self, task_ids: Sequence[str]) -> Optional[ResumeItem]:
         with self._lock:
             for i, item in enumerate(self._items):
                 if item.task_id in task_ids:
+                    del self._items[i]
+                    self._queued_count[item.task_id] -= 1
                     self._resumed_count[item.task_id] += 1
-                    return self._items.pop(i)
+                    return item
             return None
 
     def __len__(self) -> int:
diff --git a/tests/test_ltraj_resume.py b/tests/test_ltraj_resume.py
index df2ed9ad..f6a2fa6a 100644
--- a/tests/test_ltraj_resume.py
+++ b/tests/test_ltraj_resume.py
@@ -80,6 +80,37 @@ def test_resume_queue_max_attempts_bounds_a_chronically_slow_task():
     assert len(q) == 0, "nothing may remain queued past the cap"
 
 
+def test_resume_queue_duplicate_pushes_before_any_pop_are_bounded():
+    """A burst of pushes for the same task, before the first pop, must not all
+    be accepted: the cap counts *queued* copies too, not just popped ones
+    (PR #198 review: `push` checked only `_resumed_count`, so three pushes
+    with no pop in between all landed and were all handed out)."""
+    q = ResumeQueue(max_attempts=1)
+    item = ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5})
+    assert q.push(item), "first push is always accepted"
+    assert not q.push(item), "a second queued copy must be refused"
+    assert not q.push(item), "a third queued copy must be refused"
+    assert len(q) == 1, "only the first push may land"
+    assert q.dropped == 2, "the two refusals must be counted"
+    # And the single queued copy is what `pop_for` hands out, once.
+    assert q.pop_for(["a"]) is not None, "the accepted item is resumed"
+    assert q.pop_for(["a"]) is None, "nothing is left to hand out twice"
+
+
+def test_resume_queue_pop_counts_toward_the_cap_too():
+    """`pop()` (not just `pop_for`) is a resume and must feed the same bound:
+    after a pop, a re-push of the same task is refused (PR #198 review: `pop()`
+    never updated the count, so a consumer that used it could exceed the cap)."""
+    q = ResumeQueue(max_attempts=1)
+    item = ResumeItem(task_id="a", turn=0, conversation=[],
+                      version_at_checkpoint={"art": 5})
+    assert q.push(item)
+    assert q.pop() is not None, "the item is resumed via pop()"
+    assert not q.push(item), (
+        "a task popped once must not be re-queued past the cap")
+
+
 # ---------------------------------------------------------------------------
 # the engine path
 # ---------------------------------------------------------------------------

From 7371596b91d5553c3a81ac4fe25b918e3269b263 Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:01:14 +0800
Subject: [PATCH 5/7] fix(resume): align wording with after-return detection
 and prove newer-head re-run

Two secondary follow-ups from the queue review.

1. 'Abandons a rollout' overstated what detection does: run(rendered,
   task) -> output is opaque, so a rollout that overruns its estimate is
   noticed once it has returned, not cancelled mid-flight. Aligned the
   async_evolve docstring, docs/api.md and the changelog entry with that
   behaviour.

2. The against_the_newer_head test proposed None, so it exercised
   re-execution without establishing that a newer version was used. It now
   records the rendered artifact per rollout: t0 is a chronic straggler
   while another shard commits a clue, and the test asserts t0's re-run
   renders the post-clue head.
---
 CHANGELOG.md                 |  8 +++--
 agentdescent/async_evolve.py | 14 +++++----
 agentdescent/scheduler.py    |  8 ++---
 docs/api.md                  |  2 +-
 tests/test_ltraj_resume.py   | 58 ++++++++++++++++++++++--------------
 5 files changed, 53 insertions(+), 37 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 11b4c89c..d784c964 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -98,10 +98,12 @@ value-directed evaluation or Genesis blind-mode proposals.
   Wired end to end: the worker asks the governor `candidate_worth(ctx)` before the self-verify rollout; the aggregator reports **every** measured decision exactly once (`observe(ctx, committed)`), with oracle rejection and acceptance rejection as negative labels — a model fit on commits alone never leaves warm-up. **Every commit is attributed to a worker's actual pre-spend vector**: a fused candidate keeps its first contributor's context, and a tournament-selected single proposal is matched to its own winning card — never a losing first entrant, never a fused union size. The feature vector is **scale-normalised**: the raw-op diff size (1..100+) would swamp the advantage (a z-score) and blast radius in a raw logistic fit, so `size` is mapped through `log1p` to `[0, 1]` — verified by a regression where a high-advantage, large-diff candidate still predicts commit. `allocator=None` (the default) is the old behaviour byte for byte — the clock thresholds remain the *ceiling* the allocator may never exceed. 25 tests in `tests/test_allocator.py` (incl. negative-outcome, sync/async feature-parity, end-to-end fused + single-winner regressions, the A/B arm wiring, and the size-scale regression). **`bench/ab_allocator.py`** is the equal-budget A/B (off vs on, real datasets HotpotQA/finer, per-seed quality + measured tokens + a fired check that distinguishes "the allocator helped" from "it never ran"); exported from the package root; `docs/api.md` regenerated.
 
 - **`async_evolve(resume_queue=...)` — the L-traj resume half, the straggler
-  re-run.** The async runtime abandons a rollout that overruns its predicted
-  cost and counts it (`result.stragglers`). With a shared
+  re-run.** Detection is **after-return**: `run(rendered, task) -> output` is
+  opaque, so a rollout that overruns its predicted cost is noticed once it has
+  finished (it is not cancelled mid-flight) and counted
+  (`result.stragglers`). With a shared
   :class:`~agentdescent.scheduler.ResumeQueue` passed in, it now *records* the
-  abandoned task and the version it measured instead of dropping it; the next
+  detected task and the version it measured instead of dropping it; the next
   idle worker whose shard owns that task re-runs it against the **current**
   head. A straggler measured version N, its re-run measures version N+k, so the
   pair is a free cross-version A/B signal — the design's L-traj note made
diff --git a/agentdescent/async_evolve.py b/agentdescent/async_evolve.py
index 9768eed6..2b861b03 100644
--- a/agentdescent/async_evolve.py
+++ b/agentdescent/async_evolve.py
@@ -126,9 +126,11 @@ def async_evolve(
     stall_patience: int = 50,
     duration_estimator: Optional["DurationEstimator"] = None,
     straggler_factor: float = 3.0,
-    #: L-traj resume: a shared queue of abandoned stragglers, each carrying the
-    #: task and the version it was rolled out against. A worker that overruns
-    #: its predicted cost is abandoned -- but with a queue present its task is
+    #: L-traj resume: a shared queue of detected stragglers, each carrying the
+    #: task and the version it was rolled out against. Detection is
+    #: **after-return**: ``run(rendered, task) -> output`` is opaque, so a
+    #: rollout that overruns its predicted cost is noticed once it has returned
+    #: (it is not cancelled mid-flight). With a queue present its task is
     #: recorded instead of just counted, and the next idle worker re-runs it
     #: against the **latest** head. The straggler measured an old version, the
     #: re-run measures the new one, so the two are a free cross-version A/B
@@ -337,10 +339,12 @@ def async_evolve(
     resume_queue:
         Turn a counted straggler into a **re-run**. A shared
         :class:`~agentdescent.scheduler.ResumeQueue` passed here receives every
-        abandoned rollout's task and the version it measured; the next idle
+        detected straggler's task and the version it measured; the next idle
         worker whose shard owns that task re-runs it against the **current**
         head. A straggler measured version N, its re-run measures version N+k,
-        so the pair is a free cross-version A/B signal. ``None`` (the default)
+        so the pair is a free cross-version A/B signal. Detection is
+        **after-return** -- a rollout that overruns its estimate is noticed
+        once it has finished, not cancelled mid-flight. ``None`` (the default)
         keeps the old behaviour exactly: stragglers are counted and dropped.
         ``result.resumed`` reports the re-runs.
     shutdown_grace:
diff --git a/agentdescent/scheduler.py b/agentdescent/scheduler.py
index 0bc51e95..63e9494d 100644
--- a/agentdescent/scheduler.py
+++ b/agentdescent/scheduler.py
@@ -320,7 +320,7 @@ def __init__(self, p90_multiplier: float = 2.0,
         self._resumed_count: Dict[str, int] = defaultdict(int)
         #: task_id -> how many of that task are currently sitting in the queue.
         #: The other half of the bound: a burst of pushes for the same task
-        #: before any pop must not all be accepted (PR #198 review).
+        #: before any pop must not all be accepted.
         self._queued_count: Dict[str, int] = defaultdict(int)
         #: How many pushes were refused because the task was past the cap.
         self.dropped = 0
@@ -338,10 +338,8 @@ def push(self, item: ResumeItem) -> bool:
         :attr:`max_attempts` times -- the bounded-retry guard against a
         chronically slow task monopolising a worker's shard. The bound counts
         *both* the resumes already handed out and the copies of the task still
-        queued, so a burst of pushes before any pop cannot exceed the cap
-        (PR #198 review: ``push`` checked only the popped count, so three
-        pushes with no pop in between were all accepted). Returns whether the
-        item was accepted."""
+        queued, so a burst of pushes before any pop cannot exceed the cap.
+        Returns whether the item was accepted."""
         with self._lock:
             total = (self._resumed_count[item.task_id]
                      + self._queued_count[item.task_id])
diff --git a/docs/api.md b/docs/api.md
index 9d785f1a..0313b402 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -3202,7 +3202,7 @@ async_evolve(
 | `stall_patience` | `int` | `50` | Merger sweeps that may pass with cards arriving and nothing committing before every worker is forced to resync, regardless of `async_ratio`. Without it a lag budget larger than the staleness tolerance **livelocks** under the Guarded policy: workers propose against a snapshot too old for the policy to accept, every card is discarded, head never moves, and the lag budget therefore never triggers a refresh either. |
 | `duration_estimator` | `Optional['DurationEstimator']` | `None` | Pass a `DurationEstimator` to fit `seconds ~ intercept + slope * len(prompt)` online and count rollouts that overran their own estimate by more than `straggler_factor` (`result.stragglers`). This is the design's **L-traj** mechanism, which until now lived only in the reference runtime and so was unreachable from the API a real workload uses. Detection only: resuming a partial rollout would need it to expose its turns, and `run(rendered, task) -> output` is opaque. |
 | `straggler_factor` | `float` | `3.0` | As `duration_estimator`. |
-| `resume_queue` | `Optional['ResumeQueue']` | `None` | Turn a counted straggler into a **re-run**. A shared `ResumeQueue` passed here receives every abandoned rollout's task and the version it measured; the next idle worker whose shard owns that task re-runs it against the **current** head. A straggler measured version N, its re-run measures version N+k, so the pair is a free cross-version A/B signal. `None` (the default) keeps the old behaviour exactly: stragglers are counted and dropped. `result.resumed` reports the re-runs. |
+| `resume_queue` | `Optional['ResumeQueue']` | `None` | Turn a counted straggler into a **re-run**. A shared `ResumeQueue` passed here receives every detected straggler's task and the version it measured; the next idle worker whose shard owns that task re-runs it against the **current** head. A straggler measured version N, its re-run measures version N+k, so the pair is a free cross-version A/B signal. Detection is **after-return** -- a rollout that overruns its estimate is noticed once it has finished, not cancelled mid-flight. `None` (the default) keeps the old behaviour exactly: stragglers are counted and dropped. `result.resumed` reports the re-runs. |
 | `task_sampler` | `Optional['TaskSampler']` | `None` | Which task a worker takes next from its shard. |
 | `on_round` | `Optional[Callable[[RoundInfo], None]]` | `None` | Called with each `RoundInfo` as a merger sweep completes -- progress for a long run. It runs on the merger thread and must be cheap and thread-safe; an exception is reported, not fatal. |
 | `stop_when` | `Optional[Callable[[RoundInfo], bool]]` | `None` | Asked after `on_round` with the same `RoundInfo`; `True` ends the run with `stop_reason="stop_when"` -- the caller's own budget (dollars, a deadline, a kill file), checked between merger sweeps like the built-in bounds. Same thread and the same rules as `on_round`. |
diff --git a/tests/test_ltraj_resume.py b/tests/test_ltraj_resume.py
index f6a2fa6a..97d44ecd 100644
--- a/tests/test_ltraj_resume.py
+++ b/tests/test_ltraj_resume.py
@@ -51,7 +51,7 @@ def test_resume_queue_pop_for_never_hands_same_task_twice():
 def test_resume_queue_push_refuses_a_task_resumed_past_max_attempts():
     """A task that keeps being a straggler must stop being re-queued once it
     has been resumed `max_attempts` times -- otherwise it starves the worker's
-    fresh tasks (PR #198 review: repeated re-queue has no bound)."""
+    fresh tasks."""
     q = ResumeQueue(max_attempts=1)
     item = ResumeItem(task_id="a", turn=0, conversation=[],
                       version_at_checkpoint={"art": 5})
@@ -82,9 +82,7 @@ def test_resume_queue_max_attempts_bounds_a_chronically_slow_task():
 
 def test_resume_queue_duplicate_pushes_before_any_pop_are_bounded():
     """A burst of pushes for the same task, before the first pop, must not all
-    be accepted: the cap counts *queued* copies too, not just popped ones
-    (PR #198 review: `push` checked only `_resumed_count`, so three pushes
-    with no pop in between all landed and were all handed out)."""
+    be accepted: the cap counts *queued* copies too, not just popped ones."""
     q = ResumeQueue(max_attempts=1)
     item = ResumeItem(task_id="a", turn=0, conversation=[],
                       version_at_checkpoint={"art": 5})
@@ -100,8 +98,7 @@ def test_resume_queue_duplicate_pushes_before_any_pop_are_bounded():
 
 def test_resume_queue_pop_counts_toward_the_cap_too():
     """`pop()` (not just `pop_for`) is a resume and must feed the same bound:
-    after a pop, a re-push of the same task is refused (PR #198 review: `pop()`
-    never updated the count, so a consumer that used it could exceed the cap)."""
+    after a pop, a re-push of the same task is refused."""
     q = ResumeQueue(max_attempts=1)
     item = ResumeItem(task_id="a", turn=0, conversation=[],
                       version_at_checkpoint={"art": 5})
@@ -116,35 +113,50 @@ def test_resume_queue_pop_counts_toward_the_cap_too():
 # ---------------------------------------------------------------------------
 
 def test_async_re_runs_a_straggler_against_the_newer_head():
-    """A worker whose rollout overruns its estimate is re-run; result.resumed
-    reports it, and the re-run happens against the current head."""
+    """A worker whose rollout overruns its estimate is re-run against a head
+    that has since advanced -- not re-executed for its own sake. The `run`
+    records the artifact each rollout saw, so the test can tell the two apart:
+    t0 is slow (a straggler) while another shard commits a clue; t0's re-run
+    must render the artifact *with* the clue, i.e. a newer version than the
+    one its abandoned rollout measured."""
     import time
-    slow_first = {"flag": True}
+    renders = []                     # (task_id, rendered) per rollout
 
     def run(rendered, task):
-        if slow_first["flag"]:
-            slow_first["flag"] = False
-            time.sleep(0.05)          # overruns the estimator's tiny prior
-        return "yes"
-
-    # A `DurationEstimator` with a tiny prior predicts ~0s, so the first
-    # rollout of every task is a straggler by construction.
+        renders.append((task.id, rendered))
+        if task.id == "t0":
+            time.sleep(0.2)          # t0 is always slow: a chronic straggler
+        return "correct" if "clue" in rendered else "wrong"
+
+    # Only t0 stays unsolved; every other task that lands a proposal commits a
+    # clue, advancing the head between t0's abandoned rollout and its re-run.
+    def propose(rendered, task, output, reward):
+        if task.id != "t0" and "clue" not in rendered:
+            return "add a clue"
+        return None
+
+    # A `DurationEstimator` with a tiny prior predicts ~0s, so every rollout
+    # of t0 (0.2s) is a straggler by construction.
     estimator = DurationEstimator(prior=0.0001, min_samples=100)
     q = ResumeQueue()
     tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
 
     result = async_evolve(
-        tasks, lambda t, o: 1.0 if o == "yes" else 0.0,
-        run=run, propose=lambda rd, t, o, r: None,
+        tasks, lambda t, o: 1.0 if o == "correct" else 0.0,
+        run=run, propose=propose,
         strategy=AppendRules(),
         n_workers=2, duration_estimator=estimator, resume_queue=q,
         straggler_factor=2.0, max_seconds=10.0, held_out_frac=0.5)
-    # every task overran on its first try; at least one was re-run.
+    # t0 is always a straggler; it must have been re-run at least once.
     assert result.stragglers >= 1
     assert result.resumed >= 1, f"expected re-runs, got {result.resumed}"
-    # A queued straggler is re-run when a worker is free; some may legitimately
-    # still be queued when the run ends (no worker got to them), which is fine.
-    assert result.resumed <= result.stragglers
+    t0_renders = [r for tid, r in renders if tid == "t0"]
+    assert len(t0_renders) >= 2, (
+        "t0 must be rolled out at least twice (abandoned + re-run)")
+    assert "clue" not in t0_renders[0], (
+        "the first rollout measured the pre-clue head")
+    assert "clue" in t0_renders[-1], (
+        "the re-run must render the newer head, not re-run the old one")
 
 
 def test_async_without_resume_queue_keeps_old_behaviour():
@@ -172,7 +184,7 @@ def run(rendered, task):
 
 def test_a_chronically_slow_task_cannot_be_resumed_forever():
     """A task that is *always* a straggler gets bounded re-runs, not an
-    infinite resume loop starving the worker's fresh tasks (PR #198 review)."""
+    infinite resume loop starving the worker's fresh tasks."""
     import time
 
     def run(rendered, task):

From 837f08214c77f8ff4ed9f65f245551ec213c391a Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:01:17 +0800
Subject: [PATCH 6/7] docs(api): regenerate reference for the bounded pop()
 accounting

The api-reference sync check (test_the_api_reference_is_in_sync_with_the_package)
failed because the generated docs had no row for ResumeQueue.pop(), which now
documents its role in the max_attempts bound. Regenerate docs/api.md.
---
 docs/api.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/docs/api.md b/docs/api.md
index 0313b402..51c0be86 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -2947,6 +2947,7 @@ Turn-level checkpoints of timed-out rollouts (partial rollout).
 
 | method | what it does |
 |---|---|
+| `pop() -> Optional[ResumeItem]` | Pop the next resume. Counts as a resume for that task, feeding the `max_attempts` bound in `push`. |
 | `pop_for(task_ids: Sequence[str]) -> Optional[ResumeItem]` | Pop the first resume whose task belongs to `task_ids`. |
 | `push(item: ResumeItem) -> bool` | Queue `item`; `False` when this task has been resumed too often. |
 

From 6c94da6f69907037048455b534508a794cc1bf40 Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 13 Oct 2026 21:01:20 +0800
Subject: [PATCH 7/7] test(resume): associate the newer-head assertion with the
 resumed call

The regression passed even when the resumed call rendered the old state,
because a later ordinary rollout of the same task supplied the clue-bearing
render -- it proved re-execution happened, not that the resume used the
newer head. Each pop_for hand-off now flags the worker call that follows it
(thread-local, consumed by run), so the test asserts the resumed call
itself rendered the post-clue head. A negative control forces the resumed
call to render the stale state and proves the assertion fails.
---
 tests/test_ltraj_resume.py | 101 ++++++++++++++++++++++++++++++++-----
 1 file changed, 87 insertions(+), 14 deletions(-)

diff --git a/tests/test_ltraj_resume.py b/tests/test_ltraj_resume.py
index 97d44ecd..9da873fa 100644
--- a/tests/test_ltraj_resume.py
+++ b/tests/test_ltraj_resume.py
@@ -114,16 +114,23 @@ def test_resume_queue_pop_counts_toward_the_cap_too():
 
 def test_async_re_runs_a_straggler_against_the_newer_head():
     """A worker whose rollout overruns its estimate is re-run against a head
-    that has since advanced -- not re-executed for its own sake. The `run`
-    records the artifact each rollout saw, so the test can tell the two apart:
-    t0 is slow (a straggler) while another shard commits a clue; t0's re-run
-    must render the artifact *with* the clue, i.e. a newer version than the
-    one its abandoned rollout measured."""
+    that has since advanced -- not re-executed for its own sake.
+
+    Each `pop_for()` hand-off is associated with the worker call that follows
+    it (a thread-local flag consumed by `run`), so the assertion is about the
+    *resumed call specifically*, not any later ordinary rollout of the same
+    task: t0 is slow (a straggler) while another shard commits a clue, and the
+    resumed t0 call must render the artifact *with* the clue -- a newer version
+    than the one its abandoned rollout measured."""
+    import threading
     import time
-    renders = []                     # (task_id, rendered) per rollout
+    renders = []                     # (task_id, rendered, is_resumed) per call
+    resumed_flag = threading.local()
 
     def run(rendered, task):
-        renders.append((task.id, rendered))
+        is_resumed = getattr(resumed_flag, "pending", False)
+        resumed_flag.pending = False
+        renders.append((task.id, rendered, is_resumed))
         if task.id == "t0":
             time.sleep(0.2)          # t0 is always slow: a chronic straggler
         return "correct" if "clue" in rendered else "wrong"
@@ -141,6 +148,16 @@ def propose(rendered, task, output, reward):
     q = ResumeQueue()
     tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
 
+    # Flag the worker call that a `pop_for` hand-off immediately precedes.
+    real_pop_for = q.pop_for
+
+    def flagged_pop_for(task_ids):
+        item = real_pop_for(task_ids)
+        if item is not None:
+            resumed_flag.pending = True
+        return item
+
+    q.pop_for = flagged_pop_for
     result = async_evolve(
         tasks, lambda t, o: 1.0 if o == "correct" else 0.0,
         run=run, propose=propose,
@@ -150,13 +167,69 @@ def propose(rendered, task, output, reward):
     # t0 is always a straggler; it must have been re-run at least once.
     assert result.stragglers >= 1
     assert result.resumed >= 1, f"expected re-runs, got {result.resumed}"
-    t0_renders = [r for tid, r in renders if tid == "t0"]
-    assert len(t0_renders) >= 2, (
-        "t0 must be rolled out at least twice (abandoned + re-run)")
-    assert "clue" not in t0_renders[0], (
-        "the first rollout measured the pre-clue head")
-    assert "clue" in t0_renders[-1], (
-        "the re-run must render the newer head, not re-run the old one")
+    resumed_t0 = [r for tid, r, is_res in renders
+                  if tid == "t0" and is_res]
+    assert resumed_t0, (
+        "no t0 call was flagged as following a pop_for hand-off; the resume "
+        f"mechanism never fired: {renders!r}")
+    assert "clue" in resumed_t0[-1], (
+        "the resumed call itself must render the newer head; "
+        f"resumed t0 calls saw {resumed_t0!r}")
+
+
+def test_async_stale_resume_negative_control_fails():
+    """If the resumed call re-runs the *old* (pre-clue) state, the newer-head
+    assertion above must fail -- the test discriminates the resumed call's
+    version from any later ordinary rollout (review: the regression passed
+    even when the resumed call rendered the old state, because a later
+    ordinary t0 rollout supplied the clue-bearing render)."""
+    import threading
+    import time
+    import pytest
+    renders = []
+    resumed_flag = threading.local()
+
+    def run(rendered, task):
+        is_resumed = getattr(resumed_flag, "pending", False)
+        resumed_flag.pending = False
+        # The negative control: a resumed call renders the stale head, even
+        # though the artifact has since gained the clue.
+        if is_resumed and "clue" in rendered:
+            rendered = rendered.replace("clue", "")
+        renders.append((task.id, rendered, is_resumed))
+        if task.id == "t0":
+            time.sleep(0.2)
+        return "correct" if "clue" in rendered else "wrong"
+
+    def propose(rendered, task, output, reward):
+        if task.id != "t0" and "clue" not in rendered:
+            return "add a clue"
+        return None
+
+    estimator = DurationEstimator(prior=0.0001, min_samples=100)
+    q = ResumeQueue()
+    tasks = [Task(id=f"t{i}", prompt="q") for i in range(4)]
+    real_pop_for = q.pop_for
+
+    def flagged_pop_for(task_ids):
+        item = real_pop_for(task_ids)
+        if item is not None:
+            resumed_flag.pending = True
+        return item
+
+    q.pop_for = flagged_pop_for
+    result = async_evolve(
+        tasks, lambda t, o: 1.0 if o == "correct" else 0.0,
+        run=run, propose=propose,
+        strategy=AppendRules(),
+        n_workers=2, duration_estimator=estimator, resume_queue=q,
+        straggler_factor=2.0, max_seconds=10.0, held_out_frac=0.5)
+    resumed_t0 = [r for tid, r, is_res in renders
+                  if tid == "t0" and is_res]
+    assert resumed_t0, "premise: the resume mechanism fired"
+    with pytest.raises(AssertionError):
+        assert "clue" in resumed_t0[-1], (
+            "the resumed call must render the newer head")
 
 
 def test_async_without_resume_queue_keeps_old_behaviour():