feat(redteam): public as_target_session and default run_attack max_turns Part 3

AGENTS.md

@@ -106,16 +106,17 @@ strands-evals/
 │   │       ├── README.md                     # Module quick-start + walkthrough
 │   │       ├── case.py                       # RedTeamCase + RedTeamConfig
 │   │       ├── experiment.py                 # RedTeamExperiment (case × strategy cross-product)
-│   │       ├── task.py                       # _build_attacker_task; MAX_ALLOWED_TURNS = 50 (hard cap)
+│   │       ├── task.py                       # _build_attacker_task
 │   │       ├── utils.py                      # _put_model_field (used by to_dict)
 │   │       ├── report.py                     # RedTeamReport / AttackResult / GroupedSummary
 │   │       ├── evaluators/                   # AttackSuccessEvaluator
 │   │       ├── generators/                   # AdversarialCaseGenerator + TargetSpec
 │   │       ├── strategies/                   # AttackStrategy base + per-strategy subpackages
-│   │       │   ├── base.py                   # AttackStrategy ABC + AttackRunResult
+│   │       │   ├── base.py                   # AttackStrategy ABC + AttackRunResult; MAX_ALLOWED_TURNS = 50 (hard cap)
 │   │       │   ├── _common.py                # Shared helpers
 │   │       │   ├── target_session.py         # TargetSession Protocol + StrandsAgentSession,
-│   │       │   │                             # StrandsMultiAgentSession, TargetCheckpoint, ToolUseEntry
+│   │       │   │                             # StrandsMultiAgentSession, TargetCheckpoint, ToolUseEntry,
+│   │       │   │                             # as_target_session
 │   │       │   ├── bad_likert_judge/         # BadLikertJudgeStrategy
 │   │       │   ├── crescendo/                # CrescendoStrategy + crescendo_v0 prompt
 │   │       │   ├── goat/                     # GoatStrategy + goat_v0 prompt

src/strands_evals/experimental/redteam/README.md

@@ -84,7 +84,7 @@ contract and talk to the target only through `target_session.invoke(...)`.
 | `BadLikertJudgeStrategy` | Casts the target as a harmfulness-rating judge, elicits a top-score example | no | [Unit 42](https://unit42.paloaltonetworks.com/multi-turn-technique-jailbreaks-llms/) |
 | `SequentialBreakStrategy` | Hides the harmful request among benign siblings in one narrative scaffold | no | [arXiv:2411.06426](https://arxiv.org/abs/2411.06426) |
 
-Each strategy accepts a `max_turns=` kwarg (its own per-attack ceiling); the task runner additionally caps every strategy at a hard `MAX_ALLOWED_TURNS = 50` (`task.py`), so a strategy configured higher will be silently clamped. Pass `label="..."` to compare two instances of the same strategy in one experiment (e.g. `CrescendoStrategy(max_turns=5, label="cresc_short")`); duplicate labels raise at construction.
+Each strategy accepts a `max_turns=` kwarg (its own per-attack ceiling); `run_attack` additionally caps every strategy at a hard `MAX_ALLOWED_TURNS = 50` (`strategies/base.py`, exported from `strands_evals.experimental.redteam`), so a strategy configured higher will be silently clamped. A custom task can omit `run_attack`'s `max_turns`, which defaults to that ceiling; a larger value is clamped to it. Pass `label="..."` to compare two instances of the same strategy in one experiment (e.g. `CrescendoStrategy(max_turns=5, label="cresc_short")`); duplicate labels raise at construction.
 
 `PromptStrategy` (a no-attacker-LLM, system-prompt-template strategy) and the `BUILTIN_STRATEGIES` registry are the extension points for adding new template-driven strategies without subclassing — see `strategies/prompt_strategy/`. The full set of exported symbols (experiment, cases, evaluator, target sessions) is the `__all__` of `strands_evals.experimental.redteam`.
 

src/strands_evals/experimental/redteam/__init__.py

@@ -4,6 +4,7 @@
 from .generators import AdversarialCaseGenerator, TargetSpec
 from .report import AttackResult, GroupedSummary, RedTeamReport
 from .strategies import (
+    MAX_ALLOWED_TURNS,
     AttackRunResult,
     AttackStrategy,
     BadLikertJudgeStrategy,
@@ -16,10 +17,12 @@
     StrandsMultiAgentSession,
     TargetCheckpoint,
     TargetSession,
+    as_target_session,
 )
 from .types import RISK_CATEGORIES, AttackGoal, RedTeamConfig
 
 __all__ = [
+    "MAX_ALLOWED_TURNS",
     "RISK_CATEGORIES",
     "AdversarialCaseGenerator",
     "AttackGoal",
@@ -43,4 +46,5 @@
     "TargetCheckpoint",
     "TargetSession",
     "TargetSpec",
+    "as_target_session",
 ]

src/strands_evals/experimental/redteam/strategies/__init__.py

@@ -1,5 +1,5 @@
 from .bad_likert_judge import BadLikertJudgeStrategy
-from .base import AttackRunResult, AttackStrategy
+from .base import MAX_ALLOWED_TURNS, AttackRunResult, AttackStrategy
 from .crescendo import CrescendoStrategy
 from .goat import GoatStrategy
 from .pair import PairStrategy
@@ -12,6 +12,7 @@
     TargetCheckpoint,
     TargetSession,
     ToolUseEntry,
+    as_target_session,
 )
 
 # Ready-made strategy instances users can pass to RedTeamExperiment(attack_strategies=[...]).
@@ -24,6 +25,7 @@
 
 __all__ = [
     "BUILTIN_STRATEGIES",
+    "MAX_ALLOWED_TURNS",
     "AttackRunResult",
     "AttackStrategy",
     "BadLikertJudgeStrategy",
@@ -37,4 +39,5 @@
     "TargetCheckpoint",
     "TargetSession",
     "ToolUseEntry",
+    "as_target_session",
 ]

src/strands_evals/experimental/redteam/strategies/bad_likert_judge/__init__.py

@@ -108,11 +108,12 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
         goal = case.config.attack_goal
+        max_turns = self._turn_cap(max_turns)
 
         # The payload appears at turn 2; a smaller budget would send the role-set turn and never elicit,
         # reporting succeeded=False indistinguishably from "defended".

src/strands_evals/experimental/redteam/strategies/base.py

@@ -4,14 +4,18 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from strands.models.model import Model
 
 if TYPE_CHECKING:
     from ..case import RedTeamCase
     from .target_session import TargetSession
 
+# Hard ceiling on turns per attack. `run_attack` defaults `max_turns` to it and clamps any larger value; a
+# strategy's own ctor `max_turns` is the per-attack budget and wins when smaller.
+MAX_ALLOWED_TURNS = 50
+
 
 @dataclass
 class AttackRunResult:
@@ -43,6 +47,8 @@ class AttackStrategy(ABC):
     `snapshot`/`restore` if they backtrack).
     """
 
+    MAX_ALLOWED_TURNS: ClassVar[int] = MAX_ALLOWED_TURNS
+
     def __init__(self, *, label: str | None = None) -> None:
         """Initialize the strategy.
 
@@ -69,7 +75,7 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
@@ -86,13 +92,23 @@ def run_attack(
             case: The red team case carrying the attack goal.
             target_session: Session for invoking the target, snapshotting/restoring its state, and reading
                 its tool-use `trace`.
-            max_turns: Experiment-level ceiling. A strategy with its own `max_turns` should run
-                `min(self._max_turns, max_turns)`.
+            max_turns: Experiment-level ceiling; defaults to, and is clamped to, `MAX_ALLOWED_TURNS`.
+                Implementations resolve it with `self._turn_cap(max_turns, own=...)`.
             model: Model for any strategy-internal LLM calls; ctor model takes precedence.
             **kwargs: Reserved for forward compatibility.
         """
         ...
 
+    def _turn_cap(self, max_turns: int | None, own: int | None = None) -> int:
+        """Resolve the turn budget for one attack.
+
+        Args:
+            max_turns: The `run_attack` ceiling; `None` means `MAX_ALLOWED_TURNS`, larger values are clamped to it.
+            own: The strategy's own ctor budget, if any; wins when smaller.
+        """
+        cap = self.MAX_ALLOWED_TURNS if max_turns is None else min(max_turns, self.MAX_ALLOWED_TURNS)
+        return cap if own is None else min(own, cap)
+
     def reset(self) -> None:  # noqa: B027
         """Per-case reset hook; no-op by default. Override only if `self` holds mutable state."""
 

src/strands_evals/experimental/redteam/strategies/crescendo/__init__.py

@@ -161,7 +161,7 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
@@ -171,7 +171,7 @@ def run_attack(
         # Judge factories: each scoring/refusal call gets a fresh judge with no carryover.
         make_judge = partial(self._build_judge, resolved_model)
         make_refusal_judge = partial(self._build_refusal_judge, resolved_model)
-        turn_cap = min(self._max_turns, max_turns)
+        turn_cap = self._turn_cap(max_turns, own=self._max_turns)
 
         conversation: list[dict[str, Any]] = []
         pruned_branches: list[dict[str, Any]] = []

src/strands_evals/experimental/redteam/strategies/goat/__init__.py

@@ -145,15 +145,15 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
         goal = case.config.attack_goal
         resolved_model = self._model or model
         attacker = self._build_attacker(goal, resolved_model)
         make_judge = partial(self._build_judge, resolved_model)
-        turn_cap = min(self._max_turns, max_turns)
+        turn_cap = self._turn_cap(max_turns, own=self._max_turns)
 
         conversation: list[dict[str, Any]] = []
         attacks_used: list[str] = []

src/strands_evals/experimental/redteam/strategies/pair/__init__.py

@@ -145,15 +145,15 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
         goal = case.config.attack_goal
         resolved_model = self._model or model
         attacker = self._build_attacker(goal, resolved_model)
         make_judge = partial(self._build_judge, resolved_model)
-        turn_cap = min(self._max_turns, max_turns)
+        turn_cap = self._turn_cap(max_turns, own=self._max_turns)
 
         conversation: list[dict[str, Any]] = []
         last_response = ""

src/strands_evals/experimental/redteam/strategies/prompt_strategy/__init__.py

@@ -42,12 +42,12 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
         goal = case.config.attack_goal
-        turn_cap = min(self._max_turns, max_turns)
+        turn_cap = self._turn_cap(max_turns, own=self._max_turns)
         actor_profile = ActorProfile(
             traits=case.config.traits,
             context=goal.context,

src/strands_evals/experimental/redteam/strategies/sequentialbreak/__init__.py

@@ -130,13 +130,13 @@ def run_attack(
         case: RedTeamCase,
         target_session: TargetSession,
         *,
-        max_turns: int,
+        max_turns: int | None = None,
         model: Model | str | None = None,
         **kwargs: Any,
     ) -> AttackRunResult:
         goal = case.config.attack_goal
         make_judge = partial(self._build_judge, self._model or model)
-        turn_cap = min(self._max_turns, max_turns)
+        turn_cap = self._turn_cap(max_turns, own=self._max_turns)
         variants = self._variants[:turn_cap]
 
         conversation: list[dict[str, Any]] = []

src/strands_evals/experimental/redteam/strategies/target_session.py

@@ -341,11 +341,66 @@ def _multi_agent_result_text(result: Any) -> str:
     return str(result)
 
 
+def as_target_session(target: Agent | MultiAgentBase | TargetSession) -> TargetSession:
+    """Wrap a target in the `TargetSession` a strategy's `run_attack` expects.
+
+    Use this in a custom task to turn a freshly built target into a session:
+
+        session = as_target_session(build_agent())
+        result = strategy.run_attack(case, session)  # max_turns defaults to MAX_ALLOWED_TURNS
+
+    Args:
+        target: A `strands.Agent` (wrapped in `StrandsAgentSession`), a `MultiAgentBase` such as a `Graph` or
+            `Swarm` (wrapped in `StrandsMultiAgentSession`), or a ready `TargetSession` (returned as is).
+
+    Returns:
+        A `TargetSession` driving `target`.
+
+    Raises:
+        TypeError: If `target` is none of the above. A custom `TargetSession` must expose
+            `invoke`/`reset`/`snapshot`/`restore` and a `trace: list`.
+    """
+    return _build_session(target)
+
+
+def _build_session(
+    agent: Agent | MultiAgentBase | TargetSession,
+    *,
+    baseline: Any = None,
+) -> TargetSession:
+    """Wrap an `Agent` / `MultiAgentBase`, or pass a `TargetSession` through.
+
+    Args:
+        agent: The target to wrap, or a ready `TargetSession`.
+        baseline: Clean snapshot the wrapped session resets to between cases. Ignored for a passed-in
+            `TargetSession`. Typed `Any` because the two session types use different opaque baseline shapes.
+
+    Raises:
+        TypeError: If `agent` is not an `Agent`, `MultiAgentBase`, or a structural `TargetSession` (must expose
+            `invoke`/`reset`/`snapshot`/`restore` and a `trace: list`).
+    """
+    if isinstance(agent, Agent):
+        return StrandsAgentSession(agent, baseline=baseline)
+    if isinstance(agent, MultiAgentBase):
+        return StrandsMultiAgentSession(agent, baseline=baseline)
+    # Structural check: TargetSession is a Protocol. The `trace: list` check is
+    # load-bearing because the task runner dereferences `.trace` directly.
+    has_methods = all(callable(getattr(agent, method, None)) for method in ("invoke", "reset", "snapshot", "restore"))
+    if has_methods and isinstance(getattr(agent, "trace", None), list):
+        return agent
+    raise TypeError(
+        f"agent must be a strands.Agent, strands.multiagent.MultiAgentBase, or a TargetSession, "
+        f"got {type(agent).__name__!r}; wrap a custom target in a TargetSession so the strategy "
+        "can snapshot/restore its state."
+    )
+
+
 __all__ = [
     "MALFORMED_TOOL_NAME",
     "StrandsAgentSession",
     "StrandsMultiAgentSession",
     "TargetCheckpoint",
     "TargetSession",
     "ToolUseEntry",
+    "as_target_session",
 ]

src/strands_evals/experimental/redteam/task.py

@@ -12,12 +12,11 @@
 
 from .case import RedTeamCase
 from .strategies import AttackStrategy
-from .strategies.target_session import StrandsAgentSession, StrandsMultiAgentSession, TargetSession
+from .strategies.base import MAX_ALLOWED_TURNS as MAX_ALLOWED_TURNS  # re-exported for the pre-move import path
+from .strategies.target_session import StrandsMultiAgentSession, TargetSession, _build_session, as_target_session
 
 logger = logging.getLogger(__name__)
 
-MAX_ALLOWED_TURNS = 50
-
 
 def _build_attacker_task(
     agent: Agent | MultiAgentBase | TargetSession | None,
@@ -31,7 +30,7 @@ def _build_attacker_task(
     """Build a `task(case) -> {"output": conversation, "trajectory": tool_uses}` callable.
 
     Looks up each case's strategy by `metadata["strategy"]` and delegates the multi-turn loop to
-    `strategy.run_attack`, injecting a `TargetSession`. `MAX_ALLOWED_TURNS` is the hard ceiling. Run metadata
+    `strategy.run_attack`, injecting a `TargetSession` and the strategy's `MAX_ALLOWED_TURNS` ceiling. Run metadata
     is recorded into `run_meta` keyed by case name.
 
     Args:
@@ -124,7 +123,7 @@ def task_fn(case: RedTeamCase) -> dict:
 
         # No baseline: each case starts from a freshly built target, and `session.reset()` only
         # needs to clear the per-case trace.
-        session = _build_session(make_target(), baseline=None)
+        session = as_target_session(make_target())
         session.reset()
 
         # CPython dict assignment for a single distinct key is atomic, and case names are unique
@@ -147,7 +146,8 @@ def _run_attack(
     Errors propagate: the base `Experiment` retries throttling, records any other failure as an
     error reason (which the report classifies as errored), and skips caching the failed case.
     """
-    result = strategy.run_attack(case, session, max_turns=MAX_ALLOWED_TURNS, model=model)
+    # Passed explicitly: custom strategies written against the old contract declare `max_turns` without a default.
+    result = strategy.run_attack(case, session, max_turns=strategy.MAX_ALLOWED_TURNS, model=model)
     if run_meta is not None and case.name is not None:
         run_meta[case.name] = {**result.metadata, "pruned_branches": result.pruned_branches}
     return {
@@ -180,38 +180,6 @@ def _resolve_target_source(
     )
 
 
-def _build_session(
-    agent: Agent | MultiAgentBase | TargetSession,
-    *,
-    baseline: Any = None,
-) -> TargetSession:
-    """Wrap an `Agent` / `MultiAgentBase`, or pass a `TargetSession` through.
-
-    Args:
-        agent: The target to wrap, or a ready `TargetSession`.
-        baseline: Clean snapshot the wrapped session resets to between cases. Ignored for a passed-in
-            `TargetSession`. Typed `Any` because the two session types use different opaque baseline shapes.
-
-    Raises:
-        TypeError: If `agent` is not an `Agent`, `MultiAgentBase`, or a structural `TargetSession` (must expose
-            `invoke`/`reset`/`snapshot`/`restore` and a `trace: list`).
-    """
-    if isinstance(agent, Agent):
-        return StrandsAgentSession(agent, baseline=baseline)
-    if isinstance(agent, MultiAgentBase):
-        return StrandsMultiAgentSession(agent, baseline=baseline)
-    # Structural check: TargetSession is a Protocol. The `trace: list` check is
-    # load-bearing because the task runner dereferences `.trace` directly.
-    has_methods = all(callable(getattr(agent, method, None)) for method in ("invoke", "reset", "snapshot", "restore"))
-    if has_methods and isinstance(getattr(agent, "trace", None), list):
-        return agent
-    raise TypeError(
-        f"agent must be a strands.Agent, strands.multiagent.MultiAgentBase, or a TargetSession, "
-        f"got {type(agent).__name__!r}; wrap a custom target in a TargetSession so the strategy "
-        "can snapshot/restore its state."
-    )
-
-
 def _resolve_case_strategy(case: RedTeamCase, by_label: dict[str, AttackStrategy]) -> AttackStrategy:
     """Look up the strategy for `case` from its `metadata["strategy"]` label."""
     metadata = case.metadata or {}