almost finished

Author: xzrderek

eval_protocol/pytest/default_single_turn_rollout_process.py

@@ -35,11 +35,15 @@ async def process_row(row: EvaluationRow) -> EvaluationRow:
             # Single-level reasoning effort: expect `reasoning_effort` only
             effort_val = None
 
-            if "reasoning_effort" in config.completion_params:
+            if (
+                "reasoning_effort" in config.completion_params
+                and config.completion_params["reasoning_effort"] is not None
+            ):
                 effort_val = str(config.completion_params["reasoning_effort"])  # flat shape
             elif (
                 isinstance(config.completion_params.get("extra_body"), dict)
                 and "reasoning_effort" in config.completion_params["extra_body"]
+                and config.completion_params["extra_body"]["reasoning_effort"] is not None
             ):
                 # Accept if user passed it directly inside extra_body
                 effort_val = str(config.completion_params["extra_body"]["reasoning_effort"])  # already in extra_body

eval_protocol/pytest/evaluation_test.py

@@ -47,7 +47,6 @@
     aggregate,
     create_dynamically_parameterized_wrapper,
     deep_update_dict,
-    execute_function,
     extract_effort_tag,
     generate_parameter_combinations,
     log_eval_status_and_rows,
@@ -333,6 +332,11 @@ def evaluation_test(  # noqa: C901
 
     active_logger: DatasetLogger = logger if logger else default_logger
 
+    # Apply override from pytest flags if present
+    num_runs = parse_ep_num_runs(num_runs)
+    max_concurrent_rollouts = parse_ep_max_concurrent_rollouts(max_concurrent_rollouts)
+    max_dataset_rows = parse_ep_max_rows(max_dataset_rows)
+
     def decorator(
         test_func: TestFunction,
     ):
@@ -481,10 +485,7 @@ def create_wrapper_with_signature() -> Callable:
             async def wrapper_body(**kwargs):
                 eval_metadata = None
 
-                # 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)]
+                all_results: List[List[EvaluationRow]] = [[] for _ in range(num_runs)]
 
                 experiment_id = generate_id()
 
@@ -508,10 +509,9 @@ def _log_eval_error(
                                 data_jsonl.extend(load_jsonl(p))
                         else:
                             data_jsonl = load_jsonl(ds_arg)
-                        # Apply env override for max rows if present
-                        effective_max_rows = parse_ep_max_rows(max_dataset_rows)
-                        if effective_max_rows is not None:
-                            data_jsonl = data_jsonl[:effective_max_rows]
+                        # Apply override for max rows if present
+                        if max_dataset_rows is not None:
+                            data_jsonl = data_jsonl[:max_dataset_rows]
                         data = dataset_adapter(data_jsonl)
                     elif "input_messages" in kwargs and kwargs["input_messages"] is not None:
                         # Support either a single row (List[Message]) or many rows (List[List[Message]])
@@ -563,7 +563,7 @@ def _log_eval_error(
                         name=test_func.__name__,
                         description=test_func.__doc__,
                         status="running",
-                        num_runs=effective_num_runs,
+                        num_runs=num_runs,
                         aggregation_method=aggregation_method,
                         passed_threshold=threshold,
                         passed=None,
@@ -589,15 +589,15 @@ def _log_eval_error(
                     config = RolloutProcessorConfig(
                         completion_params=completion_params,
                         mcp_config_path=mcp_config_path or "",
-                        max_concurrent_rollouts=effective_max_concurrent_rollouts,
+                        max_concurrent_rollouts=max_concurrent_rollouts,
                         server_script_path=server_script_path,
                         steps=steps,
                         logger=active_logger,
                         kwargs=rollout_processor_kwargs or {},
                         exception_handler_config=exception_handler_config,
                     )
 
-                    for i in range(effective_num_runs):
+                    for i in range(num_runs):
                         # Regenerate outputs each run by deep-copying the pristine dataset
                         # so model responses are not reused across runs.
                         run_id = generate_id()
@@ -617,7 +617,7 @@ def _log_eval_error(
                             processed_rows_in_run.append(row)
 
                         # prepare parallel eval helper function
-                        semaphore = asyncio.Semaphore(effective_max_concurrent_rollouts)
+                        semaphore = asyncio.Semaphore(max_concurrent_rollouts)
 
                         async def _execute_eval_with_semaphore(**inner_kwargs):
                             async with semaphore:
@@ -665,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=effective_max_concurrent_rollouts,
+                                    max_concurrent_rollouts=max_concurrent_rollouts,
                                     server_script_path=server_script_path,
                                     steps=steps,
                                     logger=active_logger,
@@ -739,8 +739,7 @@ async def _collect_result(config, lst):
                     # rollout_id is used to differentiate the result from different completion_params
                     if mode == "groupwise":
                         results_by_group = [
-                            [[] for _ in range(effective_num_runs)]
-                            for _ in range(len(original_completion_params_list))
+                            [[] for _ in range(num_runs)] for _ in range(len(original_completion_params_list))
                         ]
                         for i_run, result in enumerate(all_results):
                             for r in result:
@@ -755,7 +754,7 @@ async def _collect_result(config, lst):
                                 mode,
                                 original_completion_params_list[rollout_id],
                                 test_func.__name__,
-                                effective_num_runs,
+                                num_runs,
                             )
                     else:
                         postprocess(
@@ -766,7 +765,7 @@ async def _collect_result(config, lst):
                             mode,
                             completion_params,
                             test_func.__name__,
-                            effective_num_runs,
+                            num_runs,
                         )
 
                 except AssertionError:
@@ -845,7 +844,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,  # TODO: fix this
+                "max_rollout_concurrency": max_concurrent_rollouts,
                 "max_evaluation_concurrency": max_concurrent_evaluations,
             }
 

eval_protocol/pytest/plugin.py

@@ -83,6 +83,7 @@ def pytest_addoption(parser) -> None:
         default="true",
         choices=["true", "false"],
         help=(
+            # TODO: this is not working as expected
             "Whether to fail the entire rollout when permanent failures occur after max retries. "
             "Default: true (fail on permanent failures). Set to 'false' to continue with remaining rollouts."
         ),