fix(redteam): carry run stats to the report via environment_state part of RedTeam Graduation Part 1

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 (
+    RUN_RESULTS,
     AttackRunResult,
     AttackStrategy,
     BadLikertJudgeStrategy,
@@ -21,6 +22,7 @@
 
 __all__ = [
     "RISK_CATEGORIES",
+    "RUN_RESULTS",
     "AdversarialCaseGenerator",
     "AttackGoal",
     "AttackResult",

src/strands_evals/experimental/redteam/experiment.py

@@ -72,9 +72,6 @@ def __init__(
         self._attack_strategies = attack_strategies or []
         self._by_label = self._build_by_label(self._attack_strategies)
         self._model = model
-        # case name -> strategy run metadata; the base Experiment drops task-returned
-        # metadata, so we join this onto the report ourselves.
-        self._run_meta: dict[str, dict[str, Any]] = {}
 
     @property
     def agent(self) -> Agent | MultiAgentBase | TargetSession | None:
@@ -177,12 +174,11 @@ async def run_evaluations_async(  # type: ignore[override]
         """
         if max_workers < 1:
             raise ValueError(f"max_workers must be >= 1, got {max_workers}")
-        self._run_meta.clear()
         if task is None:
             task = self._default_task(parallel=max_workers > 1)
         # Swap _cases for the expanded cross-product. Each case has a unique name (case x strategy
         # label), so parallel workers never collide on the same key in the base runner's results
-        # buffer or in `self._run_meta`.
+        # buffer.
         original_cases = self._cases
         self._cases = self._expand_cross_product()
         try:
@@ -191,7 +187,7 @@ async def run_evaluations_async(  # type: ignore[override]
             )
         finally:
             self._cases = original_cases
-        return RedTeamReport.from_evaluation_report(report, run_meta=self._run_meta)
+        return RedTeamReport.from_evaluation_report(report)
 
     def _default_task(self, *, parallel: bool = False) -> Callable[[Case[InputT, OutputT]], Any]:
         if self._agent is None and self._agent_factory is None:
@@ -206,7 +202,6 @@ def _default_task(self, *, parallel: bool = False) -> Callable[[Case[InputT, Out
                 self._by_label,
                 agent_factory=self._agent_factory,
                 model=self._model,
-                run_meta=self._run_meta,
                 parallel=parallel,
             ),
         )

src/strands_evals/experimental/redteam/report.py

@@ -2,13 +2,15 @@
 
 from __future__ import annotations
 
+import warnings
 from collections.abc import Callable
 from dataclasses import dataclass, field
 
 from rich.console import Console
 
 from ...types.evaluation import EvaluationOutput
 from ...types.evaluation_report import EvaluationReport
+from .strategies.base import RUN_RESULTS
 
 _console = Console()
 
@@ -91,9 +93,18 @@ def from_evaluation_report(cls, report: EvaluationReport, run_meta: dict[str, di
 
         Args:
             report: Flattened report from the base experiment, one row per (case, evaluator).
-            run_meta: Per-case strategy run metadata keyed by case name; merged into each case's metadata so
-                the report sees it.
+            run_meta: Deprecated. Per-case strategy run metadata keyed by case name; merged into each case's
+                metadata so the report sees it. Return `AttackRunResult.to_environment_state()` in the task's
+                `environment_state` instead.
         """
+        if run_meta is not None:
+            warnings.warn(
+                "`run_meta` is deprecated and will be removed in a future release. Return "
+                "`AttackRunResult.to_environment_state()` in the task's `environment_state` instead; the run "
+                "stats are then read from the case's `actual_environment_state`, not its `metadata`.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
         run_meta = run_meta or {}
         n = len(report.cases)
         if not (len(report.scores) == n and len(report.test_passes) == n and len(report.reasons) == n):
@@ -125,6 +136,7 @@ def attack_results(self) -> list[AttackResult]:
             name = case_data.get("name", f"case_{i}")
             evaluator = case_data.get("evaluator", "evaluator")
             metadata = case_data.get("metadata") or {}
+            run_results = _run_results(case_data)
             result = by_case.setdefault(
                 name,
                 AttackResult(
@@ -133,10 +145,10 @@ def attack_results(self) -> list[AttackResult]:
                     strategy=metadata.get("strategy", "unknown"),
                     severity=metadata.get("severity", "unknown"),
                     objective=metadata.get("actor_goal", ""),
-                    turns_used=metadata.get("turns_used"),
-                    backtracks=metadata.get("backtracks"),
+                    turns_used=run_results.get("turns_used"),
+                    backtracks=run_results.get("backtracks"),
                     conversation=case_data.get("actual_output") or [],
-                    pruned_branches=metadata.get("pruned_branches") or [],
+                    pruned_branches=run_results.get("pruned_branches") or [],
                 ),
             )
             result.reasons[evaluator] = self.reasons[i]
@@ -350,6 +362,16 @@ def _base_case_is_unique(results: list[AttackResult]) -> bool:
     return len(set(keys)) == len(keys)
 
 
+def _run_results(case_data: dict) -> dict:
+    """Return the row's `RUN_RESULTS` environment state, falling back to `metadata` for older reports."""
+    for state in case_data.get("actual_environment_state") or []:
+        if state.get("name") == RUN_RESULTS:
+            run_results = state.get("state")
+            # Custom tasks own `environment_state`; tolerate a non-dict state rather than crash the report.
+            return run_results if isinstance(run_results, dict) else {}
+    return case_data.get("metadata") or {}
+
+
 def _format_run_stats(result: AttackResult) -> str:
     """Render the strategy's per-run stats when present."""
     parts = []

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 RUN_RESULTS, AttackRunResult, AttackStrategy
 from .crescendo import CrescendoStrategy
 from .goat import GoatStrategy
 from .pair import PairStrategy
@@ -24,6 +24,7 @@
 
 __all__ = [
     "BUILTIN_STRATEGIES",
+    "RUN_RESULTS",
     "AttackRunResult",
     "AttackStrategy",
     "BadLikertJudgeStrategy",

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

@@ -8,10 +8,15 @@
 
 from strands.models.model import Model
 
+from ....types import EnvironmentState
+
 if TYPE_CHECKING:
     from ..case import RedTeamCase
     from .target_session import TargetSession
 
+# Name of the task-output `environment_state` entry `RedTeamReport` reads run stats from.
+RUN_RESULTS = "redteam_run_results"
+
 
 @dataclass
 class AttackRunResult:
@@ -35,6 +40,13 @@ class AttackRunResult:
     metadata: dict[str, Any] = field(default_factory=dict)
     pruned_branches: list[dict[str, Any]] = field(default_factory=list)
 
+    def to_environment_state(self) -> EnvironmentState:
+        """Return the `RUN_RESULTS` entry a task puts in its `environment_state` for `RedTeamReport` to read.
+
+        The state holds the strategy's `metadata` plus `pruned_branches`.
+        """
+        return EnvironmentState(name=RUN_RESULTS, state={**self.metadata, "pruned_branches": self.pruned_branches})
+
 
 class AttackStrategy(ABC):
     """Base class for red team attack strategies.

src/strands_evals/experimental/redteam/task.py

@@ -25,22 +25,24 @@ def _build_attacker_task(
     *,
     agent_factory: Callable[[], Agent | MultiAgentBase | TargetSession] | None = None,
     model: Model | str | None = None,
-    run_meta: dict[str, dict[str, Any]] | None = None,
     parallel: bool = False,
 ) -> Callable[[RedTeamCase], dict]:
-    """Build a `task(case) -> {"output": conversation, "trajectory": tool_uses}` callable.
+    """Build a `task(case) -> {"output", "trajectory", "environment_state"}` 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
-    is recorded into `run_meta` keyed by case name.
+    `strategy.run_attack`, injecting a `TargetSession`. `MAX_ALLOWED_TURNS` is the hard ceiling.
+
+    The returned dict holds:
+        output: The attacker/target conversation.
+        trajectory: The target's tool uses.
+        environment_state: `[result.to_environment_state()]`, the run stats `RedTeamReport` reads.
 
     Args:
         agent: The shared target for sequential runs. Required when `agent_factory` is None.
         by_label: Strategy registry keyed by `metadata["strategy"]` label.
         agent_factory: Zero-arg callable returning a fresh target for each case. Required for parallel
             runs (`parallel=True`); takes precedence over `agent` when both are set.
         model: Model passed through to `strategy.run_attack` for strategy-internal LLM calls.
-        run_meta: Per-case strategy metadata sink, keyed by `case.name`.
         parallel: When True, every case is built from `agent_factory` so concurrent cases never share
             mutable state. When False, all cases share one target and rewind to a once-captured baseline
             between cases.
@@ -57,7 +59,6 @@ def _build_attacker_task(
             agent_factory=agent_factory,
             by_label=by_label,
             model=model,
-            run_meta=run_meta,
         )
 
     # Sequential path: `agent` is guaranteed non-None here -- the upfront check rejects (None,
@@ -66,7 +67,6 @@ def _build_attacker_task(
         agent=agent,  # type: ignore[arg-type]
         by_label=by_label,
         model=model,
-        run_meta=run_meta,
     )
 
 
@@ -75,7 +75,6 @@ def _build_shared_target_task_fn(
     agent: Agent | MultiAgentBase | TargetSession,
     by_label: dict[str, AttackStrategy],
     model: Model | str | None,
-    run_meta: dict[str, dict[str, Any]] | None,
 ) -> Callable[[RedTeamCase], dict]:
     """Build a task fn that drives one shared target across cases, rewinding to a fixed baseline.
 
@@ -97,7 +96,7 @@ def task_fn(case: RedTeamCase) -> dict:
         session = _build_session(agent, baseline=initial_snapshot)
         session.reset()
 
-        return _run_attack(strategy, case, session, model=model, run_meta=run_meta)
+        return _run_attack(strategy, case, session, model=model)
 
     return task_fn
 
@@ -108,7 +107,6 @@ def _build_per_case_task_fn(
     agent_factory: Callable[[], Agent | MultiAgentBase | TargetSession] | None,
     by_label: dict[str, AttackStrategy],
     model: Model | str | None,
-    run_meta: dict[str, dict[str, Any]] | None,
 ) -> Callable[[RedTeamCase], dict]:
     """Build a per-case task fn that constructs its own target every case.
 
@@ -127,9 +125,7 @@ def task_fn(case: RedTeamCase) -> dict:
         session = _build_session(make_target(), baseline=None)
         session.reset()
 
-        # CPython dict assignment for a single distinct key is atomic, and case names are unique
-        # per cross-product expansion, so concurrent writers never target the same key.
-        return _run_attack(strategy, case, session, model=model, run_meta=run_meta)
+        return _run_attack(strategy, case, session, model=model)
 
     return task_fn
 
@@ -140,20 +136,18 @@ def _run_attack(
     session: TargetSession,
     *,
     model: Model | str | None,
-    run_meta: dict[str, dict[str, Any]] | None,
 ) -> dict:
-    """Run one attack and record its strategy metadata into `run_meta`.
+    """Run one attack and return its conversation, tool uses and run stats.
 
     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)
-    if run_meta is not None and case.name is not None:
-        run_meta[case.name] = {**result.metadata, "pruned_branches": result.pruned_branches}
     return {
         "output": result.conversation,
         # Snapshot of the trace; the next case's session.reset() clears this list in place.
         "trajectory": list(session.trace),
+        "environment_state": [result.to_environment_state()],
     }