attempted fix

Author: xzrderek

eval_protocol/pytest/evaluation_test.py

@@ -449,6 +449,8 @@ def _log_eval_error(status: Status, rows: list[EvaluationRow] | None, passed: bo
                         finally:
                             if output_buffer:
                                 await output_buffer.close()
+                            await rollout_processor.acleanup()
+                            rollout_processor.cleanup()
 
                         for res in priority_results:
                             run_idx = (res.execution_metadata.extra or {}).get("run_index", 0)
@@ -697,15 +699,19 @@ async def _collect_result(config, lst):
                         # Lazy import (cached after first import above)
                         from eval_protocol.pytest.default_mcp_gym_rollout_processor import MCPGymRolloutProcessor
 
-                        if isinstance(rollout_processor, MCPGymRolloutProcessor):
-                            # For MCPGymRolloutProcessor, create and execute tasks one at a time to avoid port conflicts
-                            for run_idx in range(num_runs):
-                                task = asyncio.create_task(execute_run(run_idx, config))
-                                await task
-                        else:
-                            # For other processors, create all tasks at once and run in parallel
-                            # Concurrency is now controlled by the shared semaphore in each rollout processor
-                            await run_tasks_with_run_progress(execute_run, num_runs, config)
+                        try:
+                            if isinstance(rollout_processor, MCPGymRolloutProcessor):
+                                # For MCPGymRolloutProcessor, create and execute tasks one at a time to avoid port conflicts
+                                for run_idx in range(num_runs):
+                                    task = asyncio.create_task(execute_run(run_idx, config))
+                                    await task
+                            else:
+                                # For other processors, create all tasks at once and run in parallel
+                                # Concurrency is now controlled by the shared semaphore in each rollout processor
+                                await run_tasks_with_run_progress(execute_run, num_runs, config)
+                        finally:
+                            await rollout_processor.acleanup()
+                            rollout_processor.cleanup()
 
                         experiment_duration_seconds = time.perf_counter() - experiment_start_time
 

eval_protocol/pytest/evaluation_test_utils.py

@@ -476,8 +476,12 @@ async def execute_row_with_backoff_and_log(
                 yield result
 
     finally:
-        await rollout_processor.acleanup()
-        rollout_processor.cleanup()
+        # Cleanup is intentionally NOT called here. rollout_processor_with_retry
+        # is invoked per-run, but the processor (and its session) is shared
+        # across parallel runs.  Closing per-run would kill in-flight requests
+        # in other runs.  Cleanup is called once after all runs complete in
+        # evaluation_test.py.
+        pass
 
 
 def sanitize_filename(text: str) -> str:

eval_protocol/pytest/remote_rollout_processor.py

@@ -49,15 +49,13 @@ def __init__(
         self._timeout_seconds = timeout_seconds
         self._tracing_adapter = FireworksTracingAdapter(base_url=self._model_base_url)
         self._session: Optional[aiohttp.ClientSession] = None
-        self._active_runs = 0
 
     def _get_or_create_session(self) -> aiohttp.ClientSession:
         if self._session is None or self._session.closed:
             self._session = aiohttp.ClientSession()
         return self._session
 
     def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
-        self._active_runs += 1
         tasks: List[asyncio.Task[EvaluationRow]] = []
 
         # Start with constructor values
@@ -208,26 +206,17 @@ async def _sem_wrapper(r: EvaluationRow) -> EvaluationRow:
         tasks = [asyncio.create_task(_sem_wrapper(row)) for row in rows]
         return tasks
 
-    def _should_close_session(self) -> bool:
-        self._active_runs = max(0, self._active_runs - 1)
-        return self._active_runs == 0 and self._session is not None and not self._session.closed
-
     async def acleanup(self) -> None:
-        """Async cleanup — only closes the session when the last run finishes.
-
-        rollout_processor_with_retry calls acleanup() per-run, but the session
-        is shared across parallel runs.  Closing it early would cancel in-flight
-        requests in other runs.
-        """
-        if self._should_close_session():
-            await self._session.close()  # type: ignore[union-attr]
+        """Async cleanup - preferred when you can await."""
+        if self._session and not self._session.closed:
+            await self._session.close()
 
     def cleanup(self) -> None:
-        """Sync cleanup — best-effort fallback when not in an async context."""
-        if self._should_close_session():
+        """Sync cleanup - best-effort, schedules close if event loop is running."""
+        if self._session and not self._session.closed:
             try:
                 loop = asyncio.get_running_loop()
-                loop.create_task(self._session.close())  # type: ignore[union-attr]
+                loop.create_task(self._session.close())
             except RuntimeError:
                 logger.warning(
                     "RemoteRolloutProcessor.cleanup() called outside of async context. "