feat(async): L-traj resume — a straggler is re-run against a newer head, not just counted

CHANGELOG.md

@@ -97,6 +97,28 @@ 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.** 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
+  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
+  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.
+
+  **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.

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,16 @@ def async_evolve(
     stall_patience: int = 50,
     duration_estimator: Optional["DurationEstimator"] = None,
     straggler_factor: float = 3.0,
+    #: 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
+    #: 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,
@@ -326,6 +336,17 @@ 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
+        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.
     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
@@ -514,6 +535,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 +693,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 +831,24 @@ 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:
+                        # `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}))
             consecutive, warned = 0, False                    # a clean rollout resets
             health.record_success()
             with counter_lock:
@@ -1225,6 +1274,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:

agentdescent/evolution.py

@@ -1297,8 +1297,13 @@ 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).
+    #: 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
@@ -1554,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,
@@ -1682,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

agentdescent/scheduler.py

@@ -272,44 +272,114 @@ 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:
     """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."""
-
-    def __init__(self, p90_multiplier: float = 2.0) -> None:
+    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.
+
+    **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). 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.
+        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
         # GIL-atomic but pop() + the emptiness check are not.
         self._lock = threading.Lock()
 
     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. 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.
+        Returns whether the item was accepted."""
         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:
+            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``.
+
+        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.
+
+        Popping counts as a resume for that task, which feeds the
+        :attr:`max_attempts` bound in :meth:`push`."""
         with self._lock:
-            return self._items.pop(0) if self._items else None
+            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 item
+            return None
 
     def __len__(self) -> int:
         return len(self._items)