fix(agent): mark agent infrastructure and simulated-user errors as unscored

evalbench/evaluator/agentevaluator.py

@@ -10,6 +10,8 @@
 from dataset.evalgeminicliinput import EvalGeminiCliRequest
 from generators.models import get_generator
 from generators.models.agent_cli import AgentCliGenerator
+from generators.models.noop_agent import NoopAgentGenerator
+from generators.models.passthrough import NOOPGenerator
 from mp import mprunner
 from util.config import get_eval_case_timeout
 from work.agentgenwork import AgentGenWork
@@ -20,6 +22,24 @@
 from typing import Dict
 
 
+def _infra_error(generator: AgentCliGenerator, result: Any) -> str | None:
+    """Returns an error message if the model API hit a quota or server error."""
+    returncode = getattr(result, "returncode", 0)
+    # 124 is a timeout, which counts as a failure.
+    if returncode == 124:
+        return None
+    api_error = generator.model_api_error(result)
+    if not api_error:
+        return None
+    return f"Agent infrastructure error (exit {returncode}): {api_error.strip()[-500:]}"
+
+
+def _is_noop(simulated_user: SimulatedUser) -> bool:
+    """Returns True if the simulated user returns an empty reply by design."""
+    return isinstance(simulated_user.model_generator,
+                      (NOOPGenerator, NoopAgentGenerator))
+
+
 class AgentEvaluator:
     def __init__(
         self,
@@ -162,6 +182,7 @@ def process_scenario(
             else self.eval_case_timeout_seconds
         )
         last_result = None
+        generated_error = None
 
         for turn in range(max_turns):
             remaining_timeout_seconds = None
@@ -241,13 +262,29 @@ def process_scenario(
                 "agent": result.stdout
             })
 
+            generated_error = _infra_error(self.generator, result)
+            if generated_error:
+                logging.error(generated_error)
+                break
+
             if turn < max_turns - 1:
                 if simulated_user:
-                    next_response = simulated_user.get_next_response(
-                        conversation_plan,
-                        conversation_history,
-                        result.stdout
-                    )
+                    try:
+                        next_response = simulated_user.get_next_response(
+                            conversation_plan,
+                            conversation_history,
+                            result.stdout
+                        )
+                        # QueryGenerator.generate returns "" when out of quota.
+                        # ClaudeGenerator returns None on any error.
+                        if (not str(next_response or "").strip()
+                                and not _is_noop(simulated_user)):
+                            raise RuntimeError(
+                                "empty reply from the simulated user model")
+                    except Exception as e:
+                        generated_error = f"Simulated user error: {e}"
+                        logging.error(generated_error)
+                        break
                     if "TERMINATE" in next_response:
                         logging.info("Simulated user terminated conversation.")
                         break
@@ -268,7 +305,8 @@ def process_scenario(
             accumulated_skills,
             eval_result,
             job_id,
-            metadata
+            metadata,
+            generated_error,
         )
 
     def _log_cli_result(self, turn: int, max_turns: int, result: subprocess.CompletedProcess):
@@ -289,7 +327,8 @@ def _finalize_scenario(
         accumulated_skills: List[str],
         eval_result: Any,
         job_id: str,
-        metadata: Dict[str, Any]
+        metadata: Dict[str, Any],
+        generated_error: str | None = None,
     ):
         """Finalizes the scenario by scoring and appending results."""
         timed_out = (
@@ -305,7 +344,7 @@ def _finalize_scenario(
             "returncode": getattr(last_result, "returncode", 0),
             "timed_out": timed_out,
             "prompt_generator_error": None,
-            "generated_error": None,
+            "generated_error": generated_error,
             "sql_generator_error": None,
             "golden_error": None,
             "generated_sql": "skipped",

evalbench/generators/models/agent_cli.py

@@ -114,3 +114,13 @@ def extract_tools(self, stdout: str) -> list:
     @abstractmethod
     def extract_skills(self, stdout: str) -> list:
         raise NotImplementedError("Subclasses must implement this method")
+
+    def model_api_error(
+        self, result: subprocess.CompletedProcess
+    ) -> Optional[str]:
+        """Returns the model API error if it is a quota or server error.
+
+        The tool under test cannot cause these errors, so the trial is
+        unscored. Each CLI overrides this. By default, every error is a failure.
+        """
+        return None

evalbench/generators/models/claude_code.py

@@ -947,6 +947,16 @@ def _parse_stream_json(
                     # Assistant message with content blocks
                     message = event.get("message", {})
                     content_blocks = message.get("content", [])
+                    if event.get("error") and not event.get("parent_tool_use_id"):
+                        # Set when the API call fails, e.g. "rate_limit".
+                        # A subagent error is skipped, because the main
+                        # agent receives it as a tool result and continues.
+                        final_obj["api_error"] = {
+                            "type": event["error"],
+                            "message": "".join(
+                                b.get("text", "") for b in content_blocks
+                                if isinstance(b, dict)),
+                        }
                     for block in content_blocks:
                         if not isinstance(block, dict):
                             continue
@@ -1168,6 +1178,16 @@ def _parse_stream_json(
     def version(self) -> str:
         return self.claude_code_version
 
+    def model_api_error(self, result: subprocess.CompletedProcess) -> Optional[str]:
+        """Returns the API error for "rate_limit" or "server_error" (any 5xx)."""
+        try:
+            api_error = json.loads(result.stdout or "{}").get("api_error") or {}
+        except (json.JSONDecodeError, AttributeError):
+            return None
+        if api_error.get("type") in ("rate_limit", "server_error"):
+            return api_error.get("message") or api_error["type"]
+        return None
+
     def parse_response(self, stdout: str) -> dict:
         if not stdout:
             return {}

evalbench/generators/models/codex_cli.py

@@ -20,6 +20,14 @@
 )
 _SECRET_MANAGER_URL_PREFIX = "secret_manager://"
 
+# Quota and server errors in a Codex `turn.failed` message. "at capacity" is
+# CodexErr::ServerOverloaded and "high demand" is CodexErr::InternalServerError.
+_CODEX_QUOTA_OR_SERVER_RE = re.compile(
+    r"\b(?:429|500|502|503|504)\b|quota|usage limit|rate limit"
+    r"|too many requests|stream disconnected before completion"
+    r"|at capacity|high demand",
+    re.IGNORECASE)
+
 
 def _strip_sm_prefix(value: str) -> str:
     if value.startswith(_SECRET_MANAGER_URL_PREFIX):
@@ -1014,6 +1022,13 @@ def item_payload(item: dict) -> dict:
                 usage = event.get("usage", {}) or usage
                 continue
 
+            if event_type == "turn.failed":
+                # Only the final error counts. `error` events are retries.
+                error = event.get("error")
+                final_obj["turn_error"] = str(
+                    error.get("message", "") if isinstance(error, dict) else error or "")
+                continue
+
             if event_type == "error":
                 # surface the error text into the response field
                 msg = event.get("message", "")
@@ -1242,6 +1257,16 @@ def _coerce_json(value):
     def version(self) -> str:
         return self.codex_cli_version
 
+    def model_api_error(self, result: subprocess.CompletedProcess) -> Optional[str]:
+        """Returns the `turn.failed` message if it is a quota or server error."""
+        try:
+            turn_error = json.loads(result.stdout or "{}").get("turn_error") or ""
+        except (json.JSONDecodeError, AttributeError):
+            return None
+        if _CODEX_QUOTA_OR_SERVER_RE.search(turn_error):
+            return turn_error
+        return None
+
     def parse_response(self, stdout: str) -> dict:
         if not stdout:
             return {}

evalbench/generators/models/gemini_cli.py

@@ -11,6 +11,14 @@
 from util.context import rpc_id_var
 from util.sessionmgr import validate_session_id
 
+# Quota and server errors in a Gemini "[API Error: ...]" message. A bare
+# "quota" does not match, because a missing quota project is a config error.
+_GEMINI_QUOTA_OR_SERVER_RE = re.compile(
+    r"\b(?:429|500|502|503|504)\b|RESOURCE_EXHAUSTED|UNAVAILABLE|\bINTERNAL\b"
+    r"|DEADLINE_EXCEEDED|quota exceeded|exceeded[^\n]{0,40}quota|exhausted"
+    r"|rate limit|too many requests|overloaded",
+    re.IGNORECASE)
+
 
 class CLICommand:
     def __init__(self, cli, prompt, env=None, resume=False, yolo=True, cwd=None,
@@ -1044,6 +1052,28 @@ def _parse_stream_json(self, stream_output: str) -> str:
     def version(self) -> str:
         return self.gemini_cli_version
 
+    def model_api_error(self, result: subprocess.CompletedProcess) -> Optional[str]:
+        """Returns the final API error if it is a quota or server error.
+
+        On failure, stdout is the raw stream-json. Gemini CLI formats fatal
+        model errors as "[API Error: ...]". Tool errors and retry logs use
+        other formats.
+        """
+        for line in reversed((result.stdout or "").splitlines()):
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(event, dict) or event.get("type") != "result":
+                continue
+            if event.get("status") != "error":
+                return None
+            message = str((event.get("error") or {}).get("message", ""))
+            if message.startswith("[API Error:") and _GEMINI_QUOTA_OR_SERVER_RE.search(message):
+                return message
+            return None
+        return None
+
     def parse_response(self, stdout: str) -> dict:
         if not stdout:
             return {}

evalbench/reporting/analyzer.py

@@ -220,6 +220,12 @@ def analyze_one_metric(
                 "skills_best_practices",
             ]
             if metric_name in non_binary_metrics:
+                # Error rows score 0. Leave them out of the average.
+                if "generated_error" in df_metric.columns:
+                    df_metric = df_metric[
+                        df_metric["generated_error"].isna()
+                        | (df_metric["generated_error"] == "")
+                    ]
                 avg_val = (
                     df_metric["score"].mean()
                     if (not df_metric.empty and "score" in df_metric.columns)