FEAT: Add BenchmarkInitializer for adversarial-target fan-out

Registers one AttackTechniqueSpec variant per (adversarial-capable
technique, adversarial-tagged target) pair into AttackTechniqueRegistry
with the live target bound onto adversarial_chat. Variants are named
f"{source}__{target_name}" and tagged ["benchmark_fanout",
f"model:{target_name}"] so the benchmark scenario can discover them via
tag query in a later commit.

Adversarial-capability is determined by reusing _spec_needs_adversarial
from scenario_techniques (multi-turn attacks + crescendo-style simulated
conversations). Single-turn techniques without an adversarial chat
target (prompt_sending, role_play, many_shot, context_compliance) are
not fanned — the benchmark holds the objective target constant and
varies the adversarial chat helper across runs.

Placed at pyrit/setup/initializers/benchmark.py (top level) alongside
AIRTInitializer and SimpleInitializer, not under components/. The
components/ initializers (TargetInitializer, ScorerInitializer,
ScenarioTechniqueInitializer) are auto-bundled building blocks that
populate their registries during every PyRIT setup. BenchmarkInitializer
is the opposite shape: a user-opted workflow profile named after the
use case, listed in .pyrit_conf when the user wants a benchmarking
trial. The placement convention is itself underspecified in the
codebase and is tracked for follow-up.

Parameter contract (target_names: list[str] | None):

The optional target_names parameter is declared via
PyRITInitializer.supported_parameters, which is the single source of
truth shared across three consumer sites:

1. .pyrit_conf YAML:

       initializers:
         - name: benchmark
           args:
             target_names:
               - adversarial_chat_singleturn
               - adversarial_chat_reasoning

   ConfigurationLoader._resolve_initializers calls
   instance.set_params_from_args(args=config.args) and then
   _validate_params against supported_parameters, so unknown keys fail
   fast at config-load time. Omitting the args block uses the default
   (fan over every adversarial-tagged target).

2. CLI (--list-initializers via frontend_core._print_initializer_meta):
   reads metadata.supported_parameters and prints name + description +
   default for each declared parameter, so users discover what they can
   put in .pyrit_conf without reading source.

3. GUI backend (InitializerService): wraps each declared parameter as an
   InitializerParameterSummary({name, description, default}) on the
   RegisteredInitializer Pydantic model. The GUI renders form fields
   from this metadata.

All three paths terminate at the same self.params dict that
initialize_async reads via self.params.get("target_names"), so adding,
renaming, or retyping the parameter is a single-site change.

target_names narrows fan-out to a subset of adversarial targets by
registry name; unknown names raise ValueError listing both the unknowns
and the discovered set. Empty discovery raises ValueError naming the
ADVERSARIAL_CHAT_* env vars and the TargetInitializer ordering
dependency (closes one of the failure-mode follow-ups surfaced in the
previous commit).

Failure modes audited during this change but not addressed here
(tracked for the PR description batch):
- AttackTechniqueRegistry.register_from_specs is first-write-wins on
  name collision with no log entry. Disjoint name spaces between
  ScenarioTechniqueInitializer and BenchmarkInitializer mean this is
  inert today; future extensions that produce colliding names would be
  silently no-op'd.
- BenchmarkInitializer's TargetRegistry walk is a snapshot at init
  time; later mutations to TargetRegistry leave the fanned specs
  holding stale references. Failure surfaces at API-call time, not at
  registration.
- Top-level vs components/ initializer placement convention is
  implicit; this commit picks "top-level for workflow profiles",
  matching AIRT/Simple. Worth a CONTRIBUTING note when convention is
  formalized.

Author: Victor Valbuena

pyrit/setup/initializers/__init__.py

@@ -6,6 +6,7 @@
 from pyrit.common.deprecation import print_deprecation_message
 from pyrit.common.parameter import Parameter
 from pyrit.setup.initializers.airt import AIRTInitializer
+from pyrit.setup.initializers.benchmark import BenchmarkInitializer
 from pyrit.setup.initializers.components.scenarios import ScenarioTechniqueInitializer
 from pyrit.setup.initializers.components.scorers import ScorerInitializer
 from pyrit.setup.initializers.components.targets import TargetInitializer
@@ -18,6 +19,7 @@
     "Parameter",
     "PyRITInitializer",
     "AIRTInitializer",
+    "BenchmarkInitializer",
     "ScenarioTechniqueInitializer",
     "ScorerInitializer",
     "TargetInitializer",

pyrit/setup/initializers/benchmark.py

@@ -0,0 +1,198 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Benchmark initializer that fans adversarial-capable scenario techniques across
+adversarial targets discovered in ``TargetRegistry``.
+
+This is the entry point for bootstrapping an ``AdversarialBenchmark`` trial.
+It queries ``TargetRegistry`` for entries tagged ``ADVERSARIAL`` (via
+``TagQuery.all("adversarial")``), then for every adversarial-capable
+technique in ``SCENARIO_TECHNIQUES`` builds one fanned
+``AttackTechniqueSpec`` per discovered target. Each fanned spec binds the
+live target onto ``adversarial_chat`` and is registered into
+``AttackTechniqueRegistry`` tagged ``["benchmark_fanout", f"model:{name}"]``
+so the benchmark scenario can discover them via tag query in a later commit.
+
+The ``target_names`` parameter (optional, settable from ``.pyrit_conf``)
+narrows the fan-out to a specific subset of adversarial targets by registry
+name. Unknown names raise ``ValueError``.
+
+Discovery returning no adversarial-tagged targets raises ``ValueError`` with
+an actionable message pointing at the ``ADVERSARIAL_CHAT_*`` env vars and
+the ``TargetInitializer`` dependency.
+"""
+
+import dataclasses
+import logging
+
+from pyrit.common.parameter import Parameter
+from pyrit.registry import TargetRegistry
+from pyrit.registry.object_registries.attack_technique_registry import (
+    AttackTechniqueRegistry,
+    AttackTechniqueSpec,
+)
+from pyrit.registry.tag_query import TagQuery
+from pyrit.scenario.core.scenario_techniques import SCENARIO_TECHNIQUES, _spec_needs_adversarial
+from pyrit.setup.initializers.components.targets import TargetInitializerTags
+from pyrit.setup.initializers.pyrit_initializer import PyRITInitializer
+
+logger = logging.getLogger(__name__)
+
+
+#: Default discovery query used when no ``target_names`` override is provided.
+#: Resolves every ``TargetRegistry`` entry tagged ``ADVERSARIAL`` (which today
+#: includes ``adversarial_chat`` plus the ``ADVERSARIAL_CHAT_{SINGLETURN,
+#: MULTITURN,REASONING}`` variants — all set by ``TargetInitializer``).
+DEFAULT_ADVERSARIAL_TAG_QUERY: TagQuery = TagQuery.all(TargetInitializerTags.ADVERSARIAL.value)
+
+
+class BenchmarkInitializer(PyRITInitializer):
+    """
+    Fan adversarial-capable scenario techniques across discovered adversarial targets.
+
+    For every ``AttackTechniqueSpec`` in ``SCENARIO_TECHNIQUES`` that uses an
+    adversarial chat target (multi-turn attacks plus crescendo-style
+    simulated conversations), this initializer registers one variant per
+    discovered adversarial target with the target bound onto
+    ``adversarial_chat``. Variants are named
+    ``f"{source_spec.name}__{target_name}"`` (e.g. ``red_teaming__adversarial_chat_singleturn``)
+    and carry the additional strategy tags ``"benchmark_fanout"`` and
+    ``f"model:{target_name}"`` so the benchmark scenario can query the
+    registry by tag in a later commit.
+
+    Parameters (declared via :attr:`supported_parameters`):
+
+    * ``target_names`` (``list[str]``, optional): Narrow fan-out to a
+      specific subset of adversarial targets by registry name. When omitted,
+      every target matching :data:`DEFAULT_ADVERSARIAL_TAG_QUERY` is used.
+
+    Raises (at ``initialize_async``):
+
+    * ``ValueError`` — no adversarial-tagged targets are registered. The
+      error names the ``ADVERSARIAL_CHAT_*`` env vars to set and the
+      ``TargetInitializer`` dependency.
+    * ``ValueError`` — any name in ``target_names`` does not match a
+      discovered adversarial-tagged target. The error lists discovered names.
+
+    Prerequisites: ``TargetInitializer`` must have run first so adversarial
+    env-driven targets are present in ``TargetRegistry``. Registering the
+    base scenario-technique catalog (``ScenarioTechniqueInitializer`` or an
+    equivalent caller of ``register_scenario_techniques``) is also expected
+    if users will select non-benchmark strategies in the same session;
+    ``BenchmarkInitializer`` itself only registers the fanned variants.
+    Per-name idempotent via ``AttackTechniqueRegistry.register_from_specs``:
+    running the initializer twice with the same registry state is a no-op.
+    """
+
+    @property
+    def supported_parameters(self) -> list[Parameter]:
+        """Declare the optional ``target_names`` narrowing parameter."""
+        return [
+            Parameter(
+                name="target_names",
+                description=(
+                    "Optional list of adversarial target registry names to narrow benchmark fan-out. "
+                    'When omitted, every target matching TagQuery.all("adversarial") is used.'
+                ),
+                default=None,
+                param_type=list[str],
+            ),
+        ]
+
+    async def initialize_async(self) -> None:
+        """
+        Discover adversarial targets and register fanned specs into the technique registry.
+
+        Raises:
+            ValueError: If no adversarial-tagged targets are registered in
+                ``TargetRegistry``, or if ``self.params['target_names']``
+                contains a name not in the discovered set.
+        """
+        target_registry = TargetRegistry.get_registry_singleton()
+        discovered_entries = target_registry.get_by_tag_query(query=DEFAULT_ADVERSARIAL_TAG_QUERY)
+        if not discovered_entries:
+            raise ValueError(
+                "BenchmarkInitializer: no adversarial-tagged targets registered in TargetRegistry. "
+                "Set ADVERSARIAL_CHAT_* env vars (see .env_example) and ensure TargetInitializer runs "
+                "before BenchmarkInitializer (e.g. via .pyrit_conf initializer ordering)."
+            )
+
+        selected_entries = self._narrow_by_target_names(discovered_entries=discovered_entries)
+
+        fanned_specs = self._build_fanned_specs(target_entries=selected_entries)
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        attack_registry.register_from_specs(fanned_specs)
+
+        logger.info(
+            "BenchmarkInitializer: registered %d fanned spec(s) across %d adversarial target(s): %s",
+            len(fanned_specs),
+            len(selected_entries),
+            ", ".join(entry.name for entry in selected_entries),
+        )
+
+    def _narrow_by_target_names(self, *, discovered_entries: list) -> list:
+        """
+        Filter ``discovered_entries`` to the names in ``self.params['target_names']``, if set.
+
+        Args:
+            discovered_entries: The full set of adversarial-tagged registry entries.
+
+        Returns:
+            list: ``discovered_entries`` unchanged when no ``target_names`` param
+            is set, otherwise the subset whose ``name`` is in the requested set.
+
+        Raises:
+            ValueError: If any name in ``self.params['target_names']`` is not
+                present in ``discovered_entries``.
+        """
+        target_names_param = self.params.get("target_names")
+        if not target_names_param:
+            return discovered_entries
+
+        requested = set(target_names_param)
+        discovered_names = {entry.name for entry in discovered_entries}
+        unknown = requested - discovered_names
+        if unknown:
+            raise ValueError(
+                f"BenchmarkInitializer: unknown target_names {sorted(unknown)}. "
+                f"Discovered adversarial targets: {sorted(discovered_names)}."
+            )
+        return [entry for entry in discovered_entries if entry.name in requested]
+
+    def _build_fanned_specs(self, *, target_entries: list) -> list[AttackTechniqueSpec]:
+        """
+        Build fanned ``AttackTechniqueSpec``s for every (adversarial-capable technique, target) pair.
+
+        Adversarial-capability is determined by ``_spec_needs_adversarial``
+        (re-used from ``scenario_techniques``): a spec needs an adversarial
+        chat target when its attack class accepts ``attack_adversarial_config``
+        or its ``seed_technique`` has a simulated conversation. Non-adversarial
+        techniques (e.g. ``prompt_sending``, ``role_play``) are skipped — the
+        benchmark holds the objective target constant and varies the
+        adversarial chat helper across runs.
+
+        Args:
+            target_entries: The adversarial-tagged registry entries to fan over.
+
+        Returns:
+            list[AttackTechniqueSpec]: One fanned spec per (adversarial-capable
+            technique, target entry) pair, with the live target bound onto
+            ``adversarial_chat`` and benchmark-specific strategy tags appended.
+        """
+        fanned: list[AttackTechniqueSpec] = []
+        for source_spec in SCENARIO_TECHNIQUES:
+            if not _spec_needs_adversarial(source_spec):
+                continue
+            fanned.extend(
+                dataclasses.replace(
+                    source_spec,
+                    name=f"{source_spec.name}__{entry.name}",
+                    adversarial_chat=entry.instance,
+                    adversarial_chat_key=None,
+                    strategy_tags=[*source_spec.strategy_tags, "benchmark_fanout", f"model:{entry.name}"],
+                )
+                for entry in target_entries
+            )
+        return fanned

tests/unit/setup/test_benchmark_initializer.py

@@ -0,0 +1,181 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Tests for BenchmarkInitializer."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from pyrit.prompt_target import PromptTarget
+from pyrit.registry import TargetRegistry
+from pyrit.registry.object_registries.attack_technique_registry import AttackTechniqueRegistry
+from pyrit.setup.initializers import BenchmarkInitializer
+from pyrit.setup.initializers.benchmark import DEFAULT_ADVERSARIAL_TAG_QUERY
+from pyrit.setup.initializers.components.targets import TargetInitializerTags
+
+
+@pytest.fixture(autouse=True)
+def reset_registries():
+    """Reset technique and target registries between tests."""
+    AttackTechniqueRegistry.reset_instance()
+    TargetRegistry.reset_instance()
+    yield
+    AttackTechniqueRegistry.reset_instance()
+    TargetRegistry.reset_instance()
+
+
+def _register_adversarial_target(*, name: str) -> PromptTarget:
+    """Register a mock adversarial-tagged target and return the instance."""
+    target = MagicMock(spec=PromptTarget)
+    target.capabilities.includes.return_value = True
+    registry = TargetRegistry.get_registry_singleton()
+    registry.register_instance(target, name=name, tags=[TargetInitializerTags.ADVERSARIAL.value])
+    return target
+
+
+class TestBenchmarkInitializerBasic:
+    """Class metadata tests."""
+
+    def test_can_be_created(self):
+        init = BenchmarkInitializer()
+        assert init is not None
+
+    def test_required_env_vars_is_empty(self):
+        """Initializer takes no required env vars; discovery happens via TargetRegistry."""
+        init = BenchmarkInitializer()
+        assert init.required_env_vars == []
+
+    def test_supported_parameters_declares_target_names(self):
+        init = BenchmarkInitializer()
+        names = [p.name for p in init.supported_parameters]
+        assert "target_names" in names
+
+    def test_default_adversarial_tag_query_matches_adversarial_only(self):
+        """The default discovery query is exactly ``TagQuery.all("adversarial")``."""
+        assert DEFAULT_ADVERSARIAL_TAG_QUERY.matches({"adversarial"})
+        assert not DEFAULT_ADVERSARIAL_TAG_QUERY.matches({"default"})
+        assert not DEFAULT_ADVERSARIAL_TAG_QUERY.matches(set())
+
+
+class TestBenchmarkInitializerFanOut:
+    """Tests for the fan-out registration behavior."""
+
+    async def test_fans_out_one_spec_per_target_per_adversarial_technique(self):
+        """N targets * M adversarial-capable techniques = N*M fanned specs in the attack registry."""
+        _register_adversarial_target(name="adv_a")
+        _register_adversarial_target(name="adv_b")
+
+        init = BenchmarkInitializer()
+        await init.initialize_async()
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        fanned = attack_registry.get_by_tag(tag="benchmark_fanout")
+        assert len(fanned) > 0
+        assert len(fanned) % 2 == 0, "Expected an even count: every adversarial technique fanned across both targets"
+
+        fanned_names = {entry.name for entry in fanned}
+        for name in fanned_names:
+            assert "__" in name, f"Fanned spec name '{name}' missing '__' separator"
+
+    async def test_fanned_spec_names_use_source_double_underscore_target(self):
+        """Spec naming contract: ``f'{source_spec.name}__{target_name}'``."""
+        _register_adversarial_target(name="adv_single")
+
+        init = BenchmarkInitializer()
+        await init.initialize_async()
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        fanned = attack_registry.get_by_tag(tag="model:adv_single")
+        assert len(fanned) > 0
+        for entry in fanned:
+            assert entry.name.endswith("__adv_single")
+            source_name = entry.name.split("__", 1)[0]
+            assert source_name and "__" not in source_name
+
+    async def test_fanned_specs_carry_benchmark_and_model_tags(self):
+        """Each fanned spec is tagged ``benchmark_fanout`` plus ``f'model:{name}'``."""
+        _register_adversarial_target(name="adv_a")
+        _register_adversarial_target(name="adv_b")
+
+        init = BenchmarkInitializer()
+        await init.initialize_async()
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        for entry in attack_registry.get_by_tag(tag="benchmark_fanout"):
+            assert "benchmark_fanout" in entry.tags
+            model_tags = [tag for tag in entry.tags if tag.startswith("model:")]
+            assert len(model_tags) == 1, f"Expected exactly one model:* tag on {entry.name}, got {model_tags}"
+            assert model_tags[0] in ("model:adv_a", "model:adv_b")
+
+    async def test_registration_is_idempotent_across_re_init(self):
+        """Re-running initialize_async produces the same registry state (per-name idempotent)."""
+        _register_adversarial_target(name="adv_a")
+
+        init = BenchmarkInitializer()
+        await init.initialize_async()
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        first_count = len(attack_registry.get_by_tag(tag="benchmark_fanout"))
+
+        await init.initialize_async()
+        second_count = len(attack_registry.get_by_tag(tag="benchmark_fanout"))
+
+        assert first_count == second_count
+
+
+class TestBenchmarkInitializerTargetNamesNarrowing:
+    """Tests for the optional ``target_names`` parameter."""
+
+    async def test_target_names_narrows_to_subset(self):
+        """When ``target_names`` is set, only those entries are fanned."""
+        _register_adversarial_target(name="adv_a")
+        _register_adversarial_target(name="adv_b")
+        _register_adversarial_target(name="adv_c")
+
+        init = BenchmarkInitializer()
+        init.params = {"target_names": ["adv_a", "adv_c"]}
+        await init.initialize_async()
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        model_b_specs = attack_registry.get_by_tag(tag="model:adv_b")
+        assert model_b_specs == []
+
+        model_a_specs = attack_registry.get_by_tag(tag="model:adv_a")
+        model_c_specs = attack_registry.get_by_tag(tag="model:adv_c")
+        assert len(model_a_specs) > 0
+        assert len(model_c_specs) > 0
+
+    async def test_target_names_unknown_raises_with_discovered_list(self):
+        """Unknown ``target_names`` raise ``ValueError`` naming both the unknowns and the discovered set."""
+        _register_adversarial_target(name="adv_a")
+
+        init = BenchmarkInitializer()
+        init.params = {"target_names": ["nonexistent"]}
+
+        with pytest.raises(ValueError, match=r"nonexistent.*adv_a"):
+            await init.initialize_async()
+
+    async def test_empty_target_names_param_falls_back_to_default_query(self):
+        """An empty ``target_names`` list is treated as "no narrowing" (same as omitting it)."""
+        _register_adversarial_target(name="adv_a")
+
+        init = BenchmarkInitializer()
+        init.params = {"target_names": []}
+        await init.initialize_async()
+
+        attack_registry = AttackTechniqueRegistry.get_registry_singleton()
+        assert len(attack_registry.get_by_tag(tag="model:adv_a")) > 0
+
+
+class TestBenchmarkInitializerErrorMessages:
+    """Tests for the actionable error message on empty discovery."""
+
+    async def test_no_adversarial_targets_raises_with_actionable_message(self):
+        """``ValueError`` must name ``ADVERSARIAL_CHAT_*`` env vars and the ``TargetInitializer`` dependency."""
+        init = BenchmarkInitializer()
+        with pytest.raises(ValueError) as exc_info:
+            await init.initialize_async()
+
+        msg = str(exc_info.value)
+        assert "ADVERSARIAL_CHAT_" in msg
+        assert "TargetInitializer" in msg