diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f29c80f..7a078290 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Reports now include an `attempts` field listing the status, error, usage, and timeout of every attempt of a task repetition. `ResultLogger` writes it too. (PR: #71) + ### Changed +- tau2 and gaia2 `load_tasks()` now default to `max_retries=0`, matching the original benchmarks, which do not retry failed runs. (PR: #71) + ### Fixed +- `TaskProtocol.max_retries` and `TimeoutAction.RETRY`/`EXTEND` now take effect. Failures outside the agent's control (`setup_failed`, `environment_error`, `user_error`, `unknown_execution_error`) are retried up to `max_retries` times. Timeouts are retried once with the same (`RETRY`) or doubled (`EXTEND`) timeout. Previously both settings were silently ignored. `TaskProtocol` now rejects negative `max_retries`. (PR: #71) + ### Removed ## [0.5.1] - 2026-07-05 diff --git a/docs/guides/exception-handling.md b/docs/guides/exception-handling.md index 98a088c7..cd33e74f 100644 --- a/docs/guides/exception-handling.md +++ b/docs/guides/exception-handling.md @@ -190,7 +190,13 @@ The success rate (65.22%) reflects `60 / 92` rather than `60 / 100`. ## Rerunning Failed Tasks -Infrastructure errors are often transient. Tasks with infrastructure failures can be rerun: +Infrastructure errors are often transient. To retry them automatically, set `max_retries` on the task's protocol. This retries `setup_failed`, `environment_error`, `user_error`, and `unknown_execution_error` with the same seeds. Every attempt is listed in the report's `attempts` field. + +```python +task = Task(query="...", protocol=TaskProtocol(max_retries=2)) +``` + +Alternatively, tasks with infrastructure failures can be rerun manually: ```python results = benchmark.run(tasks) diff --git a/maseval/benchmark/gaia2/data_loader.py b/maseval/benchmark/gaia2/data_loader.py index 05224dbe..dafbe946 100644 --- a/maseval/benchmark/gaia2/data_loader.py +++ b/maseval/benchmark/gaia2/data_loader.py @@ -94,7 +94,7 @@ class Gaia2JudgeEngineConfig: # ARE scenarios/config.py:20: DEFAULT_SCENARIO_TIMEOUT = 1860 DEFAULT_TIMEOUT_SECONDS = 1860.0 # 31 minutes per task (matching ARE) -DEFAULT_MAX_RETRIES = 1 +DEFAULT_MAX_RETRIES = 0 # ARE does not retry failed scenarios # HuggingFace dataset info HF_DATASET_ID = "meta-agents-research-environments/gaia2" @@ -129,7 +129,7 @@ def load_tasks( limit: Maximum number of tasks to load (across all capabilities) timeout_seconds: Maximum execution time per task. Default 1860 (31 minutes, matching ARE's DEFAULT_SCENARIO_TIMEOUT). Set to None to disable timeout. - max_retries: Maximum retry attempts. Default 1 (skip on failure). + max_retries: Maximum retries after an infrastructure failure (see `TaskProtocol`). Default 0 (no retries). Returns: TaskQueue with Task objects. @@ -230,7 +230,7 @@ def _convert_gaia2_to_maseval( row: Raw row from HuggingFace dataset scenario: ARE BenchmarkScenario object timeout_seconds: Maximum execution time per task - max_retries: Maximum retry attempts + max_retries: Maximum retries after an infrastructure failure config_capability: The capability from the HuggingFace config name Returns: diff --git a/maseval/benchmark/tau2/data_loader.py b/maseval/benchmark/tau2/data_loader.py index e79ec8a1..c8cb78c8 100644 --- a/maseval/benchmark/tau2/data_loader.py +++ b/maseval/benchmark/tau2/data_loader.py @@ -39,7 +39,7 @@ # Default execution protocol settings DEFAULT_TIMEOUT_SECONDS = 600.0 # 10 minutes per task -DEFAULT_MAX_RETRIES = 1 # Skip on first failure +DEFAULT_MAX_RETRIES = 0 # tau2-bench v0.2.0 does not retry failed simulations # GitHub raw content URLs for v0.2.0 tag GITHUB_BASE = "https://raw.githubusercontent.com/sierra-research/tau2-bench" @@ -274,7 +274,7 @@ def load_tasks( limit: Maximum number of tasks to load timeout_seconds: Maximum execution time per task in seconds. Default 600 (10 minutes). Set to None to disable timeout. - max_retries: Maximum retry attempts for transient failures. Default 1 (skip on failure). + max_retries: Maximum retries after an infrastructure failure (see `TaskProtocol`). Default 0 (no retries). Returns: TaskQueue containing Task objects with: @@ -349,7 +349,7 @@ def _convert_tau2_task_to_maseval( split: Split name domain_config: Domain configuration with policy and db_path timeout_seconds: Maximum execution time per task in seconds - max_retries: Maximum retry attempts for transient failures + max_retries: Maximum retries after an infrastructure failure Returns: MASEval Task object diff --git a/maseval/core/benchmark.py b/maseval/core/benchmark.py index 619bd139..f71be979 100644 --- a/maseval/core/benchmark.py +++ b/maseval/core/benchmark.py @@ -8,7 +8,7 @@ import logging from .evaluator import Evaluator -from .task import Task, BaseTaskQueue, SequentialTaskQueue +from .task import Task, BaseTaskQueue, SequentialTaskQueue, TimeoutAction from .environment import Environment from .agent import AgentAdapter from .model import ModelAdapter @@ -68,6 +68,17 @@ class TaskExecutionStatus(Enum): TASK_EXECUTION_FAILED = "task_execution_failed" +# Failures outside the agent's control, retried up to `TaskProtocol.max_retries` times. +_RETRYABLE_STATUSES = frozenset( + { + TaskExecutionStatus.SETUP_FAILED, + TaskExecutionStatus.ENVIRONMENT_ERROR, + TaskExecutionStatus.USER_ERROR, + TaskExecutionStatus.UNKNOWN_EXECUTION_ERROR, + } +) + + class Benchmark(ABC): """Abstract base class for orchestrating multi-agent system execution and evaluation. @@ -552,7 +563,7 @@ def _build_report( Returns: Report dictionary with keys: ``task_id``, ``repeat_idx``, ``status``, - ``error``, ``traces``, ``config``, ``usage``, ``eval``, ``task``. + ``error``, ``traces``, ``config``, ``usage``, ``eval``, ``attempts``, ``task``. """ if status is not TaskExecutionStatus.SUCCESS and error is None: # Defensive: a non-success report must always carry error details. @@ -570,6 +581,7 @@ def _build_report( "config": config if config is not None else {}, "usage": usage, "eval": eval_results, + "attempts": [], # Filled in by _execute_task_repetition "task": { "query": task.query, "metadata": dict(task.metadata), @@ -1072,16 +1084,65 @@ def _execute_task_repetition( agent_data: Dict[str, Any], repeat_idx: int, ) -> Dict[str, Any]: - """Execute a single task repetition with timeout handling. + """Execute a single task repetition, retrying failed attempts as configured in ``task.protocol``. + + Called from both sequential and parallel execution paths. + + Args: + task: The task to execute. + agent_data: Agent configuration for this task. + repeat_idx: Repetition index (0 to n_task_repeats-1). + + Returns: + Report of the last attempt, with ``attempts`` listing every attempt. + """ + protocol = task.protocol + timeout_seconds = protocol.timeout_seconds + attempts: List[Dict[str, Any]] = [] + failure_retries = 0 + timeout_retried = False + + while True: + report = self._execute_task_attempt(task, agent_data, repeat_idx, timeout_seconds) + attempts.append( + { + "status": report["status"], + "error": report["error"], + "usage": report["usage"], + "timeout_seconds": timeout_seconds, + } + ) + status = TaskExecutionStatus(report["status"]) + + if status in _RETRYABLE_STATUSES and failure_retries < protocol.max_retries: + # Infrastructure failure: retry up to max_retries + failure_retries += 1 + elif status is TaskExecutionStatus.TASK_TIMEOUT and protocol.timeout_action is not TimeoutAction.SKIP and not timeout_retried: + # Timeout: retry once, EXTEND doubles the timeout + timeout_retried = True + if protocol.timeout_action is TimeoutAction.EXTEND and timeout_seconds is not None: + timeout_seconds *= 2 + else: + report["attempts"] = attempts + return report + + def _execute_task_attempt( + self, + task: Task, + agent_data: Dict[str, Any], + repeat_idx: int, + timeout_seconds: Optional[float], + ) -> Dict[str, Any]: + """Execute a single attempt of a task repetition with timeout handling. - This method encapsulates the complete execution of one task repetition, - including setup, execution, trace collection, and evaluation. It is - designed to be called from both sequential and parallel execution paths. + This method encapsulates the complete execution of one attempt, + including setup, execution, trace collection, and evaluation. Args: task: The task to execute. agent_data: Agent configuration for this task. repeat_idx: Repetition index (0 to n_task_repeats-1). + timeout_seconds: Timeout for this attempt, or None for no timeout. Returns: Report dictionary containing execution results. @@ -1097,8 +1158,7 @@ def _execute_task_repetition( agents_dict: Dict[str, AgentAdapter] = {} # Create execution context with optional timeout - timeout = task.protocol.timeout_seconds - context = TaskContext(deadline=timeout) + context = TaskContext(deadline=timeout_seconds) # Create scoped seed generator for this task+repetition task_seed_gen = self._seed_generator.for_task(str(task.id)).for_repetition(repeat_idx) diff --git a/maseval/core/callbacks/result_logger.py b/maseval/core/callbacks/result_logger.py index 1b681e47..bf25fcc0 100644 --- a/maseval/core/callbacks/result_logger.py +++ b/maseval/core/callbacks/result_logger.py @@ -170,6 +170,7 @@ def _filter_report(self, report: Dict) -> Dict: "repeat_idx": report.get("repeat_idx"), "status": report.get("status"), "error": report.get("error"), + "attempts": report.get("attempts"), } if self.include_traces and "traces" in report: diff --git a/maseval/core/task.py b/maseval/core/task.py index 1f96cd77..37d1eef9 100644 --- a/maseval/core/task.py +++ b/maseval/core/task.py @@ -19,11 +19,14 @@ class TimeoutAction(Enum): - """Action to take when a task timeout occurs.""" + """Action to take when a task timeout occurs. + + A timed-out task is retried at most once, independent of `TaskProtocol.max_retries`. + """ SKIP = "skip" # Mark as timed out, continue to next task RETRY = "retry" # Retry once with same timeout - EXTEND = "extend" # Double timeout and retry + EXTEND = "extend" # Double timeout and retry once @dataclass @@ -34,6 +37,10 @@ class TaskProtocol: task content (query, environment_data, etc.). It controls the interface between the task and MASEval's execution engine. + A retry reruns the full task repetition with the same seeds. Every attempt is + listed in the report's `attempts` field. With a benchmark `fail_on_*` flag set, + errors are raised without retrying. + Note: Timeout checking is cooperative and currently only occurs at execution phase boundaries (after setup, before execution, before evaluation). Timeout detection @@ -41,10 +48,15 @@ class TaskProtocol: Attributes: timeout_seconds: Maximum execution time for this task. None means no timeout. - timeout_action: Action to take when timeout occurs. - max_retries: Maximum retry attempts for transient failures (not timeouts). + timeout_action: Action to take when timeout occurs. See `TimeoutAction`. + max_retries: Maximum retries after a failure outside the agent's control (`setup_failed`, + `environment_error`, `user_error`, `unknown_execution_error`). Agent errors and + evaluation failures are never retried. priority: Execution priority (higher = sooner). Used by adaptive task queues. tags: Arbitrary tags for filtering or grouping tasks. + + Raises: + ValueError: If `max_retries` is negative. """ timeout_seconds: Optional[float] = None @@ -53,6 +65,10 @@ class TaskProtocol: priority: int = 0 tags: Dict[str, Any] = field(default_factory=dict) + def __post_init__(self) -> None: + if self.max_retries < 0: + raise ValueError(f"TaskProtocol.max_retries must be 0 or greater, got {self.max_retries}.") + def to_dict(self) -> Dict[str, Any]: """Convert to a JSON-serializable dictionary. diff --git a/tests/test_core/test_benchmark/test_report_schema.py b/tests/test_core/test_benchmark/test_report_schema.py index 6091013b..878f6f52 100644 --- a/tests/test_core/test_benchmark/test_report_schema.py +++ b/tests/test_core/test_benchmark/test_report_schema.py @@ -25,7 +25,7 @@ # Canonical top-level keys every report must carry. -REPORT_KEYS = {"task_id", "repeat_idx", "status", "error", "traces", "config", "usage", "eval", "task"} +REPORT_KEYS = {"task_id", "repeat_idx", "status", "error", "traces", "config", "usage", "eval", "attempts", "task"} ERROR_KEYS = {"error_type", "error_message", "traceback"} diff --git a/tests/test_core/test_benchmark/test_task_retries.py b/tests/test_core/test_benchmark/test_task_retries.py new file mode 100644 index 00000000..b325fd4c --- /dev/null +++ b/tests/test_core/test_benchmark/test_task_retries.py @@ -0,0 +1,242 @@ +"""Tests for task retries configured via ``TaskProtocol``. + +These tests verify that ``max_retries`` retries failures outside the agent's +control, that ``timeout_action`` retries timeouts, and that every attempt is +recorded in the report's ``attempts`` field. +""" + +import time + +import pytest + +from maseval import Task, TaskQueue +from maseval.core.exceptions import AgentError, EnvironmentError, UserError +from maseval.core.task import TaskProtocol, TimeoutAction +from conftest import DummyBenchmark, DummyModelAdapter + +MODEL_USAGE = {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15} + + +class FlakyBenchmark(DummyBenchmark): + """Benchmark whose attempts fail as scripted. + + Each attempt pops one delay (slept during setup) and one error (raised by + ``run_agents``). Each attempt also records its agent seed and calls a model once. + """ + + def __init__(self, errors=(), delays=(), **kwargs): + super().__init__(**kwargs) + self.errors = list(errors) + self.delays = list(delays) + self.agent_seeds = [] + + def setup_environment(self, agent_data, task, seed_generator): + if self.delays: + time.sleep(self.delays.pop(0)) + return super().setup_environment(agent_data, task, seed_generator) + + def setup_agents(self, agent_data, environment, task, user, seed_generator): + self.agent_seeds.append(seed_generator.derive_seed("agent")) + return super().setup_agents(agent_data, environment, task, user, seed_generator) + + def run_agents(self, agents, task, environment, query): + model = DummyModelAdapter(model_id="agent_model", usage=MODEL_USAGE) + self.register("models", "agent_model", model) + model.chat([{"role": "user", "content": query}]) + if self.errors: + raise self.errors.pop(0) + return super().run_agents(agents, task, environment, query) + + +class FlakySetupBenchmark(DummyBenchmark): + """Fails ``setup_environment`` for the first ``n_failures`` attempts.""" + + def __init__(self, n_failures, **kwargs): + super().__init__(**kwargs) + self.n_failures = n_failures + + def setup_environment(self, agent_data, task, seed_generator): + if self.n_failures > 0: + self.n_failures -= 1 + raise RuntimeError("setup boom") + return super().setup_environment(agent_data, task, seed_generator) + + +class _FailingEvaluator: + def filter_traces(self, traces): + return traces + + def __call__(self, traces, final_answer=None): + raise ValueError("eval boom") + + +class EvaluationFailureBenchmark(DummyBenchmark): + def setup_evaluators(self, environment, task, agents, user, seed_generator): + return [_FailingEvaluator()] + + +def _task(**protocol_kwargs): + return TaskQueue([Task(query="q", environment_data={}, protocol=TaskProtocol(**protocol_kwargs))]) + + +def _statuses(report): + return [attempt["status"] for attempt in report["attempts"]] + + +# -------------------------------------------------------------------------- +# max_retries +# -------------------------------------------------------------------------- + + +@pytest.mark.core +class TestFailureRetries: + """Tests for retries of failures outside the agent's control.""" + + @pytest.mark.parametrize("num_workers", [1, 2], ids=["sequential", "parallel"]) + def test_retries_until_success(self, num_workers): + """A failed attempt is retried and the report describes the successful one.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")] * 2, num_workers=num_workers) + + reports = benchmark.run(_task(max_retries=2), agent_data={"model": "test"}) + + assert len(reports) == 1 + assert reports[0]["status"] == "success" + assert reports[0]["error"] is None + assert _statuses(reports[0]) == ["environment_error", "environment_error", "success"] + + @pytest.mark.parametrize( + "error, status", + [ + (EnvironmentError("boom"), "environment_error"), + (UserError("boom"), "user_error"), + (RuntimeError("boom"), "unknown_execution_error"), + ], + ) + def test_infrastructure_failures_are_retried(self, error, status): + """Environment, user, and unclassified execution errors are retried.""" + benchmark = FlakyBenchmark(errors=[error]) + + reports = benchmark.run(_task(max_retries=1), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == [status, "success"] + + def test_setup_failure_is_retried(self): + """Setup failures are retried.""" + benchmark = FlakySetupBenchmark(n_failures=1) + + reports = benchmark.run(_task(max_retries=1), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["setup_failed", "success"] + + def test_agent_error_is_not_retried(self): + """Agent errors count against the agent and are never retried.""" + benchmark = FlakyBenchmark(errors=[AgentError("bad tool call")]) + + reports = benchmark.run(_task(max_retries=3), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["agent_error"] + assert len(benchmark.setup_environment_calls) == 1 + + def test_evaluation_failure_is_not_retried(self): + """Evaluation failures are never retried.""" + benchmark = EvaluationFailureBenchmark() + + reports = benchmark.run(_task(max_retries=3), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["evaluation_failed"] + assert len(benchmark.setup_environment_calls) == 1 + + def test_no_retries_by_default(self): + """With the default protocol a failed task runs once.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")]) + + reports = benchmark.run(_task(), agent_data={"model": "test"}) + + assert reports[0]["status"] == "environment_error" + assert _statuses(reports[0]) == ["environment_error"] + + def test_gives_up_after_max_retries(self): + """Once retries are used up, the report describes the last failed attempt.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")] * 3) + + reports = benchmark.run(_task(max_retries=1), agent_data={"model": "test"}) + + assert reports[0]["status"] == "environment_error" + assert "env down" in reports[0]["error"]["error_message"] + assert _statuses(reports[0]) == ["environment_error", "environment_error"] + + def test_fail_on_task_error_raises_without_retry(self): + """Fail-fast flags raise on the first failure.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")], fail_on_task_error=True) + + with pytest.raises(EnvironmentError, match="env down"): + benchmark.run(_task(max_retries=3), agent_data={"model": "test"}) + + assert len(benchmark.setup_environment_calls) == 1 + + def test_attempts_reuse_seeds(self): + """Every attempt of a repetition derives the same seeds.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")], seed=42) + + benchmark.run(_task(max_retries=1), agent_data={"model": "test"}) + + assert len(benchmark.agent_seeds) == 2 + assert benchmark.agent_seeds[0] is not None + assert benchmark.agent_seeds[0] == benchmark.agent_seeds[1] + + def test_attempts_record_usage(self): + """Each attempt carries its own usage, so failed attempts are not lost from cost totals.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")]) + + reports = benchmark.run(_task(max_retries=1), agent_data={"model": "test"}) + + for attempt in reports[0]["attempts"]: + assert attempt["usage"]["models"]["agent_model"]["input_tokens"] == 10 + assert benchmark.usage.to_dict()["input_tokens"] == 20 + + +# -------------------------------------------------------------------------- +# timeout_action +# -------------------------------------------------------------------------- + + +@pytest.mark.core +class TestTimeoutRetries: + """Tests for retries of timed-out tasks.""" + + def test_skip_does_not_retry(self): + """SKIP records the timeout without retrying.""" + benchmark = FlakyBenchmark(delays=[0.05]) + + reports = benchmark.run(_task(timeout_seconds=0.01, timeout_action=TimeoutAction.SKIP), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["task_timeout"] + + def test_retry_retries_once_with_same_timeout(self): + """RETRY reruns a timed-out task once with the same timeout.""" + benchmark = FlakyBenchmark(delays=[0.05, 0.05]) + + reports = benchmark.run(_task(timeout_seconds=0.01, timeout_action=TimeoutAction.RETRY), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["task_timeout", "task_timeout"] + assert [a["timeout_seconds"] for a in reports[0]["attempts"]] == [0.01, 0.01] + + def test_extend_retries_once_with_double_timeout(self): + """EXTEND reruns a timed-out task once with double the timeout.""" + benchmark = FlakyBenchmark(delays=[0.05, 0.05]) + + reports = benchmark.run(_task(timeout_seconds=0.01, timeout_action=TimeoutAction.EXTEND), agent_data={"model": "test"}) + + assert _statuses(reports[0]) == ["task_timeout", "task_timeout"] + assert [a["timeout_seconds"] for a in reports[0]["attempts"]] == [0.01, 0.02] + + def test_timeout_retry_does_not_use_max_retries(self): + """Timeout retries and failure retries have separate budgets.""" + benchmark = FlakyBenchmark(errors=[EnvironmentError("env down")], delays=[0.2]) + + reports = benchmark.run( + _task(timeout_seconds=0.05, timeout_action=TimeoutAction.RETRY, max_retries=1), + agent_data={"model": "test"}, + ) + + assert _statuses(reports[0]) == ["task_timeout", "environment_error", "success"] diff --git a/tests/test_core/test_callbacks/test_result_logger.py b/tests/test_core/test_callbacks/test_result_logger.py index b2af93c7..378b4428 100644 --- a/tests/test_core/test_callbacks/test_result_logger.py +++ b/tests/test_core/test_callbacks/test_result_logger.py @@ -135,18 +135,20 @@ def test_filter_report_all_included(self): assert "eval" in filtered def test_filter_report_preserves_status_and_error(self): - """Test that status and error fields are always included in filtered reports. + """Test that status, error, and attempts fields are always included in filtered reports. These are core metadata fields (like task_id and repeat_idx) that must always be present so persisted results can distinguish successes from failures. """ logger = MockResultLogger(include_traces=False, include_config=False, include_eval=False) + attempts = [{"status": "agent_error", "error": {"type": "AgentError"}, "usage": None, "timeout_seconds": None}] report = { "task_id": "task_0", "repeat_idx": 0, "status": "agent_error", "error": {"type": "AgentError", "message": "Tool call failed"}, + "attempts": attempts, "traces": {"agent": "trace_data"}, "config": {"model": "gpt-4"}, "eval": {"score": 0.0}, @@ -156,6 +158,7 @@ def test_filter_report_preserves_status_and_error(self): assert filtered["status"] == "agent_error" assert filtered["error"] == {"type": "AgentError", "message": "Tool call failed"} + assert filtered["attempts"] == attempts assert "traces" not in filtered assert "config" not in filtered assert "eval" not in filtered diff --git a/tests/test_core/test_task_protocol.py b/tests/test_core/test_task_protocol.py index 291feaa0..bc3b6f8f 100644 --- a/tests/test_core/test_task_protocol.py +++ b/tests/test_core/test_task_protocol.py @@ -58,6 +58,11 @@ def test_custom_values(self): assert protocol.priority == 10 assert protocol.tags == {"category": "hard", "group": "A"} + def test_negative_max_retries_raises(self): + """Negative max_retries should be rejected.""" + with pytest.raises(ValueError, match="max_retries"): + TaskProtocol(max_retries=-1) + def test_tags_isolation(self): """Tags dict should be independent per instance.""" p1 = TaskProtocol()