feat: tool_discovery solver — agent learns the API through tools

fle/env/gym_env/environment.py

@@ -1,4 +1,5 @@
 import logging
+import time
 
 import gymnasium as gym
 import numpy as np
@@ -27,6 +28,26 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _call_with_retry(fn, what: str, delays=(0, 2, 5)):
+    """Call fn(), retrying transient failures with short backoff.
+
+    A single transient failure (engine busy, RCON hiccup, malformed
+    response) must not kill a multi-hour rollout. The underlying error is
+    included in the raised exception so real failures stay diagnosable.
+    """
+    last_error = None
+    for attempt, delay in enumerate(delays, start=1):
+        if delay:
+            time.sleep(delay)
+        try:
+            return fn()
+        except Exception as e:
+            last_error = e
+            logger.warning(f"Error {what} (attempt {attempt}/{len(delays)}): {e}")
+    raise Exception(f"Error {what}: {last_error}") from last_error
+
+
 # need to do this since gym doesn't work with numpy>=2.0 otherwise.
 np.bool8 = np.dtype(np.bool)
 
@@ -304,12 +325,9 @@ def get_observation(
         if self.enable_vision:
             map_image = namespace._render().to_base64()
 
-        # Get entity observations
-        try:
-            entities = namespace.get_entities()
-        except Exception as e:
-            logger.warning(f"Error getting entities: {e}")
-            raise Exception("Error getting entities while getting observation") from e
+        entities = _call_with_retry(
+            namespace.get_entities, "getting entities while getting observation"
+        )
 
         entity_obs = [e.__dict__ for e in entities]
 
@@ -463,7 +481,9 @@ def step(
             )
             terminated = task_success.success
 
-        production_score, automated_production_score = namespace.score()
+        production_score, automated_production_score = _call_with_retry(
+            namespace.score, "getting score after step"
+        )
         if not automated_production_score:
             automated_production_score = 0
         # Calculate reward

fle/eval/inspect/integration/eval_set.py

@@ -32,6 +32,9 @@
     factorio_reasoning_only_solver,
     factorio_pruned_gamestate_solver,
 )
+from fle.eval.inspect.integration.solver_tool_discovery import (
+    factorio_tool_discovery_solver,
+)
 from fle.eval.inspect.integration.scorers import (
     throughput_proportion_scorer,
     production_score_tracker,
@@ -57,6 +60,7 @@
     "balanced": factorio_balanced_solver,
     "reasoning_only": factorio_reasoning_only_solver,
     "pruned_gamestate": factorio_pruned_gamestate_solver,
+    "tool_discovery": factorio_tool_discovery_solver,
 }
 
 

fle/eval/inspect/integration/simple_server_pool.py

@@ -72,6 +72,35 @@ def _load_api_keys(self):
                     "No Anthropic API keys found. Set ANTHROPIC_API_KEYS or ANTHROPIC_API_KEY"
                 )
 
+    @staticmethod
+    def _server_reachable(idx: int, timeout: float = 1.0) -> bool:
+        """Check whether the Factorio container for run_idx is reachable."""
+        import socket
+
+        try:
+            with socket.create_connection(("localhost", 27000 + idx), timeout=timeout):
+                return True
+        except OSError:
+            return False
+
+    def _probe_reachable_indices(self) -> List[int]:
+        """Return the run_idx values in range whose servers actually respond.
+
+        Previously every index in range was assumed to exist; allocating a
+        run_idx with no container behind it made that sample fail deep inside
+        make_factorio_env, and Inspect recorded it as an errored sample while
+        the eval as a whole "completed" — silently degrading results (e.g.
+        1/8 real rollouts with a single container).
+        """
+        # External server override: make_factorio_env ignores run_idx, so any
+        # index is valid and there is nothing to probe on localhost.
+        if os.getenv("FACTORIO_SERVER_ADDRESS") or os.getenv("FACTORIO_SERVER_PORT"):
+            return list(range(self.start_idx, self.end_idx))
+
+        return [
+            i for i in range(self.start_idx, self.end_idx) if self._server_reachable(i)
+        ]
+
     async def initialize(self):
         """Initialize the pool with available run_idx values"""
         if self._initialized:
@@ -80,12 +109,29 @@ async def initialize(self):
         # Load API keys
         self._load_api_keys()
 
-        # Populate queue with available run_idx values in the configured range
-        for i in range(self.start_idx, self.end_idx):
+        # Only hand out indices whose servers are actually reachable, so
+        # rollouts queue for real containers instead of crashing on ghosts.
+        reachable = self._probe_reachable_indices()
+        if not reachable:
+            raise RuntimeError(
+                f"No reachable Factorio servers in configured range "
+                f"{self.start_idx}-{self.end_idx - 1} (probed localhost tcp "
+                f"{27000 + self.start_idx}-{27000 + self.end_idx - 1}). "
+                f"Start containers with 'fle cluster start -n N' and retry."
+            )
+        if len(reachable) < self.max_servers:
+            logger.warning(
+                f"Only {len(reachable)}/{self.max_servers} Factorio servers in range "
+                f"{self.start_idx}-{self.end_idx - 1} are reachable: {reachable}. "
+                f"Rollouts will share the reachable servers."
+            )
+        self.max_servers = len(reachable)
+
+        for i in reachable:
             await self.available_indices.put(i)
 
         logger.info(
-            f"Initialized SimpleServerPool with servers {self.start_idx}-{self.end_idx - 1} "
+            f"Initialized SimpleServerPool with servers {reachable} "
             f"({self.max_servers} total) and {len(self._api_keys)} API keys"
         )
         self._initialized = True
@@ -111,16 +157,36 @@ async def get_run_idx(self) -> int:
         """
         await self.initialize()
 
+        run_idx = await self._acquire_index()
+        logger.info(f"Allocated run_idx {run_idx} (server factorio_{run_idx})")
+        return run_idx
+
+    # How long a rollout may wait for a free server before the sample fails.
+    ACQUIRE_TIMEOUT_SECONDS = 1800
+
+    async def _acquire_index(self) -> int:
+        """Take a run_idx, waiting for one to free up if all are in use.
+
+        Waiting (rather than raising immediately) lets rollouts share a
+        smaller server fleet — e.g. 8 epochs on 1 container run sequentially
+        instead of 7 of them erroring out.
+        """
         if self.available_indices.empty():
+            logger.warning(
+                f"All {self.max_servers} reachable servers in use; waiting for a "
+                f"free one (start more with 'fle cluster start -n N' to parallelize)"
+            )
+        try:
+            run_idx = await asyncio.wait_for(
+                self.available_indices.get(), timeout=self.ACQUIRE_TIMEOUT_SECONDS
+            )
+        except asyncio.TimeoutError:
             raise RuntimeError(
-                f"All {self.max_servers} servers are in use. "
-                f"Consider reducing --max-connections or starting more containers with 'fle cluster start -n N'"
+                f"Timed out after {self.ACQUIRE_TIMEOUT_SECONDS}s waiting for a free "
+                f"Factorio server ({self.max_servers} reachable, all busy). "
+                f"Start more containers with 'fle cluster start -n N' or reduce --max-connections."
             )
-
-        run_idx = await self.available_indices.get()
         self.allocated_indices.add(run_idx)
-
-        logger.info(f"Allocated run_idx {run_idx} (server factorio_{run_idx})")
         return run_idx
 
     async def get_server_allocation(self) -> ServerAllocation:
@@ -136,14 +202,7 @@ async def get_server_allocation(self) -> ServerAllocation:
         """
         await self.initialize()
 
-        if self.available_indices.empty():
-            raise RuntimeError(
-                f"All {self.max_servers} servers are in use. "
-                f"Consider reducing --max-connections or starting more containers with 'fle cluster start -n N'"
-            )
-
-        run_idx = await self.available_indices.get()
-        self.allocated_indices.add(run_idx)
+        run_idx = await self._acquire_index()
 
         # Get next API key
         api_key, key_index = await self._get_next_api_key()