sandbox pull: SWE-bench per-instance pre-pull with resume (es port)

- naming corrected to the Docker Hub truth es verified 500/500:
  swebench/sweb.eval.x86_64.{instance_id.lower(), __->_1776_}:latest
  (my repo-base rewrite was wrong; per-instance images ARE published)
- 'evalharness sandbox pull swe_bench_verified': concurrent pulls,
  resume state in <cache-dir>/swe_pull_state.json, --dry-run N,
  --retry-failed; self-contained (own dataset registry, no evalscope
  import)
- save/load intentionally omitted per user call (pull-only for now)

Co-Authored-By: Claude <noreply@anthropic.com>

evalharness/cli.py

@@ -95,6 +95,19 @@ def _cmd_data_unload(args) -> int:
     return 0
 
 
+def _cmd_sandbox_pull(args) -> int:
+    """evalharness sandbox pull swe_bench_verified [--max-workers 4]"""
+    from evalharness.sandbox.pull_swe import main as _pull_main
+
+    _ov = _overrides(args)
+    flag_names = ['dataset', 'max_workers', 'dry_run', 'retry_failed']
+    pull_args = [getattr(args, f, v) for f, v in
+                 zip(flag_names, ['swe_bench_verified', 4, 0, False])]
+    return _pull_main([pull_args[0], '--max-workers', str(pull_args[1])]
+                      + (['--dry-run', str(pull_args[2])] if pull_args[2] else [])
+                      + (['--retry-failed'] if pull_args[3] else []))
+
+
 def _cmd_sandbox_prefetch(args) -> int:
     from evalharness.data import get_dataset
     from evalharness.sandbox import docker_available, images_for_dataset, prefetch_images
@@ -1320,6 +1333,15 @@ def build_parser() -> argparse.ArgumentParser:
     _add_override_flags(p)
     p.set_defaults(func=_cmd_sandbox_prefetch)
 
+    p = bsub.add_parser('pull', help='pre-pull SWE-bench per-instance images '
+                                     '(resume + retry; port of es pull_swe_bench_images)')
+    p.add_argument('dataset', nargs='?', default='swe_bench_verified')
+    p.add_argument('--max-workers', type=int, default=4)
+    p.add_argument('--dry-run', type=int, default=0, help='only pull first N images')
+    p.add_argument('--retry-failed', action='store_true')
+    _add_override_flags(p)
+    p.set_defaults(func=_cmd_sandbox_pull)
+
     # ---- viz ----
     vz = sub.add_parser('viz', help='render saved EvalReport artifacts')
     zsub = vz.add_subparsers(dest='viz_command', required=True)

evalharness/data/datasets/swe_bench_verified.py

@@ -20,17 +20,11 @@ from ..spec import DatasetSpec
 def swe_bench_verified():
     def to_sample(record: dict) -> Sample:
         instance_id = record['instance_id']
-        # image naming: local swebench/ set uses {repo}_1776_{repo}-{num}; the
+        # EXACT Docker Hub naming (verified by es's pull state: 500/500):
-        # newer official layout is sweb.eval.x86_64.{repo}__{repo}-{num}
+        # swebench/sweb.eval.x86_64.{instance_id.lower() with __->_1776_}:latest
-        # OFFICIAL docker layout: repo-level BASE images
+        # -- per-instance images ARE published under the swebench/ namespace
-        # (swebench/sweb.eval.x86_64.{repo}) -- per-instance images are
+        img = ('swebench/sweb.eval.x86_64.'
-        # NEVER published; the harness builds them on top of the base
+               + instance_id.lower().replace('__', '_1776_') + ':latest')
-        _repo = instance_id.rsplit('-', 1)[0]
-        img = f'swebench/sweb.eval.x86_64.{_repo}:latest'
-        if '__' in instance_id:
-            repo, num = instance_id.rsplit('-', 1)
-            r = repo.split('__')[0]
-            img = f'swebench/sweb.eval.x86_64.{r}_1776_{r.split("__")[-1]}-{num}'
         return Sample(
             input=record['problem_statement'],
             target=record['patch'],  # gold patch (for oracle/oracle-check only)

evalharness/sandbox/pull_swe.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""SWE-bench 逐题镜像预拉取（断点续传、并发、失败重试）。
+
+移植自 es 的 pull_swe_bench_images.py，去掉 evalscope 依赖（用我们自己的
+数据集注册表），加进 sandbox 子命令：
+
+    evalharness sandbox pull swe_bench_verified --max-workers 4
+    evalharness sandbox pull swe_bench_verified --dry-run 3      # 只试前 3 个
+    evalharness sandbox pull swe_bench_verified --retry-failed
+
+进度文件: <cache-dir>/swe_pull_state.json（done/failed/skipped），
+Ctrl+C 后重跑自动跳过已完成的。
+"""
+
+import json
+import subprocess
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+
+_print_lock = threading.Lock()
+
+
+def _log(msg):
+    with _print_lock:
+        print(msg, flush=True)
+
+
+def swe_image_names(dataset_name: str):
+    """数据集 → 官方逐题镜像名列表（与 dataset 插件的命名完全一致）。"""
+    from evalharness import get_dataset
+
+    ds = get_dataset(dataset_name)
+    ds.materialize()
+    names = []
+    for s in ds:
+        if s.sandbox and s.sandbox.image:
+            names.append(s.sandbox.image)
+    return names
+
+
+def _state_path(cache_dir: Path) -> Path:
+    return cache_dir / 'swe_pull_state.json'
+
+
+def _load_state(cache_dir: Path) -> dict:
+    p = _state_path(cache_dir)
+    if not p.exists():
+        return {'done': [], 'failed': [], 'skipped': []}
+    try:
+        state = json.load(open(p, encoding='utf-8'))
+        for k in ('done', 'failed', 'skipped'):
+            state.setdefault(k, [])
+        return state
+    except Exception:
+        return {'done': [], 'failed': [], 'skipped': []}
+
+
+def _save_state(cache_dir: Path, state: dict):
+    p = _state_path(cache_dir)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    tmp = p.with_suffix('.tmp')
+    tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')
+    tmp.replace(p)
+
+
+def _pull_one(image: str) -> tuple:
+    """docker pull 单个镜像，实时透传输出。返回 (returncode, 尾部输出)。"""
+    proc = subprocess.Popen(['docker', 'pull', image],
+                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                            text=True, bufsize=1)
+    lines = []
+    for line in proc.stdout:
+        line = line.rstrip('\n')
+        lines.append(line)
+        _log(f'   {line}')
+        if len(lines) > 50:
+            lines.pop(0)
+    proc.wait()
+    return proc.returncode, '\n'.join(lines)
+
+
+def run_pull(dataset: str, cache_dir: str, max_workers: int = 4,
+             dry_run: int = 0, retry_failed: bool = False):
+    cdir = Path(cache_dir)
+    state = _load_state(cdir)
+    if retry_failed:
+        n = len(state['failed'])
+        state['failed'] = []
+        _log(f'🔄 重试 {n} 个之前失败的镜像')
+
+    names = swe_image_names(dataset)
+    if dry_run:
+        names = names[:dry_run]
+    local = set()
+    r = subprocess.run(['docker', 'images', '--format', '{{.Repository}}:{{.Tag}}'],
+                       capture_output=True, text=True)
+    if r.returncode == 0:
+        local = {l.strip() for l in r.stdout.splitlines() if l.strip()}
+
+    todo = []
+    for nm in names:
+        if nm in state['done'] or nm in state['skipped']:
+            continue
+        if nm in local:
+            state['skipped'].append(nm)
+            _save_state(cdir, state)
+            continue
+        todo.append(nm)
+    _log(f'共 {len(names)} 个镜像: 已完成/本地 {len(names) - len(todo)}, '
+         f'待拉 {len(todo)}, 并发 {max_workers}')
+
+    def work(nm):
+        _log(f'⬇ START {nm}')
+        rc, out = _pull_one(nm)
+        if rc == 0:
+            state['done'].append(nm)
+        else:
+            state['failed'].append(nm)
+        _save_state(cdir, state)
+        return nm, rc, out[-300:] if rc else ''
+
+    fails = []
+    with ThreadPoolExecutor(max_workers=max_workers) as pool:
+        futs = {pool.submit(work, nm): nm for nm in todo}
+        for i, fut in enumerate(as_completed(futs), 1):
+            nm, rc, out = fut.result()
+            mark = '✓' if rc == 0 else '✗'
+            _log(f'[{i}/{len(todo)}] {mark} {nm}')
+            if rc:
+                fails.append((nm, out))
+
+    _log(f'\n完成: {len(state["done"])} 成功, {len(state["skipped"])} 跳过, '
+         f'{len(state["failed"])} 失败 (进度存 {_state_path(cdir)})')
+    for nm, out in fails[:5]:
+        _log(f'  失败样例 {nm}: {out[:150]}')
+    return 0 if not fails else 1
+
+
+def main(argv) -> int:
+    import argparse
+
+    p = argparse.ArgumentParser(description='Pre-pull SWE-bench per-instance '
+                                            'images with resume')
+    p.add_argument('dataset', nargs='?', default='swe_bench_verified')
+    p.add_argument('--max-workers', type=int, default=4)
+    p.add_argument('--dry-run', type=int, default=0,
+                   help='only pull the first N images')
+    p.add_argument('--retry-failed', action='store_true')
+    p.add_argument('--cache-dir', default='')
+    a = p.parse_args(argv)
+
+    cache = a.cache_dir
+    if not cache:
+        import os
+        cache = os.environ.get('EVALHARNESS_CACHE',
+                               str(Path.home() / '.cache/evalharness'))
+    return run_pull(a.dataset, cache, a.max_workers, a.dry_run, a.retry_failed)
+
+
+if __name__ == '__main__':
+    sys.exit(main(sys.argv[1:]))