diff --git a/evalharness/model/adapter.py b/evalharness/model/adapter.py index 603921a..000efc9 100644 --- a/evalharness/model/adapter.py +++ b/evalharness/model/adapter.py @@ -450,9 +450,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] diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index bdbc4b8..b181cff 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -438,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 @@ -464,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);