From 797c361edfc032f0c374834789aa498d83376fea Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 3 Jun 2026 12:11:04 -0700 Subject: [PATCH 1/4] Convert ScenarioResult and ScenarioIdentifier to Pydantic v2 (Phase 7) Relocate get_scorer_evaluation_metrics to pyrit.score.scenario_metrics.get_scenario_metrics, make display_group_map a public field, and keep to_dict/from_dict as deprecated shims preserving the legacy wire shape. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- pyrit/memory/memory_models.py | 8 +- pyrit/models/scenario_result.py | 306 ++++++------------ .../scenarios/benchmark/adversarial.py | 4 +- pyrit/score/__init__.py | 2 + pyrit/score/scenario_metrics.py | 41 +++ tests/unit/models/test_import_boundary.py | 4 - tests/unit/models/test_scenario_result.py | 53 ++- .../output/scenario_result/test_pretty.py | 2 +- .../scenario/benchmark/test_adversarial.py | 8 +- tests/unit/score/test_scenario_metrics.py | 46 +++ 10 files changed, 253 insertions(+), 221 deletions(-) create mode 100644 pyrit/score/scenario_metrics.py create mode 100644 tests/unit/score/test_scenario_metrics.py diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index e70a421a6d..7e852436c8 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -1072,7 +1072,7 @@ def __init__(self, *, entry: ScenarioResult) -> None: self.attack_results_json = json.dumps(serialized_attack_results) # Serialize display_group_map if present - self.display_group_map_json = json.dumps(entry._display_group_map) if entry._display_group_map else None + self.display_group_map_json = json.dumps(entry.display_group_map) if entry.display_group_map else None self.error_message = entry.error_message self.error_type = entry.error_type @@ -1126,14 +1126,14 @@ def get_scenario_result(self) -> ScenarioResult: attack_results=attack_results, objective_scorer_identifier=scorer_identifier, scenario_run_state=self.scenario_run_state, - labels=self.labels, + labels=self.labels or {}, creation_time=self.timestamp, number_tries=self.number_tries, completion_time=self.completion_time, - display_group_map=display_group_map, + display_group_map=display_group_map or {}, error_message=self.error_message, error_type=self.error_type, - metadata=dict(self.scenario_metadata) if self.scenario_metadata else None, + metadata=dict(self.scenario_metadata) if self.scenario_metadata else {}, ) def get_conversation_ids_by_attack_name(self) -> dict[str, list[str]]: diff --git a/pyrit/models/scenario_result.py b/pyrit/models/scenario_result.py index 3680d83b7d..e511c2bedb 100644 --- a/pyrit/models/scenario_result.py +++ b/pyrit/models/scenario_result.py @@ -6,171 +6,130 @@ import logging import uuid from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Literal +from typing import Any, Literal -import pyrit -from pyrit.models import AttackOutcome, AttackResult +from pydantic import BaseModel, ConfigDict, Field -if TYPE_CHECKING: - from pyrit.models.identifiers.component_identifier import ComponentIdentifier - from pyrit.score.scorer_evaluation.scorer_metrics import ScorerMetrics +import pyrit +from pyrit.common.deprecation import print_deprecation_message +from pyrit.models.identifiers.component_identifier import ( # noqa: TC001 (runtime-required by Pydantic field annotations) + ComponentIdentifier, +) +from pyrit.models.results.attack_result import AttackOutcome, AttackResult logger = logging.getLogger(__name__) -class ScenarioIdentifier: +class ScenarioIdentifier(BaseModel): """ - Scenario result class for aggregating results from multiple AtomicAttacks. + Identifier describing the executed scenario. """ - def __init__( - self, - name: str, - description: str = "", - scenario_version: int = 1, - init_data: dict[str, Any] | None = None, - pyrit_version: str | None = None, - ) -> None: - """ - Initialize a ScenarioIdentifier. + model_config = ConfigDict(extra="forbid", populate_by_name=True) - Args: - name (str): Name of the scenario. - description (str): Description of the scenario. - scenario_version (int): Version of the scenario. - init_data (Optional[dict]): Initialization data. - pyrit_version (Optional[str]): PyRIT version string. If None, uses current version. - - """ - self.name = name - self.description = description - self.version = scenario_version - self.pyrit_version = pyrit_version if pyrit_version is not None else pyrit.__version__ - self.init_data = init_data + #: Name of the scenario. + name: str + #: Description of the scenario. + description: str = "" + #: Version of the scenario. Accepts the legacy ``scenario_version`` kwarg/wire key. + version: int = Field(default=1, alias="scenario_version") + #: PyRIT version string. Defaults to the current installed version. + pyrit_version: str = Field(default_factory=lambda: pyrit.__version__) + #: Optional initialization data. + init_data: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: """ Serialize to a JSON-compatible dictionary. + Deprecated: use ``model_dump(by_alias=True)`` instead. + Returns: dict[str, Any]: Serialized payload. """ - return { - "name": self.name, - "description": self.description, - "scenario_version": self.version, - "pyrit_version": self.pyrit_version, - "init_data": self.init_data, - } + print_deprecation_message( + old_item="ScenarioIdentifier.to_dict()", + new_item="ScenarioIdentifier.model_dump(by_alias=True)", + removed_in="0.16.0", + ) + return self.model_dump(by_alias=True) @classmethod def from_dict(cls, data: dict[str, Any]) -> ScenarioIdentifier: """ Reconstruct a ScenarioIdentifier from a dictionary. + Deprecated: use ``model_validate(...)`` instead. + Args: - data (dict[str, Any]): Dictionary as produced by to_dict(). + data (dict[str, Any]): Dictionary as produced by ``model_dump(by_alias=True)``. Returns: ScenarioIdentifier: Reconstructed instance. """ - return cls( - name=data["name"], - description=data.get("description", ""), - scenario_version=data.get("scenario_version", 1), - init_data=data.get("init_data"), - pyrit_version=data.get("pyrit_version") or "unknown", + print_deprecation_message( + old_item="ScenarioIdentifier.from_dict(...)", + new_item="ScenarioIdentifier.model_validate(...)", + removed_in="0.16.0", ) + return cls.model_validate(data) ScenarioRunState = Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED"] -class ScenarioResult: +class ScenarioResult(BaseModel): """ Scenario result class for aggregating scenario results. """ - def __init__( - self, - *, - scenario_identifier: ScenarioIdentifier, - objective_target_identifier: ComponentIdentifier | None, - attack_results: dict[str, list[AttackResult]], - objective_scorer_identifier: ComponentIdentifier | None, - scenario_run_state: ScenarioRunState = "CREATED", - labels: dict[str, str] | None = None, - creation_time: datetime | None = None, - completion_time: datetime | None = None, - number_tries: int = 0, - id: uuid.UUID | None = None, # noqa: A002 - display_group_map: dict[str, str] | None = None, - error_message: str | None = None, - error_type: str | None = None, - error_attack_result_ids: list[str] | None = None, - metadata: dict[str, Any] | None = None, - ) -> None: - """ - Initialize a scenario result. - - Args: - scenario_identifier (ScenarioIdentifier): Identifier for the executed scenario. - objective_target_identifier (ComponentIdentifier): Target identifier. - attack_results (dict[str, List[AttackResult]]): Results grouped by atomic attack name. - objective_scorer_identifier (ComponentIdentifier | None): Objective scorer identifier, - or None if the scenario has no objective scorer. - scenario_run_state (ScenarioRunState): Current scenario run state. - labels (Optional[dict[str, str]]): Optional labels. - creation_time (datetime | None): When the scenario result was created. - completion_time (Optional[datetime]): Optional completion timestamp. - number_tries (int): Number of run attempts. - id (Optional[uuid.UUID]): Optional scenario result ID. - display_group_map (Optional[dict[str, str]]): Optional mapping of - atomic_attack_name → display group label. Used by the console - printer to aggregate results for user-facing output. - error_message (Optional[str]): Scenario-level error message when the run fails. - error_type (Optional[str]): Exception class name when the run fails. - error_attack_result_ids (Optional[list[str]]): IDs of attack results that - errored during the scenario run. Defaults to an empty list. - metadata (Optional[dict[str, Any]]): Free-form JSON metadata persisted - with the scenario result. Currently used to record - ``objective_hashes`` — the objective ``sha256`` set chosen - on the first run, replayed on resume so a fresh - ``random.sample`` can't silently change which objectives the - scenario operates on. Keys are not part of any public contract - and may evolve. - - """ - self.id = id if id is not None else uuid.uuid4() - self.scenario_identifier = scenario_identifier - - self.objective_target_identifier = objective_target_identifier - - self.objective_scorer_identifier = objective_scorer_identifier - - self.scenario_run_state = scenario_run_state - self.attack_results = attack_results - self.labels = labels if labels is not None else {} - self.creation_time = creation_time if creation_time is not None else datetime.now(timezone.utc) - self.completion_time = completion_time if completion_time is not None else datetime.now(timezone.utc) - self.number_tries = number_tries - self._display_group_map = display_group_map or {} - self.error_message = error_message - self.error_type = error_type - self.error_attack_result_ids: list[str] = list(error_attack_result_ids) if error_attack_result_ids else [] - self.metadata: dict[str, Any] = metadata if metadata is not None else {} - - @property - def display_group_map(self) -> dict[str, str]: - """Mapping of atomic_attack_name → display group label.""" - return self._display_group_map + model_config = ConfigDict( + arbitrary_types_allowed=True, + extra="forbid", + validate_assignment=False, + ) + + #: Scenario result ID. + id: uuid.UUID = Field(default_factory=uuid.uuid4) # noqa: A003 + #: Identifier for the executed scenario. + scenario_identifier: ScenarioIdentifier + #: Target identifier. + objective_target_identifier: ComponentIdentifier | None + #: Objective scorer identifier, or None if the scenario has no objective scorer. + objective_scorer_identifier: ComponentIdentifier | None + #: Results grouped by atomic attack name. + attack_results: dict[str, list[AttackResult]] + #: Current scenario run state. + scenario_run_state: ScenarioRunState = "CREATED" + #: Optional labels. + labels: dict[str, str] = Field(default_factory=dict) + #: When the scenario result was created. + creation_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + #: Optional completion timestamp. + completion_time: datetime | None = Field(default_factory=lambda: datetime.now(timezone.utc)) + #: Number of run attempts. + number_tries: int = 0 + #: Mapping of ``atomic_attack_name`` -> display group label. Used by the console + #: printer to aggregate results for user-facing output. + display_group_map: dict[str, str] = Field(default_factory=dict) + #: Scenario-level error message when the run fails. + error_message: str | None = None + #: Exception class name when the run fails. + error_type: str | None = None + #: IDs of attack results that errored during the scenario run. + error_attack_result_ids: list[str] = Field(default_factory=list) + #: Free-form JSON metadata persisted with the scenario result. Currently used to record + #: ``objective_hashes`` — the objective ``sha256`` set chosen on the first run, replayed + #: on resume so a fresh ``random.sample`` can't silently change which objectives the + #: scenario operates on. Keys are not part of any public contract and may evolve. + metadata: dict[str, Any] = Field(default_factory=dict) def get_strategies_used(self) -> list[str]: """ Get the list of strategies used in this scenario. Returns: - List[str]: Atomic attack strategy names present in the results. + list[str]: Atomic attack strategy names present in the results. """ return list(self.attack_results.keys()) @@ -181,18 +140,18 @@ def get_display_groups(self) -> dict[str, list[AttackResult]]: When a ``display_group_map`` was provided, results from multiple ``atomic_attack_name`` keys that share the same display group are - merged into a single list. When no map was provided, this returns + merged into a single list. When no map was provided, this returns the same structure as ``attack_results`` (identity mapping). Returns: dict[str, list[AttackResult]]: Results grouped by display label. """ - if not self._display_group_map: + if not self.display_group_map: return dict(self.attack_results) grouped: dict[str, list[AttackResult]] = {} for attack_name, results in self.attack_results.items(): - group = self._display_group_map.get(attack_name, attack_name) + group = self.display_group_map.get(attack_name, attack_name) grouped.setdefault(group, []).extend(results) return grouped @@ -201,11 +160,11 @@ def get_objectives(self, *, atomic_attack_name: str | None = None) -> list[str]: Get the list of unique objectives for this scenario. Args: - atomic_attack_name (Optional[str]): Name of specific atomic attack to include. + atomic_attack_name (str | None): Name of specific atomic attack to include. If None, includes objectives from all atomic attacks. Defaults to None. Returns: - List[str]: Deduplicated list of objectives. + list[str]: Deduplicated list of objectives. """ objectives: list[str] = [] @@ -231,7 +190,7 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i Get the success rate of this scenario. Args: - atomic_attack_name (Optional[str]): Name of specific atomic attack to calculate rate for. + atomic_attack_name (str | None): Name of specific atomic attack to calculate rate for. If None, calculates rate across all atomic attacks. Defaults to None. Returns: @@ -270,10 +229,10 @@ def normalize_scenario_name(scenario_name: str) -> str: This is the inverse of ScenarioRegistry._class_name_to_scenario_name(). Args: - scenario_name: The scenario name to normalize. + scenario_name (str): The scenario name to normalize. Returns: - The normalized scenario name suitable for database queries. + str: The normalized scenario name suitable for database queries. """ # Check if it looks like snake_case (contains underscore and is lowercase) @@ -285,97 +244,38 @@ def normalize_scenario_name(scenario_name: str) -> str: # Already PascalCase or other format, return as-is return scenario_name - def get_scorer_evaluation_metrics(self) -> ScorerMetrics | None: - """ - Get the evaluation metrics for the scenario's scorer from the scorer evaluation registry. - - Returns: - ScorerMetrics: The evaluation metrics object, or None if not found. - - """ - # import here to avoid circular imports - from pyrit.models.identifiers.evaluation_identifier import ScorerEvaluationIdentifier - from pyrit.score.scorer_evaluation.scorer_metrics_io import ( - find_objective_metrics_by_eval_hash, - ) - - if not self.objective_scorer_identifier: - return None - - eval_hash = ScorerEvaluationIdentifier(self.objective_scorer_identifier).eval_hash - - return find_objective_metrics_by_eval_hash(eval_hash=eval_hash) - def to_dict(self) -> dict[str, Any]: """ Serialize this scenario result to a JSON-compatible dictionary. + Deprecated: use ``model_dump(mode="json", by_alias=True)`` instead. + Returns: dict[str, Any]: Serialized payload suitable for REST APIs or persistence. """ - return { - "id": str(self.id), - "scenario_identifier": self.scenario_identifier.to_dict(), - "objective_target_identifier": ( - self.objective_target_identifier.model_dump() if self.objective_target_identifier else None - ), - "objective_scorer_identifier": ( - self.objective_scorer_identifier.model_dump() if self.objective_scorer_identifier else None - ), - "scenario_run_state": self.scenario_run_state, - "attack_results": {name: [r.to_dict() for r in results] for name, results in self.attack_results.items()}, - "display_group_map": self._display_group_map, - "labels": self.labels, - "creation_time": self.creation_time.isoformat() if self.creation_time else None, - "completion_time": self.completion_time.isoformat() if self.completion_time else None, - "number_tries": self.number_tries, - "error_attack_result_ids": self.error_attack_result_ids, - "error_message": self.error_message, - "error_type": self.error_type, - } + print_deprecation_message( + old_item="ScenarioResult.to_dict()", + new_item="ScenarioResult.model_dump(mode='json', by_alias=True)", + removed_in="0.16.0", + ) + return self.model_dump(mode="json", by_alias=True) @classmethod def from_dict(cls, data: dict[str, Any]) -> ScenarioResult: """ Reconstruct a ScenarioResult from a dictionary. + Deprecated: use ``model_validate(...)`` instead. + Args: - data (dict[str, Any]): Dictionary as produced by to_dict(). + data (dict[str, Any]): Dictionary as produced by ``model_dump(mode="json")``. Returns: ScenarioResult: Reconstructed instance. """ - from pyrit.models.identifiers.component_identifier import ComponentIdentifier - - result = cls( - id=uuid.UUID(data["id"]) if data.get("id") else None, - scenario_identifier=ScenarioIdentifier.from_dict(data["scenario_identifier"]), - objective_target_identifier=( - ComponentIdentifier.model_validate(data["objective_target_identifier"]) - if data.get("objective_target_identifier") - else None - ), - objective_scorer_identifier=( - ComponentIdentifier.model_validate(data["objective_scorer_identifier"]) - if data.get("objective_scorer_identifier") - else None - ), - scenario_run_state=data.get("scenario_run_state", "CREATED"), - attack_results={ - name: [AttackResult.from_dict(r) for r in results] - for name, results in data.get("attack_results", {}).items() - }, - display_group_map=data.get("display_group_map"), - labels=data.get("labels"), - creation_time=(datetime.fromisoformat(data["creation_time"]) if data.get("creation_time") else None), - completion_time=(datetime.fromisoformat(data["completion_time"]) if data.get("completion_time") else None), - number_tries=data.get("number_tries", 0), - error_attack_result_ids=data.get("error_attack_result_ids"), - error_message=data.get("error_message"), - error_type=data.get("error_type"), + print_deprecation_message( + old_item="ScenarioResult.from_dict(...)", + new_item="ScenarioResult.model_validate(...)", + removed_in="0.16.0", ) - # Preserve missing completion_time: __init__ defaults it to now(), but a - # still-running scenario shouldn't be marked as completed-at-load-time. - if not data.get("completion_time"): - result.completion_time = None # type: ignore[ty:invalid-assignment] - return result + return cls.model_validate(data) diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index f6d57d2d29..e08ef6d3d1 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -357,13 +357,13 @@ async def run_async(self) -> ScenarioResult: Returns: ScenarioResult: The scenario result with cached attack results merged into ``attack_results`` and cached display groups merged into - ``_display_group_map``. + ``display_group_map``. """ result = await super().run_async() if self._precomputed_cached_results: for attack_name, prior_results in self._precomputed_cached_results.items(): result.attack_results.setdefault(attack_name, []).extend(prior_results) - result._display_group_map.update(self._precomputed_cached_display_groups) + result.display_group_map.update(self._precomputed_cached_display_groups) return result def _collect_cached_completion_pairs(self, *, atomic_attacks: list[AtomicAttack]) -> set[str]: diff --git a/pyrit/score/__init__.py b/pyrit/score/__init__.py index f406bd795a..40a1317aee 100644 --- a/pyrit/score/__init__.py +++ b/pyrit/score/__init__.py @@ -25,6 +25,7 @@ from pyrit.score.float_scale.self_ask_general_float_scale_scorer import SelfAskGeneralFloatScaleScorer from pyrit.score.float_scale.self_ask_likert_scorer import LikertScaleEvalFiles, LikertScalePaths, SelfAskLikertScorer from pyrit.score.float_scale.self_ask_scale_scorer import SelfAskScaleScorer +from pyrit.score.scenario_metrics import get_scenario_metrics from pyrit.score.scorer import Scorer from pyrit.score.scorer_evaluation.metrics_type import MetricsType, RegistryUpdateBehavior from pyrit.score.scorer_evaluation.scorer_metrics import ( @@ -165,6 +166,7 @@ def __getattr__(name: str) -> object: "ScorerMetricsWithIdentity", "get_all_harm_metrics", "get_all_objective_metrics", + "get_scenario_metrics", "find_objective_metrics_by_eval_hash", "ScorerPromptValidator", "SelfAskCategoryScorer", diff --git a/pyrit/score/scenario_metrics.py b/pyrit/score/scenario_metrics.py new file mode 100644 index 0000000000..215da111f3 --- /dev/null +++ b/pyrit/score/scenario_metrics.py @@ -0,0 +1,41 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Scorer-evaluation glue for scenario results. + +This lives in ``pyrit.score`` rather than on ``ScenarioResult`` because looking up +evaluation metrics requires the scoring layer; the data model must not depend on it. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from pyrit.models.identifiers.evaluation_identifier import ScorerEvaluationIdentifier +from pyrit.score.scorer_evaluation.scorer_metrics_io import find_objective_metrics_by_eval_hash + +if TYPE_CHECKING: + from pyrit.models.scenario_result import ScenarioResult + from pyrit.score.scorer_evaluation.scorer_metrics import ScorerMetrics + + +def get_scenario_metrics(scenario_result: ScenarioResult) -> ScorerMetrics | None: + """ + Get the evaluation metrics for a scenario's objective scorer. + + Args: + scenario_result (ScenarioResult): The scenario result whose objective scorer + metrics should be looked up. + + Returns: + ScorerMetrics | None: The evaluation metrics object, or None if the scenario + has no objective scorer or no matching metrics are registered. + + """ + if not scenario_result.objective_scorer_identifier: + return None + + eval_hash = ScorerEvaluationIdentifier(scenario_result.objective_scorer_identifier).eval_hash + + return find_objective_metrics_by_eval_hash(eval_hash=eval_hash) diff --git a/tests/unit/models/test_import_boundary.py b/tests/unit/models/test_import_boundary.py index b4c33795e2..235d86e5d5 100644 --- a/tests/unit/models/test_import_boundary.py +++ b/tests/unit/models/test_import_boundary.py @@ -59,10 +59,6 @@ "pyrit.models.data_type_serializer": { "pyrit.memory": "phase-9", }, - "pyrit.models.scenario_result": { - "pyrit.score.scorer_evaluation.scorer_metrics": "phase-7", - "pyrit.score.scorer_evaluation.scorer_metrics_io": "phase-7", - }, "pyrit.models.storage_io": { "pyrit.auth": "phase-9", }, diff --git a/tests/unit/models/test_scenario_result.py b/tests/unit/models/test_scenario_result.py index fbf46e19da..1bfaf66dac 100644 --- a/tests/unit/models/test_scenario_result.py +++ b/tests/unit/models/test_scenario_result.py @@ -268,10 +268,15 @@ def test_scenario_result_to_dict_from_dict_roundtrip(): ) roundtripped = ScenarioResult.from_dict(original.to_dict()) assert original.to_dict() == roundtripped.to_dict() + # The nested identifier must preserve the legacy ``scenario_version`` wire key. + assert "scenario_version" in original.to_dict()["scenario_identifier"] + assert "version" not in original.to_dict()["scenario_identifier"] -def test_scenario_identifier_from_dict_missing_pyrit_version_yields_unknown(): - """from_dict should preserve 'unknown' rather than fabricating the current version when missing.""" +def test_scenario_identifier_from_dict_missing_pyrit_version_uses_current(): + """A payload missing pyrit_version now resolves to the current version via the Pydantic default.""" + import pyrit + data = { "name": "Legacy", "description": "loaded from older payload", @@ -280,7 +285,7 @@ def test_scenario_identifier_from_dict_missing_pyrit_version_yields_unknown(): # pyrit_version intentionally absent } identifier = ScenarioIdentifier.from_dict(data) - assert identifier.pyrit_version == "unknown" + assert identifier.pyrit_version == pyrit.__version__ def test_scenario_result_from_dict_preserves_missing_completion_time(): @@ -300,3 +305,45 @@ def test_scenario_result_from_dict_preserves_missing_completion_time(): roundtripped = ScenarioResult.from_dict(original.to_dict()) assert roundtripped.completion_time is None assert roundtripped.scenario_run_state == "IN_PROGRESS" + + +def test_scenario_identifier_to_dict_from_dict_emit_deprecation_warnings(): + import pytest + + identifier = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + with pytest.warns(DeprecationWarning): + payload = identifier.to_dict() + with pytest.warns(DeprecationWarning): + ScenarioIdentifier.from_dict(payload) + + +def test_scenario_result_to_dict_from_dict_emit_deprecation_warnings(): + import pytest + + scenario_id = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + result = ScenarioResult( + scenario_identifier=scenario_id, + objective_target_identifier=ComponentIdentifier.from_dict({}), + objective_scorer_identifier=None, + attack_results={}, + ) + with pytest.warns(DeprecationWarning): + payload = result.to_dict() + with pytest.warns(DeprecationWarning): + ScenarioResult.from_dict(payload) + + +def test_scenario_result_display_group_map_is_public_field(): + scenario_id = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + result = ScenarioResult( + scenario_identifier=scenario_id, + objective_target_identifier=ComponentIdentifier.from_dict({}), + objective_scorer_identifier=None, + attack_results={"crescendo": []}, + display_group_map={"crescendo": "Crescendo Attack"}, + ) + assert "display_group_map" in ScenarioResult.model_fields + assert result.display_group_map == {"crescendo": "Crescendo Attack"} + # Mutable and writable (used by benchmark merge logic). + result.display_group_map["foundry"] = "Foundry" + assert result.display_group_map["foundry"] == "Foundry" diff --git a/tests/unit/output/scenario_result/test_pretty.py b/tests/unit/output/scenario_result/test_pretty.py index 3e5b10a5c2..b2f8cced9c 100644 --- a/tests/unit/output/scenario_result/test_pretty.py +++ b/tests/unit/output/scenario_result/test_pretty.py @@ -35,7 +35,7 @@ def _scenario_result( objective_target_identifier=_target_identifier(**(target_params or {})), attack_results=attack_results or {"strategy_a": [_attack_result()]}, objective_scorer_identifier=objective_scorer_identifier, - display_group_map=display_group_map, + display_group_map=display_group_map or {}, ) diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 10e823d9d8..7355f53b31 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -1138,7 +1138,7 @@ async def test_precomputed_results_injected_into_attack_results(self): # Base run_async produced only the non-skipped attack's result base_scenario_result = MagicMock() base_scenario_result.attack_results = {"technique_c__adv_target_harmbench": [result_z]} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async() @@ -1164,12 +1164,12 @@ async def test_display_group_map_updated_for_cached_attacks(self): base_scenario_result = MagicMock() base_scenario_result.attack_results = {} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async() - assert result._display_group_map["technique_a__adv_target_harmbench"] == "adv_target" + assert result.display_group_map["technique_a__adv_target_harmbench"] == "adv_target" async def test_no_injection_when_no_cached_attacks(self): """When all attacks were executed freshly, attack_results is returned unchanged.""" @@ -1181,7 +1181,7 @@ async def test_no_injection_when_no_cached_attacks(self): result_z = MagicMock(spec=AttackResult) base_scenario_result = MagicMock() base_scenario_result.attack_results = {"technique_c__adv_target_harmbench": [result_z]} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async() diff --git a/tests/unit/score/test_scenario_metrics.py b/tests/unit/score/test_scenario_metrics.py new file mode 100644 index 0000000000..aeef6fb355 --- /dev/null +++ b/tests/unit/score/test_scenario_metrics.py @@ -0,0 +1,46 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +from unittest.mock import MagicMock, patch + +from pyrit.models.scenario_result import ScenarioResult +from pyrit.score.scenario_metrics import get_scenario_metrics + + +def test_get_scenario_metrics_returns_none_without_scorer_identifier(): + scenario_result = MagicMock(spec=ScenarioResult) + scenario_result.objective_scorer_identifier = None + + assert get_scenario_metrics(scenario_result) is None + + +def test_get_scenario_metrics_delegates_to_find_by_eval_hash(): + scenario_result = MagicMock(spec=ScenarioResult) + scenario_result.objective_scorer_identifier = MagicMock() + + metrics = MagicMock() + + with ( + patch("pyrit.score.scenario_metrics.ScorerEvaluationIdentifier") as mock_eval_identifier, + patch("pyrit.score.scenario_metrics.find_objective_metrics_by_eval_hash", return_value=metrics) as mock_find, + ): + mock_eval_identifier.return_value.eval_hash = "abc123" + + result = get_scenario_metrics(scenario_result) + + assert result is metrics + mock_eval_identifier.assert_called_once_with(scenario_result.objective_scorer_identifier) + mock_find.assert_called_once_with(eval_hash="abc123") + + +def test_get_scenario_metrics_returns_none_when_no_metrics_found(): + scenario_result = MagicMock(spec=ScenarioResult) + scenario_result.objective_scorer_identifier = MagicMock() + + with ( + patch("pyrit.score.scenario_metrics.ScorerEvaluationIdentifier") as mock_eval_identifier, + patch("pyrit.score.scenario_metrics.find_objective_metrics_by_eval_hash", return_value=None), + ): + mock_eval_identifier.return_value.eval_hash = "abc123" + + assert get_scenario_metrics(scenario_result) is None From cefaf0272fc41bedd1c3dbf9a33d76df1031d6fe Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 3 Jun 2026 12:27:19 -0700 Subject: [PATCH 2/4] Remove unused get_scenario_metrics dead code The helper had no production callers (the original ScenarioResult method also had none), so relocating it in Phase 7 just moved dead code. Remove the module, its test, and the pyrit.score export. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- pyrit/score/__init__.py | 2 - pyrit/score/scenario_metrics.py | 41 -------------------- tests/unit/score/test_scenario_metrics.py | 46 ----------------------- 3 files changed, 89 deletions(-) delete mode 100644 pyrit/score/scenario_metrics.py delete mode 100644 tests/unit/score/test_scenario_metrics.py diff --git a/pyrit/score/__init__.py b/pyrit/score/__init__.py index 40a1317aee..f406bd795a 100644 --- a/pyrit/score/__init__.py +++ b/pyrit/score/__init__.py @@ -25,7 +25,6 @@ from pyrit.score.float_scale.self_ask_general_float_scale_scorer import SelfAskGeneralFloatScaleScorer from pyrit.score.float_scale.self_ask_likert_scorer import LikertScaleEvalFiles, LikertScalePaths, SelfAskLikertScorer from pyrit.score.float_scale.self_ask_scale_scorer import SelfAskScaleScorer -from pyrit.score.scenario_metrics import get_scenario_metrics from pyrit.score.scorer import Scorer from pyrit.score.scorer_evaluation.metrics_type import MetricsType, RegistryUpdateBehavior from pyrit.score.scorer_evaluation.scorer_metrics import ( @@ -166,7 +165,6 @@ def __getattr__(name: str) -> object: "ScorerMetricsWithIdentity", "get_all_harm_metrics", "get_all_objective_metrics", - "get_scenario_metrics", "find_objective_metrics_by_eval_hash", "ScorerPromptValidator", "SelfAskCategoryScorer", diff --git a/pyrit/score/scenario_metrics.py b/pyrit/score/scenario_metrics.py deleted file mode 100644 index 215da111f3..0000000000 --- a/pyrit/score/scenario_metrics.py +++ /dev/null @@ -1,41 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT license. - -""" -Scorer-evaluation glue for scenario results. - -This lives in ``pyrit.score`` rather than on ``ScenarioResult`` because looking up -evaluation metrics requires the scoring layer; the data model must not depend on it. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from pyrit.models.identifiers.evaluation_identifier import ScorerEvaluationIdentifier -from pyrit.score.scorer_evaluation.scorer_metrics_io import find_objective_metrics_by_eval_hash - -if TYPE_CHECKING: - from pyrit.models.scenario_result import ScenarioResult - from pyrit.score.scorer_evaluation.scorer_metrics import ScorerMetrics - - -def get_scenario_metrics(scenario_result: ScenarioResult) -> ScorerMetrics | None: - """ - Get the evaluation metrics for a scenario's objective scorer. - - Args: - scenario_result (ScenarioResult): The scenario result whose objective scorer - metrics should be looked up. - - Returns: - ScorerMetrics | None: The evaluation metrics object, or None if the scenario - has no objective scorer or no matching metrics are registered. - - """ - if not scenario_result.objective_scorer_identifier: - return None - - eval_hash = ScorerEvaluationIdentifier(scenario_result.objective_scorer_identifier).eval_hash - - return find_objective_metrics_by_eval_hash(eval_hash=eval_hash) diff --git a/tests/unit/score/test_scenario_metrics.py b/tests/unit/score/test_scenario_metrics.py deleted file mode 100644 index aeef6fb355..0000000000 --- a/tests/unit/score/test_scenario_metrics.py +++ /dev/null @@ -1,46 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT license. - -from unittest.mock import MagicMock, patch - -from pyrit.models.scenario_result import ScenarioResult -from pyrit.score.scenario_metrics import get_scenario_metrics - - -def test_get_scenario_metrics_returns_none_without_scorer_identifier(): - scenario_result = MagicMock(spec=ScenarioResult) - scenario_result.objective_scorer_identifier = None - - assert get_scenario_metrics(scenario_result) is None - - -def test_get_scenario_metrics_delegates_to_find_by_eval_hash(): - scenario_result = MagicMock(spec=ScenarioResult) - scenario_result.objective_scorer_identifier = MagicMock() - - metrics = MagicMock() - - with ( - patch("pyrit.score.scenario_metrics.ScorerEvaluationIdentifier") as mock_eval_identifier, - patch("pyrit.score.scenario_metrics.find_objective_metrics_by_eval_hash", return_value=metrics) as mock_find, - ): - mock_eval_identifier.return_value.eval_hash = "abc123" - - result = get_scenario_metrics(scenario_result) - - assert result is metrics - mock_eval_identifier.assert_called_once_with(scenario_result.objective_scorer_identifier) - mock_find.assert_called_once_with(eval_hash="abc123") - - -def test_get_scenario_metrics_returns_none_when_no_metrics_found(): - scenario_result = MagicMock(spec=ScenarioResult) - scenario_result.objective_scorer_identifier = MagicMock() - - with ( - patch("pyrit.score.scenario_metrics.ScorerEvaluationIdentifier") as mock_eval_identifier, - patch("pyrit.score.scenario_metrics.find_objective_metrics_by_eval_hash", return_value=None), - ): - mock_eval_identifier.return_value.eval_hash = "abc123" - - assert get_scenario_metrics(scenario_result) is None From 9fed72a7919a99c24c7cde65c71f3e72036e2371 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 3 Jun 2026 12:43:24 -0700 Subject: [PATCH 3/4] Make ScenarioRunState a str Enum and fix ScoreEntry None handling - Convert ScenarioRunState from a Literal alias to a str-backed Enum (matches house style; backward-compatible at all string call sites). - Move ScenarioResult.normalize_scenario_name to the end (static methods last). - Guard ScoreEntry.__init__ against a None scorer_class_identifier and drop a now-unused type: ignore, fixing ty errors surfaced after merging origin/main. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- pyrit/memory/memory_models.py | 27 ++++++------ pyrit/models/scenario_result.py | 75 +++++++++++++++++++-------------- 2 files changed, 58 insertions(+), 44 deletions(-) diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index 7e852436c8..69c9306457 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -401,14 +401,17 @@ def __init__(self, *, entry: Score) -> None: self.score_rationale = entry.score_rationale self.score_metadata = entry.score_metadata or {} normalized_scorer = entry.scorer_class_identifier - # Ensure eval_hash is set before truncation so it survives the DB round-trip - if normalized_scorer.eval_hash is None: - normalized_scorer = normalized_scorer.with_eval_hash( - ScorerEvaluationIdentifier(normalized_scorer).eval_hash + if normalized_scorer is None: + self.scorer_class_identifier = {} + else: + # Ensure eval_hash is set before truncation so it survives the DB round-trip + if normalized_scorer.eval_hash is None: + normalized_scorer = normalized_scorer.with_eval_hash( + ScorerEvaluationIdentifier(normalized_scorer).eval_hash + ) + self.scorer_class_identifier = normalized_scorer.model_dump( + context={"max_value_length": MAX_IDENTIFIER_VALUE_LENGTH}, ) - self.scorer_class_identifier = normalized_scorer.model_dump( - context={"max_value_length": MAX_IDENTIFIER_VALUE_LENGTH}, - ) self.prompt_request_response_id = entry.message_piece_id if entry.message_piece_id else None self.timestamp = entry.timestamp # Store in both columns for backward compatibility @@ -439,7 +442,7 @@ def get_score(self) -> Score: score_category=self.score_category, score_rationale=self.score_rationale, score_metadata=self.score_metadata, - scorer_class_identifier=scorer_identifier, # type: ignore[ty:invalid-argument-type] + scorer_class_identifier=scorer_identifier, message_piece_id=self.prompt_request_response_id, timestamp=_ensure_utc(self.timestamp), objective=self.objective, @@ -975,8 +978,8 @@ class ScenarioResultEntry(Base): scenario_init_data (dict): Optional initialization parameters used to configure the scenario. objective_target_identifier (dict): Identifier for the target being evaluated in the scenario. objective_scorer_identifier (dict): Optional identifier for the scorer used to evaluate results. - scenario_run_state (Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED"]): Current execution state - of the scenario. + scenario_run_state (str): Current execution state of the scenario + (one of CREATED, IN_PROGRESS, COMPLETED, FAILED, CANCELLED). attack_results_json (str): JSON-serialized dictionary mapping attack names to conversation IDs. Format: {"attack_name": ["conversation_id1", "conversation_id2", ...]}. The full AttackResult objects are stored in AttackResultEntries and can be queried by conversation_id. @@ -1003,9 +1006,7 @@ class ScenarioResultEntry(Base): scenario_init_data: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) objective_target_identifier: Mapped[dict[str, str]] = mapped_column(JSON, nullable=False) objective_scorer_identifier: Mapped[dict[str, str] | None] = mapped_column(JSON, nullable=True) - scenario_run_state: Mapped[Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED"]] = mapped_column( - String, nullable=False, default="CREATED" - ) + scenario_run_state: Mapped[str] = mapped_column(String, nullable=False, default="CREATED") attack_results_json: Mapped[str] = mapped_column(Unicode, nullable=False) display_group_map_json: Mapped[str | None] = mapped_column(Unicode, nullable=True) labels: Mapped[dict[str, str] | None] = mapped_column(JSON, nullable=True) diff --git a/pyrit/models/scenario_result.py b/pyrit/models/scenario_result.py index e511c2bedb..63f832db8d 100644 --- a/pyrit/models/scenario_result.py +++ b/pyrit/models/scenario_result.py @@ -6,7 +6,8 @@ import logging import uuid from datetime import datetime, timezone -from typing import Any, Literal +from enum import Enum +from typing import Any from pydantic import BaseModel, ConfigDict, Field @@ -75,7 +76,19 @@ def from_dict(cls, data: dict[str, Any]) -> ScenarioIdentifier: return cls.model_validate(data) -ScenarioRunState = Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED"] +class ScenarioRunState(str, Enum): + """ + Lifecycle state of a scenario run. + + Inherits from ``str`` so values serialize naturally in Pydantic models and + REST responses, and compare equal to their string form. + """ + + CREATED = "CREATED" + IN_PROGRESS = "IN_PROGRESS" + COMPLETED = "COMPLETED" + FAILED = "FAILED" + CANCELLED = "CANCELLED" class ScenarioResult(BaseModel): @@ -100,7 +113,7 @@ class ScenarioResult(BaseModel): #: Results grouped by atomic attack name. attack_results: dict[str, list[AttackResult]] #: Current scenario run state. - scenario_run_state: ScenarioRunState = "CREATED" + scenario_run_state: ScenarioRunState = ScenarioRunState.CREATED #: Optional labels. labels: dict[str, str] = Field(default_factory=dict) #: When the scenario result was created. @@ -216,34 +229,6 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i successful_results = sum(1 for result in all_results if result.outcome == AttackOutcome.SUCCESS) return int((successful_results / total_results) * 100) - @staticmethod - def normalize_scenario_name(scenario_name: str) -> str: - """ - Normalize a scenario name to match the stored class name format. - - Converts CLI-style snake_case names (e.g., "foundry" or "content_harms") to - PascalCase class names (e.g., "Foundry" or "ContentHarms") for database queries. - If the input is already in PascalCase or doesn't match the snake_case pattern, - it is returned unchanged. - - This is the inverse of ScenarioRegistry._class_name_to_scenario_name(). - - Args: - scenario_name (str): The scenario name to normalize. - - Returns: - str: The normalized scenario name suitable for database queries. - - """ - # Check if it looks like snake_case (contains underscore and is lowercase) - if "_" in scenario_name and scenario_name == scenario_name.lower(): - # Convert snake_case to PascalCase - # e.g., "content_harms" -> "ContentHarms" - parts = scenario_name.split("_") - return "".join(part.capitalize() for part in parts) - # Already PascalCase or other format, return as-is - return scenario_name - def to_dict(self) -> dict[str, Any]: """ Serialize this scenario result to a JSON-compatible dictionary. @@ -279,3 +264,31 @@ def from_dict(cls, data: dict[str, Any]) -> ScenarioResult: removed_in="0.16.0", ) return cls.model_validate(data) + + @staticmethod + def normalize_scenario_name(scenario_name: str) -> str: + """ + Normalize a scenario name to match the stored class name format. + + Converts CLI-style snake_case names (e.g., "foundry" or "content_harms") to + PascalCase class names (e.g., "Foundry" or "ContentHarms") for database queries. + If the input is already in PascalCase or doesn't match the snake_case pattern, + it is returned unchanged. + + This is the inverse of ScenarioRegistry._class_name_to_scenario_name(). + + Args: + scenario_name (str): The scenario name to normalize. + + Returns: + str: The normalized scenario name suitable for database queries. + + """ + # Check if it looks like snake_case (contains underscore and is lowercase) + if "_" in scenario_name and scenario_name == scenario_name.lower(): + # Convert snake_case to PascalCase + # e.g., "content_harms" -> "ContentHarms" + parts = scenario_name.split("_") + return "".join(part.capitalize() for part in parts) + # Already PascalCase or other format, return as-is + return scenario_name From 3f4cf1bc1371875d79a86dc5e2aef6d591dadc0a Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 3 Jun 2026 12:53:44 -0700 Subject: [PATCH 4/4] MAINT Address PR review: hoist test imports, simplify pyrit_version default - Move inline imports in test_scenario_result.py to the top of the file - Simplify pyrit_version from default_factory to a plain default in ScenarioIdentifier and ComponentIdentifier (sibling identifiers) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../identifiers/component_identifier.py | 2 +- pyrit/models/scenario_result.py | 2 +- tests/unit/models/test_scenario_result.py | 19 ++++++------------- 3 files changed, 8 insertions(+), 15 deletions(-) diff --git a/pyrit/models/identifiers/component_identifier.py b/pyrit/models/identifiers/component_identifier.py index 948ad2eeec..d3d0c71933 100644 --- a/pyrit/models/identifiers/component_identifier.py +++ b/pyrit/models/identifiers/component_identifier.py @@ -168,7 +168,7 @@ class ComponentIdentifier(BaseModel): #: may have been truncated. hash: Optional[str] = None #: Version tag for storage. Not included in the content hash. - pyrit_version: str = Field(default_factory=lambda: pyrit.__version__) + pyrit_version: str = Field(default=pyrit.__version__) #: Evaluation hash. Computed by EvaluationIdentifier subclasses and attached #: to the identifier so it survives DB round-trips with truncated params. eval_hash: Optional[str] = None diff --git a/pyrit/models/scenario_result.py b/pyrit/models/scenario_result.py index 63f832db8d..60227b96a4 100644 --- a/pyrit/models/scenario_result.py +++ b/pyrit/models/scenario_result.py @@ -35,7 +35,7 @@ class ScenarioIdentifier(BaseModel): #: Version of the scenario. Accepts the legacy ``scenario_version`` kwarg/wire key. version: int = Field(default=1, alias="scenario_version") #: PyRIT version string. Defaults to the current installed version. - pyrit_version: str = Field(default_factory=lambda: pyrit.__version__) + pyrit_version: str = Field(default=pyrit.__version__) #: Optional initialization data. init_data: dict[str, Any] | None = None diff --git a/tests/unit/models/test_scenario_result.py b/tests/unit/models/test_scenario_result.py index 1bfaf66dac..e15abc53e7 100644 --- a/tests/unit/models/test_scenario_result.py +++ b/tests/unit/models/test_scenario_result.py @@ -2,9 +2,15 @@ # Licensed under the MIT license. import uuid +from datetime import datetime, timezone +import pytest + +import pyrit from pyrit.models import ComponentIdentifier +from pyrit.models.conversation_reference import ConversationReference, ConversationType from pyrit.models.results.attack_result import AttackOutcome, AttackResult +from pyrit.models.retry_event import RetryEvent from pyrit.models.scenario_result import ScenarioIdentifier, ScenarioResult @@ -47,8 +53,6 @@ def test_init_with_all_params(self): assert si.pyrit_version == "1.0.0" def test_init_default_pyrit_version(self): - import pyrit - si = ScenarioIdentifier(name="X") assert si.pyrit_version == pyrit.__version__ @@ -201,11 +205,6 @@ def test_scenario_identifier_to_dict_from_dict_roundtrip(): def test_scenario_result_to_dict_from_dict_roundtrip(): - from datetime import datetime, timezone - - from pyrit.models.conversation_reference import ConversationReference, ConversationType - from pyrit.models.retry_event import RetryEvent - scenario_id = ScenarioIdentifier( name="ContentHarms", description="Tests content harm scenarios", @@ -275,8 +274,6 @@ def test_scenario_result_to_dict_from_dict_roundtrip(): def test_scenario_identifier_from_dict_missing_pyrit_version_uses_current(): """A payload missing pyrit_version now resolves to the current version via the Pydantic default.""" - import pyrit - data = { "name": "Legacy", "description": "loaded from older payload", @@ -308,8 +305,6 @@ def test_scenario_result_from_dict_preserves_missing_completion_time(): def test_scenario_identifier_to_dict_from_dict_emit_deprecation_warnings(): - import pytest - identifier = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") with pytest.warns(DeprecationWarning): payload = identifier.to_dict() @@ -318,8 +313,6 @@ def test_scenario_identifier_to_dict_from_dict_emit_deprecation_warnings(): def test_scenario_result_to_dict_from_dict_emit_deprecation_warnings(): - import pytest - scenario_id = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") result = ScenarioResult( scenario_identifier=scenario_id,