From 39f3c2664004f2d44ba181b6eeb42d7adce6d7e5 Mon Sep 17 00:00:00 2001 From: Utkarsh Bahuguna Date: Thu, 24 Sep 2026 13:23:33 +0530 Subject: [PATCH 1/3] FIX Scenario success rates ignore ERROR attempts recovered by retry or resume --- pyrit/models/results/scenario_result.py | 93 +++++++++++++++++-- pyrit/output/scenario_result/json.py | 3 +- pyrit/output/scenario_result/pretty.py | 3 +- tests/unit/models/test_scenario_result.py | 86 ++++++++++++++++- .../unit/output/scenario_result/test_json.py | 14 +++ 5 files changed, 190 insertions(+), 9 deletions(-) diff --git a/pyrit/models/results/scenario_result.py b/pyrit/models/results/scenario_result.py index e13ca5613e..eb8c883dd2 100644 --- a/pyrit/models/results/scenario_result.py +++ b/pyrit/models/results/scenario_result.py @@ -12,6 +12,8 @@ from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator # Runtime-required by Pydantic field / computed-field annotations. +from pyrit.models.identifiers.atomic_attack_identifier import AtomicAttackIdentifier +from pyrit.models.identifiers.component_identifier import ComponentIdentifier from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier # noqa: TC001 from pyrit.models.identifiers.scorer_identifier import ScorerIdentifier # noqa: TC001 from pyrit.models.identifiers.target_identifier import TargetIdentifier # noqa: TC001 @@ -20,6 +22,61 @@ logger = logging.getLogger(__name__) +def _attempt_unit_key(attack_result: AttackResult) -> str: + """ + Return the execution unit (seed group, falling back to objective) an attack attempt belongs to. + + Mirrors the unit identity the GUI progress read model uses, so retried and resumed attempts of the + same unit can be recognized. + + Returns: + str: A key identifying the unit within its atomic attack. + """ + attribution_data = attack_result.attribution_data + seed_group_id = attribution_data.get("seed_group_id") if isinstance(attribution_data, dict) else None + if seed_group_id: + return f"seed_group:{seed_group_id}" + atomic_identifier = attack_result.atomic_attack_identifier + if isinstance(atomic_identifier, ComponentIdentifier): + typed_identifier = AtomicAttackIdentifier.from_component_identifier(atomic_identifier) + if typed_identifier.seed_identifiers: + return f"seed_group:{typed_identifier.logical_seed_group_id}" + return f"objective:{attack_result.objective}" + + +def _attempt_order_key(attack_result: AttackResult) -> tuple[datetime, str]: + """ + Return a deterministic chronological key for one attack attempt. + + Returns: + tuple[datetime, str]: The attempt timestamp and its attack result ID. + """ + return attack_result.timestamp, str(attack_result.attack_result_id) + + +def _without_superseded_errors(results: list[AttackResult]) -> list[AttackResult]: + """ + Drop ERROR attempts that a later attempt of the same execution unit superseded. + + Non-error results are always kept, and the original order is preserved. + + Returns: + list[AttackResult]: The results without superseded ERROR attempts. + """ + latest_by_unit: dict[str, tuple[datetime, str]] = {} + for result in results: + unit = _attempt_unit_key(result) + order_key = _attempt_order_key(result) + if unit not in latest_by_unit or order_key > latest_by_unit[unit]: + latest_by_unit[unit] = order_key + return [ + result + for result in results + if result.outcome != AttackOutcome.ERROR + or _attempt_order_key(result) == latest_by_unit[_attempt_unit_key(result)] + ] + + __all__ = ["ScenarioResult", "ScenarioRunState"] @@ -167,7 +224,7 @@ def get_techniques_used(self) -> list[str]: """ return list(self.get_display_groups().keys()) - def get_display_groups(self) -> dict[str, list[AttackResult]]: + def get_display_groups(self, *, latest_attempts_only: bool = False) -> dict[str, list[AttackResult]]: """ Aggregate attack results by display group. @@ -176,14 +233,19 @@ def get_display_groups(self) -> dict[str, list[AttackResult]]: merged into a single list. When no map was provided, this returns the same structure as ``attack_results`` (identity mapping). + Args: + latest_attempts_only (bool): When True, superseded ERROR attempts are dropped first + (see ``get_latest_attack_results``). Defaults to False. + Returns: dict[str, list[AttackResult]]: Results grouped by display label. """ + attack_results = self.get_latest_attack_results() if latest_attempts_only else self.attack_results if not self.display_group_map: - return dict(self.attack_results) + return dict(attack_results) grouped: dict[str, list[AttackResult]] = {} - for attack_name, results in self.attack_results.items(): + for attack_name, results in attack_results.items(): group = self.display_group_map.get(attack_name, attack_name) grouped.setdefault(group, []).extend(results) return grouped @@ -218,10 +280,28 @@ def get_objectives(self, *, atomic_attack_name: str | None = None) -> list[str]: return list(set(objectives)) + def get_latest_attack_results(self) -> dict[str, list[AttackResult]]: + """ + Get attack results without superseded ERROR attempts. + + Retries and resumed runs persist a new result for the same execution unit (atomic attack + and seed group, falling back to the objective) and keep the earlier ERROR attempts as + history. This drops ERROR attempts that a later attempt of the same unit superseded, so + recovered errors are not counted, as in the GUI progress view. Non-error results are + always kept. + + Returns: + dict[str, list[AttackResult]]: The remaining results, grouped by atomic attack name. + """ + return {name: _without_superseded_errors(results) for name, results in self.attack_results.items()} + def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> int: """ Get the success rate of this scenario. + ERROR attempts that were later retried or resumed are not counted (see + ``get_latest_attack_results``), so recovered errors do not lower the rate. + Args: 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. @@ -230,15 +310,16 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i int: Success rate as a percentage (0-100). """ + latest_results = self.get_latest_attack_results() if not atomic_attack_name: # Calculate rate across all atomic attacks all_results = [] - for results in self.attack_results.values(): + for results in latest_results.values(): all_results.extend(results) else: # Calculate rate for specific atomic attack - if atomic_attack_name in self.attack_results: - all_results = self.attack_results[atomic_attack_name] + if atomic_attack_name in latest_results: + all_results = latest_results[atomic_attack_name] else: return 0 diff --git a/pyrit/output/scenario_result/json.py b/pyrit/output/scenario_result/json.py index 1dddf0c9ff..00d871bcd5 100644 --- a/pyrit/output/scenario_result/json.py +++ b/pyrit/output/scenario_result/json.py @@ -118,11 +118,12 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]: target = resolve_target_info(result.objective_target_identifier) display_groups = result.get_display_groups() + latest_groups = result.get_display_groups(latest_attempts_only=True) groups = [ { "name": group_name, "num_results": len(group_results), - "success_rate": group_success_rate(group_results), + "success_rate": group_success_rate(latest_groups.get(group_name, [])), } for group_name, group_results in display_groups.items() ] diff --git a/pyrit/output/scenario_result/pretty.py b/pyrit/output/scenario_result/pretty.py index d17fc18d4b..e88687ede0 100644 --- a/pyrit/output/scenario_result/pretty.py +++ b/pyrit/output/scenario_result/pretty.py @@ -214,9 +214,10 @@ async def render_async( lines.append(self._render_section_header("Per-Group Breakdown")) display_groups = result.get_display_groups() + latest_groups = result.get_display_groups(latest_attempts_only=True) group_summaries: list[tuple[str, int, int]] = [ - (group_name, len(group_results), group_success_rate(group_results)) + (group_name, len(group_results), group_success_rate(latest_groups.get(group_name, []))) for group_name, group_results in display_groups.items() ] diff --git a/tests/unit/models/test_scenario_result.py b/tests/unit/models/test_scenario_result.py index 540c183774..cbee4d145d 100644 --- a/tests/unit/models/test_scenario_result.py +++ b/tests/unit/models/test_scenario_result.py @@ -2,7 +2,7 @@ # Licensed under the MIT license. import uuid -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pyrit.models import ( ComponentIdentifier, @@ -145,6 +145,90 @@ def test_objective_achieved_rate_by_name(self): assert sr.objective_achieved_rate(atomic_attack_name="s2") == 0 assert sr.objective_achieved_rate(atomic_attack_name="missing") == 0 + def _attempt(self, *, outcome, objective="obj", minutes=0, seed_group_id=None): + return AttackResult( + conversation_id=str(uuid.uuid4()), + objective=objective, + outcome=outcome, + timestamp=datetime(2026, 1, 1, tzinfo=UTC) + timedelta(minutes=minutes), + attribution_data={"seed_group_id": seed_group_id} if seed_group_id else None, + ) + + def _scenario(self, attack_results): + return make_scenario_result( + scenario_name="TestScenario", + objective_target_identifier=ComponentIdentifier.model_validate({}), + attack_results=attack_results, + objective_scorer_identifier=ComponentIdentifier.model_validate({}), + ) + + def test_objective_achieved_rate_ignores_errors_recovered_by_retry(self): + sr = self._scenario( + { + "s1": [ + self._attempt(outcome=AttackOutcome.SUCCESS, objective="a", minutes=0), + self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=1), + self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=2), + self._attempt(outcome=AttackOutcome.SUCCESS, objective="b", minutes=3), + ] + } + ) + + assert sr.objective_achieved_rate() == 100 + assert sr.objective_achieved_rate(atomic_attack_name="s1") == 100 + assert [r.objective for r in sr.get_latest_attack_results()["s1"]] == ["a", "b"] + + def test_objective_achieved_rate_counts_unrecovered_error_once(self): + sr = self._scenario( + { + "s1": [ + self._attempt(outcome=AttackOutcome.SUCCESS, objective="a", minutes=0), + self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=1), + self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=2), + ] + } + ) + + latest = sr.get_latest_attack_results()["s1"] + assert [r.outcome for r in latest] == [AttackOutcome.SUCCESS, AttackOutcome.ERROR] + assert sr.objective_achieved_rate() == 50 + + def test_errors_are_only_superseded_within_the_same_seed_group(self): + sr = self._scenario( + { + "s1": [ + self._attempt(outcome=AttackOutcome.ERROR, objective="same", minutes=0, seed_group_id="g1"), + self._attempt(outcome=AttackOutcome.SUCCESS, objective="same", minutes=1, seed_group_id="g2"), + ] + } + ) + + assert len(sr.get_latest_attack_results()["s1"]) == 2 + assert sr.objective_achieved_rate() == 50 + + def test_errors_are_only_superseded_within_the_same_atomic_attack(self): + sr = self._scenario( + { + "s1": [self._attempt(outcome=AttackOutcome.ERROR, objective="same", minutes=0)], + "s2": [self._attempt(outcome=AttackOutcome.SUCCESS, objective="same", minutes=1)], + } + ) + + assert sr.objective_achieved_rate() == 50 + + def test_get_display_groups_latest_attempts_only(self): + sr = self._scenario( + { + "s1": [ + self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=0), + self._attempt(outcome=AttackOutcome.SUCCESS, objective="b", minutes=1), + ] + } + ) + + assert len(sr.get_display_groups()["s1"]) == 2 + assert [r.outcome for r in sr.get_display_groups(latest_attempts_only=True)["s1"]] == [AttackOutcome.SUCCESS] + def test_normalize_scenario_name_snake_case(self): assert ScenarioResult.normalize_scenario_name("content_harms") == "ContentHarms" assert ScenarioResult.normalize_scenario_name("foundry") == "foundry" diff --git a/tests/unit/output/scenario_result/test_json.py b/tests/unit/output/scenario_result/test_json.py index 081e87e66e..e9f8e3b1a3 100644 --- a/tests/unit/output/scenario_result/test_json.py +++ b/tests/unit/output/scenario_result/test_json.py @@ -3,6 +3,7 @@ import json import uuid +from datetime import timedelta import pytest from unit.mocks import make_scenario_result @@ -86,6 +87,19 @@ async def test_overview_reports_scenario_and_stats(printer): assert {g["name"] for g in payload["groups"]} == {"technique_a", "technique_b"} +async def test_overview_success_rates_ignore_errors_recovered_by_retry(printer): + error = _attack_result(outcome=AttackOutcome.ERROR, objective="obj") + retried = _attack_result(outcome=AttackOutcome.SUCCESS, objective="obj") + retried.timestamp = error.timestamp + timedelta(seconds=1) + result = _scenario_result(attack_results={"technique_a": [error, retried]}) + + payload = json.loads(await printer.render_async(result)) + + assert payload["stats"]["overall_success_rate"] == 100 + assert payload["stats"]["total_results"] == 2 + assert payload["groups"] == [{"name": "technique_a", "num_results": 2, "success_rate": 100}] + + async def test_overview_prefers_underlying_model_name(printer): result = _scenario_result(target_params={"model_name": "deploy", "underlying_model_name": "gpt-4o"}) payload = json.loads(await printer.render_async(result)) From 7374c79e018c66f20c41de9f57e7cdc54a5eeb4a Mon Sep 17 00:00:00 2001 From: Utkarsh Bahuguna Date: Fri, 25 Sep 2026 02:29:35 +0530 Subject: [PATCH 2/3] Move scenario success statistics into pyrit.analytics Execution-unit identity, latest-attempt selection, counts, denominators and rounding now live in pyrit.analytics.scenario_statistics, with the shared result types in pyrit.models. The backend read model, the pretty/JSON/HTML reports and SDK callers all use it. ScenarioResult.objective_achieved_rate is deprecated because pyrit.models cannot import pyrit.analytics. Adds parity tests that check SDK, run detail, run history, live progress and the JSON report agree on the same saved history, and fixes a test that leaked the SPA mount onto the shared app. --- doc/code/framework.md | 1 + pyproject.toml | 2 + pyrit/analytics/__init__.py | 2 + pyrit/analytics/scenario_statistics.py | 396 ++++++++++++++++++ .../services/scenario_progress_read_model.py | 237 ++--------- pyrit/models/__init__.py | 4 + pyrit/models/results/scenario_result.py | 104 +---- pyrit/models/scenario_progress.py | 38 +- pyrit/output/_derivation.py | 24 +- pyrit/output/scenario_result/json.py | 13 +- pyrit/output/scenario_result/pretty.py | 12 +- .../analytics/test_scenario_statistics.py | 152 +++++++ .../test_scenario_statistics_parity.py | 245 +++++++++++ tests/unit/backend/test_main.py | 28 +- tests/unit/models/test_scenario_result.py | 109 +---- .../unit/output/scenario_result/test_json.py | 14 - .../output/scenario_result/test_pretty.py | 8 +- tests/unit/output/test_derivation.py | 30 +- tests/unit/scenario/core/test_scenario.py | 15 +- 19 files changed, 985 insertions(+), 449 deletions(-) create mode 100644 pyrit/analytics/scenario_statistics.py create mode 100644 tests/unit/analytics/test_scenario_statistics.py create mode 100644 tests/unit/analytics/test_scenario_statistics_parity.py diff --git a/doc/code/framework.md b/doc/code/framework.md index 7d29890f96..77eb1c8e7e 100644 --- a/doc/code/framework.md +++ b/doc/code/framework.md @@ -299,6 +299,7 @@ The below talks about responsibilities of most modules in the PyRIT library - This is where cross-run analysis belongs: e.g. "which attack performed best for this objective?", "how often did a technique succeed?", or "which responses match known content?". - **Does not own**: live, in-attack decisions — any decision made *during* an attack is a scorer's job. Analytics only operates on stored results, after the fact. - Today it includes `ConversationAnalytics` (inspecting conversation history), `analyze_results` / `AttackStats` (aggregating outcomes across techniques), and text-matching strategies (`ExactTextMatching`, `ApproximateTextMatching`). +- `compute_scenario_statistics` is the single calculation of scenario success statistics. It owns execution-unit identity (atomic attack, technique configuration, and seed group), latest-attempt selection, counts, denominators, and rounding. `ScenarioResult`'s SDK callers, the GUI backend, and the console, JSON, and HTML reports all present its results (`ScenarioExecutionStatistics`, `ScenarioExecutionUnit`, and `ScenarioProgressCounts` in `pyrit.models`) instead of calculating their own. - Shared analytics contracts (filters, dimensions, typed values, reports, facets, result pages, and `AttackStats`) live in `pyrit.models.analytics`. They validate data without querying memory or calculating statistics. `AttackResultSelection` defines selection modes without changing existing callers. - Filter-bound cursor and label-normalization helpers live in `pyrit.common.pagination`. The backend pagination module retains compatibility exports, including History's invalid-cursor first-page fallback. diff --git a/pyproject.toml b/pyproject.toml index 6be2b689f2..ed143ab50d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -169,6 +169,8 @@ filterwarnings = [ # Keep the in-repo suite on the scorable/expectation contract. Tests that cover the # shim itself opt back in with pytest.warns. "error:Scorer\\.score_async:DeprecationWarning", + # Rates come from pyrit.analytics; tests that cover the deprecated method opt back in with pytest.warns. + "error:ScenarioResult\\.objective_achieved_rate:DeprecationWarning", ] [tool.ty] diff --git a/pyrit/analytics/__init__.py b/pyrit/analytics/__init__.py index bbbd1def2b..d23e841073 100644 --- a/pyrit/analytics/__init__.py +++ b/pyrit/analytics/__init__.py @@ -11,12 +11,14 @@ if TYPE_CHECKING: from pyrit.analytics.conversation_analytics import ConversationAnalytics from pyrit.analytics.result_analysis import AttackStats, analyze_results, get_cached_results_for_technique + from pyrit.analytics.scenario_statistics import compute_scenario_statistics from pyrit.analytics.text_matching import ApproximateTextMatching, ExactTextMatching, TextMatching _LAZY_EXPORTS: dict[str, str | tuple[str, str | None]] = { "analyze_results": "pyrit.analytics.result_analysis", "ApproximateTextMatching": "pyrit.analytics.text_matching", "AttackStats": "pyrit.analytics.result_analysis", + "compute_scenario_statistics": "pyrit.analytics.scenario_statistics", "ConversationAnalytics": "pyrit.analytics.conversation_analytics", "ExactTextMatching": "pyrit.analytics.text_matching", "get_cached_results_for_technique": "pyrit.analytics.result_analysis", diff --git a/pyrit/analytics/scenario_statistics.py b/pyrit/analytics/scenario_statistics.py new file mode 100644 index 0000000000..48077556c1 --- /dev/null +++ b/pyrit/analytics/scenario_statistics.py @@ -0,0 +1,396 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Effective execution-unit statistics for scenario runs. + +This is the single implementation of scenario success statistics. The SDK (``ScenarioResult``), the +GUI backend, and the console, JSON, and HTML reports all derive their numbers from it, so they cannot +drift apart. It owns: + +- execution-unit identity: an atomic group (atomic attack name plus technique configuration) and a + logical seed group, resolved against the saved run plan when one exists; +- attempt selection: each unit counts once, by its latest attempt (timestamp, then attempt ID); +- counts, denominators, and rounding: the success percentage is succeeded units over completed units, + truncated to an integer. + +Historical attempt, error, and retry counts are reported separately from the effective-unit counts. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +from pydantic import ValidationError + +from pyrit.common.utils import to_sha256 +from pyrit.models import ( + SCENARIO_RUN_PLAN_METADATA_KEY, + AtomicAttackIdentifier, + AttackOutcome, + AttackResult, + ComponentIdentifier, + ScenarioExecutionStatistics, + ScenarioExecutionUnit, + ScenarioProgressCounts, + ScenarioRunPlan, + ScenarioRunPlanAtomicGroup, + config_hash, +) + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping, Sequence + + from pyrit.models import ScenarioResult + +logger = logging.getLogger(__name__) + +#: Default for ``compute_scenario_statistics(plan=...)``: use the plan saved in the result's metadata. +SAVED_RUN_PLAN: Any = object() + + +@dataclass(frozen=True, slots=True) +class ScenarioPlanLookup: + """Pre-indexed run-plan data used while resolving persisted attempts to execution units.""" + + groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] + groups_by_name: dict[str, tuple[ScenarioRunPlanAtomicGroup, ...]] + seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] + planned_units: frozenset[ScenarioExecutionUnit] + + @classmethod + def from_plan(cls, *, plan: ScenarioRunPlan | None) -> ScenarioPlanLookup: + """ + Build constant-time lookup tables for one run plan. + + Returns: + ScenarioPlanLookup: Indexed plan data. + """ + if plan is None: + return cls( + groups_by_identity={}, + groups_by_name={}, + seed_ids_by_group_and_objective={}, + planned_units=frozenset(), + ) + + groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] = {} + grouped_by_name: dict[str, list[ScenarioRunPlanAtomicGroup]] = {} + seeds_by_id = {seed.id: seed for seed in plan.seed_groups} + seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] = {} + planned_units: set[ScenarioExecutionUnit] = set() + for group in plan.atomic_groups: + groups_by_identity[(group.atomic_attack_name, group.technique_eval_hash)] = group + grouped_by_name.setdefault(group.atomic_attack_name, []).append(group) + seed_ids_by_objective: dict[str, list[str]] = {} + for seed_id in group.seed_group_ids: + seed = seeds_by_id[seed_id] + seed_ids_by_objective.setdefault(seed.objective_sha256, []).append(seed_id) + seed_ids_by_group_and_objective.update( + { + (group.id, objective_sha256): tuple(seed_ids) + for objective_sha256, seed_ids in seed_ids_by_objective.items() + } + ) + planned_units.update( + ScenarioExecutionUnit(atomic_group_id=group.id, seed_group_id=seed_group_id) + for seed_group_id in group.seed_group_ids + ) + + return cls( + groups_by_identity=groups_by_identity, + groups_by_name={name: tuple(groups) for name, groups in grouped_by_name.items()}, + seed_ids_by_group_and_objective=seed_ids_by_group_and_objective, + planned_units=frozenset(planned_units), + ) + + def resolve_group( + self, + *, + atomic_attack_name: str, + technique_eval_hash: str | None, + ) -> ScenarioRunPlanAtomicGroup | None: + """ + Resolve one planned group from persisted attribution. + + Returns: + ScenarioRunPlanAtomicGroup | None: The uniquely matching group. + """ + if technique_eval_hash is not None: + return self.groups_by_identity.get((atomic_attack_name, technique_eval_hash)) + matching_groups = self.groups_by_name.get(atomic_attack_name, ()) + return matching_groups[0] if len(matching_groups) == 1 else None + + +@dataclass(frozen=True, slots=True) +class ScenarioAttempt: + """One persisted attempt, resolved to the execution unit it belongs to.""" + + unit: ScenarioExecutionUnit + atomic_attack_name: str + outcome: AttackOutcome + timestamp: datetime + attempt_id: str + total_retries: int + + +def load_scenario_run_plan(scenario_result: ScenarioResult) -> ScenarioRunPlan | None: + """ + Load the run plan saved in a scenario result's metadata. + + Returns: + ScenarioRunPlan | None: The saved plan, or None for results persisted without one. + """ + raw_plan = (scenario_result.metadata or {}).get(SCENARIO_RUN_PLAN_METADATA_KEY) + if raw_plan is None: + return None + try: + return ScenarioRunPlan.model_validate(raw_plan) + except (ValidationError, ValueError): + logger.warning( + "Scenario result %s has an invalid saved run plan; counting it as a legacy run.", scenario_result.id + ) + return None + + +def resolve_execution_unit( + *, + atomic_attack_name: str, + technique_eval_hash: str | None, + attributed_seed_group_id: str | None, + atomic_attack_identifier: AtomicAttackIdentifier | None, + objective: str, + objective_sha256: str | None, + plan_lookup: ScenarioPlanLookup, +) -> ScenarioExecutionUnit: + """ + Resolve one persisted attempt to its execution unit. + + The atomic group is the planned group matching the atomic attack name and technique configuration, + or a hash of both for attempts the plan does not describe. The seed group comes from, in order: the + persisted attribution, the atomic identifier's logical seed group, a unique objective match in the + planned group, and finally a hash of the objective (legacy rows). + + Returns: + ScenarioExecutionUnit: The resolved execution unit. + """ + atomic_group_id = config_hash( + {"atomic_attack_name": atomic_attack_name, "technique_eval_hash": technique_eval_hash or ""} + ) + planned_group = plan_lookup.resolve_group( + atomic_attack_name=atomic_attack_name, + technique_eval_hash=technique_eval_hash, + ) + if planned_group is not None: + atomic_group_id = planned_group.id + + seed_group_id = attributed_seed_group_id or "" + if not seed_group_id and atomic_attack_identifier is not None and atomic_attack_identifier.seed_identifiers: + seed_group_id = atomic_attack_identifier.logical_seed_group_id + if not seed_group_id: + matching_seed_ids = plan_lookup.seed_ids_by_group_and_objective.get( + (atomic_group_id, objective_sha256 or to_sha256(objective)), + (), + ) + if len(matching_seed_ids) == 1: + seed_group_id = matching_seed_ids[0] + if not seed_group_id: + seed_group_id = config_hash({"objective": objective}) + return ScenarioExecutionUnit(atomic_group_id=atomic_group_id, seed_group_id=seed_group_id) + + +def resolve_attack_result_attempt( + *, + atomic_attack_name: str, + attack_result: AttackResult, + plan_lookup: ScenarioPlanLookup, +) -> ScenarioAttempt: + """ + Resolve one hydrated attack result to a scenario attempt. + + Returns: + ScenarioAttempt: The attempt and its execution unit. + """ + attribution_data = attack_result.attribution_data if isinstance(attack_result.attribution_data, dict) else {} + eval_hash = attribution_data.get("parent_eval_hash") + attributed_seed_group_id = attribution_data.get("seed_group_id") + atomic_identifier = attack_result.atomic_attack_identifier + typed_identifier = ( + AtomicAttackIdentifier.from_component_identifier(atomic_identifier) + if isinstance(atomic_identifier, ComponentIdentifier) + else None + ) + # Hydrated rows (and test doubles) may carry a non-integer retry count; treat it as zero. + retries = getattr(attack_result, "total_retries", 0) + return ScenarioAttempt( + unit=resolve_execution_unit( + atomic_attack_name=atomic_attack_name, + technique_eval_hash=str(eval_hash) if eval_hash is not None else None, + attributed_seed_group_id=str(attributed_seed_group_id) if attributed_seed_group_id else None, + atomic_attack_identifier=typed_identifier, + objective=str(attack_result.objective), + objective_sha256=None, + plan_lookup=plan_lookup, + ), + atomic_attack_name=atomic_attack_name, + outcome=attack_result.outcome, + timestamp=_timestamp_order_key(attack_result.timestamp), + attempt_id=str(attack_result.attack_result_id), + total_retries=retries if isinstance(retries, int) else 0, + ) + + +def retry_pressure(*, attempts_per_unit: Iterable[int], persisted_retries: Iterable[int]) -> int: + """ + Combine per-attempt retries with re-attempts of the same execution unit. + + Returns: + int: Total retry pressure. + """ + within_attempts = sum(max(0, retries) for retries in persisted_retries) + repeated_units = sum(max(0, count - 1) for count in attempts_per_unit) + return within_attempts + repeated_units + + +def success_percentage(*, succeeded: int, completed: int) -> int | None: + """ + Return the success percentage for effective execution units. + + Returns: + int | None: ``succeeded / completed`` as a truncated integer percentage, or None with no + completed units. + """ + return int((succeeded / completed) * 100) if completed else None + + +def count_execution_units( + *, + units: Iterable[ScenarioExecutionUnit], + attempts_by_unit: Mapping[ScenarioExecutionUnit, Sequence[Any]], + planned: int | None, +) -> ScenarioProgressCounts: + """ + Count effective execution units from chronologically ordered attempts. + + Each item in ``attempts_by_unit`` must expose ``outcome`` and ``total_retries`` and be ordered + oldest first; the last attempt decides the unit's outcome. + + Returns: + ScenarioProgressCounts: Completed and succeeded units plus historical errors and retries. + """ + completed = 0 + succeeded = 0 + errors = 0 + retries = 0 + for unit in units: + attempts = attempts_by_unit.get(unit, ()) + if not attempts: + continue + completed += 1 + succeeded += int(attempts[-1].outcome == AttackOutcome.SUCCESS) + errors += sum(int(attempt.outcome == AttackOutcome.ERROR) for attempt in attempts) + retries += retry_pressure( + attempts_per_unit=[len(attempts)], + persisted_retries=[attempt.total_retries for attempt in attempts], + ) + return ScenarioProgressCounts( + completed=completed, + planned=planned, + succeeded=succeeded, + success_percentage=success_percentage(succeeded=succeeded, completed=completed), + errors=errors, + retries=retries, + ) + + +def compute_scenario_statistics( + scenario_result: ScenarioResult, + *, + plan: ScenarioRunPlan | None = SAVED_RUN_PLAN, +) -> ScenarioExecutionStatistics: + """ + Calculate effective execution-unit statistics for a scenario result. + + With a run plan, counts cover the planned units, and attempts that match no planned unit are + reported as unattributed. Without one (legacy results), every resolved unit counts and display + groups come from ``display_group_map``. + + Args: + scenario_result (ScenarioResult): The scenario result with its hydrated attack results. + plan (ScenarioRunPlan | None): The run plan, or None to count the result as a legacy run. + Defaults to the plan saved in the result's metadata. + + Returns: + ScenarioExecutionStatistics: Overall, per atomic attack, and per display group counts. + """ + if plan is SAVED_RUN_PLAN: + plan = load_scenario_run_plan(scenario_result) + plan_lookup = ScenarioPlanLookup.from_plan(plan=plan) + + attempts = [ + resolve_attack_result_attempt( + atomic_attack_name=atomic_attack_name, + attack_result=attack_result, + plan_lookup=plan_lookup, + ) + for atomic_attack_name, results in scenario_result.attack_results.items() + for attack_result in results + ] + attempts.sort(key=lambda attempt: (attempt.timestamp, attempt.attempt_id)) + attempts_by_unit: dict[ScenarioExecutionUnit, list[ScenarioAttempt]] = {} + for attempt in attempts: + attempts_by_unit.setdefault(attempt.unit, []).append(attempt) + + units_by_name: dict[str, list[ScenarioExecutionUnit]] = {} + units_by_display_group: dict[str, list[ScenarioExecutionUnit]] = {} + if plan is not None: + for group in plan.atomic_groups: + group_units = [ + ScenarioExecutionUnit(atomic_group_id=group.id, seed_group_id=seed_group_id) + for seed_group_id in group.seed_group_ids + ] + units_by_name.setdefault(group.atomic_attack_name, []).extend(group_units) + units_by_display_group.setdefault(group.display_group, []).extend(group_units) + counted_units = list(dict.fromkeys(unit for units in units_by_name.values() for unit in units)) + planned: int | None = len(counted_units) + else: + for unit, unit_attempts in attempts_by_unit.items(): + name = unit_attempts[0].atomic_attack_name + units_by_name.setdefault(name, []).append(unit) + display_group = scenario_result.display_group_map.get(name, name) + units_by_display_group.setdefault(display_group, []).append(unit) + counted_units = list(attempts_by_unit) + planned = None + + counted = set(counted_units) + unattributed_attempts = sum( + len(unit_attempts) for unit, unit_attempts in attempts_by_unit.items() if unit not in counted + ) + + def _count(units: Sequence[ScenarioExecutionUnit]) -> ScenarioProgressCounts: + return count_execution_units( + units=units, + attempts_by_unit=attempts_by_unit, + planned=len(units) if planned is not None else None, + ) + + return ScenarioExecutionStatistics( + overall=_count(counted_units), + atomic_attacks={name: _count(units) for name, units in units_by_name.items()}, + display_groups={name: _count(units) for name, units in units_by_display_group.items()}, + attempts=len(attempts), + unattributed_attempts=unattributed_attempts, + ) + + +def _timestamp_order_key(timestamp: object) -> datetime: + """ + Normalize potentially malformed timestamps from mutable result objects. + + Returns: + datetime: The timestamp or a stable earliest-time fallback. + """ + return timestamp if isinstance(timestamp, datetime) else datetime.min.replace(tzinfo=UTC) diff --git a/pyrit/backend/services/scenario_progress_read_model.py b/pyrit/backend/services/scenario_progress_read_model.py index 5eb879aa89..e9491b62f5 100644 --- a/pyrit/backend/services/scenario_progress_read_model.py +++ b/pyrit/backend/services/scenario_progress_read_model.py @@ -7,16 +7,21 @@ from collections import OrderedDict from collections.abc import Iterable, Sequence from dataclasses import dataclass, field -from datetime import UTC, datetime from threading import Lock from typing import Literal +from pyrit.analytics.scenario_statistics import ( + ScenarioPlanLookup, + compute_scenario_statistics, + count_execution_units, + resolve_attack_result_attempt, + resolve_execution_unit, + retry_pressure, +) from pyrit.common.utils import to_sha256 from pyrit.memory import AttackResultKeysetCursor from pyrit.memory.memory_interface import MemoryInterface from pyrit.models import ( - AtomicAttackIdentifier, - AttackOutcome, AttackResult, AttackTechniqueIdentifier, ComponentIdentifier, @@ -25,6 +30,7 @@ ScenarioAttackTechniqueDetails, ScenarioComponentIdentity, ScenarioDisplayGroupProgress, + ScenarioExecutionUnit, ScenarioObjectiveScorer, ScenarioObjectiveScorerMetrics, ScenarioProgressCounts, @@ -39,13 +45,17 @@ ScenarioTechniqueProgress, ScorerEvaluationIdentifier, ScorerIdentifier, - config_hash, project_behavioral_identity, ) from pyrit.score.scorer_evaluation.scorer_metrics_io import find_objective_metrics_by_eval_hash logger = logging.getLogger(__name__) +# Execution-unit identity and plan lookup live in ``pyrit.analytics`` so the SDK, backend, and reports +# share one implementation. These names remain importable from here for compatibility. +ResultUnitIdentity = ScenarioExecutionUnit +__all__ = ["ResultUnitIdentity", "ScenarioPlanLookup", "ScenarioProgressReadModel", "ScenarioProgressSnapshot"] + # Technique seeds are rendered as content, so the REST payload carries only the # fields the UI displays. All other narrowing is declared by identifier types and # applied by ``project_behavioral_identity``. @@ -53,14 +63,6 @@ _TECHNIQUE_SEED_DISPLAY_PARAMS = ("value", "data_type") -@dataclass(frozen=True, slots=True) -class ResultUnitIdentity: - """Stable identity of one planned scenario execution unit.""" - - atomic_group_id: str - seed_group_id: str - - @dataclass(frozen=True, slots=True) class ScenarioProgressSnapshot: """Immutable boundary returned after refreshing one run's progress state.""" @@ -92,79 +94,6 @@ class _ProgressCacheEntry: summary_state: _ProgressSummaryState | None = None -@dataclass(frozen=True, slots=True) -class ScenarioPlanLookup: - """Pre-indexed run-plan data used while mapping persisted results.""" - - groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] - groups_by_name: dict[str, tuple[ScenarioRunPlanAtomicGroup, ...]] - seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] - planned_units: frozenset[ResultUnitIdentity] - - @classmethod - def from_plan(cls, *, plan: ScenarioRunPlan | None) -> "ScenarioPlanLookup": - """ - Build constant-time lookup tables for one run plan. - - Returns: - ScenarioPlanLookup: Indexed plan data. - """ - if plan is None: - return cls( - groups_by_identity={}, - groups_by_name={}, - seed_ids_by_group_and_objective={}, - planned_units=frozenset(), - ) - - groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] = {} - grouped_by_name: dict[str, list[ScenarioRunPlanAtomicGroup]] = {} - seeds_by_id = {seed.id: seed for seed in plan.seed_groups} - seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] = {} - planned_units: set[ResultUnitIdentity] = set() - for group in plan.atomic_groups: - groups_by_identity[(group.atomic_attack_name, group.technique_eval_hash)] = group - grouped_by_name.setdefault(group.atomic_attack_name, []).append(group) - seed_ids_by_objective: dict[str, list[str]] = {} - for seed_id in group.seed_group_ids: - seed = seeds_by_id[seed_id] - seed_ids_by_objective.setdefault(seed.objective_sha256, []).append(seed_id) - seed_ids_by_group_and_objective.update( - { - (group.id, objective_sha256): tuple(seed_ids) - for objective_sha256, seed_ids in seed_ids_by_objective.items() - } - ) - planned_units.update( - ResultUnitIdentity(atomic_group_id=group.id, seed_group_id=seed_group_id) - for seed_group_id in group.seed_group_ids - ) - - return cls( - groups_by_identity=groups_by_identity, - groups_by_name={name: tuple(groups) for name, groups in grouped_by_name.items()}, - seed_ids_by_group_and_objective=seed_ids_by_group_and_objective, - planned_units=frozenset(planned_units), - ) - - def resolve_group( - self, - *, - atomic_attack_name: str, - technique_eval_hash: str | None, - ) -> ScenarioRunPlanAtomicGroup | None: - """ - Resolve one planned group from persisted attribution. - - Returns: - ScenarioRunPlanAtomicGroup | None: The uniquely matching group. - """ - if technique_eval_hash is not None: - return self.groups_by_identity.get((atomic_attack_name, technique_eval_hash)) - matching_groups = self.groups_by_name.get(atomic_attack_name, ()) - return matching_groups[0] if len(matching_groups) == 1 else None - - class ScenarioProgressReadModel: """Hydrate, map, cache, and summarize persisted scenario progress.""" @@ -293,42 +222,14 @@ def resolve_result_unit_identity( Returns: ResultUnitIdentity: The atomic-group and seed-group IDs. """ - atomic_identifier = attack_result.atomic_attack_identifier - typed_identifier = ( - AtomicAttackIdentifier.from_component_identifier(atomic_identifier) - if isinstance(atomic_identifier, ComponentIdentifier) - else None - ) - objective = str(attack_result.objective) - attribution_data = attack_result.attribution_data - attributed_seed_group_id = attribution_data.get("seed_group_id") if isinstance(attribution_data, dict) else None - seed_group_id = str(attributed_seed_group_id) if attributed_seed_group_id else "" - if not seed_group_id and typed_identifier is not None and typed_identifier.seed_identifiers: - seed_group_id = typed_identifier.logical_seed_group_id - - atomic_group_id = atomic_attack_name - eval_hash = attribution_data.get("parent_eval_hash") if isinstance(attribution_data, dict) else None - planned_group = plan_lookup.resolve_group( + return resolve_attack_result_attempt( atomic_attack_name=atomic_attack_name, - technique_eval_hash=str(eval_hash) if eval_hash is not None else None, - ) - if planned_group is not None: - atomic_group_id = planned_group.id - if not seed_group_id: - objective_sha256 = to_sha256(objective) - matching_seed_ids = plan_lookup.seed_ids_by_group_and_objective.get( - (planned_group.id, objective_sha256), - (), - ) - if len(matching_seed_ids) == 1: - seed_group_id = matching_seed_ids[0] - if not seed_group_id: - seed_group_id = config_hash({"objective": objective}) - return ResultUnitIdentity(atomic_group_id=atomic_group_id, seed_group_id=seed_group_id) + attack_result=attack_result, + plan_lookup=plan_lookup, + ).unit - @classmethod + @staticmethod def calculate_progress_counts( - cls, *, scenario_result: ScenarioResult, plan: ScenarioRunPlan | None, @@ -337,46 +238,15 @@ def calculate_progress_counts( """ Calculate planned-unit totals without inflating retries or error attempts. + Delegates to ``pyrit.analytics.scenario_statistics`` so run details match the SDK and reports. + Returns: tuple[int, int, int, int]: Total, completed, success-rate percentage, and successful-unit count. """ - latest_result_by_unit: dict[ResultUnitIdentity, AttackResult] = {} - for atomic_attack_name, results in scenario_result.attack_results.items(): - for attack_result in results: - unit_identity = cls.resolve_result_unit_identity( - atomic_attack_name=atomic_attack_name, - attack_result=attack_result, - plan_lookup=plan_lookup, - ) - previous = latest_result_by_unit.get(unit_identity) - if previous is None or cls._result_order_key(attack_result) > cls._result_order_key(previous): - latest_result_by_unit[unit_identity] = attack_result - - planned_units = plan_lookup.planned_units if plan is not None else frozenset(latest_result_by_unit) - total = len(planned_units) - completed_results = [result for unit, result in latest_result_by_unit.items() if unit in planned_units] - completed = len(completed_results) - succeeded = sum(result.outcome == AttackOutcome.SUCCESS for result in completed_results) - rate = int((succeeded / completed) * 100) if completed else 0 - return total, completed, rate, succeeded - - @staticmethod - def _result_order_key(attack_result: AttackResult) -> tuple[datetime, str]: - """Return a deterministic chronological key for one hydrated result attempt.""" - return ScenarioProgressReadModel._timestamp_order_key(attack_result.timestamp), str( - attack_result.attack_result_id - ) - - @staticmethod - def _timestamp_order_key(timestamp: object) -> datetime: - """ - Normalize potentially malformed timestamps from mutable result objects. - - Returns: - datetime: The timestamp or a stable earliest-time fallback. - """ - return timestamp if isinstance(timestamp, datetime) else datetime.min.replace(tzinfo=UTC) + overall = compute_scenario_statistics(scenario_result, plan=plan).overall + total = overall.planned if overall.planned is not None else overall.completed + return total, overall.completed, overall.success_percentage or 0, overall.succeeded @staticmethod def total_retry_pressure(*, attempts_per_unit: Iterable[int], persisted_retries: Iterable[int]) -> int: @@ -386,9 +256,7 @@ def total_retry_pressure(*, attempts_per_unit: Iterable[int], persisted_retries: Returns: int: Total retry pressure. """ - within_attempts = sum(max(0, retries) for retries in persisted_retries) - repeated_units = sum(max(0, count - 1) for count in attempts_per_unit) - return within_attempts + repeated_units + return retry_pressure(attempts_per_unit=attempts_per_unit, persisted_retries=persisted_retries) @staticmethod def _build_technique_details_by_group( @@ -443,28 +311,7 @@ def _build_progress_summary( attempts_by_unit.setdefault(identity, []).append(result) def aggregate(*, units: Sequence[ResultUnitIdentity], planned: int | None) -> ScenarioProgressCounts: - completed = 0 - succeeded = 0 - errors = 0 - retries = 0 - for unit in units: - attempts = attempts_by_unit.get(unit, []) - if attempts: - completed += 1 - succeeded += int(attempts[-1].outcome == AttackOutcome.SUCCESS) - errors += sum(int(attempt.outcome == AttackOutcome.ERROR) for attempt in attempts) - retries += ScenarioProgressReadModel.total_retry_pressure( - attempts_per_unit=[len(attempts)], - persisted_retries=[attempt.total_retries for attempt in attempts], - ) - return ScenarioProgressCounts( - completed=completed, - planned=planned, - succeeded=succeeded, - success_percentage=int((succeeded / completed) * 100) if completed else None, - errors=errors, - retries=retries, - ) + return count_execution_units(units=units, attempts_by_unit=attempts_by_unit, planned=planned) group_units: dict[str, list[ResultUnitIdentity]] = { group.id: [ @@ -769,32 +616,18 @@ def _map_progress_delta( """ atomic_attack_name = str(delta.attribution_data.get("parent_collection") or "") eval_hash = delta.attribution_data.get("parent_eval_hash") - atomic_group_id = config_hash( - {"atomic_attack_name": atomic_attack_name, "technique_eval_hash": eval_hash or ""} - ) - planned_group = plan_lookup.resolve_group( + attributed_seed_group_id = delta.attribution_data.get("seed_group_id") + unit = resolve_execution_unit( atomic_attack_name=atomic_attack_name, technique_eval_hash=str(eval_hash) if eval_hash is not None else None, + attributed_seed_group_id=str(attributed_seed_group_id) if attributed_seed_group_id else None, + atomic_attack_identifier=delta.atomic_attack_identifier, + objective=delta.objective, + objective_sha256=delta.objective_sha256, + plan_lookup=plan_lookup, ) - if planned_group is not None: - atomic_group_id = planned_group.id - attributed_seed_group_id = delta.attribution_data.get("seed_group_id") - seed_group_id = str(attributed_seed_group_id) if attributed_seed_group_id else "" - if ( - not seed_group_id - and delta.atomic_attack_identifier is not None - and delta.atomic_attack_identifier.seed_identifiers - ): - seed_group_id = delta.atomic_attack_identifier.logical_seed_group_id - if not seed_group_id and delta.objective_sha256: - matching_seed_ids = plan_lookup.seed_ids_by_group_and_objective.get( - (atomic_group_id, delta.objective_sha256), - (), - ) - if len(matching_seed_ids) == 1: - seed_group_id = matching_seed_ids[0] - if not seed_group_id: - seed_group_id = config_hash({"objective": delta.objective}) + atomic_group_id = unit.atomic_group_id + seed_group_id = unit.seed_group_id return ScenarioProgressResult( attack_result_id=delta.attack_result_id, conversation_id=delta.conversation_id, diff --git a/pyrit/models/__init__.py b/pyrit/models/__init__.py index 323679e361..e2bf83c91d 100644 --- a/pyrit/models/__init__.py +++ b/pyrit/models/__init__.py @@ -142,6 +142,8 @@ ScenarioAttackTechniqueDetails, ScenarioComponentIdentity, ScenarioDisplayGroupProgress, + ScenarioExecutionStatistics, + ScenarioExecutionUnit, ScenarioObjectiveScorer, ScenarioObjectiveScorerMetrics, ScenarioProgressCounts, @@ -370,6 +372,8 @@ "ScenarioAttackTechniqueDetails": "pyrit.models.scenario_progress", "ScenarioComponentIdentity": "pyrit.models.scenario_progress", "ScenarioDisplayGroupProgress": "pyrit.models.scenario_progress", + "ScenarioExecutionStatistics": "pyrit.models.scenario_progress", + "ScenarioExecutionUnit": "pyrit.models.scenario_progress", "ScenarioObjectiveScorer": "pyrit.models.scenario_progress", "ScenarioObjectiveScorerMetrics": "pyrit.models.scenario_progress", "ScenarioProgressCounts": "pyrit.models.scenario_progress", diff --git a/pyrit/models/results/scenario_result.py b/pyrit/models/results/scenario_result.py index eb8c883dd2..542cf12b05 100644 --- a/pyrit/models/results/scenario_result.py +++ b/pyrit/models/results/scenario_result.py @@ -11,9 +11,9 @@ from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator +from pyrit.common.deprecation import print_deprecation_message + # Runtime-required by Pydantic field / computed-field annotations. -from pyrit.models.identifiers.atomic_attack_identifier import AtomicAttackIdentifier -from pyrit.models.identifiers.component_identifier import ComponentIdentifier from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier # noqa: TC001 from pyrit.models.identifiers.scorer_identifier import ScorerIdentifier # noqa: TC001 from pyrit.models.identifiers.target_identifier import TargetIdentifier # noqa: TC001 @@ -22,61 +22,6 @@ logger = logging.getLogger(__name__) -def _attempt_unit_key(attack_result: AttackResult) -> str: - """ - Return the execution unit (seed group, falling back to objective) an attack attempt belongs to. - - Mirrors the unit identity the GUI progress read model uses, so retried and resumed attempts of the - same unit can be recognized. - - Returns: - str: A key identifying the unit within its atomic attack. - """ - attribution_data = attack_result.attribution_data - seed_group_id = attribution_data.get("seed_group_id") if isinstance(attribution_data, dict) else None - if seed_group_id: - return f"seed_group:{seed_group_id}" - atomic_identifier = attack_result.atomic_attack_identifier - if isinstance(atomic_identifier, ComponentIdentifier): - typed_identifier = AtomicAttackIdentifier.from_component_identifier(atomic_identifier) - if typed_identifier.seed_identifiers: - return f"seed_group:{typed_identifier.logical_seed_group_id}" - return f"objective:{attack_result.objective}" - - -def _attempt_order_key(attack_result: AttackResult) -> tuple[datetime, str]: - """ - Return a deterministic chronological key for one attack attempt. - - Returns: - tuple[datetime, str]: The attempt timestamp and its attack result ID. - """ - return attack_result.timestamp, str(attack_result.attack_result_id) - - -def _without_superseded_errors(results: list[AttackResult]) -> list[AttackResult]: - """ - Drop ERROR attempts that a later attempt of the same execution unit superseded. - - Non-error results are always kept, and the original order is preserved. - - Returns: - list[AttackResult]: The results without superseded ERROR attempts. - """ - latest_by_unit: dict[str, tuple[datetime, str]] = {} - for result in results: - unit = _attempt_unit_key(result) - order_key = _attempt_order_key(result) - if unit not in latest_by_unit or order_key > latest_by_unit[unit]: - latest_by_unit[unit] = order_key - return [ - result - for result in results - if result.outcome != AttackOutcome.ERROR - or _attempt_order_key(result) == latest_by_unit[_attempt_unit_key(result)] - ] - - __all__ = ["ScenarioResult", "ScenarioRunState"] @@ -224,7 +169,7 @@ def get_techniques_used(self) -> list[str]: """ return list(self.get_display_groups().keys()) - def get_display_groups(self, *, latest_attempts_only: bool = False) -> dict[str, list[AttackResult]]: + def get_display_groups(self) -> dict[str, list[AttackResult]]: """ Aggregate attack results by display group. @@ -233,19 +178,14 @@ def get_display_groups(self, *, latest_attempts_only: bool = False) -> dict[str, merged into a single list. When no map was provided, this returns the same structure as ``attack_results`` (identity mapping). - Args: - latest_attempts_only (bool): When True, superseded ERROR attempts are dropped first - (see ``get_latest_attack_results``). Defaults to False. - Returns: dict[str, list[AttackResult]]: Results grouped by display label. """ - attack_results = self.get_latest_attack_results() if latest_attempts_only else self.attack_results if not self.display_group_map: - return dict(attack_results) + return dict(self.attack_results) grouped: dict[str, list[AttackResult]] = {} - for attack_name, results in attack_results.items(): + for attack_name, results in self.attack_results.items(): group = self.display_group_map.get(attack_name, attack_name) grouped.setdefault(group, []).extend(results) return grouped @@ -280,28 +220,10 @@ def get_objectives(self, *, atomic_attack_name: str | None = None) -> list[str]: return list(set(objectives)) - def get_latest_attack_results(self) -> dict[str, list[AttackResult]]: - """ - Get attack results without superseded ERROR attempts. - - Retries and resumed runs persist a new result for the same execution unit (atomic attack - and seed group, falling back to the objective) and keep the earlier ERROR attempts as - history. This drops ERROR attempts that a later attempt of the same unit superseded, so - recovered errors are not counted, as in the GUI progress view. Non-error results are - always kept. - - Returns: - dict[str, list[AttackResult]]: The remaining results, grouped by atomic attack name. - """ - return {name: _without_superseded_errors(results) for name, results in self.attack_results.items()} - def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> int: """ Get the success rate of this scenario. - ERROR attempts that were later retried or resumed are not counted (see - ``get_latest_attack_results``), so recovered errors do not lower the rate. - Args: 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. @@ -309,17 +231,25 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i Returns: int: Success rate as a percentage (0-100). + .. deprecated:: 1.2.0 + Counts every persisted attempt, including ERROR attempts that were later retried or resumed. + Use ``pyrit.analytics.compute_scenario_statistics(result)``, which counts each execution unit + once and is shared with the GUI backend and the reports. Removed in 1.4.0. """ - latest_results = self.get_latest_attack_results() + print_deprecation_message( + old_item="ScenarioResult.objective_achieved_rate", + new_item="pyrit.analytics.compute_scenario_statistics", + removed_in="1.4.0", + ) if not atomic_attack_name: # Calculate rate across all atomic attacks all_results = [] - for results in latest_results.values(): + for results in self.attack_results.values(): all_results.extend(results) else: # Calculate rate for specific atomic attack - if atomic_attack_name in latest_results: - all_results = latest_results[atomic_attack_name] + if atomic_attack_name in self.attack_results: + all_results = self.attack_results[atomic_attack_name] else: return 0 diff --git a/pyrit/models/scenario_progress.py b/pyrit/models/scenario_progress.py index 8e9ef9bd95..8397fbb285 100644 --- a/pyrit/models/scenario_progress.py +++ b/pyrit/models/scenario_progress.py @@ -6,7 +6,7 @@ from datetime import datetime from typing import Any, Literal -from pydantic import AwareDatetime, BaseModel, Field, model_validator +from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator from pyrit.models.catalog.scenario import ScenarioOverloadSummary, ScenarioTargetSummary # noqa: TC001 from pyrit.models.identifiers.atomic_attack_identifier import AtomicAttackIdentifier @@ -167,6 +167,42 @@ class ScenarioProgressCounts(BaseModel): retries: int = Field(..., ge=0) +class ScenarioExecutionUnit(BaseModel): + """ + Identity of one scenario execution unit. + + ``atomic_group_id`` identifies the atomic attack together with its technique configuration, so + two configurations that share an atomic attack name are separate units. ``seed_group_id`` + identifies the logical seed group within it. + """ + + model_config = ConfigDict(frozen=True) + + atomic_group_id: str + seed_group_id: str + + +class ScenarioExecutionStatistics(BaseModel): + """ + Effective execution-unit statistics for one scenario run, calculated by ``pyrit.analytics``. + + Each execution unit counts once, by its latest attempt, so recovered errors do not lower the success + percentage. ``attempts`` and the ``errors`` and ``retries`` of each count keep the historical attempt + history separately from the effective-unit statistics. + """ + + #: Counts across every counted execution unit. + overall: ScenarioProgressCounts + #: Counts keyed by atomic attack name (all technique configurations that share the name). + atomic_attacks: dict[str, ScenarioProgressCounts] = Field(default_factory=dict) + #: Counts keyed by display group label. + display_groups: dict[str, ScenarioProgressCounts] = Field(default_factory=dict) + #: Total persisted attempts, including superseded ones. + attempts: int = Field(default=0, ge=0) + #: Attempts that matched no planned execution unit and are excluded from the counts. + unattributed_attempts: int = Field(default=0, ge=0) + + class ScenarioTechniqueProgress(ScenarioProgressCounts): """Progress for one scenario technique.""" diff --git a/pyrit/output/_derivation.py b/pyrit/output/_derivation.py index 0280a81ff4..f1d143a5f4 100644 --- a/pyrit/output/_derivation.py +++ b/pyrit/output/_derivation.py @@ -14,8 +14,6 @@ from typing import TYPE_CHECKING, NamedTuple -from pyrit.models import AttackOutcome - if TYPE_CHECKING: from pyrit.models import AttackResult, ComponentIdentifier, ScenarioResult, Score @@ -53,21 +51,25 @@ def resolve_target_info(target_id: ComponentIdentifier | None) -> TargetInfo: ) -def group_success_rate(attacks: list[AttackResult]) -> int: +def scenario_success_rates(result: ScenarioResult) -> tuple[int, dict[str, int]]: """ - Return the percentage of *attacks* whose outcome is SUCCESS (0 when empty). + Return the overall and per-display-group success rates of a scenario result. + + The numbers come from ``pyrit.analytics.compute_scenario_statistics``, the calculation shared with + the SDK and the GUI backend, so every report shows the same effective execution-unit statistics. + Groups with no completed unit report 0. Args: - attacks (list[AttackResult]): The attacks to score. + result (ScenarioResult): The scenario result to summarize. Returns: - int: The success rate as an integer percent. + tuple[int, dict[str, int]]: The overall rate and the rate for each display group, as integer percents. """ - total = len(attacks) - if not total: - return 0 - successful = sum(1 for attack in attacks if attack.outcome == AttackOutcome.SUCCESS) - return int((successful / total) * 100) + from pyrit.analytics.scenario_statistics import compute_scenario_statistics + + statistics = compute_scenario_statistics(result) + group_rates = {name: counts.success_percentage or 0 for name, counts in statistics.display_groups.items()} + return statistics.overall.success_percentage or 0, group_rates def attack_score_display(attack: AttackResult, *, none_value: str | None = None) -> str | None: diff --git a/pyrit/output/scenario_result/json.py b/pyrit/output/scenario_result/json.py index 00d871bcd5..d002c4b078 100644 --- a/pyrit/output/scenario_result/json.py +++ b/pyrit/output/scenario_result/json.py @@ -5,7 +5,12 @@ from typing import TYPE_CHECKING, Any from pyrit.models import AttackResult, ScenarioResult -from pyrit.output._derivation import attack_score_display, group_success_rate, resolve_target_info, select_attacks +from pyrit.output._derivation import ( + attack_score_display, + resolve_target_info, + scenario_success_rates, + select_attacks, +) from pyrit.output.scenario_result.base import ScenarioResultPrinterBase, ScenarioView from pyrit.output.sink import Sink @@ -117,13 +122,13 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]: """ target = resolve_target_info(result.objective_target_identifier) + overall_rate, group_rates = scenario_success_rates(result) display_groups = result.get_display_groups() - latest_groups = result.get_display_groups(latest_attempts_only=True) groups = [ { "name": group_name, "num_results": len(group_results), - "success_rate": group_success_rate(latest_groups.get(group_name, [])), + "success_rate": group_rates.get(group_name, 0), } for group_name, group_results in display_groups.items() ] @@ -154,7 +159,7 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]: "stats": { "total_techniques": len(result.get_techniques_used()), "total_results": sum(len(results) for results in result.attack_results.values()), - "overall_success_rate": result.objective_achieved_rate(), + "overall_success_rate": overall_rate, "unique_objectives": len(result.get_objectives()), }, "groups": groups, diff --git a/pyrit/output/scenario_result/pretty.py b/pyrit/output/scenario_result/pretty.py index e88687ede0..242f889c47 100644 --- a/pyrit/output/scenario_result/pretty.py +++ b/pyrit/output/scenario_result/pretty.py @@ -6,7 +6,12 @@ from colorama import Fore, Style from pyrit.models import AttackOutcome, ScenarioResult -from pyrit.output._derivation import attack_score_display, group_success_rate, resolve_target_info, select_attacks +from pyrit.output._derivation import ( + attack_score_display, + resolve_target_info, + scenario_success_rates, + select_attacks, +) from pyrit.output._formatting import _PrettyPrinterMixin from pyrit.output.scenario_result.base import ScenarioResultPrinterBase, ScenarioView from pyrit.output.scorer.base import ScorerPrinterBase @@ -198,7 +203,7 @@ async def render_async( lines.append(self._render_section_header("Overall Statistics")) total_results = sum(len(results) for results in result.attack_results.values()) total_techniques = len(result.get_techniques_used()) - overall_rate = result.objective_achieved_rate() + overall_rate, group_rates = scenario_success_rates(result) lines.append(self._format_colored(f"{self._indent}📈 Summary", Style.BRIGHT)) lines.append(self._format_colored(f"{self._indent * 2}• Total Techniques: {total_techniques}", Fore.GREEN)) @@ -214,10 +219,9 @@ async def render_async( lines.append(self._render_section_header("Per-Group Breakdown")) display_groups = result.get_display_groups() - latest_groups = result.get_display_groups(latest_attempts_only=True) group_summaries: list[tuple[str, int, int]] = [ - (group_name, len(group_results), group_success_rate(latest_groups.get(group_name, []))) + (group_name, len(group_results), group_rates.get(group_name, 0)) for group_name, group_results in display_groups.items() ] diff --git a/tests/unit/analytics/test_scenario_statistics.py b/tests/unit/analytics/test_scenario_statistics.py new file mode 100644 index 0000000000..dc652e3915 --- /dev/null +++ b/tests/unit/analytics/test_scenario_statistics.py @@ -0,0 +1,152 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import uuid +from datetime import UTC, datetime, timedelta + +from pyrit.analytics.scenario_statistics import ( + ScenarioPlanLookup, + compute_scenario_statistics, + resolve_execution_unit, +) +from pyrit.common.utils import to_sha256 +from pyrit.models import ( + SCENARIO_RUN_PLAN_METADATA_KEY, + AttackOutcome, + AttackResult, + ScenarioRunPlan, + ScenarioRunPlanAtomicGroup, + ScenarioRunPlanSeedGroup, + config_hash, +) +from unit.mocks import make_scenario_result + +_T0 = datetime(2026, 9, 1, tzinfo=UTC) + + +def _result(*, objective: str, outcome: AttackOutcome, seconds: int = 0, **attribution: str) -> AttackResult: + return AttackResult( + conversation_id=str(uuid.uuid4()), + objective=objective, + outcome=outcome, + timestamp=_T0 + timedelta(seconds=seconds), + attribution_data={"parent_collection": "attack", **attribution} if attribution else None, + ) + + +def _plan() -> ScenarioRunPlan: + return ScenarioRunPlan( + atomic_groups=[ + ScenarioRunPlanAtomicGroup( + id="group", + atomic_attack_name="attack", + display_group="Attack", + technique_eval_hash="eval", + seed_group_ids=["seed-a", "seed-b"], + ) + ], + seed_groups=[ + ScenarioRunPlanSeedGroup(id="seed-a", objective_sha256=to_sha256("A"), objective="A"), + ScenarioRunPlanSeedGroup(id="seed-b", objective_sha256=to_sha256("B"), objective="B"), + ], + ) + + +def test_latest_attempt_decides_each_unit() -> None: + result = make_scenario_result( + attack_results={ + "attack": [ + _result(objective="A", outcome=AttackOutcome.SUCCESS, seconds=0), + _result(objective="A", outcome=AttackOutcome.ERROR, seconds=1), + _result(objective="B", outcome=AttackOutcome.ERROR, seconds=2), + _result(objective="B", outcome=AttackOutcome.SUCCESS, seconds=3), + ] + } + ) + + statistics = compute_scenario_statistics(result) + + assert statistics.overall.completed == 2 + assert statistics.overall.succeeded == 1 + assert statistics.overall.success_percentage == 50 + assert statistics.overall.errors == 2 + assert statistics.attempts == 4 + + +def test_empty_result_has_no_success_percentage() -> None: + statistics = compute_scenario_statistics(make_scenario_result(attack_results={"attack": []})) + + assert statistics.overall.completed == 0 + assert statistics.overall.success_percentage is None + assert statistics.attempts == 0 + + +def test_saved_plan_counts_planned_units_and_reports_unattributed_attempts() -> None: + result = make_scenario_result( + attack_results={ + "attack": [ + _result(objective="A", outcome=AttackOutcome.SUCCESS, parent_eval_hash="eval", seed_group_id="seed-a"), + _result(objective="Z", outcome=AttackOutcome.SUCCESS, parent_eval_hash="other-eval"), + ] + }, + metadata={SCENARIO_RUN_PLAN_METADATA_KEY: _plan().model_dump(mode="json")}, + ) + + statistics = compute_scenario_statistics(result) + + assert statistics.overall.planned == 2 + assert statistics.overall.completed == 1 + assert statistics.overall.success_percentage == 100 + assert statistics.unattributed_attempts == 1 + assert statistics.display_groups["Attack"].planned == 2 + + +def test_explicit_none_plan_ignores_saved_plan() -> None: + result = make_scenario_result( + attack_results={"attack": [_result(objective="A", outcome=AttackOutcome.SUCCESS)]}, + metadata={SCENARIO_RUN_PLAN_METADATA_KEY: _plan().model_dump(mode="json")}, + ) + + statistics = compute_scenario_statistics(result, plan=None) + + assert statistics.overall.planned is None + assert statistics.overall.completed == 1 + + +def test_invalid_saved_plan_counts_as_legacy_run(caplog) -> None: + result = make_scenario_result( + attack_results={"attack": [_result(objective="A", outcome=AttackOutcome.SUCCESS)]}, + metadata={SCENARIO_RUN_PLAN_METADATA_KEY: {"atomic_groups": "invalid"}}, + ) + + statistics = compute_scenario_statistics(result) + + assert statistics.overall.planned is None + assert statistics.overall.success_percentage == 100 + assert "invalid saved run plan" in caplog.text + + +def test_resolve_execution_unit_precedence() -> None: + lookup = ScenarioPlanLookup.from_plan(plan=_plan()) + + def resolve(**kwargs): + defaults = { + "atomic_attack_name": "attack", + "technique_eval_hash": "eval", + "attributed_seed_group_id": None, + "atomic_attack_identifier": None, + "objective": "A", + "objective_sha256": None, + "plan_lookup": lookup, + } + return resolve_execution_unit(**{**defaults, **kwargs}) + + # Attribution wins, then a unique objective match in the planned group, then an objective hash. + assert resolve(attributed_seed_group_id="seed-b").seed_group_id == "seed-b" + assert resolve().seed_group_id == "seed-a" + assert resolve(objective="unplanned").seed_group_id == config_hash({"objective": "unplanned"}) + # Planned groups keep their plan ID; unplanned configurations get a name-and-hash identity. + assert resolve().atomic_group_id == "group" + assert resolve(technique_eval_hash="other").atomic_group_id == config_hash( + {"atomic_attack_name": "attack", "technique_eval_hash": "other"} + ) diff --git a/tests/unit/analytics/test_scenario_statistics_parity.py b/tests/unit/analytics/test_scenario_statistics_parity.py new file mode 100644 index 0000000000..cd2bcc56f6 --- /dev/null +++ b/tests/unit/analytics/test_scenario_statistics_parity.py @@ -0,0 +1,245 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Parity coverage for scenario success statistics. + +The same saved history must produce identical effective-unit statistics through the SDK +(``compute_scenario_statistics``), the GUI API (run detail, run history list, and live progress), +and the reports (JSON printer). +""" + +import json +import uuid +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta + +import pytest + +from pyrit.analytics import compute_scenario_statistics +from pyrit.backend.services.scenario_run_service import ScenarioRunService +from pyrit.common.utils import to_sha256 +from pyrit.memory import MemoryInterface +from pyrit.models import ( + SCENARIO_RUN_PLAN_METADATA_KEY, + AttackOutcome, + AttackResult, + ComponentIdentifier, + ScenarioRunPlan, + ScenarioRunPlanAtomicGroup, + ScenarioRunPlanSeedGroup, + ScenarioRunState, +) +from pyrit.output.scenario_result.json import JsonScenarioResultPrinter +from unit.mocks import make_scenario_result + +_T0 = datetime(2026, 9, 1, tzinfo=UTC) + + +@dataclass(frozen=True) +class _Attempt: + atomic_attack_name: str + objective: str + outcome: AttackOutcome + eval_hash: str | None = "eval" + seed_group_id: str | None = None + + +@dataclass(frozen=True) +class _History: + attempts: list[_Attempt] + plan: ScenarioRunPlan | None = None + display_group_map: dict[str, str] = field(default_factory=dict) + + +def _group(*, name: str, eval_hash: str, seed_ids: list[str], display_group: str | None = None): + return ScenarioRunPlanAtomicGroup( + id=f"{name}-{eval_hash}", + atomic_attack_name=name, + display_group=display_group or name, + technique_eval_hash=eval_hash, + seed_group_ids=seed_ids, + ) + + +def _seed(seed_id: str, objective: str) -> ScenarioRunPlanSeedGroup: + return ScenarioRunPlanSeedGroup(id=seed_id, objective_sha256=to_sha256(objective), objective=objective) + + +def _plan(*groups: ScenarioRunPlanAtomicGroup, seeds: list[ScenarioRunPlanSeedGroup]) -> ScenarioRunPlan: + return ScenarioRunPlan(scenario_registry_name="test.scenario", atomic_groups=list(groups), seed_groups=seeds) + + +_HISTORIES = { + "retry_and_resume_recovered": _History( + plan=_plan( + _group(name="attack", eval_hash="eval", seed_ids=["a", "b"]), seeds=[_seed("a", "A"), _seed("b", "B")] + ), + attempts=[ + _Attempt("attack", "A", AttackOutcome.SUCCESS, seed_group_id="a"), + _Attempt("attack", "B", AttackOutcome.ERROR, seed_group_id="b"), + _Attempt("attack", "B", AttackOutcome.ERROR, seed_group_id="b"), + _Attempt("attack", "B", AttackOutcome.SUCCESS, seed_group_id="b"), + ], + ), + "unrecovered_errors": _History( + plan=_plan( + _group(name="attack", eval_hash="eval", seed_ids=["a", "b"]), seeds=[_seed("a", "A"), _seed("b", "B")] + ), + attempts=[ + _Attempt("attack", "A", AttackOutcome.SUCCESS, seed_group_id="a"), + _Attempt("attack", "B", AttackOutcome.ERROR, seed_group_id="b"), + _Attempt("attack", "B", AttackOutcome.ERROR, seed_group_id="b"), + ], + ), + "legacy_identities_without_plan": _History( + attempts=[ + _Attempt("attack", "A", AttackOutcome.SUCCESS, eval_hash=None), + _Attempt("attack", "B", AttackOutcome.ERROR, eval_hash=None), + _Attempt("attack", "B", AttackOutcome.SUCCESS, eval_hash=None), + _Attempt("other", "A", AttackOutcome.FAILURE, eval_hash=None), + ], + ), + "legacy_error_matched_by_saved_plan": _History( + plan=_plan(_group(name="attack", eval_hash="eval", seed_ids=["a"]), seeds=[_seed("a", "A")]), + attempts=[ + # An older error row without seed attribution resolves to the planned unit by objective. + _Attempt("attack", "A", AttackOutcome.ERROR), + _Attempt("attack", "A", AttackOutcome.SUCCESS, seed_group_id="a"), + ], + ), + "technique_configurations_sharing_a_name": _History( + plan=_plan( + _group(name="attack", eval_hash="eval-1", seed_ids=["a"], display_group="Attack"), + _group(name="attack", eval_hash="eval-2", seed_ids=["a"], display_group="Attack"), + seeds=[_seed("a", "A")], + ), + attempts=[ + _Attempt("attack", "A", AttackOutcome.SUCCESS, eval_hash="eval-1", seed_group_id="a"), + _Attempt("attack", "A", AttackOutcome.FAILURE, eval_hash="eval-2", seed_group_id="a"), + ], + ), + "display_groups": _History( + plan=_plan( + _group(name="base64", eval_hash="e1", seed_ids=["a", "b"], display_group="encoding"), + _group(name="rot13", eval_hash="e2", seed_ids=["a", "b"], display_group="encoding"), + _group(name="crescendo", eval_hash="e3", seed_ids=["a"], display_group="multi_turn"), + seeds=[_seed("a", "A"), _seed("b", "B")], + ), + display_group_map={"base64": "encoding", "rot13": "encoding", "crescendo": "multi_turn"}, + attempts=[ + _Attempt("base64", "A", AttackOutcome.SUCCESS, eval_hash="e1", seed_group_id="a"), + _Attempt("base64", "B", AttackOutcome.FAILURE, eval_hash="e1", seed_group_id="b"), + _Attempt("rot13", "A", AttackOutcome.ERROR, eval_hash="e2", seed_group_id="a"), + _Attempt("rot13", "A", AttackOutcome.SUCCESS, eval_hash="e2", seed_group_id="a"), + _Attempt("crescendo", "A", AttackOutcome.UNDETERMINED, eval_hash="e3", seed_group_id="a"), + ], + ), + "empty_history": _History( + plan=_plan(_group(name="attack", eval_hash="eval", seed_ids=["a"]), seeds=[_seed("a", "A")]), + attempts=[], + ), +} + +# Effective-unit success percentages each history must report everywhere (None: no completed unit). +_EXPECTED_OVERALL = { + "retry_and_resume_recovered": 100, + "unrecovered_errors": 50, + "legacy_identities_without_plan": 66, + "legacy_error_matched_by_saved_plan": 100, + "technique_configurations_sharing_a_name": 50, + "display_groups": 50, + "empty_history": None, +} + + +def _persist(memory: MemoryInterface, history: _History) -> str: + scenario_result_id = uuid.uuid4() + metadata = {SCENARIO_RUN_PLAN_METADATA_KEY: history.plan.model_dump(mode="json")} if history.plan else {} + scenario_result = make_scenario_result( + id=scenario_result_id, + scenario_name="ParityScenario", + objective_target_identifier=ComponentIdentifier(class_name="MockTarget", class_module="tests"), + scenario_run_state=ScenarioRunState.COMPLETED, + attack_results={}, + creation_time=_T0, + display_group_map=history.display_group_map, + metadata=metadata, + ) + memory.add_scenario_results_to_memory(scenario_results=[scenario_result]) + attack_results = [] + for index, attempt in enumerate(history.attempts): + attribution_data: dict[str, str] = {"parent_collection": attempt.atomic_attack_name} + if attempt.eval_hash is not None: + attribution_data["parent_eval_hash"] = attempt.eval_hash + if attempt.seed_group_id is not None: + attribution_data["seed_group_id"] = attempt.seed_group_id + attack_results.append( + AttackResult( + conversation_id=f"conversation-{index}", + objective=attempt.objective, + outcome=attempt.outcome, + timestamp=_T0 + timedelta(seconds=index), + attribution_parent_id=str(scenario_result_id), + attribution_data=attribution_data, + ) + ) + if attack_results: + memory.add_attack_results_to_memory(attack_results=attack_results) + return str(scenario_result_id) + + +@pytest.mark.parametrize("history_name", sorted(_HISTORIES)) +async def test_sdk_api_and_reports_report_identical_statistics(history_name: str, sqlite_instance) -> None: + history = _HISTORIES[history_name] + scenario_result_id = _persist(sqlite_instance, history) + expected = _EXPECTED_OVERALL[history_name] + + # SDK + [scenario_result] = sqlite_instance.get_scenario_results(scenario_result_ids=[scenario_result_id]) + sdk = compute_scenario_statistics(scenario_result) + assert sdk.overall.success_percentage == expected + + # API: run detail, history list (SQL aggregate), and live progress + service = ScenarioRunService() + detail = service.get_run_from_storage(scenario_result_id=scenario_result_id, active_error=None) + [list_item] = [item for item in service.list_runs().items if item.scenario_result_id == scenario_result_id] + progress = service.get_run_progress_from_storage( + scenario_result_id=scenario_result_id, since=None, limit=500, active_group_ids=[] + ) + assert detail is not None + assert progress is not None + assert detail.objective_achieved_rate == (expected or 0) + assert list_item.objective_achieved_rate == (expected or 0) + assert progress.summary.overall.success_percentage == expected + assert detail.completed_attacks == sdk.overall.completed == progress.summary.overall.completed + assert list_item.completed_attacks == sdk.overall.completed + assert progress.summary.overall.succeeded == sdk.overall.succeeded + assert progress.summary.overall.errors == sdk.overall.errors + + # Reports + report = json.loads(await JsonScenarioResultPrinter().render_async(scenario_result)) + assert report["stats"]["overall_success_rate"] == (expected or 0) + + # Per-group numbers agree between the saved-plan progress view and the reports. + if history.plan is not None: + progress_groups = {group.display_group: group.success_percentage for group in progress.summary.display_groups} + sdk_groups = {name: counts.success_percentage for name, counts in sdk.display_groups.items()} + assert sdk_groups == progress_groups + report_groups = {group["name"]: group["success_rate"] for group in report["groups"]} + for name, rate in report_groups.items(): + assert rate == (progress_groups.get(name) or 0) + + +def test_historical_attempt_counts_stay_separate_from_units(sqlite_instance) -> None: + scenario_result_id = _persist(sqlite_instance, _HISTORIES["retry_and_resume_recovered"]) + [scenario_result] = sqlite_instance.get_scenario_results(scenario_result_ids=[scenario_result_id]) + + statistics = compute_scenario_statistics(scenario_result) + + assert statistics.attempts == 4 + assert statistics.overall.completed == 2 + assert statistics.overall.planned == 2 + assert statistics.overall.errors == 2 + assert statistics.overall.retries == 2 + assert statistics.unattributed_attempts == 0 diff --git a/tests/unit/backend/test_main.py b/tests/unit/backend/test_main.py index d9790b81e7..a0140f2b02 100644 --- a/tests/unit/backend/test_main.py +++ b/tests/unit/backend/test_main.py @@ -291,16 +291,24 @@ def test_frontend_exists_mounts_static(self) -> None: # Create the directory so StaticFiles doesn't raise os.makedirs("/tmp/fake_frontend", exist_ok=True) - with ( - patch("pyrit.backend.main.DEV_MODE", False), - patch("pyrit.backend.main.Path") as mock_path_cls, - patch("builtins.print"), - ): - mock_path_instance = MagicMock() - mock_path_instance.parent.__truediv__ = MagicMock(return_value=mock_frontend_path) - mock_path_cls.return_value = mock_path_instance - - setup_frontend() + # setup_frontend mounts onto the shared app; restore its routes so the catch-all SPA mount + # does not leak into later tests (it turns unmatched API paths into 404s). + original_routes = list(app.router.routes) + try: + with ( + patch("pyrit.backend.main.DEV_MODE", False), + patch("pyrit.backend.main.Path") as mock_path_cls, + patch("builtins.print"), + ): + mock_path_instance = MagicMock() + mock_path_instance.parent.__truediv__ = MagicMock(return_value=mock_frontend_path) + mock_path_cls.return_value = mock_path_instance + + setup_frontend() + + assert any(getattr(route, "name", None) == "frontend" for route in app.router.routes) + finally: + app.router.routes[:] = original_routes def test_frontend_missing_warns_but_continues(self) -> None: """Test that setup_frontend warns but does not exit when frontend is missing.""" diff --git a/tests/unit/models/test_scenario_result.py b/tests/unit/models/test_scenario_result.py index cbee4d145d..98cda6f8fd 100644 --- a/tests/unit/models/test_scenario_result.py +++ b/tests/unit/models/test_scenario_result.py @@ -2,7 +2,9 @@ # Licensed under the MIT license. import uuid -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime + +import pytest from pyrit.models import ( ComponentIdentifier, @@ -109,10 +111,10 @@ def test_get_objectives_by_attack_name(self): def test_objective_achieved_rate_all(self): results = [ - _make_attack_result(outcome=AttackOutcome.SUCCESS), - _make_attack_result(outcome=AttackOutcome.FAILURE), - _make_attack_result(outcome=AttackOutcome.SUCCESS), - _make_attack_result(outcome=AttackOutcome.UNDETERMINED), + _make_attack_result(objective="obj1", outcome=AttackOutcome.SUCCESS), + _make_attack_result(objective="obj2", outcome=AttackOutcome.FAILURE), + _make_attack_result(objective="obj3", outcome=AttackOutcome.SUCCESS), + _make_attack_result(objective="obj4", outcome=AttackOutcome.UNDETERMINED), ] sr = make_scenario_result( scenario_name="TestScenario", @@ -120,7 +122,8 @@ def test_objective_achieved_rate_all(self): attack_results={"s1": results}, objective_scorer_identifier=ComponentIdentifier.model_validate({}), ) - assert sr.objective_achieved_rate() == 50 + with pytest.warns(DeprecationWarning, match="compute_scenario_statistics"): + assert sr.objective_achieved_rate() == 50 def test_objective_achieved_rate_empty(self): sr = make_scenario_result( @@ -129,7 +132,8 @@ def test_objective_achieved_rate_empty(self): attack_results={"s1": []}, objective_scorer_identifier=ComponentIdentifier.model_validate({}), ) - assert sr.objective_achieved_rate() == 0 + with pytest.warns(DeprecationWarning, match="compute_scenario_statistics"): + assert sr.objective_achieved_rate() == 0 def test_objective_achieved_rate_by_name(self): sr = make_scenario_result( @@ -141,93 +145,10 @@ def test_objective_achieved_rate_by_name(self): }, objective_scorer_identifier=ComponentIdentifier.model_validate({}), ) - assert sr.objective_achieved_rate(atomic_attack_name="s1") == 100 - assert sr.objective_achieved_rate(atomic_attack_name="s2") == 0 - assert sr.objective_achieved_rate(atomic_attack_name="missing") == 0 - - def _attempt(self, *, outcome, objective="obj", minutes=0, seed_group_id=None): - return AttackResult( - conversation_id=str(uuid.uuid4()), - objective=objective, - outcome=outcome, - timestamp=datetime(2026, 1, 1, tzinfo=UTC) + timedelta(minutes=minutes), - attribution_data={"seed_group_id": seed_group_id} if seed_group_id else None, - ) - - def _scenario(self, attack_results): - return make_scenario_result( - scenario_name="TestScenario", - objective_target_identifier=ComponentIdentifier.model_validate({}), - attack_results=attack_results, - objective_scorer_identifier=ComponentIdentifier.model_validate({}), - ) - - def test_objective_achieved_rate_ignores_errors_recovered_by_retry(self): - sr = self._scenario( - { - "s1": [ - self._attempt(outcome=AttackOutcome.SUCCESS, objective="a", minutes=0), - self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=1), - self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=2), - self._attempt(outcome=AttackOutcome.SUCCESS, objective="b", minutes=3), - ] - } - ) - - assert sr.objective_achieved_rate() == 100 - assert sr.objective_achieved_rate(atomic_attack_name="s1") == 100 - assert [r.objective for r in sr.get_latest_attack_results()["s1"]] == ["a", "b"] - - def test_objective_achieved_rate_counts_unrecovered_error_once(self): - sr = self._scenario( - { - "s1": [ - self._attempt(outcome=AttackOutcome.SUCCESS, objective="a", minutes=0), - self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=1), - self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=2), - ] - } - ) - - latest = sr.get_latest_attack_results()["s1"] - assert [r.outcome for r in latest] == [AttackOutcome.SUCCESS, AttackOutcome.ERROR] - assert sr.objective_achieved_rate() == 50 - - def test_errors_are_only_superseded_within_the_same_seed_group(self): - sr = self._scenario( - { - "s1": [ - self._attempt(outcome=AttackOutcome.ERROR, objective="same", minutes=0, seed_group_id="g1"), - self._attempt(outcome=AttackOutcome.SUCCESS, objective="same", minutes=1, seed_group_id="g2"), - ] - } - ) - - assert len(sr.get_latest_attack_results()["s1"]) == 2 - assert sr.objective_achieved_rate() == 50 - - def test_errors_are_only_superseded_within_the_same_atomic_attack(self): - sr = self._scenario( - { - "s1": [self._attempt(outcome=AttackOutcome.ERROR, objective="same", minutes=0)], - "s2": [self._attempt(outcome=AttackOutcome.SUCCESS, objective="same", minutes=1)], - } - ) - - assert sr.objective_achieved_rate() == 50 - - def test_get_display_groups_latest_attempts_only(self): - sr = self._scenario( - { - "s1": [ - self._attempt(outcome=AttackOutcome.ERROR, objective="b", minutes=0), - self._attempt(outcome=AttackOutcome.SUCCESS, objective="b", minutes=1), - ] - } - ) - - assert len(sr.get_display_groups()["s1"]) == 2 - assert [r.outcome for r in sr.get_display_groups(latest_attempts_only=True)["s1"]] == [AttackOutcome.SUCCESS] + with pytest.warns(DeprecationWarning, match="compute_scenario_statistics"): + assert sr.objective_achieved_rate(atomic_attack_name="s1") == 100 + assert sr.objective_achieved_rate(atomic_attack_name="s2") == 0 + assert sr.objective_achieved_rate(atomic_attack_name="missing") == 0 def test_normalize_scenario_name_snake_case(self): assert ScenarioResult.normalize_scenario_name("content_harms") == "ContentHarms" diff --git a/tests/unit/output/scenario_result/test_json.py b/tests/unit/output/scenario_result/test_json.py index e9f8e3b1a3..081e87e66e 100644 --- a/tests/unit/output/scenario_result/test_json.py +++ b/tests/unit/output/scenario_result/test_json.py @@ -3,7 +3,6 @@ import json import uuid -from datetime import timedelta import pytest from unit.mocks import make_scenario_result @@ -87,19 +86,6 @@ async def test_overview_reports_scenario_and_stats(printer): assert {g["name"] for g in payload["groups"]} == {"technique_a", "technique_b"} -async def test_overview_success_rates_ignore_errors_recovered_by_retry(printer): - error = _attack_result(outcome=AttackOutcome.ERROR, objective="obj") - retried = _attack_result(outcome=AttackOutcome.SUCCESS, objective="obj") - retried.timestamp = error.timestamp + timedelta(seconds=1) - result = _scenario_result(attack_results={"technique_a": [error, retried]}) - - payload = json.loads(await printer.render_async(result)) - - assert payload["stats"]["overall_success_rate"] == 100 - assert payload["stats"]["total_results"] == 2 - assert payload["groups"] == [{"name": "technique_a", "num_results": 2, "success_rate": 100}] - - async def test_overview_prefers_underlying_model_name(printer): result = _scenario_result(target_params={"model_name": "deploy", "underlying_model_name": "gpt-4o"}) payload = json.loads(await printer.render_async(result)) diff --git a/tests/unit/output/scenario_result/test_pretty.py b/tests/unit/output/scenario_result/test_pretty.py index 1ff5bbb023..f2c6dcfcd9 100644 --- a/tests/unit/output/scenario_result/test_pretty.py +++ b/tests/unit/output/scenario_result/test_pretty.py @@ -152,7 +152,9 @@ async def test_write_async_raises_when_scorer_identifier_present_without_scorer_ ) async def test_write_async_color_bands_for_success_rate(patch_central_database, capsys, expected_rate, attack_outcomes): p = PrettyScenarioResultMemoryPrinter(enable_colors=True) - result = _scenario_result(attack_results={"s": [_attack_result(outcome=o) for o in attack_outcomes]}) + result = _scenario_result( + attack_results={"s": [_attack_result(outcome=o, objective=f"obj{i}") for i, o in enumerate(attack_outcomes)]} + ) await p.write_async(result) out = capsys.readouterr().out assert f"Overall Success Rate: {expected_rate}%" in out @@ -219,8 +221,8 @@ async def test_write_async_sorts_groups_by_success_rate_descending(patch_central "low": [_attack_result(outcome=AttackOutcome.FAILURE)], "high": [_attack_result(outcome=AttackOutcome.SUCCESS)], "mid": [ - _attack_result(outcome=AttackOutcome.SUCCESS), - _attack_result(outcome=AttackOutcome.FAILURE), + _attack_result(outcome=AttackOutcome.SUCCESS, objective="obj1"), + _attack_result(outcome=AttackOutcome.FAILURE, objective="obj2"), ], }, ) diff --git a/tests/unit/output/test_derivation.py b/tests/unit/output/test_derivation.py index 03a25009cd..4c6b7d33ed 100644 --- a/tests/unit/output/test_derivation.py +++ b/tests/unit/output/test_derivation.py @@ -8,9 +8,9 @@ from pyrit.models import AttackOutcome, AttackResult, ComponentIdentifier, Score, ScoreStatus from pyrit.output._derivation import ( attack_score_display, - group_success_rate, resolve_scorer_name, resolve_target_info, + scenario_success_rates, select_attacks, ) @@ -47,21 +47,27 @@ def test_resolve_target_info_missing_fields_are_none(): assert info.endpoint is None -# --- group_success_rate --- +# --- scenario_success_rates --- -def test_group_success_rate_empty_is_zero(): - assert group_success_rate([]) == 0 +def test_scenario_success_rates_empty_is_zero(): + result = make_scenario_result(scenario_name="S", attack_results={"s1": []}) + assert scenario_success_rates(result) == (0, {}) -def test_group_success_rate_counts_success(): - attacks = [ - _attack(outcome=AttackOutcome.SUCCESS), - _attack(outcome=AttackOutcome.FAILURE), - _attack(outcome=AttackOutcome.SUCCESS), - _attack(outcome=AttackOutcome.UNDETERMINED), - ] - assert group_success_rate(attacks) == 50 +def test_scenario_success_rates_per_display_group(): + result = make_scenario_result( + scenario_name="S", + attack_results={ + "base64": [ + AttackResult(conversation_id="c1", objective="o1", outcome=AttackOutcome.SUCCESS), + AttackResult(conversation_id="c2", objective="o2", outcome=AttackOutcome.FAILURE), + ], + "rot13": [AttackResult(conversation_id="c3", objective="o1", outcome=AttackOutcome.SUCCESS)], + }, + display_group_map={"base64": "encoding", "rot13": "encoding"}, + ) + assert scenario_success_rates(result) == (66, {"encoding": 66}) # --- attack_score_display --- diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index a22d43cca5..76b708d024 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -9,6 +9,7 @@ import pytest +from pyrit.analytics import compute_scenario_statistics from pyrit.executor.attack import PromptSendingAttack, RedTeamingAttack from pyrit.executor.attack.core import AttackExecutorResult from pyrit.memory import CentralMemory @@ -836,10 +837,10 @@ def test_scenario_result_with_empty_results(self): ) assert len(result.attack_results["base64"]) == 0 - assert result.objective_achieved_rate() == 0 + assert compute_scenario_statistics(result).overall.success_percentage is None - def test_scenario_result_objective_achieved_rate(self, sample_attack_results): - """Test objective_achieved_rate calculation.""" + def test_scenario_result_success_percentage(self, sample_attack_results): + """Test the effective success percentage of a scenario result.""" # All successful result = make_scenario_result( scenario_name="Test", @@ -851,19 +852,19 @@ def test_scenario_result_objective_achieved_rate(self, sample_attack_results): attack_results={"base64": sample_attack_results}, objective_scorer_identifier=_TEST_SCORER_ID, ) - assert result.objective_achieved_rate() == 100 + assert compute_scenario_statistics(result).overall.success_percentage == 100 # Mixed outcomes mixed_results = sample_attack_results[:3] + [ AttackResult( conversation_id="conv-fail", - objective="objective", + objective="objective-fail", outcome=AttackOutcome.FAILURE, executed_turns=1, ), AttackResult( conversation_id="conv-fail2", - objective="objective", + objective="objective-fail2", outcome=AttackOutcome.FAILURE, executed_turns=1, ), @@ -878,7 +879,7 @@ def test_scenario_result_objective_achieved_rate(self, sample_attack_results): attack_results={"base64": mixed_results}, objective_scorer_identifier=_TEST_SCORER_ID, ) - assert result2.objective_achieved_rate() == 60 # 3 out of 5 + assert compute_scenario_statistics(result2).overall.success_percentage == 60 # 3 out of 5 @pytest.mark.usefixtures("patch_central_database") From 9b3a9f49a9b39e99784c88c66ca0315394d90a99 Mon Sep 17 00:00:00 2001 From: Utkarsh Bahuguna Date: Sat, 26 Sep 2026 16:17:15 +0530 Subject: [PATCH 3/3] Key report group rates by display_group_map and address review feedback - Reports fold per-atomic-attack counts with display_group_map, the same keys they group results by, via combine_execution_counts. - Reports show effective units next to raw attempts (num_results vs num_attempts, total_results vs total_attempts). - compute_scenario_statistics takes use_saved_plan instead of a sentinel; drop the unused ScenarioPlanLookup.planned_units. - Parity test compares group key sets and unit counts directly. - Framework doc calls out the SQL history aggregate as a second implementation kept in parity by the test. --- doc/code/framework.md | 2 +- pyrit/analytics/scenario_statistics.py | 55 ++++++++++----- .../services/scenario_progress_read_model.py | 2 +- pyrit/output/_derivation.py | 60 +++++++++++++--- pyrit/output/scenario_result/html.py | 8 ++- pyrit/output/scenario_result/json.py | 19 ++--- pyrit/output/scenario_result/pretty.py | 27 ++++--- .../analytics/test_scenario_statistics.py | 4 +- .../test_scenario_statistics_parity.py | 28 ++++++-- .../unit/output/scenario_result/test_json.py | 24 ++++++- .../output/scenario_result/test_pretty.py | 5 +- tests/unit/output/test_derivation.py | 70 +++++++++++++++++-- 12 files changed, 229 insertions(+), 75 deletions(-) diff --git a/doc/code/framework.md b/doc/code/framework.md index e37001beb1..c05bfd2a6d 100644 --- a/doc/code/framework.md +++ b/doc/code/framework.md @@ -320,7 +320,7 @@ The below talks about responsibilities of most modules in the PyRIT library - This is where cross-run analysis belongs: e.g. "which attack performed best for this objective?", "how often did a technique succeed?", or "which responses match known content?". - **Does not own**: live, in-attack decisions — any decision made *during* an attack is a scorer's job. Analytics only operates on stored results, after the fact. - Today it includes `ConversationAnalytics` (inspecting conversation history), `analyze_results` / `AttackStats` (aggregating outcomes across techniques), and text-matching strategies (`ExactTextMatching`, `ApproximateTextMatching`). -- `compute_scenario_statistics` is the single calculation of scenario success statistics. It owns execution-unit identity (atomic attack, technique configuration, and seed group), latest-attempt selection, counts, denominators, and rounding. `ScenarioResult`'s SDK callers, the GUI backend, and the console, JSON, and HTML reports all present its results (`ScenarioExecutionStatistics`, `ScenarioExecutionUnit`, and `ScenarioProgressCounts` in `pyrit.models`) instead of calculating their own. +- `compute_scenario_statistics` calculates scenario success statistics. It owns execution-unit identity (atomic attack, technique configuration, and seed group), latest-attempt selection, counts, denominators, and rounding. SDK callers, the GUI backend's run detail and progress views, and the console, JSON, and HTML reports all present its results (`ScenarioExecutionStatistics`, `ScenarioExecutionUnit`, and `ScenarioProgressCounts` in `pyrit.models`) instead of calculating their own. The one exception is the GUI run-history list, which aggregates the same statistics in SQL (`MemoryInterface._build_scenario_history_aggregate_statement`) so it can page over many runs; `tests/unit/analytics/test_scenario_statistics_parity.py` keeps the two implementations in agreement. - Shared analytics contracts (filters, dimensions, typed values, reports, facets, result pages, and `AttackStats`) live in `pyrit.models.analytics`. They validate data without querying memory or calculating statistics. `AttackResultSelection` defines selection modes without changing existing callers. - Filter-bound cursor and label-normalization helpers live in `pyrit.common.pagination`. The backend pagination module retains compatibility exports, including History's invalid-cursor first-page fallback. diff --git a/pyrit/analytics/scenario_statistics.py b/pyrit/analytics/scenario_statistics.py index 48077556c1..1abf511411 100644 --- a/pyrit/analytics/scenario_statistics.py +++ b/pyrit/analytics/scenario_statistics.py @@ -22,7 +22,7 @@ import logging from dataclasses import dataclass from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Protocol from pydantic import ValidationError @@ -48,8 +48,15 @@ logger = logging.getLogger(__name__) -#: Default for ``compute_scenario_statistics(plan=...)``: use the plan saved in the result's metadata. -SAVED_RUN_PLAN: Any = object() + +class _CountableAttempt(Protocol): + """An attempt that can be counted: both ``ScenarioAttempt`` and ``ScenarioProgressResult`` qualify.""" + + @property + def outcome(self) -> AttackOutcome: ... + + @property + def total_retries(self) -> int: ... @dataclass(frozen=True, slots=True) @@ -59,7 +66,6 @@ class ScenarioPlanLookup: groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] groups_by_name: dict[str, tuple[ScenarioRunPlanAtomicGroup, ...]] seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] - planned_units: frozenset[ScenarioExecutionUnit] @classmethod def from_plan(cls, *, plan: ScenarioRunPlan | None) -> ScenarioPlanLookup: @@ -74,14 +80,12 @@ def from_plan(cls, *, plan: ScenarioRunPlan | None) -> ScenarioPlanLookup: groups_by_identity={}, groups_by_name={}, seed_ids_by_group_and_objective={}, - planned_units=frozenset(), ) groups_by_identity: dict[tuple[str, str], ScenarioRunPlanAtomicGroup] = {} grouped_by_name: dict[str, list[ScenarioRunPlanAtomicGroup]] = {} seeds_by_id = {seed.id: seed for seed in plan.seed_groups} seed_ids_by_group_and_objective: dict[tuple[str, str], tuple[str, ...]] = {} - planned_units: set[ScenarioExecutionUnit] = set() for group in plan.atomic_groups: groups_by_identity[(group.atomic_attack_name, group.technique_eval_hash)] = group grouped_by_name.setdefault(group.atomic_attack_name, []).append(group) @@ -95,16 +99,11 @@ def from_plan(cls, *, plan: ScenarioRunPlan | None) -> ScenarioPlanLookup: for objective_sha256, seed_ids in seed_ids_by_objective.items() } ) - planned_units.update( - ScenarioExecutionUnit(atomic_group_id=group.id, seed_group_id=seed_group_id) - for seed_group_id in group.seed_group_ids - ) return cls( groups_by_identity=groups_by_identity, groups_by_name={name: tuple(groups) for name, groups in grouped_by_name.items()}, seed_ids_by_group_and_objective=seed_ids_by_group_and_objective, - planned_units=frozenset(planned_units), ) def resolve_group( @@ -269,7 +268,7 @@ def success_percentage(*, succeeded: int, completed: int) -> int | None: def count_execution_units( *, units: Iterable[ScenarioExecutionUnit], - attempts_by_unit: Mapping[ScenarioExecutionUnit, Sequence[Any]], + attempts_by_unit: Mapping[ScenarioExecutionUnit, Sequence[_CountableAttempt]], planned: int | None, ) -> ScenarioProgressCounts: """ @@ -306,10 +305,33 @@ def count_execution_units( ) +def combine_execution_counts(counts: Iterable[ScenarioProgressCounts]) -> ScenarioProgressCounts: + """ + Combine counts for disjoint sets of execution units, such as the atomic attacks of one display group. + + Returns: + ScenarioProgressCounts: The summed counts with the success percentage recomputed. ``planned`` is + None unless every input has one. + """ + counts = list(counts) + completed = sum(item.completed for item in counts) + succeeded = sum(item.succeeded for item in counts) + planned = [item.planned for item in counts] + return ScenarioProgressCounts( + completed=completed, + planned=sum(value for value in planned if value is not None) if all(v is not None for v in planned) else None, + succeeded=succeeded, + success_percentage=success_percentage(succeeded=succeeded, completed=completed), + errors=sum(item.errors for item in counts), + retries=sum(item.retries for item in counts), + ) + + def compute_scenario_statistics( scenario_result: ScenarioResult, *, - plan: ScenarioRunPlan | None = SAVED_RUN_PLAN, + plan: ScenarioRunPlan | None = None, + use_saved_plan: bool = True, ) -> ScenarioExecutionStatistics: """ Calculate effective execution-unit statistics for a scenario result. @@ -320,13 +342,14 @@ def compute_scenario_statistics( Args: scenario_result (ScenarioResult): The scenario result with its hydrated attack results. - plan (ScenarioRunPlan | None): The run plan, or None to count the result as a legacy run. - Defaults to the plan saved in the result's metadata. + plan (ScenarioRunPlan | None): An explicit run plan. Defaults to None. + use_saved_plan (bool): When ``plan`` is None, whether to use the plan saved in the result's + metadata. Pass False to count the result as a legacy run. Defaults to True. Returns: ScenarioExecutionStatistics: Overall, per atomic attack, and per display group counts. """ - if plan is SAVED_RUN_PLAN: + if plan is None and use_saved_plan: plan = load_scenario_run_plan(scenario_result) plan_lookup = ScenarioPlanLookup.from_plan(plan=plan) diff --git a/pyrit/backend/services/scenario_progress_read_model.py b/pyrit/backend/services/scenario_progress_read_model.py index e9491b62f5..ae6d11e8c4 100644 --- a/pyrit/backend/services/scenario_progress_read_model.py +++ b/pyrit/backend/services/scenario_progress_read_model.py @@ -244,7 +244,7 @@ def calculate_progress_counts( tuple[int, int, int, int]: Total, completed, success-rate percentage, and successful-unit count. """ - overall = compute_scenario_statistics(scenario_result, plan=plan).overall + overall = compute_scenario_statistics(scenario_result, plan=plan, use_saved_plan=False).overall total = overall.planned if overall.planned is not None else overall.completed return total, overall.completed, overall.success_percentage or 0, overall.succeeded diff --git a/pyrit/output/_derivation.py b/pyrit/output/_derivation.py index f1d143a5f4..a6d903225f 100644 --- a/pyrit/output/_derivation.py +++ b/pyrit/output/_derivation.py @@ -14,6 +14,8 @@ from typing import TYPE_CHECKING, NamedTuple +from pyrit.analytics.scenario_statistics import combine_execution_counts, compute_scenario_statistics + if TYPE_CHECKING: from pyrit.models import AttackResult, ComponentIdentifier, ScenarioResult, Score @@ -51,25 +53,63 @@ def resolve_target_info(target_id: ComponentIdentifier | None) -> TargetInfo: ) -def scenario_success_rates(result: ScenarioResult) -> tuple[int, dict[str, int]]: +class GroupStatistics(NamedTuple): + """Effective-unit statistics for one display group, alongside its raw attempt count.""" + + name: str + units: int + attempts: int + success_rate: int + + +class ScenarioOverview(NamedTuple): + """Overall and per-display-group statistics for a scenario report.""" + + units: int + attempts: int + success_rate: int + groups: list[GroupStatistics] + + +def scenario_overview(result: ScenarioResult) -> ScenarioOverview: """ - Return the overall and per-display-group success rates of a scenario result. + Summarize a scenario result for the reports. - The numbers come from ``pyrit.analytics.compute_scenario_statistics``, the calculation shared with - the SDK and the GUI backend, so every report shows the same effective execution-unit statistics. - Groups with no completed unit report 0. + The numbers come from ``pyrit.analytics.compute_scenario_statistics``, the calculation shared with the + SDK and the GUI backend. Groups follow ``result.get_display_groups()``: each group folds the per-atomic- + attack counts of the atomic attacks ``display_group_map`` assigns to it, so the rate is always keyed the + same way the printers group their results. ``units`` counts effective execution units (the success-rate + denominator); ``attempts`` counts every persisted attempt, including retries. Args: result (ScenarioResult): The scenario result to summarize. Returns: - tuple[int, dict[str, int]]: The overall rate and the rate for each display group, as integer percents. + ScenarioOverview: The overall and per-group statistics. """ - from pyrit.analytics.scenario_statistics import compute_scenario_statistics - statistics = compute_scenario_statistics(result) - group_rates = {name: counts.success_percentage or 0 for name, counts in statistics.display_groups.items()} - return statistics.overall.success_percentage or 0, group_rates + groups: list[GroupStatistics] = [] + for group_name, group_results in result.get_display_groups().items(): + atomic_attack_names = [ + name for name in result.attack_results if result.display_group_map.get(name, name) == group_name + ] + counts = combine_execution_counts( + statistics.atomic_attacks[name] for name in atomic_attack_names if name in statistics.atomic_attacks + ) + groups.append( + GroupStatistics( + name=group_name, + units=counts.completed, + attempts=len(group_results), + success_rate=counts.success_percentage or 0, + ) + ) + return ScenarioOverview( + units=statistics.overall.completed, + attempts=statistics.attempts, + success_rate=statistics.overall.success_percentage or 0, + groups=groups, + ) def attack_score_display(attack: AttackResult, *, none_value: str | None = None) -> str | None: diff --git a/pyrit/output/scenario_result/html.py b/pyrit/output/scenario_result/html.py index bb575db1ee..f43fc132bd 100644 --- a/pyrit/output/scenario_result/html.py +++ b/pyrit/output/scenario_result/html.py @@ -51,6 +51,8 @@
{{ report.overview.stats.total_techniques }}
Attack results
{{ report.overview.stats.total_results }}
+ {% if report.overview.stats.total_attempts is defined %}
Attempts
+
{{ report.overview.stats.total_attempts }}
{% endif %}
Objectives
{{ report.overview.stats.unique_objectives }}
@@ -64,9 +66,11 @@

Per-group breakdown

- + {% for g in report.overview.groups %} - + + + {% endfor %}
GroupResultsSuccess rate
GroupResultsAttemptsSuccess rate
{{ g.name }}{{ g.num_results }}{{ g.success_rate }}%
{{ g.name }}{{ g.num_results }}{{ g.num_attempts if g.num_attempts is defined else g.num_results }}{{ g.success_rate }}%
diff --git a/pyrit/output/scenario_result/json.py b/pyrit/output/scenario_result/json.py index d002c4b078..6941e72e9c 100644 --- a/pyrit/output/scenario_result/json.py +++ b/pyrit/output/scenario_result/json.py @@ -8,7 +8,7 @@ from pyrit.output._derivation import ( attack_score_display, resolve_target_info, - scenario_success_rates, + scenario_overview, select_attacks, ) from pyrit.output.scenario_result.base import ScenarioResultPrinterBase, ScenarioView @@ -122,15 +122,15 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]: """ target = resolve_target_info(result.objective_target_identifier) - overall_rate, group_rates = scenario_success_rates(result) - display_groups = result.get_display_groups() + overview = scenario_overview(result) groups = [ { - "name": group_name, - "num_results": len(group_results), - "success_rate": group_rates.get(group_name, 0), + "name": group.name, + "num_results": group.units, + "num_attempts": group.attempts, + "success_rate": group.success_rate, } - for group_name, group_results in display_groups.items() + for group in overview.groups ] scorer_identifier = result.objective_scorer_identifier @@ -158,8 +158,9 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]: "scorer": scorer, "stats": { "total_techniques": len(result.get_techniques_used()), - "total_results": sum(len(results) for results in result.attack_results.values()), - "overall_success_rate": overall_rate, + "total_results": overview.units, + "total_attempts": overview.attempts, + "overall_success_rate": overview.success_rate, "unique_objectives": len(result.get_objectives()), }, "groups": groups, diff --git a/pyrit/output/scenario_result/pretty.py b/pyrit/output/scenario_result/pretty.py index 242f889c47..2880a1665f 100644 --- a/pyrit/output/scenario_result/pretty.py +++ b/pyrit/output/scenario_result/pretty.py @@ -9,7 +9,7 @@ from pyrit.output._derivation import ( attack_score_display, resolve_target_info, - scenario_success_rates, + scenario_overview, select_attacks, ) from pyrit.output._formatting import _PrettyPrinterMixin @@ -201,13 +201,14 @@ async def render_async( lines = [] lines.append(self._render_section_header("Overall Statistics")) - total_results = sum(len(results) for results in result.attack_results.values()) total_techniques = len(result.get_techniques_used()) - overall_rate, group_rates = scenario_success_rates(result) + overview = scenario_overview(result) + overall_rate = overview.success_rate lines.append(self._format_colored(f"{self._indent}📈 Summary", Style.BRIGHT)) lines.append(self._format_colored(f"{self._indent * 2}• Total Techniques: {total_techniques}", Fore.GREEN)) - lines.append(self._format_colored(f"{self._indent * 2}• Total Attack Results: {total_results}", Fore.GREEN)) + lines.append(self._format_colored(f"{self._indent * 2}• Total Attack Results: {overview.units}", Fore.GREEN)) + lines.append(self._format_colored(f"{self._indent * 2}• Total Attempts: {overview.attempts}", Fore.GREEN)) lines.append( self._format_colored( f"{self._indent * 2}• Overall Success Rate: {overall_rate}%", self._get_rate_color(overall_rate) @@ -218,24 +219,20 @@ async def render_async( lines.append(self._format_colored(f"{self._indent * 2}• Unique Objectives: {len(objectives)}", Fore.GREEN)) lines.append(self._render_section_header("Per-Group Breakdown")) - display_groups = result.get_display_groups() - - group_summaries: list[tuple[str, int, int]] = [ - (group_name, len(group_results), group_rates.get(group_name, 0)) - for group_name, group_results in display_groups.items() - ] + group_summaries = list(overview.groups) if self._sort_groups_by_success_rate: # Stable sort so groups with equal rates retain their original relative order. - group_summaries.sort(key=lambda item: item[2], reverse=True) + group_summaries.sort(key=lambda group: group.success_rate, reverse=True) - for group_name, total_group, group_rate in group_summaries: + for group in group_summaries: lines.append("\n") - lines.append(self._format_colored(f"{self._indent}🔸 Group: {group_name}", Style.BRIGHT)) - lines.append(self._format_colored(f"{self._indent * 2}• Number of Results: {total_group}", Fore.YELLOW)) + lines.append(self._format_colored(f"{self._indent}🔸 Group: {group.name}", Style.BRIGHT)) + lines.append(self._format_colored(f"{self._indent * 2}• Number of Results: {group.units}", Fore.YELLOW)) + lines.append(self._format_colored(f"{self._indent * 2}• Attempts: {group.attempts}", Fore.YELLOW)) lines.append( self._format_colored( - f"{self._indent * 2}• Success Rate: {group_rate}%", self._get_rate_color(group_rate) + f"{self._indent * 2}• Success Rate: {group.success_rate}%", self._get_rate_color(group.success_rate) ) ) diff --git a/tests/unit/analytics/test_scenario_statistics.py b/tests/unit/analytics/test_scenario_statistics.py index dc652e3915..a3499ade17 100644 --- a/tests/unit/analytics/test_scenario_statistics.py +++ b/tests/unit/analytics/test_scenario_statistics.py @@ -101,13 +101,13 @@ def test_saved_plan_counts_planned_units_and_reports_unattributed_attempts() -> assert statistics.display_groups["Attack"].planned == 2 -def test_explicit_none_plan_ignores_saved_plan() -> None: +def test_use_saved_plan_false_counts_a_legacy_run() -> None: result = make_scenario_result( attack_results={"attack": [_result(objective="A", outcome=AttackOutcome.SUCCESS)]}, metadata={SCENARIO_RUN_PLAN_METADATA_KEY: _plan().model_dump(mode="json")}, ) - statistics = compute_scenario_statistics(result, plan=None) + statistics = compute_scenario_statistics(result, use_saved_plan=False) assert statistics.overall.planned is None assert statistics.overall.completed == 1 diff --git a/tests/unit/analytics/test_scenario_statistics_parity.py b/tests/unit/analytics/test_scenario_statistics_parity.py index cd2bcc56f6..13debc907d 100644 --- a/tests/unit/analytics/test_scenario_statistics_parity.py +++ b/tests/unit/analytics/test_scenario_statistics_parity.py @@ -114,6 +114,7 @@ def _plan(*groups: ScenarioRunPlanAtomicGroup, seeds: list[ScenarioRunPlanSeedGr _group(name="attack", eval_hash="eval-2", seed_ids=["a"], display_group="Attack"), seeds=[_seed("a", "A")], ), + display_group_map={"attack": "Attack"}, attempts=[ _Attempt("attack", "A", AttackOutcome.SUCCESS, eval_hash="eval-1", seed_group_id="a"), _Attempt("attack", "A", AttackOutcome.FAILURE, eval_hash="eval-2", seed_group_id="a"), @@ -221,14 +222,27 @@ async def test_sdk_api_and_reports_report_identical_statistics(history_name: str report = json.loads(await JsonScenarioResultPrinter().render_async(scenario_result)) assert report["stats"]["overall_success_rate"] == (expected or 0) - # Per-group numbers agree between the saved-plan progress view and the reports. + # Per-group numbers agree between the SDK, the saved-plan progress view, and the reports. Compare + # key sets first so a group missing from one view fails instead of reading as 0%. + sdk_groups = {name: (counts.completed, counts.success_percentage) for name, counts in sdk.display_groups.items()} + report_groups = { + group["name"]: (group["num_results"], group["success_rate"]) + for group in report["groups"] + if group["num_attempts"] + } + # Reports list the groups that have results; planned groups with no attempts only appear in the SDK/API. + sdk_groups_with_results = {name: value for name, value in sdk_groups.items() if value[0]} + assert set(report_groups) == set(sdk_groups_with_results) + assert report_groups == { + name: (completed, rate or 0) for name, (completed, rate) in sdk_groups_with_results.items() + } if history.plan is not None: - progress_groups = {group.display_group: group.success_percentage for group in progress.summary.display_groups} - sdk_groups = {name: counts.success_percentage for name, counts in sdk.display_groups.items()} - assert sdk_groups == progress_groups - report_groups = {group["name"]: group["success_rate"] for group in report["groups"]} - for name, rate in report_groups.items(): - assert rate == (progress_groups.get(name) or 0) + progress_groups = { + group.display_group: (group.completed, group.success_percentage) + for group in progress.summary.display_groups + } + assert set(progress_groups) == set(sdk_groups) + assert progress_groups == sdk_groups def test_historical_attempt_counts_stay_separate_from_units(sqlite_instance) -> None: diff --git a/tests/unit/output/scenario_result/test_json.py b/tests/unit/output/scenario_result/test_json.py index 081e87e66e..8b1f738ea4 100644 --- a/tests/unit/output/scenario_result/test_json.py +++ b/tests/unit/output/scenario_result/test_json.py @@ -3,6 +3,7 @@ import json import uuid +from datetime import timedelta import pytest from unit.mocks import make_scenario_result @@ -65,8 +66,8 @@ async def test_overview_reports_scenario_and_stats(printer): target_params={"model_name": "gpt-test", "endpoint": "https://example.com"}, attack_results={ "technique_a": [ - _attack_result(outcome=AttackOutcome.SUCCESS), - _attack_result(outcome=AttackOutcome.FAILURE), + _attack_result(outcome=AttackOutcome.SUCCESS, objective="obj1"), + _attack_result(outcome=AttackOutcome.FAILURE, objective="obj2"), ], "technique_b": [_attack_result(outcome=AttackOutcome.SUCCESS)], }, @@ -82,7 +83,8 @@ async def test_overview_reports_scenario_and_stats(printer): assert payload["target"]["endpoint"] == "https://example.com" assert payload["stats"]["total_techniques"] == 2 assert payload["stats"]["total_results"] == 3 - assert payload["stats"]["unique_objectives"] == 1 + assert payload["stats"]["total_attempts"] == 3 + assert payload["stats"]["unique_objectives"] == 3 assert {g["name"] for g in payload["groups"]} == {"technique_a", "technique_b"} @@ -209,3 +211,19 @@ def test_build_conversations_empty_entries(): async def test_memory_printer_constructs_without_args(patch_central_database): assert isinstance(JsonScenarioResultMemoryPrinter(), JsonScenarioResultPrinter) + + +async def test_overview_separates_units_from_attempts(printer): + retried = [ + AttackResult(conversation_id=str(uuid.uuid4()), objective="obj", outcome=AttackOutcome.ERROR), + AttackResult(conversation_id=str(uuid.uuid4()), objective="obj", outcome=AttackOutcome.SUCCESS), + ] + retried[1].timestamp = retried[0].timestamp + timedelta(seconds=1) + result = _scenario_result(attack_results={"technique_a": retried}) + + payload = json.loads(await printer.render_async(result)) + + assert payload["stats"]["total_results"] == 1 + assert payload["stats"]["total_attempts"] == 2 + assert payload["stats"]["overall_success_rate"] == 100 + assert payload["groups"] == [{"name": "technique_a", "num_results": 1, "num_attempts": 2, "success_rate": 100}] diff --git a/tests/unit/output/scenario_result/test_pretty.py b/tests/unit/output/scenario_result/test_pretty.py index f2c6dcfcd9..6a21649186 100644 --- a/tests/unit/output/scenario_result/test_pretty.py +++ b/tests/unit/output/scenario_result/test_pretty.py @@ -57,8 +57,8 @@ async def test_write_async_renders_full_summary(printer, capsys): target_params={"model_name": "gpt-test", "endpoint": "https://example.com"}, attack_results={ "technique_a": [ - _attack_result(outcome=AttackOutcome.SUCCESS), - _attack_result(outcome=AttackOutcome.FAILURE), + _attack_result(outcome=AttackOutcome.SUCCESS, objective="obj1"), + _attack_result(outcome=AttackOutcome.FAILURE, objective="obj2"), ], "technique_b": [_attack_result(outcome=AttackOutcome.SUCCESS)], }, @@ -77,6 +77,7 @@ async def test_write_async_renders_full_summary(printer, capsys): assert "Overall Statistics" in out assert "Total Techniques: 2" in out assert "Total Attack Results: 3" in out + assert "Total Attempts: 3" in out assert "Per-Group Breakdown" in out assert "technique_a" in out assert "technique_b" in out diff --git a/tests/unit/output/test_derivation.py b/tests/unit/output/test_derivation.py index 4c6b7d33ed..7f70e4ddab 100644 --- a/tests/unit/output/test_derivation.py +++ b/tests/unit/output/test_derivation.py @@ -5,12 +5,24 @@ from unit.mocks import make_scenario_result -from pyrit.models import AttackOutcome, AttackResult, ComponentIdentifier, Score, ScoreStatus +from pyrit.common.utils import to_sha256 +from pyrit.models import ( + SCENARIO_RUN_PLAN_METADATA_KEY, + AttackOutcome, + AttackResult, + ComponentIdentifier, + ScenarioRunPlan, + ScenarioRunPlanAtomicGroup, + ScenarioRunPlanSeedGroup, + Score, + ScoreStatus, +) from pyrit.output._derivation import ( + GroupStatistics, attack_score_display, resolve_scorer_name, resolve_target_info, - scenario_success_rates, + scenario_overview, select_attacks, ) @@ -47,15 +59,19 @@ def test_resolve_target_info_missing_fields_are_none(): assert info.endpoint is None -# --- scenario_success_rates --- +# --- scenario_overview --- -def test_scenario_success_rates_empty_is_zero(): +def test_scenario_overview_empty_is_zero(): result = make_scenario_result(scenario_name="S", attack_results={"s1": []}) - assert scenario_success_rates(result) == (0, {}) + overview = scenario_overview(result) + + assert (overview.units, overview.attempts, overview.success_rate) == (0, 0, 0) + assert overview.groups == [GroupStatistics(name="s1", units=0, attempts=0, success_rate=0)] -def test_scenario_success_rates_per_display_group(): + +def test_scenario_overview_folds_atomic_attacks_by_display_group(): result = make_scenario_result( scenario_name="S", attack_results={ @@ -67,7 +83,47 @@ def test_scenario_success_rates_per_display_group(): }, display_group_map={"base64": "encoding", "rot13": "encoding"}, ) - assert scenario_success_rates(result) == (66, {"encoding": 66}) + + overview = scenario_overview(result) + + assert overview.success_rate == 66 + assert overview.groups == [GroupStatistics(name="encoding", units=3, attempts=3, success_rate=66)] + + +def test_scenario_overview_uses_display_group_map_even_when_plan_labels_differ(): + # The saved plan labels the group differently from display_group_map; the report must still + # key its rates the way it groups results (by display_group_map) instead of showing 0%. + plan = ScenarioRunPlan( + atomic_groups=[ + ScenarioRunPlanAtomicGroup( + id="g", + atomic_attack_name="base64", + display_group="Plan Label", + technique_eval_hash="e", + seed_group_ids=["s"], + ) + ], + seed_groups=[ScenarioRunPlanSeedGroup(id="s", objective_sha256=to_sha256("o"), objective="o")], + ) + result = make_scenario_result( + scenario_name="S", + attack_results={ + "base64": [ + AttackResult( + conversation_id="c1", + objective="o", + outcome=AttackOutcome.SUCCESS, + attribution_data={"parent_collection": "base64", "parent_eval_hash": "e", "seed_group_id": "s"}, + ) + ] + }, + display_group_map={"base64": "encoding"}, + metadata={SCENARIO_RUN_PLAN_METADATA_KEY: plan.model_dump(mode="json")}, + ) + + overview = scenario_overview(result) + + assert overview.groups == [GroupStatistics(name="encoding", units=1, attempts=1, success_rate=100)] # --- attack_score_display ---