feat(eval): add parallel task execution support via max_workers in eval_set

CHANGELOG.md

@@ -7,6 +7,10 @@ All notable changes to this project are documented here. The format is based on
 
 ## [Unreleased]
 
+### Added
+
+- **Core & CLI:** added `max_workers` parameter to `eval_set()` and `--max-workers` flag to `inspect-robots eval-set` for concurrent task execution.
+
 ### Fixed
 
 - **Voice plugin (0.5.1):** operator-ended trials now cut `--speak` narration

src/inspect_robots/cli.py

@@ -387,6 +387,12 @@ def build_parser() -> argparse.ArgumentParser:
         help="passed through to eval_set(); resumption of a partial run is "
         "accepted but not yet honored",
     )
+    p_eval_set.add_argument(
+        "--max-workers",
+        type=int,
+        default=1,
+        help="number of parallel task worker threads for task execution (default: 1)",
+    )
 
     p_inspect = sub.add_parser("inspect", help="print a saved eval log")
     p_inspect.add_argument("log", help="path to an EvalLog JSON file")
@@ -1851,6 +1857,7 @@ def _cmd_eval_set(args: argparse.Namespace) -> int:
                 retry_attempts=args.retry_attempts,
                 operator_input=operator_input,
                 grader=grader,
+                max_workers=args.max_workers,
             )
         except KeyboardInterrupt:
             # eval_set writes one log per task; eval() persists a cancelled log

src/inspect_robots/eval.py

@@ -650,43 +650,59 @@ def eval_set(
     before_scoring: Callable[[TrialRecord, Scene], None] | None = None,
     grader: Grader | str | None = None,
     retry_attempts: int = 0,
+    max_workers: int = 1,
 ) -> tuple[bool, list[EvalLog]]:
     """Run a set of tasks and return ``(success, logs)`` (mirrors Inspect AI).
 
     ``success`` is ``True`` iff every task's log has ``status == "success"``.
 
+    ``max_workers`` controls parallel task execution. When ``max_workers > 1``,
+    tasks are executed concurrently using a thread pool worker execution queue.
+
     ``grader``/``before_scoring`` follow ``eval()``'s contract (one pre-scoring
     hook, not both) and are resolved once here, so every task shares the same
     grader instance.
 
-    Caller-supplied ``sinks`` are reused across the set's sequential runs. Each
+    Caller-supplied ``sinks`` are reused across the set's runs. Each
     sink must reset its per-run state in ``on_eval_start`` and tolerate one
     complete lifecycle per task.
 
     Resumption of a partially-completed run (skipping already-finished scenes via
     a stable run id) is reserved for a follow-up: ``retry_attempts`` is accepted
     now so callers don't get retrofitted, but is not yet honored.
     """
+    if max_workers < 1:
+        raise ConfigError("max_workers must be >= 1")
     before_scoring = _grading_hook(grader, before_scoring)
     task_list = [tasks] if isinstance(tasks, Task | str) else list(tasks)
     logs: list[EvalLog] = []
-    for task in task_list:
-        logs.extend(
-            eval(
-                task,
-                policy,
-                embodiment,
-                log_dir=log_dir,
-                sinks=sinks,
-                seed=seed,
-                fail_on_error=fail_on_error,
-                controller=controller,
-                approver=approver,
-                remap=remap,
-                store_frames=store_frames,
-                operator_input=operator_input,
-                before_scoring=before_scoring,
-            )
+
+    def _eval_one(target_task: Task | str) -> list[EvalLog]:
+        return eval(
+            target_task,
+            policy,
+            embodiment,
+            log_dir=log_dir,
+            sinks=sinks,
+            seed=seed,
+            fail_on_error=fail_on_error,
+            controller=controller,
+            approver=approver,
+            remap=remap,
+            store_frames=store_frames,
+            operator_input=operator_input,
+            before_scoring=before_scoring,
         )
+
+    if max_workers > 1 and len(task_list) > 1:
+        from concurrent.futures import ThreadPoolExecutor
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            for task_logs in executor.map(_eval_one, task_list):
+                logs.extend(task_logs)
+    else:
+        for task in task_list:
+            logs.extend(_eval_one(task))
+
     success = all(log.status == "success" for log in logs)
     return success, logs

tests/test_eval_orchestration.py

@@ -966,6 +966,38 @@ def judge(record: TrialRecord, scene: Scene) -> None:
     assert logs[0].results.metrics["operator"] == 1.0
 
 
+def test_eval_set_max_workers_validation(tmp_path: Path) -> None:
+    from inspect_robots.errors import ConfigError
+
+    with pytest.raises(ConfigError, match="max_workers must be >= 1"):
+        eval_set(
+            _task(),
+            ScriptedPolicy(),
+            CubePickEmbodiment(),
+            log_dir=str(tmp_path),
+            max_workers=0,
+        )
+
+
+def test_eval_set_parallel_execution(tmp_path: Path) -> None:
+    task1 = _task(scorer=operator_scorer())
+    task2 = _task(scorer=operator_scorer())
+
+    def judge(record: TrialRecord, scene: Scene) -> None:
+        record.operator_judgement = "pass"
+
+    success, logs = eval_set(
+        [task1, task2],
+        ScriptedPolicy(),
+        CubePickEmbodiment(),
+        log_dir=str(tmp_path),
+        before_scoring=judge,
+        max_workers=2,
+    )
+    assert success
+    assert len(logs) == 2
+
+
 # --------------------------------------------------------------------------- #
 # 9. Attended operator input is trial-scoped and persisted beside each epoch.
 # --------------------------------------------------------------------------- #