Skip to content
Open
16 changes: 11 additions & 5 deletions .github/instructions/scenarios.instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,8 @@ the constructor — no classmethod indirection required.

4. **Implement `_build_atomic_attacks_async(self, *, context)`** — this is the single
abstract extension point every scenario must define (see "AtomicAttack Construction" below).
Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.
Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns
the attacks plus the names of selected techniques that had no registered factory.

## Constructor Pattern

Expand Down Expand Up @@ -191,7 +192,7 @@ via `build_matrix_atomic_attacks`/`MatrixAtomicAttackBuilder`, pass a `display_g
callback that maps each `MatrixCombo` to a group string:

```python
build_matrix_atomic_attacks(
attacks, skipped = build_matrix_atomic_attacks(
context=context,
objective_scorer=self._objective_scorer,
display_group_fn=lambda combo: combo.technique_name, # default: group by technique
Expand Down Expand Up @@ -221,23 +222,28 @@ half-initialized `self._*` state to build attacks — read everything from `cont
### Zero-boilerplate matrix scenarios

Scenarios whose construction is the plain technique × dataset cross-product delegate to the
`build_matrix_atomic_attacks` helper in one line (see `Cyber`, `RapidResponse`):
`build_matrix_atomic_attacks` helper (see `Cyber`, `RapidResponse`). It returns the attacks
**and** the names of selected techniques that had no registered factory; record that second
value on `self._skipped_techniques` so the run reports the drop instead of hiding it:

```python
from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks

async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]:
return build_matrix_atomic_attacks(
attacks, skipped = build_matrix_atomic_attacks(
context=context,
objective_scorer=self._objective_scorer,
technique_converters=self._technique_converters, # optional CLI converter stacks
)
self._skipped_techniques = skipped
return attacks
```

`build_matrix_atomic_attacks`:
1. Calls `resolve_technique_factories(context=context)` to map the selected techniques to their
registered `AttackTechniqueFactory` instances (reads the `AttackTechniqueRegistry` singleton;
techniques with no registered factory are dropped).
techniques with no registered factory are dropped, warned about, and returned as the second
value, so a partial miss is visible in the run summary rather than silent).
2. Iterates every (technique × dataset) pair from `context.seed_groups_by_dataset`.
3. Calls `factory.create()` with the objective target, conditional scorer override, and any
per-technique converters (from `--techniques <technique>:converter.<name>`) as
Expand Down
12 changes: 8 additions & 4 deletions doc/code/scenarios/0_scenarios.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@
"2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:\n",
" - `technique_class`: Your technique enum class\n",
" - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point.\n",
" Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.\n",
" Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory.\n",
"\n",
"3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box.\n",
" - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=[\"my_dataset\"])`)\n",
Expand Down Expand Up @@ -171,13 +171,17 @@
"\n",
" # Implement the single abstract extension point. Matrix-shaped scenarios delegate\n",
" # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping\n",
" # (default groups by technique; here we group by dataset instead).\n",
" # (default groups by technique; here we group by dataset instead). The helper also\n",
" # reports which selected techniques had no registered factory, so record that skip\n",
" # list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary.\n",
" async def _build_atomic_attacks_async(self, *, context):\n",
" return build_matrix_atomic_attacks(\n",
" attacks, skipped = build_matrix_atomic_attacks(\n",
" context=context,\n",
" objective_scorer=self._objective_scorer,\n",
" display_group_fn=lambda combo: combo.dataset_name,\n",
" )"
" )\n",
" self._skipped_techniques = skipped\n",
" return attacks"
]
},
{
Expand Down
10 changes: 7 additions & 3 deletions doc/code/scenarios/0_scenarios.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@
# 2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:
# - `technique_class`: Your technique enum class
# - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point.
# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.
# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory.
#
# 3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box.
# - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=["my_dataset"])`)
Expand Down Expand Up @@ -148,13 +148,17 @@ def __init__(

# Implement the single abstract extension point. Matrix-shaped scenarios delegate
# to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping
# (default groups by technique; here we group by dataset instead).
# (default groups by technique; here we group by dataset instead). The helper also
# reports which selected techniques had no registered factory, so record that skip
# list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary.
async def _build_atomic_attacks_async(self, *, context):
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,
)
self._skipped_techniques = skipped
return attacks


# %% [markdown]
Expand Down
7 changes: 7 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,12 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari
completed_attacks = total_attacks
techniques_used = scenario_result.get_techniques_used()

# 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(set(scenario_result.metadata.get("skipped_techniques", []) or []))

# 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 +647,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 @@ -387,6 +387,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
87 changes: 74 additions & 13 deletions pyrit/scenario/core/matrix_atomic_attack_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,34 @@
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.
"""


@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

Expand Down Expand Up @@ -136,7 +164,7 @@ def resolve_technique_factories(
*,
context: ScenarioContext,
extra_factories: dict[str, AttackTechniqueFactory] | None = None,
) -> dict[str, AttackTechniqueFactory]:
) -> TechniqueResolution:
Comment thread
romanlutz marked this conversation as resolved.
"""
Resolve a run's selected techniques to their registered ``AttackTechniqueFactory`` instances.

Expand All @@ -152,19 +180,48 @@ def resolve_technique_factories(
name.

Returns:
dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by
the selected techniques.
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,
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 TechniqueResolution(resolved=resolved, skipped=missing)


def build_matrix_atomic_attacks(

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.

This function now returns (attacks, skipped) but is still annotated as list[AtomicAttack], which causes four ty errors in this helper and its callers. It also breaks the documented custom-scenario pattern in doc/code/scenarios/0_scenarios.py, where _build_atomic_attacks_async() directly returns this helper and now returns a tuple instead of the required attack list. Please fix the return contract and update both synchronized documentation files, or preserve the existing list-returning API.

Expand All @@ -174,7 +231,7 @@ def build_matrix_atomic_attacks(
display_group_fn: Callable[[MatrixCombo], str] | None = None,
technique_converters: dict[str, list[Converter]] | None = None,
extra_factories: dict[str, AttackTechniqueFactory] | None = None,
) -> list[AtomicAttack]:
) -> tuple[list[AtomicAttack], list[str]]:
"""
Build a matrix-shaped scenario's atomic attacks from its resolved context in one call.

Expand Down Expand Up @@ -205,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),
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:
Expand Down
3 changes: 3 additions & 0 deletions pyrit/scenario/core/scenario.py
Original file line number Diff line number Diff line change
Expand Up @@ -695,6 +695,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] = []
Expand Down
4 changes: 3 additions & 1 deletion pyrit/scenario/scenarios/airt/cyber.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
6 changes: 5 additions & 1 deletion pyrit/scenario/scenarios/airt/jailbreak.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,11 @@ 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

selected_names = {technique.value for technique in context.scenario_techniques}
missing = selected_names - set(technique_factories)
if missing:
Expand Down
4 changes: 3 additions & 1 deletion pyrit/scenario/scenarios/airt/leakage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
4 changes: 3 additions & 1 deletion pyrit/scenario/scenarios/airt/multilingual.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
4 changes: 3 additions & 1 deletion pyrit/scenario/scenarios/airt/rapid_response.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
4 changes: 3 additions & 1 deletion pyrit/scenario/scenarios/benchmark/adversarial.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,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,
Expand Down
Loading