diff --git a/doc/code/framework.md b/doc/code/framework.md
index 621e037a80..c05bfd2a6d 100644
--- a/doc/code/framework.md
+++ b/doc/code/framework.md
@@ -320,6 +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` 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/pyproject.toml b/pyproject.toml
index df88176d1d..56c224911d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -170,6 +170,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..1abf511411
--- /dev/null
+++ b/pyrit/analytics/scenario_statistics.py
@@ -0,0 +1,419 @@
+# 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, Protocol
+
+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__)
+
+
+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)
+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, ...]]
+
+ @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={},
+ )
+
+ 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, ...]] = {}
+ 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()
+ }
+ )
+
+ 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,
+ )
+
+ 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[_CountableAttempt]],
+ 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 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 = None,
+ use_saved_plan: bool = True,
+) -> 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): 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 None and use_saved_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..ae6d11e8c4 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, 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
@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 bc34c01116..bdbbc91e7f 100644
--- a/pyrit/models/__init__.py
+++ b/pyrit/models/__init__.py
@@ -142,6 +142,8 @@
ScenarioAttackTechniqueDetails,
ScenarioComponentIdentity,
ScenarioDisplayGroupProgress,
+ ScenarioExecutionStatistics,
+ ScenarioExecutionUnit,
ScenarioObjectiveScorer,
ScenarioObjectiveScorerMetrics,
ScenarioProgressCounts,
@@ -374,6 +376,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 e13ca5613e..542cf12b05 100644
--- a/pyrit/models/results/scenario_result.py
+++ b/pyrit/models/results/scenario_result.py
@@ -11,6 +11,8 @@
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.scenario_identifier import ScenarioIdentifier # noqa: TC001
from pyrit.models.identifiers.scorer_identifier import ScorerIdentifier # noqa: TC001
@@ -229,7 +231,16 @@ 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.
"""
+ 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 = []
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..a6d903225f 100644
--- a/pyrit/output/_derivation.py
+++ b/pyrit/output/_derivation.py
@@ -14,7 +14,7 @@
from typing import TYPE_CHECKING, NamedTuple
-from pyrit.models import AttackOutcome
+from pyrit.analytics.scenario_statistics import combine_execution_counts, compute_scenario_statistics
if TYPE_CHECKING:
from pyrit.models import AttackResult, ComponentIdentifier, ScenarioResult, Score
@@ -53,21 +53,63 @@ def resolve_target_info(target_id: ComponentIdentifier | None) -> TargetInfo:
)
-def group_success_rate(attacks: list[AttackResult]) -> 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 percentage of *attacks* whose outcome is SUCCESS (0 when empty).
+ 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. 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:
- attacks (list[AttackResult]): The attacks to score.
+ result (ScenarioResult): The scenario result to summarize.
Returns:
- int: The success rate as an integer percent.
+ ScenarioOverview: The overall and per-group statistics.
"""
- 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)
+ statistics = compute_scenario_statistics(result)
+ 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 }}
Attempts
+
{{ report.overview.stats.total_attempts }}
Objectives
{{ report.overview.stats.unique_objectives }}
Per-group breakdown
- | Group | Results | Success rate |
|---|
+ | Group | Results | Attempts | Success rate |
|---|
{% for g in report.overview.groups %}
- | {{ 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 }}% |
{% endfor %}
diff --git a/pyrit/output/scenario_result/json.py b/pyrit/output/scenario_result/json.py
index 1dddf0c9ff..6941e72e9c 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_overview,
+ select_attacks,
+)
from pyrit.output.scenario_result.base import ScenarioResultPrinterBase, ScenarioView
from pyrit.output.sink import Sink
@@ -117,14 +122,15 @@ def _build_overview(self, result: ScenarioResult) -> dict[str, Any]:
"""
target = resolve_target_info(result.objective_target_identifier)
- display_groups = result.get_display_groups()
+ overview = scenario_overview(result)
groups = [
{
- "name": group_name,
- "num_results": len(group_results),
- "success_rate": group_success_rate(group_results),
+ "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
@@ -152,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": result.objective_achieved_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 d17fc18d4b..2880a1665f 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_overview,
+ select_attacks,
+)
from pyrit.output._formatting import _PrettyPrinterMixin
from pyrit.output.scenario_result.base import ScenarioResultPrinterBase, ScenarioView
from pyrit.output.scorer.base import ScorerPrinterBase
@@ -196,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 = result.objective_achieved_rate()
+ 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)
@@ -213,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_success_rate(group_results))
- 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
new file mode 100644
index 0000000000..a3499ade17
--- /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_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, use_saved_plan=False)
+
+ 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..13debc907d
--- /dev/null
+++ b/tests/unit/analytics/test_scenario_statistics_parity.py
@@ -0,0 +1,259 @@
+# 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")],
+ ),
+ 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"),
+ ],
+ ),
+ "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 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.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:
+ 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 2862f37c53..ef20e2a87c 100644
--- a/tests/unit/backend/test_main.py
+++ b/tests/unit/backend/test_main.py
@@ -308,16 +308,24 @@ def test_frontend_exists_mounts_static(self, tmp_path: Path) -> None:
mock_frontend_path.exists.return_value = True
mock_frontend_path.__str__ = lambda self: str(tmp_path)
- 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 540c183774..98cda6f8fd 100644
--- a/tests/unit/models/test_scenario_result.py
+++ b/tests/unit/models/test_scenario_result.py
@@ -4,6 +4,8 @@
import uuid
from datetime import UTC, datetime
+import pytest
+
from pyrit.models import (
ComponentIdentifier,
ConversationReference,
@@ -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,9 +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
+ 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 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 1ff5bbb023..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
@@ -152,7 +153,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 +222,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..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,
- group_success_rate,
resolve_scorer_name,
resolve_target_info,
+ scenario_overview,
select_attacks,
)
@@ -47,21 +59,71 @@ def test_resolve_target_info_missing_fields_are_none():
assert info.endpoint is None
-# --- group_success_rate ---
+# --- scenario_overview ---
+
+
+def test_scenario_overview_empty_is_zero():
+ result = make_scenario_result(scenario_name="S", attack_results={"s1": []})
+
+ 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_group_success_rate_empty_is_zero():
- assert group_success_rate([]) == 0
+def test_scenario_overview_folds_atomic_attacks_by_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"},
+ )
+
+ 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)
-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
+ assert overview.groups == [GroupStatistics(name="encoding", units=1, attempts=1, success_rate=100)]
# --- 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")