MAINT: Refactoring scenario strategy selection

doc/code/scenarios/0_scenarios.ipynb

@@ -56,7 +56,7 @@
     "1. **Strategy Enum**: Create a `ScenarioStrategy` enum that defines the available strategies for your scenario.\n",
     "   - Each enum member is defined as `(value, tags)` where value is a string and tags is a set of strings\n",
     "   - Include an `ALL` aggregate strategy that expands to all available strategies\n",
-    "   - Optionally implement `supports_composition()` and `validate_composition()` for strategy composition rules\n",
+    "   - Optionally override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`)\n",
     "\n",
     "2. **Scenario Class**: Extend `Scenario` and implement these abstract methods:\n",
     "   - `get_strategy_class()`: Return your strategy enum class\n",
@@ -108,7 +108,6 @@
     "    Scenario,\n",
     "    ScenarioStrategy,\n",
     ")\n",
-    "from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy\n",
     "from pyrit.score.true_false.true_false_scorer import TrueFalseScorer\n",
     "from pyrit.setup import initialize_pyrit_async\n",
     "\n",
@@ -162,19 +161,14 @@
     "        Build atomic attacks based on selected strategies.\n",
     "\n",
     "        This method is called by initialize_async() after strategies are prepared.\n",
-    "        Use self._scenario_composites to access the selected strategies.\n",
+    "        Use self._scenario_strategies to access the resolved strategy list.\n",
     "        \"\"\"\n",
     "        atomic_attacks = []\n",
     "\n",
     "        # objective_target is guaranteed to be non-None by parent class validation\n",
     "        assert self._objective_target is not None\n",
     "\n",
-    "        # Extract individual strategy values from the composites\n",
-    "        selected_strategies = ScenarioCompositeStrategy.extract_single_strategy_values(\n",
-    "            self._scenario_composites, strategy_type=MyStrategy\n",
-    "        )\n",
-    "\n",
-    "        for strategy in selected_strategies:\n",
+    "        for strategy in self._scenario_strategies:\n",
     "            # self._dataset_config is set by the parent class\n",
     "            seed_groups = self._dataset_config.get_all_seed_groups()\n",
     "\n",
@@ -185,7 +179,7 @@
     "            )\n",
     "            atomic_attacks.append(\n",
     "                AtomicAttack(\n",
-    "                    atomic_attack_name=strategy,\n",
+    "                    atomic_attack_name=strategy.value,\n",
     "                    attack=attack,\n",
     "                    seed_groups=seed_groups,  # type: ignore[arg-type]\n",
     "                    memory_labels=self._memory_labels,\n",

doc/code/scenarios/0_scenarios.py

@@ -62,7 +62,7 @@
 # 1. **Strategy Enum**: Create a `ScenarioStrategy` enum that defines the available strategies for your scenario.
 #    - Each enum member is defined as `(value, tags)` where value is a string and tags is a set of strings
 #    - Include an `ALL` aggregate strategy that expands to all available strategies
-#    - Optionally implement `supports_composition()` and `validate_composition()` for strategy composition rules
+#    - Optionally override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`)
 #
 # 2. **Scenario Class**: Extend `Scenario` and implement these abstract methods:
 #    - `get_strategy_class()`: Return your strategy enum class
@@ -96,7 +96,6 @@
     Scenario,
     ScenarioStrategy,
 )
-from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy
 from pyrit.score.true_false.true_false_scorer import TrueFalseScorer
 from pyrit.setup import initialize_pyrit_async
 
@@ -150,19 +149,14 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         Build atomic attacks based on selected strategies.
 
         This method is called by initialize_async() after strategies are prepared.
-        Use self._scenario_composites to access the selected strategies.
+        Use self._scenario_strategies to access the resolved strategy list.
         """
         atomic_attacks = []
 
         # objective_target is guaranteed to be non-None by parent class validation
         assert self._objective_target is not None
 
-        # Extract individual strategy values from the composites
-        selected_strategies = ScenarioCompositeStrategy.extract_single_strategy_values(
-            self._scenario_composites, strategy_type=MyStrategy
-        )
-
-        for strategy in selected_strategies:
+        for strategy in self._scenario_strategies:
             # self._dataset_config is set by the parent class
             seed_groups = self._dataset_config.get_all_seed_groups()
 
@@ -173,7 +167,7 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
             )
             atomic_attacks.append(
                 AtomicAttack(
-                    atomic_attack_name=strategy,
+                    atomic_attack_name=strategy.value,
                     attack=attack,
                     seed_groups=seed_groups,  # type: ignore[arg-type]
                     memory_labels=self._memory_labels,

doc/code/scenarios/1_scenario_parameters.ipynb

@@ -160,8 +160,8 @@
    "id": "10",
    "metadata": {},
    "source": [
-    "**Composite strategies** — multiple converters applied together in sequence using\n",
-    "`ScenarioCompositeStrategy`:"
+    "**Composite strategies** — pair an attack with one or more converters using `FoundryComposite`.\n",
+    "For example, to run Crescendo with Base64 encoding applied:"
    ]
   },
   {
@@ -171,9 +171,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from pyrit.scenario import ScenarioCompositeStrategy\n",
+    "from pyrit.scenario.scenarios.foundry import FoundryComposite\n",
     "\n",
-    "composite_strategy = [ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])]"
+    "composite_strategy = [FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64])]"
    ]
   },
   {
@@ -194,7 +194,7 @@
     "scenario_strategies = [\n",
     "    FoundryStrategy.Base64,\n",
     "    FoundryStrategy.Binary,\n",
-    "    ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]),\n",
+    "    FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar]),\n",
     "]"
    ]
   },

doc/code/scenarios/1_scenario_parameters.py

@@ -79,13 +79,13 @@
 aggregate_strategy = [FoundryStrategy.EASY]
 
 # %% [markdown]
-# **Composite strategies** — multiple converters applied together in sequence using
-# `ScenarioCompositeStrategy`:
+# **Composite strategies** — pair an attack with one or more converters using `FoundryComposite`.
+# For example, to run Crescendo with Base64 encoding applied:
 
 # %%
-from pyrit.scenario import ScenarioCompositeStrategy
+from pyrit.scenario.scenarios.foundry import FoundryComposite
 
-composite_strategy = [ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])]
+composite_strategy = [FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64])]
 
 # %% [markdown]
 # You can mix all three types in a single list:
@@ -94,7 +94,7 @@
 scenario_strategies = [
     FoundryStrategy.Base64,
     FoundryStrategy.Binary,
-    ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]),
+    FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar]),
 ]
 
 # %% [markdown]

pyrit/scenario/core/dataset_configuration.py

@@ -20,7 +20,7 @@
     from collections.abc import Sequence
 
     from pyrit.models.seeds.seed import Seed
-    from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy
+    from pyrit.scenario.core.scenario_strategy import ScenarioStrategy
 
 # Key used when seed_groups are provided directly (not from a named dataset)
 EXPLICIT_SEED_GROUPS_KEY = "_explicit_seed_groups"
@@ -38,7 +38,7 @@ class DatasetConfiguration:
         dataset_names (Optional[List[str]]): Names of datasets to load from memory.
         max_dataset_size (Optional[int]): If set, randomly samples up to this many SeedGroups
             from the configured dataset source (without replacement, so no duplicates).
-        scenario_composites (Optional[Sequence[ScenarioCompositeStrategy]]): The scenario
+        scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The scenario
             strategies being executed. Subclasses can use this to filter or customize
             which seed groups are loaded based on the selected strategies.
     """
@@ -49,7 +49,7 @@ def __init__(
         seed_groups: Optional[list[SeedGroup]] = None,
         dataset_names: Optional[list[str]] = None,
         max_dataset_size: Optional[int] = None,
-        scenario_composites: Optional[Sequence[ScenarioCompositeStrategy]] = None,
+        scenario_strategies: Optional[Sequence[ScenarioStrategy]] = None,
     ) -> None:
         """
         Initialize a DatasetConfiguration.
@@ -59,7 +59,7 @@ def __init__(
             dataset_names (Optional[List[str]]): Names of datasets to load from memory.
             max_dataset_size (Optional[int]): If set, randomly samples up to this many SeedGroups
                 (without replacement).
-            scenario_composites (Optional[Sequence[ScenarioCompositeStrategy]]): The scenario
+            scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The scenario
                 strategies being executed. Subclasses can use this to filter or customize
                 which seed groups are loaded.
 
@@ -82,7 +82,7 @@ def __init__(
         self._seed_groups = list(seed_groups) if seed_groups is not None else None
         self.max_dataset_size = max_dataset_size
         self._dataset_names = list(dataset_names) if dataset_names is not None else None
-        self._scenario_composites = scenario_composites
+        self._scenario_strategies = scenario_strategies
 
     def get_seed_groups(self) -> dict[str, list[SeedGroup]]:
         """

pyrit/scenario/core/scenario.py

@@ -29,10 +29,7 @@
 from pyrit.scenario.core.atomic_attack import AtomicAttack
 from pyrit.scenario.core.attack_technique import AttackTechnique
 from pyrit.scenario.core.dataset_configuration import DatasetConfiguration
-from pyrit.scenario.core.scenario_strategy import (
-    ScenarioCompositeStrategy,
-    ScenarioStrategy,
-)
+from pyrit.scenario.core.scenario_strategy import ScenarioStrategy
 from pyrit.score import Scorer, SelfAskRefusalScorer, TrueFalseInverterScorer, TrueFalseScorer
 
 if TYPE_CHECKING:
@@ -114,8 +111,8 @@ def __init__(
 
         self._include_baseline = include_default_baseline
 
-        # Store prepared strategy composites for use in _get_atomic_attacks_async
-        self._scenario_composites: list[ScenarioCompositeStrategy] = []
+        # Store prepared strategies for use in _get_atomic_attacks_async
+        self._scenario_strategies: list[ScenarioStrategy] = []
 
         # Store original objectives for each atomic attack (before any mutations)
         # Key: atomic_attack_name, Value: tuple of original objectives
@@ -186,12 +183,34 @@ def _get_default_objective_scorer(self) -> TrueFalseScorer:
         logger.info(f"No registered default objective scorer found, using fallback: {type(scorer).__name__}")
         return scorer
 
+    def _prepare_strategies(
+        self,
+        strategies: Optional[Sequence[ScenarioStrategy]],
+    ) -> list[ScenarioStrategy]:
+        """
+        Resolve strategy inputs into a concrete list for this scenario.
+
+        The default implementation calls resolve() on the strategy class, which handles
+        None (use default), empty list (baseline-only), and aggregate expansion.
+
+        Subclasses with complex composition semantics (e.g., RedTeamAgent with
+        FoundryComposite) should override this to build their own composite types.
+
+        Args:
+            strategies: Strategy inputs from initialize_async. None means use default,
+                [] means baseline-only, otherwise a list of strategies to resolve.
+
+        Returns:
+            list[ScenarioStrategy]: Ordered, deduplicated concrete strategies.
+        """
+        return self._strategy_class.resolve(strategies, default=self.get_default_strategy())
+
     @apply_defaults
     async def initialize_async(
         self,
         *,
         objective_target: PromptTarget = REQUIRED_VALUE,  # type: ignore[assignment]
-        scenario_strategies: Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]] = None,
+        scenario_strategies: Optional[Sequence[ScenarioStrategy]] = None,
         dataset_config: Optional[DatasetConfiguration] = None,
         max_concurrency: int = 10,
         max_retries: int = 0,
@@ -210,10 +229,8 @@ async def initialize_async(
 
         Args:
             objective_target (PromptTarget): The target system to attack.
-            scenario_strategies (Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]]):
-                The strategies to execute. Can be a list of bare ScenarioStrategy enums or
-                ScenarioCompositeStrategy instances for advanced composition. Bare enums are
-                automatically wrapped into composites. If None, uses the default aggregate
+            scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The strategies to execute.
+                Can be a list of ScenarioStrategy enum members. If None, uses the default aggregate
                 from the scenario's configuration.
             dataset_config (Optional[DatasetConfiguration]): Configuration for the dataset source.
                 Use this to specify dataset names or maximum dataset size from the CLI.
@@ -246,11 +263,7 @@ async def initialize_async(
         self._memory_labels = memory_labels or {}
 
         # Prepare scenario strategies using the stored configuration
-        # Allow empty strategies when include_baseline is True (baseline-only execution)
-        self._scenario_composites = self._strategy_class.prepare_scenario_strategies(
-            scenario_strategies,
-            default_aggregate=self.get_default_strategy(),
-        )
+        self._scenario_strategies = self._prepare_strategies(scenario_strategies)
 
         self._atomic_attacks = await self._get_atomic_attacks_async()