feat(mcp-readability): compliance orchestrator, LLM judge, and metrics scorer

The evaluation half of the MCP-readability work, on top of the mcp_tools
generator. For each endpoint the orchestrator fetches tools (rendered as a
man page), computes deterministic size metrics, gathers applicable waivers,
and judges the man page against the style guide with an LLM. One result row
per endpoint is emitted through the shared EvalBench reporters (CSV/BigQuery).

- McpReadabilityOrchestrator (orchestrator: mcp_readability), driven entirely
  by datasets/mcp_readability/run_config.yaml.
- McpToolMetricsScorer: deterministic tool count / estimated tokens /
  token-budget usage.
- McpStyleComplianceScorer: LLM judge scoring the man page vs the style guide
  (P0/P1/P2 findings, compliance score, waived rules), JSON output.
- enums + exceptions helpers, aligned to the endpoints/exceptions schema and
  the readability_judge run-config block.

Reuses the standard evalbench.py report path with NO changes to evalbench.py:
- process() emits a real scores_tf (one standard style_compliance score row
  per endpoint; pass = SUCCESS with no P0 findings), so the run takes the
  existing results+scores branch and the shared analyzer produces a P0-clean
  compliance rate. run_config declares scorers: [style_compliance].
- dataset_config is made optional in the shared helpers: set_session_configs
  always sets it (default None) and load_dataset_from_json returns {} for a
  falsy path, so the datasetless orchestrator loads an empty dataset.
- CsvReporter.store no-ops on None/empty frames (mirroring BigQueryReporter),
  so subset-only report writes are safe.

Testing: pytest evalbench/test/mcp_readability_test.py
evalbench/test/mcp_tool_metrics_test.py evalbench/test/evalbench_test.py
— 31 passing, including an offline end-to-end orchestrator run, an
analyzer-integration test proving the emitted scores aggregate correctly, and
the existing evalbench.py tests (unchanged).

Author: Akangsha Goel

datasets/mcp_readability/run_config.yaml

@@ -12,6 +12,13 @@
 
 orchestrator: mcp_readability
 
+# Metric the shared analyzer aggregates for the scores / summary reports. The
+# orchestrator emits one `style_compliance` score row per endpoint (pass = no P0
+# findings); it does its own scoring, so this list is only a label for the
+# analyzer.
+scorers:
+  - style_compliance
+
 # Inputs (paths relative to the repo root / run cwd).
 endpoints_config: datasets/mcp_readability/endpoints.yaml
 exceptions_config: datasets/mcp_readability/exceptions.yaml          # optional

evalbench/dataset/dataset.py

@@ -242,6 +242,10 @@ def load_json(json_file_path):
 
 
 def load_dataset_from_json(json_file_path, config):
+    # No dataset path (e.g. orchestrators driven by their run config rather than a
+    # prompt dataset): nothing to load. flatten_dataset({}) yields [].
+    if not json_file_path:
+        return {}
     input_items = {}
     dataset_format = config.get("dataset_format", "evalbench-standard-format")
     if dataset_format == "bird-interact-format":

evalbench/evaluator/__init__.py

@@ -8,6 +8,7 @@
 from evaluator.dataengineeringagentorchestrator import (
     DataEngineeringAgentOrchestrator,
 )
+from evaluator.mcp_readability import McpReadabilityOrchestrator
 import logging
 
 
@@ -27,6 +28,10 @@ def get_orchestrator(config, db_configs, setup_config, report_progress=False):
         return DataEngineeringAgentOrchestrator(
             config, db_configs, setup_config, report_progress
         )
+    elif orchestrator_type == "mcp_readability":
+        return McpReadabilityOrchestrator(
+            config, db_configs, setup_config, report_progress
+        )
     else:
         return Orchestrator(config, db_configs, setup_config, report_progress)
 

evalbench/evaluator/mcp_readability/__init__.py

@@ -0,0 +1,3 @@
+from evaluator.mcp_readability.orchestrator import McpReadabilityOrchestrator
+
+__all__ = ["McpReadabilityOrchestrator"]

evalbench/evaluator/mcp_readability/enums.py

@@ -0,0 +1,72 @@
+"""Enums for the MCP style-guide readability/compliance check.
+
+They are defined as plain Python enums here so the feature is self-contained;
+values can be aligned to an external schema later without touching call sites.
+All enums are written to the results CSV using their ``.name``.
+"""
+
+from enum import Enum
+
+
+class EndpointType(Enum):
+    """Deployment channel / reporting metadata for an endpoint.
+
+    Used both to label result rows and (optionally) to filter which endpoints
+    are checked via ``endpoint_types`` in the run config.
+    """
+
+    ENDPOINT_TYPE_UNSPECIFIED = 0
+    PROD = 1
+    AUTOPUSH = 2
+    STAGING = 3
+    DEV = 4
+
+
+class CheckStatus(Enum):
+    """Whether the compliance check *ran* successfully (not its findings).
+
+    Compliance results are captured separately via the p0/p1/p2 issue counts and
+    compliance_score; this status only reflects whether the eval completed.
+
+    - ``SUCCESS``: the eval ran end-to-end.
+    - ``FETCH_ERROR``: failed to retrieve tools data from the endpoint.
+    - ``ANALYSIS_ERROR``: error during LLM analysis or result parsing.
+    - ``INTERNAL_ERROR``: other script/system error.
+    """
+
+    CHECK_STATUS_UNSPECIFIED = 0
+    SUCCESS = 1
+    FETCH_ERROR = 2
+    ANALYSIS_ERROR = 3
+    INTERNAL_ERROR = 4
+
+
+def _coerce(enum_cls, value, default):
+    """Coerce a config value (str / int / enum / None) into ``enum_cls``.
+
+    Unknown values fall back to ``default`` rather than raising, so a typo in a
+    config file degrades gracefully instead of aborting the whole run.
+    """
+    if value is None:
+        return default
+    if isinstance(value, enum_cls):
+        return value
+    if isinstance(value, int):
+        try:
+            return enum_cls(value)
+        except ValueError:
+            return default
+    name = str(value).strip().upper()
+    # Allow both short ("PROD") and fully-qualified ("ENDPOINT_TYPE_PROD")
+    # spellings, plus the raw member name.
+    members = enum_cls.__members__
+    if name in members:
+        return members[name]
+    for member_name, member in members.items():
+        if member_name.endswith("_" + name):
+            return member
+    return default
+
+
+def coerce_endpoint_type(value) -> EndpointType:
+    return _coerce(EndpointType, value, EndpointType.ENDPOINT_TYPE_UNSPECIFIED)

evalbench/evaluator/mcp_readability/exceptions.py

@@ -0,0 +1,71 @@
+"""Loading and matching of per-endpoint style-rule exceptions (waivers).
+
+An exceptions file lets an operator waive a specific style requirement for a
+specific endpoint when it legitimately cannot comply. Waived rules are passed to
+the scorer so they are excluded from the P0/P1/P2 counts and surfaced separately
+in the feedback.
+
+File schema (see ``datasets/mcp_readability/exceptions.yaml``)::
+
+    exceptions:
+      - product_name: "Cloud SQL"     # any of product_name / endpoint_type
+        rule_id: "tool-names"
+        reason: "..."
+      - endpoint_type: AUTOPUSH       # "*" or omitted field = match-all
+        rule_id: "use-enums"
+        reason: "..."
+"""
+
+import logging
+from util.config import load_yaml_config
+
+
+def load_exceptions(path: str) -> list[dict]:
+    """Load the exceptions list from a YAML file. Missing/empty -> []."""
+    if not path:
+        return []
+    parsed = load_yaml_config(path)
+    if not parsed:
+        return []
+    exceptions = parsed.get("exceptions") or []
+    if not isinstance(exceptions, list):
+        logging.warning(
+            "mcp_readability: 'exceptions' in %s is not a list; ignoring.", path
+        )
+        return []
+    return exceptions
+
+
+def _matches(field_value, exception_value) -> bool:
+    """A matcher field matches when it is absent, '*', or equal (case-insensitive)."""
+    if exception_value is None or exception_value == "*":
+        return True
+    if field_value is None:
+        return False
+    return str(field_value).strip().lower() == str(exception_value).strip().lower()
+
+
+def applicable_exceptions(endpoint: dict, all_exceptions: list[dict]) -> list[dict]:
+    """Return exceptions whose matchers all apply to ``endpoint``.
+
+    Matchers considered: ``product_name`` and ``endpoint_type`` (the endpoint's
+    identity in #469's ``endpoints.yaml``). Each exception keeps its ``rule_id``
+    and ``reason`` for the scorer prompt.
+    """
+    matched = []
+    for exc in all_exceptions:
+        if not isinstance(exc, dict):
+            continue
+        if not exc.get("rule_id"):
+            continue
+        if (
+            _matches(endpoint.get("product_name"), exc.get("product_name"))
+            and _matches(endpoint.get("endpoint_type"), exc.get("endpoint_type"))
+        ):
+            matched.append(
+                {
+                    "rule_id": exc.get("rule_id"),
+                    "reason": exc.get("reason", ""),
+                }
+            )
+    return matched