--auto-concurrency: let the adaptive gate drive request concurrency

- new flag wraps even a single endpoint as a one-member pool so the
  per-endpoint AdaptiveGate takes over; --concurrency becomes the gate's
  STARTING point (global semaphore lifted to the gate ceiling of 96)
- demand-driven AIMD fallback for endpoints without /metrics (404,
  gateway-stripped, non-sglang): ramp +1 while callers wait on acquire
  and the interval is failure-free; the old code early-returned on
  fetch errors and never adapted at all. Real GLM endpoint verified:
  /metrics is 404, so this fallback is the live path there
- probe fetch moved to a thread: a blocked urlopen parked the whole
  event loop for its 4s timeout
- current limit surfaces on the progress bar as a magenta 'gate N'
  field (pushed on every change)

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

README.md

@@ -161,6 +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` 是起点不是上限 |
 | `--resume [PATH]` | 断点续跑；默认 `<cache-dir>/ckpt/<bench>.jsonl` |
 | `--env NAME` | agent 环境（`bfcl_mock` 等） |
 | `--perf` | 采集流式 TTFT / ITL / 重试率入报告 |

evalharness/cli.py

@@ -400,6 +400,12 @@ def _compose_model_spec(args):
         if not model:
             raise SystemExit('error: --api-url requires --model (model name)')
         model = f'{internal_provider}/{api_url.rstrip("/")}?{model}'
+    # --auto-concurrency: the per-endpoint adaptive gate decides how many
+    # requests fly; it lives in the pool layer, so wrap even a single
+    # endpoint as a one-member pool (round-robin of 1 = identity)
+    if getattr(args, 'auto_concurrency', False) and api_url \
+            and internal_provider == 'openai':
+        model = f'openai-pool/{api_url.rstrip("/")}?{model.split("?", 1)[1]}'
     return _model_with_flags(model, args)
 
 
@@ -618,6 +624,12 @@ def _cmd_eval_run(args) -> int:
     run_started = _time.time()
     total_runs = len(args.datasets)
     model_spec = _compose_model_spec(args)
+    if getattr(args, 'auto_concurrency', False) and model_spec:
+        # the gate starts where the user's manual concurrency was, then
+        # adapts on its own signals (see pool.AdaptiveGate)
+        from evalharness.model.pool import AdaptiveGate
+
+        AdaptiveGate.INITIAL = float(max(2, getattr(args, 'concurrency', 8)))
     if not out_dir and model_spec and not args.out:
         # always persist results: default dir = evalharness-results/<stamp>-<model>/
         import re as _re
@@ -744,7 +756,10 @@ def _cmd_eval_run(args) -> int:
                     if _repeats > 1:
                         print(f'\n[repeat {_rep + 1}/{_repeats}]', flush=True)
                     report = asyncio.run(run_eval(
-                        ds, model_spec, concurrency=args.concurrency,
+                        ds, model_spec,
+                        concurrency=max(args.concurrency, 96)
+                        if getattr(args, 'auto_concurrency', False)
+                        else args.concurrency,
                         limit=args.limit, limit_per_task=args.limit_per_task,
                         gen_kwargs=_gen_kw or None,
                         max_input_tokens=_mit,
@@ -1067,6 +1082,12 @@ 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='',

evalharness/model/pool.py

@@ -154,17 +154,30 @@ class AdaptiveGate:
     LO = 2          # 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)
 
     def __init__(self, adapter: ModelAdapter):
         self.adapter = adapter
-        self.limit = 8.0          # float for smooth x0.7; compare with int()
+        self.limit = self.INITIAL  # float for smooth x0.7; compare with int()
         self._inflight = 0
         self._cond: Optional[asyncio.Condition] = None
         self._task: Optional[asyncio.Task] = None
         self._stopped = False
         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.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0,
-                      'backoff_queue': 0}
+                      'backoff_queue': 0, 'ramp_demand': 0}
+
+    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:
+            try:
+                fn(max(1, int(self.limit)))
+            except Exception:
+                pass
 
     # ---- gate semantics ----
     async def acquire(self) -> None:
@@ -181,10 +194,12 @@ class AdaptiveGate:
             self._cond = asyncio.Condition()
             self._inflight = 0
             self._stopped = False
+            self._push_limit()
             if self._task is not None:
                 self._task.cancel()  # dead task from the closed loop; no-op
             self._task = loop.create_task(self._probe_loop())
         while self._inflight >= max(1, int(self.limit)):
+            self._contended += 1   # demand exceeded the cap: potential ramp fuel
             await self._cond.acquire()
             try:
                 await self._cond.wait()
@@ -195,10 +210,12 @@ class AdaptiveGate:
     def release(self, ok: bool) -> None:
         self._inflight = max(0, self._inflight - 1)
         if not ok:  # multiplicative decrease -- survival first
+            self._interval_fails += 1
             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._wake()
 
     def _wake(self) -> None:
@@ -215,15 +232,37 @@ class AdaptiveGate:
             self._cond.notify_all()
 
     # ---- 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)
+            self.stats['ramp_demand'] += 1
+            self._push_limit()
+        self._interval_fails = 0
+        self._contended = 0
+        self._wake()
+
     async def _probe_once(self) -> None:
         import urllib.request
 
         url = f'{self.adapter.api_base.rstrip("/")}/metrics'
-        try:
+
+        def _fetch():
             with urllib.request.urlopen(url, timeout=4) as resp:
-                text = resp.read().decode('utf-8', 'ignore')
+                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)
         except Exception:
-            return  # no metrics (or busy): hold current limit
+            self._no_signal_ramp()  # no metrics: demand-driven fallback
+            return
         running = queue = None
         for line in text.splitlines():
             if line.startswith('sglang:num_running_reqs'):
@@ -232,17 +271,22 @@ class AdaptiveGate:
                 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))):
             # underfed: no queue and running below our own cap -> ramp up
             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:

evalharness/progress/rich_terminal.py

@@ -46,6 +46,7 @@ class RichTerminalProgress:
             TextColumn("• Completed {task.completed}/{task.total} [dim]{task.fields[new]}[/dim]"),
             TextColumn("• in-flight [yellow]{task.fields[inflight]}[/yellow] ([yellow]{task.fields[cur]}[/yellow])"),
             TextColumn("• [red]retries {task.fields[retries]}[/red]"),
+            TextColumn("[magenta]{task.fields[gate]}[/magenta]"),
             TextColumn("• [dim]{task.fields[rate]}/s[/dim]"),
             TextColumn("• [green]{task.fields[elapsed]}[/green]"),
             TextColumn("• [cyan]eta {task.fields[eta]}[/cyan]"),
@@ -102,6 +103,7 @@ class RichTerminalProgress:
             inflight=0,
             cur="0s",
             retries=0,
+            gate='',
             elapsed="0s",
             eta="-",
         )
@@ -130,13 +132,13 @@ class RichTerminalProgress:
             self.task_id = self.progress.add_task(
                 desc, total=total, completed=min(completed, total),
                 new=self._new_txt(completed), rate='0.00', inflight=0,
-                cur='0s', elapsed='0s', eta='-', retries=0)
+                cur='0s', elapsed='0s', eta='-', retries=0, gate='')
         else:
             self.progress.update(self.task_id, description=desc,
                                  total=total, completed=min(completed, total),
                                  new=self._new_txt(completed), rate='0.00',
                                  inflight=0, cur='0s', elapsed='0s', eta='-',
-                                 retries=0)
+                                 retries=0, gate='')
         # ALWAYS recreate the heartbeat: the previous one may have died during
         # pause/resume cycles between benchmarks (stale reference -> silent
         # death -> frozen clock while the spinner still animates)
@@ -200,6 +202,11 @@ class RichTerminalProgress:
         if self.task_id is not None:
             self.progress.update(self.task_id, retries=n)
 
+    def set_gate(self, n: int):
+        """Show the adaptive concurrency gate's current limit ('gate 12')."""
+        if self.task_id is not None:
+            self.progress.update(self.task_id, gate=f'gate {n}')
+
     def rollback(self):
         if self.disabled:
             return