MAINT Decompose scorer execution policy and result finalization

pyrit/score/message_scorer.py

@@ -6,6 +6,7 @@
 import asyncio
 import logging
 from abc import abstractmethod
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, cast
 
 from pyrit.common.deprecation import print_deprecation_message
@@ -42,6 +43,22 @@
 MESSAGE_BATCH_REMOVED_IN = "1.3.0"
 
 
+@dataclass(frozen=True, kw_only=True)
+class _PreparedMessageScoringInput:
+    """The effective input and policy decision for one message-scoring call."""
+
+    source_message: Message
+    scoring_message: Message | None
+    expectation: ScoringExpectation | None
+    anchor: Scorable | None = None
+    should_skip: bool = False
+
+    @property
+    def objective(self) -> str | None:
+        """The objective attached to the effective expectation."""
+        return self.expectation.objective if self.expectation else None
+
+
 def extract_objective_from_previous_turn(*, message: Message, memory: MemoryInterface) -> str:
     """
     Read the text of the turn before an assistant message and use it as the objective.
@@ -759,120 +776,197 @@ async def _score_resolved_message_async(
             PyritException: If scoring raises a PyRIT exception (re-raised with enhanced context).
             RuntimeError: If scoring raises a non-PyRIT exception (wrapped with scorer context).
         """
-        objective = expectation.objective if expectation else None
+        scoring_input = self._prepare_message_scoring_input(
+            message=message,
+            expectation=expectation,
+            infer_objective_from_request=infer_objective_from_request,
+            anchor=anchor,
+            role_filter=role_filter,
+            skip_on_error_result=skip_on_error_result,
+        )
+        if scoring_input.should_skip:
+            return []
 
-        if not _legacy_policy_allows_message(
+        scores = await self._execute_message_scoring_async(scoring_input=scoring_input)
+        return self._finalize_message_scores(scoring_input=scoring_input, scores=scores)
+
+    def _prepare_message_scoring_input(
+        self,
+        *,
+        message: Message,
+        expectation: ScoringExpectation | None,
+        infer_objective_from_request: bool,
+        anchor: Scorable | None,
+        role_filter: ChatMessageRole | None,
+        skip_on_error_result: bool,
+    ) -> _PreparedMessageScoringInput:
+        """
+        Apply message policy and return the effective input for execution.
+
+        Args:
+            message (Message): The acquired message.
+            expectation (ScoringExpectation | None): What to look for.
+            infer_objective_from_request (bool): Whether to infer a missing objective.
+            anchor (Scorable | None): The scorable the caller named, when there was one.
+            role_filter (ChatMessageRole | None): Deprecated compatibility filter.
+            skip_on_error_result (bool): Deprecated compatibility policy.
+
+        Returns:
+            _PreparedMessageScoringInput: The effective input and skip decision.
+        """
+        objective = expectation.objective if expectation else None
+        should_skip = not _legacy_policy_allows_message(
             message=message,
             role_filter=role_filter,
             skip_on_error_result=skip_on_error_result,
             should_score_blocked_content=self.should_score_blocked_content,
-        ):
-            return []
+        )
 
         # A role this scorer does not read means the evidence is not its to judge, which is
         # neither a verdict nor a failed acquisition. The scorer says nothing at all.
-        if not self._reads_any_role(message=message, anchor=anchor):
+        if not should_skip and not self._reads_any_role(message=message, anchor=anchor):
             logger.debug("Skipping scoring: the scorer does not read this message's role.")
-            return []
+            should_skip = True
 
-        scoring_message = self._build_scoring_message(message=message)
+        scoring_message = None if should_skip else self._build_scoring_message(message=message)
+        effective_expectation = expectation
 
-        if infer_objective_from_request and (not objective):
+        if not should_skip and infer_objective_from_request and not objective:
             objective = extract_objective_from_previous_turn(message=message, memory=self._memory)
 
-        effective_expectation = expectation
-        if expectation is None and objective is not None:
+        if not should_skip and expectation is None and objective is not None:
             effective_expectation = ScoringExpectation(objective=objective)
-        elif expectation is not None and objective != expectation.objective:
+        elif not should_skip and expectation is not None and objective != expectation.objective:
             effective_expectation = ScoringExpectation(
                 objective=objective,
                 conditions=expectation.conditions,
             )
 
-        if scoring_message is None:
-            scores = self._build_fallback_score(message=message, objective=objective)
-            self._finalize_message_scores(message=message, scores=scores, anchor=anchor)
-            return scores
+        if scoring_message is not None:
+            self._validate_scoring_message(message=scoring_message, objective=objective)
+
+        return _PreparedMessageScoringInput(
+            source_message=message,
+            scoring_message=scoring_message,
+            expectation=effective_expectation,
+            anchor=anchor,
+            should_skip=should_skip,
+        )
+
+    async def _execute_message_scoring_async(
+        self,
+        *,
+        scoring_input: _PreparedMessageScoringInput,
+    ) -> list[Score]:
+        """
+        Run scorer code with the scorer-layer exception policy.
 
-        self._validate_scoring_message(message=scoring_message, objective=objective)
+        Args:
+            scoring_input (_PreparedMessageScoringInput): The prepared scoring input.
+
+        Returns:
+            list[Score]: The scores produced by the scorer or its blocked fallback.
+
+        Raises:
+            ScorerLLMResponseBlockedException: If the scorer's own LLM response is blocked
+                and ``raise_if_scorer_blocks`` is True.
+            PyritException: If scoring raises a PyRIT exception.
+            RuntimeError: If scoring raises a non-PyRIT exception.
+        """
+        scoring_message = scoring_input.scoring_message
+        if scoring_message is None:
+            return []
 
         try:
-            scores = await self._score_prepared_message_async(
+            return await self._score_prepared_message_async(
                 message=scoring_message,
-                expectation=effective_expectation,
+                expectation=scoring_input.expectation,
             )
-        except ScorerLLMResponseBlockedException as e:
+        except ScorerLLMResponseBlockedException as error:
             # The scorer's own LLM response was content-filtered. By default this is a real
             # error and propagates; when raise_if_scorer_blocks is False, no verdict was
             # reached, so the score is undetermined rather than a definitive negative. The
             # decision lives here in the scorer, not the transport (see doc/code/framework.md).
             if self.raise_if_scorer_blocks:
-                e.message = f"Error in scorer {self.__class__.__name__}: {e.message}"
-                e.args = (f"Status Code: {e.status_code}, Message: {e.message}",)
+                error.message = f"Error in scorer {self.__class__.__name__}: {error.message}"
+                error.args = (f"Status Code: {error.status_code}, Message: {error.message}",)
                 raise
             logger.info(
                 "Scorer %s LLM response was blocked by content filtering; "
                 "returning an undetermined score (raise_if_scorer_blocks=False).",
                 self.__class__.__name__,
             )
             first_piece = scoring_message.message_pieces[0]
-            scores = [
+            return [
                 self._build_undetermined_score(
                     rationale=(
                         "The scorer's own LLM response was blocked by content filtering "
                         "(raise_if_scorer_blocks is False), so no verdict was reachable."
                     ),
                     description="Scorer response blocked; no verdict was reachable.",
                     message_piece_id=first_piece.id or first_piece.original_prompt_id,
-                    objective=objective,
+                    objective=scoring_input.objective,
                 )
             ]
-        except PyritException as e:
-            # Re-raise PyRIT exceptions with enhanced context while preserving type for retry decorators
-            e.message = f"Error in scorer {self.__class__.__name__}: {e.message}"
-            e.args = (f"Status Code: {e.status_code}, Message: {e.message}",)
+        except PyritException as error:
+            error.message = f"Error in scorer {self.__class__.__name__}: {error.message}"
+            error.args = (f"Status Code: {error.status_code}, Message: {error.message}",)
             raise
-        except Exception as e:
-            # Wrap non-PyRIT exceptions for better error tracing
-            raise RuntimeError(f"Error in scorer {self.__class__.__name__}: {str(e)}") from e
+        except Exception as error:
+            raise RuntimeError(f"Error in scorer {self.__class__.__name__}: {str(error)}") from error
 
-        if not scores and scoring_message.message_pieces and not self._get_supported_pieces(scoring_message):
-            scores = self._build_fallback_score(message=message, objective=objective)
-
-        self._finalize_message_scores(message=scoring_message, scores=scores, anchor=anchor)
-
-        return scores
-
-    def _validate_scoring_message(self, *, message: Message, objective: str | None) -> None:
+    def _finalize_message_scores(
+        self,
+        *,
+        scoring_input: _PreparedMessageScoringInput,
+        scores: list[Score],
+    ) -> list[Score]:
         """
-        Validate the acquired message before it reaches the leaf scorer.
+        Apply fallback behavior and canonical evidence anchors.
 
         Args:
-            message (Message): The acquired message.
-            objective (str | None): The objective supplied for scoring.
+            scoring_input (_PreparedMessageScoringInput): The prepared scoring input.
+            scores (list[Score]): The scores produced by execution.
+
+        Returns:
+            list[Score]: The finalized scores.
         """
-        self._validator.validate(message, objective=objective)
+        scoring_message = scoring_input.scoring_message
+        if not scores and (
+            scoring_message is None
+            or (scoring_message.message_pieces and not self._get_supported_pieces(scoring_message))
+        ):
+            scores = self._build_fallback_score(
+                message=scoring_input.source_message,
+                objective=scoring_input.objective,
+            )
 
-    def _finalize_message_scores(
-        self,
-        *,
-        message: Message,
-        scores: list[Score],
-        anchor: Scorable | None,
-    ) -> None:
-        """Apply legacy and canonical evidence anchors to completed message scores."""
-        persisted_piece_ids = self._get_persisted_piece_ids(message=message) if anchor is None else None
+        finalization_message = scoring_message or scoring_input.source_message
+        persisted_piece_ids = (
+            self._get_persisted_piece_ids(message=finalization_message) if scoring_input.anchor is None else None
+        )
         self._drop_ephemeral_score_links(
-            message=message,
+            message=finalization_message,
             scores=scores,
             persisted_piece_ids=persisted_piece_ids,
         )
         self._stamp_scorable(
-            message=message,
+            message=finalization_message,
             scores=scores,
-            anchor=anchor,
+            anchor=scoring_input.anchor,
             persisted_piece_ids=persisted_piece_ids,
         )
+        return scores
+
+    def _validate_scoring_message(self, *, message: Message, objective: str | None) -> None:
+        """
+        Validate the acquired message before it reaches the leaf scorer.
+
+        Args:
+            message (Message): The acquired message.
+            objective (str | None): The objective supplied for scoring.
+        """
+        self._validator.validate(message, objective=objective)
 
     async def _score_prepared_message_async(
         self,