perf(eval): pipeline graph setup and correct the wall-clock cost model

The evolution sweep paid `sanitize` + `analyze --pdg --index-only` for every
unique task SHA on the critical path, one at a time, with nothing overlapping.
`_run_sweep` now starts the next unpaid SHA's clone template and graph snapshot
on a prefetch thread as soon as the current task's cells are dispatched, so
every SHA but the first hides behind a paid session wave. The thread is joined
before that SHA is used and before the trees tempdir is torn down, and a
prefetch failure is recorded against the SHA exactly as an inline failure is.

Tasks whose cells are all reusable comparator rows are not prefetched: they
never build a graph, so priming one would be pure cost.

Adds `measure_evolution_cost.py`, the cost model behind these numbers. It reads
the review corpus, the evolve defaults, and the workflow's workers default —
it does not start a session. Its first version charged `copy_isolated_tree`
once per paid cell, serially. `run_cell` clones inside its own pool worker, so
the clones in a wave overlap and only one is on the critical path per wave;
the model now charges `ceil(cells / workers)` waves.

Estimated review generation at workers=3: cold 21570s, weekly 7710s.

Wall clock is quantised by `ceil(cells_per_task / workers)`. A cold review task
is 9 cells, so workers=4 buys the wall clock of workers=3 and pays host
contention for it. Documented in the workflow's rollout checklist.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.github/workflows/gitnexus-skill-evolution.yml

@@ -71,6 +71,9 @@
 #     Dispatch defaults to 3; pass workers=1 only to debug a contended host.
 #     Weekly generations reuse matching incumbent/CE cells from the previous
 #     artifact so the paid matrix is the new candidate, not a 54-cell replay.
+#     Wall clock is quantised by ceil(cells_per_task / workers), and a review
+#     task is 9 cells cold, so 4 costs host contention for exactly the wall
+#     clock of 3. The next step up that buys anything is 5 (3 waves -> 2).
 name: GitNexus skill evolution
 
 on:

eval/tests/test_measure_evolution_cost.py

@@ -0,0 +1,70 @@
+"""Cost model for the evolution wall clock: waves, not cells; overlap, not wishes."""
+
+from __future__ import annotations
+
+from workflow_bench.measure_evolution_cost import (
+    CELL_COPY_SECONDS,
+    GRAPH_ANALYZE_SECONDS,
+    MEAN_SESSION_SECONDS,
+    TEMPLATE_SANITIZE_SECONDS,
+    cell_setup_seconds,
+    graph_pipeline_enabled,
+    paid_cells_per_task,
+    pipelined_wall_seconds,
+    session_wall_seconds,
+    setup_wall_seconds,
+    unique_paid_shas,
+)
+
+TASKS = [
+    {"id": "a", "ref": "sha-1"},
+    {"id": "b", "ref": "sha-2"},
+    {"id": "c", "ref": "sha-1"},
+]
+
+
+def test_weekly_reuse_pays_only_candidate_cells():
+    assert paid_cells_per_task(TASKS, runs=3, weekly=True, reuse_enabled=True) == [3, 3, 3]
+    assert paid_cells_per_task(TASKS, runs=3, weekly=False, reuse_enabled=True) == [9, 9, 9]
+    assert paid_cells_per_task(TASKS, runs=3, weekly=True, reuse_enabled=False) == [9, 9, 9]
+
+
+def test_cell_clones_are_charged_once_per_wave():
+    # run_cell clones inside its own pool worker: three siblings clone at once.
+    assert cell_setup_seconds(9, 3, True) == 3 * CELL_COPY_SECONDS
+    assert cell_setup_seconds(9, 1, True) == 9 * CELL_COPY_SECONDS
+    assert cell_setup_seconds(0, 3, True) == 0
+    # Without templates the cell pays a full clone+sanitize, still per wave.
+    assert cell_setup_seconds(9, 3, False) == 3 * TEMPLATE_SANITIZE_SECONDS
+
+
+def test_setup_counts_each_sha_once_and_each_task_wave_once():
+    assert unique_paid_shas(TASKS, [9, 9, 9]) == 2
+    assert unique_paid_shas(TASKS, [9, 0, 0]) == 1
+    assert setup_wall_seconds(
+        unique_shas=2,
+        paid_per_task=[9, 9, 9],
+        workers=3,
+        clone_templates_enabled=True,
+    ) == 2 * (TEMPLATE_SANITIZE_SECONDS + GRAPH_ANALYZE_SECONDS) + 9 * CELL_COPY_SECONDS
+
+
+def test_pipelining_hides_every_sha_but_the_first():
+    paid = [9, 9, 9]
+    waves = [3 * MEAN_SESSION_SECONDS] * 3
+    serial = session_wall_seconds(paid, 3) + setup_wall_seconds(
+        unique_shas=2,
+        paid_per_task=paid,
+        workers=3,
+        clone_templates_enabled=True,
+    )
+    pipelined = pipelined_wall_seconds(
+        TASKS, paid, waves, workers=3, clone_templates_enabled=True
+    )
+    # sha-2's setup fits inside task a's session wave; sha-1 is already ready.
+    assert serial - pipelined == TEMPLATE_SANITIZE_SECONDS + GRAPH_ANALYZE_SECONDS
+
+
+def test_pipeline_flag_reads_the_runner_not_the_wish():
+    assert graph_pipeline_enabled("def _run_sweep(): pass") == 0
+    assert graph_pipeline_enabled("graph_prefetch = GraphPrefetch(...)") == 1

eval/tests/test_workflow_bench.py

@@ -5,6 +5,7 @@ import os
 import re
 import shlex
 import subprocess
+import threading
 from pathlib import Path
 
 import pytest
@@ -15,12 +16,15 @@ from workflow_bench.runner import (
     broken_incumbent_arms,
     build_parser,
     infra_error_record,
+    next_graph_prefetch_target,
     normalized_model_identifier,
     parse_shortstat,
+    prefetch_next_graph,
     render_report,
     savings,
     select_tasks,
     systemic_outage_streak,
+    task_has_planned_paid_cells,
 )
 
 
@@ -515,3 +519,94 @@ def test_run_evolution_script_is_the_shared_ci_and_local_entrypoint():
     assert "--include-expensive" in argv
     assert "claude-sonnet-5" not in argv
     assert printed.stderr  # rewrite notice goes to stderr
+
+
+def test_planned_paid_cells_treat_missing_reuse_as_paid():
+    task = {"id": "review-pr-2718-defect"}
+    assert task_has_planned_paid_cells(
+        task,
+        arms=["ce_review", "review", "candidate_review"],
+        runs=3,
+        reusable_rows={},
+        reuse_source=None,
+    )
+    reuse_source = Path("/tmp/seed")
+    rows = {
+        (task["id"], arm, run_idx): {}
+        for run_idx in range(3)
+        for arm in ("ce_review", "review", "candidate_review")
+    }
+    assert not task_has_planned_paid_cells(
+        task,
+        arms=["ce_review", "review", "candidate_review"],
+        runs=3,
+        reusable_rows=rows,
+        reuse_source=reuse_source,
+    )
+    del rows[(task["id"], "candidate_review", 0)]
+    assert task_has_planned_paid_cells(
+        task,
+        arms=["ce_review", "review", "candidate_review"],
+        runs=3,
+        reusable_rows=rows,
+        reuse_source=reuse_source,
+    )
+
+
+def test_next_graph_prefetch_skips_ready_shas_and_fully_reused_tasks(tmp_path: Path):
+    first = {"id": "review-a"}
+    second = {"id": "review-b"}
+    third = {"id": "review-c"}
+    reuse_source = tmp_path / "seed"
+    reused_second = {
+        (second["id"], arm, 0): {} for arm in ("ce_review", "review", "candidate_review")
+    }
+    target = next_graph_prefetch_target(
+        [
+            (first, {"repo_identity": "/repo", "resolved_sha": "aaa"}),
+            (second, {"repo_identity": "/repo", "resolved_sha": "bbb"}),
+            (third, {"repo_identity": "/repo", "resolved_sha": "ccc"}),
+        ],
+        arms=["ce_review", "review", "candidate_review"],
+        runs=1,
+        reusable_rows=reused_second,
+        reuse_source=reuse_source,
+        ready_keys={("/repo", "aaa")},
+    )
+    assert target is not None
+    task, binding, key = target
+    assert task["id"] == "review-c"
+    assert key == ("/repo", "ccc")
+    assert binding["resolved_sha"] == "ccc"
+
+
+def test_prefetch_next_graph_runs_ensure_on_a_background_thread(monkeypatch):
+    started = threading.Event()
+    seen: list[tuple[str, str]] = []
+
+    def fake_ensure(**kwargs):
+        seen.append(kwargs["graph_key"])
+        started.set()
+
+    monkeypatch.setattr("workflow_bench.runner.ensure_task_graph", fake_ensure)
+    cancel = threading.Event()
+    job = prefetch_next_graph(
+        task={"id": "review-b"},
+        binding={"repo_identity": "/repo", "resolved_sha": "bbb"},
+        graph_key=("/repo", "bbb"),
+        trees=Path("/tmp"),
+        task_asset_cache=None,
+        claude_bin="claude",
+        bwrap_bin="bwrap",
+        sandbox_backend="bwrap",
+        runtime_mounts=(),
+        clone_templates={},
+        clone_template_errors={},
+        graph_snapshots={},
+        graph_snapshot_errors={},
+        cancel_event=cancel,
+    )
+    assert job.key == ("/repo", "bbb")
+    assert started.wait(timeout=2)
+    job.join()
+    assert seen == [("/repo", "bbb")]

eval/workflow_bench/measure_evolution_cost.py

@@ -0,0 +1,380 @@
+#!/usr/bin/env python3
+"""Cheap cost model for the skill-evolution review generation.
+
+This is the ce-optimize measurement harness. It does not start Claude or
+replay Actions run 33962002890. Wall clock is session waves plus the
+pre-sweep setup the runner still pays before ``sweep_task_cells``:
+``make_worktree`` + ``sanitize_clone_for_hidden_oracles`` and
+``analyze --pdg --index-only``.
+
+Weekly assumes a matching seed so every reusable comparator cell is skipped.
+Cold assumes an empty seed. Candidate cells are never treated as reusable.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+# Agent-session mean from Actions run 33962002890 (review profile). The
+# runner builds graphs before cells start, so setup is added separately.
+MEAN_SESSION_SECONDS = 1140
+# Conservative mean for `analyze --pdg --index-only` of a sanitized
+# GitNexus snapshot on the evolution box. The hard timeout is 3600s.
+GRAPH_ANALYZE_SECONDS = 600
+# `git clone --no-local` plus sanitize_clone_for_hidden_oracles (repack /
+# prune / fsck). README called this "minutes" per cell before templates.
+TEMPLATE_SANITIZE_SECONDS = 180
+# copy_isolated_tree of an already-sanitized template (reflink or copy).
+# run_cell does this inside its pool worker, so it is a per-wave cost.
+CELL_COPY_SECONDS = 15
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+EVAL_ROOT = REPO_ROOT / "eval"
+REVIEW_TASKS = EVAL_ROOT / "workflow_bench" / "tasks.review.scenarios.yaml"
+EVOLVE_PY = EVAL_ROOT / "workflow_bench" / "evolve.py"
+RUNNER_PY = EVAL_ROOT / "workflow_bench" / "runner.py"
+ARTIFACTS_PY = EVAL_ROOT / "workflow_bench" / "runner_artifacts.py"
+REUSE_PY = EVAL_ROOT / "workflow_bench" / "comparator_reuse.py"
+WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gitnexus-skill-evolution.yml"
+
+REVIEW_ARMS = ("ce_review", "review", "candidate_review")
+CANDIDATE_ARM = "candidate_review"
+REUSABLE_ARMS = frozenset({"review", "ce_review"})
+
+SUITE_FILES = (
+    "tests/test_comparator_reuse.py",
+    "tests/test_evolve.py",
+    "tests/test_sanitized_graph.py",
+    "tests/test_workflow_bench_sessions.py",
+    "tests/test_session_progress.py",
+    "tests/test_measure_evolution_cost.py",
+    "tests/test_workflow_bench.py",
+)
+
+
+def _read(path: Path) -> str:
+    return path.read_text(encoding="utf-8")
+
+
+def review_tasks(text: str) -> list[dict[str, str]]:
+    tasks: list[dict[str, str]] = []
+    current: dict[str, str] | None = None
+    for raw in text.splitlines():
+        line = raw.strip()
+        if line.startswith("id:"):
+            if current is not None:
+                tasks.append(current)
+            current = {"id": line.split(":", 1)[1].strip()}
+        elif line.startswith("ref:") and current is not None:
+            current["ref"] = line.split(":", 1)[1].strip()
+    if current is not None:
+        tasks.append(current)
+    return tasks
+
+
+def evolve_default(name: str, text: str) -> int:
+    match = re.search(
+        rf'add_argument\("--{re.escape(name)}".*?default=(\d+)',
+        text,
+        flags=re.S,
+    )
+    if match is None:
+        raise ValueError(f"evolve.py is missing --{name} default")
+    return int(match.group(1))
+
+
+def workflow_dispatch_workers(text: str) -> int:
+    match = re.search(
+        r"^\s+workers:\n(?:.*\n)*?^\s+default: '(\d+)'",
+        text,
+        flags=re.M,
+    )
+    if match is None:
+        raise ValueError("workflow_dispatch workers default is missing")
+    return int(match.group(1))
+
+
+def feature_enabled() -> tuple[int, int]:
+    evolve = _read(EVOLVE_PY)
+    runner = _read(RUNNER_PY)
+    artifacts = _read(ARTIFACTS_PY)
+    reuse = int(
+        REUSE_PY.is_file()
+        and "--reuse-results" in evolve
+        and "select_reusable_comparator_rows" in runner
+        and "CANDIDATE" in _read(REUSE_PY)
+    )
+    templates = int(
+        "def copy_isolated_tree" in artifacts
+        and "clone_templates" in runner
+        and "clone_template" in runner
+    )
+    return reuse, templates
+
+
+def paid_cells_per_task(
+    tasks: list[dict[str, str]],
+    *,
+    runs: int,
+    weekly: bool,
+    reuse_enabled: bool,
+) -> list[int]:
+    per_task: list[int] = []
+    for _task in tasks:
+        paid = 0
+        for _run in range(runs):
+            for arm in REVIEW_ARMS:
+                if weekly and reuse_enabled and arm in REUSABLE_ARMS:
+                    continue
+                paid += 1
+        per_task.append(paid)
+    return per_task
+
+
+def unique_paid_shas(tasks: list[dict[str, str]], paid_per_task: list[int]) -> int:
+    seen: set[str] = set()
+    for task, paid in zip(tasks, paid_per_task, strict=True):
+        ref = task.get("ref", "")
+        if paid > 0 and ref:
+            seen.add(ref)
+    return len(seen)
+
+
+def sha_setup_seconds(clone_templates_enabled: bool) -> int:
+    if clone_templates_enabled:
+        return TEMPLATE_SANITIZE_SECONDS + GRAPH_ANALYZE_SECONDS
+    return GRAPH_ANALYZE_SECONDS
+
+
+def cell_setup_seconds(paid_cells: int, workers: int, clone_templates_enabled: bool) -> int:
+    """Per-cell clone cost, charged once per wave rather than once per cell.
+
+    ``run_cell`` clones inside its own pool worker, so the siblings in a wave
+    clone concurrently and only one clone sits on the critical path per wave.
+    """
+
+    if paid_cells < 1:
+        return 0
+    waves = math.ceil(paid_cells / workers)
+    if clone_templates_enabled:
+        return waves * CELL_COPY_SECONDS
+    return waves * TEMPLATE_SANITIZE_SECONDS
+
+
+def setup_wall_seconds(
+    *,
+    unique_shas: int,
+    paid_per_task: list[int],
+    workers: int,
+    clone_templates_enabled: bool,
+) -> int:
+    """Fully serial SHA setup plus per-wave clone work, task by task."""
+
+    if unique_shas < 1 or sum(paid_per_task) < 1:
+        return 0
+    cells = sum(
+        cell_setup_seconds(paid, workers, clone_templates_enabled) for paid in paid_per_task
+    )
+    return unique_shas * sha_setup_seconds(clone_templates_enabled) + cells
+
+
+def graph_pipeline_enabled(runner_text: str) -> int:
+    """True only when the runner prefetches the next SHA during paid sessions."""
+
+    return int(
+        bool(
+            re.search(
+                r"(graph_prefetch|prefetch_next_graph|pipeline_next_sha|prefetch_clone_template)",
+                runner_text,
+            )
+        )
+    )
+
+
+def pipelined_wall_seconds(
+    tasks: list[dict[str, str]],
+    paid_per_task: list[int],
+    session_per_task: list[int],
+    *,
+    workers: int,
+    clone_templates_enabled: bool,
+) -> int:
+    """Overlap the next unseen SHA's setup with the current task's session wave.
+
+    The first SHA still pays setup up front. Later SHAs hide behind the
+    previous task's paid sessions when that wave is longer than SHA setup.
+    Per-cell copies stay on the task's critical path.
+    """
+
+    setup = sha_setup_seconds(clone_templates_enabled)
+    elapsed = 0
+    ready: set[str] = set()
+    in_flight: tuple[str, int] | None = None
+    for index, (task, paid, session) in enumerate(zip(tasks, paid_per_task, session_per_task, strict=True)):
+        if paid <= 0:
+            continue
+        sha = task.get("ref", "")
+        if sha not in ready:
+            if in_flight is not None and in_flight[0] == sha:
+                elapsed = max(elapsed, in_flight[1])
+                ready.add(sha)
+                in_flight = None
+            else:
+                elapsed += setup
+                if sha:
+                    ready.add(sha)
+        session_start = elapsed
+        elapsed += session + cell_setup_seconds(paid, workers, clone_templates_enabled)
+        if in_flight is None:
+            for later_task, later_paid in zip(tasks[index + 1 :], paid_per_task[index + 1 :], strict=True):
+                later_sha = later_task.get("ref", "")
+                if later_paid > 0 and later_sha and later_sha not in ready:
+                    in_flight = (later_sha, session_start + setup)
+                    break
+        elif in_flight[1] <= elapsed:
+            ready.add(in_flight[0])
+            in_flight = None
+    return elapsed
+
+
+def session_wall_seconds(paid_per_task: list[int], workers: int) -> int:
+    if workers < 1:
+        raise ValueError("workers must be at least 1")
+    total = 0
+    for paid in paid_per_task:
+        if paid <= 0:
+            continue
+        total += math.ceil(paid / workers) * MEAN_SESSION_SECONDS
+    return total
+
+
+def _pytest_python() -> list[str]:
+    venv_python = EVAL_ROOT / ".venv" / "bin" / "python"
+    if venv_python.is_file():
+        return [str(venv_python)]
+    if (EVAL_ROOT / "uv.lock").is_file():
+        return ["uv", "run", "--locked", "--extra", "dev", "python"]
+    return [sys.executable]
+
+
+def suite_passed() -> int:
+    files = [name for name in SUITE_FILES if (EVAL_ROOT / name).is_file()]
+    if not files:
+        return 0
+    cmd = [
+        *_pytest_python(),
+        "-m",
+        "pytest",
+        *files,
+        "-q",
+        "--tb=no",
+        "--no-header",
+    ]
+    try:
+        completed = subprocess.run(
+            cmd,
+            cwd=EVAL_ROOT,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=240,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return 0
+    return int(completed.returncode == 0)
+
+
+def main() -> int:
+    tasks = review_tasks(_read(REVIEW_TASKS))
+    evolve = _read(EVOLVE_PY)
+    runs = evolve_default("runs", evolve)
+    promotion_min_runs = evolve_default("promotion-min-runs", evolve)
+    workers = workflow_dispatch_workers(_read(WORKFLOW))
+    reuse_enabled, clone_templates_enabled = feature_enabled()
+    pipeline_enabled = graph_pipeline_enabled(_read(RUNNER_PY))
+    weekly_paid = paid_cells_per_task(
+        tasks,
+        runs=runs,
+        weekly=True,
+        reuse_enabled=bool(reuse_enabled),
+    )
+    cold_paid = paid_cells_per_task(
+        tasks,
+        runs=runs,
+        weekly=False,
+        reuse_enabled=bool(reuse_enabled),
+    )
+    weekly_sessions = session_wall_seconds(weekly_paid, workers)
+    cold_sessions = session_wall_seconds(cold_paid, workers)
+    weekly_session_waves = [
+        0 if paid <= 0 else math.ceil(paid / workers) * MEAN_SESSION_SECONDS for paid in weekly_paid
+    ]
+    cold_session_waves = [
+        0 if paid <= 0 else math.ceil(paid / workers) * MEAN_SESSION_SECONDS for paid in cold_paid
+    ]
+    weekly_setup = setup_wall_seconds(
+        unique_shas=unique_paid_shas(tasks, weekly_paid),
+        paid_per_task=weekly_paid,
+        workers=workers,
+        clone_templates_enabled=bool(clone_templates_enabled),
+    )
+    cold_setup = setup_wall_seconds(
+        unique_shas=unique_paid_shas(tasks, cold_paid),
+        paid_per_task=cold_paid,
+        workers=workers,
+        clone_templates_enabled=bool(clone_templates_enabled),
+    )
+    if pipeline_enabled:
+        weekly_wall = pipelined_wall_seconds(
+            tasks,
+            weekly_paid,
+            weekly_session_waves,
+            workers=workers,
+            clone_templates_enabled=bool(clone_templates_enabled),
+        )
+        cold_wall = pipelined_wall_seconds(
+            tasks,
+            cold_paid,
+            cold_session_waves,
+            workers=workers,
+            clone_templates_enabled=bool(clone_templates_enabled),
+        )
+        weekly_setup = weekly_wall - weekly_sessions
+        cold_setup = cold_wall - cold_sessions
+    else:
+        weekly_wall = weekly_sessions + weekly_setup
+        cold_wall = cold_sessions + cold_setup
+    payload = {
+        "estimated_weekly_wall_seconds": weekly_wall,
+        "estimated_cold_wall_seconds": cold_wall,
+        "suite_passed": suite_passed(),
+        "promotion_min_runs": promotion_min_runs,
+        "review_task_count": len(tasks),
+        "candidate_cells": len(tasks) * runs,
+        "paid_weekly_cells": sum(weekly_paid),
+        "paid_cold_cells": sum(cold_paid),
+        "workers": workers,
+        "unique_task_shas": len({task.get("ref", "") for task in tasks if task.get("ref")}),
+        "reuse_enabled": reuse_enabled,
+        "clone_templates_enabled": clone_templates_enabled,
+        "graph_pipeline_enabled": pipeline_enabled,
+        "mean_session_seconds": MEAN_SESSION_SECONDS,
+        "session_weekly_seconds": weekly_sessions,
+        "session_cold_seconds": cold_sessions,
+        "setup_weekly_seconds": weekly_setup,
+        "setup_cold_seconds": cold_setup,
+        "graph_analyze_seconds": GRAPH_ANALYZE_SECONDS,
+        "template_sanitize_seconds": TEMPLATE_SANITIZE_SECONDS,
+    }
+    json.dump(payload, sys.stdout, separators=(",", ":"))
+    sys.stdout.write("\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

eval/workflow_bench/runner.py

@@ -1755,6 +1755,162 @@ def _comparator_reuse_expectation(
     )
 
 
+def task_has_planned_paid_cells(
+    task: Mapping[str, Any],
+    *,
+    arms: Sequence[str],
+    runs: int,
+    reusable_rows: Mapping[tuple[str, str, int], object],
+    reuse_source: Path | None,
+) -> bool:
+    """True when at least one planned cell is not a reusable comparator row."""
+
+    task_id = str(task["id"])
+    for run_idx in range(runs):
+        for arm in arms:
+            if reuse_source is None or (task_id, arm, run_idx) not in reusable_rows:
+                return True
+    return False
+
+
+def next_graph_prefetch_target(
+    remaining: Sequence[tuple[Mapping[str, Any], Mapping[str, Any]]],
+    *,
+    arms: Sequence[str],
+    runs: int,
+    reusable_rows: Mapping[tuple[str, str, int], object],
+    reuse_source: Path | None,
+    ready_keys: set[tuple[str, str]],
+) -> tuple[Mapping[str, Any], Mapping[str, Any], tuple[str, str]] | None:
+    """Next later task that still needs a clone template and sanitized graph."""
+
+    for task, binding in remaining:
+        if not task_has_planned_paid_cells(
+            task,
+            arms=arms,
+            runs=runs,
+            reusable_rows=reusable_rows,
+            reuse_source=reuse_source,
+        ):
+            continue
+        key = (str(binding["repo_identity"]), str(binding["resolved_sha"]))
+        if key in ready_keys:
+            continue
+        return task, binding, key
+    return None
+
+
+def ensure_task_graph(
+    *,
+    task: Mapping[str, Any],
+    repo: Path,
+    task_sha: str,
+    graph_key: tuple[str, str],
+    trees: Path,
+    task_asset_cache: TaskAssetCache,
+    claude_bin: Path | str,
+    bwrap_bin: Path | str,
+    sandbox_backend: str,
+    runtime_mounts: Sequence[ReadOnlyMount],
+    clone_templates: dict[tuple[str, str], tuple[Path, str]],
+    clone_template_errors: dict[tuple[str, str], BaseException],
+    graph_snapshots: dict[tuple[str, str], SanitizedGraphSnapshot],
+    graph_snapshot_errors: dict[tuple[str, str], BaseException],
+) -> None:
+    """Build one SHA's sanitized clone template and graph. Idempotent per key."""
+
+    if graph_key in graph_snapshots or graph_key in graph_snapshot_errors:
+        return
+    try:
+        validate_no_prebuilt_graph_assets(task)
+        if graph_key not in clone_templates and graph_key not in clone_template_errors:
+            template = make_worktree(repo, task_sha, trees)
+            template_head = sanitize_clone_for_hidden_oracles(template)
+            clone_templates[graph_key] = (template, template_head)
+        clone_template: Path | None = None
+        template_head: str | None = None
+        if graph_key in clone_templates:
+            clone_template, template_head = clone_templates[graph_key]
+        if graph_key in clone_template_errors:
+            graph_snapshot_errors[graph_key] = clone_template_errors[graph_key]
+            return
+        graph_snapshots[graph_key] = prepare_sanitized_graph(
+            task,
+            repo=repo,
+            resolved_sha=task_sha,
+            parent=trees,
+            cache=task_asset_cache,
+            claude_bin=claude_bin,
+            bwrap_bin=bwrap_bin,
+            sandbox_backend=sandbox_backend,
+            runtime_mounts=runtime_mounts,
+            clone_template=clone_template,
+            sanitized_head=template_head,
+        )
+    except (ManagedProcessError, OSError, SandboxError, RuntimeError, ValueError) as exc:
+        graph_snapshot_errors[graph_key] = exc
+        clone_template_errors.setdefault(graph_key, exc)
+
+
+@dataclass
+class GraphPrefetch:
+    """In-flight clone+graph build for a later task SHA."""
+
+    key: tuple[str, str]
+    thread: threading.Thread
+
+    def join(self) -> None:
+        self.thread.join()
+
+
+def prefetch_next_graph(
+    *,
+    task: Mapping[str, Any],
+    binding: Mapping[str, Any],
+    graph_key: tuple[str, str],
+    trees: Path,
+    task_asset_cache: TaskAssetCache,
+    claude_bin: Path | str,
+    bwrap_bin: Path | str,
+    sandbox_backend: str,
+    runtime_mounts: Sequence[ReadOnlyMount],
+    clone_templates: dict[tuple[str, str], tuple[Path, str]],
+    clone_template_errors: dict[tuple[str, str], BaseException],
+    graph_snapshots: dict[tuple[str, str], SanitizedGraphSnapshot],
+    graph_snapshot_errors: dict[tuple[str, str], BaseException],
+    cancel_event: threading.Event,
+) -> GraphPrefetch:
+    """Start clone+graph prep for the next unpaid SHA during paid sessions."""
+
+    repo = Path(binding["repo_identity"])
+    task_sha = str(binding["resolved_sha"])
+
+    def run() -> None:
+        if cancel_event.is_set():
+            return
+        print(f"[prefetch_next_graph] clone+graph for {task_sha}")
+        ensure_task_graph(
+            task=task,
+            repo=repo,
+            task_sha=task_sha,
+            graph_key=graph_key,
+            trees=trees,
+            task_asset_cache=task_asset_cache,
+            claude_bin=claude_bin,
+            bwrap_bin=bwrap_bin,
+            sandbox_backend=sandbox_backend,
+            runtime_mounts=runtime_mounts,
+            clone_templates=clone_templates,
+            clone_template_errors=clone_template_errors,
+            graph_snapshots=graph_snapshots,
+            graph_snapshot_errors=graph_snapshot_errors,
+        )
+
+    thread = threading.Thread(target=run, name="prefetch_next_graph", daemon=False)
+    thread.start()
+    return GraphPrefetch(key=graph_key, thread=thread)
+
+
 def _run_sweep(
     args: argparse.Namespace,
     *,
@@ -1840,151 +1996,180 @@ def _run_sweep(
         graph_snapshot_errors: dict[tuple[str, str], BaseException] = {}
         clone_templates: dict[tuple[str, str], tuple[Path, str]] = {}
         clone_template_errors: dict[tuple[str, str], BaseException] = {}
-        for task, task_binding, oracle_snapshot in zip(
-            tasks,
-            task_bindings,
-            oracle_snapshots,
-            strict=True,
-        ):
-            if outage_tripped or cancel_event.is_set():
-                break
-            repo = Path(task_binding["repo_identity"])
-            task_sha = task_binding["resolved_sha"]
-            per_arm: dict[str, list[dict[str, Any]]] = {a: [] for a in args.arms}
-            planned = [(run_idx, arm) for run_idx in range(args.runs) for arm in args.arms]
-            reused_records: list[tuple[int, str, dict[str, Any]]] = []
-            paid_cells: list[tuple[int, str]] = []
-            for run_idx, arm in planned:
-                prior = reusable_rows.get((task["id"], arm, run_idx))
-                if prior is None or reuse_source is None:
-                    paid_cells.append((run_idx, arm))
-                    continue
-                try:
-                    reused_records.append(
-                        (
-                            run_idx,
-                            arm,
-                            materialize_reused_row(prior, source_dir=reuse_source, dest_dir=out_dir),
-                        )
-                    )
-                except (OSError, SandboxError, ValueError) as exc:
-                    print(
-                        f"[{task['id']}][{arm}][run {run_idx}] comparator reuse "
-                        f"failed ({exc}); running a paid cell"
-                    )
-                    paid_cells.append((run_idx, arm))
+        graph_prefetch: GraphPrefetch | None = None
+        sweep_rows = list(zip(tasks, task_bindings, oracle_snapshots, strict=True))
 
-            asset_snapshot: TaskAssetSnapshot | None = None
-            asset_snapshot_error: BaseException | None = None
-            graph_key = (str(repo), task_sha)
-            graph_snapshot: SanitizedGraphSnapshot | None = graph_snapshots.get(graph_key)
-            graph_snapshot_error: BaseException | None = graph_snapshot_errors.get(graph_key)
-            clone_template: Path | None = None
-            template_head: str | None = None
-            if paid_cells:
-                try:
-                    validate_no_prebuilt_graph_assets(task)
-                    if graph_key not in clone_templates and graph_key not in clone_template_errors:
-                        template = make_worktree(repo, task_sha, Path(trees))
-                        template_head = sanitize_clone_for_hidden_oracles(template)
-                        clone_templates[graph_key] = (template, template_head)
-                    if graph_key in clone_templates:
-                        clone_template, template_head = clone_templates[graph_key]
-                    if graph_key in clone_template_errors and graph_snapshot_error is None:
-                        graph_snapshot_error = clone_template_errors[graph_key]
-                    if graph_snapshot is None and graph_snapshot_error is None:
-                        graph_snapshot = prepare_sanitized_graph(
+        def _join_graph_prefetch() -> None:
+            nonlocal graph_prefetch
+            if graph_prefetch is not None:
+                graph_prefetch.join()
+                graph_prefetch = None
+
+        try:
+            for index, (task, task_binding, oracle_snapshot) in enumerate(sweep_rows):
+                if outage_tripped or cancel_event.is_set():
+                    break
+                repo = Path(task_binding["repo_identity"])
+                task_sha = task_binding["resolved_sha"]
+                per_arm: dict[str, list[dict[str, Any]]] = {a: [] for a in args.arms}
+                planned = [(run_idx, arm) for run_idx in range(args.runs) for arm in args.arms]
+                reused_records: list[tuple[int, str, dict[str, Any]]] = []
+                paid_cells: list[tuple[int, str]] = []
+                for run_idx, arm in planned:
+                    prior = reusable_rows.get((task["id"], arm, run_idx))
+                    if prior is None or reuse_source is None:
+                        paid_cells.append((run_idx, arm))
+                        continue
+                    try:
+                        reused_records.append(
+                            (
+                                run_idx,
+                                arm,
+                                materialize_reused_row(prior, source_dir=reuse_source, dest_dir=out_dir),
+                            )
+                        )
+                    except (OSError, SandboxError, ValueError) as exc:
+                        print(
+                            f"[{task['id']}][{arm}][run {run_idx}] comparator reuse "
+                            f"failed ({exc}); running a paid cell"
+                        )
+                        paid_cells.append((run_idx, arm))
+
+                asset_snapshot: TaskAssetSnapshot | None = None
+                asset_snapshot_error: BaseException | None = None
+                graph_key = (str(repo), task_sha)
+                if graph_prefetch is not None and graph_prefetch.key == graph_key:
+                    _join_graph_prefetch()
+                if paid_cells and graph_key not in graph_snapshots and graph_key not in graph_snapshot_errors:
+                    ensure_task_graph(
+                        task=task,
+                        repo=repo,
+                        task_sha=task_sha,
+                        graph_key=graph_key,
+                        trees=Path(trees),
+                        task_asset_cache=task_asset_cache,
+                        claude_bin=args.claude_bin,
+                        bwrap_bin=bwrap_bin,
+                        sandbox_backend=sandbox_backend,
+                        runtime_mounts=runtime_mounts,
+                        clone_templates=clone_templates,
+                        clone_template_errors=clone_template_errors,
+                        graph_snapshots=graph_snapshots,
+                        graph_snapshot_errors=graph_snapshot_errors,
+                    )
+                graph_snapshot = graph_snapshots.get(graph_key)
+                graph_snapshot_error = graph_snapshot_errors.get(graph_key)
+                clone_template, template_head = clone_templates.get(graph_key, (None, None))
+                if paid_cells:
+                    try:
+                        # Prepared here, once, rather than lazily inside the first cell:
+                        # TaskAssetCache is a plain dict, so a lazy build would be a
+                        # read-then-write race the moment cells stop running serially.
+                        asset_snapshot = task_asset_cache.prepare(
                             task,
                             repo=repo,
                             resolved_sha=task_sha,
-                            parent=Path(trees),
-                            cache=task_asset_cache,
-                            claude_bin=args.claude_bin,
-                            bwrap_bin=bwrap_bin,
-                            sandbox_backend=sandbox_backend,
-                            runtime_mounts=runtime_mounts,
-                            clone_template=clone_template,
-                            sanitized_head=template_head,
+                            expected_dependency_binding=task_binding,
                         )
-                        graph_snapshots[graph_key] = graph_snapshot
-                except (ManagedProcessError, OSError, SandboxError, RuntimeError, ValueError) as exc:
-                    graph_snapshot_error = exc
-                    graph_snapshot_errors[graph_key] = exc
-                    clone_template_errors.setdefault(graph_key, exc)
-                try:
-                    # Prepared here, once, rather than lazily inside the first cell:
-                    # TaskAssetCache is a plain dict, so a lazy build would be a
-                    # read-then-write race the moment cells stop running serially.
-                    asset_snapshot = task_asset_cache.prepare(
-                        task,
-                        repo=repo,
-                        resolved_sha=task_sha,
-                        expected_dependency_binding=task_binding,
+                    except (OSError, SandboxError, ValueError) as exc:
+                        asset_snapshot_error = exc
+                cell_context = TaskCellContext(
+                    task=task,
+                    oracle_snapshot=oracle_snapshot,
+                    repo=repo,
+                    task_sha=task_sha,
+                    graph_snapshot=graph_snapshot,
+                    graph_snapshot_error=graph_snapshot_error,
+                    asset_snapshot=asset_snapshot,
+                    asset_snapshot_error=asset_snapshot_error,
+                    args=args,
+                    out_dir=out_dir,
+                    ce_plugin_snapshot=ce_plugin_snapshot,
+                    trees_dir=Path(trees),
+                    bwrap_bin=bwrap_bin,
+                    sandbox_backend=sandbox_backend,
+                    runtime_mounts=runtime_mounts,
+                    candidate_overlay=candidate_overlay,
+                    overlay_digest=overlay_digest,
+                    clone_template=clone_template,
+                    sanitized_head=template_head,
+                )
+
+                def announce(run_idx: int, arm: str) -> None:
+                    nonlocal started_cells
+                    started_cells += 1
+                    print(
+                        f"[{task['id']}][{arm}][run {run_idx}] starting "
+                        f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
                     )
-                except (OSError, SandboxError, ValueError) as exc:
-                    asset_snapshot_error = exc
-            cell_context = TaskCellContext(
-                task=task,
-                oracle_snapshot=oracle_snapshot,
-                repo=repo,
-                task_sha=task_sha,
-                graph_snapshot=graph_snapshot,
-                graph_snapshot_error=graph_snapshot_error,
-                asset_snapshot=asset_snapshot,
-                asset_snapshot_error=asset_snapshot_error,
-                args=args,
-                out_dir=out_dir,
-                ce_plugin_snapshot=ce_plugin_snapshot,
-                trees_dir=Path(trees),
-                bwrap_bin=bwrap_bin,
-                sandbox_backend=sandbox_backend,
-                runtime_mounts=runtime_mounts,
-                candidate_overlay=candidate_overlay,
-                overlay_digest=overlay_digest,
-                clone_template=clone_template,
-                sanitized_head=template_head,
-            )
 
-            def announce(run_idx: int, arm: str) -> None:
-                nonlocal started_cells
-                started_cells += 1
-                print(
-                    f"[{task['id']}][{arm}][run {run_idx}] starting "
-                    f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
-                )
+                def keep(run_idx: int, arm: str, record: dict[str, Any]) -> None:
+                    per_arm[arm].append(record)
+                    with results_path.open("a") as fh:
+                        # Redact any API token a session-error stderr_tail echoed
+                        # into error_detail before it enters the uploaded
+                        # results.jsonl artifact (transcripts are redacted; this
+                        # sink was not).
+                        fh.write(redact_text(json.dumps(record), credential_secrets(args)) + "\n")
+                    print(cell_progress_line(task["id"], arm, run_idx, record))
+                    failure = cell_failure_detail_line(task["id"], arm, run_idx, record, credential_secrets(args))
+                    if failure:
+                        print(failure)
 
-            def keep(run_idx: int, arm: str, record: dict[str, Any]) -> None:
-                per_arm[arm].append(record)
-                with results_path.open("a") as fh:
-                    # Redact any API token a session-error stderr_tail echoed
-                    # into error_detail before it enters the uploaded
-                    # results.jsonl artifact (transcripts are redacted; this
-                    # sink was not).
-                    fh.write(redact_text(json.dumps(record), credential_secrets(args)) + "\n")
-                print(cell_progress_line(task["id"], arm, run_idx, record))
-                failure = cell_failure_detail_line(task["id"], arm, run_idx, record, credential_secrets(args))
-                if failure:
-                    print(failure)
-
-            for run_idx, arm, record in reused_records:
-                started_cells += 1
-                print(
-                    f"[{task['id']}][{arm}][run {run_idx}] reused comparator "
-                    f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
-                )
-                keep(run_idx, arm, record)
-            if paid_cells:
-                outage_streak, outage_tripped = sweep_task_cells(
-                    paid_cells,
-                    workers=args.workers,
-                    run=partial(run_cell, cell_context),
-                    on_start=announce,
-                    on_record=keep,
-                    outage_streak=outage_streak,
-                    outage_limit=args.outage_streak,
-                    cancel_event=cancel_event,
-                )
-            results[task["id"]] = {a: aggregate(rs) for a, rs in per_arm.items() if rs}
+                for run_idx, arm, record in reused_records:
+                    started_cells += 1
+                    print(
+                        f"[{task['id']}][{arm}][run {run_idx}] reused comparator "
+                        f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
+                    )
+                    keep(run_idx, arm, record)
+                if paid_cells:
+                    if graph_prefetch is None and not cancel_event.is_set():
+                        ready_keys = (
+                            set(clone_templates)
+                            | set(clone_template_errors)
+                            | set(graph_snapshots)
+                            | set(graph_snapshot_errors)
+                        )
+                        target = next_graph_prefetch_target(
+                            [(later_task, later_binding) for later_task, later_binding, _ in sweep_rows[index + 1 :]],
+                            arms=args.arms,
+                            runs=args.runs,
+                            reusable_rows=reusable_rows,
+                            reuse_source=reuse_source,
+                            ready_keys=ready_keys,
+                        )
+                        if target is not None:
+                            later_task, later_binding, later_key = target
+                            graph_prefetch = prefetch_next_graph(
+                                task=later_task,
+                                binding=later_binding,
+                                graph_key=later_key,
+                                trees=Path(trees),
+                                task_asset_cache=task_asset_cache,
+                                claude_bin=args.claude_bin,
+                                bwrap_bin=bwrap_bin,
+                                sandbox_backend=sandbox_backend,
+                                runtime_mounts=runtime_mounts,
+                                clone_templates=clone_templates,
+                                clone_template_errors=clone_template_errors,
+                                graph_snapshots=graph_snapshots,
+                                graph_snapshot_errors=graph_snapshot_errors,
+                                cancel_event=cancel_event,
+                            )
+                    outage_streak, outage_tripped = sweep_task_cells(
+                        paid_cells,
+                        workers=args.workers,
+                        run=partial(run_cell, cell_context),
+                        on_start=announce,
+                        on_record=keep,
+                        outage_streak=outage_streak,
+                        outage_limit=args.outage_streak,
+                        cancel_event=cancel_event,
+                    )
+                results[task["id"]] = {a: aggregate(rs) for a, rs in per_arm.items() if rs}
+        finally:
+            _join_graph_prefetch()
 
     selection_report = [
         "## Run provenance",