From d82db4cc5fcb9fe67a356dea7ea6023187d1dbf8 Mon Sep 17 00:00:00 2001 From: feiiiiii5 <204683769+feiiiiii5@users.noreply.github.com> Date: Sun, 23 Aug 2026 01:55:33 +0800 Subject: [PATCH 1/4] fix(scenario): warn when selected attack techniques have no registered factory resolve_technique_factories silently dropped techniques whose factory was not registered, so a typo or an unregistered custom technique shrank the run without any signal. Emit one warning naming the missing technique(s) in selection order (deduplicated), and update the docstring accordingly. Fixes #2461 --- .../core/matrix_atomic_attack_builder.py | 28 +++++++++--- .../core/test_matrix_atomic_attack_builder.py | 44 +++++++++++++++++++ 2 files changed, 66 insertions(+), 6 deletions(-) diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 40a35ca474..053dcd38b9 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -153,18 +153,34 @@ 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. """ 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: + logger.warning( + "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), + ) + + return resolved def build_matrix_atomic_attacks( diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 044ce78512..4c60be3c6b 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -15,6 +15,7 @@ * optional baseline emission from the flattened seed groups. """ +import logging from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -408,6 +409,49 @@ 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_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") From 1b3a4811bf2d29a27a5f95aaad17c1654b39e8ed Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Mon, 24 Aug 2026 19:27:07 +0800 Subject: [PATCH 2/4] fix(scenario): surface skipped techniques and fail fast when no selection resolves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the silent-technique-drop warning: - A nonempty technique selection that resolves to zero factories now raises TechniqueResolutionError instead of running baseline-only with a success status — a silently empty evaluation. Partial misses keep the warn-and-continue behavior. The error subclasses ValueError so existing handlers keep working, mirroring DatasetConstraintError. - Skipped selections are now visible in normal flows, not just pyrit_backend.log: ScenarioRunSummary gains skipped_techniques, populated by comparing the scenario identifier's resolved techniques against built display groups (execution-progress independent), and pyrit_scan renders a "Skipped:" line so users see when selected techniques were left out. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../backend/services/scenario_run_service.py | 13 ++++ pyrit/cli/_output.py | 3 + pyrit/models/catalog/scenario.py | 4 ++ .../core/matrix_atomic_attack_builder.py | 24 +++++++ .../test_scenario_run_service_summary.py | 68 +++++++++++++++++++ tests/unit/cli/test_output.py | 18 +++++ .../core/test_matrix_atomic_attack_builder.py | 23 +++++++ 7 files changed, 153 insertions(+) create mode 100644 tests/unit/backend/services/test_scenario_run_service_summary.py diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index 7ba66d0f43..6896eada19 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -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 []) + 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] = [] @@ -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(), diff --git a/pyrit/cli/_output.py b/pyrit/cli/_output.py index 53136e15c8..9c8b994ea1 100644 --- a/pyrit/cli/_output.py +++ b/pyrit/cli/_output.py @@ -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: diff --git a/pyrit/models/catalog/scenario.py b/pyrit/models/catalog/scenario.py index 488ccf8c78..1218e7d061 100644 --- a/pyrit/models/catalog/scenario.py +++ b/pyrit/models/catalog/scenario.py @@ -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)") diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 053dcd38b9..5224f409b6 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -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 @@ -155,6 +166,10 @@ def resolve_technique_factories( dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by 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 @@ -180,6 +195,15 @@ def resolve_technique_factories( ", ".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 diff --git a/tests/unit/backend/services/test_scenario_run_service_summary.py b/tests/unit/backend/services/test_scenario_run_service_summary.py new file mode 100644 index 0000000000..ba56157493 --- /dev/null +++ b/tests/unit/backend/services/test_scenario_run_service_summary.py @@ -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``.""" + + 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"] diff --git a/tests/unit/cli/test_output.py b/tests/unit/cli/test_output.py index 84c18fa05b..04bbb38474 100644 --- a/tests/unit/cli/test_output.py +++ b/tests/unit/cli/test_output.py @@ -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", diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 4c60be3c6b..e3e9cef5fd 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -27,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, @@ -432,6 +433,28 @@ def test_no_warning_when_all_techniques_resolve(self, caplog): 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( From a1600c20da6c9b8357d67d8b40eae7320bfad297 Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Tue, 25 Aug 2026 11:34:11 +0800 Subject: [PATCH 3/4] fix(scenario): persist authoritative skipped-technique names from resolution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round two on the skipped-techniques summary: - resolve_technique_factories() now returns a TechniqueResolution (resolved factories + skipped names) instead of a bare dict; jailbreak/adversarial record resolution.skipped and Scenario persists it into ScenarioResult.metadata["skipped_techniques"] at result creation. The run summary reads that authoritative record — display groups are presentation data (grouped by dataset/target/template depending on scenario) and cannot reveal which factories resolved. - tests/unit/backend/test_scenario_run_service.py's shared fixture now builds a real ScenarioResult (model_construct) carrying scenario_identifier and metadata, fixing the 28 spec-mock failures; added a custom-display-group regression proving labels don't affect reported skips. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../backend/services/scenario_run_service.py | 14 ++---- .../core/matrix_atomic_attack_builder.py | 29 +++++++++--- pyrit/scenario/core/scenario.py | 3 ++ pyrit/scenario/scenarios/airt/jailbreak.py | 5 +- .../scenarios/benchmark/adversarial.py | 4 +- .../test_scenario_run_service_summary.py | 47 ++++++++++++------- .../unit/backend/test_scenario_run_service.py | 45 +++++++++++------- .../core/test_matrix_atomic_attack_builder.py | 15 ++++-- 8 files changed, 105 insertions(+), 57 deletions(-) diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index 6896eada19..dc7a4aa9d5 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -599,16 +599,12 @@ 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 []) - built_labels = set(scenario_result.display_group_map.values()) + # Authoritative skip record persisted at resolution time by the scenario + # (see resolve_technique_factories). Display groups are presentation data — + # they group by dataset/target/template depending on the scenario, so they + # cannot be used to derive which factories resolved. 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) + set(scenario_result.metadata.get("skipped_techniques", []) or []) ) # Surface per-attack errors and retry pressure regardless of overall run status: diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 5224f409b6..9179049759 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -40,6 +40,23 @@ class TechniqueResolutionError(ValueError): """ +@dataclass(frozen=True) +class TechniqueResolution: + """ + Outcome of resolving a scenario's selected techniques to factories. + + Attributes: + resolved (dict[str, AttackTechniqueFactory]): Factories keyed by technique + name, ordered by the selection. + skipped (list[str]): Selected techniques with no registered factory, in + selection order. This is the authoritative record for surfacing skips — + display groups are presentation data and cannot be used to derive it. + """ + + resolved: dict[str, AttackTechniqueFactory] + skipped: list[str] + + if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence @@ -147,7 +164,7 @@ def resolve_technique_factories( *, context: ScenarioContext, extra_factories: dict[str, AttackTechniqueFactory] | None = None, -) -> dict[str, AttackTechniqueFactory]: +) -> TechniqueResolution: """ Resolve a run's selected techniques to their registered ``AttackTechniqueFactory`` instances. @@ -163,9 +180,9 @@ def resolve_technique_factories( name. Returns: - dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by - the selected techniques. Techniques with no registered factory are skipped with a - warning naming them, so the caller can proceed with whatever techniques exist. + TechniqueResolution: ``resolved`` maps technique name to factory ordered by the + selection; ``skipped`` lists selected techniques with no registered factory in + selection order (also emitted as a warning so callers proceed knowingly). Raises: TechniqueResolutionError: If the selection is nonempty but no technique resolves, @@ -204,7 +221,7 @@ def resolve_technique_factories( "extra_factories) so the run has at least one technique to execute." ) - return resolved + return TechniqueResolution(resolved=resolved, skipped=missing) def build_matrix_atomic_attacks( @@ -254,7 +271,7 @@ def build_matrix_atomic_attacks( memory_labels=context.memory_labels, ) return builder.build( - technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories), + technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories).resolved, dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, technique_converters=technique_converters, diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index b607c1dee0..fe5cc990fa 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -687,6 +687,9 @@ def _build_initial_scenario_metadata(self) -> dict[str, Any]: dict[str, Any]: Metadata payload for the new ScenarioResult. """ metadata: dict[str, Any] = {} + # Authoritative record of selected techniques that had no registered factory, + # captured during attack construction (see resolve_technique_factories). + metadata["skipped_techniques"] = sorted(set(getattr(self, "_skipped_techniques", []) or [])) if getattr(self._dataset_config, "max_dataset_size", None) is None: return metadata hashes: list[str] = [] diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 710c7eba7c..f4bc3603a2 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -313,7 +313,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_jailbreaks = self._resolve_templates() num_attempts = self.params.get("num_jailbreak_attempts", 1) - technique_factories = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) + resolution = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) + technique_factories = resolution.resolved + # Authoritative skip record for the run summary (display groups are presentation data). + self._skipped_techniques = resolution.skipped # ``jailbreak_system_prompt`` is delivered separately (native system prompt, no converter); # every other technique goes through the inline converter path. diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 006e695c75..72bb3f9b21 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -222,7 +222,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list ) resolved_targets = self._resolve_adversarial_targets(target_names=target_names) - technique_factories = resolve_technique_factories(context=context) + resolution = resolve_technique_factories(context=context) + technique_factories = resolution.resolved + self._skipped_techniques = resolution.skipped builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, diff --git a/tests/unit/backend/services/test_scenario_run_service_summary.py b/tests/unit/backend/services/test_scenario_run_service_summary.py index ba56157493..eca63ce8ce 100644 --- a/tests/unit/backend/services/test_scenario_run_service_summary.py +++ b/tests/unit/backend/services/test_scenario_run_service_summary.py @@ -21,48 +21,59 @@ def _service() -> ScenarioRunService: return service -def _result(*, techniques, display_groups) -> ScenarioResult: - return ScenarioResult( - scenario_identifier=ScenarioIdentifier(name="scenario", techniques=techniques), +def _result(*, techniques, skipped, display_group_map=None) -> ScenarioResult: + return ScenarioResult.model_construct( + scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=techniques), attack_results={}, - display_group_map=display_groups, + display_group_map=display_group_map or {}, + metadata={"skipped_techniques": sorted(skipped)}, ) @pytest.mark.usefixtures("patch_central_database") class TestBuildResponseSkippedTechniques: - """Selected techniques with no built attack cell surface as ``skipped_techniques``.""" + """``skipped_techniques`` comes from the resolution-time record the scenario persists.""" - def test_selected_without_built_label_is_reported_skipped(self): - result = _result(techniques=["alpha", "ghost"], display_groups={"alpha::ds": "alpha"}) + def test_persisted_skips_are_reported_verbatim(self): + result = _result(techniques=["alpha", "ghost"], skipped=["ghost"], display_group_map={"cell": "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)"}) + def test_display_group_labels_do_not_affect_reporting(self): + """Custom display-group functions rename cells; skips must still surface verbatim.""" + result = _result( + techniques=["alpha", "ghost"], + skipped=["ghost"], + display_group_map={"cell-a": "alpha (hard mode)", "cell-b": "ghost (hard mode)"}, + ) summary = _service()._build_response_from_db(scenario_result=result) - assert summary.skipped_techniques == [] + assert summary.skipped_techniques == ["ghost"] def test_no_selection_reports_no_skips(self): - result = _result(techniques=None, display_groups={}) + result = _result(techniques=None, skipped=[]) 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"}, - ) + result = _result(techniques=["zeta", "alpha", "alpha"], skipped=["zeta", "alpha", "alpha"]) summary = _service()._build_response_from_db(scenario_result=result) assert summary.skipped_techniques == ["alpha", "zeta"] + + def test_legacy_results_without_metadata_report_no_skips(self): + result = ScenarioResult.model_construct( + scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=["x"]), + attack_results={}, + metadata={}, + ) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == [] diff --git a/tests/unit/backend/test_scenario_run_service.py b/tests/unit/backend/test_scenario_run_service.py index 8ce480c62f..df0b6ac8b1 100644 --- a/tests/unit/backend/test_scenario_run_service.py +++ b/tests/unit/backend/test_scenario_run_service.py @@ -19,6 +19,7 @@ from pyrit.converter import Converter from pyrit.models import AttackOutcome, ScenarioResult, ScenarioRunState from pyrit.models.catalog.scenario import RunScenarioRequest +from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier from pyrit.scenario.core import DatasetAttackConfiguration, DatasetConfiguration from pyrit.scenario.core.scenario_technique import ScenarioTechnique from unit.mocks import make_scenario_result @@ -88,23 +89,33 @@ def _make_db_scenario_result( run_state: ScenarioRunState = ScenarioRunState.IN_PROGRESS, attack_results: dict | None = None, ) -> MagicMock: - """Create a mock ScenarioResult as returned by CentralMemory.""" - sr = MagicMock(spec=ScenarioResult) - sr.id = result_id - sr.scenario_name = scenario_name - sr.scenario_version = 1 - sr.scenario_run_state = run_state - sr.get_techniques_used.return_value = [] - sr.attack_results = attack_results or {} - sr.number_tries = 1 - sr.creation_time = datetime(2025, 1, 1, tzinfo=timezone.utc) - sr.completion_time = datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc) - sr.labels = {} - sr.objective_achieved_rate.return_value = 0 - sr.get_display_groups.return_value = {} - sr.display_group_map = {} - sr.error_message = None - sr.error_type = None + """Create a real ``ScenarioResult`` as CentralMemory would return it. + + A real object (not a spec'd mock) because the summary builder reads + ``scenario_identifier`` / ``metadata``, which a spec'd mock does not define. + """ + sr = ScenarioResult.model_construct( + id=result_id, + scenario_identifier=ScenarioIdentifier(class_name=scenario_name, techniques=[]), + attack_results=attack_results or {}, + scenario_run_state=run_state, + number_tries=1, + creation_time=datetime(2025, 1, 1, tzinfo=timezone.utc), + completion_time=datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc), + labels={}, + display_group_map={}, + error_message=None, + error_type=None, + metadata={"skipped_techniques": []}, + ) + + # Individual tests configure these; real methods would need live data. + from unittest.mock import MagicMock as MockFactory + + object.__setattr__(sr, "get_techniques_used", MockFactory(return_value=[])) + object.__setattr__(sr, "objective_achieved_rate", MockFactory(return_value=0)) + object.__setattr__(sr, "get_display_groups", MockFactory(return_value={})) + return sr diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index e3e9cef5fd..d2756e3bee 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -400,14 +400,16 @@ def test_keeps_only_selected_in_order(self): } context = _context(techniques=[_technique("beta"), _technique("alpha")]) with _patch_registry(factories): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["beta", "alpha"] def test_drops_techniques_without_factory(self): factories = {"alpha": _mock_factory(name="alpha")} context = _context(techniques=[_technique("alpha"), _technique("missing")]) with _patch_registry(factories): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["alpha"] def test_warns_when_dropping_techniques_without_factory(self, caplog): @@ -443,7 +445,7 @@ def test_raises_when_all_selected_techniques_missing(self): def test_empty_selection_resolves_without_error(self): context = _context(techniques=[]) with _patch_registry({}): - assert resolve_technique_factories(context=context) == {} + assert resolve_technique_factories(context=context).resolved == {} def test_partial_miss_still_warns_and_continues(self, caplog): factories = {"alpha": _mock_factory(name="alpha")} @@ -452,7 +454,8 @@ def test_partial_miss_still_warns_and_continues(self, caplog): _patch_registry(factories), caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), ): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["alpha"] def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog): @@ -481,11 +484,13 @@ def test_extra_factories_merged_and_override_registry(self): local_only = _mock_factory(name="local") context = _context(techniques=[_technique("alpha"), _technique("local")]) with _patch_registry(registry_factories): - resolved = resolve_technique_factories( + resolution = resolve_technique_factories( context=context, extra_factories={"alpha": local_alpha, "local": local_only}, ) + resolved = resolution.resolved assert list(resolved.keys()) == ["alpha", "local"] + assert resolution.skipped == [] assert resolved["alpha"] is local_alpha # extra overrides the registry factory of the same name assert resolved["local"] is local_only # local-only factory is selectable without global registration From 6b6755178492bbd821c2dba206fa93a5aec04413 Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Tue, 25 Aug 2026 13:41:45 +0800 Subject: [PATCH 4/4] fix(scenario): propagate skipped techniques through build_matrix_atomic_attacks and multilingual Review round two follow-ups: build_matrix_atomic_attacks now returns (attacks, skipped) so Cyber/Leakage/RapidResponse persist the authoritative skip record before ScenarioResult creation, and Multilingual consumes resolution.resolved while recording resolution.skipped (its dict-style use of the old return type caused 9 TypeErrors after the TechniqueResolution change). Jailbreak's mocked resolver updated to the new contract. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../core/matrix_atomic_attack_builder.py | 12 +++-- pyrit/scenario/scenarios/airt/cyber.py | 4 +- pyrit/scenario/scenarios/airt/leakage.py | 4 +- pyrit/scenario/scenarios/airt/multilingual.py | 4 +- .../scenario/scenarios/airt/rapid_response.py | 4 +- tests/unit/scenario/airt/test_jailbreak.py | 6 ++- .../core/test_matrix_atomic_attack_builder.py | 54 +++++++++++-------- 7 files changed, 57 insertions(+), 31 deletions(-) diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 9179049759..1b970d5853 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -262,21 +262,25 @@ def build_matrix_atomic_attacks( can offer techniques without registering them globally. Returns: - list[AtomicAttack]: The generated atomic attacks, baseline first when - ``context.include_baseline`` is set. + tuple[list[AtomicAttack], list[str]]: The generated atomic attacks (baseline + first when ``context.include_baseline`` is set) and the names of selected + techniques that had no registered factory, so callers can persist the + authoritative skip record before creating the ``ScenarioResult``. """ builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=objective_scorer, memory_labels=context.memory_labels, ) - return builder.build( - technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories).resolved, + resolution = resolve_technique_factories(context=context, extra_factories=extra_factories) + attacks = builder.build( + technique_factories=resolution.resolved, dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, technique_converters=technique_converters, include_baseline=context.include_baseline, ) + return attacks, resolution.skipped class MatrixAtomicAttackBuilder: diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 2f622c7b41..1c5af7c6af 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -123,8 +123,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, ) + self._skipped_techniques = skipped + return attacks diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 264035f0ae..5113b3c5d3 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -140,9 +140,11 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, extra_factories={factory.name: factory for factory in _leakage_factories()}, ) + self._skipped_techniques = skipped + return attacks diff --git a/pyrit/scenario/scenarios/airt/multilingual.py b/pyrit/scenario/scenarios/airt/multilingual.py index bc70be31e6..a1c48afb0d 100644 --- a/pyrit/scenario/scenarios/airt/multilingual.py +++ b/pyrit/scenario/scenarios/airt/multilingual.py @@ -308,10 +308,12 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_languages = self._resolve_languages() adversarial_chat = self._adversarial_chat or get_default_adversarial_target() strategies = set(self.params.get("translation_strategies") or [_TRANSLATION, _RANDOM_TRANSLATION]) - technique_factories = resolve_technique_factories( + resolution = resolve_technique_factories( context=context, extra_factories=_extra_default_factories(), ) + technique_factories = resolution.resolved + self._skipped_techniques = resolution.skipped builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=self._objective_scorer, diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 4fd292bbe8..d11ead6b2f 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -124,9 +124,11 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.dataset_name, technique_converters=self._technique_converters, ) + self._skipped_techniques = skipped + return attacks diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index b055e087aa..93aa1ce2de 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -19,6 +19,7 @@ from pyrit.registry.components.scenario_registry import ScenarioRegistry from pyrit.scenario.core import BaselineAttackPolicy from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory +from pyrit.scenario.core.matrix_atomic_attack_builder import TechniqueResolution from pyrit.scenario.scenarios.airt.jailbreak import ( _DEFAULT_NUM_JAILBREAKS, _DEFAULT_TECHNIQUES, @@ -392,7 +393,10 @@ async def test_missing_runtime_factory_is_rejected( self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups ): with _patch_seed_groups(mock_memory_seed_groups): - with patch("pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", return_value={}): + with patch( + "pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", + return_value=TechniqueResolution(resolved={}, skipped=[]), + ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) scenario.set_params_from_args(args=_default_args(mock_objective_target, jailbreak_names=["aim.yaml"])) with pytest.raises(ValueError, match="no longer available.*prompt_sending"): diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index d2756e3bee..b52331d271 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -495,6 +495,7 @@ def test_extra_factories_merged_and_override_registry(self): assert resolved["local"] is local_only # local-only factory is selectable without global registration +@pytest.mark.usefixtures("patch_central_database") @pytest.mark.usefixtures("patch_central_database") class TestBuildMatrixAtomicAttacks: """``build_matrix_atomic_attacks`` wires the context into the builder in one call.""" @@ -502,46 +503,53 @@ class TestBuildMatrixAtomicAttacks: def test_builds_cross_product_grouped_by_technique(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert [a.atomic_attack_name for a in result] == ["tech_ds"] - assert result[0].display_group == "tech" + attacks, skipped = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert [a.atomic_attack_name for a in attacks] == ["tech_ds"] + assert attacks[0].display_group == "tech" + assert skipped == [] def test_custom_display_group_fn(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks( + attacks, _ = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), display_group_fn=lambda combo: combo.dataset_name, ) - assert result[0].display_group == "ds" + assert attacks[0].display_group == "ds" def test_no_baseline_emitted_when_context_disables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, include_baseline=False, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert all(a.atomic_attack_name != "baseline" for a in result) + attacks, _ = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert all(a.atomic_attack_name != "baseline" for a in attacks) def test_baseline_emitted_when_context_enables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, include_baseline=True, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert result[0].atomic_attack_name == "baseline" - assert [a.atomic_attack_name for a in result] == ["baseline", "tech_ds"] + attacks, _ = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert attacks[0].atomic_attack_name == "baseline" + assert [a.atomic_attack_name for a in attacks] == ["baseline", "tech_ds"] def test_technique_converters_forwarded(self): from pyrit.converter import Converter @@ -553,7 +561,7 @@ def test_technique_converters_forwarded(self): factory = _mock_factory(name="tech") converter = MagicMock(spec=Converter) with _patch_registry({"tech": factory}): - build_matrix_atomic_attacks( + attacks, _ = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), technique_converters={"tech": [converter]}, @@ -563,15 +571,17 @@ def test_technique_converters_forwarded(self): assert len(extra) == 1 def test_extra_factories_used_for_selection(self): + local_alpha = _mock_factory(name="alpha") + local_only = _mock_factory(name="local") context = _context( - techniques=[_technique("local")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + techniques=[_technique("alpha"), _technique("local")], + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) - # The selected technique exists only in extra_factories, not the registry. - with _patch_registry({"other": _mock_factory(name="other")}): - result = build_matrix_atomic_attacks( + with _patch_registry({}): + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), - extra_factories={"local": _mock_factory(name="local")}, + extra_factories={"alpha": local_alpha, "local": local_only}, ) - assert [a.atomic_attack_name for a in result] == ["local_ds"] + assert [a.atomic_attack_name for a in attacks] == ["alpha_ds", "local_ds"] + assert skipped == []