Progress coverage for minute-scale work: overall bench bar (which benchmark of N, covers loading/scoring), byte-level download bars for dataset fetches (ModelScope/HF raw, tty-only), one shared reporter across benchmarks (caller-owned lifecycle)

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

evalharness/cli.py

@@ -471,6 +471,7 @@ def _cmd_eval_run(args) -> int:
         Path(out_dir).mkdir(parents=True, exist_ok=True)
     console = _rich_console()
     _print_run_plan(console, args, model_spec)
+    _shared_reporter = None
     for i, name in enumerate(args.datasets):
         t0 = _time.time()
         try:
@@ -490,8 +491,16 @@ def _cmd_eval_run(args) -> int:
 
                     if RichTerminalProgress is not None:
                         # share ONE console: phase lines printed by another
-                        # writer during the live bar interleave incorrectly
+                        # writer during the live bar interleave incorrectly.
-                        progress_reporter = RichTerminalProgress(console=console)
+                        # One reporter for the WHOLE run: overall bar (which
+                        # benchmark) + sample bar (which sample), reused per
+                        # benchmark via reset_samples().
+                        if _shared_reporter is None:
+                            _shared_reporter = RichTerminalProgress(console=console)
+                            _shared_reporter.owned_externally = True
+                        progress_reporter = _shared_reporter
+                        if total_runs > 1:
+                            progress_reporter.set_overall(total_runs, i, 'benches')
 
                 def status_callback(msg, _idx=i + 1, _name=name,
                                     _reporter=progress_reporter,
@@ -569,6 +578,8 @@ def _cmd_eval_run(args) -> int:
                          'lat_p50': _pct(0.50), 'lat_p90': _pct(0.90),
                          'trunc': sum(1 for f in fins if f == 'length'),
                          'groups': groups, 'ok': True})
+            if progress_reporter is not None:
+                progress_reporter.advance_overall()
             _print_benchmark_result(console, i + 1, total_runs, name,
                                     'done', _time.time() - t0)
         except Exception as e:
@@ -586,6 +597,8 @@ def _cmd_eval_run(args) -> int:
             _print_benchmark_result(console, i + 1, total_runs, name,
                                     'failed', _time.time() - t0)
 
+    if _shared_reporter is not None:
+        _shared_reporter.close()
     if all_reports and out_dir:
         try:
             from evalharness.viz import render as _render

evalharness/data/loader.py

@@ -13,6 +13,7 @@ Supported sources:
 import csv
 import json
 import os
+import sys
 import re
 import shutil
 import urllib.request
@@ -212,6 +213,53 @@ def _parquet_ok(path: Path) -> bool:
         return False
 
 
+
+def _download_with_progress(resp, out, filename: str) -> None:
+    """Stream a download with a rich byte-level progress bar.
+
+    Used for the minute-scale dataset fetches (ModelScope blobs, HF raw
+    files); silent (plain streaming) when rich is unavailable or output
+    is redirected."""
+    total = 0
+    try:
+        total = int(resp.headers.get('Content-Length') or 0)
+    except (TypeError, ValueError):
+        total = 0
+    if not sys.stdout.isatty():
+        while True:
+            chunk = resp.read(1 << 20)
+            if not chunk:
+                break
+            out.write(chunk)
+        return
+    try:
+        from rich.progress import (BarColumn, DownloadColumn, Progress,
+                                   SpinnerColumn, TextColumn,
+                                   TimeElapsedColumn, TransferSpeedColumn)
+
+        pg = Progress(SpinnerColumn(),
+                      TextColumn('[cyan]{task.fields[name]}[/cyan]'),
+                      BarColumn(), DownloadColumn(), TransferSpeedColumn(),
+                      TextColumn('•'), TimeElapsedColumn(),
+                      transient=True)
+        with pg:
+            t = pg.add_task('download', total=total or None, name=filename)
+            while True:
+                chunk = resp.read(1 << 20)
+                if not chunk:
+                    break
+                out.write(chunk)
+                pg.advance(t, len(chunk))
+        return
+    except ImportError:
+        pass
+    while True:
+        chunk = resp.read(1 << 20)
+        if not chunk:
+            break
+        out.write(chunk)
+
+
 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)
@@ -225,11 +273,7 @@ def _ms_download(repo: str, path: str, dest_dir: Path) -> Path:
     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:
-        while True:
+        _download_with_progress(resp, out, dest.name)
-            chunk = resp.read(1 << 20)
-            if not chunk:
-                break
-            out.write(chunk)
     os.replace(tmp, dest)
     return dest
 
@@ -369,11 +413,7 @@ def _hf_download(repo: str, path: str, dest_dir: Path) -> Path:
     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:
-        while True:
+        _download_with_progress(resp, out, dest.name)
-            chunk = resp.read(1 << 20)
-            if not chunk:
-                break
-            out.write(chunk)
     os.replace(tmp, dest)
     return dest
 

evalharness/model/runner.py

@@ -405,7 +405,7 @@ async def generate_predictions(
             status_callback(f'checkpoint restored: {len(restored)} ready, {len(pending)} pending')
 
     if progress_reporter is not None:
-        progress_reporter.start(len(work), dataset_name, completed=len(restored))
+        progress_reporter.reset_samples(len(work), dataset_name, completed=len(restored))
 
     async def run_one(i_s):
         i, s = i_s
@@ -441,7 +441,10 @@ async def generate_predictions(
             status_callback(f'generation complete: {len(preds)} responses')
         return preds, usages, total_usage
     finally:
-        if progress_reporter is not None:
+        # reporter lifecycle belongs to the CALLER (CLI reuses one reporter
+        # across benchmarks and closes it after the whole run); only close
+        # here when nobody external passed it in
+        if progress_reporter is not None and not getattr(progress_reporter, 'owned_externally', False):
             progress_reporter.close()
 
 

evalharness/progress/rich_terminal.py

@@ -20,6 +20,8 @@ class RichTerminalProgress:
         # accept an EXTERNAL console: CLI phase messages and the live bar must
         # share one console, or the two writers interleave and repaint wrongly
         self.console = console or Console()
+        self.overall_id = None
+        self.overall_total = 0
         self.progress = Progress(
             SpinnerColumn(),
             TextColumn("[progress.description]{task.description}"),
@@ -41,8 +43,29 @@ class RichTerminalProgress:
         self.inflight = 0
         self.heartbeat_task = None
 
+    def set_overall(self, total: int, done: int, label: str = 'benches'):
+        """Multi-benchmark runs: a bar above the sample bar showing which
+        benchmark we are on (also covers loading/scoring phases)."""
+        self.overall_total = total
+        if self.overall_id is None:
+            self.progress.start()
+            # the shared column set reads fields from EVERY task: give the
+            # overall task the same fields or rendering raises KeyError
+            self.overall_id = self.progress.add_task(
+                f'[cyan]{label}[/cyan]', total=total, completed=min(done, total),
+                inflight=0, rate='0.00', success=done, failed=0,
+                waiting='00:00', last_result='')
+        else:
+            self.progress.update(self.overall_id, completed=min(done, total))
+
+    def advance_overall(self):
+        if self.overall_id is not None:
+            self.progress.advance(self.overall_id)
+
     def start(self, total: int, description: str, completed: int = 0):
         self.started = time.monotonic()
+        if self.task_id is not None:
+            return  # one live reporter at a time; reuse across benchmarks
         self.progress.start()
         self.task_id = self.progress.add_task(
             f"[green]{description}",
@@ -57,6 +80,23 @@ class RichTerminalProgress:
         )
         self.heartbeat_task = asyncio.create_task(self._heartbeat())
 
+    def reset_samples(self, total: int, description: str, completed: int = 0):
+        """Start (or re-target) the per-sample task for the next benchmark."""
+        self.started = time.monotonic()
+        self.inflight = 0
+        if self.task_id is None:
+            self.progress.start()
+            self.task_id = self.progress.add_task(
+                f'[green]{description}', total=total, completed=min(completed, total),
+                success=completed, failed=0, rate='0.00', inflight=0,
+                waiting='00:00', last_result='restored')
+            self.heartbeat_task = asyncio.create_task(self._heartbeat())
+        else:
+            self.progress.update(self.task_id, description=f'[green]{description}',
+                                 total=total, completed=min(completed, total),
+                                 success=completed, failed=0, rate='0.00',
+                                 inflight=0, last_result='restored')
+
     def begin_sample(self, label: str):
         if self.task_id is None:
             return