feat(redteam): add Crescendo multi-turn attack strategy

src/strands_evals/experimental/redteam/__init__.py

@@ -3,21 +3,32 @@
 from .experiment import RedTeamExperiment
 from .generators import AdversarialCaseGenerator, TargetSpec
 from .report import AttackResult, GroupedSummary, RedTeamReport
-from .strategies import AttackStrategy, PromptStrategy
+from .strategies import (
+    AttackRunResult,
+    AttackStrategy,
+    CrescendoStrategy,
+    PromptStrategy,
+    TargetCheckpoint,
+    TargetSession,
+)
 from .types import RISK_CATEGORIES, AttackGoal, RedTeamConfig
 
 __all__ = [
     "RISK_CATEGORIES",
     "AdversarialCaseGenerator",
     "AttackGoal",
     "AttackResult",
+    "AttackRunResult",
     "AttackStrategy",
     "AttackSuccessEvaluator",
+    "CrescendoStrategy",
     "GroupedSummary",
     "PromptStrategy",
     "RedTeamCase",
     "RedTeamConfig",
     "RedTeamExperiment",
     "RedTeamReport",
+    "TargetCheckpoint",
+    "TargetSession",
     "TargetSpec",
 ]

src/strands_evals/experimental/redteam/case.py

@@ -15,10 +15,7 @@ class RedTeamCase(Case[InputT, OutputT]):
 
     @model_validator(mode="after")
     def _sync_metadata_from_config(self) -> Self:
-        dump = {
-            **self.config.attack_goal.model_dump(),
-            "strategy": self.config.strategy,
-        }
+        dump = dict(self.config.attack_goal.model_dump())
         if self.metadata is None:
             self.metadata = dump
         else:

src/strands_evals/experimental/redteam/experiment.py

@@ -17,24 +17,29 @@
 from ...types import InputT, OutputT
 from .evaluators import AttackSuccessEvaluator
 from .report import RedTeamReport
+from .strategies import AttackStrategy
+from .strategies.target_session import TargetSession
 from .task import _build_attacker_task
 
 
 class RedTeamExperiment(Experiment[InputT, OutputT]):
     """Experiment specialized for red team evaluation.
 
-    When ``target`` is supplied, ``run_evaluations()`` builds a default
-    multi-turn attacker task internally; pass an explicit ``task`` to
-    customize. Returns a :class:`RedTeamReport`.
+    Holds the attack strategies and runs the case × strategy cross-product:
+    every case is attacked by every strategy. When ``agent`` is supplied,
+    ``run_evaluations()`` builds a default multi-turn attacker task internally;
+    pass an explicit ``task`` to customize. Returns a :class:`RedTeamReport`.
 
     Example:
         ```python
         from strands_evals.experimental.redteam import (
-            AdversarialCaseGenerator, RedTeamExperiment,
+            AdversarialCaseGenerator, RedTeamExperiment, CrescendoStrategy,
         )
 
-        cases = AdversarialCaseGenerator(model=model).generate_cases(target=agent)
-        experiment = RedTeamExperiment(cases=cases, target=agent, max_turns=10)
+        cases = AdversarialCaseGenerator(model=model).generate_cases(agent=agent)
+        experiment = RedTeamExperiment(
+            cases=cases, agent=agent, attack_strategies=[CrescendoStrategy(max_turns=10)]
+        )
         report = experiment.run_evaluations()
         report.display()
         ```
@@ -44,25 +49,69 @@ def __init__(
         self,
         cases: list[Case[InputT, OutputT]] | None = None,
         *,
-        target: Agent | Callable[[str], Any] | None = None,
+        agent: Agent | TargetSession | None = None,
+        attack_strategies: list[AttackStrategy] | None = None,
         evaluators: list[Evaluator[InputT, OutputT]] | None = None,
-        max_turns: int = 10,
         model: Model | str | None = None,
     ):
         super().__init__(
             cases=cases,
             evaluators=evaluators or [AttackSuccessEvaluator(model=model)],
         )
-        self._target = target
-        self._max_turns = max_turns
+        self._agent = agent
+        self._attack_strategies = attack_strategies or []
+        self._by_label = self._build_by_label(self._attack_strategies)
         self._model = model
+        # case name -> strategy run metadata (turns_used, backtracks, ...); the
+        # default task records into this and we join it onto the report, since the
+        # base Experiment doesn't carry task-returned metadata into EvaluationData.
+        self._run_meta: dict[str, dict[str, Any]] = {}
+
+    @staticmethod
+    def _build_by_label(strategies: list[AttackStrategy]) -> dict[str, AttackStrategy]:
+        by_label: dict[str, AttackStrategy] = {}
+        for strategy in strategies:
+            if strategy.label in by_label:
+                raise ValueError(
+                    f"Duplicate strategy label {strategy.label!r}. "
+                    "Pass distinct label= values to compare same-type strategies."
+                )
+            by_label[strategy.label] = strategy
+        return by_label
+
+    def _expand_cross_product(self) -> list[Case[InputT, OutputT]]:
+        """Expand held cases into the case × strategy cross-product.
+
+        Each work item is a copy of the case tagged with one strategy's label in
+        ``metadata["strategy"]`` and a unique name ``"{case}__{label}"`` (so the
+        evaluation_data_store cache keys stay unique). The base worker still sees
+        a plain case queue.
+
+        Pure: returns a new list and never mutates ``self._cases``, so reusing
+        the experiment across runs does not re-expand an already-expanded list.
+
+        Returns:
+            The expanded case list, or the held cases unchanged when no
+            strategies were supplied.
+        """
+        if not self._attack_strategies:
+            return list(self._cases)
+        expanded: list[Case[InputT, OutputT]] = []
+        for case in self._cases:
+            for strategy in self._attack_strategies:
+                item = case.model_copy(deep=True)
+                item.name = f"{case.name}__{strategy.label}"
+                metadata = dict(item.metadata or {})
+                metadata["strategy"] = strategy.label
+                item.metadata = metadata
+                expanded.append(item)
+        return expanded
 
     def run_evaluations(  # type: ignore[override]
         self,
         task: Callable[[Case[InputT, OutputT]], Any] | None = None,
         evaluation_data_store: EvaluationDataStore | None = None,
     ) -> RedTeamReport:
-        task = task or self._default_task()
         if inspect.iscoroutinefunction(task):
             raise ValueError("Async task is not supported. Please use run_evaluations_async instead.")
         return asyncio.run(self.run_evaluations_async(task, max_workers=1, evaluation_data_store=evaluation_data_store))
@@ -73,24 +122,45 @@ async def run_evaluations_async(  # type: ignore[override]
         max_workers: int = 1,
         evaluation_data_store: EvaluationDataStore | None = None,
     ) -> RedTeamReport:
-        # max_workers=1: parallel runs would interleave on the shared target Agent.
-        task = task or self._default_task()
-        report = await super().run_evaluations_async(
-            task, max_workers=max_workers, evaluation_data_store=evaluation_data_store
-        )
-        return RedTeamReport.from_evaluation_report(report)
+        # Parallel workers would interleave on the shared target Agent and on each
+        # strategy instance's per-case state, so red team runs are strictly sequential.
+        if max_workers != 1:
+            raise ValueError("RedTeamExperiment requires max_workers=1 (shared target Agent and strategy state).")
+        self._run_meta.clear()
+        if task is None:
+            task = self._default_task()
+        # Swap self._cases to the expanded cross-product only for the duration of the
+        # base run, then restore it, so the experiment can be re-run without re-expanding
+        # (validated by test_run_evaluations_twice_is_idempotent). This temporary mutation
+        # is safe ONLY because max_workers=1 is enforced above: no other worker or
+        # event-loop task reads self._cases concurrently. Passing the expanded list down
+        # without mutating self is cleaner and is part of the standalone (composition)
+        # refactor tracked in the fast-follow plan; the base API doesn't accept an
+        # explicit case list today.
+        original_cases = self._cases
+        self._cases = self._expand_cross_product()
+        try:
+            # base now returns a single flattened EvaluationReport (#241), each case row
+            # tagged with its evaluator; RedTeamReport wraps that and joins run_meta.
+            report = await super().run_evaluations_async(
+                task, max_workers=max_workers, evaluation_data_store=evaluation_data_store
+            )
+        finally:
+            self._cases = original_cases
+        return RedTeamReport.from_evaluation_report(report, run_meta=self._run_meta)
 
     def _default_task(self) -> Callable[[Case[InputT, OutputT]], Any]:
-        if self._target is None:
+        if self._agent is None:
             raise ValueError(
-                "RedTeamExperiment requires either `target` at construction "
+                "RedTeamExperiment requires either `agent` at construction "
                 "or an explicit `task` argument to run_evaluations()."
             )
         return cast(
             Callable[[Case[InputT, OutputT]], Any],
             _build_attacker_task(
-                target=self._target,
-                max_turns=self._max_turns,
+                self._agent,
+                self._by_label,
                 model=self._model,
+                run_meta=self._run_meta,
             ),
         )

src/strands_evals/experimental/redteam/generators/adversarial.py

@@ -12,12 +12,6 @@
 from strands.models.model import Model
 
 from ..case import RedTeamCase
-from ..strategies import (
-    BUILTIN_STRATEGIES,
-    DEFAULT_STRATEGY,
-    AttackStrategy,
-    resolve_strategy,
-)
 from ..types import DEFAULT_SEVERITY, RISK_CATEGORIES, AttackGoal, RedTeamConfig
 from .prompt_templates import get_template as _get_prompt_template
 
@@ -47,12 +41,6 @@ class _RiskCategorySelection(BaseModel):
     categories: list[str] = Field(description="Selected risk category keys relevant to the target")
 
 
-def _resolve_strategies(strategies: list[AttackStrategy | str] | None) -> list[AttackStrategy]:
-    if not strategies:
-        return [BUILTIN_STRATEGIES[DEFAULT_STRATEGY]]
-    return [resolve_strategy(s) for s in strategies]
-
-
 _REQUIRED_TARGET_KEYS = ("system_prompt", "tools")
 
 
@@ -71,7 +59,7 @@ def _extract_tool_info(agent: Agent) -> dict:
                 }
             )
     except (AttributeError, KeyError, TypeError) as e:
-        logger.warning("Failed to extract tools from agent: %s", e)
+        logger.warning("error=<%s> | failed to extract tools from agent", e)
 
     return {
         "system_prompt": agent.system_prompt or "",
@@ -95,7 +83,7 @@ class AdversarialCaseGenerator:
     Example:
         ```python
         cases = AdversarialCaseGenerator(model=model).generate_cases(
-            target=agent,
+            agent=agent,
             risk_categories=["guideline_bypass", "data_exfiltration"],
             num_cases=3,
         )
@@ -113,37 +101,33 @@ def __init__(
     def generate_cases(
         self,
         *,
-        target: Agent | TargetSpec,
+        agent: Agent | TargetSpec,
         risk_categories: list[str] | None = None,
         num_cases: int = 5,
-        attack_strategies: list[AttackStrategy | str] | None = None,
     ) -> list[RedTeamCase]:
-        """Generate cases without building an experiment. Use with TargetSpec or for case reuse."""
+        """Generate strategy-agnostic cases for an agent. Use with TargetSpec or for case reuse."""
         return asyncio.run(
             self.generate_cases_async(
-                target=target,
+                agent=agent,
                 risk_categories=risk_categories,
                 num_cases=num_cases,
-                attack_strategies=attack_strategies,
             )
         )
 
     async def generate_cases_async(
         self,
         *,
-        target: Agent | TargetSpec,
+        agent: Agent | TargetSpec,
         risk_categories: list[str] | None = None,
         num_cases: int = 5,
-        attack_strategies: list[AttackStrategy | str] | None = None,
     ) -> list[RedTeamCase]:
         """Async variant of :meth:`generate_cases`."""
-        target_info = _coerce_target(target)
+        target_info = _coerce_target(agent)
         resolved_categories = risk_categories or await self._infer_risk_categories(target_info)
         return await self._generate_cases(
             target_info=target_info,
             risk_categories=resolved_categories,
             num_cases=num_cases,
-            attack_strategies=attack_strategies,
         )
 
     async def _generate_cases(
@@ -152,16 +136,13 @@ async def _generate_cases(
         target_info: dict,
         risk_categories: list[str],
         num_cases: int = 5,
-        attack_strategies: list[AttackStrategy | str] | None = None,
     ) -> list[RedTeamCase]:
         for risk_category in risk_categories:
             if risk_category not in RISK_CATEGORIES:
                 raise ValueError(
                     f"Unknown risk category: '{risk_category}'. Available categories: {list(RISK_CATEGORIES)}"
                 )
 
-        resolved_strategies = _resolve_strategies(attack_strategies)
-
         cases: list[RedTeamCase] = []
         for risk_category in risk_categories:
             generated = await self._generate_cases_for_category(
@@ -172,32 +153,23 @@ async def _generate_cases(
 
             severity = DEFAULT_SEVERITY.get(risk_category, "medium")
             for i, attack in enumerate(generated):
-                for strategy in resolved_strategies:
-                    template = strategy.system_prompt_template
-                    if template is None:
-                        raise NotImplementedError(
-                            f"Strategy {type(strategy).__name__!r} does not expose system_prompt_template. "
-                            "Only system-prompt-based strategies are currently supported."
-                        )
-                    config = RedTeamConfig(
-                        attack_goal=AttackGoal(
-                            risk_category=risk_category,
-                            actor_goal=attack.actor_goal,
-                            context=attack.target_context,
-                            severity=severity,
-                            success_criteria=attack.success_criteria,
-                        ),
-                        traits=attack.traits,
-                        system_prompt_template=template,
-                        strategy=strategy.name,
-                    )
-                    cases.append(
-                        RedTeamCase(
-                            name=f"{risk_category}_{i}__{strategy.name}",
-                            input=attack.opening_message,
-                            config=config,
-                        )
+                config = RedTeamConfig(
+                    attack_goal=AttackGoal(
+                        risk_category=risk_category,
+                        actor_goal=attack.actor_goal,
+                        context=attack.target_context,
+                        severity=severity,
+                        success_criteria=attack.success_criteria,
+                    ),
+                    traits=attack.traits,
+                )
+                cases.append(
+                    RedTeamCase(
+                        name=f"{risk_category}_{i}",
+                        input=attack.opening_message,
+                        config=config,
                     )
+                )
 
         return cases
 
@@ -213,11 +185,11 @@ async def _infer_risk_categories(self, target_info: dict) -> list[str]:
         response = await agent.invoke_async(prompt, structured_output_model=_RiskCategorySelection)
         result = cast(_RiskCategorySelection, response.structured_output)
         if result is None:
-            logger.warning("Risk-category inference returned no structured output; using all categories.")
+            logger.warning("reason=<no_structured_output> | risk-category inference empty | using all")
             return list(RISK_CATEGORIES.keys())
         valid = [c for c in result.categories if c in RISK_CATEGORIES]
         if not valid:
-            logger.warning("No recognized risk categories inferred (got %s); using all.", result.categories)
+            logger.warning("got=<%s> | no recognized risk categories inferred | using all", result.categories)
             return list(RISK_CATEGORIES.keys())
         return valid