feat: Add support for eval case level timeouts

datasets/gemini-cli-tools/example_run_timeout_config.yaml

@@ -0,0 +1,30 @@
+############################################################
+### Dataset / Eval Items
+############################################################
+dataset_config: datasets/gemini-cli-tools/gemini-cli-fake.evalset.json
+dataset_format: gemini-cli-format
+
+# Orchestrator Configuration
+orchestrator: geminicli
+model_config: datasets/model_configs/gemini_cli_fake_model.yaml
+simulated_user_model_config: datasets/model_configs/gemini_2.5_pro_model.yaml
+
+# Per-scenario evaluation timeout duration (e.g., '300s', '5m', '1h30m')
+eval_case_timeout: 5m
+
+############################################################
+### Scorer Related Configs
+############################################################
+scorers:
+  trajectory_matcher: {}
+  turn_count: {}
+  end_to_end_latency: {}
+  tool_call_latency: {}
+  token_consumption: {}
+
+############################################################
+### Reporting Related Configs
+############################################################
+reporting:
+  csv:
+    output_directory: 'results'

docs/configs/run-config.md

@@ -19,6 +19,7 @@ This section defines the primary resources used during evaluation, including the
 | `num_trials`      | Optional     | Number of trials to run for each prompt. |
 | `scenarios`      | Optional     | A list of specific scenario IDs to run (only applies to scenario-based agentic datasets like `gemini-cli-format` or `cortado-format`). Defaults to empty (runs all scenarios). |
 | `scenario_pattern` | Optional     | A glob pattern of scenario IDs to run (only applies to scenario-based agentic datasets). Defaults to None (runs all scenarios). |
+| `eval_case_timeout` | Optional     | Maximum execution duration allowed per individual scenario (e.g. `'300s'`, `'5m'`, `'1h30m'`). Applicable to scenario-based agent evaluations. |
 ---
 
 ## 2. Prompt and Generation Modules

evalbench/evaluator/agentevaluator.py

@@ -4,12 +4,14 @@
 import logging
 import os
 import shutil
+import time
 import threading
 
 from dataset.evalgeminicliinput import EvalGeminiCliRequest
 from generators.models import get_generator
 from generators.models.agent_cli import AgentCliGenerator
 from mp import mprunner
+from util.config import get_eval_case_timeout
 from work.agentgenwork import AgentGenWork
 from evaluator.simulateduser import SimulatedUser
 from work.agentscorework import AgentScoreWork
@@ -24,6 +26,7 @@ def __init__(
         config,
     ):
         self.config = config
+        self.eval_case_timeout_seconds = get_eval_case_timeout(config)
 
         model_config_path = config.get("model_config")
         if not isinstance(model_config_path, str):
@@ -142,7 +145,46 @@ def process_scenario(
                     )
 
         session_id = None
+
+        # Dynamically calculate the effective timeout, taking into account
+        # both the scenario timeout and the global timeout.
+        start_time = time.monotonic()
+        scenario_timeout = get_eval_case_timeout(scenario)
+        effective_case_timeout_seconds = (
+            scenario_timeout
+            if scenario_timeout is not None
+            else self.eval_case_timeout_seconds
+        )
+        last_result = None
+
         for turn in range(max_turns):
+            remaining_timeout_seconds = None
+            if effective_case_timeout_seconds is not None:
+                elapsed_seconds = time.monotonic() - start_time
+                remaining_timeout_seconds = effective_case_timeout_seconds - elapsed_seconds
+                if remaining_timeout_seconds <= 0:
+                    logging.warning(
+                        f"Eval case timeout ({effective_case_timeout_seconds}s) reached before turn {turn + 1}."
+                    )
+                    timeout_msg = f"TimeoutError: Scenario timed out after {effective_case_timeout_seconds}s before turn {turn + 1}"
+                    if last_result is None:
+                        last_result = subprocess.CompletedProcess(
+                            args=[self.agent_version],
+                            returncode=124,
+                            stdout="",
+                            stderr=timeout_msg,
+                        )
+                    else:
+                        current_stderr = getattr(last_result, "stderr", "") or ""
+                        new_stderr = f"{current_stderr}\n{timeout_msg}".strip()
+                        last_result = subprocess.CompletedProcess(
+                            args=getattr(last_result, "args", [self.agent_version]),
+                            returncode=124,
+                            stdout=getattr(last_result, "stdout", "") or "",
+                            stderr=new_stderr,
+                        )
+                    break
+
             logging.info(
                 f"Turn {turn + 1}/{max_turns} - Prompt: {current_prompt}")
             if isinstance(self.generator, AgentCliGenerator):
@@ -155,7 +197,9 @@ def process_scenario(
                     cwd=resolved_work_dir,
                 )
                 try:
-                    result = self.generator.safe_generate(cli_cmd)
+                    result = self.generator.safe_generate(
+                        cli_cmd, timeout_seconds=remaining_timeout_seconds
+                    )
                     if result.stdout:
                         parsed = self.generator.parse_response(result.stdout)
                         if parsed.get("session_id"):
@@ -205,18 +249,22 @@ def process_scenario(
                 else:
                     break
 
-        if last_result:
-            self._finalize_scenario(
-                scenario,
-                last_result,
-                conversation_history,
-                accumulated_tools,
-                accumulated_skills,
-                eval_result,
-                job_id,
-                metadata
+        if last_result is None:
+            last_result = subprocess.CompletedProcess(
+                args=[self.agent_version], returncode=0, stdout="", stderr=""
             )
 
+        self._finalize_scenario(
+            scenario,
+            last_result,
+            conversation_history,
+            accumulated_tools,
+            accumulated_skills,
+            eval_result,
+            job_id,
+            metadata
+        )
+
     def _log_cli_result(self, turn: int, max_turns: int, result: subprocess.CompletedProcess):
         generator_name = self.generator.name
         logging.info(
@@ -238,12 +286,18 @@ def _finalize_scenario(
         metadata: Dict[str, Any]
     ):
         """Finalizes the scenario by scoring and appending results."""
+        timed_out = (
+            getattr(last_result, "returncode", 0) == 124
+            or "TimeoutError" in (getattr(last_result, "stderr", "") or "")
+        )
+
         # Prepare intermediate eval_output with all necessary data for scoring
         eval_output_data = {
             "eval_id": scenario["id"],
-            "stdout": last_result.stdout,
-            "stderr": last_result.stderr,
-            "returncode": last_result.returncode,
+            "stdout": getattr(last_result, "stdout", "") or "",
+            "stderr": getattr(last_result, "stderr", "") or "",
+            "returncode": getattr(last_result, "returncode", 0),
+            "timed_out": timed_out,
             "prompt_generator_error": None,
             "generated_error": None,
             "sql_generator_error": None,

evalbench/evaluator/evaluator.py

@@ -72,7 +72,8 @@ def __init__(
         self.sqlexec_runners = runner_config.get("sqlexec_runners", 10)
         self.scoring_runners = runner_config.get("scoring_runners", 10)
         self.task_timeout_seconds = runner_config.get(
-            "task_timeout_seconds", 600)
+            "task_timeout_seconds", 600
+        )
         self.num_trials = self.config.get("num_trials", 1)
 
     def evaluate(

evalbench/generators/models/agent_cli.py

@@ -1,4 +1,6 @@
 from abc import abstractmethod
+import subprocess
+from typing import Optional
 
 from mcp import types as mcp_types
 
@@ -96,7 +98,9 @@ def create_command(
         raise NotImplementedError("Subclasses must implement this method")
 
     @abstractmethod
-    def safe_generate(self, cli_cmd):
+    def safe_generate(
+        self, cli_cmd, timeout_seconds: Optional[float] = None
+    ) -> subprocess.CompletedProcess:
         raise NotImplementedError("Subclasses must implement this method")
 
     @abstractmethod

evalbench/generators/models/agy_cli.py

@@ -9,7 +9,9 @@
 import sys
 import tempfile
 import weakref
+from typing import Optional, Union, Dict, List
 from util.context import rpc_id_var
+from util.config import parse_timeout_seconds
 
 # Default CLI label reported in metadata. The executed binary is installed
 # per-session at self.agy_bin (see _ensure_agy_installed).
@@ -107,7 +109,7 @@ def _validate_timeout(timeout):
                 )
             # Strict regex for common units (s, m, h).
             # Allows things like "20m", "1h30m", "300s".
-            if not re.match(r'^(\d+(s|m|h))+$', timeout):
+            if not re.match(r"^(\d+(s|m|h))+$", timeout):
                 raise ValueError(
                     f"Invalid timeout format: '{timeout}'. "
                     "Must be a valid duration string (e.g., '20m', '1h30m', '300s')."
@@ -866,13 +868,13 @@ def _base_agy_command(
             command.append("--continue")
         return command
 
-    def generate_internal(self, cli_cmd):
+    def generate_internal(self, cli_cmd, timeout_seconds=None):
         if not isinstance(cli_cmd, CLICommand):
             cli_cmd = CLICommand(self.agy_bin, str(cli_cmd))
-        return self._run_agy_cli(cli_cmd)
+        return self._run_agy_cli(cli_cmd, timeout_seconds=timeout_seconds)
 
     def _execute_cli_command(
-        self, command, env=None, cwd=None
+        self, command, env=None, cwd=None, timeout_seconds=None
     ) -> subprocess.CompletedProcess:
         try:
             return subprocess.run(
@@ -882,6 +884,16 @@ def _execute_cli_command(
                 check=False,
                 env=env,
                 cwd=cwd if cwd else self.fake_home,
+                timeout=timeout_seconds,
+            )
+        except subprocess.TimeoutExpired as e:
+            stdout_str = e.stdout if isinstance(e.stdout, str) else (e.stdout.decode() if e.stdout else "")
+            stderr_str = f"TimeoutError: Command timed out after {timeout_seconds} seconds"
+            if e.stderr:
+                err_text = e.stderr if isinstance(e.stderr, str) else e.stderr.decode()
+                stderr_str = f"{stderr_str}\n{err_text}"
+            return subprocess.CompletedProcess(
+                command, 124, stdout_str, stderr_str
             )
         except FileNotFoundError:
             return subprocess.CompletedProcess(
@@ -893,7 +905,7 @@ def _execute_cli_command(
                 command, 1, "", f"An unexpected error occurred: {e}"
             )
 
-    def _run_agy_cli(self, cli_cmd: CLICommand):
+    def _run_agy_cli(self, cli_cmd: CLICommand, timeout_seconds=None):
         env = self._merged_env(cli_cmd.env)
         # The executable is always this session's sandbox binary, regardless of
         # the label carried on cli_cmd.cli (the evaluator passes agent_version,
@@ -904,7 +916,7 @@ def _run_agy_cli(self, cli_cmd: CLICommand):
             timeout=self.timeout,
         )
         cwd = cli_cmd.cwd if cli_cmd.cwd else self.fake_home
-        result = self._execute_cli_command(command, env=env, cwd=cwd)
+        result = self._execute_cli_command(command, env=env, cwd=cwd, timeout_seconds=timeout_seconds)
 
         # Parse whenever agy emitted a stream, even on a non-zero exit: a
         # timed-out/errored run still ends in a ``result`` event carrying real
@@ -1198,9 +1210,9 @@ def extract_skills(self, stdout: str) -> list:
             return []
 
     def safe_generate(
-        self, cli_cmd: CLICommand
+        self, cli_cmd: CLICommand, timeout_seconds: Optional[float] = None
     ) -> subprocess.CompletedProcess:
-        result = self.generate_internal(cli_cmd)
+        result = self.generate_internal(cli_cmd, timeout_seconds=timeout_seconds)
         if isinstance(result, str):
             return subprocess.CompletedProcess(
                 args=[], returncode=0, stdout=result

evalbench/generators/models/claude_code.py

@@ -9,6 +9,7 @@
 import re
 import shutil
 import time
+from typing import Optional, Union, Dict, List
 import uuid
 from util.context import rpc_id_var
 
@@ -599,21 +600,31 @@ def _install_plugin(self, plugin_id: str, env: dict | None = None):
         else:
             logging.info(f"Successfully installed plugin '{plugin_id}'")
 
-    def generate_internal(self, cli_cmd):
+    def generate_internal(self, cli_cmd, timeout_seconds=None):
         if not isinstance(cli_cmd, CLICommand):
             cli_cmd = CLICommand(self.claude_code_version, str(cli_cmd))
-        return self._run_claude_code(cli_cmd)
+        return self._run_claude_code(cli_cmd, timeout_seconds=timeout_seconds)
 
     def _execute_cli_command(
         self, command: list[str], env: dict[str, str] | None = None,
-        cwd: str | None = None,
+        cwd: str | None = None, timeout_seconds: float | int | None = None,
     ) -> subprocess.CompletedProcess:
         try:
             result = subprocess.run(
                 command, capture_output=True, text=True, check=False, env=env,
-                cwd=cwd if cwd else self.fake_home, stdin=subprocess.DEVNULL
+                cwd=cwd if cwd else self.fake_home, stdin=subprocess.DEVNULL,
+                timeout=timeout_seconds,
             )
             return result
+        except subprocess.TimeoutExpired as e:
+            stdout_str = e.stdout if isinstance(e.stdout, str) else (e.stdout.decode() if e.stdout else "")
+            stderr_str = f"TimeoutError: Command timed out after {timeout_seconds} seconds"
+            if e.stderr:
+                err_text = e.stderr if isinstance(e.stderr, str) else e.stderr.decode()
+                stderr_str = f"{stderr_str}\n{err_text}"
+            return subprocess.CompletedProcess(
+                command, 124, stdout_str, stderr_str
+            )
         except FileNotFoundError:
             return subprocess.CompletedProcess(
                 command, 127, "", f"Error: Command not found: {command[0]}"
@@ -633,7 +644,7 @@ def _session_id_headers() -> str:
         session_id = f"sess-{int(time.time())}-{uuid.uuid4().hex[:8]}"
         return f"X-Vertex-Ai-Session-Id: {session_id}"
 
-    def _run_claude_code(self, cli_cmd: CLICommand):
+    def _run_claude_code(self, cli_cmd: CLICommand, timeout_seconds=None):
         env = os.environ.copy()
         env.update(self.env)
         env.update(cli_cmd.env)
@@ -697,7 +708,7 @@ def _run_claude_code(self, cli_cmd: CLICommand):
 
         logging.info(f"Running Claude Code CLI: {' '.join(command)}")
 
-        result = self._execute_cli_command(command, env=env, cwd=cli_cmd.cwd)
+        result = self._execute_cli_command(command, env=env, cwd=cli_cmd.cwd, timeout_seconds=timeout_seconds)
         if result.stdout:
             result.stdout = self._parse_stream_json(result.stdout)
 
@@ -1078,8 +1089,10 @@ def extract_skill_scripts(self, stdout: str) -> list[str]:
             return []
         return self._extract_script_names(by_name)
 
-    def safe_generate(self, cli_cmd: CLICommand) -> subprocess.CompletedProcess:
-        result = self.generate_internal(cli_cmd)
+    def safe_generate(
+        self, cli_cmd: CLICommand, timeout_seconds: Optional[float] = None
+    ) -> subprocess.CompletedProcess:
+        result = self.generate_internal(cli_cmd, timeout_seconds=timeout_seconds)
         if isinstance(result, str):
             return subprocess.CompletedProcess(args=[], returncode=0, stdout=result)