FEAT: Run atomic attacks in parallel (better) within a scenario

pyrit/executor/attack/core/attack_executor.py

@@ -120,19 +120,33 @@ class AttackExecutor:
     from seed groups.
     """
 
-    def __init__(self, *, max_concurrency: int = 1) -> None:
+    def __init__(self, *, max_concurrency: int = 1, semaphore: Optional[asyncio.Semaphore] = None) -> None:
         """
         Initialize the attack executor with configurable concurrency control.
 
         Args:
             max_concurrency: Maximum number of concurrent attack executions (default: 1).
+                Ignored when ``semaphore`` is provided.
+            semaphore: Optional externally-owned ``asyncio.Semaphore`` used to gate
+                concurrent objective execution. When provided, all concurrency control
+                (both seed-group parameter building and attack execution) is delegated
+                to this semaphore, allowing a parent (e.g., a Scenario) to share a
+                single budget across many executors. When ``None`` (default), a new
+                semaphore is created from ``max_concurrency``.
 
         Raises:
             ValueError: If max_concurrency is not a positive integer.
         """
         if max_concurrency <= 0:
             raise ValueError(f"max_concurrency must be a positive integer, got {max_concurrency}")
         self._max_concurrency = max_concurrency
+        self._external_semaphore = semaphore
+
+    def _get_semaphore(self) -> asyncio.Semaphore:
+        """Return the externally-supplied semaphore, or a fresh one sized to max_concurrency."""
+        if self._external_semaphore is not None:
+            return self._external_semaphore
+        return asyncio.Semaphore(self._max_concurrency)
 
     async def execute_attack_from_seed_groups_async(
         self,
@@ -193,7 +207,7 @@ async def execute_attack_from_seed_groups_async(
 
         # Build params list using from_seed_group_async with concurrency control
         # This can take time if the SeedSimulatedConversation generation is included
-        semaphore = asyncio.Semaphore(self._max_concurrency)
+        semaphore = self._get_semaphore()
 
         async def build_params(i: int, sg: SeedAttackGroup) -> AttackParameters:
             async with semaphore:
@@ -309,7 +323,7 @@ async def _execute_with_params_list_async(
         Returns:
             AttackExecutorResult with completed results and any incomplete objectives.
         """
-        semaphore = asyncio.Semaphore(self._max_concurrency)
+        semaphore = self._get_semaphore()
 
         async def run_one(index: int, params: AttackParameters) -> AttackStrategyResultT:
             async with semaphore:

pyrit/scenario/core/atomic_attack.py

@@ -13,6 +13,7 @@
 have a common interface for scenarios.
 """
 
+import asyncio
 import logging
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -303,6 +304,7 @@ async def run_async(
         *,
         max_concurrency: int = 1,
         return_partial_on_failure: bool = True,
+        semaphore: asyncio.Semaphore | None = None,
         **attack_params: Any,
     ) -> AttackExecutorResult[AttackResult]:
         """
@@ -321,10 +323,16 @@ async def run_async(
 
         Args:
             max_concurrency (int): Maximum number of concurrent attack executions.
-                Defaults to 1 for sequential execution.
+                Defaults to 1 for sequential execution. Ignored when ``semaphore``
+                is provided.
             return_partial_on_failure (bool): If True, returns partial results even when
                 some objectives don't complete execution. If False, raises an exception on
                 any execution failure. Defaults to True.
+            semaphore (asyncio.Semaphore | None): Optional externally-owned semaphore
+                used to gate objective execution. Allows a parent (e.g., a Scenario
+                running multiple atomic attacks in parallel) to share a single
+                concurrency budget across all of them. When provided, takes precedence
+                over ``max_concurrency``.
             **attack_params: Additional parameters to pass to the attack strategy.
 
         Returns:
@@ -334,7 +342,7 @@ async def run_async(
         Raises:
             ValueError: If the attack execution fails completely and return_partial_on_failure=False.
         """
-        executor = AttackExecutor(max_concurrency=max_concurrency)
+        executor = AttackExecutor(max_concurrency=max_concurrency, semaphore=semaphore)
 
         logger.info(
             f"Starting atomic attack execution with {len(self._seed_groups)} seed groups "