Eval robustness: fresh HarborEnv client per rollout (openenv reconnect bug); failed checkpoint evals retried next poll

eval_adapter.py

@@ -22,7 +22,6 @@ Config via env:
 
 import json
 import os
-import threading
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from harbor_env import HarborEnv  # re-export of openenv.harbor.client.HarborEnv
@@ -37,13 +36,14 @@ N_TASKS = int(os.environ.get("N_TASKS", "50"))
 CONCURRENCY = int(os.environ.get("CONCURRENCY", "8"))
 OUT = os.environ.get("OUT", "/app/eval_results.json")
 
-_local = threading.local()
-
-
 def run_one(task_index, model):
-    """One graded rollout. Returns (correct, failed, error_or_none)."""
-    if not hasattr(_local, "env"):
-        _local.env = HarborEnv(base_url=ENV_URL)
+    """One graded rollout. Returns (correct, failed, error_or_none).
+
+    A FRESH client per rollout: HarborSessionFactory documents that shared
+    clients break under concurrency, and a reused websocket hits an openenv
+    reconnect bug (mcp_client close() returning None) that kills the run.
+    """
+    env = HarborEnv(base_url=ENV_URL)
     kwargs = dict(
         split=TEST_SPLIT,
         task_index=task_index,
@@ -58,13 +58,19 @@ def run_one(task_index, model):
         agent_timeout_sec=600,
     )
     try:
-        result = _local.env.run_rollout(**kwargs)
-    except TypeError:
-        # older server without eval_sampling: retry without it
-        kwargs.pop("eval_sampling")
-        result = _local.env.run_rollout(**kwargs)
+        try:
+            result = env.run_rollout(**kwargs)
+        except TypeError:
+            # older server without eval_sampling: retry without it
+            kwargs.pop("eval_sampling")
+            result = env.run_rollout(**kwargs)
     except Exception as exc:
         return task_index, model, None, True, f"{type(exc).__name__}: {exc}"
+    finally:
+        try:
+            env.close()
+        except Exception:
+            pass
     if not result.ok or result.reward is None:
         return task_index, model, None, True, getattr(result, "error", None) or "not ok"
     return task_index, model, float(result.reward) == 1.0, False, None

eval_watch.py

@@ -93,7 +93,13 @@ def main():
             loaded.append(ckpt)
             if len(loaded) > 4:  # stay under --max-loras
                 lora_rpc("unload_lora_adapter", loaded.pop(0))
-            run_eval(ckpt, f"/tmp/{ckpt}_cells.json")
+            try:
+                run_eval(ckpt, f"/tmp/{ckpt}_cells.json")
+            except subprocess.CalledProcessError as exc:
+                # Transient eval failure (client reconnect flake, sandbox hiccup):
+                # skip it — the next poll recomputes pending and retries it.
+                print(f"[watch] {ckpt} eval failed ({exc}); will retry next poll", flush=True)
+                continue
             upload(f"/tmp/{ckpt}_cells.json", f"eval/{ckpt}_results.json")
             p1, n = pass_at_1(f"/tmp/{ckpt}_cells.json")
             pb, nb = pass_at_1("/tmp/base_cells.json")