Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions pyrit/backend/services/scenario_run_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -599,6 +599,18 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari
completed_attacks = total_attacks
techniques_used = scenario_result.get_techniques_used()

# Techniques the user selected but that never produced an attack cell were
# skipped during resolution (no registered factory for them). Compare against
# the built display groups rather than executed results so in-progress runs
# don't report not-yet-run techniques as skipped.
selected = set(scenario_result.scenario_identifier.techniques or [])

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

display_group_map is presentation data and cannot reliably tell us which factories resolved. Rapid Response groups by dataset, Adversarial Benchmark by target, and Jailbreak by template, so their successfully built techniques will be reported as skipped here. Please persist the exact missing names from resolve_technique_factories() and expose that authoritative list instead.

built_labels = set(scenario_result.display_group_map.values())
skipped_techniques = sorted(
technique
for technique in selected
if technique not in built_labels and not any(technique in label for label in built_labels)
)

# Surface per-attack errors and retry pressure regardless of overall run status:
# a COMPLETED scenario can still hide errored objectives or rate-limit retries.
failed_attacks: list[AttackErrorSummary] = []
Expand Down Expand Up @@ -641,6 +653,7 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari
error=error,
error_type=error_type,
techniques_used=techniques_used,
skipped_techniques=skipped_techniques,
total_attacks=total_attacks,
completed_attacks=completed_attacks,
objective_achieved_rate=scenario_result.objective_achieved_rate(),
Expand Down
3 changes: 3 additions & 0 deletions pyrit/cli/_output.py
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,9 @@ def print_scenario_run_summary(*, run: ScenarioRunSummary) -> None:
if run.techniques_used:
print(f" Techniques: {', '.join(run.techniques_used)}")

if run.skipped_techniques:
print(f" Skipped: {', '.join(run.skipped_techniques)} (no registered factory)")

if run.failed_attacks:
print(f"\n Failed Attacks ({len(run.failed_attacks)}):")
for failed in run.failed_attacks:
Expand Down
4 changes: 4 additions & 0 deletions pyrit/models/catalog/scenario.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,10 @@ class ScenarioRunSummary(BaseModel):
error: str | None = Field(None, description="Error message if status is FAILED")
error_type: str | None = Field(None, description="Exception class name if status is FAILED")
techniques_used: list[str] = Field(default_factory=list, description="Technique names that were executed")
skipped_techniques: list[str] = Field(
default_factory=list,
description="Selected techniques that were skipped because no factory was registered for them",
)
total_attacks: int = Field(0, ge=0, description="Total number of attack results persisted for this run")
completed_attacks: int = Field(0, ge=0, description="Number of attacks that reached a terminal outcome")
objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)")
Expand Down
52 changes: 46 additions & 6 deletions pyrit/scenario/core/matrix_atomic_attack_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,17 @@
from pyrit.scenario.core.atomic_attack import AtomicAttack
from pyrit.scenario.core.attack_technique import AttackTechnique


class TechniqueResolutionError(ValueError):
"""
Raised when a scenario selects techniques but none of them resolve to a factory.

Subclasses ``ValueError`` so existing ``except ValueError`` handlers keep working,
mirroring ``DatasetConstraintError``. Partial misses (some techniques resolve)
only warn, so a run still proceeds with the techniques that do exist.
"""


if TYPE_CHECKING:
from collections.abc import Callable, Mapping, Sequence

Expand Down Expand Up @@ -153,18 +164,47 @@ def resolve_technique_factories(

Returns:
dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by
the selected techniques.
the selected techniques. Techniques with no registered factory are skipped with a
warning naming them, so the caller can proceed with whatever techniques exist.

Raises:
TechniqueResolutionError: If the selection is nonempty but no technique resolves,
since running only the baseline would silently defeat the selection.
"""
from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry

all_factories = dict(AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise())
if extra_factories:
all_factories.update(extra_factories)
return {
technique.value: all_factories[technique.value]
for technique in context.scenario_techniques
if technique.value in all_factories
}

resolved: dict[str, AttackTechniqueFactory] = {}
missing: list[str] = []
seen_missing: set[str] = set()
for technique in context.scenario_techniques:
if technique.value in all_factories:
resolved[technique.value] = all_factories[technique.value]
elif technique.value not in seen_missing:
missing.append(technique.value)
seen_missing.add(technique.value)

if missing:
Comment thread
romanlutz marked this conversation as resolved.
logger.warning(
Comment thread
romanlutz marked this conversation as resolved.
"Skipping %d selected attack technique(s) with no registered factory: %s. "
"Register the technique(s) (or pass them via extra_factories) to include them in the run.",
len(missing),
", ".join(missing),
)

if context.scenario_techniques and not resolved:
# A nonempty selection that resolves to nothing would otherwise run the
# baseline only while reporting success — a silently empty evaluation.
raise TechniqueResolutionError(
f"All {len(context.scenario_techniques)} selected attack technique(s) have no registered "
f"factory: {', '.join(missing)}. Register the technique(s) (or pass them via "
"extra_factories) so the run has at least one technique to execute."
)

return resolved


def build_matrix_atomic_attacks(
Expand Down
68 changes: 68 additions & 0 deletions tests/unit/backend/services/test_scenario_run_service_summary.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""Tests for skipped-technique diagnostics in ``ScenarioRunService`` run summaries."""

import pytest

from pyrit.backend.services.scenario_run_service import ScenarioRunService
from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier
from pyrit.models.results.scenario_result import ScenarioResult


def _service() -> ScenarioRunService:
from unittest.mock import MagicMock

service = object.__new__(ScenarioRunService)
service._active_tasks = {}
# The error fallback path queries persisted error results; none exist here.
service._memory = MagicMock()
service._memory.get_attack_results.return_value = []
return service


def _result(*, techniques, display_groups) -> ScenarioResult:
return ScenarioResult(
scenario_identifier=ScenarioIdentifier(name="scenario", techniques=techniques),
attack_results={},
display_group_map=display_groups,
)


@pytest.mark.usefixtures("patch_central_database")
class TestBuildResponseSkippedTechniques:
"""Selected techniques with no built attack cell surface as ``skipped_techniques``."""

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new focused tests pass, but tests/unit/backend/test_scenario_run_service.py now has 28 failures because its MagicMock(spec=ScenarioResult) fixture does not define scenario_identifier. Please update that shared fixture, preferably to use a real ScenarioResult, and add a custom display-group regression case.


def test_selected_without_built_label_is_reported_skipped(self):
result = _result(techniques=["alpha", "ghost"], display_groups={"alpha::ds": "alpha"})

summary = _service()._build_response_from_db(scenario_result=result)

assert summary.skipped_techniques == ["ghost"]
assert summary.techniques_used is not None

def test_decorated_display_label_still_counts_as_built(self):
# Custom ``display_group_fn`` may decorate technique names; a label that
# contains the technique name must not be reported as skipped.
result = _result(techniques=["alpha"], display_groups={"cell-1": "alpha (hard mode)"})

summary = _service()._build_response_from_db(scenario_result=result)

assert summary.skipped_techniques == []

def test_no_selection_reports_no_skips(self):
result = _result(techniques=None, display_groups={})

summary = _service()._build_response_from_db(scenario_result=result)

assert summary.skipped_techniques == []

def test_skips_are_sorted_and_deduplicated(self):
result = _result(
techniques=["zeta", "alpha", "alpha"],
display_groups={"mid::ds": "mid"},
)

summary = _service()._build_response_from_db(scenario_result=result)

assert summary.skipped_techniques == ["alpha", "zeta"]
18 changes: 18 additions & 0 deletions tests/unit/cli/test_output.py
Original file line number Diff line number Diff line change
Expand Up @@ -541,6 +541,24 @@ def test_print_scenario_run_summary_completed(capsys):
assert "Completed:" not in captured.out


def test_print_scenario_run_summary_lists_skipped_techniques(capsys):
run = _make_run(
scenario_name="partial",
scenario_result_id="id",
status=ScenarioRunState.COMPLETED,
total_attacks=2,
completed_attacks=2,
objective_achieved_rate=50,
techniques_used=["s1"],
skipped_techniques=["ghost_tech"],
)
_output.print_scenario_run_summary(run=run)
captured = capsys.readouterr()
assert "Skipped:" in captured.out
assert "ghost_tech" in captured.out
assert "no registered factory" in captured.out


def test_print_scenario_run_summary_with_error(capsys):
run = _make_run(
scenario_name="failing",
Expand Down
67 changes: 67 additions & 0 deletions tests/unit/scenario/core/test_matrix_atomic_attack_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
* optional baseline emission from the flattened seed groups.
"""

import logging
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

Expand All @@ -26,6 +27,7 @@
from pyrit.scenario.core.matrix_atomic_attack_builder import (
MatrixAtomicAttackBuilder,
MatrixCombo,
TechniqueResolutionError,
build_baseline_atomic_attack,
build_matrix_atomic_attacks,
resolve_technique_factories,
Expand Down Expand Up @@ -408,6 +410,71 @@ def test_drops_techniques_without_factory(self):
resolved = resolve_technique_factories(context=context)
assert list(resolved.keys()) == ["alpha"]

def test_warns_when_dropping_techniques_without_factory(self, caplog):
factories = {"alpha": _mock_factory(name="alpha")}
context = _context(techniques=[_technique("alpha"), _technique("missing")])
with (
_patch_registry(factories),
caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"),
):
resolve_technique_factories(context=context)
assert any("missing" in record.message for record in caplog.records)

def test_no_warning_when_all_techniques_resolve(self, caplog):
factories = {
"alpha": _mock_factory(name="alpha"),
"beta": _mock_factory(name="beta"),
}
context = _context(techniques=[_technique("alpha"), _technique("beta")])
with (
_patch_registry(factories),
caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"),
):
resolve_technique_factories(context=context)
assert not [record for record in caplog.records if record.levelno == logging.WARNING]

def test_raises_when_all_selected_techniques_missing(self):
"""A nonempty selection resolving to nothing must fail loudly, not run baseline-only."""
factories = {"alpha": _mock_factory(name="alpha")}
context = _context(techniques=[_technique("missing_a"), _technique("missing_b")])
with _patch_registry(factories), pytest.raises(TechniqueResolutionError, match="missing_a"):
resolve_technique_factories(context=context)

def test_empty_selection_resolves_without_error(self):
context = _context(techniques=[])
with _patch_registry({}):
assert resolve_technique_factories(context=context) == {}

def test_partial_miss_still_warns_and_continues(self, caplog):
factories = {"alpha": _mock_factory(name="alpha")}
context = _context(techniques=[_technique("alpha"), _technique("missing")])
with (
_patch_registry(factories),
caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"),
):
resolved = resolve_technique_factories(context=context)
assert list(resolved.keys()) == ["alpha"]

def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog):
factories = {"alpha": _mock_factory(name="alpha")}
context = _context(
techniques=[
_technique("missing_a"),
_technique("alpha"),
_technique("missing_b"),
_technique("missing_a"),
]
)
with (
_patch_registry(factories),
caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"),
):
resolve_technique_factories(context=context)
messages = [record.message for record in caplog.records if record.levelno == logging.WARNING]
assert len(messages) == 1
assert messages[0].index("missing_a") < messages[0].index("missing_b")
assert messages[0].count("missing_a") == 1 # duplicates are deduplicated

def test_extra_factories_merged_and_override_registry(self):
registry_factories = {"alpha": _mock_factory(name="alpha")}
local_alpha = _mock_factory(name="alpha")
Expand Down