Run: {escaped_timestamp} | {total} generations | {len(df["model"].unique())} models
-
Pass Rate
= 50 else "#e74c3c"}">{pass_rate}%
{passed}/{len(eligible)} eligible
-
Outcomes
{passed} / {validation_failed}
passed / failed
{ineligible} ineligible / {generation_errors} generation errors / {check_errors} check errors
+
Pass Rate
{pass_rate_value}
{eligible_summary}
{exception_summary}
+
Outcomes
{passed} / {validation_failed}
passed / failed
{ineligible} ineligible / {generation_errors} generation errors / {rate_limited} rate limited / {check_errors} check errors
Total Reported Cost
${total_reported_cost:.4f}
{known_costs}/{total} costs reported
{"".join(chart_divs)}
diff --git a/src/conductor_eval/evaluator.py b/src/conductor_eval/evaluator.py
index 4276071..0fdfe71 100644
--- a/src/conductor_eval/evaluator.py
+++ b/src/conductor_eval/evaluator.py
@@ -20,6 +20,7 @@
from uuid import uuid4
from conductor_core import EngineConfig, GenerationRequest, LoopGenerationEngine
+from conductor_core.errors import ProviderRateLimitError
from conductor_core.music import DURATION_KEYWORDS, get_model_info
from conductor_core.providers import ollama as ollama_api
from mido import MidiFile
@@ -36,6 +37,7 @@
polyphony_test,
scale_test,
)
+from conductor_eval.outcomes import get_overall_status
from conductor_eval.paths import get_evaluations_dir
DIRECT_EVALUATION_CONFIRMATION = "RUN CLOUD EVALUATION"
@@ -131,6 +133,7 @@ class Evaluator:
"""
SCALES = ["major", "minor"]
+ MAX_CLOUD_CONCURRENCY = 4
AVAILABLE_TESTS = {
"scale": scale_test,
@@ -230,6 +233,7 @@ def evaluate(
tests: list[str] = ["scale", "duration"],
test_reasoning: bool = False,
test_params: dict[str, dict] | None = None,
+ max_cloud_concurrency: int = MAX_CLOUD_CONCURRENCY,
) -> dict:
"""
Run evaluation across all specified combinations.
@@ -244,6 +248,7 @@ def evaluate(
test_reasoning: If True, test all thinking modes and effort levels for compatible models.
test_params: Explicit keyword arguments for named tests. Duration parameters override
prompt detection; omitted duration parameters still use keyword detection.
+ max_cloud_concurrency: Maximum simultaneous requests to each cloud provider.
Returns:
dict: Summary of evaluation results.
@@ -253,6 +258,12 @@ def evaluate(
"""
if run_name is None:
raise ValueError("run_name is required")
+ if (
+ isinstance(max_cloud_concurrency, bool)
+ or not isinstance(max_cloud_concurrency, int)
+ or max_cloud_concurrency <= 0
+ ):
+ raise ValueError("max_cloud_concurrency must be a positive integer")
tests = self._with_required_scale_test(tests)
test_params = self._validate_test_params(tests, test_params)
@@ -281,6 +292,7 @@ def evaluate(
"test_params": test_params,
"test_reasoning": test_reasoning,
"temperature": self.temperature,
+ "max_cloud_concurrency": max_cloud_concurrency,
}
with open(run_path / "config.json", "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
@@ -305,7 +317,13 @@ def evaluate(
# Run async tasks (cloud providers)
if async_tasks:
async_results = asyncio.run(
- self._run_async_batch(async_tasks, run_path, tests, logger)
+ self._run_async_batch(
+ async_tasks,
+ run_path,
+ tests,
+ logger,
+ max_cloud_concurrency,
+ )
)
all_results.extend(async_results)
@@ -820,9 +838,10 @@ async def _run_async_batch(
run_path: Path,
tests_to_run: list[str],
logger: logging.Logger,
+ max_cloud_concurrency: int = MAX_CLOUD_CONCURRENCY,
) -> list[dict]:
"""
- Run tasks asynchronously with rate limiting.
+ Run cloud tasks with an independent concurrency cap for each provider.
Args:
tasks: List of task dictionaries
@@ -832,17 +851,8 @@ async def _run_async_batch(
Returns:
list: List of result dictionaries
"""
- # Build semaphores from RPM
- semaphores = {}
- for provider in ["OpenAI", "Anthropic", "Google"]:
- if provider in self.model_info["models"]:
- rpms = []
- for model in self.model_info["models"][provider].keys():
- rate_info = self.model_info["models"][provider][model].get("rate_limits", {})
- rpm = rate_info.get("RPM", 60)
- rpms.append(rpm)
- max_concurrent = max(1, min(rpms) // 60) if rpms else 1
- semaphores[provider] = asyncio.Semaphore(max_concurrent)
+ providers = {task["provider"] for task in tasks}
+ semaphores = {provider: asyncio.Semaphore(max_cloud_concurrency) for provider in providers}
results = []
total_tasks = len(tasks)
@@ -859,7 +869,7 @@ async def _run_async_batch(
async def run_single_task(task: dict) -> dict:
provider = task["provider"]
- async with semaphores.get(provider, asyncio.Semaphore(1)):
+ async with semaphores[provider]:
return await asyncio.to_thread(
self._run_single,
task=task,
@@ -1103,7 +1113,9 @@ def _run_single(
)
result["error"] = str(e)
result["tests"]["overall_pass"] = False
- result["tests"]["overall_status"] = "generation_error"
+ result["tests"]["overall_status"] = (
+ "rate_limited" if isinstance(e, ProviderRateLimitError) else "generation_error"
+ )
# Still save the result even on failure
self._save_results(result, None, [], run_path, task)
return result
@@ -1179,6 +1191,7 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
"successful_generations": 0,
"failed_generations": 0,
"generation_error_generations": 0,
+ "rate_limited_generations": 0,
"check_error_generations": 0,
"validation_failed_generations": 0,
"ineligible_generations": 0,
@@ -1199,18 +1212,15 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
}
for r in all_results:
- tests = r.get("tests", {})
- if r.get("error"):
- outcome = "generation_error"
- else:
- outcome = tests.get(
- "overall_status", "passed" if tests.get("overall_pass", False) else "failed"
- )
+ outcome = get_overall_status(r)
# Totals
if r.get("error"):
summary["totals"]["failed_generations"] += 1
- summary["totals"]["generation_error_generations"] += 1
+ if outcome == "rate_limited":
+ summary["totals"]["rate_limited_generations"] += 1
+ else:
+ summary["totals"]["generation_error_generations"] += 1
else:
summary["totals"]["successful_generations"] += 1
@@ -1251,6 +1261,7 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
"passed": 0,
"failed": 0,
"generation_errors": 0,
+ "rate_limited": 0,
"check_errors": 0,
"validation_failed": 0,
"ineligible": 0,
@@ -1269,7 +1280,10 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
m["passed"] += 1
if r.get("error"):
m["failed"] += 1
- m["generation_errors"] += 1
+ if outcome == "rate_limited":
+ m["rate_limited"] += 1
+ else:
+ m["generation_errors"] += 1
elif outcome == "failed":
m["validation_failed"] += 1
elif outcome == "ineligible":
@@ -1295,6 +1309,7 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
"passed": 0,
"validation_failed": 0,
"generation_errors": 0,
+ "rate_limited": 0,
"check_errors": 0,
"ineligible": 0,
"eligible": 0,
@@ -1307,6 +1322,8 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
summary["by_root"][root]["validation_failed"] += 1
elif outcome == "generation_error":
summary["by_root"][root]["generation_errors"] += 1
+ elif outcome == "rate_limited":
+ summary["by_root"][root]["rate_limited"] += 1
elif outcome == "ineligible":
summary["by_root"][root]["ineligible"] += 1
elif outcome == "check_error":
@@ -1322,6 +1339,7 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
"passed": 0,
"validation_failed": 0,
"generation_errors": 0,
+ "rate_limited": 0,
"check_errors": 0,
"ineligible": 0,
"eligible": 0,
@@ -1334,6 +1352,8 @@ def _generate_summary(self, all_results: list[dict], config: dict) -> dict:
summary["by_scale"][scale]["validation_failed"] += 1
elif outcome == "generation_error":
summary["by_scale"][scale]["generation_errors"] += 1
+ elif outcome == "rate_limited":
+ summary["by_scale"][scale]["rate_limited"] += 1
elif outcome == "ineligible":
summary["by_scale"][scale]["ineligible"] += 1
elif outcome == "check_error":
diff --git a/src/conductor_eval/outcomes.py b/src/conductor_eval/outcomes.py
new file mode 100644
index 0000000..afab44f
--- /dev/null
+++ b/src/conductor_eval/outcomes.py
@@ -0,0 +1,13 @@
+"""Shared result outcome compatibility helpers."""
+
+
+def get_overall_status(result: dict) -> str:
+ """Return the persisted outcome, with fallbacks for legacy results."""
+ tests = result.get("tests", {})
+ if "overall_status" in tests:
+ return tests["overall_status"]
+ if result.get("error"):
+ return "generation_error"
+ if tests.get("overall_pass", False):
+ return "passed"
+ return "failed"
diff --git a/tests/test_analysis.py b/tests/test_analysis.py
index d2d457d..dff5160 100644
--- a/tests/test_analysis.py
+++ b/tests/test_analysis.py
@@ -5,6 +5,7 @@
from conductor_eval.analysis import (
_build_combined_html,
+ _format_pass_rate_summary,
_overall_pass_rate_percent,
build_chord_performance_by_model,
build_cost_by_model,
@@ -28,6 +29,30 @@
)
+def test_format_pass_rate_summary_uses_compact_outcome_text():
+ assert _format_pass_rate_summary(4, 8, []) == (
+ "50.0%",
+ "4 / 8 eligible passed",
+ "No generation errors",
+ "#2ecc71",
+ )
+ assert _format_pass_rate_summary(0, 0, [("generation_error", 1)]) == (
+ "N/A",
+ "No eligible generations",
+ "1 generation error",
+ "#5dade2",
+ )
+ assert _format_pass_rate_summary(1, 4, [("rate_limited", 2)])[:3] == (
+ "25.0%",
+ "1 / 4 eligible passed",
+ "2 rate limited",
+ )
+ assert (
+ _format_pass_rate_summary(1, 2, [("ineligible", 2), ("check_error", 1)])[2]
+ == "3 run exceptions"
+ )
+
+
def test_metric_charts_exclude_unknown_costs_and_failed_latencies():
df = pd.DataFrame(
[
@@ -386,6 +411,18 @@ def test_load_run_excludes_check_errors_from_overall_pass_rates(tmp_path):
assert not build_pass_rate_by_model(df).data
+def test_load_run_preserves_rate_limited_outcome(tmp_path):
+ run_path = _write_run(tmp_path, {"overall_status": "rate_limited"})
+
+ df, _, _ = load_run(run_path)
+ row = df.iloc[0]
+
+ assert row["overall_status"] == "rate_limited"
+ assert row["overall_pass"] is None
+ assert not row["overall_eligible"]
+ assert not build_pass_rate_by_model(df).data
+
+
def test_duration_adherence_groups_by_note_length_and_excludes_not_run_rows():
df = pd.DataFrame(
[
diff --git a/tests/test_conductor_eval_evaluator.py b/tests/test_conductor_eval_evaluator.py
index 57b74a6..c8928b6 100644
--- a/tests/test_conductor_eval_evaluator.py
+++ b/tests/test_conductor_eval_evaluator.py
@@ -1,15 +1,18 @@
import json
import logging
import threading
+import time
from datetime import datetime
from types import SimpleNamespace
import pytest
from conductor_core import GenerationRequest
+from conductor_core.errors import ProviderRateLimitError
from mido import Message, MidiFile
import conductor_eval.evaluator as evaluator_module
from conductor_eval import EvalEngineAdapter, Evaluator
+from conductor_eval.outcomes import get_overall_status
def test_texture_checks_are_available():
@@ -614,6 +617,7 @@ def fail_scale_check(*args):
({"overall_pass": False, "overall_status": "failed"}, True),
({"overall_pass": False, "overall_status": "ineligible"}, False),
({"overall_pass": False, "overall_status": "generation_error"}, False),
+ ({"overall_pass": False, "overall_status": "rate_limited"}, False),
({"overall_pass": False, "overall_status": "check_error"}, False),
({"overall_pass": False}, True),
],
@@ -622,9 +626,99 @@ def test_overall_eligibility_contract_includes_only_valid_verdicts(test_results,
assert Evaluator._is_overall_eligible(test_results) is expected
+@pytest.mark.parametrize(
+ ("result", "expected"),
+ [
+ ({"tests": {"overall_status": "rate_limited"}, "error": "throttled"}, "rate_limited"),
+ ({"tests": {}, "error": "provider failed"}, "generation_error"),
+ ({"tests": {"overall_pass": True}, "error": None}, "passed"),
+ ({"tests": {"overall_pass": False}, "error": None}, "failed"),
+ ({}, "failed"),
+ ],
+)
+def test_get_overall_status_preserves_status_and_supports_legacy_results(result, expected):
+ assert get_overall_status(result) == expected
+
+
+@pytest.mark.parametrize("value", [True, 0, -1, 1.5, "4"])
+def test_evaluate_rejects_invalid_cloud_concurrency_before_output(tmp_path, value):
+ output_dir = tmp_path / "evaluations"
+ evaluator = Evaluator(output_dir=output_dir)
+
+ with pytest.raises(ValueError, match="max_cloud_concurrency must be a positive integer"):
+ evaluator.evaluate(
+ prompts="melody",
+ roots=["C"],
+ models=[],
+ run_name="invalid-concurrency",
+ max_cloud_concurrency=value,
+ )
+
+ assert not output_dir.exists()
+
+
+def test_async_batch_caps_each_provider_independently(monkeypatch, tmp_path):
+ evaluator = Evaluator(output_dir=tmp_path / "evaluations")
+ active = {"OpenAI": 0, "Anthropic": 0}
+ peaks = {"OpenAI": 0, "Anthropic": 0}
+ total_active = 0
+ total_peak = 0
+ lock = threading.Lock()
+
+ def run_single(task, **kwargs):
+ nonlocal total_active, total_peak
+ provider = task["provider"]
+ with lock:
+ active[provider] += 1
+ total_active += 1
+ peaks[provider] = max(peaks[provider], active[provider])
+ total_peak = max(total_peak, total_active)
+ time.sleep(0.05)
+ with lock:
+ active[provider] -= 1
+ total_active -= 1
+ return {
+ "provider": provider,
+ "model": task["model"],
+ "root": "C",
+ "scale": "major",
+ "metrics": {},
+ "tests": {"overall_pass": True, "overall_status": "passed"},
+ "error": None,
+ }
+
+ monkeypatch.setattr(evaluator, "_run_single", run_single)
+ tasks = [
+ {"provider": provider, "model": f"model-{index}"}
+ for provider in active
+ for index in range(4)
+ ]
+
+ results = evaluator_module.asyncio.run(
+ evaluator._run_async_batch(
+ tasks,
+ tmp_path,
+ ["scale"],
+ logging.Logger("test"),
+ max_cloud_concurrency=2,
+ )
+ )
+
+ assert len(results) == 8
+ assert peaks == {"OpenAI": 2, "Anthropic": 2}
+ assert total_peak == 4
+
+
def test_summary_separates_all_outcomes_and_excludes_noneligible_results(tmp_path):
evaluator = Evaluator(output_dir=tmp_path / "evaluations")
- statuses = ["passed", "failed", "ineligible", "generation_error", "check_error"]
+ statuses = [
+ "passed",
+ "failed",
+ "ineligible",
+ "generation_error",
+ "rate_limited",
+ "check_error",
+ ]
results = [
{
"model": "model",
@@ -636,7 +730,13 @@ def test_summary_separates_all_outcomes_and_excludes_noneligible_results(tmp_pat
"overall_pass": status == "passed",
"overall_status": status,
},
- "error": "generation failed" if status == "generation_error" else None,
+ "error": (
+ "provider throttled"
+ if status == "rate_limited"
+ else "generation failed"
+ if status == "generation_error"
+ else None
+ ),
}
for status in statuses
]
@@ -648,13 +748,46 @@ def test_summary_separates_all_outcomes_and_excludes_noneligible_results(tmp_pat
assert summary["totals"]["validation_failed_generations"] == 1
assert summary["totals"]["ineligible_generations"] == 1
assert summary["totals"]["generation_error_generations"] == 1
+ assert summary["totals"]["rate_limited_generations"] == 1
assert summary["totals"]["check_error_generations"] == 1
assert summary["totals"]["overall_pass_rate"] == 0.5
assert summary["by_model"]["model"]["check_errors"] == 1
+ assert summary["by_model"]["model"]["rate_limited"] == 1
assert summary["by_root"]["C"]["check_errors"] == 1
assert summary["by_scale"]["major"]["check_errors"] == 1
+def test_rate_limit_error_is_persisted_as_distinct_outcome(monkeypatch, tmp_path):
+ class RateLimitedAdapter:
+ def __init__(self, output_dir):
+ self.output_dir = output_dir
+
+ def generate(self, **kwargs):
+ raise ProviderRateLimitError("OpenAI", "account rate exceeded")
+
+ monkeypatch.setattr(evaluator_module, "EvalEngineAdapter", RateLimitedAdapter)
+ evaluator = Evaluator(output_dir=tmp_path / "evaluations")
+ task = {
+ "provider": "OpenAI",
+ "model": "test-model",
+ "full_prompt": "prompt in C major",
+ "original_prompt": "prompt",
+ "root": "C",
+ "scale": "major",
+ "use_thinking": False,
+ "effort": None,
+ "variation_name": "standard",
+ "task_id": "task-rate-limit-0123456789abcdef-1",
+ }
+ run_path = tmp_path / "run"
+ run_path.mkdir()
+
+ result = evaluator._run_single(task, run_path, ["scale"])
+
+ assert result["tests"]["overall_status"] == "rate_limited"
+ assert result["tests"]["overall_pass"] is False
+
+
def test_generate_tasks_copies_test_params_to_each_task(tmp_path):
evaluator = Evaluator(output_dir=tmp_path / "evaluations")
test_params = {"polyphony": {"min_voices": 3}}