fix(eval): re-run infra-retryable verifier-errored tasks on resume

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",

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)})"

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(