diff --git a/datasets/gemini-cli-tools/example_run_timeout_config.yaml b/datasets/gemini-cli-tools/example_run_timeout_config.yaml
new file mode 100644
index 00000000..8932efb8
--- /dev/null
+++ b/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'
diff --git a/docs/configs/run-config.md b/docs/configs/run-config.md
index 0c5fcfd6..670b3b1e 100644
--- a/docs/configs/run-config.md
+++ b/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
diff --git a/evalbench/evaluator/agentevaluator.py b/evalbench/evaluator/agentevaluator.py
index f464a83b..0272badc 100644
--- a/evalbench/evaluator/agentevaluator.py
+++ b/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,
diff --git a/evalbench/evaluator/evaluator.py b/evalbench/evaluator/evaluator.py
index 8d6b9de0..0b89ee35 100644
--- a/evalbench/evaluator/evaluator.py
+++ b/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(
diff --git a/evalbench/generators/models/agent_cli.py b/evalbench/generators/models/agent_cli.py
index 41010e6c..ddee1b4a 100644
--- a/evalbench/generators/models/agent_cli.py
+++ b/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
diff --git a/evalbench/generators/models/agy_cli.py b/evalbench/generators/models/agy_cli.py
index 37eee849..15de1de9 100644
--- a/evalbench/generators/models/agy_cli.py
+++ b/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
diff --git a/evalbench/generators/models/claude_code.py b/evalbench/generators/models/claude_code.py
index ef10983d..ebcad6de 100644
--- a/evalbench/generators/models/claude_code.py
+++ b/evalbench/generators/models/claude_code.py
@@ -8,6 +8,7 @@
 import sys
 import re
 import shutil
+from typing import Optional, Union, Dict, List
 from util.context import rpc_id_var
 
 
@@ -597,21 +598,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]}"
@@ -621,7 +632,7 @@ def _execute_cli_command(
                 command, 1, "", f"An unexpected error occurred: {e}"
             )
 
-    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)
@@ -673,7 +684,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)
 
@@ -1054,8 +1065,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)
 
diff --git a/evalbench/generators/models/codex_cli.py b/evalbench/generators/models/codex_cli.py
index 35f6e4ab..a6bcece2 100644
--- a/evalbench/generators/models/codex_cli.py
+++ b/evalbench/generators/models/codex_cli.py
@@ -9,6 +9,7 @@
 import sys
 import threading
 import time
+from typing import Optional, Union, Dict, List
 from util.context import rpc_id_var
 
 
@@ -703,10 +704,10 @@ def _toml_value(cls, value) -> str:
             return "{ " + inner + " }"
         return json.dumps(str(value))
 
-    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.codex_cli_version, str(cli_cmd))
-        return self._run_codex_cli(cli_cmd)
+        return self._run_codex_cli(cli_cmd, timeout_seconds=timeout_seconds)
 
     _EV_ITEM_STARTED = "item.started"
     _EV_ITEM_UPDATED = "item.updated"
@@ -719,6 +720,7 @@ def generate_internal(self, cli_cmd):
 
     def _execute_cli_command(
         self, command: list[str], env: dict[str, str] | None = None, cwd: str | None = None,
+        timeout_seconds: float | int | None = None,
     ) -> tuple[subprocess.CompletedProcess, dict[str, int]]:
         """Runs the Codex CLI with line-streamed stdout so we can stamp the
         wall-clock time at which each NDJSON event arrives.
@@ -762,17 +764,32 @@ def _drain_stderr():
         started_at_ms: dict[str, float] = {}
         tool_durations: dict[str, int] = {}
 
+        def _drain_stdout():
+            try:
+                for line in proc.stdout:
+                    arrival_ms = time.monotonic() * 1000
+                    stdout_lines.append(line)
+                    self._stamp_tool_event(
+                        line, arrival_ms, started_at_ms, tool_durations,
+                    )
+            except Exception as e:
+                logging.warning(f"stdout stream read failed: {e}")
+
+        stdout_thread = threading.Thread(target=_drain_stdout, daemon=True)
+        stdout_thread.start()
+
         try:
-            for line in proc.stdout:
-                arrival_ms = time.monotonic() * 1000
-                stdout_lines.append(line)
-                self._stamp_tool_event(
-                    line, arrival_ms, started_at_ms, tool_durations,
-                )
-        except Exception as e:
-            logging.warning(f"stdout stream read failed: {e}")
+            proc.wait(timeout=timeout_seconds)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait()
+            stdout_thread.join(timeout=5)
+            stderr_thread.join(timeout=5)
+            return subprocess.CompletedProcess(
+                command, 124, "".join(stdout_lines), f"TimeoutError: Command timed out after {timeout_seconds} seconds"
+            ), tool_durations
 
-        proc.wait()
+        stdout_thread.join(timeout=5)
         stderr_thread.join(timeout=5)
 
         completed = subprocess.CompletedProcess(
@@ -813,7 +830,7 @@ def _stamp_tool_event(
             if t0 is not None:
                 tool_durations[item_id] = max(0, int(arrival_ms - t0))
 
-    def _run_codex_cli(self, cli_cmd: CLICommand):
+    def _run_codex_cli(self, cli_cmd: CLICommand, timeout_seconds=None):
         env = os.environ.copy()
         env.update(self.env)
         env.update(cli_cmd.env)
@@ -866,7 +883,7 @@ def _run_codex_cli(self, cli_cmd: CLICommand):
         logging.info(f"Running Codex CLI: {' '.join(command)}")
 
         start_ms = time.monotonic()
-        result, tool_durations = self._execute_cli_command(command, env=env, cwd=cli_cmd.cwd)
+        result, tool_durations = self._execute_cli_command(command, env=env, cwd=cli_cmd.cwd, timeout_seconds=timeout_seconds)
         duration_ms = int((time.monotonic() - start_ms) * 1000)
         if result.stdout:
             result.stdout = self._parse_stream_json(
@@ -1294,8 +1311,10 @@ def add_skill(name: str):
 
         return items
 
-    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)
         if not result.stdout and result.returncode != 0:
diff --git a/evalbench/generators/models/gemini_cli.py b/evalbench/generators/models/gemini_cli.py
index a72e1231..0f0d7757 100644
--- a/evalbench/generators/models/gemini_cli.py
+++ b/evalbench/generators/models/gemini_cli.py
@@ -7,6 +7,7 @@
 import re
 import shutil
 import sys
+from typing import Optional, Union, Dict, List
 from util.context import rpc_id_var
 
 
@@ -784,16 +785,16 @@ def _patch_manifest_sensitive(self, ext_path):
         except Exception as e:
             logging.error(f"Failed to patch manifest at {manifest_path}: {e}")
 
-    def generate_internal(self, cli_cmd: CLICommand | str):
+    def generate_internal(self, cli_cmd: CLICommand | str, timeout_seconds: float | int | None = None):
         if not isinstance(cli_cmd, CLICommand):
             cli_cmd = CLICommand(self.gemini_cli_version, str(cli_cmd))
-        return self._run_gemini_cli(cli_cmd)
+        return self._run_gemini_cli(cli_cmd, timeout_seconds=timeout_seconds)
 
     def _execute_cli_command(
-        self, command: list[str], env: dict[str, str] | None = None, cwd: str | None = None
+        self, command: list[str], env: dict[str, 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)
+            result = subprocess.run(command, capture_output=True, text=True, check=False, env=env, cwd=cwd if cwd else self.fake_home, timeout=timeout_seconds)
             # Filter out benign schema warnings from json decoder from stderr to reduce noise
             if result.stderr:
                 result.stderr = "\n".join(
@@ -804,6 +805,15 @@ def _execute_cli_command(
                     ]
                 )
             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]}"
@@ -813,7 +823,7 @@ def _execute_cli_command(
                 command, 1, "", f"An unexpected error occurred: {e}"
             )
 
-    def _run_gemini_cli(self, cli_cmd: CLICommand):
+    def _run_gemini_cli(self, cli_cmd: CLICommand, timeout_seconds: float | int | None = None):
         gemini_settings_path = os.path.join(self.gemini_home, "settings.json")
 
         if not os.path.exists(gemini_settings_path):
@@ -847,7 +857,9 @@ def _run_gemini_cli(self, cli_cmd: CLICommand):
             ]
         )
 
-        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.returncode == 0 and result.stdout:
             result.stdout = self._parse_stream_json(result.stdout)
 
@@ -1072,8 +1084,10 @@ def extract_skills(self, stdout: str) -> list[str]:
         except (KeyError, TypeError):
             return []
 
-    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)
 
diff --git a/evalbench/generators/models/noop_agent.py b/evalbench/generators/models/noop_agent.py
index f43e29cb..6bb76443 100644
--- a/evalbench/generators/models/noop_agent.py
+++ b/evalbench/generators/models/noop_agent.py
@@ -9,6 +9,8 @@
 
 import subprocess
 
+from typing import Optional
+
 from .agent_cli import AgentCliGenerator
 
 
@@ -31,7 +33,9 @@ def create_command(
     ):
         return []
 
-    def safe_generate(self, cli_cmd):
+    def safe_generate(
+        self, cli_cmd, timeout_seconds: Optional[float] = None
+    ) -> subprocess.CompletedProcess:
         return subprocess.CompletedProcess(
             args=[], returncode=0, stdout="", stderr=""
         )
diff --git a/evalbench/test/test_eval_case_timeout.py b/evalbench/test/test_eval_case_timeout.py
new file mode 100644
index 00000000..2e4d01ba
--- /dev/null
+++ b/evalbench/test/test_eval_case_timeout.py
@@ -0,0 +1,475 @@
+import concurrent.futures
+import subprocess
+import time
+import unittest
+from unittest.mock import MagicMock, patch
+
+from evaluator.agentevaluator import AgentEvaluator
+from evaluator.evaluator import Evaluator, _process_futures_with_timeout
+from generators.models.agent_cli import AgentCliGenerator
+from generators.models.gemini_cli import GeminiCliGenerator
+from generators.models.codex_cli import CodexCliGenerator
+from generators.models.claude_code import ClaudeCodeGenerator
+from generators.models.agy_cli import AgyCliGenerator
+from util.config import get_eval_case_timeout, parse_timeout_seconds
+from work.agentscorework import AgentScoreWork
+
+
+class TestEvalCaseTimeoutConfig(unittest.TestCase):
+    """Unit tests for parse_timeout_seconds and get_eval_case_timeout."""
+
+    def test_parse_timeout_seconds_numeric(self):
+        self.assertEqual(parse_timeout_seconds(300), 300.0)
+        self.assertEqual(parse_timeout_seconds(12.5), 12.5)
+        self.assertEqual(parse_timeout_seconds(0), 0.0)
+        with self.assertRaises(ValueError):
+            parse_timeout_seconds(-10)
+        self.assertIsNone(parse_timeout_seconds(None))
+
+    def test_parse_timeout_seconds_strings(self):
+        self.assertEqual(parse_timeout_seconds("300"), 300.0)
+        self.assertEqual(parse_timeout_seconds("45.5"), 45.5)
+        self.assertEqual(parse_timeout_seconds("300s"), 300.0)
+        self.assertEqual(parse_timeout_seconds("5m"), 300.0)
+        self.assertEqual(parse_timeout_seconds("1h"), 3600.0)
+        self.assertEqual(parse_timeout_seconds("1h30m"), 5400.0)
+        self.assertEqual(parse_timeout_seconds("1h\t30m"), 5400.0)
+        self.assertEqual(parse_timeout_seconds("5m 30s"), 330.0)
+        self.assertEqual(parse_timeout_seconds("10  s"), 10.0)
+        self.assertEqual(parse_timeout_seconds("1d"), 86400.0)
+        self.assertEqual(parse_timeout_seconds("500ms"), 0.5)
+        self.assertEqual(parse_timeout_seconds(" 10m "), 600.0)
+
+    def test_parse_timeout_seconds_invalid(self):
+        with self.assertRaises(ValueError):
+            parse_timeout_seconds("invalid")
+        self.assertIsNone(parse_timeout_seconds(""))
+        with self.assertRaises(TypeError):
+            parse_timeout_seconds([])
+        with self.assertRaises(TypeError):
+            parse_timeout_seconds({})
+        with self.assertRaises(TypeError):
+            parse_timeout_seconds(True)
+        with self.assertRaises(ValueError):
+            parse_timeout_seconds("1h and 30m")
+
+    def test_get_eval_case_timeout_top_level(self):
+        self.assertEqual(
+            get_eval_case_timeout({"eval_case_timeout": 300}), 300.0
+        )
+        self.assertEqual(
+            get_eval_case_timeout({"eval_case_timeout": "5m"}), 300.0
+        )
+        self.assertEqual(
+            get_eval_case_timeout({"eval_case_timeout": "1h30m"}), 5400.0
+        )
+
+    def test_get_eval_case_timeout_default_none(self):
+        self.assertIsNone(get_eval_case_timeout({}))
+        self.assertIsNone(get_eval_case_timeout({"runners": {}}))
+        self.assertIsNone(get_eval_case_timeout({"runners": {"task_timeout_seconds": 400}}))
+        self.assertIsNone(get_eval_case_timeout({"orchestrator": "geminicli"}))
+
+
+class TestAgentEvaluatorTimeout(unittest.TestCase):
+    """Unit tests for AgentEvaluator timeout enforcement."""
+
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_agent_evaluator_initializes_timeout(self, mock_get_generator):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_get_generator.return_value = mock_generator
+
+        # With timeout in run_config
+        config_with_timeout = {
+            "model_config": "dummy.yaml",
+            "eval_case_timeout": "300s",
+        }
+        evaluator = AgentEvaluator(config_with_timeout)
+        self.assertEqual(evaluator.eval_case_timeout_seconds, 300.0)
+
+        # Without timeout (default)
+        config_without_timeout = {
+            "model_config": "dummy.yaml",
+        }
+        evaluator_default = AgentEvaluator(config_without_timeout)
+        self.assertIsNone(evaluator_default.eval_case_timeout_seconds)
+
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_process_scenario_times_out_multi_turn(self, mock_get_generator):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_generator.fake_home = "/tmp/fake_home"
+        mock_generator.create_command.return_value = MagicMock()
+        mock_generator.parse_response.return_value = {"session_id": "s1"}
+        mock_generator.extract_tools.return_value = []
+        mock_generator.extract_skills.return_value = []
+        mock_get_generator.return_value = mock_generator
+
+        config = {
+            "model_config": "dummy.yaml",
+            "eval_case_timeout": 0.05,  # 50ms timeout
+        }
+        evaluator = AgentEvaluator(config)
+        evaluator._finalize_scenario = MagicMock()
+
+        # Generator simulates a slow execution
+        def slow_safe_generate(cli_cmd, timeout_seconds=None):
+            time.sleep(0.08)
+            return subprocess.CompletedProcess(
+                args=["mock"], returncode=0, stdout='{"response": "ok"}', stderr=""
+            )
+
+        mock_generator.safe_generate.side_effect = slow_safe_generate
+
+        simulated_user = MagicMock()
+        simulated_user.get_next_response.return_value = "next question"
+
+        scenario = {
+            "id": "scenario_timeout_test",
+            "starting_prompt": "Hello",
+            "max_turns": 3,
+        }
+
+        evaluator.process_scenario(
+            scenario=scenario,
+            eval_result=MagicMock(),
+            job_id="job_123",
+            metadata={},
+            simulated_user=simulated_user,
+        )
+
+        # Verify finalize_scenario was called with the result
+        evaluator._finalize_scenario.assert_called_once()
+        args, kwargs = evaluator._finalize_scenario.call_args
+        last_result = args[1]
+        self.assertIsNotNone(last_result)
+        self.assertEqual(last_result.returncode, 124)
+        self.assertIn("TimeoutError", last_result.stderr)
+        # Should have only executed 1 turn due to timeout
+        self.assertEqual(mock_generator.safe_generate.call_count, 1)
+
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_process_scenario_times_out_before_turn_1(self, mock_get_generator):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_get_generator.return_value = mock_generator
+
+        config = {
+            "model_config": "dummy.yaml",
+            "eval_case_timeout": 0.0,  # 0s timeout -> immediately expired
+        }
+        evaluator = AgentEvaluator(config)
+        evaluator._finalize_scenario = MagicMock()
+
+        scenario = {
+            "id": "scenario_immediate_timeout",
+            "starting_prompt": "Hello",
+            "max_turns": 3,
+        }
+
+        evaluator.process_scenario(
+            scenario=scenario,
+            eval_result=MagicMock(),
+            job_id="job_123",
+            metadata={},
+        )
+
+        # Finalize scenario must be called even if timed out before turn 1
+        evaluator._finalize_scenario.assert_called_once()
+        args, kwargs = evaluator._finalize_scenario.call_args
+        last_result = args[1]
+        self.assertEqual(last_result.returncode, 124)
+        self.assertIn("TimeoutError", last_result.stderr)
+        self.assertEqual(mock_generator.safe_generate.call_count, 0)
+
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_process_scenario_scenario_level_timeout_override(self, mock_get_generator):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_generator.fake_home = "/tmp/fake_home"
+        mock_generator.create_command.return_value = MagicMock()
+        mock_generator.parse_response.return_value = {"session_id": "s1"}
+        mock_generator.extract_tools.return_value = []
+        mock_generator.extract_skills.return_value = []
+        mock_get_generator.return_value = mock_generator
+
+        # Run config has NO timeout set
+        config = {"model_config": "dummy.yaml"}
+        evaluator = AgentEvaluator(config)
+        evaluator._finalize_scenario = MagicMock()
+
+        def slow_safe_generate(cli_cmd, timeout_seconds=None):
+            time.sleep(0.08)
+            return subprocess.CompletedProcess(
+                args=["mock"], returncode=0, stdout='{"response": "ok"}', stderr=""
+            )
+
+        mock_generator.safe_generate.side_effect = slow_safe_generate
+
+        simulated_user = MagicMock()
+        simulated_user.get_next_response.return_value = "next question"
+
+        # Scenario overrides timeout locally to 50ms
+        scenario = {
+            "id": "scenario_override_test",
+            "starting_prompt": "Hello",
+            "max_turns": 3,
+            "eval_case_timeout": "50ms",
+        }
+
+        evaluator.process_scenario(
+            scenario=scenario,
+            eval_result=MagicMock(),
+            job_id="job_123",
+            metadata={},
+            simulated_user=simulated_user,
+        )
+
+        # Should have only executed 1 turn due to scenario-level 50ms timeout
+        self.assertEqual(mock_generator.safe_generate.call_count, 1)
+
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_process_scenario_records_timeout_error_in_result(self, mock_get_generator):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_generator.fake_home = "/tmp/fake_home"
+        mock_generator.create_command.return_value = MagicMock()
+        mock_generator.parse_response.return_value = {}
+        mock_generator.extract_tools.return_value = []
+        mock_generator.extract_skills.return_value = []
+        mock_get_generator.return_value = mock_generator
+
+        config = {
+            "model_config": "dummy.yaml",
+            "eval_case_timeout": 0.05,
+        }
+        evaluator = AgentEvaluator(config)
+
+        # Simulate safe_generate returning a TimeoutError
+        mock_generator.safe_generate.return_value = subprocess.CompletedProcess(
+            args=["mock"],
+            returncode=124,
+            stdout="",
+            stderr="TimeoutError: Command timed out after 0.05 seconds",
+        )
+
+        with patch("evaluator.agentevaluator.AgentScoreWork") as mock_score_work:
+            mock_instance = MagicMock()
+            mock_score_work.return_value = mock_instance
+
+            scenario = {
+                "id": "scenario_timeout_err",
+                "starting_prompt": "Execute task",
+                "max_turns": 2,
+            }
+
+            evaluator.process_scenario(
+                scenario=scenario,
+                eval_result=MagicMock(scoring_results=[]),
+                job_id="job_123",
+                metadata={},
+            )
+
+            # Check that AgentScoreWork received eval_output with job_id, timed_out, and artifacts populated
+            mock_score_work.assert_called_once()
+            eval_output = mock_score_work.call_args[1]["eval_output"]
+            self.assertEqual(eval_output["job_id"], "job_123")
+            self.assertIn("TimeoutError", eval_output["stderr"])
+            self.assertIsNone(eval_output["generated_error"])
+            self.assertTrue(eval_output["timed_out"])
+            self.assertEqual(eval_output["returncode"], 124)
+
+    def test_agent_score_work_keeps_generated_error_none(self):
+        eval_output = {
+            "eval_id": "test_err_prop",
+            "timed_out": True,
+            "accumulated_tools": [],
+            "scenario": {"starting_prompt": "prompt", "expected_trajectory": []},
+            "metadata": {},
+            "job_id": "job_123",
+        }
+        score_work = AgentScoreWork(
+            config={},
+            eval_output=eval_output,
+            scoring_results=[],
+        )
+        with patch("scorers.score.compare") as mock_compare:
+            score_work.run()
+            mock_compare.assert_called_once()
+            eval_output_item = mock_compare.call_args[1]["eval_output_item"]
+            self.assertIsNone(eval_output_item["generated_error"])
+
+    @patch("evaluator.agentevaluator.SimulatedUser")
+    @patch("evaluator.agentevaluator.get_generator")
+    def test_evaluate_agent_cli_returns_job_id_and_outputs_on_timeout(self, mock_get_generator, mock_sim_user):
+        mock_generator = MagicMock(spec=AgentCliGenerator)
+        mock_generator.name = "mock_agent_cli"
+        mock_generator.version = "1.0"
+        mock_generator.fake_home = "/tmp/fake_home"
+        mock_generator.create_command.return_value = MagicMock()
+        mock_generator.parse_response.return_value = {}
+        mock_generator.extract_tools.return_value = []
+        mock_generator.extract_skills.return_value = []
+        mock_get_generator.return_value = mock_generator
+
+        # Subprocess times out
+        mock_generator.safe_generate.return_value = subprocess.CompletedProcess(
+            args=["mock"],
+            returncode=124,
+            stdout="partial stdout output",
+            stderr="TimeoutError: Command timed out after 0.05 seconds",
+        )
+
+        config = {
+            "model_config": "dummy.yaml",
+            "simulated_user_model_config": "dummy_user.yaml",
+            "eval_case_timeout": 0.05,
+            "runners": {"agent_runners": 1},
+        }
+        evaluator = AgentEvaluator(config)
+
+        import json
+        import types
+        mock_request = types.SimpleNamespace(
+            payload=json.dumps({
+                "scenarios": [{
+                    "id": "scenario_timeout_upload",
+                    "starting_prompt": "Run long task",
+                    "max_turns": 3,
+                }]
+            }),
+            agent_results=[],
+            scoring_results=[],
+        )
+
+        mock_sim_user.return_value.get_next_response.return_value = "TERMINATE"
+
+        with patch("evaluator.agentevaluator.AgentScoreWork"):
+            eval_outputs, scoring_results = evaluator._evaluate_agent_cli(
+                dataset=[mock_request],
+                job_id="job_timeout_456",
+                run_time=MagicMock(),
+            )
+
+            # Validate that outputs returned for reporting/upload retain job_id and artifacts
+            self.assertEqual(len(eval_outputs), 1)
+            output = eval_outputs[0]
+            self.assertEqual(output["job_id"], "job_timeout_456")
+            self.assertEqual(output["eval_id"], "scenario_timeout_upload")
+            self.assertEqual(output["returncode"], 124)
+            self.assertTrue(output["timed_out"])
+            self.assertIsNone(output["generated_error"])
+            self.assertEqual(output["stdout"], "partial stdout output")
+
+
+class TestEvaluatorFuturesTimeout(unittest.TestCase):
+    """Unit tests for OneShot Evaluator task timeout handling."""
+
+    def test_evaluator_configures_timeout(self):
+        # eval_case_timeout does NOT override task_timeout_seconds (kept separate)
+        config_with_eval_case_timeout = {
+            "eval_case_timeout": "2m",
+        }
+        evaluator = Evaluator(config_with_eval_case_timeout)
+        self.assertEqual(evaluator.task_timeout_seconds, 600)
+        self.assertEqual(evaluator.scoring_runners, 10)
+        self.assertEqual(evaluator.num_trials, 1)
+
+        config_with_runner_timeout = {
+            "runners": {"task_timeout_seconds": 450, "scoring_runners": 5},
+            "num_trials": 3,
+        }
+        evaluator_runner = Evaluator(config_with_runner_timeout)
+        self.assertEqual(evaluator_runner.task_timeout_seconds, 450)
+        self.assertEqual(evaluator_runner.scoring_runners, 5)
+        self.assertEqual(evaluator_runner.num_trials, 3)
+
+    def test_process_futures_with_timeout_times_out(self):
+        mock_future = concurrent.futures.Future()
+        future_map = {mock_future: {"id": "test_1"}}
+
+        # Generator that yields with timeout
+        results = list(
+            _process_futures_with_timeout(
+                [mock_future],
+                future_map,
+                timeout=0.05,
+            )
+        )
+
+        self.assertEqual(len(results), 1)
+        future, eval_out, timed_out = results[0]
+        self.assertTrue(timed_out)
+        self.assertEqual(eval_out["id"], "test_1")
+
+
+class TestGeneratorsTimeout(unittest.TestCase):
+    """Unit tests for CLI generators timeout execution."""
+
+    @patch("subprocess.run")
+    def test_gemini_cli_run_timeout(self, mock_subprocess_run):
+        mock_subprocess_run.side_effect = subprocess.TimeoutExpired(
+            cmd=["npm", "exec"], timeout=10.0, output="partial", stderr="err"
+        )
+        gen = GeminiCliGenerator.__new__(GeminiCliGenerator)
+        gen.fake_home = "/tmp/fake_home"
+        gen.gemini_home = "/tmp/fake_home/.gemini"
+        gen.gemini_cli_version = "@google/gemini-cli@0.36.0"
+        gen.env = {}
+
+        result = gen._execute_cli_command(["npm", "exec"], timeout_seconds=10.0)
+        self.assertEqual(result.returncode, 124)
+        self.assertIn("TimeoutError", result.stderr)
+
+    @patch("subprocess.Popen")
+    def test_codex_cli_run_timeout(self, mock_popen):
+        mock_proc = MagicMock()
+        mock_proc.stdout = iter(["item 1\n", "item 2\n"])
+        mock_proc.stderr = iter([])
+        mock_proc.wait.side_effect = [
+            subprocess.TimeoutExpired(cmd=["codex"], timeout=5.0),
+            0,
+        ]
+        mock_popen.return_value = mock_proc
+
+        gen = CodexCliGenerator.__new__(CodexCliGenerator)
+        gen.fake_home = "/tmp/fake_home"
+
+        result, durations = gen._execute_cli_command(["codex"], timeout_seconds=5.0)
+        self.assertEqual(result.returncode, 124)
+        self.assertIn("TimeoutError", result.stderr)
+        mock_proc.kill.assert_called_once()
+
+    @patch("subprocess.run")
+    def test_claude_code_run_timeout(self, mock_subprocess_run):
+        mock_subprocess_run.side_effect = subprocess.TimeoutExpired(
+            cmd=["claude"], timeout=5.0
+        )
+        gen = ClaudeCodeGenerator.__new__(ClaudeCodeGenerator)
+        gen.fake_home = "/tmp/fake_home"
+
+        result = gen._execute_cli_command(["claude"], timeout_seconds=5.0)
+        self.assertEqual(result.returncode, 124)
+        self.assertIn("TimeoutError", result.stderr)
+
+    @patch("subprocess.run")
+    def test_agy_cli_run_timeout(self, mock_subprocess_run):
+        mock_subprocess_run.side_effect = subprocess.TimeoutExpired(
+            cmd=["agy"], timeout=5.0
+        )
+        gen = AgyCliGenerator.__new__(AgyCliGenerator)
+        gen.fake_home = "/tmp/fake_home"
+
+        result = gen._execute_cli_command(["agy"], timeout_seconds=5.0)
+        self.assertEqual(result.returncode, 124)
+        self.assertIn("TimeoutError", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/evalbench/util/config.py b/evalbench/util/config.py
index df6a8b01..67f83950 100644
--- a/evalbench/util/config.py
+++ b/evalbench/util/config.py
@@ -4,8 +4,9 @@
 import os
 import csv
 import random
+import re
 import string
-from typing import List
+from typing import List, Optional
 
 from pyaml_env import parse_config
 import pandas as pd
@@ -240,3 +241,72 @@ def breakdown_db_configs_by_dialect(db_configs: list[dict]):
         else:
             db_configs_by_dialect[dialect] = [db_config]
     return db_configs_by_dialect
+
+
+def parse_timeout_seconds(val) -> Optional[float]:
+    """Parses a timeout duration (int, float, or string like '300', '5m', '1h', '500ms') into seconds as float.
+
+    Returns None if val is None or empty string.
+    Raises ValueError if val is negative or has an invalid format string.
+    Raises TypeError if val is not int, float, str, or None.
+    """
+    if val is None:
+        return None
+    if isinstance(val, bool):
+        raise TypeError(f"Timeout cannot be a boolean: {val}")
+    if isinstance(val, (int, float)):
+        if val < 0:
+            raise ValueError(f"Timeout cannot be negative: {val}")
+        return float(val)
+    if not isinstance(val, str):
+        raise TypeError(
+            f"Timeout must be int, float, str, or None, got {type(val).__name__}"
+        )
+
+    s = val.strip()
+    if not s:
+        return None
+
+    try:
+        f = float(s)
+        if f < 0:
+            raise ValueError(f"Timeout cannot be negative: {val}")
+        return f
+    except ValueError:
+        pass
+
+    matches = list(re.finditer(r"(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)", s))
+    if not matches:
+        raise ValueError(f"Invalid timeout format: '{val}'")
+
+    matched_len = sum(len(m.group(1)) + len(m.group(2)) for m in matches)
+    if matched_len != len(re.sub(r"\s+", "", s)):
+        raise ValueError(f"Invalid timeout format: '{val}'")
+
+    unit_multipliers = {
+        "ms": 0.001,
+        "s": 1.0,
+        "m": 60.0,
+        "h": 3600.0,
+        "d": 86400.0,
+    }
+
+    total = 0.0
+    for m in matches:
+        amount = float(m.group(1))
+        unit = m.group(2)
+        total += amount * unit_multipliers[unit]
+
+    if total < 0:
+        raise ValueError(f"Timeout cannot be negative: {val}")
+
+    return total
+
+
+def get_eval_case_timeout(config: dict) -> Optional[float]:
+    """Extracts and parses eval_case_timeout setting from run config or scenario."""
+    if not isinstance(config, dict):
+        return None
+
+    val = config.get("eval_case_timeout")
+    return parse_timeout_seconds(val) if val is not None else None