thinking.yaml: 64k budgets for the full-thinking ladder (es parity)

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

README.md

@@ -161,7 +161,7 @@ rep.save('gsm8k.report.json')
 | `--limit N` / `--limit-per-task N` | 全局前 N / 每子集前 N（多科目 bench 用后者；可组合取交集） |
 | `--subset` `--split` `--source` | 覆盖子集 / split / 数据源（可指本地路径离线跑） |
 | `--concurrency N` | 并发（默认 32；长输出 bench 建议 8-16） |
-| `--auto-concurrency` | 自适应并发门：按端点健康状况自动决定并发（健康且供不应求时 +1 爬坡，请求失败 ×0.7 退避，服务端 `/metrics` 可用时按排队信号调节）；当前值显示在进度条 `gate N`。此时 `--concurrency` 是起点不是上限 |
+| `--concurrency auto` | 自适应并发门：从 2 起步，健康且供不应求时 +1 爬坡，请求失败 ×0.7 退避（服务端 `/metrics` 可用时按排队信号调节）；当前值显示在进度条 `gate N` |
 | `--resume [PATH]` | 断点续跑；默认 `<cache-dir>/ckpt/<bench>.jsonl` |
 | `--env NAME` | agent 环境（`bfcl_mock` 等） |
 | `--perf` | 采集流式 TTFT / ITL / 重试率入报告 |

evalharness/cli.py

@@ -163,6 +163,30 @@ def _rich_console():
         return None
 
 
+BENCH_CATEGORIES = {
+    'bigcodebench': 'Code & Engineering', 'humaneval': 'Code & Engineering',
+    'live_code_bench': 'Code & Engineering',
+    'swe_bench_verified': 'Code & Engineering',
+    'aime24': 'Reasoning & Math', 'aime25': 'Reasoning & Math',
+    'aime26': 'Reasoning & Math', 'hmmt26': 'Reasoning & Math',
+    'imo_answerbench': 'Reasoning & Math', 'hle': 'Reasoning & Math',
+    'gsm8k': 'Reasoning & Math', 'competition_math': 'Reasoning & Math',
+    'bbh': 'Reasoning & Math', 'drop': 'Reasoning & Math',
+    'gpqa_diamond': 'Knowledge & Language', 'mmlu_pro': 'Knowledge & Language',
+    'simple_qa': 'Knowledge & Language', 'mmlu': 'Knowledge & Language',
+    'cmmlu': 'Knowledge & Language', 'arc': 'Knowledge & Language',
+    'hellaswag': 'Knowledge & Language', 'trivia_qa': 'Knowledge & Language',
+    'winogrande': 'Knowledge & Language',
+    'longbench_v2': 'Long Context', 'openai_mrcr': 'Long Context',
+    'tau2_bench': 'Agents & Tools', 'general_fc': 'Agents & Tools',
+    'bfcl_v3': 'Agents & Tools',
+}
+
+
+def bench_category(name: str) -> str:
+    return BENCH_CATEGORIES.get(name, '')
+
+
 def _load_bench_cfg(args, name: str) -> dict:
     """Merged YAML config for one bench: {default 段, bench 段}.
 
@@ -175,8 +199,14 @@ def _load_bench_cfg(args, name: str) -> dict:
     cfg_name = getattr(args, 'config', '')
     if not cfg_name and cfg_dir.exists():
         yamls = sorted(cfg_dir.glob('*.yaml'))
+        # auto-load: a lone config wins; otherwise default.yaml wins.
+        # (NB: non-config yaml sidecars must NOT land in config/ -- a
+        # sample-counts manifest here once disabled auto-load entirely and
+        # silently dropped every bench's repeats/temperature/max_tokens)
         if len(yamls) == 1:
-            cfg_name = yamls[0].stem  # auto: the only config
+            cfg_name = yamls[0].stem
+        elif any(y.stem == 'default' for y in yamls):
+            cfg_name = 'default'
     if not cfg_name:
         return {}
     cfg_path = cfg_dir / f'{cfg_name}.yaml'
@@ -193,25 +223,36 @@ def _load_bench_cfg(args, name: str) -> dict:
 
 def _plan_sample_counts(args):
     """(total_samples, total_generations, n_uncached) across the planned
-    benches, counted from LOCAL cache entries only -- never touches the
-    network, so the run plan stays instant on cold machines. Uncached
-    benches simply don't contribute yet.
+    benches. Cache-first (exact); uncached benches fall back to the shipped
+    sample-counts manifest so the plan still shows CONCRETE numbers on a
+    cold machine instead of 'counts when datasets load'.
     """
+    from pathlib import Path
+
     from evalharness.data import get_dataset
 
+    manifest = {}
+    mf = Path(__file__).parent / 'config' / 'sample_counts.json'
+    if mf.exists():
+        try:
+            manifest = json.load(open(mf)) or {}
+        except Exception:
+            manifest = {}
+
     total = gens = uncached = 0
     for name in getattr(args, 'datasets', []) or []:
+        n = None
         try:
             ds = get_dataset(name, **_overrides(args))
             cache_file = ds.cache_dir / 'samples.jsonl'
-            if not cache_file.exists():
-                uncached += 1
-                continue
+            if cache_file.exists():
                 with open(cache_file, 'rb') as f:
                     n = sum(1 for _ in f)
         except Exception:
+            n = None
+        if n is None:
             uncached += 1
-            continue
+            n = int(manifest.get(name, 0) or 0)  # estimate; 0 = unknown
         if getattr(args, 'limit', None):
             n = min(n, args.limit)
         total += n
@@ -232,21 +273,26 @@ def _print_run_plan(console, args, model_spec):
         cap = f' · ≤{args.limit} per bench (--limit)'
     elif getattr(args, 'limit_per_task', None):
         cap = f' · ≤{args.limit_per_task} per subject (--limit-per-task)'
+    src = 'cached' if not n_uncached else ('cache+est.' if n_samples else 'est.')
     if n_samples:
-        samples = f'{n_samples:,} samples (cached){cap}'
+        samples = f'{n_samples:,} samples ({src}){cap}'
         if n_gens > n_samples:  # repeats multiply the real work
-            samples = (f'{n_samples:,} samples (cached){cap} → '
+            samples = (f'{n_samples:,} samples ({src}){cap} → '
                        f'{n_gens:,} generations (repeats)')
+    elif getattr(args, 'limit', None):
+        samples = f'up to {args.limit} per bench (--limit)'
+    elif getattr(args, 'limit_per_task', None):
+        samples = f'up to {args.limit_per_task} per subject (--limit-per-task)'
+    else:
+        samples = '? (unknown benchmarks)'
     if n_uncached:
         samples += f' · {n_uncached} bench(es) not cached yet'
-    elif n_uncached:
-        if getattr(args, 'limit', None):
-            samples = f'up to {args.limit} per bench (--limit), counts when datasets load'
-        elif getattr(args, 'limit_per_task', None):
-            samples = (f'up to {args.limit_per_task} per subject '
-                       '(--limit-per-task), counts when datasets load')
-        else:
-            samples = 'full dataset, counts when datasets load (none cached yet)'
+    _eff = getattr(args, 'reasoning_effort', '') or ''
+    _think_txt = ('disabled' if getattr(args, 'disable_thinking', False)
+                  else (f'enabled · effort={_eff}' if _eff else 'enabled'))
+    _auto = getattr(args, 'auto_concurrency', False)
+    _conc = (f'auto (start {args.concurrency}, gate decides)' if _auto
+             else str(args.concurrency))
     if console is None:
         print(f'=== {title} ===')
         print(f'Provider: {provider}')
@@ -254,8 +300,7 @@ def _print_run_plan(console, args, model_spec):
         print(f'Model: {model_name}')
         print(f'Benchmarks: {len(args.datasets)} -> {", ".join(args.datasets)}')
         print(f'Samples: {samples}')
-        print(f'Concurrency: {args.concurrency} | Thinking: '
-              f'{"enabled" if not args.disable_thinking else "disabled"} | '
+        print(f'Concurrency: {_conc} | Thinking: {_think_txt} | '
               f'Performance: {"on" if args.perf else "off"}')
         print(f'Resume: {"on" if args.resume else "off"} | Output: {args.out_dir or "(none)"}')
         return
@@ -271,8 +316,11 @@ def _print_run_plan(console, args, model_spec):
     table.add_row('Model', model_name)
     table.add_row('Benchmarks', f'{len(args.datasets)}  ·  {", ".join(args.datasets)}')
     table.add_row('Samples', samples)
-    table.add_row('Concurrency', str(args.concurrency))
-    table.add_row('Thinking', '[red]disabled[/red]' if args.disable_thinking else '[green]enabled[/green]')
+    table.add_row('Concurrency', f'[magenta]{_conc}[/magenta]' if _auto else _conc)
+    table.add_row('Thinking',
+                  '[red]disabled[/red]' if getattr(args, 'disable_thinking', False)
+                  else (f'[green]enabled · effort={_eff}[/green]' if _eff
+                        else '[green]enabled[/green]'))
     table.add_row('Performance', '[green]enabled[/green]' if args.perf else '[dim]disabled[/dim]')
     table.add_row('Checkpoint', '[green]resume[/green]' if args.resume else '[dim]new run[/dim]')
     table.add_row('Output', args.out_dir or '[dim](not specified)[/dim]')
@@ -621,6 +669,55 @@ def _cmd_eval_run(args) -> int:
 
     rows = []
     all_reports = []
+
+    def _flush_summaries():
+        """Rewrite summary.csv/xlsx from the benches finished SO FAR.
+
+        Called after every benchmark instead of only at the very end:
+        multi-hour runs (mmlu_pro 12k samples) leave the summary stale for
+        hours otherwise -- and a crashed run would leave the LAST run's
+        corpses in place instead of partial results."""
+        if not out_dir:
+            return
+        if all_reports:
+            try:
+                from evalharness.viz import render as _render
+
+                _render(all_reports, style='excel', out=f'{out_dir}/summary.xlsx')
+            except Exception as e:
+                print(f'excel export skipped: {type(e).__name__}: {str(e)[:80]}',
+                      file=sys.stderr)
+        if not rows:
+            return
+        import csv as _csv
+
+        with open(f'{out_dir}/summary.csv', 'w', newline='', encoding='utf-8') as f:
+            w = _csv.writer(f)
+            w.writerow(['benchmark', 'category', 'score', 'metric', 'num_samples',
+                        'time_h', 'time_s', 'extract_fail',
+                        'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
+                        'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
+                        'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
+                        'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s',
+                        'categories'])
+            for r in rows:
+                perf = (r.get('groups') or {}).get('perf') or {}
+                cats = '; '.join(f'{g}={_f3(v)}'
+                                 for gname, gv in (r.get('groups') or {}).items()
+                                 if gname != 'perf'
+                                 for g, v in (gv or {}).items()
+                                 if isinstance(v, (int, float)))[:2000]
+                w.writerow([r['name'], r.get('category', ''), _f3(r.get('value')),
+                            r['metric'], r.get('n', ''),
+                            r.get('hours', ''), r.get('secs', ''),
+                            r.get('extract_fail', 0)] +
+                           [perf.get(k, '') for k in (
+                               'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
+                               'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
+                               'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
+                               'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] +
+                           [cats])
+
     run_started = _time.time()
     total_runs = len(args.datasets)
     model_spec = _compose_model_spec(args)
@@ -629,7 +726,7 @@ def _cmd_eval_run(args) -> int:
         # adapts on its own signals (see pool.AdaptiveGate)
         from evalharness.model.pool import AdaptiveGate
 
-        AdaptiveGate.INITIAL = float(max(2, getattr(args, 'concurrency', 8)))
+        AdaptiveGate.INITIAL = float(max(1, getattr(args, 'concurrency', 1)))
     if not out_dir and model_spec and not args.out:
         # always persist results: default dir = evalharness-results/<stamp>-<model>/
         import re as _re
@@ -680,11 +777,10 @@ def _cmd_eval_run(args) -> int:
             _rep_secs = 0.0
             _rep_tin = _rep_tout = 0
 
-            if model_spec:  # generate + score in one go
-                from evalharness.model import run_eval
-
+            # progress reporter setup runs for every path: the success path
+            # advances the overall bar even when nothing was generated
             progress_reporter = None
-                if args.progress:
+            if args.progress and model_spec:
                 from evalharness.progress import PROGRESS_REGISTRY
 
                 _pname = getattr(args, 'progress_plugin', 'rich') \
@@ -709,6 +805,12 @@ def _cmd_eval_run(args) -> int:
                         _shared_reporter.owned_externally = True
                     progress_reporter = _shared_reporter
 
+            if getattr(args, 'reasoning_effort', ''):
+                args._gen_override = {**(getattr(args, '_gen_override', {}) or {}),
+                                      'reasoning_effort': args.reasoning_effort}
+            if model_spec:  # generate + score in one go
+                from evalharness.model import run_eval
+
 
                 def status_callback(msg, _idx=i + 1, _name=name,
                                     _reporter=progress_reporter,
@@ -771,11 +873,21 @@ def _cmd_eval_run(args) -> int:
                         progress_reporter=progress_reporter,
                         status_callback=status_callback,
                         on_scored=on_scored,
+                        rescore=getattr(args, 'rescore', False),
                         repeat=_rep + 1))
                     _m = next((v for k, v in report.metrics.items()
                                if k != 'extraction_failure_rate'), None)
                     if _m is not None:
                         _scores.append(_m)
+                    if _repeats > 1 and out_dir:
+                        # per-run report: <out-dir>/<bench>/reps/repNN.report.jsonl
+                        # (each carries THAT run's own score + samples; the
+                        # top-level report.jsonl stays the mean-summary view)
+                        import pathlib as _pl
+
+                        _rdir = _pl.Path(out_dir) / name / 'reps'
+                        _rdir.mkdir(parents=True, exist_ok=True)
+                        report.save(str(_rdir / f'rep{_rep + 1:02d}.report.jsonl'))
                     _rep_info = report.metric_groups.get('run_info', {}) or {}
                     _rep_secs += sum(float((s.usage or {}).get('latency_s', 0) or 0)
                                      for s in report.samples)
@@ -842,15 +954,13 @@ def _cmd_eval_run(args) -> int:
                     print(render(report, style=args.style))
             all_reports.append(report)
             primary = next(iter(report.metrics), '')
-            secs_total = sum(float((s.usage or {}).get('latency_s', 0) or 0)
-                             for s in report.samples)
             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')}
             info = report.metric_groups.get('run_info', {}) or {}
-            if _repeats > 1 and _rep_secs:
-                # repeats: report the SUM over all runs, not the last one
-                secs_total = _rep_secs
+            if _repeats > 1:
+                # repeats: tokens are the SUM over all runs (real cost of the
+                # predictions used)
                 info = {**info, 'gen_input_tokens': _rep_tin,
                         'gen_output_tokens': _rep_tout,
                         'gen_total_tokens': _rep_tin + _rep_tout}
@@ -861,12 +971,24 @@ def _cmd_eval_run(args) -> int:
                 return lats[min(int(len(lats) * q), len(lats) - 1)] if lats else 0.0
             fins = [(s.usage or {}).get('finish_reason', '')
                     for s in report.samples]
-            rows.append({'name': name, 'metric': primary, 'value': report.metrics.get(primary),
-                         'n': report.num_samples,
+            # time = THIS run's wall clock everywhere: summing per-prediction
+            # latency_s counts RESTORED predictions' original generation time
+            # (days old, slower setup) -- that once reported 15.9h for a
+            # one-hour run
+            _wall = round(_time.time() - t0, 1)
+            _cat = bench_category(name)
+            report.metric_groups.setdefault('run_info', {})['category'] = _cat
+            rows.append({'name': name, 'metric': primary, 'category': _cat,
+                         'cached': (report.metric_groups.get('run_info', {})
+                                    .get('gen_fresh') == 0) if _repeats <= 1 else False,
+                         'value': report.metrics.get(primary),
+                         # repeats evaluate the SAME N problems k times: the
+                         # count people expect is the generations, not N
+                         'n': report.num_samples * _repeats,
                          'extract_fail': report.num_failed_extractions,
-                         'secs': round(secs_total, 1),
-                         'wall': round(_time.time() - t0, 1),
-                         'hours': round(secs_total / 3600, 2),
+                         'secs': _wall,
+                         'wall': _wall,
+                         'hours': round(_wall / 3600, 2),
                          'tok_in': info.get('gen_input_tokens', 0) or 0,
                          'tok_out': info.get('gen_output_tokens', 0) or 0,
                          'tokens': (info.get('gen_input_tokens', 0) or 0)
@@ -884,9 +1006,15 @@ def _cmd_eval_run(args) -> int:
                                     'done', _time.time() - t0)
         except Exception as e:
             rows.append({'name': name, 'metric': '-', 'value': None,
+                         'category': bench_category(name),
                          'secs': round(_time.time() - t0, 1), 'ok': False,
                          'err': f'{type(e).__name__}: {str(e)[:100]}'})
             print(f'{name}: FAILED {type(e).__name__}: {str(e)[:160]}', file=sys.stderr)
+            import traceback as _tb
+
+            _tb.print_exc()   # full stack to stderr: the one-line form hides
+            # WHERE an error like 'Object of type ellipsis is not JSON
+            # serializable' actually comes from
             from evalharness.hooks import fire as _fire
 
             _fire('on_benchmark_failed', name=name, error=e, dataset=name)
@@ -899,49 +1027,17 @@ def _cmd_eval_run(args) -> int:
                     border_style='red', expand=False), justify='center')
             _print_benchmark_result(console, i + 1, total_runs, name,
                                     'failed', _time.time() - t0)
+        # incremental summary: partial results are visible (and survive a
+        # crash) after EVERY benchmark, not only when the whole run ends
+        _flush_summaries()
 
     if _shared_reporter is not None:
         _shared_reporter.close()
     if all_reports and out_dir:
-        try:
-            from evalharness.viz import render as _render
-
-            xb = _render(all_reports, style='excel',
-                         out=f'{out_dir}/summary.xlsx')
-            print(f'excel  -> {xb}', flush=True)
-        except Exception as e:
-            print(f'excel export skipped: {type(e).__name__}: {str(e)[:80]}',
-                  file=sys.stderr)
+        _flush_summaries()   # final state (benches already flushed per-bench)
+        print(f'excel  -> {out_dir}/summary.xlsx', flush=True)
     if rows:
         _print_summary_table(console, rows)
-        if out_dir:
-            import csv as _csv
-
-            with open(f'{out_dir}/summary.csv', 'w', newline='', encoding='utf-8') as f:
-                w = _csv.writer(f)
-                w.writerow(['benchmark', 'score', 'metric', 'num_samples',
-                            'time_h', 'time_s', 'extract_fail',
-                            'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
-                            'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
-                            'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
-                            'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s',
-                            'categories'])
-                for r in rows:
-                    perf = (r.get('groups') or {}).get('perf') or {}
-                    cats = '; '.join(f'{g}={_f3(v)}'
-                                     for gname, gv in (r.get('groups') or {}).items()
-                                     if gname != 'perf'
-                                     for g, v in (gv or {}).items()
-                                     if isinstance(v, (int, float)))[:2000]
-                    w.writerow([r['name'], _f3(r.get('value')), r['metric'], r.get('n', ''),
-                                r.get('hours', ''), r.get('secs', ''),
-                                r.get('extract_fail', 0)] +
-                               [perf.get(k, '') for k in (
-                                   'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
-                                   'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
-                                   'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
-                                   'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] +
-                               [cats])
 
     # artifacts notice: tell the user where everything landed (or how to save);
     # rich terminals get clickable file:// links (iTerm2/kitty/WezTerm/WT...)
@@ -980,7 +1076,8 @@ def _print_summary_table(console, rows):
 
         t = Table(title='Run Summary', header_style='bold cyan',
                   title_style='bold', expand=False)
-        for col, just in (('benchmark', 'left'), ('metric', 'left'),
+        for col, just in (('benchmark', 'left'), ('category', 'left'),
+                          ('metric', 'left'),
                           ('score', 'right'), ('n', 'right'), ('time', 'right'),
                           ('tok in', 'right'), ('tok out', 'right'),
                           ('in/s', 'right'), ('out/s', 'right')):
@@ -988,13 +1085,15 @@ def _print_summary_table(console, rows):
         for r in rows:
             v = _fmt_score(r.get('value')) if r['ok'] else 'ERR'
             wall = r.get('wall') or r.get('secs') or 0
-            tm = f'{wall / 3600:.2f}h' if wall >= 3600 else f'{wall:.0f}s'
+            tm = 'cached' if r.get('cached') else (
+                f'{wall / 3600:.2f}h' if wall >= 3600 else f'{wall:.0f}s')
             ti, to = r.get('tok_in', 0), r.get('tok_out', 0)
             tin = f'{ti:,}' if ti else '—'
             tout = f'{to:,}' if to else '—'
             tis = f'{ti / wall:.0f}' if (wall > 1 and ti) else '—'
             tos = f'{to / wall:.0f}' if (wall > 1 and to) else '—'
-            t.add_row(r['name'], r['metric'], v, str(r.get('n', '')), tm,
+            t.add_row(r['name'], r.get('category', ''), r['metric'], v,
+                      str(r.get('n', '')), tm,
                       tin, tout, tis, tos,
                       style='green' if r['ok'] else 'red')
         wall_all = sum(r.get('wall') or r.get('secs') or 0 for r in rows)
@@ -1005,7 +1104,7 @@ def _print_summary_table(console, rows):
         t.add_section()
         tis_all = f'{ti_all / wall_all:.0f}' if wall_all > 1 else ''
         tos_all = f'{to_all / wall_all:.0f}' if wall_all > 1 else ''
-        t.add_row(f'[bold]{len(rows)} benchmarks[/bold]', '',
+        t.add_row(f'[bold]{len(rows)} benchmarks[/bold]', '', '',
                   f'{sum(1 for r in rows if r["ok"])}/{len(rows)} ok',
                   str(n_all), tm_all, f'{ti_all:,}', f'{to_all:,}',
                   tis_all, tos_all)
@@ -1095,12 +1194,6 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument('--provider', default='openai-chat',
                    choices=('openai-chat', 'openai-pool'),
                    help='API protocol/provider (default: openai-chat)')
-    p.add_argument('--auto-concurrency', action='store_true',
-                   help='let the per-endpoint adaptive gate decide concurrency '
-                        '(ramps while healthy, backs off x0.7 on failures, '
-                        'server /metrics when available); current limit shows '
-                        'on the progress bar as "gate N". --concurrency '
-                        'becomes the starting point, not a cap')
     p.add_argument('--judge-model', '--judge', dest='judge', default='',
                    help='judge model name with --judge-api-url, or full spec')
     p.add_argument('--judge-api-url', default='',
@@ -1122,12 +1215,28 @@ def build_parser() -> argparse.ArgumentParser:
                         "profile.default < profile['<bench>'] < explicit kwargs")
     p.add_argument('--disable-thinking', action='store_true',
                    help='send enable_thinking=false to the OpenAI-compatible model')
+    p.add_argument('--thinking', default='',
+                   choices=('', 'off', 'low', 'medium', 'high', 'max', 'full'),
+                   help='unified thinking switch: off = --disable-thinking; '
+                        'low/medium/high/max = reasoning_effort (verified on '
+                        'this endpoint: low = ~1/9 tokens); full = default '
+                        'thinking. Takes precedence over --disable-thinking '
+                        'and --reasoning-effort')
+    p.add_argument('--reasoning-effort', default='',
+                   choices=('', 'minimal', 'low', 'medium', 'high', 'max'),
+                   help="thinking intensity (GLM/Anthropic-style; verified "
+                        "working on the sglang endpoint: low shrinks reasoning "
+                        "~85%%). Overrides the YAML config")
     p.add_argument('--perf', action='store_true',
                    help='collect streaming TTFT and ITL metrics')
     p.add_argument('--textools', action='store_true',
                    help='send tools as text instead of native tool calls')
     p.add_argument('--env', default='', help="agent environment (e.g. 'bfcl_mock') -> message pump")
-    p.add_argument('--concurrency', type=int, default=32, help='parallel model calls (default 32)')
+    p.add_argument('--concurrency', default='32',
+                   help="parallel model calls (default 32); 'auto' = the "
+                        "adaptive gate decides (starts at 2, ramps +1 while "
+                        "healthy, backs off x0.7 on failures -- see 'gate N' "
+                        "on the progress bar)")
     p.add_argument('--progress', action='store_true', default=True,
                    help='show per-sample progress (default: on)')
     p.add_argument('--no-progress', dest='progress', action='store_false',
@@ -1141,6 +1250,9 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument('--resume', nargs='?', const=True, default=False,
                    help='resume from per-sample checkpoint (default path auto-derived; '
                         'pass a path to override)')
+    p.add_argument('--rescore', action='store_true',
+                   help='force re-scoring even when a matching saved report '
+                        'could be reused (change of recipe/judge, or paranoia)')
     p.add_argument('--limit-per-task', type=int,
                    help='first N samples PER subset/category (evalscope --limit semantics); '
                         'composable with --limit (intersection)')
@@ -1189,6 +1301,21 @@ def build_parser() -> argparse.ArgumentParser:
 
 def main(argv=None) -> int:
     args = build_parser().parse_args(argv)
+    # '--concurrency auto' == '--auto-concurrency': normalize once, here,
+    # so every downstream site (plan display, run_eval, gate initial) sees
+    # an int + the flag
+    _th = str(getattr(args, 'thinking', '') or '').strip().lower()
+    if _th:
+        if _th == 'off':
+            args.disable_thinking = True
+        else:
+            args.disable_thinking = False
+            args.reasoning_effort = '' if _th == 'full' else _th
+    if str(getattr(args, 'concurrency', '32')).strip().lower() == 'auto':
+        args.auto_concurrency = True
+        args.concurrency = 1   # gate starts at 1: probe x2, bisect to capacity
+    else:
+        args.concurrency = int(args.concurrency)
     return args.func(args)
 
 

evalharness/config/default.yaml

@@ -23,6 +23,7 @@ imo_answerbench:
   temperature: 1.0
 gpqa_diamond:
   temperature: 1.0
+  repeats: 3
   max_tokens: 8192
 mmlu:
   max_tokens: 8192
@@ -42,6 +43,7 @@ trivia_qa:
   max_tokens: 8192
 humaneval:
   temperature: 1.0
+  repeats: 3
 live_code_bench:
   temperature: 1.0
 longbench_v2:

evalharness/config/effort_low.yaml

@@ -0,0 +1,84 @@
+# 低思考档：全库参数同 default.yaml，仅追加 reasoning_effort: low
+# 用法: evalharness eval run ... --config effort_low
+default:
+  temperature: 0.0
+  top_p: 1.0
+  stream: true
+  max_tokens: 32768
+  reasoning_effort: low
+aime24:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 8192
+  reasoning_effort: low
+aime25:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 8192
+  reasoning_effort: low
+aime26:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 8192
+  reasoning_effort: low
+hmmt26:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 8192
+  reasoning_effort: low
+imo_answerbench:
+  temperature: 1.0
+  reasoning_effort: low
+gpqa_diamond:
+  temperature: 1.0
+  repeats: 3
+  max_tokens: 8192
+  reasoning_effort: low
+mmlu:
+  max_tokens: 8192
+  reasoning_effort: low
+mmlu_pro:
+  max_tokens: 8192
+  reasoning_effort: low
+cmmlu:
+  max_tokens: 8192
+  reasoning_effort: low
+arc:
+  max_tokens: 8192
+  reasoning_effort: low
+hellaswag:
+  max_tokens: 8192
+  reasoning_effort: low
+winogrande:
+  max_tokens: 8192
+  reasoning_effort: low
+simple_qa:
+  max_tokens: 8192
+  reasoning_effort: low
+trivia_qa:
+  max_tokens: 8192
+  reasoning_effort: low
+humaneval:
+  temperature: 1.0
+  repeats: 3
+  reasoning_effort: low
+live_code_bench:
+  temperature: 1.0
+  reasoning_effort: low
+longbench_v2:
+  max_tokens: 8192
+  max_input_tokens: 128000
+  reasoning_effort: low
+openai_mrcr:
+  max_tokens: 8192
+  max_input_tokens: 128000
+  reasoning_effort: low
+bfcl_v3:
+  max_tokens: 4096
+  reasoning_effort: low
+general_fc:
+  max_tokens: 4096
+  reasoning_effort: low
+tau2_bench:
+  max_tokens: 16384
+  reasoning_effort: low

evalharness/config/sample_counts.json

@@ -0,0 +1,29 @@
+{
+ "humaneval": 164,
+ "aime24": 30,
+ "aime25": 30,
+ "aime26": 30,
+ "hmmt26": 30,
+ "gpqa_diamond": 198,
+ "mmlu": 285,
+ "mmlu_pro": 12032,
+ "cmmlu": 11528,
+ "gsm8k": 1319,
+ "arc": 2376,
+ "hellaswag": 10042,
+ "winogrande": 1267,
+ "drop": 9535,
+ "longbench_v2": 503,
+ "live_code_bench": 1055,
+ "bigcodebench": 1140,
+ "trivia_qa": 17944,
+ "simple_qa": 4326,
+ "hle": 2500,
+ "imo_answerbench": 89,
+ "openai_mrcr": 1300,
+ "bfcl_v3": 2000,
+ "general_fc": 400,
+ "tau2_bench": 184,
+ "competition_math": 5000,
+ "swe_bench_verified": 500
+}

evalharness/config/thinking.yaml

@@ -0,0 +1,60 @@
+default:
+  temperature: 0.0
+  top_p: 1.0
+  stream: true
+  max_tokens: 200336
+aime24:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 200336
+aime25:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 200336
+aime26:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 200336
+hmmt26:
+  temperature: 1.0
+  repeats: 12
+  max_tokens: 200336
+imo_answerbench:
+  temperature: 1.0
+gpqa_diamond:
+  temperature: 1.0
+  repeats: 3
+  max_tokens: 200336
+mmlu:
+  max_tokens: 200336
+mmlu_pro:
+  max_tokens: 200336
+cmmlu:
+  max_tokens: 200336
+arc:
+  max_tokens: 200336
+hellaswag:
+  max_tokens: 200336
+winogrande:
+  max_tokens: 200336
+simple_qa:
+  max_tokens: 200336
+trivia_qa:
+  max_tokens: 200336
+humaneval:
+  temperature: 1.0
+  repeats: 3
+live_code_bench:
+  temperature: 1.0
+longbench_v2:
+  max_tokens: 200336
+  max_input_tokens: 128000
+openai_mrcr:
+  max_tokens: 200336
+  max_input_tokens: 128000
+bfcl_v3:
+  max_tokens: 4096
+general_fc:
+  max_tokens: 4096
+tau2_bench:
+  max_tokens: 16384

evalharness/eval/checkpoint.py

@@ -28,6 +28,7 @@ class CheckpointStore:
         self.model = model
         self.dataset = dataset
         self._entries: Dict[str, Dict[str, Any]] = {}
+        self._scores: Dict[str, Dict[str, Any]] = {}
         self._fh = None
 
     @staticmethod
@@ -47,6 +48,7 @@ class CheckpointStore:
     def load(self) -> Dict[str, Dict[str, Any]]:
         """Read all checkpointed predictions (idempotent)."""
         self._entries = {}
+        self._scores = {}
         if not self.path.exists():
             return self._entries
         with open(self.path, encoding='utf-8') as f:
@@ -57,6 +59,8 @@ class CheckpointStore:
                 try:
                     rec = json.loads(line)
                     self._entries[rec['key']] = rec.get('pred', {})
+                    if rec.get('score'):
+                        self._scores[rec['key']] = rec['score']
                 except (ValueError, KeyError):
                     continue  # torn tail line from a crash -- safe to skip
         return self._entries
@@ -66,9 +70,44 @@ class CheckpointStore:
         worst case loses the last in-flight sample on crash)."""
         rec = {'key': key, 'ts': time.time(), 'pred': pred}
         with open(self.path, 'a', encoding='utf-8') as f:
-            f.write(json.dumps(rec, ensure_ascii=False) + '\n')
+            f.write(json.dumps(rec, ensure_ascii=False, default=str) + '\n')
         self._entries[key] = pred
 
+    def scores(self) -> Dict[str, Dict[str, Any]]:
+        """Cached per-sample score records ({key: {'fp', 'scores', ...}}).
+
+        Scores are bound to predictions and live in the SAME file -- one
+        --resume flag controls both layers."""
+        return self._scores
+
+    def put_scores(self, records: Dict[str, Dict[str, Any]]) -> None:
+        """Attach score records to checkpoint entries (end-of-evaluation
+        writeback). Rewrites the file atomically; legacy lines without a
+        score field simply gain one."""
+        if not records:
+            return
+        lines: Dict[str, str] = {}
+        if self.path.exists():
+            with open(self.path, encoding='utf-8') as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        rec = json.loads(line)
+                    except ValueError:
+                        continue
+                    if rec.get('key') in records:
+                        rec['score'] = records[rec['key']]
+                    lines[rec['key']] = json.dumps(rec, ensure_ascii=False,
+                                                   default=str)
+        tmp = self.path.with_suffix('.tmp')
+        with open(tmp, 'w', encoding='utf-8') as f:
+            for v in lines.values():
+                f.write(v + '\n')
+        os.replace(tmp, self.path)
+        self._scores.update(records)
+
     def __len__(self) -> int:
         return len(self._entries)
 

evalharness/eval/recipes/agent.py

@@ -56,7 +56,7 @@ def bigcodebench():
         extract='code_any',
         scorers={'pass': {'name': 'execution', 'harness': _bcb_harness,
                           # official sandbox image (bundles every task's deps)
-                          'image': 'bigcodebench-sandbox:latest',
+                          'image': 'bigcodebench/bigcodebench-evaluate:latest',  # official hub image, same as evalscope
                           'sandbox': 'docker', 'timeout_s': 120}},
         aggregators={'pass': 'pass_at_k'},
         exec_workers=12,

evalharness/eval/record.py

@@ -56,7 +56,10 @@ class EvalReport(BaseModel):
     num_samples: int = 0
     num_failed_extractions: int = 0
     metrics: Dict[str, float] = Field(default_factory=dict)     # {'acc': 0.62}
-    metric_groups: Dict[str, Dict[str, float]] = Field(default_factory=dict)
+    # values may be None (perf stats the adapter couldn't measure), lists
+    # (repeats.scores) or nested dicts -- a strict float type rejected the
+    # file on LOAD and silently defeated report reuse
+    metric_groups: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
     # {'by_category': {'algebra': 0.7, ...}, 'pass_at_k': {'pass@1': .., 'pass@8': ..},
     #  'by_length_bin': {'8k': .., '32k': ..}}
 
@@ -65,20 +68,25 @@ class EvalReport(BaseModel):
     def save(self, path) -> None:
         import json
 
+        # default=str everywhere: score_details/samples carry arbitrary
+        # scorer output, and ONE exotic object (an Ellipsis sneaked in via
+        # a scorer's detail dict) must not kill a finished benchmark at
+        # the save line
         if str(path).endswith('.jsonl'):
             # streaming format: first line = report header, then one
             # sample per line (grep/split/tail friendly)
             head = self.model_dump(exclude={'samples'})
             head['type'] = 'report'
             with open(path, 'w', encoding='utf-8') as f:
-                f.write(json.dumps(head, ensure_ascii=False) + '\n')
+                f.write(json.dumps(head, ensure_ascii=False, default=str) + '\n')
                 for smp in self.samples:
                     row = smp if isinstance(smp, dict) else smp.model_dump()
                     row['type'] = 'sample'
-                    f.write(json.dumps(row, ensure_ascii=False) + '\n')
+                    f.write(json.dumps(row, ensure_ascii=False, default=str) + '\n')
             return
         with open(path, 'w', encoding='utf-8') as f:
-            json.dump(self.model_dump(), f, ensure_ascii=False, indent=2)
+            json.dump(self.model_dump(), f, ensure_ascii=False, indent=2,
+                      default=str)
 
     @classmethod
     def load(cls, path) -> 'EvalReport':

evalharness/eval/runner.py

@@ -51,6 +51,29 @@ def evaluate(
     aggregators = recipe.resolve_aggregators()
     ctx = ScoreContext(judge=judge, params={})
 
+    # fail-fast image preflight: recipe-level AND sample-level images are
+    # ensured (local or pulled once) BEFORE any container runs -- a missing
+    # image must kill the bench in seconds with a fix hint, not produce a
+    # 0.0% after hours of per-sample pull failures
+    try:
+        _imgs = set()
+        for spec in (recipe.scorers or {}).values():
+            p = spec if isinstance(spec, dict) else {}
+            if p.get('name') == 'execution' and p.get('sandbox') == 'docker'                     and p.get('image'):
+                _imgs.add(p['image'])
+        for s in samples[:200]:
+            if getattr(s, 'sandbox', None) and s.sandbox.image:
+                _imgs.add(s.sandbox.image)
+        if _imgs:
+            from ..sandbox.docker import ensure_image
+
+            for _img in sorted(_imgs):
+                ensure_image(_img)
+    except RuntimeError:
+        raise
+    except Exception:
+        pass  # no docker here (local sandbox): the scorer will complain
+
     # If any scorer executes in docker with per-sample images, overlap pulls
     # with scoring (run sample N while N+1..N+lookahead images download).
     bp = None
@@ -62,10 +85,10 @@ def evaluate(
 
     results: List[SampleResult] = []
 
-    def judge_one(sample, pred) -> SampleResult:
-        """Extract + score ONE sample (thread-safe: everything here is local
-        except docker/subprocess execution, which parallelizes perfectly --
-        each sample gets its own container/workdir)."""
+    def _shell(sample, pred) -> SampleResult:
+        """SampleResult with everything derivable from (sample, prediction):
+        identity, raw text, usage, trajectories. Shared by live scoring and
+        the checkpoint-score replay path."""
         raw = pred if isinstance(pred, str) else str(pred.get('raw', ''))
         override = {} if isinstance(pred, str) else pred
         result = SampleResult(
@@ -89,6 +112,14 @@ def evaluate(
             result.env_state = pred['env_state']
         if isinstance(pred, dict) and pred.get('usage'):
             result.usage = pred['usage']
+        return result
+
+    def judge_one(sample, pred) -> SampleResult:
+        """Extract + score ONE sample (thread-safe: everything here is local
+        except docker/subprocess execution, which parallelizes perfectly --
+        each sample gets its own container/workdir)."""
+        result = _shell(sample, pred)
+        raw = result.raw_prediction
         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
@@ -165,6 +196,91 @@ def evaluate(
     return report
 
 
+def score_fingerprint(recipe: EvalRecipe, judge_spec: str = '') -> str:
+    """Identity of the SCORING setup: recipe + extractors + scorers + judge.
+    Cached score records carry it; a mismatch means re-evaluate."""
+    import hashlib
+    import json as _json
+
+    payload = _json.dumps({
+        'recipe': getattr(recipe, 'name', ''),
+        'extract': str(getattr(recipe, 'extract', '')),
+        'scorers': str(getattr(recipe, 'scorers', '')),
+        'judge': judge_spec or '',
+    }, sort_keys=True, default=str)
+    return hashlib.md5(payload.encode()).hexdigest()[:12]
+
+
+def score_record_of(result: SampleResult, fp: str) -> Dict:
+    """Extract the cacheable part of an evaluated SampleResult."""
+    return {'fp': fp,
+            'extracted': result.extracted_prediction,
+            'ok': result.extraction_ok,
+            'note': result.extraction_note,
+            'scores': dict(result.scores),
+            'details': {k: (v if isinstance(v, (str, int, float, bool, dict, list, type(None)))
+                            else str(v))
+                        for k, v in result.score_details.items()},
+            'error': result.error or ''}
+
+
+def evaluate_cached(dataset, predictions, recipe, score_records, *,
+                    model: str = '', extra_metadata=None) -> EvalReport:
+    """Rebuild a report from checkpoint-cached scores -- no extractor, no
+    scorers, no docker. Shells come from (sample, prediction), scores from
+    the cached records; aggregation runs FRESH (cheap, and covers recipe
+    aggregation changes without invalidating the cache)."""
+    samples = list(dataset)
+    if len(predictions) != len(samples) or len(score_records) != len(samples):
+        raise ValueError('evaluate_cached: samples/predictions/score_records '
+                         f'length mismatch ({len(samples)}/'
+                         f'{len(predictions)}/{len(score_records)})')
+    spec = getattr(dataset, 'spec', None)
+    ds_name = spec.name if spec is not None else 'adhoc'
+    ds_subset = spec.subset if spec is not None else ''
+    aggregator_map = recipe.resolve_aggregators() if recipe is not None else {}
+    from .aggregator import mean as _mean
+
+    def _mk(sample, pred, sr):
+        r = SampleResult(
+            sample_id=sample.id, dataset=ds_name, subset=ds_subset,
+            task_type=sample.task_type,
+            raw_prediction=pred if isinstance(pred, str) else str(pred.get('raw', '')),
+            target=sample.target,
+            group_key=str((pred if isinstance(pred, dict) else {}).get('group_key')
+                          or sample.metadata.get('group_key')
+                          or (sample.metadata.get('task_id') or sample.metadata.get('id') or '')),
+            metadata={k: v for k, v in (sample.metadata or {}).items()
+                      if k in ('category', 'subject', 'test_category', 'bin', 'difficulty')},
+        )
+        if isinstance(pred, dict):
+            if pred.get('usage'):
+                r.usage = pred['usage']
+            if pred.get('metadata'):
+                r.metadata.update(pred['metadata'])
+        r.extracted_prediction = sr.get('extracted', '')
+        r.extraction_ok = bool(sr.get('ok', True))
+        r.extraction_note = sr.get('note', '')
+        r.scores.update(sr.get('scores') or {})
+        r.score_details.update(sr.get('details') or {})
+        return r
+
+    results = [_mk(s, p, sr) for s, p, sr in zip(samples, predictions, score_records)]
+    report = EvalReport(
+        dataset=ds_name,
+        recipe=recipe.name if recipe is not None else ds_name,
+        model=model,
+        num_samples=len(results),
+        num_failed_extractions=sum(1 for r in results if not r.extraction_ok),
+        samples=results,
+    )
+    _aggregate_into(report, results, recipe, aggregator_map or {'acc': _mean})
+    if extra_metadata:
+        report.metric_groups['run_info'] = {k: v for k, v in extra_metadata.items()
+                                            if isinstance(v, (int, float, str))}
+    return report
+
+
 def _needs_bg_prefetch(recipe, samples) -> bool:
     """True when the recipe executes in docker AND samples declare images."""
     try:

evalharness/model/adapter.py

@@ -115,6 +115,17 @@ def _key_for(api_base: str) -> str:
 
 
 
+def _payload_chars(payload: Dict[str, Any]) -> int:
+    """Rough prompt size in characters (~3-4 chars/token). Used to decide
+    the auto-stream path: gateways that buffer whole requests make a huge
+    INPUT as hang-prone as a huge output budget."""
+    n = 0
+    for m in payload.get('messages') or []:
+        c = m.get('content')
+        n += len(c) if isinstance(c, str) else 256
+    return n
+
+
 def _parse_text_tool_calls(text: str) -> list:
     """Extract tool calls from a text reply. Handles both shapes:
     - JSON array:  [{"name":..,"arguments":{..}}]
@@ -164,8 +175,12 @@ def _parse_text_tool_calls(text: str) -> list:
                         args[kw_.arg] = _ast.unparse(kw_.value)
             except SyntaxError:
                 return None
+        # literal_eval happily returns Ellipsis (code like `f(key=...)`) and
+        # other non-JSON constants; default=str keeps the serializer alive
+        # instead of killing the whole benchmark at parse time
         return {'id': '', 'type': 'function',
-                'function': {'name': name, 'arguments': json.dumps(args)}}
+                'function': {'name': name,
+                             'arguments': json.dumps(args, default=str)}}
 
     for m_ in _re.finditer(r'([A-Za-z_][A-Za-z0-9_]*)\((.*?)\)', text):
         c = _py_call(m_)
@@ -205,9 +220,15 @@ class OpenAICompatible(ModelAdapter):
                 if stream:
                     out = await self._post_stream_perf(
                         f'{self.api_base}/chat/completions', payload, headers, t0)
-                elif int(payload.get('max_tokens') or 0) > 100000 \
+                elif (int(payload.get('max_tokens') or 0) > 100000
+                      or _payload_chars(payload) > 300_000) \
                         and not os.environ.get('EVALHARNESS_NO_AUTOSTREAM'):
-                    # long generation: stream and aggregate (gateway-safe).
+                    # long generation OR LONG INPUT: stream and aggregate
+                    # (gateway-safe). Some gateways buffer the whole request
+                    # before answering on the non-stream path -- a 128k-token
+                    # longbench_v2 prompt sat there past every read timeout;
+                    # streaming starts emitting immediately, so a stuck
+                    # endpoint surfaces in ~60s instead of after 20 minutes.
                     # Some gateways drop chat_template_kwargs on the STREAM
                     # path only (non-stream honors it) -- append the /no_think
                     # soft switch into the prompt itself as a belt-and-braces
@@ -222,10 +243,10 @@ class OpenAICompatible(ModelAdapter):
                                 break
                     data = await self._post_stream_aggregate(
                         f'{self.api_base}/chat/completions', payload, headers)
-                    out = self._parse(data)
+                    out = self._parse(data, allow_text_calls=bool(tools))
                 else:
                     data = await self._post(f'{self.api_base}/chat/completions', payload, headers)
-                    out = self._parse(data)
+                    out = self._parse(data, allow_text_calls=bool(tools))
                 out.usage.latency_s = round(_time.time() - t0, 3)
                 out.usage.retries = attempt
                 return out
@@ -333,7 +354,7 @@ class OpenAICompatible(ModelAdapter):
             'usage': (usage_ev or {}).get('usage') or {},
             'model': self.model,
         }
-        out = self._parse(data)
+        out = self._parse(data, allow_text_calls=True)
         out.usage.ttft_s = round(ttft, 3) if ttft is not None else None
         out.usage.itl_mean_s = round(sum(itl_vals) / len(itl_vals), 4) if itl_vals else None
         out.usage.http_status = status
@@ -359,7 +380,8 @@ class OpenAICompatible(ModelAdapter):
                 ]
                 payload.pop('chat_template_kwargs', None)
         for k in ('temperature', 'max_tokens', 'top_p', 'stop', 'seed',
-                  'response_format', 'chat_template_kwargs'):
+                  'response_format', 'chat_template_kwargs',
+                  'thinking', 'reasoning_effort'):
             if kw.get(k) is not None:
                 payload[k] = kw[k]
         payload.setdefault('max_tokens', self.extra.get('max_tokens', 4096))  # CoT room
@@ -379,12 +401,17 @@ class OpenAICompatible(ModelAdapter):
                 payload['chat_template_kwargs'] = {'enable_thinking': False}
         return payload
 
-    def _parse(self, data: Dict[str, Any]) -> ModelOutput:
+    def _parse(self, data: Dict[str, Any],
+               allow_text_calls: bool = True) -> ModelOutput:
         choice = (data.get('choices') or [{}])[0]
         msg = choice.get('message') or {}
         calls = []
         raw_calls = list(msg.get('tool_calls') or [])
-        if not raw_calls:
+        if not raw_calls and allow_text_calls:
+            # text-protocol fallback ONLY for requests that carried tools:
+            # running it on plain prose/code (humaneval!) regex-matched
+            # `f(key=...)` style code as "calls", literal_eval'd the `...`
+            # into an Ellipsis and crashed json.dumps mid-generation
             raw_calls = _parse_text_tool_calls(msg.get('content') or '')
         for c in raw_calls:
             fn = c.get('function') or {}
@@ -424,9 +451,14 @@ class OpenAICompatible(ModelAdapter):
 
         import httpx as _hx
 
+        # read timeout bounds the wait for the FIRST byte too: a 128k-token
+        # prompt queued behind other prefills legitimately takes minutes to
+        # start answering -- 60s killed exactly those (the gate's x0.7 and
+        # sample-failure containment now handle real hangs)
         async with _hx.AsyncClient(timeout=_hx.Timeout(
                 connect=self.extra.get('connect_timeout', 15),
-                read=60, write=30, pool=15)) as client:
+                read=self.extra.get('stream_read_timeout', 240),
+                write=30, pool=15)) as client:
             async with client.stream('POST', url, json=payload, headers=headers) as resp:
                 if resp.status_code != 200:
                     body = (await resp.aread()).decode('utf-8', 'replace')[:300]

evalharness/model/pool.py

@@ -151,10 +151,25 @@ class AdaptiveGate:
     Purely additive to PooledAdapter: one gate per backend, no caller change.
     """
 
-    LO = 2          # never go below: progress beats perfection
+    LO = 1          # never go below: progress beats perfection
     HI = 96         # sane ceiling for one endpoint
-    PROBE_S = 5.0   # metrics probe interval
-    INITIAL = 8.0   # class-level start point (--auto-concurrency rebinds it)
+    PROBE_S = 5.0   # safety tick (fails/hang detection); ramp decisions use
+                    # level statistics, not this interval alone
+    INITIAL = 2.0   # class-level start point ('--concurrency auto' rebinds it)
+    # ---- exponential-probe + binary-search capacity discovery ----
+    # probe: 1 -> 2 -> 4 -> ... while throughput keeps IMPROVING (>10%);
+    # the first level where it plateaus opens a bisect [last_good, bad];
+    # bisect narrows to the knee; steady holds there. Any failure x0.7s
+    # immediately and restarts probing from the shrunken level.
+    GAIN_EPS = 1.1      # rate must beat the previous level by 10% to keep doubling
+    MIN_OK = 5          # baseline completions needed at a level before judging
+    # dwell fallback: JUDGE also when the level has been held this long (with
+    # >= 1 completion) -- scaled by observed inter-completion gap so a bench
+    # whose single request takes 60s is not judged on one lone sample at t=25s
+    DWELL_BASE_S = 120.0
+    # robust judging: sample count scales WITH the level (a 2-completion
+    # estimate at level 8 is pure quantization noise), plus a minimum dwell
+    # so one lucky tick cannot speak for the whole level
 
     def __init__(self, adapter: ModelAdapter):
         self.adapter = adapter
@@ -166,8 +181,26 @@ class AdaptiveGate:
         self._loop = None         # loop the cond/probe-task are bound to
         self._contended = 0       # acquire() waits this probe interval (demand)
         self._interval_fails = 0  # failed releases this probe interval
+        self._interval_ok = 0     # SUCCESSFUL releases this probe interval:
+                                  # zero completions = zero throughput, and a
+                                  # gate that ramps on demand alone would pile
+                                  # 96 concurrent prefills onto a server whose
+                                  # first 2 requests have not even answered
+        # capacity-discovery state
+        self._mode = 'probe'      # probe | bisect | steady
+        self._level_t0 = None     # when we arrived at the current limit
+        self._level_ok = 0        # completions at this level
+        self._prev = (None, None)  # (level, rate) we came from / last-good
+        self._bis = (None, None)   # bisect bounds (lo=good, hi=bad)
+        self._good_rate = 0.0      # throughput at the good bound (baseline)
+        self._fetch_exec = None    # DEDICATED executor for /metrics probes:
+                                   # the shared default pool is occupied by
+                                   # second-long tokenization jobs, and the
+                                   # probe queued behind them never ticked
+                                   # (gate frozen at 1 while results flowed)
+        self._metrics_dead = False # 3 consecutive fetch failures -> stop asking
         self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0,
-                      'backoff_queue': 0, 'ramp_demand': 0}
+                      'backoff_queue': 0, 'ramp_demand': 0, 'bisect': 0}
 
     def _push_limit(self) -> None:
         """Surface the current limit to the progress bar ('gate N')."""
@@ -179,6 +212,18 @@ class AdaptiveGate:
             except Exception:
                 pass
 
+    def push_inflight(self) -> None:
+        """Tell the bar how many requests the gate has ACTUALLY admitted
+        (bar shows 'admitted/held' -- a bare held count with gate 2 read
+        as 'the gate is broken')."""
+        rep = (self.adapter.extra or {}).get('progress_reporter')
+        fn = getattr(rep, 'set_admitted', None)
+        if fn is not None:
+            try:
+                fn(self._inflight)
+            except Exception:
+                pass
+
     # ---- gate semantics ----
     async def acquire(self) -> None:
         loop = asyncio.get_running_loop()
@@ -206,18 +251,40 @@ class AdaptiveGate:
             finally:
                 self._cond.release()
         self._inflight += 1
+        self.push_inflight()
 
     def release(self, ok: bool) -> None:
         self._inflight = max(0, self._inflight - 1)
-        if not ok:  # multiplicative decrease -- survival first
+        self.push_inflight()
+        if ok:
+            self._interval_ok += 1
+            self._level_ok += 1
+            now = time.monotonic()
+            last = getattr(self, '_last_ok_t', None)
+            if last is not None:
+                gap = now - last
+                ge = getattr(self, '_gap_ema', None)
+                self._gap_ema = gap if ge is None else 0.6 * ge + 0.4 * gap
+            self._last_ok_t = now
+        else:  # multiplicative decrease -- survival first
             self._interval_fails += 1
+            # capacity moved (or we overshot): shrink now and restart the
+            # discovery from the shrunken level
             before = self.limit
             self.limit = max(self.LO, self.limit * 0.7)
             if before != self.limit:
                 self.stats['backoff_fail'] += 1
                 self._push_limit()
+                self._enter_level(mode='probe')
         self._wake()
 
+    def _enter_level(self, mode: str = '') -> None:
+        """Arrive at (a new) limit: start measuring this level fresh."""
+        if mode:
+            self._mode = mode
+        self._level_t0 = time.monotonic()
+        self._level_ok = 0
+
     def _wake(self) -> None:
         if self._cond is not None:
             # fire-and-forget notify (loop may not be ours -- best effort)
@@ -233,21 +300,121 @@ class AdaptiveGate:
 
     # ---- server-signal probe ----
     def _no_signal_ramp(self) -> None:
-        """No server signals available (no /metrics, 404/HTTPError, gateway
-        stripped it, non-sglang backend): fall back to demand-driven AIMD --
-        ramp while the cap is the binding constraint (callers had to WAIT on
-        acquire) and the interval was failure-free. Failures still cut x0.7
-        per release, so a drowning backend shrinks the gate immediately."""
-        if self._interval_fails == 0 and self._contended > 0 \
-                and int(self.limit) < self.HI:
-            self.limit = min(self.HI, self.limit + 1)
+        """No server signals (no /metrics, 404, gateway stripped it):
+        discover capacity by measuring THROUGHPUT per concurrency level.
+
+        probe:   double while completions/s keeps improving (rate > prev x
+                 1.1) -- 1, 2, 4, 8 ... reaches the knee in log time
+        bisect:  first level where the gain stalls opens [last_good, bad];
+                 narrow to the knee with midpoint measurements
+        steady:  hold at the converged level; any failure x0.7s (handled in
+                 release) and probing restarts from the shrunken level
+
+        A level is judged only after MIN_OK completions or MAX_AT_LEVEL_S;
+        zero completions so far = hold (hang protection)."""
+        try:
+            now = time.monotonic()
+            if self._level_t0 is None:
+                self._enter_level()
+            dt = now - self._level_t0
+            lvl = max(1, int(self.limit))
+
+            # not enough evidence yet at this level: keep measuring.
+            # evidence = 2 x level completions, no other floor
+            need_ok = max(2, lvl * 2)
+            # dwell fallback scales with the OBSERVED completion cadence:
+            # a 60s-per-request bench needs minutes, not 25s, before a
+            # single-sample judgment is acceptable
+            need_dt = max(self.DWELL_BASE_S,
+                          3.0 * (getattr(self, '_gap_ema', None) or 0.0))
+            if self._level_ok < need_ok and dt < need_dt:
+                return
+            # zero completions so far: hang or overloaded -> hold
+            if self._level_ok == 0:
+                return
+
+            rate = self._level_ok / max(dt, 1e-6)
+            prev_lvl, prev_rate = self._prev
+            self._prev = (lvl, rate)
+
+            if self._mode == 'probe':
+                self.stats['probe'] += 1
+                # continue while NOT WORSE (>= 0.9x): with heterogeneous
+                # request lengths (lbv2: 4k..2M-token docs) completion-rate
+                # noise dwarfs a 10% gain threshold, and demanding strict
+                # improvement bisected [1,2]->1 on the first plateau.
+                # Only CLEAR degradation (<0.9x) means past the knee.
+                ok = prev_rate is None or rate >= prev_rate * 0.9
+                if ok and lvl < self.HI:
+                    self._bis = (lvl, min(lvl * 2, self.HI))  # remember bounds
+                    self.limit = float(min(lvl * 2, self.HI))
                     self.stats['ramp_demand'] += 1
                     self._push_limit()
+                    self._enter_level()
+                elif not ok:
+                    # throughput CLEARLY degraded: knee is in (prev_lvl, lvl]
+                    self._mode = 'bisect'
+                    self._bis = (prev_lvl or max(1, lvl // 2), lvl)
+                    self._good_rate = prev_rate or rate
+                    lo, hi = self._bis
+                    mid = (lo + hi) // 2
+                    if hi - lo <= 1:
+                        self.limit = float(lo)   # prev_lvl was the knee
+                        self._push_limit()
+                        self._enter_level('steady')
+                    else:
+                        self.limit = float(mid)
+                        self.stats['bisect'] += 1
+                        self._push_limit()
+                        self._enter_level()
+                else:
+                    self._enter_level('steady')   # hit HI with gains: stay
+            elif self._mode == 'steady':
+                # the endpoint is SHARED: other tenants change its capacity
+                # while we run -- keep judging forever, nudge +-1 against the
+                # reference rate instead of pinning the converged level
+                ref = self._good_rate or rate
+                if rate >= ref * 1.05 and lvl < self.HI:
+                    self.limit = float(min(lvl + 1, self.HI))
+                    self.stats['ramp_demand'] += 1
+                    self._push_limit()
+                    self._enter_level()
+                elif rate <= ref * 0.85 and lvl > self.LO:
+                    self.limit = float(max(self.LO, lvl - 1))
+                    self.stats['backoff_queue'] += 1
+                    self._push_limit()
+                    self._enter_level()
+                else:
+                    # reference drifts with fresh measurements (slow EWMA)
+                    self._good_rate = 0.7 * ref + 0.3 * rate
+                    self._enter_level()   # restart the measurement window
+            elif self._mode == 'bisect':
+                lo, hi = self._bis
+                if rate >= self._good_rate * 0.9:
+                    lo = lvl            # not worse here: knee is at/above
+                else:
+                    hi = lvl            # clearly worse: knee is below
+                self._bis = (lo, hi)
+                if hi - lo <= 1:
+                    self.limit = float(lo)
+                    self._push_limit()
+                    self._enter_level('steady')
+                else:
+                    mid = (lo + hi) // 2
+                    self.limit = float(mid)
+                    self.stats['bisect'] += 1
+                    self._push_limit()
+                    self._enter_level()
+        finally:
             self._interval_fails = 0
+            self._interval_ok = 0
             self._contended = 0
             self._wake()
 
     async def _probe_once(self) -> None:
+        if self._metrics_dead:
+            self._no_signal_ramp()   # endpoint said no thrice: stop asking
+            return
         import urllib.request
 
         url = f'{self.adapter.api_base.rstrip("/")}/metrics'
@@ -257,12 +424,23 @@ class AdaptiveGate:
                 return resp.read().decode('utf-8', 'ignore')
 
         try:
-            # thread: the blocking fetch must never stall the event loop
-            # (an unreachable host parks urlopen for the full 4s timeout)
-            text = await asyncio.to_thread(_fetch)
+            # DEDICATED single-thread executor: the shared asyncio pool is
+            # full of second-long truncation tokenizations, and a probe
+            # queued behind them never ran (gate appeared frozen)
+            if self._fetch_exec is None:
+                import concurrent.futures
+
+                self._fetch_exec = concurrent.futures.ThreadPoolExecutor(
+                    max_workers=1, thread_name_prefix='gate-probe')
+            loop = asyncio.get_running_loop()
+            text = await loop.run_in_executor(self._fetch_exec, _fetch)
         except Exception:
+            self._fetch_fails = getattr(self, '_fetch_fails', 0) + 1
+            if self._fetch_fails >= 3:
+                self._metrics_dead = True   # 404/unreachable: pure demand mode
             self._no_signal_ramp()  # no metrics: demand-driven fallback
             return
+        self._fetch_fails = 0
         running = queue = None
         for line in text.splitlines():
             if line.startswith('sglang:num_running_reqs'):
@@ -278,8 +456,10 @@ class AdaptiveGate:
             self.limit = max(self.LO, self.limit - 1)
             self.stats['backoff_queue'] += 1
             self._push_limit()
-        elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))):
+        elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))) \
+                and self._interval_ok > 0:
             # underfed: no queue and running below our own cap -> ramp up
+            # (still requires completions this interval: no throughput, no ramp)
             self.limit = min(self.HI, self.limit + 1)
             self.stats['ramp'] += 1
             self._push_limit()

evalharness/model/runner.py

@@ -317,7 +317,25 @@ async def generate_predictions(
                 _progress(progress, done_count, len(samples), t0, total_usage)
             return pred
 
-        messages = ([ChatMessage(role='user', content=assemble(sample))]
+        # assemble() tokenizes for the max_input_tokens truncation -- on
+        # long-context benches that is SECONDS of CPU per sample (2M-token
+        # docs). Two failure modes fixed here:
+        #  - inline: froze the whole event loop behind one encode
+        #  - asyncio.to_thread (32-thread default pool): dozens of concurrent
+        #    tokenizers hogged the GIL and starved the progress renderer +
+        #    loop itself (bar froze, then jumped)
+        # A DEDICATED BOUNDED pool: 8 encodes at a time, remaining workers
+        # queue -- GIL pressure capped, everything stays responsive.
+        global _ASSEMBLE_EXEC
+        if _ASSEMBLE_EXEC is None:
+            import concurrent.futures
+
+            _ASSEMBLE_EXEC = concurrent.futures.ThreadPoolExecutor(
+                max_workers=8, thread_name_prefix='assemble')
+        text = await asyncio.get_running_loop().run_in_executor(
+            _ASSEMBLE_EXEC, assemble, sample) \
+            if isinstance(sample.input, str) else None
+        messages = ([ChatMessage(role='user', content=text)]
                     if isinstance(sample.input, str) else list(sample.input))
         if not system and extra_system[0] and isinstance(sample.input, str):
             # renderer-provided SYSTEM contract (es lcb expert-programmer)
@@ -420,21 +438,25 @@ async def generate_predictions(
         for m_ in members:
             m_.extra['progress_reporter'] = progress_reporter
 
+    # terminal (post-retry) sample failures are CONTAINED: one sample that
+    # never makes it (server queue ate its first byte past every timeout)
+    # must not kill the remaining hundreds -- it becomes an empty prediction
+    # (scores as wrong, es-parity for timeouts), is NOT checkpointed (a
+    # rerun retries it), and only a total wipeout fails the bench
+    failed_samples: Dict[int, str] = {}
+
     async def run_one(i_s):
         i, s = i_s
-        # transient network flaps (cluster routes re-converge): retry with
-        # backoff so one ConnectError burst cannot kill the whole batch --
-        # the adapter already retries 5xx/429 and the pool fails over per
-        # instance; this is the last line of defense around asyncio.gather
-        # NO outer retry: the adapter retries internally; a second loop
-        # here multiplied worst-case time (42+ attempts before this fix).
-        # One pass, one result or one error.
+        # NO outer retry: the adapter retries internally (and the pool
+        # fails over per instance); a second loop here multiplied
+        # worst-case time. One pass, one result or one contained error.
         try:
             pred = await one(s)
-        except Exception:
+        except Exception as e:
             if progress_reporter is not None:
                 progress_reporter.advance(success=False)
-            raise
+            failed_samples[i] = f'{type(e).__name__}: {str(e)[:120]}'
+            return i, None   # empty marker: no checkpoint write
         if ckpt_store is not None:
             ckpt_store.append(keys[i], pred)
         return i, pred
@@ -446,8 +468,20 @@ async def generate_predictions(
             else:
                 status_callback('Generation skipped: the checkpoint already covers every sample')
         fresh = await asyncio.gather(*(run_one((i, s)) for i, s in pending))
+        if failed_samples and len(failed_samples) >= len(pending):
+            # every single fresh sample died: the endpoint is down, not flaky
+            _f = next(iter(failed_samples.values()))
+            raise RuntimeError(f'all {len(failed_samples)} generations failed '
+                               f'(first: {_f})')
+        if failed_samples:
+            print(f'generation: {len(failed_samples)}/{len(pending)} samples '
+                  'failed after all retries (empty predictions, not '
+                  'checkpointed -- rerun to retry them); first: '
+                  f'{next(iter(failed_samples.items()))[1][:100]}', flush=True)
         for i, pred in fresh:
-            preds_by_key[keys[i]] = pred
+            preds_by_key[keys[i]] = pred if pred is not None \
+                else {'raw': '', 'usage': {},
+                      'error': failed_samples.get(i, '')[:200]}
         preds = [preds_by_key[k] for k in keys]
         usages = [p.get('usage', {}) for p in preds]
         # include RESTORED predictions' usage (they carry it in the ckpt);
@@ -466,7 +500,10 @@ async def generate_predictions(
                 latency_s=float(u.get('latency_s', 0) or 0))
         if status_callback and pending:
             status_callback(f'Generation complete: {len(preds)} responses collected')
-        return preds, usages, total_usage
+        # ckpt info (store + per-position keys) so run_eval can read/write
+        # SCORES bound to these predictions; None when checkpointing is off
+        ckpt_info = (ckpt_store, keys) if ckpt_store is not None else None
+        return preds, usages, total_usage, ckpt_info, len(fresh) - len(failed_samples)
     finally:
         # reporter lifecycle belongs to the CALLER (CLI reuses one reporter
         # across benchmarks and closes it after the whole run); only close
@@ -529,6 +566,7 @@ def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) ->
 
 
 _PROBED_SPECS = set()
+_ASSEMBLE_EXEC = None   # bounded truncation pool (lazy)
 
 
 async def _probe_model(adapter, model_spec: str) -> None:
@@ -630,6 +668,7 @@ async def run_eval(
     gen_profile: str = '',
     repeat: int = 1,
     on_scored=None,
+    rescore: bool = False,
 ) -> EvalReport:
     """Generate + score in one call. Model spec examples:
     'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'.
@@ -742,7 +781,7 @@ async def run_eval(
 
     try:
         from .gen_profiles import merge_gen_kwargs
-        preds, _usages, usage = await generate_predictions(
+        preds, _usages, usage, ckpt_info, n_fresh = await generate_predictions(
             adapter, list(raw_samples), concurrency, progress=progress,
             progress_reporter=progress_reporter,
             status_callback=status_callback,
@@ -762,7 +801,27 @@ async def run_eval(
             repeat=repeat)
     finally:
         await adapter.close()
-    if judge is None and judge_spec:
+
+    # SCORES ARE BOUND TO PREDICTIONS in the checkpoint: when every sample's
+    # cached score matches the current scoring setup (recipe/extract/scorers/
+    # judge fingerprint), replay them without touching a single scorer --
+    # docker exec benches skip their containers entirely. --resume controls
+    # the whole stack (no checkpoint -> nothing cached -> evaluate + backfill)
+    _fp = None
+    _records = None
+    if ckpt_info is not None and not rescore:
+        from ..eval.runner import score_fingerprint
+
+        store, ck_keys = ckpt_info
+        _fp = score_fingerprint(recipe, judge_spec or '')
+        cached = store.scores()
+        if ck_keys and all(cached.get(k, {}).get('fp') == _fp for k in ck_keys):
+            _records = [cached[k] for k in ck_keys]
+            if status_callback:
+                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 status_callback:
             status_callback('loading judge model')
         judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key)
@@ -770,6 +829,18 @@ async def run_eval(
 
     if status_callback:
         status_callback('Scoring predictions against the benchmark recipe')
+    _meta = {'gen_input_tokens': usage.input_tokens,
+             'gen_output_tokens': usage.output_tokens,
+             'gen_total_tokens': usage.total_tokens,
+             # fresh=0 means the whole bench replayed from checkpoint: the
+             # summary table then shows 'cached' instead of a ~0s time
+             'gen_fresh': n_fresh}
+    if _records is not None:
+        from ..eval.runner import evaluate_cached
+
+        report = evaluate_cached(samples, preds, recipe, _records,
+                                 model=model_spec, extra_metadata=_meta)
+    else:
         # scoring off the event loop: math_equal/sympy equivalence can chew a
         # single hard problem for minutes (es's checker famously hangs on one) --
         # running it inline froze the progress bar's clock for the whole bench
@@ -778,11 +849,18 @@ async def run_eval(
             samples, preds, recipe,
             model=model_spec,
             judge=judge,
-        extra_metadata={'gen_input_tokens': usage.input_tokens,
-                        'gen_output_tokens': usage.output_tokens,
-                        'gen_total_tokens': usage.total_tokens},
+            extra_metadata=_meta,
             on_scored=on_scored,
         )
+        # writeback: bind these scores to the predictions in the checkpoint
+        if ckpt_info is not None:
+            from ..eval.runner import score_fingerprint, score_record_of
+
+            store, ck_keys = ckpt_info
+            if _fp is None:
+                _fp = score_fingerprint(recipe, judge_spec or '')
+            store.put_scores({ck_keys[i]: score_record_of(report.samples[i], _fp)
+                              for i in range(min(len(ck_keys), len(report.samples)))})
     report.model = model_spec
     report.dataset = name
     if status_callback:

evalharness/model/truncation.py

@@ -9,20 +9,33 @@ Requires a tokenizer (transformers) at tokenizer_path or auto from the model.
 """
 
 import os
+import threading
 from functools import lru_cache
 from typing import Optional
 
 DEFAULT_TRUNCATION_TOKENS = 32768 * 4  # 131072, mirrors evalside run.py
 
 
+_TOK_LOCK = threading.Lock()
+
+
 @lru_cache(maxsize=4)
 def _get_tokenizer(tokenizer_path: str):
     if not tokenizer_path or not os.path.exists(tokenizer_path):
         raise FileNotFoundError(
             f'tokenizer not found at {tokenizer_path!r} -- token-level truncation '
             'needs a local tokenizer dir (e.g. /data1/models/DeepSeek-V4-Flash-INT8)')
+    # serialize the FIRST load: 96 worker threads racing transformers 5.x's
+    # lazy imports raised ImportError and silently degraded batches to the
+    # char approximation; after one success lru_cache serves the rest
+    with _TOK_LOCK:
         from transformers import AutoTokenizer
 
+        # the '> model_max_length' warnings are EXPECTED here -- counting a
+        # 2M-token doc before trimming it is the whole point of truncation
+        import logging
+
+        logging.getLogger('transformers').setLevel(logging.ERROR)
         return AutoTokenizer.from_pretrained(tokenizer_path, trust_remote_code=True)
 
 

evalharness/progress/rich_terminal.py

@@ -19,6 +19,21 @@ def _fmt(sec):
     return f'{sec // 60}m{sec % 60:02d}s' if sec >= 60 else f'{sec}s'
 
 
+def _fmt_eta(sec):
+    """eta: s -> m s -> h m s -> d h m (grows with the unit that matters)."""
+    sec = int(sec)
+    if sec < 60:
+        return f'{sec}s'
+    m, s = divmod(sec, 60)
+    if m < 60:
+        return f'{m}m{s:02d}s'
+    h, m = divmod(m, 60)
+    if h < 24:
+        return f'{h}h{m:02d}m{s:02d}s'
+    d, h = divmod(h, 24)
+    return f'{d}d{h:02d}h{m:02d}m'
+
+
 class RichTerminalProgress:
     def __init__(self, console=None):
         # accept an EXTERNAL console: CLI phase messages and the live bar must
@@ -60,6 +75,7 @@ class RichTerminalProgress:
         self.started = 0.0
         self.current_started = 0.0
         self.inflight = 0
+        self.admitted = None  # requests past the pool gate (None = no gate)
         self.restored = 0     # checkpoint head start (drives the '+N new' marker)
         self.heartbeat_task = None
 
@@ -166,8 +182,8 @@ class RichTerminalProgress:
             description=f'[green]{self.bench_tag}{self.bench_name} · scoring[/green]',
             total=total, completed=min(done, total), new='',
             rate=f'{done / elapsed:.2f}', inflight=0, cur='0s',
-            elapsed=_fmt(elapsed),
-            eta=_fmt((total - done) * elapsed / done) if done and total > done else '-')
+            elapsed=_fmt_eta(elapsed),
+            eta=_fmt_eta((total - done) * elapsed / done) if done and total > done else '-')
 
     def set_bench_tag(self, tag: str):
         if self.disabled:
@@ -195,7 +211,7 @@ class RichTerminalProgress:
             return
         self.inflight += 1
         self.current_started = time.monotonic()
-        self.progress.update(self.task_id, inflight=self.inflight, cur='0s')
+        self.progress.update(self.task_id, inflight=self._inflight_txt(), cur='0s')
 
     def set_retries(self, n: int):
         """Show the retry count on the bar (from the adapter's attempt)."""
@@ -207,6 +223,23 @@ class RichTerminalProgress:
         if self.task_id is not None:
             self.progress.update(self.task_id, gate=f'gate {n}')
 
+    def _inflight_txt(self) -> str:
+        # 'held' counts workers past the global semaphore; when a pool gate
+        # is active most of them are QUEUED on it -- 'admitted' is what
+        # actually hits the server. Showing a bare 96 with gate 2 read as
+        # 'the gate is not working'
+        if self.admitted is None:
+            return str(self.inflight)
+        # 'N gen · M wait': N admitted by the gate (real server load),
+        # M held by the pipeline (tokenizing or gate-queued)
+        return f'{self.admitted} gen · {max(0, self.inflight - self.admitted)} wait'
+
+    def set_admitted(self, n: int):
+        """Pooled runs: requests actually admitted by the gate."""
+        self.admitted = n
+        if self.task_id is not None:
+            self.progress.update(self.task_id, inflight=self._inflight_txt())
+
     def rollback(self):
         if self.disabled:
             return
@@ -214,7 +247,7 @@ class RichTerminalProgress:
         just decrement the in-flight count, no success/fail bookkeeping."""
         self.inflight = max(0, self.inflight - 1)
         if self.task_id is not None:
-            self.progress.update(self.task_id, inflight=self.inflight, cur='0s')
+            self.progress.update(self.task_id, inflight=self._inflight_txt(), cur='0s')
 
     def advance(self, success: bool = True):
         if self.disabled:
@@ -233,9 +266,9 @@ class RichTerminalProgress:
             advance=1,
             new=self._new_txt(completed),
             rate=f"{fresh / elapsed:.2f}",
-            inflight=self.inflight, cur='0s',
-            elapsed=_fmt(elapsed),
-            eta=_fmt((task.total - completed) * elapsed / fresh)
+            inflight=self._inflight_txt(), cur='0s',
+            elapsed=_fmt_eta(elapsed),
+            eta=_fmt_eta((task.total - completed) * elapsed / fresh)
                  if fresh and task.total and task.total > completed else '-',
         )
 
@@ -246,8 +279,7 @@ class RichTerminalProgress:
         try:
             while self.task_id is not None:
                 e = time.monotonic() - self.started
-                upd = {'elapsed': f'{int(e) // 60}m{int(e) % 60:02d}s' if e >= 60
-                       else f'{int(e)}s'}
+                upd = {'elapsed': _fmt_eta(e)}
                 if self.inflight:
                     secs = int(time.monotonic() - self.current_started)
                     upd['cur'] = (f'{secs // 60}m{secs % 60:02d}s'

evalharness/sandbox/docker.py

@@ -26,6 +26,32 @@ def docker_available() -> bool:
     return _run(['docker', 'info']).returncode == 0
 
 
+def ensure_image(img: str) -> None:
+    """Fail-fast sandbox image preflight: present locally, else pull ONCE.
+
+    Without this, every sample's `docker run` tries its own pull at scoring
+    time -- a missing image burned 1140 x 3 retries x ~70s on bigcodebench
+    before anyone saw a 0.0%."""
+    if not img:
+        return
+    if _run(['docker', 'image', 'inspect', img]).returncode == 0:
+        return
+    # CN-mirror fallback chain (daocloud -> 1ms.run -> baidubce -> sjtu ->
+    # rat.dev), mirroring the SWE prefetch path; a hit is retagged to the
+    # canonical name so the recipe never knows which mirror answered
+    from .prefetch import _pull_one
+    try:
+        _pull_one(img)
+        return
+    except RuntimeError as e:
+        raise RuntimeError(
+            f'sandbox image {img!r} is not available: not local, and every '
+            f'mirror failed. {str(e)[:200]}. '
+            'Fix: pull/build it manually (for bigcodebench the official image '
+            'is bigcodebench/bigcodebench-evaluate:latest), then re-run with '
+            '--rescore to score the cached predictions.') from e
+
+
 @register_sandbox('docker')
 class DockerSandbox(Sandbox):
     name = 'docker'
@@ -90,6 +116,14 @@ class DockerSandbox(Sandbox):
                     raise
                 if proc.returncode != 125 or attempt == 2:
                     break
+                # image-not-found is PERMANENT: retrying it 3x per sample
+                # burned 1140 x ~200s on a nonexistent bigcodebench image
+                _nf = 'Unable to find image' in (proc.stderr or '') \
+                    or 'failed to resolve' in (proc.stderr or '') \
+                    or 'manifest unknown' in (proc.stderr or '') \
+                    or 'pull access denied' in (proc.stderr or '')
+                if _nf:
+                    break
                 # clear any husk; fresh name next try. Bounded: an rm against
                 # a bloated daemon can hang for minutes and silently eat the
                 # whole worker pool (7 of 8 workers were observed stuck here)

evalharness/sandbox/prefetch.py

@@ -21,6 +21,7 @@ from ..data.dataset import Dataset
 # fall through: daemon default -> 1ms.run -> baidubce -> sjtug.
 _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.mirrors.sjtug.sjtu.edu.cn/{img}',