concurrency + reliability overhaul for agentic workloads

- AdaptiveGate rewritten (Netflix Gradient2): window-vs-window per-stream
  speed gradient, count-driven windows with admission stamps, no thresholds
  or mode state machine; failures x0.7 + 30s drain pause
- session-level admission for multi-turn agents (_SessionGate): in-progress
  sessions hold slots until done, newcomers queue at the door; capacity
  follows the model gate's discovered limit (CONCUR-style continuity)
- image service: memory-first register (zero docker calls for known
  images), TTL-cached docker images listing, optimistic ready when the
  daemon is unreachable (docker save contention no longer kills runs);
  es tar loading removed in favor of ModelScope shipping (ms_images.py
  per-image tar upload/pull with round-trip verification)
- runner: circuit breaker (12 consecutive failures abort the bench),
  first-failure error printed immediately
- swe_agentic: image wait / docker run / rm off the event loop; exec
  timeout becomes an observation the agent can react to; container gets
  curlrc + git low-speed aborts (stalled github downloads fail fast)
- eval run excludes its own endpoints from http_proxy (a sick personal
  proxy read as 'endpoint dead' and killed whole runs)
- progress bar shows failed count; swe agentic exec_workers 2 -> 4

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

evalharness/agent/envs/swe_agentic.py

@@ -24,6 +24,66 @@ from ..loop import Environment, register_env
 
 SENTINEL = 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'
 
+
+class _SessionGate:
+    """Session-level admission for multi-turn agent benches (CONCUR-style).
+
+    Turn-level fairness lets EVERY started session crawl at 1/N bandwidth
+    -- fine for single-turn benches, terrible for 100-turn agents: first
+    completions arrive absurdly late and all contexts stay hot. Here a
+    session HOLDS its slot for its whole life (execution continuity: an
+    in-progress session always outranks a new one), and the number of
+    admitted sessions follows the model gate's discovered limit.
+
+    No preemption: a shrink only blocks NEW sessions; running ones drain
+    naturally.
+    """
+
+    def __init__(self):
+        self.active = 0
+        self._cond = None
+        self._loop = None
+
+    def _limit(self, adapter) -> int:
+        gates = getattr(adapter, '_gates', None)
+        if gates:
+            try:
+                return max(1, int(gates[0].limit))
+            except Exception:
+                pass
+        return int(__import__('os').environ.get(
+            'EVALHARNESS_SESSION_SLOTS', '8') or 8)
+
+    async def enter(self, adapter):
+        import asyncio
+
+        loop = asyncio.get_running_loop()
+        if self._cond is None or self._loop is not loop:
+            self._loop = loop
+            self._cond = asyncio.Condition()
+            self.active = 0        # fresh loop: nothing in flight
+        while self.active >= self._limit(adapter):
+            await self._cond.acquire()
+            try:
+                await self._cond.wait()
+            finally:
+                self._cond.release()
+        self.active += 1
+
+    async def exit(self):
+        import asyncio
+
+        self.active = max(0, self.active - 1)
+        if self._cond is not None:
+            try:
+                async with self._cond:
+                    self._cond.notify_all()
+            except RuntimeError:
+                pass
+
+
+_SESSION_GATE = _SessionGate()
+
 # mini-swe-agent swebench.yaml contract (verbatim sections that matter)
 INSTANCE_TEMPLATE = """<pr_description>
 Consider the following PR description:
@@ -144,12 +204,36 @@ class SWEAgenticEnvironment(Environment):
                      image, 'tail', '-f', '/dev/null'], timeout=120)
         if r.returncode != 0:
             raise RuntimeError(f'start container failed: {r.stderr[:200]}')
+        # network hardening: github.com is reachable from these containers
+        # but large transfers stall mid-stream (GFW throttling) -- an agent
+        # curl/git then hangs until the 600s exec cap. Make curl and git
+        # abort stalled transfers themselves so the agent sees a FAST
+        # failure and reroutes instead of burning 10 minutes:
+        #   curl: ~/.curlrc applies to every agent-invoked curl
+        #   git:  lowSpeed abort on <1KB/s sustained 30s
+        _docker(['exec', name, 'bash', '-lc',
+                 'printf "connect-timeout 15\\nspeed-time 30\\nspeed-limit 1024\\n" > /root/.curlrc 2>/dev/null; '
+                 'git config --global http.lowSpeedLimit 1024 2>/dev/null; '
+                 'git config --global http.lowSpeedTime 30 2>/dev/null'], timeout=30)
         return name
 
     def _exec(self, cmd: str, timeout: int = 600) -> Dict[str, Any]:
         # bash -lc: swebench images activate the per-instance testbed via
         # shell startup files (es parity: _SWE_BENCH_INTERPRETER)
-        r = _docker(['exec', self.container, 'bash', '-lc', cmd], timeout=timeout)
+        try:
+            r = _docker(['exec', self.container, 'bash', '-lc', cmd],
+                        timeout=timeout)
+        except subprocess.TimeoutExpired:
+            # a hung command (e.g. curl to an unreachable github from a CN
+            # container) must be an OBSERVATION the agent can react to --
+            # letting it raise killed the whole task 600s in, mid-loop,
+            # after all its previous turns were already spent
+            return {'exit': 124,
+                    'out': f'COMMAND TIMED OUT after {timeout}s (killed). '
+                           'The command may be blocked on the network -- '
+                           'github.com is often unreachable from this '
+                           'container; try a mirror (ghproxy) or proceed '
+                           'without the download.'}
         out = (r.stdout or '') + (r.stderr or '')
         # cap observation: agents choke on 100k-char dumps
         if len(out) > 30000:
@@ -165,16 +249,38 @@ class SWEAgenticEnvironment(Environment):
     async def run_task(self, adapter, sample: Sample, max_turns: int = 250,
                        system: str = '', user_adapter=None, gen_kwargs=None,
                        **kw) -> Dict[str, Any]:
+        # SESSION admission BEFORE the container: hold arriving sessions at
+        # the door instead of letting all of them start containers and then
+        # crawl at 1/N turn bandwidth
+        await _SESSION_GATE.enter(adapter)
+        try:
+            return await self._run_session(adapter, sample, max_turns,
+                                           system, user_adapter, gen_kwargs)
+        finally:
+            await _SESSION_GATE.exit()
+
+    async def _run_session(self, adapter, sample: Sample, max_turns: int = 250,
+                           system: str = '', user_adapter=None,
+                           gen_kwargs=None, **kw) -> Dict[str, Any]:
         image = (sample.sandbox and sample.sandbox.image) or ''
         if not image:
             raise RuntimeError('swe_agentic: sample has no sandbox image '
                                '(dataset plugin must declare it)')
         # register on ARRIVAL: the background pool starts pulling while
-        # earlier samples are still generating
+        # earlier samples are still generating. OFF the loop: register runs
+        # `docker images` (2-4s against a 534-image daemon) and 100 tasks
+        # doing that serially ON the loop froze the whole runner at startup
         from ...sandbox.image_service import get_image_service
 
-        get_image_service().register([image])
+        svc = get_image_service()
-        self.container = self._start(image)
+        await asyncio.to_thread(svc.register, [image])
+        # NEVER wait on the image barrier on the event loop: one blocked
+        # wait_ready froze the WHOLE runner (0.00/s, every other task
+        # stalled behind it). Park the wait (and the docker run) in a
+        # thread; _start's own wait then returns instantly (already ready)
+        if not await asyncio.to_thread(svc.wait_ready, image, 3600):
+            raise RuntimeError(f'image {image} unavailable after all sources')
+        self.container = await asyncio.to_thread(self._start, image)
         ps = (sample.metadata or {}).get('problem_statement') or sample.input_text
         messages = [ChatMessage(role='user', content=INSTANCE_TEMPLATE.format(
             problem_statement=ps, sentinel=SENTINEL))]
@@ -232,7 +338,7 @@ class SWEAgenticEnvironment(Environment):
                     self._exec, 'cd /testbed && git diff')
                 patch = res['out'].strip()
         finally:
-            self._stop()
+            await asyncio.to_thread(self._stop)
 
         md = sample.metadata or {}
         return {

evalharness/cli.py

@@ -751,6 +751,25 @@ def _cmd_eval_run(args) -> int:
                                'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] +
                            [cats])
 
+    # the endpoints this run drives are DIRECT infrastructure, but httpx
+    # honors http_proxy/https_proxy by default (trust_env) -- a personal
+    # proxy in the launching shell silently routed every eval request
+    # through it, and a sick proxy read as "endpoint dead" (fast-fail
+    # cascades killed whole runs while the endpoint was perfectly fine).
+    # Exclude our own endpoints from proxying, whatever client is used.
+    import urllib.parse as _up
+
+    _np = {x.strip() for x in
+           os.environ.get('no_proxy', os.environ.get('NO_PROXY', '')).split(',')
+           if x.strip()}
+    for _u in filter(None, (getattr(args, 'api_url', ''),
+                            getattr(args, 'judge_api_url', ''))):
+        _h = _up.urlparse(_u if '//' in _u else f'//{_u}').hostname
+        if _h:
+            _np.add(_h)
+    if _np:
+        os.environ['no_proxy'] = os.environ['NO_PROXY'] = ','.join(sorted(_np))
+
     run_started = _time.time()
     total_runs = len(args.datasets)
     model_spec = _compose_model_spec(args)

evalharness/config/default.yaml

@@ -71,4 +71,4 @@ swe_bench_verified_agentic:
   max_tokens: 4096           # 单轮 bash 命令生成预算（mini-swe-agent 口径）
   temperature: 0.0
   env: swe_agentic           # 多轮 agent：bash 探索 /testbed + sentinel 提交
-  max_turns: 250             # mini-swe-agent 默认步数
+  max_turns: 100             # mini-swe-agent 默认步数

evalharness/config/thinking.yaml

@@ -60,7 +60,7 @@ tau2_bench:
   max_tokens: 16384
 
 swe_bench_verified_agentic:
-  max_tokens: 4096
+  max_tokens: 8196
   temperature: 0.0
   env: swe_agentic
   max_turns: 250

evalharness/eval/recipes/agent.py

@@ -445,7 +445,9 @@ def swe_bench_verified_agentic():
         extract='identity',
         scorers={'resolved': {'name': 'env_reward',
                               'backend': _swe_official_reward}},
-        exec_workers=2,
+        # official swebench eval spins ONE container per instance (~5 min
+        # each); 2 workers stretched 500 instances to ~20h of scoring tail
+        exec_workers=4,
         description='SWE-bench Verified AGENTIC (mini-swe-agent protocol): '
                     'multi-turn bash agent explores /testbed in the '
                     'per-instance container, submits a git diff; scored by '

evalharness/model/pool.py

@@ -14,8 +14,9 @@ expires, so one sick endpoint cannot absorb its share of the load.
 """
 
 import asyncio
-import contextlib
 import itertools
+import os
+import statistics
 import time
 from typing import Dict, List, Optional
 
@@ -103,7 +104,12 @@ class PooledAdapter(ModelAdapter):
             adapter = self._next()
             try:
                 i = self.adapters.index(adapter)
-                await self._gates[i].acquire()
+                stamp = await self._gates[i].acquire()
+                # NB: service time ONLY -- starting the clock before acquire
+                # bills the gate's own QUEUE wait as service slowness, and
+                # the shrinking medians convinced the gate the endpoint
+                # was degrading under load it was not under
+                t0 = time.monotonic()
                 ok = False
                 try:
                     out = await adapter.generate(messages, tools=tools, **kw)
@@ -113,7 +119,9 @@ class PooledAdapter(ModelAdapter):
                     # this finally AND False in the except handler, double-
                     # decrementing _inflight (gate over-admits) and applying
                     # the x0.7 backoff twice per failure
-                    self._gates[i].release(ok)
+                    self._gates[i].release(
+                        ok, stamp=stamp, dur_s=time.monotonic() - t0,
+                        tokens=(out.usage.output_tokens if ok else 0))
                 self.usage = self.usage + out.usage
                 self.stats['ok'] += 1
                 self._mark(adapter, True)
@@ -137,73 +145,57 @@ class PooledAdapter(ModelAdapter):
 
 
 class AdaptiveGate:
-    """Per-endpoint adaptive concurrency limiter (AIMD + server signals).
+    """Per-endpoint adaptive concurrency limiter (Netflix Gradient2 style).
 
-    Goal: keep the backend SATURATED (high XPU util / throughput) without
+    One formula, no thresholds, no server telemetry, no mode state machine:
-    pushing it over the cliff (500s / child crashes). Signals:
+
-      - server /metrics: num_queue_reqs > 0 means WE are pushing too hard
+        when a measurement window fills:
-        for the current mix; idle (no queue, low running) means room to grow
+            gradient = median per-stream speed this window / previous window
-      - request failures: multiplicative decrease (survive first)
+            limit    = limit * gradient + sqrt(limit)
-    Control law (classic AIMD):
+
-      +1 concurrency per probe interval when the endpoint looks underfed
+    - healthy (gradient ~1): grows by sqrt(limit) -- big steps at low
-      -1 when the server reports a queue (gentle)
+      limits (fast escape from 1), small relative steps at high limits
-      x0.7 on any failed request (fast backoff), floor at LO
+      (conservative where overshoot is fatal)
-    Purely additive to PooledAdapter: one gate per backend, no caller change.
+    - degraded (gradient <1): multiplicative shrink proportional to how
+      much each stream actually slowed
+    - window-vs-window comparison self-normalizes context growth: swe
+      sessions decode slower as they mature, and an all-time-best baseline
+      would wrongly shrink the gate forever for that alone
+    - windows are COUNT driven (min(10, 2*limit) same-stamp samples):
+      sparse workloads judge slowly by construction -- TCP's trick of
+      using RTT as the clock, inherited for free. Time-based windows
+      judge on thin evidence (a 30s window at limit 1 sees zero swe turns)
+    - every sample is stamped with the limit at admission; only
+      current-stamp samples close a window. Early completions at a new
+      level are survivorship-biased (short requests finish first) and
+      always read optimistic
+    - per-stream speed = completion_tokens/duration; gateways that report
+      no usage fall back to 1/duration (completion frequency)
+    - a failed request: limit x0.7 + a 30s admission pause so the server
+      can drain (anti death-spiral). Fully dead endpoints are the
+      runner-level circuit breaker's job.
     """
 
-    LO = 1          # never go below: progress beats perfection
+    LO = 1
-    HI = 96         # sane ceiling for one endpoint
+    HI = 96
-    PROBE_S = 5.0   # safety tick (fails/hang detection); ramp decisions use
+    INITIAL = 1.0     # '--concurrency auto' rebinds this
-                    # level statistics, not this interval alone
+    SILENCE_S = 30.0  # post-failure drain pause before admitting again
-    INITIAL = 2.0   # class-level start point ('--concurrency auto' rebinds it)
+    WINDOW_MAX = 10   # samples to close a window (2*limit when smaller)
-    # ---- 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
-        self.limit = self.INITIAL  # float for smooth x0.7; compare with int()
+        self.limit = self.INITIAL
         self._inflight = 0
         self._cond: Optional[asyncio.Condition] = None
-        self._task: Optional[asyncio.Task] = None
+        self._loop = None
         self._stopped = False
-        self._loop = None         # loop the cond/probe-task are bound to
+        self._quiet_until = 0.0              # post-failure drain window
-        self._contended = 0       # acquire() waits this probe interval (demand)
+        self._window: List[tuple] = []       # (stamp, speed) since last close
-        self._interval_fails = 0  # failed releases this probe interval
+        self._prev_median: Optional[float] = None
-        self._interval_ok = 0     # SUCCESSFUL releases this probe interval:
+        self.stats = {'window': 0, 'grow': 0, 'shrink': 0, 'backoff_fail': 0}
-                                  # 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, 'bisect': 0}
 
+    # ---- bar plumbing ----
     def _push_limit(self) -> None:
-        """Surface the current limit to the progress bar ('gate N')."""
         rep = (self.adapter.extra or {}).get('progress_reporter')
         fn = getattr(rep, 'set_gate', None)
         if fn is not None:
@@ -213,9 +205,6 @@ class AdaptiveGate:
                 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:
@@ -225,262 +214,95 @@ class AdaptiveGate:
                 pass
 
     # ---- gate semantics ----
-    async def acquire(self) -> None:
+    async def acquire(self) -> int:
+        """Admit one request; the return value is its admission stamp
+        (the limit it was admitted under) -- pass it back to release()."""
         loop = asyncio.get_running_loop()
         if self._cond is None or self._loop is not loop or self._stopped:
-            # lazy init OR LOOP CHANGE: pools are cached across benchmarks,
+            # lazy init OR LOOP CHANGE: pools survive across asyncio.run()
-            # and the CLI runs asyncio.run() per bench (per repeat!) -- a new
+            # boundaries and a Condition is loop-bound -- rebuild on the
-            # run means a new event loop while this gate survives. A Condition
+            # new loop (nothing is in flight there by construction)
-            # is loop-bound: reusing the old one raises "bound to a different
-            # event loop" under contention, and the old probe task died with
-            # the closed loop. Rebuild both; inflight resets to 0 (nothing is
-            # in flight on a fresh loop by construction).
             self._loop = loop
             self._cond = asyncio.Condition()
             self._inflight = 0
             self._stopped = False
             self._push_limit()
-            if self._task is not None:
+        while (self._inflight >= max(1, int(self.limit))
-                self._task.cancel()  # dead task from the closed loop; no-op
+               or time.monotonic() < self._quiet_until):
-            self._task = loop.create_task(self._probe_loop())
+            # timed wait: silence expiry must re-open admission even if no
-        while self._inflight >= max(1, int(self.limit)):
+            # request completes to wake us
-            self._contended += 1   # demand exceeded the cap: potential ramp fuel
             await self._cond.acquire()
             try:
-                await self._cond.wait()
+                await asyncio.wait_for(self._cond.wait(), timeout=5.0)
+            except asyncio.TimeoutError:
+                pass
             finally:
                 self._cond.release()
         self._inflight += 1
         self.push_inflight()
+        return max(1, int(self.limit))
 
-    def release(self, ok: bool) -> None:
+    def release(self, ok: bool, stamp: int = 0, dur_s: float = 0.0,
+                tokens: float = 0.0) -> None:
         self._inflight = max(0, self._inflight - 1)
-        self.push_inflight()
+        if not ok:
-        if ok:
+            self.stats['backoff_fail'] += 1
-            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._quiet_until = time.monotonic() + self.SILENCE_S
-                self.stats['backoff_fail'] += 1
+            self._push_limit()
-                self._push_limit()
+        else:
-                self._enter_level(mode='probe')
+            speed = (tokens / dur_s if tokens > 0
+                     else 1.0 / dur_s if dur_s > 0 else 0.0)
+            if speed > 0:
+                self._window.append((stamp, speed))
+                self._maybe_close_window()
+        self.push_inflight()
         self._wake()
 
-    def _enter_level(self, mode: str = '') -> None:
+    def _maybe_close_window(self) -> None:
-        """Arrive at (a new) limit: start measuring this level fresh."""
+        cur = max(1, int(self.limit))
-        if mode:
+        fresh = [s for st, s in self._window if st == cur]
-            self._mode = mode
+        if len(fresh) < min(self.WINDOW_MAX, 2 * cur):
-        self._level_t0 = time.monotonic()
+            return                       # not enough evidence yet: keep waiting
-        self._level_ok = 0
+        med = statistics.median(fresh)
+        if self._prev_median:
+            # dampers (Netflix/Envoy both do this): small-window medians
+            # are noisy -- a bare ratio swings the limit 5 -> 1 -> 9 -> 2
+            # in simulation. Clamp the gradient band and smooth the
+            # reference so one jittery window cannot whipsaw the gate.
+            g = max(0.5, min(1.5, med / self._prev_median))
+            if g < 0.95:
+                # clearly degraded: PURE multiplicative shrink -- adding
+                # sqrt headroom here let a halved speed at low limits
+                # still RAISE the limit (sim: 3.4 x 0.5 + sqrt(3.4) = 3.6)
+                new = self.limit * g
+                self.stats['shrink'] += 1
+            else:
+                # healthy (or within noise): gentle sqrt probe upward
+                new = self.limit * g + self.limit ** 0.5
+                if new > self.limit:
+                    self.stats['grow'] += 1
+            self.limit = max(float(self.LO), min(float(self.HI), new))
+            self._push_limit()
+        self._prev_median = (med if self._prev_median is None
+                             else 0.6 * self._prev_median + 0.4 * med)
+        self._window = []
+        self.stats['window'] += 1
 
     def _wake(self) -> None:
-        if self._cond is not None:
+        if self._cond is None:
-            # fire-and-forget notify (loop may not be ours -- best effort)
+            return
-            try:
+        try:
-                fut = asyncio.ensure_future(self._notify_all())
+            asyncio.get_running_loop().create_task(self._notify_all())
-                fut.add_done_callback(lambda f: None)
+        except RuntimeError:
-            except RuntimeError:
+            pass                         # no running loop (shutdown)
-                pass
 
     async def _notify_all(self) -> None:
         async with self._cond:
             self._cond.notify_all()
 
-    # ---- server-signal probe ----
-    def _no_signal_ramp(self) -> None:
-        """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'
-
-        def _fetch():
-            with urllib.request.urlopen(url, timeout=4) as resp:
-                return resp.read().decode('utf-8', 'ignore')
-
-        try:
-            # 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'):
-                running = float(line.rsplit(' ', 1)[-1])
-            elif line.startswith('sglang:num_queue_reqs'):
-                queue = float(line.rsplit(' ', 1)[-1])
-        self.stats['probe'] += 1
-        if queue is None and running is None:
-            self._no_signal_ramp()
-            return
-        if queue is not None and queue >= 2:
-            # server is queuing OUR excess: gentle additive decrease
-            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))) \
-                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()
-        else:
-            self.stats['hold_queue'] += 1
-        self._interval_fails = 0
-        self._contended = 0
-        self._wake()
-
-    async def _probe_loop(self) -> None:
-        import contextlib
-
-        while not self._stopped:
-            with contextlib.suppress(Exception):
-                await self._probe_once()
-            await asyncio.sleep(self.PROBE_S)
-
     def stop(self) -> None:
         self._stopped = True
-        if self._task is not None:
-            self._task.cancel()
 
     def report(self) -> Dict[str, float]:
         return {'limit': max(1, int(self.limit)), 'inflight': self._inflight,

evalharness/model/runner.py

@@ -446,10 +446,13 @@ async def generate_predictions(
             get_image_service().register(sorted(_imgs))
     except Exception:
         pass
-        # let the adapter surface retry attempts to the bar
+    # let the adapter surface retry attempts / gate limits to the bar
-        members = getattr(adapter, 'adapters', [adapter])
+    # (NB: this was indented INSIDE the except above -- it only ran when
-        for m_ in members:
+    # image registration THREW, so the gate never pushed its limit and
-            m_.extra['progress_reporter'] = progress_reporter
+    # the bar showed a bare in-flight count with no 'gate N')
+    members = getattr(adapter, 'adapters', [adapter])
+    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)
@@ -457,6 +460,13 @@ async def generate_predictions(
     # (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] = {}
+    # circuit breaker: a dead/wedged endpoint makes EVERY sample fail fast
+    # (~7s each) -- without this the runner burns the whole bench in 15
+    # minutes marking everything failed (a 465-sample run lost 98/100 to
+    # a dead sglang before anyone noticed). N consecutive failures with
+    # zero successes in between = abort now; --resume retries later.
+    _consec_fail = [0]
+    CB_ABORT_AT = 12
 
     async def run_one(i_s):
         i, s = i_s
@@ -469,7 +479,19 @@ async def generate_predictions(
             if progress_reporter is not None:
                 progress_reporter.advance(success=False)
             failed_samples[i] = f'{type(e).__name__}: {str(e)[:120]}'
+            if len(failed_samples) == 1:
+                # surface the FIRST failure's error immediately -- the
+                # end-of-generation summary arrives minutes later and
+                # every fast-fail cascade diagnosis died waiting for it
+                print(f'⚡ FIRST FAILURE: {failed_samples[i]}', flush=True)
+            _consec_fail[0] += 1
+            if _consec_fail[0] >= CB_ABORT_AT:
+                raise RuntimeError(
+                    f'circuit breaker: {_consec_fail[0]} consecutive '
+                    f'failures, endpoint appears down (first: '
+                    f'{next(iter(failed_samples.values()))[:150]})')
             return i, None   # empty marker: no checkpoint write
+        _consec_fail[0] = 0
         if ckpt_store is not None:
             ckpt_store.append(keys[i], pred)
         return i, pred

evalharness/progress/rich_terminal.py

@@ -141,6 +141,7 @@ class RichTerminalProgress:
         self.inflight = 0
         self.bench_name = description
         self.restored = max(completed, 0)  # checkpoint head start this bench
+        self._failed = 0                   # per-bench failure counter reset
         self._scoring_for = None           # next scoring phase retargets anew
         desc = f'[green]{self.bench_tag}{description} · generating[/green]'
         if self.task_id is None:
@@ -175,6 +176,7 @@ class RichTerminalProgress:
         self.started = time.monotonic()
         self.inflight = 0
         self.restored = 0
+        self._failed = 0
         self.admitted = None
         self.progress.update(
             self.task_id,
@@ -277,14 +279,21 @@ class RichTerminalProgress:
         task = self.progress.tasks[self.task_id]
         completed = task.completed + 1
         self.inflight = max(0, self.inflight - 1)
+        if not success:
+            # failed samples advance the bar too (the bench must finish) --
+            # but surface the count so 'Completed 310/500' cannot masquerade
+            # as progress when 275 of them died (e.g. image-wait timeouts)
+            self._failed = getattr(self, '_failed', 0) + 1
         elapsed = max(time.monotonic() - self.started, 1e-6)
         # rate/eta over THIS RUN's fresh samples only: counting the restored
         # head start would print 141/6s = 23/s when 1 sample was generated
         fresh = max(completed - getattr(self, 'restored', 0), 0)
+        failed_txt = (f' [red]· {self._failed} failed[/red]'
+                      if getattr(self, '_failed', 0) else '')
         self.progress.update(
             self.task_id,
             advance=1,
-            new=self._new_txt(completed),
+            new=self._new_txt(completed) + failed_txt,
             rate=f"{fresh / elapsed:.2f}",
             inflight=self._inflight_txt(), cur='0s',
             elapsed=_fmt_eta(elapsed),

evalharness/sandbox/image_service.py

@@ -2,11 +2,10 @@
 
 One service per process. Benches declare the images they will need (the
 runner registers them when a task arrives); a small worker pool pulls
-them in the background — local tar batches first (es's swebench_v
+them in the background from the CN mirror chain. Sample execution calls
-shipment, 500 images offline), network mirror chain second. Sample
+``ensure``/``wait_ready`` BEFORE ``docker run``: an image still in flight
-execution calls ``ensure``/``wait_ready`` BEFORE ``docker run``: an image
+BLOCKS that sample (no failure, no timeout at 120s) until the pull
-still in flight BLOCKS that sample (no failure, no timeout at 120s) until
+service delivers it or exhausts all sources.
-the pull service delivers it or exhausts all sources.
 
 This replaces the two broken timings:
   - gen-phase `docker run` implicit pull (zero output, 120s timeout kill)
@@ -18,51 +17,29 @@ import os
 import queue
 import subprocess
 import threading
-from pathlib import Path
 from typing import Dict, Iterable, List, Optional
 
-# local tar shipments searched before the network (dir -> glob pattern)
-_TAR_SOURCES = [
-    ('/data1/sora/evalscope/docker/swebench_v', 'swebench_batch_*.tar.gz'),
-    ('/data1/sora/evalscope/docker', 'bigcodebench-sandbox.tar.gz'),
-]
-
 
 def _docker(args, timeout=300):
     return subprocess.run(['docker'] + args, capture_output=True, text=True,
                           timeout=timeout)
 
 
+_LOCAL_TTL_S = 60.0          # `docker images` on a daemon busy with saves
+_LOCAL_CACHE = [0.0, set()]  # [expires_at, names] -- one query per TTL
+
+
 def _local_images() -> set:
+    import time
+
+    now = time.monotonic()
+    if now < _LOCAL_CACHE[0]:
+        return _LOCAL_CACHE[1]
     r = _docker(['images', '--format', '{{.Repository}}:{{.Tag}}'], timeout=60)
-    return set(r.stdout.split()) if r.returncode == 0 else set()
+    names = set(r.stdout.split()) if r.returncode == 0 else set()
-
+    _LOCAL_CACHE[0] = now + _LOCAL_TTL_S
-
+    _LOCAL_CACHE[1] = names
-def _tar_manifest_tags(path: str) -> List[str]:
+    return names
-    """RepoTags inside a docker-save tar ( OCI layout or legacy )."""
-    import tarfile
-
-    try:
-        with tarfile.open(path, 'r:gz') as t:
-            for name in ('manifest.json', 'index.json'):
-                try:
-                    member = t.getmember(name)
-                except KeyError:
-                    continue
-                data = json.load(t.extractfile(member))
-                if name == 'manifest.json':
-                    return [tg for e in data for tg in (e.get('RepoTags') or [])]
-                # OCI index: ref lives in annotations; best effort
-                refs = []
-                for m in data.get('manifests', []):
-                    ref = (m.get('annotations') or {}).get(
-                        'org.opencontainers.image.ref.name', '')
-                    if ref:
-                        refs.append(ref)
-                return refs
-    except Exception:
-        return []
-    return []
 
 
 class ImageService:
@@ -76,7 +53,6 @@ class ImageService:
         self._events: Dict[str, threading.Event] = {}
         self._queue: 'queue.Queue[Optional[str]]' = queue.Queue()
         self._threads: List[threading.Thread] = []
-        self._tar_index: Optional[Dict[str, str]] = None  # tag -> tar path
         self._started = False
         self._pulled: List[str] = []            # for atexit release
 
@@ -92,19 +68,37 @@ class ImageService:
             self._threads.append(t)
 
     def register(self, images: Iterable[str]):
-        """Declare upcoming images (called when tasks arrive). Idempotent;
+        """Declare upcoming images (called when tasks arrive). Idempotent
-        already-local images skip the queue entirely."""
+        and MEMORY-FIRST: the run bulk-registers everything once, so the
+        per-task register must be a dict hit with ZERO docker calls --
+        `docker images` per task (x100) against a daemon busy with
+        concurrent `docker save`s timed out at 60s and cascaded whole
+        runs into failure. Only genuinely UNKNOWN images trigger one
+        (TTL-cached) docker query for the batch; if even that fails under
+        daemon pressure, mark ready optimistically and let `docker run`
+        be the referee -- it has its own timeout and a clear error."""
         self.start()
-        local = _local_images()
+        unknown = []
-        for img in images:
+        with self._lock:
-            if not img:
+            for img in images:
-                continue
+                if not img:
-            with self._lock:
+                    continue
                 if img in self._wanted:
                     continue
                 self._wanted[img] = 'pulling'
                 self._events[img] = threading.Event()
-            if img in local:
+                unknown.append(img)
+        if not unknown:
+            return
+        try:
+            local = _local_images()
+        except Exception:
+            local = None       # daemon query failed: optimistic path below
+        for img in unknown:
+            if local is None or img in local:
+                if local is None:
+                    self._log(f'· docker images unavailable -- assuming '
+                              f'local: {img.split("/")[-1][:44]}')
                 with self._lock:
                     self._wanted[img] = 'ready'
                 self._events[img].set()
@@ -143,16 +137,29 @@ class ImageService:
             self._queue.task_done()
 
     def _pull_one(self, image: str) -> bool:
-        # 1. local tar shipment (offline, fastest)
+        # fast path: a concurrent worker / previous run may have delivered
-        tar = self._find_in_tars(image)
+        # it already (also flips any other queued-now-local images ready)
-        if tar:
+        if image in _local_images():
-            self._log(f'· loading {image.split("/")[-1][:44]} from {Path(tar).parent.name}/{Path(tar).name}')
+            self._sweep_local()
-            r = _docker(['load', '-qi', tar], timeout=1800)
+            return True
-            if r.returncode == 0 and image in _local_images():
+        # 1. OUR ModelScope dataset shipment (CN CDN, per-image tars). The
-                self._pulled.append(image)
+        #    swebench/* namespace is 403 on every public CN mirror, so this
-                return True
+        #    is the only network source for those images. Opt out with
-            self._log(f'  tar load failed, falling back to network')
+        #    EVALHARNESS_MS_IMAGE_REPO='' .
-        # 2. mirror chain (progress + watchdog inside)
+        repo = os.environ.get('EVALHARNESS_MS_IMAGE_REPO',
+                              'SoraAmami/swebench-images')
+        if repo:
+            try:
+                from .ms_images import ms_pull
+
+                if ms_pull(image, repo):
+                    self._pulled.append(image)
+                    self._sweep_local()
+                    return True
+            except Exception:
+                pass
+        # 2. CN mirror chain (progress + idle watchdog inside); on hit the
+        #    puller retags to the canonical name so callers see it directly
         try:
             from .prefetch import _pull_one as net_pull
 
@@ -164,42 +171,23 @@ class ImageService:
             pass
         return False
 
-    # ---- tar index ----
+    def _sweep_local(self):
-    _INDEX_CACHE = '/tmp/evalharness_tar_index.json'
+        """A tar load makes ~10 images local at once: flip every queued
-
+        'pulling' image that is now local to 'ready' (and drop it from the
-    def _find_in_tars(self, image: str) -> Optional[str]:
+        release list duty) -- otherwise each waits its own full queue turn
-        # the full 50-tar index costs minutes to build (gzip full-scan);
+        just to discover it already arrived."""
-        # cache it on disk so only the FIRST process ever pays
+        with self._lock:
-        import glob
+            waiting = [img for img, st in self._wanted.items()
-
+                       if st == 'pulling']
-        if self._tar_index is None:
+        if not waiting:
-            cache = {}
+            return
-            try:
+        local = _local_images()
-                cache = json.load(open(self._INDEX_CACHE))
+        with self._lock:
-                # invalidate when the tar set changes
+            for img in waiting:
-                cur = sorted(glob.glob('/data1/sora/evalscope/docker/swebench_v/*.tar.gz'))
+                if img in local:
-                if cache.get('_files') != [os.path.basename(x) for x in cur]:
+                    self._wanted[img] = 'ready'
-                    cache = {}
+                    self._events[img].set()
-            except Exception:
+                    self._pulled.append(img)  # we caused it; release at exit
-                cache = {}
-            if cache.get('tags'):
-                self._tar_index = cache['tags']
-            else:
-                self._tar_index = {}
-                for d, pat in _TAR_SOURCES:
-                    if not os.path.isdir(d):
-                        continue
-                    for f in sorted(glob.glob(os.path.join(d, pat))):
-                        for tag in _tar_manifest_tags(f):
-                            self._tar_index.setdefault(tag, f)
-                try:
-                    files = [os.path.basename(x) for x in
-                             glob.glob('/data1/sora/evalscope/docker/swebench_v/*.tar.gz')]
-                    json.dump({'_files': sorted(files), 'tags': self._tar_index},
-                              open(self._INDEX_CACHE, 'w'))
-                except Exception:
-                    pass
-        return self._tar_index.get(image)
 
     # ---- misc ----
     def _log(self, msg):

evalharness/sandbox/ms_images.py

@@ -0,0 +1,214 @@
+"""ModelScope-hosted sandbox images: per-image tar upload + download.
+
+registry.modelscope.cn does not accept personal docker pushes, so images
+ship as ONE docker-save tar per image inside a ModelScope MODEL repo
+(file hosting with a stable CN CDN -- measured ~12MB/s anonymous; the
+per-task granularity means running 3 instances downloads exactly 3 tars).
+This is the only working network source for the swebench/* namespace --
+public CN mirrors (daocloud/1ms/...) 403 it.
+
+NB: it must be a MODEL repo, not a dataset repo: upload_file defaults to
+repo_type='model' and silently CREATES one when given a dataset repo_id
+(the first upload landed in a phantom model repo while the dataset repo
+stayed empty -- confusing 404s on pull).
+
+Upload (resumable; skips tars already on the remote):
+    EVALHARNESS_MS_TOKEN=ms-... python -m evalharness.sandbox.ms_images \\
+        upload --repo SoraAmami/swebench-images [--limit N]
+
+Download side is wired into ImageService._pull_one (first source, before
+the docker-hub mirror chain) whenever EVALHARNESS_MS_IMAGE_REPO is set
+(defaults on for this deployment; empty string disables).
+"""
+
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from typing import List, Optional, Set
+
+# a private-repo token for downloads (public repos work without it)
+_MS_TOKEN = os.environ.get('EVALHARNESS_MS_TOKEN', '').strip()
+_MS_ENDPOINT = os.environ.get('EVALHARNESS_MS_ENDPOINT', 'https://modelscope.cn')
+# big enough for a ~4GB uncompressed save; / has less headroom than /data1
+_TMP_ROOT = os.environ.get('EVALHARNESS_MS_TMP', '/data1/sora/temp/ms_images')
+_STATE = os.path.join(_TMP_ROOT, 'uploaded.txt')   # resume manifest
+
+# image prefixes this deployment ships (swe per-instance + code benches)
+_DEFAULT_PREFIXES = ('swebench/',)
+
+
+def _docker(args, timeout=1800):
+    return subprocess.run(['docker'] + args, capture_output=True, text=True,
+                          timeout=timeout)
+
+
+def file_for(image: str) -> str:
+    """Image ref -> dataset filename: last path segment minus tag.
+
+    swebench/sweb.eval.x86_64.django_1776_django-13964:latest
+        -> sweb.eval.x86_64.django_1776_django-13964.tar.gz
+    """
+    base = image.rsplit('/', 1)[-1]
+    base = base.split(':')[0]
+    return f'{base}.tar.gz'
+
+
+def image_from_file(fname: str) -> Optional[str]:
+    """Inverse of file_for for the swe namespace (namespace is not
+    recoverable in general; callers here only ship swebench/*)."""
+    if not fname.endswith('.tar.gz'):
+        return None
+    return f"swebench/{fname[:-len('.tar.gz')]}:latest"
+
+
+def _local_images() -> Set[str]:
+    r = _docker(['images', '--format', '{{.Repository}}:{{.Tag}}'], timeout=60)
+    return set(r.stdout.split()) if r.returncode == 0 else set()
+
+
+def _remote_files(repo: str) -> Set[str]:
+    """Files currently in the MODEL repo (resume manifest source of truth)."""
+    from modelscope.hub.api import HubApi
+
+    api = HubApi()
+    if _MS_TOKEN:
+        api.login(_MS_TOKEN)
+    files: Set[str] = set()
+    for e in api.get_model_files(repo, recursive=True):
+        p = (e.get('Path') or e.get('Name') or '').lstrip('/')
+        if p.endswith('.tar.gz'):
+            files.add(p)
+    return files
+
+
+def _upload_one(api, repo: str, image: str, tmp_dir: str) -> bool:
+    tar = os.path.join(tmp_dir, file_for(image))
+    os.makedirs(tmp_dir, exist_ok=True)
+    # gzip -1: ~2x faster than default for ~10% more bytes -- upload wall
+    # time is dominated by docker save IO, not the extra size
+    r = subprocess.run(f'docker save {image} | gzip -1 > {tar}',
+                       shell=True, timeout=3600)
+    if r.returncode != 0 or not os.path.exists(tar):
+        print(f'  save failed: {image}', flush=True)
+        return False
+    try:
+        api.upload_file(repo_id=repo, path_or_fileobj=tar,
+                        path_in_repo=file_for(image),
+                        repo_type='model', token=_MS_TOKEN or None)
+        with open(_STATE, 'a') as f:
+            f.write(file_for(image) + '\n')
+        gb = os.path.getsize(tar) / 1e9
+        print(f'  uploaded {file_for(image)} ({gb:.2f} GB)', flush=True)
+        return True
+    except Exception as e:
+        print(f'  upload failed {image}: {type(e).__name__}: {str(e)[:120]}',
+              flush=True)
+        return False
+    finally:
+        try:
+            os.unlink(tar)
+        except OSError:
+            pass
+
+
+def upload(repo: str, limit: int = 0, prefixes=None) -> None:
+    """Ship every local matching image to the MODEL repo. Resumable:
+    already-remote tars are skipped (state file + remote listing)."""
+    from modelscope.hub.api import HubApi
+
+    os.makedirs(_TMP_ROOT, exist_ok=True)
+    api = HubApi()
+    if _MS_TOKEN:
+        api.login(_MS_TOKEN)
+    try:
+        api.get_model(repo)
+    except Exception:
+        print(f'creating model repo {repo} ...', flush=True)
+        api.create_model(model_id=repo)
+    done = set()
+    try:
+        done = {l.strip() for l in open(_STATE) if l.strip()}
+    except OSError:
+        pass
+    print(f'remote listing {repo} ...', flush=True)
+    try:
+        done |= _remote_files(repo)
+    except Exception as e:
+        print(f'  remote listing failed ({e}); relying on local state',
+              flush=True)
+    prefixes = prefixes or _DEFAULT_PREFIXES
+    imgs = sorted(i for i in _local_images() if i.startswith(prefixes))
+    if limit:
+        imgs = imgs[:limit]
+    todo = [i for i in imgs if file_for(i) not in done]
+    print(f'{len(imgs)} local images, {len(todo)} to upload '
+          f'({len(imgs) - len(todo)} already remote)', flush=True)
+    ok = 0
+    for n, img in enumerate(todo, 1):
+        print(f'[{n}/{len(todo)}] {img}', flush=True)
+        for attempt in range(3):
+            if _upload_one(api, repo, img, _TMP_ROOT):
+                ok += 1
+                break
+            time.sleep(10 * (attempt + 1))     # hub hiccups: backoff + retry
+    print(f'done: {ok}/{len(todo)} uploaded', flush=True)
+
+
+def ms_pull(image: str, repo: str) -> bool:
+    """Fetch one image tar from the dataset repo and docker-load it.
+    Returns True iff the image is local afterwards."""
+    import urllib.request
+
+    fname = file_for(image)
+    url = (f'{_MS_ENDPOINT}/api/v1/models/{repo}/repo'
+           f'?Revision=master&FilePath={fname}')
+    hdrs = {'User-Agent': 'evalharness'}
+    if _MS_TOKEN:
+        hdrs['Authorization'] = f'Bearer {_MS_TOKEN}'
+    os.makedirs(_TMP_ROOT, exist_ok=True)
+    fd, tar = tempfile.mkstemp(suffix='.tar.gz', dir=_TMP_ROOT)
+    os.close(fd)
+    try:
+        req = urllib.request.Request(url, headers=hdrs)
+        t0 = time.time()
+        with urllib.request.urlopen(req, timeout=600) as r, open(tar, 'wb') as f:
+            while True:
+                chunk = r.read(1 << 20)
+                if not chunk:
+                    break
+                f.write(chunk)
+        r = _docker(['load', '-qi', tar], timeout=1800)
+        if r.returncode == 0 and image in _local_images():
+            gb = os.path.getsize(tar) / 1e9
+            print(f'· ms-images: loaded {image.split("/")[-1][:50]} '
+                  f'({gb:.2f} GB in {time.time() - t0:.0f}s)', flush=True)
+            return True
+        print(f'· ms-images: load failed for {image}: '
+              f'{(r.stderr or "")[:120]}', flush=True)
+        return False
+    except Exception as e:
+        # 404 = this image was never uploaded; anything else = transient
+        code = getattr(e, 'code', None)
+        if code != 404:
+            print(f'· ms-images: fetch {fname}: {type(e).__name__}: '
+                  f'{str(e)[:100]}', flush=True)
+        return False
+    finally:
+        try:
+            os.unlink(tar)
+        except OSError:
+            pass
+
+
+if __name__ == '__main__':
+    import argparse
+
+    p = argparse.ArgumentParser()
+    p.add_argument('cmd', choices=['upload'])
+    p.add_argument('--repo', default='SoraAmami/swebench-images')
+    p.add_argument('--limit', type=int, default=0)
+    a = p.parse_args()
+    if a.cmd == 'upload':
+        upload(a.repo, limit=a.limit)