diff --git a/src/benchflow/_utils/scoring.py b/src/benchflow/_utils/scoring.py index 87491266d..0bb3f0d04 100644 --- a/src/benchflow/_utils/scoring.py +++ b/src/benchflow/_utils/scoring.py @@ -186,6 +186,8 @@ def _looks_like_infra_error(error: str) -> bool: # Predates the marker above and is kept for strings rebuilt # outside that boundary (e.g. a verifier error re-raised as text). "failed to get session command", + "failed to execute session command", + "command timed out after", "sandbox not found", "workspace not found", "api connection", diff --git a/src/benchflow/evaluation.py b/src/benchflow/evaluation.py index 277a5b987..1c96c1f41 100644 --- a/src/benchflow/evaluation.py +++ b/src/benchflow/evaluation.py @@ -1084,31 +1084,60 @@ def _get_task_dirs(self) -> list[Path]: def _get_completed_tasks(self) -> dict[str, dict]: """Load tasks that already have results with rewards or verifier errors. + Scoreless results whose verifier error is infra-retryable (per + ``RetryConfig.should_retry_verifier_error``) are not reused: the + verifier never scored the frozen workspace, so the task re-runs. + Scoped to the current job directory (``_jobs_dir / _job_name``) to - prevent cross-job contamination. When multiple result.json files - exist for the same task (retry artifacts), the newest by mtime wins. + prevent cross-job contamination. When multiple result.json files exist + for the same task (retry artifacts), a scored result always wins over + a scoreless verifier error; otherwise the newest artifact wins. Guards ENG-160: orphan retry artifacts no longer pollute resume. """ job_dir = self._jobs_dir / self._job_name if not job_dir.exists(): return {} - # Collect every result keyed by (task_name) → keep newest by mtime. - best: dict[str, tuple[float, dict]] = {} + # A completed score is durable evidence and must not be displaced by a + # newer scoreless retry artifact. Within the same scored/unscored tier, + # prefer recency and use the path as a deterministic tie-breaker. + best: dict[str, tuple[tuple[bool, float, str], dict]] = {} for rfile in job_dir.rglob("result.json"): try: r = json.loads(rfile.read_text()) task = r["task_name"] if r.get("rewards") is not None or r.get("verifier_error"): - mtime = rfile.stat().st_mtime + rank = ( + r.get("rewards") is not None, + rfile.stat().st_mtime, + str(rfile), + ) prev = best.get(task) - if prev is None or (mtime, str(rfile)) >= (prev[0], ""): - best[task] = (mtime, r) + if prev is None or rank >= prev[0]: + best[task] = (rank, r) except Exception as e: logger.debug(f"Skipping corrupt result file {rfile}: {e}") completed: dict[str, dict] = {} - for task, (_mt, r) in best.items(): + # Re-running an errored task is only safe when rollouts are + # independent. A sequential-shared job advances one persisted learner + # state in task order, so replaying an earlier task after later tasks + # committed their skills would corrupt the learning curve; there the + # errored result stays reused, matching the pre-existing behavior. + rerun_ok = self._config.job_mode != "sequential-shared" + for task, (_rank, r) in best.items(): if r.get("verifier_error"): + # A scoreless result whose verifier error is infra-retryable + # (same taxonomy as the within-run retry) records no signal + # about the task; reusing it pins a lost score forever. + retryable = self._config.retry.should_retry_verifier_error( + r["verifier_error"] + ) + if rerun_ok and r.get("rewards") is None and retryable: + logger.info( + f"Re-running verifier-errored task on resume: {task} " + f"({truncate_end(r['verifier_error'], 80)})" + ) + continue logger.info( f"Reusing completed verifier-errored task on resume: {task} " f"({truncate_end(r['verifier_error'], 80)})" diff --git a/tests/test_verify.py b/tests/test_verify.py index b82831e21..d3821839b 100644 --- a/tests/test_verify.py +++ b/tests/test_verify.py @@ -4,6 +4,7 @@ import contextlib import json import logging +import os from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch @@ -60,6 +61,20 @@ def test_classify_verifier_error_substring_order(): assert classify_verifier_error(msg) == VERIFIER_FAILED +def test_pr_1063_daytona_verifier_exec_errors_are_infra(): + """Guards PR #1063: historical Daytona exec failures must be retryable.""" + assert ( + classify_verifier_error( + "verifier crashed: Failed to execute session command: (no detail)" + ) + == VERIFIER_INFRA + ) + assert ( + classify_verifier_error("verifier crashed: Command timed out after 180 seconds") + == VERIFIER_INFRA + ) + + # RunResult with verifier_error @@ -489,11 +504,75 @@ async def test_agent_error_still_retries(self, job_factory): class TestResume: - def test_verifier_errored_is_complete(self, tmp_path, caplog): - """Guards the PR #819 fix for issue #542's misleading resume log.""" + def test_infra_verifier_errored_reruns(self, tmp_path, caplog): + """Guards PR #1063: retryable verifier infra must not pin a lost score.""" task_dir = tmp_path / "task1" / "trial-1" task_dir.mkdir(parents=True) (task_dir / "result.json").write_text( + json.dumps( + { + "task_name": "task1", + "rewards": None, + "error": None, + "verifier_error": ( + "verifier crashed: Failed to execute session command: " + "(no detail)" + ), + } + ) + ) + from benchflow.evaluation import Evaluation, EvaluationConfig + + job = Evaluation( + tasks_dir=tmp_path, jobs_dir=tmp_path, config=EvaluationConfig() + ) + with caplog.at_level(logging.INFO): + completed = job._get_completed_tasks() + assert "task1" not in completed + assert any("Re-running verifier-errored task" in m for m in caplog.messages) + + def test_sequential_shared_reuses_infra_verifier_error(self, tmp_path): + """Guards PR #1063: sequential resume must preserve its learning curve.""" + task_dir = tmp_path / "task1" / "trial-1" + task_dir.mkdir(parents=True) + (task_dir / "result.json").write_text( + json.dumps( + { + "task_name": "task1", + "rewards": None, + "error": None, + "verifier_error": "verifier timed out after 900s", + } + ) + ) + from benchflow.evaluation import Evaluation, EvaluationConfig + + job = Evaluation( + tasks_dir=tmp_path, + jobs_dir=tmp_path, + config=EvaluationConfig(job_mode="sequential-shared"), + ) + assert "task1" in job._get_completed_tasks() + + def test_pr_1063_older_scored_result_beats_newer_infra_error(self, tmp_path): + """Guards PR #1063: a later failed retry cannot erase a valid score.""" + scored_dir = tmp_path / "task1" / "trial-1" + retry_dir = tmp_path / "task1" / "trial-1-retry-1" + scored_dir.mkdir(parents=True) + retry_dir.mkdir(parents=True) + scored_path = scored_dir / "result.json" + retry_path = retry_dir / "result.json" + scored_path.write_text( + json.dumps( + { + "task_name": "task1", + "rewards": {"reward": 1.0}, + "error": None, + "verifier_error": None, + } + ) + ) + retry_path.write_text( json.dumps( { "task_name": "task1", @@ -503,6 +582,31 @@ def test_verifier_errored_is_complete(self, tmp_path, caplog): } ) ) + os.utime(scored_path, (1, 1)) + os.utime(retry_path, (2, 2)) + + from benchflow.evaluation import Evaluation, EvaluationConfig + + job = Evaluation( + tasks_dir=tmp_path, jobs_dir=tmp_path, config=EvaluationConfig() + ) + completed = job._get_completed_tasks() + assert completed["task1"]["rewards"] == {"reward": 1.0} + + def test_contract_verifier_errored_is_complete(self, tmp_path, caplog): + """Guards the PR #819 fix for issue #542's misleading resume log.""" + task_dir = tmp_path / "task1" / "trial-1" + task_dir.mkdir(parents=True) + (task_dir / "result.json").write_text( + json.dumps( + { + "task_name": "task1", + "rewards": None, + "error": None, + "verifier_error": "verifier crashed: No reward file found", + } + ) + ) from benchflow.evaluation import Evaluation, EvaluationConfig job = Evaluation(