fix(metrics): guard reward reads against a non-dict rewards field

src/benchflow/metrics.py

@@ -16,6 +16,7 @@
     classify_error,
     classify_score_outcome,
     classify_verifier_error,
+    extract_reward,
     pass_rate,
     pass_rate_excl_errors,
 )
@@ -318,13 +319,15 @@ def summary(self) -> dict[str, Any]:
         }
 
 
-def _safe_reward(rewards: dict) -> float:
+def _safe_reward(rewards: Any) -> float:
     """Extract reward value from a rewards dict, defaulting to 0 if None/missing.
 
     Prevents TypeError when comparing reward values where one is None
-    (e.g. rewards={"reward": None, "rubric": [...]}).
+    (e.g. rewards={"reward": None, "rubric": [...]}), and AttributeError when
+    the persisted ``rewards`` is not a mapping at all — the resume path feeds
+    raw json.loads payloads with no shape validation.
     """
-    val = rewards.get("reward")
+    val = rewards.get("reward") if isinstance(rewards, dict) else None
     return val if isinstance(val, (int, float)) else 0.0
 
 
@@ -361,7 +364,7 @@ def collect_metrics(
 
     tasks = []
     for task_name, r in sorted(best.items()):
-        reward = r.get("rewards", {}).get("reward") if r.get("rewards") else None
+        reward = extract_reward(r)
         # Calculate duration
         duration = 0.0
         try:

src/benchflow/skill_eval/_core.py

@@ -738,7 +738,7 @@ async def _run_job(
                     try:
                         result_data = json.loads(result_file.read_text())
                         rewards = result_data.get("rewards")
-                        if rewards:
+                        if isinstance(rewards, dict):
                             reward = rewards.get(
                                 "reward", next(iter(rewards.values()), None)
                             )

tests/test_metrics.py

@@ -232,6 +232,68 @@ def test_collect_metrics_corrupt_file(tmp_path):
     assert s["total"] == 0
 
 
+@pytest.mark.parametrize("rewards", [1.0, 1, True, [1.0], "1.0"])
+def test_collect_metrics_non_dict_rewards_counts_as_errored(tmp_path, rewards):
+    """A persisted non-dict `rewards` must not crash the aggregation.
+
+    Results are raw json.loads payloads with no shape validation, so `rewards`
+    can be any JSON value. Reading it as a mapping raised AttributeError and
+    killed `bench eval metrics` on the whole results dir. `extract_reward`
+    treats a non-mapping as no reward, which lands the task in the errored
+    bucket — the same reading `_classify_completed_outcomes` applies to the
+    identical payload.
+    """
+    trial = tmp_path / "job" / "task-a__abc"
+    trial.mkdir(parents=True)
+    (trial / "result.json").write_text(
+        json.dumps({"task_name": "task-a", "rewards": rewards})
+    )
+
+    metrics = collect_metrics(str(tmp_path))
+    s = metrics.summary()
+
+    assert metrics.tasks[0].reward is None
+    assert s["total"] == 1
+    assert s["errored"] == 1
+    assert s["errored_tasks"] == ["task-a"]
+
+
+def test_collect_metrics_non_dict_rewards_does_not_shadow_a_scored_retry(tmp_path):
+    """A malformed artifact must not win the best-result pick.
+
+    The selection step compares both artifacts' rewards; reading a non-dict
+    `rewards` raised inside the enclosing `except Exception`, which dropped the
+    *well-formed* retry for the same task and reported it as errored. Mirrors
+    test_scoring.py::test_malformed_rewards_shape_tolerated, which pins the
+    same invariant for mean_scored_reward.
+    """
+    job = tmp_path / "job"
+    for name, rewards, n_tool_calls in (
+        ("task-a__aaa", 1.0, 99),  # malformed, sorts first
+        ("task-a__zzz", {"reward": 1.0}, 7),
+    ):
+        trial = job / name
+        trial.mkdir(parents=True)
+        (trial / "result.json").write_text(
+            json.dumps(
+                {
+                    "task_name": "task-a",
+                    "rewards": rewards,
+                    "n_tool_calls": n_tool_calls,
+                }
+            )
+        )
+
+    metrics = collect_metrics(str(tmp_path))
+    s = metrics.summary()
+
+    assert s["total"] == 1
+    assert s["passed"] == 1
+    assert metrics.tasks[0].reward == 1.0
+    # The well-formed artifact is the one kept, not just its reward.
+    assert metrics.tasks[0].n_tool_calls == 7
+
+
 def test_collect_metrics_metadata(results_dir):
     """Test that benchmark/agent/model metadata is passed through."""
     metrics = collect_metrics(

tests/test_skill_eval.py

@@ -593,6 +593,52 @@ async def fake_run(self):
         assert collected["case-1"].reward == 0.0
         assert collected["case-10"].reward == 1.0
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("rewards", [1.0, 1, True, [1.0], "1.0"])
+    async def test_non_dict_rewards_does_not_crash_collection(
+        self, skill_dir, tmp_path, monkeypatch, rewards
+    ):
+        """A persisted non-dict `rewards` must not abort `bench skills eval`.
+
+        Same failure mode as collect_metrics: the rollout's result.json is a
+        raw json.loads payload, and reading `rewards` as a mapping raised
+        AttributeError — which the enclosing
+        `except (json.JSONDecodeError, KeyError)` does not catch.
+        """
+        from benchflow.evaluation import EvaluationResult
+
+        async def fake_run(self):
+            rollout_dir = self._jobs_dir / "2026-09-04__10-00-00" / "calc-001__abc123"
+            rollout_dir.mkdir(parents=True)
+            (rollout_dir / "result.json").write_text(
+                json.dumps(
+                    {
+                        "task_name": "calc-001",
+                        "rewards": rewards,
+                        "n_tool_calls": 4,
+                    }
+                )
+            )
+            return EvaluationResult(job_name="fake", config=self._config, total=1)
+
+        monkeypatch.setattr("benchflow.evaluation.Evaluation.run", fake_run)
+
+        evaluator = SkillEvaluator(skill_dir)
+        results = await evaluator._run_job(
+            tasks_dir=tmp_path / "tasks",
+            agent="gemini",
+            model="",
+            environment="docker",
+            jobs_dir=str(tmp_path / "jobs"),
+            concurrency=1,
+            with_skill=True,
+        )
+
+        collected = {result.case_id: result for result in results}
+        assert collected["calc-001"].reward is None
+        assert collected["calc-001"].error is None
+        assert collected["calc-001"].n_tool_calls == 4
+
 
 class TestGepaExport:
     def test_exports_structure(self, skill_dir, tmp_path):