feat: add status field to EvaluationOutput

src/strands_evals/evaluators/evaluator.py

@@ -79,13 +79,16 @@ 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")
+        # Filter to only graded outputs for aggregation
+        graded = [o for o in outputs if o.status == "graded"]
 
-        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)
+        # Handle empty graded list to avoid division by zero
+        if not graded:
+            return (0.0, False, "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

@@ -59,6 +59,23 @@
 _MAX_RETRY_DELAY = 240  # 4 minutes
 
 
+def _roll_up_status(evaluation_outputs: list) -> str:
+    """Derive the aggregate status for a case from its individual evaluation outputs.
+
+    Precedence (highest to lowest):
+        1. "graded" - if ANY output is graded, the roll-up is graded because
+           the evaluator produced at least one real verdict.
+        2. First output's status - when no output is graded, use the status
+           of the first output (could_not_evaluate or informational).
+        3. "graded" - fallback when outputs list is empty (defensive).
+    """
+    if not evaluation_outputs:
+        return "graded"
+    if any(o.status == "graded" for o in evaluation_outputs):
+        return "graded"
+    return evaluation_outputs[0].status
+
+
 def _get_label_from_score(evaluator: Evaluator, score: float) -> str:
     """
     Get the label from score using evaluator's _score_mapping if available.
@@ -422,6 +439,7 @@ async def _evaluate_with_retry(evaluator=evaluator, evaluation_context=evaluatio
                     "score": aggregate_score,
                     "reason": aggregate_reason or "",
                     "detailed_results": evaluation_outputs,
+                    "status": _roll_up_status(evaluation_outputs),
                 }
 
         except RetryError as e:
@@ -443,6 +461,7 @@ async def _evaluate_with_retry(evaluator=evaluator, evaluation_context=evaluatio
                 "score": 0,
                 "reason": f"Evaluator error: {str(original_exception)}",
                 "detailed_results": [],
+                "status": "could_not_evaluate",
             }
         except Exception as e:
             # Catch non-throttling errors and record as failure (error isolation)
@@ -453,6 +472,7 @@ async def _evaluate_with_retry(evaluator=evaluator, evaluation_context=evaluatio
                 "score": 0,
                 "reason": f"Evaluator error: {str(e)}",
                 "detailed_results": [],
+                "status": "could_not_evaluate",
             }
 
     async def _run_diagnosis(
@@ -569,6 +589,7 @@ async def _worker(
                                     "score": 0,
                                     "reason": f"An error occurred: {str(e)}",
                                     "detailed_results": [],
+                                    "status": "could_not_evaluate",
                                 }
                             )
                         results[index] = {
@@ -659,6 +680,7 @@ async def run_evaluations_async(
                 "detailed_results": [],
                 "diagnoses": [],
                 "recommendations": [],
+                "statuses": [],
             }
             for evaluator in self._evaluators
         }
@@ -678,21 +700,25 @@ async def run_evaluations_async(
                 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)
+                evaluator_data[eval_name]["statuses"].append(eval_result.get("status", "graded"))
 
         reports = []
         for evaluator in self._evaluators:
             eval_name = evaluator.get_name()
             data = evaluator_data[eval_name]
             scores = data["scores"]
+            statuses = data["statuses"]
+            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"],
                 reasons=data["reasons"],
                 detailed_results=data["detailed_results"],
                 diagnoses=data["diagnoses"],
                 recommendations=data["recommendations"],
+                statuses=statuses,
             )
             reports.append(report)
 

src/strands_evals/types/evaluation.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from pydantic import BaseModel
 from typing_extensions import Any, Generic, TypedDict, TypeVar
 
@@ -118,9 +120,22 @@ 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: The grading status of the evaluation output. Controls whether
+            this result is included in aggregate score computations.
+
+            - "graded" (default): The score and test_pass are real verdicts.
+              Included in all aggregations. Backward compatible with every
+              existing evaluator.
+            - "could_not_evaluate": The evaluator tried to grade but could not
+              (preconditions not met, harness failure, missing data). Excluded
+              from pass-rate and score aggregations.
+            - "informational": The evaluator does not pass/fail by design. It
+              surfaces content for human review but is excluded from numeric
+              aggregates.
     """
 
     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

@@ -1,7 +1,8 @@
 import json
 from pathlib import Path
+from typing import Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, model_validator
 
 from ..display.display_console import CollapsibleTableReportDisplay
 from ..types.evaluation import EvaluationOutput
@@ -18,6 +19,9 @@ class EvaluationReport(BaseModel):
             the evaluator that produced that row.
         test_passes: A list of booleans indicating whether the test pass or fail.
         reasons: A list of reason for each test case.
+        statuses: A list of status strings for each test case. Controls whether
+            the result is included in overall_score aggregation. Values are
+            "graded", "could_not_evaluate", or "informational".
     """
 
     overall_score: float
@@ -28,6 +32,19 @@ class EvaluationReport(BaseModel):
     detailed_results: list[list[EvaluationOutput]] = []
     diagnoses: list[dict | None] = []
     recommendations: list[str | None] = []
+    statuses: list[Literal["graded", "could_not_evaluate", "informational"]] = []
+
+    @model_validator(mode="after")
+    def _pad_statuses_to_match_scores(self) -> "EvaluationReport":
+        """Ensure statuses list length matches scores list length.
+
+        If statuses is shorter than scores (e.g., legacy data or partial construction),
+        pad with 'graded' to maintain the invariant that each score has a corresponding status.
+        """
+        expected_len = len(self.scores)
+        if len(self.statuses) < expected_len:
+            self.statuses = list(self.statuses) + ["graded"] * (expected_len - len(self.statuses))
+        return self
 
     @classmethod
     def flatten(cls, reports: list["EvaluationReport"]) -> "EvaluationReport":
@@ -40,7 +57,7 @@ def flatten(cls, reports: list["EvaluationReport"]) -> "EvaluationReport":
         if not reports:
             return cls(overall_score=0.0, scores=[], cases=[], test_passes=[])
 
-        scores, cases, passes, reasons, detailed, diags, recs = [], [], [], [], [], [], []
+        scores, cases, passes, reasons, detailed, diags, recs, statuses = [], [], [], [], [], [], [], []
 
         for report in reports:
             for i, case in enumerate(report.cases):
@@ -51,16 +68,20 @@ def flatten(cls, reports: list["EvaluationReport"]) -> "EvaluationReport":
                 detailed.append(report.detailed_results[i] if i < len(report.detailed_results) else [])
                 diags.append(report.diagnoses[i] if i < len(report.diagnoses) else None)
                 recs.append(report.recommendations[i] if i < len(report.recommendations) else None)
+                statuses.append(report.statuses[i] if i < len(report.statuses) else "graded")
+
+        graded_scores = [s for s, st in zip(scores, statuses, strict=True) if st == "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,
             reasons=reasons,
             detailed_results=detailed,
             diagnoses=diags,
             recommendations=recs,
+            statuses=statuses,
         )
 
     @staticmethod
@@ -134,6 +155,10 @@ def _display(
                 details_dict["evaluator"] = self.cases[i]["evaluator"]
             details_dict["score"] = f"{self.scores[i]:.2f}"
             details_dict["test_pass"] = self.test_passes[i]
+            # Include status when it differs from the default "graded"
+            status = self.statuses[i] if i < len(self.statuses) else "graded"
+            if status != "graded":
+                details_dict["status"] = status
             details_dict["reason"] = reason
             if include_input:
                 details_dict["input"] = self.format_input_for_display(self.cases[i].get("input"))