Contain terminal sample failures; 240s stream first-byte timeout

One sample whose stream never got a first byte (prefill queue at high
gate levels) exhausted 6 adapter retries and killed the WHOLE gather --
500 samples died with it. Terminal failures are now contained: empty
prediction (scores wrong, es-parity for timeouts), NOT checkpointed so
a rerun retries them, prominently counted; only a 100% wipeout fails
the bench. Stream-aggregate read timeout 60s -> 240s: a 128k prompt
queued behind other prefills legitimately takes minutes to start
answering; real hangs are now the gate's job (x0.7) and contained
failures rather than bench death.

Verified: 1-in-3 terminal failures -> bench completes 6 ok / 3 empty,
checkpoint holds only the 6.

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

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]

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);