mark remaining metrics as error

Author: asamal4

src/lightspeed_evaluation/pipeline/evaluation/errors.py

@@ -124,6 +124,74 @@ def mark_turn_metrics_as_error(  # pylint: disable=too-many-arguments,too-many-p
         self.results.extend(error_results)
         return error_results
 
+    def mark_remaining_turns_and_conversation_as_error(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        self,
+        conv_data: EvaluationData,
+        failed_turn_idx: int,
+        resolved_turn_metrics: list[list[str]],
+        resolved_conversation_metrics: list[str],
+        error_reason: str,
+    ) -> list[EvaluationResult]:
+        """Mark all remaining turns and conversation metrics as ERROR after API failure.
+
+        Args:
+            conv_data: Conversation data
+            failed_turn_idx: Index of the turn that failed
+            resolved_turn_metrics: Resolved metrics for all turns
+            resolved_conversation_metrics: Resolved conversation metrics
+            error_reason: Reason for error
+
+        Returns:
+            list[EvaluationResult]: ERROR results for remaining turns and conversation
+        """
+        logger.warning(
+            "Marking remaining turns (%d onwards) and conversation metrics as ERROR for %s: %s",
+            failed_turn_idx + 1,
+            conv_data.conversation_group_id,
+            error_reason,
+        )
+        error_results = []
+
+        # Mark remaining turns as ERROR (from failed_turn_idx + 1 onwards)
+        for turn_idx in range(failed_turn_idx + 1, len(conv_data.turns)):
+            turn_data = conv_data.turns[turn_idx]
+            turn_metrics = resolved_turn_metrics[turn_idx]
+
+            for metric_identifier in turn_metrics:
+                error_result = EvaluationResult(
+                    conversation_group_id=conv_data.conversation_group_id,
+                    turn_id=turn_data.turn_id,
+                    metric_identifier=metric_identifier,
+                    result="ERROR",
+                    score=None,
+                    threshold=None,
+                    reason=error_reason,
+                    query=turn_data.query,
+                    response="",
+                    execution_time=0.0,
+                )
+                error_results.append(error_result)
+
+        # Mark conversation-level metrics as ERROR
+        for metric_identifier in resolved_conversation_metrics:
+            error_result = EvaluationResult(
+                conversation_group_id=conv_data.conversation_group_id,
+                turn_id=None,  # Conversation-level
+                metric_identifier=metric_identifier,
+                result="ERROR",
+                score=None,
+                threshold=None,
+                reason=error_reason,
+                query="",
+                response="",
+                execution_time=0.0,
+            )
+            error_results.append(error_result)
+
+        # Store results internally for summary tracking
+        self.results.extend(error_results)
+        return error_results
+
     def get_error_summary(self) -> dict[str, int]:
         """Get summary of error results collected."""
         return {

src/lightspeed_evaluation/pipeline/evaluation/processor.py

@@ -101,9 +101,7 @@ def process_conversation(  # pylint: disable=too-many-locals
             ):
                 # Step 2a: Amend with API data if enabled (per turn)
                 if self.config.api.enabled:
-                    logger.debug(
-                        "Processing turn %d: %s", turn_idx, turn_data.turn_id
-                    )
+                    logger.debug("Processing turn %d: %s", turn_idx, turn_data.turn_id)
                     api_error_message, conversation_id = (
                         self.components.api_amender.amend_single_turn(
                             turn_data, conversation_id
@@ -115,13 +113,15 @@ def process_conversation(  # pylint: disable=too-many-locals
                         turn_data.turn_id,
                     )
 
-                    # If API error occurred for this turn, mark its metrics as ERROR
+                    # If API error occurred, mark current turn + remaining + conversation as ERROR
                     if api_error_message:
                         logger.error(
-                            "API error for turn %d - marking turn metrics as ERROR",
+                            "API error for turn %d - marking current turn, "
+                            "remaining turns, and conversation as ERROR",
                             turn_idx,
                         )
-                        error_results = (
+                        # Mark current turn as ERROR
+                        current_turn_errors = (
                             self.components.error_handler.mark_turn_metrics_as_error(
                                 conv_data,
                                 turn_idx,
@@ -130,8 +130,25 @@ def process_conversation(  # pylint: disable=too-many-locals
                                 api_error_message,
                             )
                         )
-                        results.extend(error_results)
-                        continue  # Skip to next turn
+                        results.extend(current_turn_errors)
+
+                        # Mark remaining turns and conversation metrics as ERROR
+                        cascade_error_reason = (
+                            f"Cascade failure from turn {turn_idx + 1} API error: "
+                            f"{api_error_message}"
+                        )
+                        error_handler = self.components.error_handler
+                        remaining_errors = error_handler.mark_remaining_turns_and_conversation_as_error(
+                            conv_data,
+                            turn_idx,
+                            resolved_turn_metrics,
+                            resolved_conversation_metrics,
+                            cascade_error_reason,
+                        )
+                        results.extend(remaining_errors)
+
+                        # Stop processing - API failure cascades to all remaining
+                        return results
 
                 # Step 2b: Process turn-level metrics for this turn
                 if turn_metrics:

tests/unit/pipeline/evaluation/test_errors.py

@@ -222,3 +222,73 @@ def test_mark_turn_metrics_as_error(self):
         assert summary["total_errors"] == 2
         assert summary["turn_errors"] == 2
         assert summary["conversation_errors"] == 0
+
+    def test_mark_remaining_turns_and_conversation_as_error(self):
+        """Test marking remaining turns and conversation metrics as error after API failure."""
+        handler = EvaluationErrorHandler()
+
+        # Setup conversation with 3 turns
+        turn1 = TurnData(turn_id="turn1", query="Query 1", response="Response 1")
+        turn2 = TurnData(turn_id="turn2", query="Query 2", response="Response 2")
+        turn3 = TurnData(turn_id="turn3", query="Query 3", response="Response 3")
+        conv_data = EvaluationData(
+            conversation_group_id="test_conv", turns=[turn1, turn2, turn3]
+        )
+
+        # Resolved metrics for all turns
+        resolved_turn_metrics = [
+            ["ragas:faithfulness"],  # turn1
+            ["custom:answer_correctness"],  # turn2
+            ["ragas:response_relevancy"],  # turn3
+        ]
+        resolved_conversation_metrics = [
+            "deepeval:conversation_completeness",
+            "deepeval:conversation_relevancy",
+        ]
+
+        # API failure happens at turn 0 (first turn)
+        failed_turn_idx = 0
+        error_reason = "Cascade failure from turn 1 API error: Connection timeout"
+
+        results = handler.mark_remaining_turns_and_conversation_as_error(
+            conv_data,
+            failed_turn_idx,
+            resolved_turn_metrics,
+            resolved_conversation_metrics,
+            error_reason,
+        )
+
+        # Should have errors for:
+        # - Turn 2 (1 metric) + Turn 3 (1 metric) + Conversation (2 metrics) = 4 total
+        assert len(results) == 4
+
+        # Check turn 2 error
+        turn2_result = results[0]
+        assert turn2_result.conversation_group_id == "test_conv"
+        assert turn2_result.turn_id == "turn2"
+        assert turn2_result.metric_identifier == "custom:answer_correctness"
+        assert turn2_result.result == "ERROR"
+        assert turn2_result.reason == error_reason
+
+        # Check turn 3 error
+        turn3_result = results[1]
+        assert turn3_result.turn_id == "turn3"
+        assert turn3_result.metric_identifier == "ragas:response_relevancy"
+        assert turn3_result.result == "ERROR"
+
+        # Check conversation-level errors
+        conv_result1 = results[2]
+        assert conv_result1.turn_id is None  # Conversation-level
+        assert conv_result1.metric_identifier == "deepeval:conversation_completeness"
+        assert conv_result1.result == "ERROR"
+
+        conv_result2 = results[3]
+        assert conv_result2.turn_id is None  # Conversation-level
+        assert conv_result2.metric_identifier == "deepeval:conversation_relevancy"
+        assert conv_result2.result == "ERROR"
+
+        # Verify summary
+        summary = handler.get_error_summary()
+        assert summary["total_errors"] == 4
+        assert summary["turn_errors"] == 2  # turn2 + turn3
+        assert summary["conversation_errors"] == 2