diff --git a/.github/workflows/gitnexus-skill-evolution.yml b/.github/workflows/gitnexus-skill-evolution.yml index 8d02917c1..44b1895e6 100644 --- a/.github/workflows/gitnexus-skill-evolution.yml +++ b/.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: diff --git a/eval/tests/test_measure_evolution_cost.py b/eval/tests/test_measure_evolution_cost.py new file mode 100644 index 000000000..07f81baa3 --- /dev/null +++ b/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 diff --git a/eval/tests/test_workflow_bench.py b/eval/tests/test_workflow_bench.py index 889cdee78..e2c80ea1f 100644 --- a/eval/tests/test_workflow_bench.py +++ b/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")] diff --git a/eval/workflow_bench/measure_evolution_cost.py b/eval/workflow_bench/measure_evolution_cost.py new file mode 100644 index 000000000..7bd9daa20 --- /dev/null +++ b/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()) diff --git a/eval/workflow_bench/runner.py b/eval/workflow_bench/runner.py index f78394bf9..3be83916b 100644 --- a/eval/workflow_bench/runner.py +++ b/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",