first pass

Author: xzrderek

eval_protocol/pytest/default_single_turn_rollout_process.py

@@ -20,19 +20,6 @@ class SingleTurnRolloutProcessor(RolloutProcessor):
 
     def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
         """Generate single turn rollout tasks and return them for external handling."""
-
-        # Quiet LiteLLM logs in test runs unless user overrode
-        try:
-            if os.environ.get("LITELLM_LOG") is None:
-                os.environ["LITELLM_LOG"] = "ERROR"
-            _llog = logging.getLogger("LiteLLM")
-            _llog.setLevel(logging.CRITICAL)
-            _llog.propagate = False
-            for _h in list(_llog.handlers):
-                _llog.removeHandler(_h)
-        except Exception:
-            pass
-
         # Do not modify global LiteLLM cache. Disable caching per-request instead.
 
         async def process_row(row: EvaluationRow) -> EvaluationRow:

eval_protocol/pytest/evaluation_test.py

@@ -52,6 +52,7 @@
     generate_parameter_combinations,
     log_eval_status_and_rows,
     parse_ep_max_rows,
+    parse_ep_max_concurrent_rollouts,
     parse_ep_num_runs,
     rollout_processor_with_retry,
     sanitize_filename,
@@ -482,6 +483,7 @@ async def wrapper_body(**kwargs):
 
                 # Apply environment override for num_runs if present
                 effective_num_runs = parse_ep_num_runs(num_runs)
+                effective_max_concurrent_rollouts = parse_ep_max_concurrent_rollouts(max_concurrent_rollouts)
                 all_results: List[List[EvaluationRow]] = [[] for _ in range(effective_num_runs)]
 
                 experiment_id = generate_id()
@@ -587,7 +589,7 @@ def _log_eval_error(
                     config = RolloutProcessorConfig(
                         completion_params=completion_params,
                         mcp_config_path=mcp_config_path or "",
-                        max_concurrent_rollouts=max_concurrent_rollouts,
+                        max_concurrent_rollouts=effective_max_concurrent_rollouts,
                         server_script_path=server_script_path,
                         steps=steps,
                         logger=active_logger,
@@ -615,7 +617,7 @@ def _log_eval_error(
                             processed_rows_in_run.append(row)
 
                         # prepare parallel eval helper function
-                        semaphore = asyncio.Semaphore(max_concurrent_evaluations)
+                        semaphore = asyncio.Semaphore(effective_max_concurrent_rollouts)
 
                         async def _execute_eval_with_semaphore(**inner_kwargs):
                             async with semaphore:
@@ -663,7 +665,7 @@ async def _execute_eval_with_semaphore(**inner_kwargs):
                                 config = RolloutProcessorConfig(
                                     completion_params=cp,
                                     mcp_config_path=mcp_config_path or "",
-                                    max_concurrent_rollouts=max_concurrent_rollouts,
+                                    max_concurrent_rollouts=effective_max_concurrent_rollouts,
                                     server_script_path=server_script_path,
                                     steps=steps,
                                     logger=active_logger,
@@ -843,7 +845,7 @@ async def dual_mode_wrapper(*args, **kwargs):
             dual_mode_wrapper._origin_func = test_func
             dual_mode_wrapper._metainfo = {
                 "mode": mode,
-                "max_rollout_concurrency": max_concurrent_rollouts,
+                "max_rollout_concurrency": max_concurrent_rollouts,  # TODO: fix this
                 "max_evaluation_concurrency": max_concurrent_evaluations,
             }
 

eval_protocol/pytest/exception_config.py

@@ -109,8 +109,7 @@ def __post_init__(self):
         # Override backoff settings from environment variables
         if "EP_MAX_RETRY" in os.environ:
             max_retry = int(os.environ["EP_MAX_RETRY"])
-            if max_retry > 0:
-                self.backoff_config.max_tries = max_retry
+            self.backoff_config.max_tries = max_retry
 
         if "EP_FAIL_ON_MAX_RETRY" in os.environ:
             fail_on_max_retry = os.environ["EP_FAIL_ON_MAX_RETRY"].lower()

eval_protocol/pytest/plugin.py

@@ -34,6 +34,12 @@ def pytest_addoption(parser) -> None:
         default=None,
         help=("Override the number of runs for evaluation_test. Pass an integer (e.g., 1, 5, 10)."),
     )
+    group.addoption(
+        "--ep-max-concurrent-rollouts",
+        action="store",
+        default=None,
+        help=("Override the maximum number of concurrent rollouts. Pass an integer (e.g., 8, 50, 100)."),
+    )
     group.addoption(
         "--ep-print-summary",
         action="store_true",
@@ -62,14 +68,13 @@ def pytest_addoption(parser) -> None:
         default=None,
         help=(
             "Set reasoning.effort for providers that support it (e.g., Fireworks) via LiteLLM extra_body. "
-            "Values: low|medium|high"
+            "Values: low|medium|high|none"
         ),
     )
     group.addoption(
         "--ep-max-retry",
         action="store",
-        type=int,
-        default=0,
+        default=None,
         help=("Failed rollouts (with rollout_status.code indicating error) will be retried up to this many times."),
     )
     group.addoption(
@@ -98,7 +103,7 @@ def _normalize_max_rows(val: Optional[str]) -> Optional[str]:
         return None
 
 
-def _normalize_num_runs(val: Optional[str]) -> Optional[str]:
+def _normalize_number(val: Optional[str]) -> Optional[str]:
     if val is None:
         return None
     s = val.strip()
@@ -131,10 +136,15 @@ def pytest_configure(config) -> None:
         os.environ["EP_MAX_DATASET_ROWS"] = norm
 
     num_runs_val = config.getoption("--ep-num-runs")
-    norm_runs = _normalize_num_runs(num_runs_val)
+    norm_runs = _normalize_number(num_runs_val)
     if norm_runs is not None:
         os.environ["EP_NUM_RUNS"] = norm_runs
 
+    max_concurrent_val = config.getoption("--ep-max-concurrent-rollouts")
+    norm_concurrent = _normalize_number(max_concurrent_val)
+    if norm_concurrent is not None:
+        os.environ["EP_MAX_CONCURRENT_ROLLOUTS"] = norm_concurrent
+
     if config.getoption("--ep-print-summary"):
         os.environ["EP_PRINT_SUMMARY"] = "1"
 
@@ -143,10 +153,13 @@ def pytest_configure(config) -> None:
         os.environ["EP_SUMMARY_JSON"] = summary_json_path
 
     max_retry = config.getoption("--ep-max-retry")
-    os.environ["EP_MAX_RETRY"] = str(max_retry)
+    norm_max_retry = _normalize_number(max_retry)
+    if norm_max_retry is not None:
+        os.environ["EP_MAX_RETRY"] = norm_max_retry
 
     fail_on_max_retry = config.getoption("--ep-fail-on-max-retry")
-    os.environ["EP_FAIL_ON_MAX_RETRY"] = fail_on_max_retry
+    if fail_on_max_retry is not None:
+        os.environ["EP_FAIL_ON_MAX_RETRY"] = fail_on_max_retry
 
     # Allow ad-hoc overrides of input params via CLI flags
     try:
@@ -178,7 +191,8 @@ def pytest_configure(config) -> None:
         if reasoning_effort:
             # Always place under extra_body to avoid LiteLLM rejecting top-level params
             eb = merged.setdefault("extra_body", {})
-            eb["reasoning_effort"] = str(reasoning_effort)
+            # Convert "none" string to None value for API compatibility
+            eb["reasoning_effort"] = None if reasoning_effort.lower() == "none" else str(reasoning_effort)
         if merged:
             os.environ["EP_INPUT_PARAMS_JSON"] = _json.dumps(merged)
     except Exception:

eval_protocol/pytest/utils.py

@@ -18,6 +18,8 @@
 )
 from eval_protocol.pytest.exception_config import ExceptionHandlerConfig, get_default_exception_handler_config
 
+import logging
+
 
 def execute_function(func: Callable, **kwargs) -> Any:
     """
@@ -160,6 +162,15 @@ def parse_ep_num_runs(default_value: int) -> int:
     return int(raw) if raw is not None else default_value
 
 
+def parse_ep_max_concurrent_rollouts(default_value: int) -> int:
+    """Read EP_MAX_CONCURRENT_ROLLOUTS env override as int.
+
+    Assumes the environment variable was already validated by plugin.py.
+    """
+    raw = os.getenv("EP_MAX_CONCURRENT_ROLLOUTS")
+    return int(raw) if raw is not None else default_value
+
+
 def deep_update_dict(base: dict, override: dict) -> dict:
     """Recursively update nested dictionaries in-place and return base."""
     for key, value in override.items():
@@ -322,10 +333,14 @@ async def execute_row_with_backoff(task: asyncio.Task, row: EvaluationRow) -> Ev
                         return result
                     except Exception as retry_error:
                         # Backoff gave up
+                        logging.error(
+                            f"❌ Rollout failed, (retried {exception_config.backoff_config.max_tries} times): {repr(retry_error)}"
+                        )
                         row.rollout_status = Status.rollout_error(str(retry_error))
                         return row
                 else:
                     # Non-retryable exception - fail immediately
+                    logging.error(f"❌ Rollout failed (non-retryable error encountered): {repr(e)}")
                     row.rollout_status = Status.rollout_error(str(e))
                     return row