README: agent-env extra dependencies (tau2-bench engine, swebench)

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

README.md

@@ -148,6 +148,18 @@ rep.save('gsm8k.report.json')
 # notebook / async 环境用 await evalharness.arun(...)
 ```
 
+### Agent 环境的额外依赖
+
+`tau2_bench` 需要官方引擎（**PyPI 上的 `tau2` 是同名无关项目，别装错**）：
+
+```bash
+pip install -e /path/to/tau2-bench          # 本机源码（es 仓库 tools/ 下有）
+# 或： pip install git+https://github.com/sierra-research/tau2-bench
+```
+
+`swe_bench_verified` 需要 `pip install swebench` + 逐题 Docker 镜像（见 §8）。
+`bfcl_v3` / `general_fc` 无额外依赖。
+
 ## 5. 参数速查
 
 | 参数 | 作用 |

evalharness/cli.py

@@ -546,7 +546,17 @@ def _compose_judge_spec(args):
     url = getattr(args, 'judge_api_url', '') or ''
     provider = getattr(args, 'judge_provider', 'openai-chat') or 'openai-chat'
     internal = {'openai-chat': 'openai', 'openai-pool': 'openai-pool'}.get(provider, provider)
-    if url and judge and '/' not in judge:
+    if judge and not url:
+        if '/' not in judge and '?' not in judge:
+            raise SystemExit(
+                f'--judge-model {judge!r} needs --judge-api-url (or pass a '
+                f'full spec like openai/http://host:8000/v1?model)')
+        return judge   # full legacy spec, used as-is
+    if url and judge:
+        # url+model ALWAYS combines (same as the main model flags): a bare
+        # model name may itself contain slashes (/data/hf_models/...), which
+        # the old 'no slash' heuristic misread as a full spec and passed
+        # through un-prefixed
         judge = f'{internal}/{url.rstrip("/")}?{judge}'
     return judge or None
 
@@ -640,7 +650,17 @@ def _compose_judge_spec(args):
     url = getattr(args, 'judge_api_url', '') or ''
     provider = getattr(args, 'judge_provider', 'openai-chat') or 'openai-chat'
     internal = {'openai-chat': 'openai', 'openai-pool': 'openai-pool'}.get(provider, provider)
-    if url and judge and '/' not in judge:
+    if judge and not url:
+        if '/' not in judge and '?' not in judge:
+            raise SystemExit(
+                f'--judge-model {judge!r} needs --judge-api-url (or pass a '
+                f'full spec like openai/http://host:8000/v1?model)')
+        return judge   # full legacy spec, used as-is
+    if url and judge:
+        # url+model ALWAYS combines (same as the main model flags): a bare
+        # model name may itself contain slashes (/data/hf_models/...), which
+        # the old 'no slash' heuristic misread as a full spec and passed
+        # through un-prefixed
         judge = f'{internal}/{url.rstrip("/")}?{judge}'
     return judge or None
 
@@ -745,6 +765,9 @@ def _cmd_eval_run(args) -> int:
         try:
             if _shared_reporter is not None and total_runs > 1:
                 _shared_reporter.set_bench_tag(f'[{i + 1}/{total_runs}]')
+            if _shared_reporter is not None:
+                _shared_reporter.begin_bench(name)   # clear the previous
+                # bench's stale scoring bar before this one's dataset load
 
             def _emit(msg, _i=i, _n=name):
                 if _shared_reporter is not None:
@@ -765,8 +788,12 @@ def _cmd_eval_run(args) -> int:
             # YAML config: per-bench generation params, AUTO-LOADED
             # (single .yaml in config/ = the default; --config overrides)
             bench_cfg = _load_bench_cfg(args, name)
+            # env: per-bench from YAML, CLI --env as fallback/default
+            _env_cfg = bench_cfg.pop('env', '') or ''
+            if _env_cfg:
+                args.env = _env_cfg
             # strip non-generation keys (they go to run_eval kwargs)
-            for k in ('judge', 'judge_url', 'env', 'max_turns',
+            for k in ('judge', 'judge_url', 'max_turns',
                       'limit', 'limit_per_task', 'concurrency'):
                 bench_cfg.pop(k, None)
 
@@ -912,7 +939,8 @@ def _cmd_eval_run(args) -> int:
                         'max': round(max(_scores), 4),
                         'std': round(_var ** 0.5, 4),
                     }
-                    _primary = next(iter(report.metrics), '')
+                    _primary = next((k for k in report.metrics
+                                         if k != 'extraction_failure_rate'), '')
                     if _primary:
                         report.metrics[f'{_primary}_last_run'] = report.metrics[_primary]
                         report.metrics[_primary] = _mean
@@ -953,7 +981,11 @@ def _cmd_eval_run(args) -> int:
                 if console is None:
                     print(render(report, style=args.style))
             all_reports.append(report)
-            primary = next(iter(report.metrics), '')
+            # primary metric = first REAL metric; extraction_failure_rate
+            # is diagnostics (mrcr showed '0.0% extraction_failure_rate' as
+            # its score in the summary because dict order put it first)
+            primary = next((k for k in report.metrics
+                            if k != 'extraction_failure_rate'), '')
             groups = {k: v for k, v in report.metric_groups.items()
                       if isinstance(v, dict) and k not in ('run_info',)
                       and not k.startswith('agg_error')}

evalharness/config/default.yaml

@@ -54,7 +54,9 @@ openai_mrcr:
   max_input_tokens: 128000
 bfcl_v3:
   max_tokens: 4096
+  env: bfcl_mock          # agent 模式：工具调用轨迹 + 官方 AST 判分
 general_fc:
   max_tokens: 4096
 tau2_bench:
   max_tokens: 16384
+  env: tau2_official      # agent 模式：官方引擎

evalharness/data/datasets/swe_bench_verified.py

@@ -22,7 +22,11 @@ def swe_bench_verified():
         instance_id = record['instance_id']
         # image naming: local swebench/ set uses {repo}_1776_{repo}-{num}; the
         # newer official layout is sweb.eval.x86_64.{repo}__{repo}-{num}
-        img = f'sweb.eval.x86_64.{instance_id}'
+        # OFFICIAL docker layout: repo-level BASE images
+        # (swebench/sweb.eval.x86_64.{repo}) -- per-instance images are
+        # NEVER published; the harness builds them on top of the base
+        _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]

evalharness/data/loader.py

@@ -13,6 +13,7 @@ Supported sources:
 import csv
 import json
 import os
+import subprocess
 import sys
 import time
 import re
@@ -261,6 +262,36 @@ def _download_with_progress(resp, out, filename: str) -> None:
         out.write(chunk)
 
 
+def _aria2_fetch(url: str, dest_dir: Path, name: str) -> bool:
+    """Multi-connection fetch via aria2c when available.
+
+    The mirror's CDN assigns wildly different edges per connection (a
+    190MB file ran at 6KB/s on one connection and 4.8MB/s on a fresh
+    one). aria2c splits the file into segments -- each segment rolls its
+    own edge dice -- and --lowest-speed-limit self-heals a stalled piece
+    by re-opening it. Returns False (caller falls back to urllib) when
+    aria2c is missing or fails.
+    """
+    import shutil as _sh
+
+    if not _sh.which('aria2c'):
+        return False
+    tmp = dest_dir / (name + '.aria2.part')
+    cmd = ['aria2c', '-x', '8', '-s', '8', '-k', '4M', '--continue=true',
+           '--file-allocation=none', '--console-log-level=warn',
+           '--summary-interval=0', '--retry-wait=3', '--max-tries=5',
+           '--lowest-speed-limit=50K', '--timeout=20',
+           '-d', str(dest_dir), '-o', tmp.name, url]
+    r = subprocess.run(cmd, capture_output=True, text=True)
+    if r.returncode == 0 and tmp.exists():
+        os.replace(tmp, dest_dir / name)
+        return True
+    tmp.unlink(missing_ok=True)
+    tmp2 = dest_dir / (tmp.name + '.aria2')
+    tmp2.unlink(missing_ok=True)
+    return False
+
+
 def _ms_download(repo: str, path: str, dest_dir: Path) -> Path:
     """Download one repo file into the raw cache (content-addressed, reused)."""
     dest = dest_dir / os.path.basename(path)
@@ -271,6 +302,8 @@ def _ms_download(repo: str, path: str, dest_dir: Path) -> Path:
         dest.unlink()
     dest_dir.mkdir(parents=True, exist_ok=True)
     url = f'{_MS_API}/{repo}/repo?Revision=master&FilePath={path}'
+    if _aria2_fetch(url, dest_dir, dest.name):
+        return dest_dir / dest.name
     tmp = dest.with_name(dest.name + f'.part-{os.getpid()}')
     req = urllib.request.Request(url, headers={'User-Agent': 'evalharness/0.1'})
     with urllib.request.urlopen(req, timeout=600) as resp, open(tmp, 'wb') as out:
@@ -411,6 +444,8 @@ def _hf_download(repo: str, path: str, dest_dir: Path) -> Path:
         dest.unlink()
     dest_dir.mkdir(parents=True, exist_ok=True)
     url = f'{_hf_base()}/datasets/{repo}/resolve/main/{path}'
+    if _aria2_fetch(url, dest_dir, dest.name):
+        return dest_dir / dest.name
     tmp = dest.with_name(dest.name + f'.part-{os.getpid()}')
     req = urllib.request.Request(url, headers={'User-Agent': 'evalharness/0.1'})
     with urllib.request.urlopen(req, timeout=1800) as resp, open(tmp, 'wb') as out:
@@ -422,11 +457,32 @@ def _hf_download(repo: str, path: str, dest_dir: Path) -> Path:
 def _load_from_hf_raw(spec: DatasetSpec, raw_dir: Optional[Path] = None) -> List[Dict[str, Any]]:
     import hashlib
 
+    # the mirror's tree API degrades at night and returns TRUNCATED listings
+    # (a repo with 9 parquets listed as [.gitattributes, README.md]); the
+    # blobs from a previous run are already in the shared store -- use them
+    blob_dir = get_cache_root() / '.raw' / hashlib.md5(spec.source.encode()).hexdigest()[:10]
+
+    def _cached_blobs() -> List[str]:
+        if not blob_dir.exists():
+            return []
+        have = [f.name for f in blob_dir.iterdir()
+                if os.path.splitext(f)[1] in _SUPPORTED_EXTS]
+        pref = [f for f in have if f.startswith(spec.subset)] or have
+        return pref
+
     files = _hf_list_files(spec.source)
-    if not files:
+    selected = _hf_match_files(spec, files) if files else []
-        raise FileNotFoundError(f'no files found on HF dataset {spec.source!r}')
-    selected = _hf_match_files(spec, files)
     if not selected:
+        cached = _cached_blobs()
+        if cached:
+            print(f'· listing unavailable/degraded -- using {len(cached)} '
+                  f'cached file(s) from {blob_dir}', flush=True)
+            records: List[Dict[str, Any]] = []
+            for f in sorted(cached):
+                records.extend(_read_file(str(blob_dir / f)))
+            if raw_dir is not None:
+                _link_or_copy_all([blob_dir / f for f in sorted(cached)], raw_dir)
+            return records
         raise FileNotFoundError(
             f'no data file for subset={spec.subset!r} split={spec.split!r} in '
             f'HF {spec.source!r}. Available (first 10): {files[:10]}'

evalharness/eval/recipes/agent.py

@@ -55,8 +55,11 @@ def bigcodebench():
         name='bigcodebench',
         extract='code_any',
         scorers={'pass': {'name': 'execution', 'harness': _bcb_harness,
-                          # official sandbox image (bundles every task's deps)
+                          # official sandbox image (bundles every task's deps);
-                          'image': 'bigcodebench/bigcodebench-evaluate:latest',  # official hub image, same as evalscope
+                          # its ENTRYPOINT is the official evaluate CLI which
+                          # swallows our runner -> override with plain python3
+                          'image': 'bigcodebench/bigcodebench-evaluate:latest',
+                          'entrypoint': 'python3',
                           'sandbox': 'docker', 'timeout_s': 120}},
         aggregators={'pass': 'pass_at_k'},
         exec_workers=12,

evalharness/eval/runner.py

@@ -120,6 +120,15 @@ def evaluate(
         each sample gets its own container/workdir)."""
         result = _shell(sample, pred)
         raw = result.raw_prediction
+        # hybrid-thinking backends sometimes inline the reasoning channel
+        # into content wrapped in <think>...</think> (or leave a stray
+        # closer): extractors then fish answers out of reasoning text
+        # ('3</think>Let me analyze...'). Strip the blocks before extract.
+        if '<think>' in raw or '</think>' in raw:
+            import re as _re0
+
+            raw = _re0.sub(r'<think>.*?</think>', '', raw, flags=_re0.S)
+            raw = raw.replace('</think>', '')
         try:
             if bp is not None and sample.sandbox and sample.sandbox.image:
                 bp.ensure(sample.sandbox.image)  # wait only if this one still pulling

evalharness/eval/scorer.py

@@ -342,7 +342,8 @@ def execution(pred: str, target, sample: Sample, ctx: ScoreContext):
     files = harness(sample, pred or '')
     result = sbx.exec(files, entry=ctx.params.get('entry', 'main.py'),
                       timeout_s=ctx.params.get('timeout_s', 30),
-                      image=ctx.params.get('image', ''))
+                      image=ctx.params.get('image', ''),
+                      entrypoint=ctx.params.get('entrypoint', ''))
     ok = result.ok
     return ({'pass': 1.0} if ok else {'pass': 0.0}), {'pass': {
         'exit_code': result.exit_code,

evalharness/model/runner.py

@@ -821,7 +821,8 @@ async def run_eval(
                 status_callback('Scores cached in checkpoint -- replaying '
                                 '(no scorers run; --rescore re-evaluates)')
 
-    if judge is None and judge_spec and _records is None:
+    if judge is None and judge_spec and _records is None \
+            and _recipe_needs_judge(recipe):
         if status_callback:
             status_callback('loading judge model')
         judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key)
@@ -829,6 +830,13 @@ async def run_eval(
 
     if status_callback:
         status_callback('Scoring predictions against the benchmark recipe')
+    # retarget the bar to scoring IMMEDIATELY (0/N): waiting for the first
+    # on_scored left the stale GENERATION counters (100%, +N new) on screen
+    # through minute-long preflights (image pulls) -- reading as 'done'
+    if progress_reporter is not None and _records is None:
+        _ss0 = getattr(progress_reporter, 'set_scoring', None)
+        if _ss0 is not None:
+            _ss0(0, len(samples))
     _meta = {'gen_input_tokens': usage.input_tokens,
              'gen_output_tokens': usage.output_tokens,
              'gen_total_tokens': usage.total_tokens,
@@ -886,6 +894,20 @@ async def run_eval(
     return report
 
 
+def _recipe_needs_judge(recipe) -> bool:
+    """Only construct the judge when a scorer actually consumes it --
+    rule-based benches (longbench_v2 etc.) never touch a judge, and
+    building one there crashed on malformed specs for no benefit."""
+    try:
+        for spec in (recipe.scorers or {}).values():
+            p = spec if isinstance(spec, dict) else {}
+            if p.get('name') in ('llm_judge', 'judge'):
+                return True
+    except Exception:
+        pass
+    return False
+
+
 def _env_user_adapter(spec: str):
     """Build (once per spec) the separate USER-simulator adapter for env
     benches (tau2 strong-user parity mode)."""

evalharness/progress/rich_terminal.py

@@ -162,6 +162,26 @@ class RichTerminalProgress:
             self.heartbeat_task.cancel()
         self.heartbeat_task = asyncio.create_task(self._heartbeat())
 
+    def begin_bench(self, name: str):
+        """New benchmark starting: immediately re-label the bar to its
+        loading phase. Without this the PREVIOUS bench's stale scoring
+        state (e.g. a failed '0/2') stayed on screen through the whole
+        dataset download of the next one."""
+        if self.disabled or self.task_id is None:
+            return
+        self.bench_name = name
+        self._last_phase = 'loading'
+        self._scoring_for = None
+        self.started = time.monotonic()
+        self.inflight = 0
+        self.restored = 0
+        self.admitted = None
+        self.progress.update(
+            self.task_id,
+            description=f'[green]{self.bench_tag}{name} · loading[/green]',
+            total=1, completed=0, new='', rate='0.00', inflight=0,
+            cur='0s', elapsed='0s', eta='-', retries=0)
+
     def set_scoring(self, done: int, total: int):
         """Retarget the SAME bar to the scoring phase: generation is finished
         and its filled 100% state is stale -- now the bar refills with judged

evalharness/sandbox/base.py

@@ -65,6 +65,8 @@ class Sandbox:
         mounts: Optional[Dict[str, str]] = None,
         timeout_s: int = 60,
         image: str = '',
+        entrypoint: str = '',   # docker-only override; accepted for
+                                # signature parity (local runs a plain python)
     ) -> ExecResult:
         """Run ``python <entry>`` with ``files`` (name->content) in isolation.
 

evalharness/sandbox/docker.py

@@ -45,11 +45,13 @@ def ensure_image(img: str) -> None:
         return
     except RuntimeError as e:
         raise RuntimeError(
-            f'sandbox image {img!r} is not available: not local, and every '
+            f'沙箱镜像 {img!r} 不可用：本地没有，全部镜像源尝试失败。'
-            f'mirror failed. {str(e)[:200]}. '
+            f'{str(e)[:150]}\n'
-            'Fix: pull/build it manually (for bigcodebench the official image '
+            '需要自己加载镜像（任选其一），然后重跑加 --rescore：\n'
-            'is bigcodebench/bigcodebench-evaluate:latest), then re-run with '
+            f'  A. 有外网的机器:  docker pull {img} && docker save {img} | gzip > img.tgz\n'
-            '--rescore to score the cached predictions.') from e
+            f'     拷回本机后:    docker load < img.tgz\n'
+            f'  B. 本机网络恢复后: docker pull {img}\n'
+            '预测都在 checkpoint 里，镜像就位后纯判分即可。') from e
 
 
 @register_sandbox('docker')
@@ -65,6 +67,7 @@ class DockerSandbox(Sandbox):
         mounts: Optional[Dict[str, str]] = None,
         timeout_s: int = 60,
         image: str = '',
+        entrypoint: str = '',
     ) -> ExecResult:
         img = image or self.DEFAULT_EXEC_IMAGE
         with tempfile.TemporaryDirectory(prefix='eh-sbx-') as host_dir:
@@ -90,8 +93,17 @@ class DockerSandbox(Sandbox):
                     out_host = Path(hpath).expanduser()
                     out_host.mkdir(parents=True, exist_ok=True)
                     cmd += ['-v', f'{out_host}:{cpath}:rw']
-            runner = ['python', f'/work/{entry}'] if entry.endswith('.py') \
+            if entrypoint:
-                else ['sh', f'/work/{entry}']
+                # images with an official-evaluator ENTRYPOINT (bigcodebench:
+                # python3 -m bigcodebench.evaluate) swallow our runner as
+                # CLI args -> override it. The entrypoint IS the interpreter
+                # now: pass only the script path, else 'python3 python
+                # /work/main.py' tries to open a file named 'python'
+                cmd += ['--entrypoint', entrypoint]
+                runner = [f'/work/{entry}']
+            else:
+                runner = ['python', f'/work/{entry}'] if entry.endswith('.py') \
+                    else ['sh', f'/work/{entry}']
             # Named + retried runs. A timed-out/killed `docker run` only kills
             # the CLI client -- the container lives on (--rm fires on EXIT)
             # and leaked containers slowly drown the daemon; the name gives

evalharness/sandbox/local.py

@@ -25,6 +25,8 @@ class LocalSandbox(Sandbox):
         mounts: Optional[Dict[str, str]] = None,
         timeout_s: int = 60,
         image: str = '',
+        entrypoint: str = '',   # docker-only override; accepted for
+                                # signature parity (local runs a plain python)
     ) -> ExecResult:
         with tempfile.TemporaryDirectory(prefix='eh-local-') as td:
             work = Path(td)

evalharness/sandbox/prefetch.py

@@ -19,15 +19,25 @@ from ..data.dataset import Dataset
 # CN mirrors tried in order before/alongside the daemon's configured mirrors.
 # Some namespaces (e.g. swebench/*) are blocked by individual CN mirrors, so we
 # fall through: daemon default -> 1ms.run -> baidubce -> sjtug.
+# Six mainstream fallbacks (trimmed from 16 per user call: the long tail
+# died anyway). History on this host: daocloud and 1ms.run have each
+# delivered the 9GB bigcodebench image; the rest are the usual suspects.
 _CN_MIRROR_FALLBACKS = [
     '{img}',  # daemon default (uses its own registry-mirrors config)
     'docker.m.daocloud.io/{img}',
     'docker.1ms.run/{img}',
-    'mirror.baidubce.com/{img}',
+    'docker.1panel.live/{img}',
-    'docker.mirrors.sjtug.sjtu.edu.cn/{img}',
     'hub.rat.dev/{img}',
+    'docker.mirrors.ustc.edu.cn/{img}',
 ]
 
+# 用户/环境可整体覆盖（逗号分隔模板，{img} 占位）
+import os as _os
+
+_env = _os.environ.get('EVALHARNESS_DOCKER_MIRRORS', '').strip()
+if _env:
+    _CN_MIRROR_FALLBACKS = [t.strip() for t in _env.split(',') if t.strip()]
+
 
 def images_for_dataset(ds: Dataset, limit: int = 0) -> List[str]:
     """Collect distinct sandbox images declared by a dataset's samples."""
@@ -93,16 +103,139 @@ def prefetch_images(images: Iterable[str], workers: int = 8) -> List[str]:
     return pulled
 
 
-def _pull_one(image: str) -> None:
+def _pull_one(image: str, verbose: bool = True) -> None:
-    """Pull via CN-mirror fallback chain; retag to the canonical name on hit."""
+    """Pull via CN-mirror fallback chain; retag to the canonical name on hit.
+
+    A multi-GB pull runs silently for many minutes with capture_output --
+    which reads as a hang -- so say WHICH source is being tried and how
+    long it took."""
+    import sys as _sys
+    import time as _time
     last_err = None
     for template in _CN_MIRROR_FALLBACKS:
         ref = template.format(img=image)
-        r = subprocess.run(['docker', 'pull', ref], capture_output=True, text=True,
+        if verbose:
-                           timeout=3600)
+            # the daemon-default entry IS exactly '{img}'; every real
+            # mirror template also CONTAINS '{img}' -- the old containment
+            # check matched both and labeled everything docker.io
+            src = ('docker.io (daemon mirrors)' if template.strip() == '{img}'
+                   else template.split('/{img}')[0])
+            print(f'· pulling sandbox image {image} via {src} ...',
+                  file=_sys.stderr, flush=True)
+        t0 = _time.monotonic()
+        # stream progress: a multi-GB pull with captured output is minutes
+        # of silence that reads as a hang. Non-TTY docker pull emits clean
+        # per-layer lines; forward the informative ones (throttled).
+        # docker 25+ hides per-layer progress when stdout is not a TTY;
+        # --progress=plain forces the id: Downloading x/y lines our ⏳
+        # summary parses (without it even a real 9GB/74s pull is silent)
+        proc = subprocess.Popen(['docker', 'pull', '--progress=plain', ref],
+                                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                bufsize=0)
+        # idle watchdog: a dead mirror emits NOTHING (0 bytes/s for hours);
+        # the old fixed 3600s timeout let one dead source stall the whole
+        # chain. 5 minutes of silence = kill and try the next mirror.
+        import os as _os
+        import selectors as _sel
+
+        _selr = _sel.DefaultSelector()
+        _selr.register(proc.stdout, _sel.EVENT_READ)
+        _fd = proc.stdout.fileno()
+        last_line = ''
+        last_show = 0.0
+        layer_prog = {}   # layer id -> (done_bytes, total_bytes): the raw
+        # per-layer events give no sense of OVERALL progress -- sum them
+        import os as _os
+        import re as _re
+
+        def _bytes(s):
+            s = s.strip()
+            mult = {'kB': 1e3, 'KB': 1e3, 'MB': 1e6, 'GB': 1e9, 'B': 1}
+            for suf, m in mult.items():
+                if s.endswith(suf):
+                    return int(float(s[:-len(suf)]) * m)
+            return int(float(s or 0))
+
+        # docker progress lines end with \r (carriage-return refresh), not
+        # \n -- readline() buffered them until a rare \n arrived, so the
+        # Downloading snapshots (and the ⏳ summary fed by them) never
+        # surfaced. Split on BOTH terminators, chunked reads.
+        buf = ''
+        _lines_iter = iter(lambda: None, 1)  # placeholder, replaced below
+
+        # no output at all -> dead mirror, switch source. ANY output resets
+        # the timer (manifest/layer announcements/DOWNLOADING refreshes), so
+        # this only fires on sources that print nothing. 10s default;
+        # override with EVALHARNESS_PULL_IDLE_S for slow-negotiating mirrors.
+        import os as _os2
+
+        _IDLE_S = float(_os2.environ.get('EVALHARNESS_PULL_IDLE_S', '300'))
+
+        def _iter_lines():
+            nonlocal buf
+            while True:
+                if not _selr.select(timeout=_IDLE_S):
+                    proc.kill()
+                    if verbose:
+                        print(f'· no data for {_IDLE_S:.0f}s -- dead mirror, '
+                              'trying next source ...',
+                              file=_sys.stderr, flush=True)
+                    return
+                chunk = _os.read(_fd, 4096)
+                if not chunk:
+                    if buf:
+                        yield buf
+                    return
+                buf += chunk.decode('utf-8', 'replace')
+                parts = buf.split('\r') if '\r' in buf else buf.split('\n')
+                if len(parts) > 1:
+                    for p in parts[:-1]:
+                        yield p
+                    buf = parts[-1]
+
+        for line in _iter_lines():
+            line = line.strip('\n ')
+            last_line = line
+            m = _re.match(r'^([0-9a-f]{12}): Downloading.*?([\d.]+[kKMG]?B)/([\d.]+[kKMG]?B)', line)
+            if m:
+                layer_prog[m.group(1)] = (_bytes(m.group(2)), _bytes(m.group(3)))
+            if 'Pull complete' in line:
+                lid = line.split(':')[0]
+                if lid in layer_prog:
+                    layer_prog[lid] = (layer_prog[lid][1], layer_prog[lid][1])
+            now = _time.monotonic()
+            key_evt = any(k in line for k in
+                          ('Pulling from', 'Status', 'error', 'denied'))
+            if verbose and key_evt:
+                print(f'  {last_line[:100]}', file=_sys.stderr, flush=True)
+                last_show = now
+            elif verbose and now - last_show > 3 and layer_prog:
+                done = sum(v[0] for v in layer_prog.values())
+                tot = sum(v[1] for v in layer_prog.values())
+                if tot:
+                    pct = done / tot * 100
+                    big = max(layer_prog.items(), key=lambda kv: kv[1][1] - kv[1][0])
+                    print(f'  ⏳ {done / 1e9:.2f}/{tot / 1e9:.2f} GB ({pct:.0f}%)'
+                          f' · 最大层 {big[0]}: {big[1][0] / 1e9:.2f}/{big[1][1] / 1e9:.2f} GB',
+                          file=_sys.stderr, flush=True)
+                last_show = now
+        proc.wait(timeout=3600)
+        r = subprocess.CompletedProcess(proc.args, proc.returncode,
+                                        stdout='', stderr=last_line)
+        if r.returncode == 0 and verbose:
+            print(f'✓ image ready in {_time.monotonic() - t0:.0f}s',
+                  file=_sys.stderr, flush=True)
         if r.returncode == 0:
-            if ref != image:  # retag the mirrored pull to the canonical name
+            if ref != image:
-                subprocess.run(['docker', 'tag', ref, image], check=False)
+                # retag BEFORE rmi and VERIFY: a silent tag failure followed
+                # by rmi deleted the only tag and lost a 342s multi-GB pull
+                tag = subprocess.run(['docker', 'tag', ref, image],
+                                     capture_output=True, text=True)
+                if tag.returncode != 0:
+                    print(f'!! docker tag {ref} -> {image} failed: '
+                          f'{(tag.stderr or "")[:120]} (keeping the mirror tag)',
+                          file=_sys.stderr, flush=True)
+                    return   # do NOT rmi: the mirror tag is the only handle
                 subprocess.run(['docker', 'rmi', ref], check=False)
             return
         last_err = (ref, (r.stderr or '').strip()[:150])