diff --git a/pyrit/score/message_scorer.py b/pyrit/score/message_scorer.py index 2dc68291d9..c65472a950 100644 --- a/pyrit/score/message_scorer.py +++ b/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,56 +776,120 @@ 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; " @@ -816,7 +897,7 @@ async def _score_resolved_message_async( 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 " @@ -824,55 +905,68 @@ async def _score_resolved_message_async( ), 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, diff --git a/tests/unit/score/test_message_scorer.py b/tests/unit/score/test_message_scorer.py index 67f7c0ef29..d50d22f462 100644 --- a/tests/unit/score/test_message_scorer.py +++ b/tests/unit/score/test_message_scorer.py @@ -5,10 +5,11 @@ import inspect import uuid from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from pyrit.exceptions import InvalidJsonException from pyrit.memory import CentralMemory, MemoryInterface from pyrit.memory.memory_models import ScorableContentEntry from pyrit.models import ( @@ -19,6 +20,7 @@ MatchesObjective, Message, MessagePiece, + PromptResponseError, Score, ScoreStatus, ScoringExpectation, @@ -425,6 +427,186 @@ async def test_no_expectation_means_no_objective(self): assert scorer.scored_objectives == [None] +@pytest.mark.usefixtures("patch_central_database") +class TestMessageScoringPolicyMatrix: + """The prepared input makes substitutions, filtering, and objective state explicit.""" + + @pytest.mark.parametrize( + ( + "role_filter", + "supported_roles", + "skip_on_error_result", + "structured_refusal", + "partial_content", + "should_score_blocked_content", + "input_response_error", + "expected_skip", + "expected_value", + "expected_response_error", + ), + [ + (None, ["assistant"], False, None, None, False, "none", False, "response", "none"), + ("user", ["assistant"], False, None, None, False, "none", True, None, None), + (None, ["user"], False, None, None, False, "none", True, None, None), + (None, ["assistant"], True, None, None, False, "blocked", True, None, None), + ( + None, + ["assistant"], + True, + "I cannot assist.", + None, + False, + "blocked", + False, + "I cannot assist.", + "blocked", + ), + ( + None, + ["assistant"], + True, + None, + "partial response", + True, + "blocked", + False, + "partial response", + "none", + ), + (None, ["assistant"], False, None, None, True, "blocked", False, None, None), + (None, ["assistant"], False, None, None, True, "processing", False, None, None), + ], + ) + def test_prepared_input_policy_matrix( + self, + role_filter: ChatMessageRole | None, + supported_roles: list[ChatMessageRole], + skip_on_error_result: bool, + structured_refusal: str | None, + partial_content: str | None, + should_score_blocked_content: bool, + input_response_error: PromptResponseError, + expected_skip: bool, + expected_value: str | None, + expected_response_error: PromptResponseError | None, + ) -> None: + has_error = input_response_error != "none" + piece = MessagePiece( + role="assistant", + original_value="error" if has_error else "response", + original_value_data_type="error" if has_error else "text", + converted_value_data_type="error" if has_error else "text", + response_error=input_response_error, + prompt_metadata={"partial_content": partial_content} if partial_content is not None else {}, + ) + if structured_refusal is not None: + piece.mark_as_structured_refusal(refusal=structured_refusal) + message = piece.to_message() + scorer = RecordingScorer(supported_roles=supported_roles) + scorer.should_score_blocked_content = should_score_blocked_content + + scoring_input = scorer._prepare_message_scoring_input( + message=message, + expectation=ScoringExpectation(objective="objective"), + infer_objective_from_request=False, + anchor=None, + role_filter=role_filter, + skip_on_error_result=skip_on_error_result, + ) + + assert scoring_input.source_message is message + assert scoring_input.should_skip is expected_skip + assert scoring_input.objective == "objective" + if expected_value is None: + assert scoring_input.scoring_message is None + else: + assert scoring_input.scoring_message is not None + assert scoring_input.scoring_message.get_value() == expected_value + assert scoring_input.scoring_message.get_piece().response_error == expected_response_error + + def test_inferred_objective_is_part_of_the_prepared_input(self, sqlite_instance: MemoryInterface) -> None: + conversation_id = str(uuid.uuid4()) + sqlite_instance.add_message_to_memory( + request=MessagePiece( + role="user", + original_value="the inferred objective", + conversation_id=conversation_id, + sequence=0, + ).to_message() + ) + message = _assistant_message("response", conversation_id=conversation_id) + scorer = RecordingScorer() + + scoring_input = scorer._prepare_message_scoring_input( + message=message, + expectation=None, + infer_objective_from_request=True, + anchor=None, + role_filter=None, + skip_on_error_result=False, + ) + + assert scoring_input.objective == "the inferred objective" + assert scoring_input.expectation == ScoringExpectation(objective="the inferred objective") + + +@pytest.mark.usefixtures("patch_central_database") +class TestMessageScoringExecutionAndFinalization: + """Execution policy and outer validation/persistence retain their exact contracts.""" + + async def test_pyrit_exception_keeps_type_and_gains_scorer_context(self) -> None: + scorer = RecordingScorer() + error = InvalidJsonException(message="invalid scorer response") + + with patch.object( + scorer, + "_score_prepared_message_async", + new=AsyncMock(side_effect=error), + ): + with pytest.raises(InvalidJsonException) as exception_info: + await scorer.score_async(scorable=MessageScorable.from_message(_assistant_message())) + + assert exception_info.value is error + assert error.message == "Error in scorer RecordingScorer: invalid scorer response" + + async def test_non_pyrit_exception_is_wrapped_with_scorer_context(self) -> None: + scorer = RecordingScorer() + error = ValueError("unexpected scorer failure") + + with patch.object( + scorer, + "_score_prepared_message_async", + new=AsyncMock(side_effect=error), + ): + with pytest.raises(RuntimeError, match="Error in scorer RecordingScorer") as exception_info: + await scorer.score_async(scorable=MessageScorable.from_message(_assistant_message())) + + assert exception_info.value.__cause__ is error + + async def test_success_validates_and_persists_scores_once(self) -> None: + scorer = RecordingScorer() + with ( + patch.object(scorer, "validate_return_scores", wraps=scorer.validate_return_scores) as validate, + patch.object(scorer._memory, "add_scores_to_memory") as persist, + ): + scores = await scorer.score_async(scorable=MessageScorable.from_message(_assistant_message())) + + validate.assert_called_once_with(scores=scores) + persist.assert_called_once_with(scores=scores) + + async def test_skipped_input_is_not_validated_or_persisted(self) -> None: + scorer = RecordingScorer(supported_roles=["user"]) + with ( + patch.object(scorer, "validate_return_scores", wraps=scorer.validate_return_scores) as validate, + patch.object(scorer._memory, "add_scores_to_memory") as persist, + ): + scores = await scorer.score_async(scorable=MessageScorable.from_message(_assistant_message())) + + assert scores == [] + validate.assert_not_called() + persist.assert_not_called() + + @pytest.mark.usefixtures("patch_central_database") class TestDeprecatedParameters: """The legacy message-shaped parameters survive one release behind a warning."""