feat: add could-not-evaluate status so non-gradable cases are excluded from scores

src/strands_evals/evaluators/evaluator.py

@@ -6,7 +6,7 @@
 from typing_extensions import Any, Generic, TypeGuard
 
 from ..extractors import TraceExtractor
-from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
+from ..types.evaluation import GRADED, EvaluationData, EvaluationOutput, InputT, OutputT
 from ..types.trace import (
     AssistantMessage,
     Context,
@@ -79,13 +79,24 @@ def _get_model_id(self, model: Model | str | None) -> str:
 
     @staticmethod
     def _default_aggregator(outputs: list[EvaluationOutput]) -> tuple[float, bool, str]:
-        # Handle empty outputs list to avoid division by zero
-        if not outputs:
-            return (0.0, False, "No evaluation outputs produced")
-
-        avg_score = sum(o.score for o in outputs) / len(outputs)
-        all_pass = all(o.test_pass for o in outputs)
-        combined_reason = " | ".join(o.reason for o in outputs if o.reason)
+        # Only "graded" outputs contribute to the score/pass aggregates. Outputs
+        # marked "could_not_evaluate" or "informational" are excluded so a
+        # non-gradable result (e.g. a harness overflow, or an evaluator whose
+        # preconditions weren't met) doesn't silently drag the average up or down.
+        graded = [o for o in outputs if o.status == GRADED]
+
+        # No gradable outputs (empty list, or every output was non-graded): there
+        # is nothing to score. Report it as a non-failure so a fully-skipped
+        # evaluator doesn't read as a quality failure.
+        if not graded:
+            if not outputs:
+                return (0.0, False, "No evaluation outputs produced")
+            combined_reason = " | ".join(o.reason for o in outputs if o.reason)
+            return (0.0, False, combined_reason or "No gradable evaluation outputs produced")
+
+        avg_score = sum(o.score for o in graded) / len(graded)
+        all_pass = all(o.test_pass for o in graded)
+        combined_reason = " | ".join(o.reason for o in graded if o.reason)
         return avg_score, all_pass, combined_reason
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:

src/strands_evals/experiment.py

@@ -45,7 +45,7 @@
 from .telemetry import get_tracer, serialize
 from .telemetry._cloudwatch_logger import _send_to_cloudwatch
 from .types.detector import DiagnosisConfig
-from .types.evaluation import EvaluationData, InputT, OutputT
+from .types.evaluation import COULD_NOT_EVALUATE, GRADED, EvaluationData, EvaluationOutput, InputT, OutputT
 from .types.evaluation_report import EvaluationReport
 from .types.trace import Session
 from .utils import is_throttling_error
@@ -83,6 +83,19 @@ def _get_label_from_score(evaluator: Evaluator, score: float) -> str:
     return "YES" if score >= 0.5 else "NO"
 
 
+def _aggregate_status(outputs: list[EvaluationOutput]) -> str:
+    """Roll a list of per-output statuses up to a single evaluator-level status.
+
+    The evaluator result is "graded" if at least one output was graded; if every
+    output was non-graded, the whole result is non-gradable and is reported as
+    "could_not_evaluate" so downstream aggregates exclude it. Mirrors the
+    graded-only filter in ``Evaluator._default_aggregator``.
+    """
+    if any(o.status == GRADED for o in outputs):
+        return GRADED
+    return COULD_NOT_EVALUATE
+
+
 class Experiment(Generic[InputT, OutputT]):
     """
     An evaluation experiment containing test cases and evaluators.
@@ -421,6 +434,7 @@ async def _evaluate_with_retry(evaluator=evaluator, evaluation_context=evaluatio
                     "test_pass": aggregate_pass,
                     "score": aggregate_score,
                     "reason": aggregate_reason or "",
+                    "status": _aggregate_status(evaluation_outputs),
                     "detailed_results": evaluation_outputs,
                 }
 
@@ -442,16 +456,21 @@ async def _evaluate_with_retry(evaluator=evaluator, evaluation_context=evaluatio
                 "test_pass": False,
                 "score": 0,
                 "reason": f"Evaluator error: {str(original_exception)}",
+                "status": COULD_NOT_EVALUATE,
                 "detailed_results": [],
             }
         except Exception as e:
-            # Catch non-throttling errors and record as failure (error isolation)
+            # A harness error (e.g. context-window overflow) is a capability
+            # failure of the evaluation, not a quality signal about the agent.
+            # Mark it could_not_evaluate so it is excluded from aggregates rather
+            # than silently recorded as a score-0 quality failure.
             return {
                 "evaluator_name": evaluator.get_name(),
                 "evaluator_type": evaluator.get_type_name(),
                 "test_pass": False,
                 "score": 0,
                 "reason": f"Evaluator error: {str(e)}",
+                "status": COULD_NOT_EVALUATE,
                 "detailed_results": [],
             }
 
@@ -656,6 +675,7 @@ async def run_evaluations_async(
                 "test_passes": [],
                 "cases": [],
                 "reasons": [],
+                "statuses": [],
                 "detailed_results": [],
                 "diagnoses": [],
                 "recommendations": [],
@@ -669,12 +689,21 @@ async def run_evaluations_async(
             recommendation = result.get("recommendation")
             for eval_result in result["evaluator_results"]:
                 eval_name = eval_result["evaluator_name"]
+                # Default to GRADED for backward compatibility with results that
+                # predate the status field (e.g. custom evaluator result dicts).
+                status = eval_result.get("status", GRADED)
                 evaluator_data[eval_name]["cases"].append(
-                    {**case_data, "evaluator": eval_name, "evaluator_type": eval_result["evaluator_type"]}
+                    {
+                        **case_data,
+                        "evaluator": eval_name,
+                        "evaluator_type": eval_result["evaluator_type"],
+                        "status": status,
+                    }
                 )
                 evaluator_data[eval_name]["scores"].append(eval_result["score"])
                 evaluator_data[eval_name]["test_passes"].append(eval_result["test_pass"])
                 evaluator_data[eval_name]["reasons"].append(eval_result["reason"])
+                evaluator_data[eval_name]["statuses"].append(status)
                 evaluator_data[eval_name]["detailed_results"].append(eval_result["detailed_results"])
                 evaluator_data[eval_name]["diagnoses"].append(diagnosis)
                 evaluator_data[eval_name]["recommendations"].append(recommendation)
@@ -684,8 +713,12 @@ async def run_evaluations_async(
             eval_name = evaluator.get_name()
             data = evaluator_data[eval_name]
             scores = data["scores"]
+            statuses = data["statuses"]
+            # overall_score reflects only gradable cases: a could_not_evaluate /
+            # informational row must not drag the aggregate up or down.
+            graded_scores = [s for s, st in zip(scores, statuses, strict=True) if st == GRADED]
             report = EvaluationReport(
-                overall_score=sum(scores) / len(scores) if scores else 0,
+                overall_score=sum(graded_scores) / len(graded_scores) if graded_scores else 0,
                 scores=scores,
                 test_passes=data["test_passes"],
                 cases=data["cases"],

src/strands_evals/types/evaluation.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing_extensions import Any, Generic, TypedDict, TypeVar
+from typing_extensions import Any, Generic, Literal, TypedDict, TypeVar
 
 from .trace import Session
 
@@ -109,6 +109,14 @@ class EvaluationData(BaseModel, Generic[InputT, OutputT]):
     expected_environment_state: list[EnvironmentState] | None = None
 
 
+# Evaluation status values for EvaluationOutput.status.
+# Kept as plain strings (not an Enum) so new values can be added without a
+# breaking change; a Literal annotation documents the currently-defined set.
+GRADED = "graded"
+COULD_NOT_EVALUATE = "could_not_evaluate"
+INFORMATIONAL = "informational"
+
+
 class EvaluationOutput(BaseModel):
     """
     Structured output for LLM-based judge.
@@ -118,9 +126,20 @@ class EvaluationOutput(BaseModel):
         test_pass: Whether the test pass or fail.
         reason: The reason for the score for each test case.
         label: The categorical label corresponding to the score.
+        status: Whether this output is a real verdict and should count toward
+            aggregates. One of:
+              - "graded" (default): score/test_pass are real verdicts.
+              - "could_not_evaluate": the evaluator tried but could not grade
+                this case (preconditions not met, missing data, harness error);
+                excluded from pass-rate and score aggregates.
+              - "informational": surfaces content for human review and never
+                counts toward a numeric aggregate.
+            Defaults to "graded", so existing evaluators and consumers are
+            unaffected.
     """
 
     score: float
     test_pass: bool
     reason: str | None = None
     label: str | None = None
+    status: Literal["graded", "could_not_evaluate", "informational"] = GRADED

src/strands_evals/types/evaluation_report.py

@@ -4,7 +4,7 @@
 from pydantic import BaseModel
 
 from ..display.display_console import CollapsibleTableReportDisplay
-from ..types.evaluation import EvaluationOutput
+from ..types.evaluation import GRADED, EvaluationOutput
 
 
 class EvaluationReport(BaseModel):
@@ -52,8 +52,13 @@ def flatten(cls, reports: list["EvaluationReport"]) -> "EvaluationReport":
                 diags.append(report.diagnoses[i] if i < len(report.diagnoses) else None)
                 recs.append(report.recommendations[i] if i < len(report.recommendations) else None)
 
+        # overall_score reflects only gradable cases. Each case dict carries a
+        # "status" (defaulting to "graded" for rows that predate the field), so a
+        # could_not_evaluate / informational row is excluded from the aggregate.
+        graded_scores = [s for s, case in zip(scores, cases, strict=False) if case.get("status", GRADED) == GRADED]
+
         return cls(
-            overall_score=sum(scores) / len(scores) if scores else 0.0,
+            overall_score=sum(graded_scores) / len(graded_scores) if graded_scores else 0.0,
             scores=scores,
             cases=cases,
             test_passes=passes,

tests/strands_evals/evaluators/test_evaluator.py

@@ -411,3 +411,61 @@ def test_format_tools_multiple_tools(self):
         result = self.evaluator._format_tools(tools)
         expected = "- tool_a: First tool\n- tool_b: Second tool\n  Parameters:\n    - x (number (required)): A number"
         assert result == expected
+
+
+class TestDefaultAggregatorStatus:
+    """_default_aggregator excludes non-graded outputs from score/pass aggregates."""
+
+    def test_all_graded_averages_as_before(self):
+        """With no status set (defaults to graded), behavior is unchanged."""
+        outputs = [
+            EvaluationOutput(score=1.0, test_pass=True, reason="a"),
+            EvaluationOutput(score=0.0, test_pass=False, reason="b"),
+        ]
+        score, all_pass, reason = Evaluator._default_aggregator(outputs)
+        assert score == 0.5
+        assert all_pass is False
+        assert reason == "a | b"
+
+    def test_could_not_evaluate_excluded_from_average(self):
+        """A could_not_evaluate row must not drag the average down."""
+        outputs = [
+            EvaluationOutput(score=1.0, test_pass=True, reason="graded"),
+            EvaluationOutput(score=0.0, test_pass=False, reason="skip", status="could_not_evaluate"),
+        ]
+        score, all_pass, _ = Evaluator._default_aggregator(outputs)
+        # Only the single graded 1.0 counts — not (1.0 + 0.0) / 2.
+        assert score == 1.0
+        assert all_pass is True
+
+    def test_informational_excluded_from_average(self):
+        """An informational row never counts toward the numeric aggregate."""
+        outputs = [
+            EvaluationOutput(score=0.6, test_pass=True, reason="graded"),
+            EvaluationOutput(score=1.0, test_pass=True, reason="fyi", status="informational"),
+        ]
+        score, _, _ = Evaluator._default_aggregator(outputs)
+        assert score == 0.6
+
+    def test_all_non_graded_returns_non_failure(self):
+        """When every output is non-graded there is nothing to score.
+
+        It must not read as a quality failure: score 0.0 / pass False is the
+        signal, but the reason makes clear it's could-not-evaluate, and the
+        experiment layer rolls this up to a could_not_evaluate status that
+        downstream aggregates exclude.
+        """
+        outputs = [
+            EvaluationOutput(score=0.0, test_pass=False, reason="no errors to grade", status="could_not_evaluate"),
+        ]
+        score, all_pass, reason = Evaluator._default_aggregator(outputs)
+        assert score == 0.0
+        assert all_pass is False
+        assert reason == "no errors to grade"
+
+    def test_empty_outputs_unchanged(self):
+        """Empty list keeps its original sentinel result."""
+        score, all_pass, reason = Evaluator._default_aggregator([])
+        assert score == 0.0
+        assert all_pass is False
+        assert reason == "No evaluation outputs produced"

tests/strands_evals/test_experiment.py

@@ -1161,6 +1161,40 @@ def echo_task(c):
     assert "Evaluator error" in report.reasons[throwing_idx]
     assert "Evaluator exploded" in report.reasons[throwing_idx]
 
+    # A harness error is a capability failure, not a quality signal: the row is
+    # tagged could_not_evaluate so downstream aggregates exclude it. The
+    # succeeding evaluator's row stays graded.
+    assert report.cases[throwing_idx]["status"] == "could_not_evaluate"
+    assert report.cases[mock_idx]["status"] == "graded"
+
+
+def test_experiment_could_not_evaluate_excluded_from_overall_score():
+    """A throwing evaluator's could_not_evaluate row must not drag overall_score down.
+
+    Two cases run through a passing evaluator (both score 1.0) and a throwing one.
+    The two graded 1.0s should give overall_score 1.0 — not (1.0 + 1.0 + 0 + 0) / 4.
+    """
+    cases = [
+        Case(name="c1", input="hello", expected_output="hello"),
+        Case(name="c2", input="world", expected_output="world"),
+    ]
+    experiment = Experiment(cases=cases, evaluators=[MockEvaluator(), ThrowingEvaluator()])
+
+    def echo_task(c):
+        return c.input
+
+    report = experiment.run_evaluations(echo_task)
+
+    # Four rows: 2 cases x 2 evaluators. Raw per-case scores are preserved.
+    assert len(report.scores) == 4
+    graded = [c for c in report.cases if c["status"] == "graded"]
+    could_not = [c for c in report.cases if c["status"] == "could_not_evaluate"]
+    assert len(graded) == 2
+    assert len(could_not) == 2
+
+    # overall_score averages only the two graded 1.0s, excluding the two skipped 0s.
+    assert report.overall_score == pytest.approx(1.0)
+
 
 def testis_throttling_error_detects_model_throttled_exception():
     """Test that ModelThrottledException is detected as throttling error"""