fix(redteam): reset target session between PAIR/SequentialBreak iterations

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

@@ -84,7 +84,8 @@ def run_attack(
 
         Args:
             case: The red team case carrying the attack goal.
-            target_session: Session for talking to the target; use `target_session.invoke(message)`.
+            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)`.
             model: Model for any strategy-internal LLM calls; ctor model takes precedence.

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

@@ -18,6 +18,7 @@
 
 from ...utils import _put_model_field
 from ..base import AttackRunResult, AttackStrategy
+from ..target_session import _single_shot_attempts
 from . import pair_v0
 
 if TYPE_CHECKING:
@@ -156,39 +157,42 @@ def run_attack(
         conversation: list[dict[str, Any]] = []
         last_response = ""
         last_raw_score: int | None = None
-        # PEAK normalized score across turns (a refinement strategy's closest approach is more informative
-        # than the last turn, which may have dipped after a peak).
         best_score: float | None = None
         succeeded = False
         iterations = 0
         target_calls = 0
         parse_failures = 0
 
-        for _ in range(turn_cap):
-            p = gen_refined_prompt(attacker, goal, last_response, last_raw_score)
-            if p is None or not p.strip():
-                parse_failures += 1
-                logger.warning("reason=<empty_prompt> | attacker produced no prompt | ending case early")
-                break
-
-            response = target_session.invoke(p)
-            target_calls += 1
-            if not response or not response.strip():
-                logger.warning("reason=<empty_response> | target returned no text | ending case early")
-                break
-
-            iterations += 1
-            conversation.append({"role": "attacker", "content": p})
-            conversation.append({"role": "target", "content": response})
-            last_response = response
-
-            turn_score, last_raw_score = success_score(make_judge, goal, response)
-            best_score = turn_score if best_score is None else max(best_score, turn_score)
-            logger.debug("iteration=<%s> score=<%.3f> | pair turn scored", iterations, turn_score)
-
-            if turn_score >= self._success_threshold:
-                succeeded = True
-                break
+        # Each P is an independent single-shot: restore the target checkpoint between iterations while
+        # preserving every iteration's tool-use trace.
+        with _single_shot_attempts(target_session) as begin_attempt:
+            for _ in range(turn_cap):
+                p = gen_refined_prompt(attacker, goal, last_response, last_raw_score)
+                if p is None or not p.strip():
+                    parse_failures += 1
+                    logger.warning("reason=<empty_prompt> | attacker produced no prompt | ending case early")
+                    break
+
+                begin_attempt()
+                response = target_session.invoke(p)
+                target_calls += 1
+                if not response or not response.strip():
+                    logger.warning("reason=<empty_response> | target returned no text | ending case early")
+                    break
+
+                iterations += 1
+                conversation.append({"role": "attacker", "content": p})
+                conversation.append({"role": "target", "content": response})
+                last_response = response
+
+                turn_score, last_raw_score = success_score(make_judge, goal, response)
+                # A refinement strategy's closest approach is more informative than a later dip.
+                best_score = turn_score if best_score is None else max(best_score, turn_score)
+                logger.debug("iteration=<%s> score=<%.3f> | pair turn scored", iterations, turn_score)
+
+                if turn_score >= self._success_threshold:
+                    succeeded = True
+                    break
 
         return AttackRunResult(
             conversation=conversation,

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

@@ -5,9 +5,9 @@
 breaching scaffold is model-dependent, so this strategy tries several variants per case and stops at the
 first breach. No attacker LLM; the inline success judge is just an early-stop gate. Append-only.
 
-On a stateful `target_session`, variants 2..N see earlier variants' refusal context, so measured ASR is a
-lower bound whenever more than one variant is tried; best-variant-first (dc_t1) minimizes this.
-`variants_tried` length > 1 flags possible contamination.
+Each scaffold variant is an independent single-shot: the strategy checkpoints `target_session` at entry
+and restores that checkpoint before every later variant, so variant N never sees variant N-1's refusal
+context. Variant ordering therefore does not bias measured ASR.
 """
 
 from __future__ import annotations
@@ -23,6 +23,7 @@
 
 from ...utils import _put_model_field
 from ..base import AttackRunResult, AttackStrategy
+from ..target_session import _single_shot_attempts
 from . import sequentialbreak_v0
 
 if TYPE_CHECKING:
@@ -145,30 +146,34 @@ def run_attack(
         score: float | None = None
         succeeded = False
 
-        for variant in variants:
-            prompt = assemble_scaffold(sequentialbreak_v0.SCAFFOLDS[variant], goal.actor_goal)
-            if not prompt.strip():
-                logger.warning("variant=<%s> | assembled empty prompt | skipping", variant)
-                continue
-
-            response = target_session.invoke(prompt)
-            target_calls += 1
-            if not response.strip():
-                logger.warning("variant=<%s> | empty target response | trying next variant", variant)
-                continue
-
-            variants_tried.append(variant)
-            conversation.append({"role": "attacker", "content": prompt})
-            conversation.append({"role": "target", "content": response})
-
-            rating = success_score(make_judge, goal, response)
-            score = rating if score is None else max(score, rating)
-            logger.debug("variant=<%s> score=<%.3f> | sequentialbreak variant scored", variant, rating)
-
-            if rating >= self._success_threshold:
-                succeeded = True
-                breached_variant = variant
-                break
+        # Each variant is an independent single-shot: restore the target checkpoint between variants while
+        # preserving every variant's tool-use trace.
+        with _single_shot_attempts(target_session) as begin_attempt:
+            for variant in variants:
+                prompt = assemble_scaffold(sequentialbreak_v0.SCAFFOLDS[variant], goal.actor_goal)
+                if not prompt.strip():
+                    logger.warning("variant=<%s> | assembled empty prompt | skipping", variant)
+                    continue
+
+                begin_attempt()
+                response = target_session.invoke(prompt)
+                target_calls += 1
+                if not response.strip():
+                    logger.warning("variant=<%s> | empty target response | trying next variant", variant)
+                    continue
+
+                variants_tried.append(variant)
+                conversation.append({"role": "attacker", "content": prompt})
+                conversation.append({"role": "target", "content": response})
+
+                rating = success_score(make_judge, goal, response)
+                score = rating if score is None else max(score, rating)
+                logger.debug("variant=<%s> score=<%.3f> | sequentialbreak variant scored", variant, rating)
+
+                if rating >= self._success_threshold:
+                    succeeded = True
+                    breached_variant = variant
+                    break
 
         return AttackRunResult(
             conversation=conversation,

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

@@ -74,8 +74,7 @@
 6. What debrief questions help trainees reflect afterwards?
 7. How should the simulation onboard a first-time trainee?"""
 
-# Insertion order is the strategy's try-order; dc_t1 is first as the paper's strongest
-# variant, which also minimizes cross-variant contamination on stateful targets.
+# Insertion order is the strategy's try-order; dc_t1 is first as the paper's strongest variant.
 SCAFFOLDS = {
     "dc_t1": DC_T1,
     "dc_t2": DC_T2,

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

@@ -9,6 +9,8 @@
 
 import copy
 import logging
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import Any, Protocol, TypedDict
 
@@ -93,6 +95,57 @@ def restore(self, checkpoint: TargetCheckpoint) -> None:
         ...
 
 
+@contextmanager
+def _single_shot_attempts(session: TargetSession) -> Iterator[Callable[[], None]]:
+    """Run a sequence of independent single-shot attempts against one `session`.
+
+    PAIR (each refined prompt P) and SequentialBreak (each scaffold variant) treat every attempt as an
+    independent single-shot, so attempt N must start from the same target state attempt 1 saw — not carry
+    attempt N-1's conversation or agent state. This context manager captures the entry state once with
+    `snapshot()` and rolls back to it before each later attempt with `restore()`, which covers the state
+    represented by the session checkpoint, unlike a baseline-less `reset()` that only clears messages.
+
+    Tool-use evidence must survive the rollbacks: the authoritative `AttackSuccessEvaluator` reads
+    `session.trace` to detect tool-call breaches. Each attempt's trace delta is accumulated and, on exit,
+    the session trace is rebuilt as the entry prefix plus every attempt's delta — once each, in order —
+    even if the caller raises mid-loop.
+
+    Args:
+        session: The target session to checkpoint at entry and restore between attempts.
+
+    Yields:
+        A zero-arg `begin_attempt` the caller MUST invoke immediately before each `invoke()`. The first
+        call is a no-op (attempt 1 uses the supplied state); each later call collects the finished
+        attempt's trace delta and restores the checkpoint. Calling it only right before `invoke()` means
+        attempts that end early (parse failure, empty prompt) never trigger a rollback.
+    """
+    checkpoint = session.snapshot()
+    trace_prefix = list(session.trace[: checkpoint.trace_len])
+    attempt_trace: list[ToolUseEntry] = []
+    started = False
+
+    def take_delta() -> list[ToolUseEntry]:
+        # Move (not copy) the delta off the session: a `restore()` that raises after this must not leave
+        # the same entries behind for the `finally` block to collect a second time.
+        delta = list(session.trace[checkpoint.trace_len :])
+        del session.trace[checkpoint.trace_len :]
+        return delta
+
+    def begin_attempt() -> None:
+        nonlocal started
+        if started:
+            attempt_trace.extend(take_delta())
+            session.restore(checkpoint)
+        started = True
+
+    try:
+        yield begin_attempt
+    finally:
+        # Keep the in-flight attempt's evidence even if the caller raised after invoke().
+        attempt_trace.extend(take_delta())
+        session.trace[:] = trace_prefix + attempt_trace
+
+
 class StrandsAgentSession:
     """A `TargetSession` backed by a `strands.Agent`, rewindable via the SDK snapshot API."""
 
@@ -214,7 +267,10 @@ def snapshot(self) -> TargetCheckpoint:
         return TargetCheckpoint(
             agent_snapshot=_MultiAgentSnapshot(
                 agents={path: agent.take_snapshot(preset="session") for path, agent in self._agent_index.items()},
-                orchestrators={path: orch.serialize_state() for path, orch in self._orch_index.items()},
+                # Deep-copy at capture: `serialize_state()` can hand back live orchestrator state (Swarm
+                # returns its `shared_context` dict), and one checkpoint is restored repeatedly here.
+                # Leaf snapshots need no copy -- `take_snapshot` already deep-copies messages and state.
+                orchestrators={path: copy.deepcopy(orch.serialize_state()) for path, orch in self._orch_index.items()},
             ),
             trace_len=len(self.trace),
         )

tests/strands_evals/experimental/redteam/test_multi_agent_session.py

@@ -6,6 +6,7 @@
 
 import pytest
 from strands import Agent
+from strands.multiagent import Swarm
 from strands.multiagent.base import MultiAgentBase
 
 from strands_evals.experimental.redteam.strategies.target_session import (
@@ -206,6 +207,27 @@ def test_snapshot_captures_every_leaf_and_orchestrator(self):
         assert set(ck.agent_snapshot.agents.keys()) == {("a",), ("sub", "y")}
         assert set(ck.agent_snapshot.orchestrators.keys()) == {(), ("sub",)}
 
+    def test_snapshot_detaches_nested_orchestrator_state(self):
+        root = _FakeOrchestrator({"a": _real_agent()})
+        root._orch_state["scratch"] = {"notes": ["seed"]}
+        s = StrandsMultiAgentSession(root)
+
+        ck = s.snapshot()
+        root._orch_state["scratch"]["notes"].append("attempt")
+
+        assert ck.agent_snapshot.orchestrators[()]["orch_state"]["scratch"] == {"notes": ["seed"]}
+
+    def test_real_swarm_snapshot_detaches_live_nested_context(self):
+        swarm = Swarm([_real_agent()])
+        live_context = swarm.state.shared_context.context
+        live_context["planner"] = {"notes": ["seed"]}
+        s = StrandsMultiAgentSession(swarm)
+
+        ck = s.snapshot()
+        live_context["planner"]["notes"].append("attempt")
+
+        assert ck.agent_snapshot.orchestrators[()]["context"]["shared_context"] == {"planner": {"notes": ["seed"]}}
+
     def test_restore_rolls_back_each_leaf_messages(self):
         leaf_a = _real_agent("seed_a")
         leaf_b = _real_agent("seed_b")