diff --git a/src/strands_evals/experimental/redteam/__init__.py b/src/strands_evals/experimental/redteam/__init__.py
index 06e62753..f439aa2a 100644
--- a/src/strands_evals/experimental/redteam/__init__.py
+++ b/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",
diff --git a/src/strands_evals/experimental/redteam/experiment.py b/src/strands_evals/experimental/redteam/experiment.py
index 77a0f471..bedc060f 100644
--- a/src/strands_evals/experimental/redteam/experiment.py
+++ b/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,
             ),
         )
diff --git a/src/strands_evals/experimental/redteam/report.py b/src/strands_evals/experimental/redteam/report.py
index 326b4981..2008a0b4 100644
--- a/src/strands_evals/experimental/redteam/report.py
+++ b/src/strands_evals/experimental/redteam/report.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from collections.abc import Callable
 from dataclasses import dataclass, field
 
@@ -9,6 +10,7 @@
 
 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 = []
diff --git a/src/strands_evals/experimental/redteam/strategies/__init__.py b/src/strands_evals/experimental/redteam/strategies/__init__.py
index 9ec2d3ff..ea41c1f9 100644
--- a/src/strands_evals/experimental/redteam/strategies/__init__.py
+++ b/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",
diff --git a/src/strands_evals/experimental/redteam/strategies/base.py b/src/strands_evals/experimental/redteam/strategies/base.py
index 667e57eb..a1557410 100644
--- a/src/strands_evals/experimental/redteam/strategies/base.py
+++ b/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.
diff --git a/src/strands_evals/experimental/redteam/task.py b/src/strands_evals/experimental/redteam/task.py
index 4f3238c4..072301ac 100644
--- a/src/strands_evals/experimental/redteam/task.py
+++ b/src/strands_evals/experimental/redteam/task.py
@@ -25,14 +25,17 @@ 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.
@@ -40,7 +43,6 @@ def _build_attacker_task(
         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()],
     }
 
 
diff --git a/tests/strands_evals/experimental/redteam/test_experiment.py b/tests/strands_evals/experimental/redteam/test_experiment.py
index b4f170fc..d08ca178 100644
--- a/tests/strands_evals/experimental/redteam/test_experiment.py
+++ b/tests/strands_evals/experimental/redteam/test_experiment.py
@@ -3,6 +3,9 @@
 import pytest
 from strands.models.model import Model
 
+from strands_evals import LocalFileTaskResultStore
+from strands_evals.evaluators import Evaluator
+from strands_evals.evaluators.prompt_templates.case_prompt_template import compose_test_prompt
 from strands_evals.experimental.redteam.case import RedTeamCase
 from strands_evals.experimental.redteam.evaluators import AttackSuccessEvaluator
 from strands_evals.experimental.redteam.experiment import RedTeamExperiment
@@ -15,8 +18,9 @@
     PromptStrategy,
     SequentialBreakStrategy,
 )
-from strands_evals.experimental.redteam.strategies.base import AttackRunResult, AttackStrategy
+from strands_evals.experimental.redteam.strategies.base import RUN_RESULTS, AttackRunResult, AttackStrategy
 from strands_evals.experimental.redteam.types import AttackGoal, RedTeamConfig
+from strands_evals.types import EvaluationOutput
 
 
 class _StubModel(Model):
@@ -193,7 +197,7 @@ async def test_parallel_report_per_case_isolation():
     """End-to-end: under `max_workers > 1`, each case's output lands on its own report row.
 
     Pins the property the parallel path is *for* -- that concurrent cases don't bleed each
-    other's outputs through the shared strategy / shared `_run_meta` dict.
+    other's outputs through the shared strategy.
     """
 
     class _EchoStrategy(AttackStrategy):
@@ -494,3 +498,101 @@ def _task(case, _cap=captured):
     assert runs[0] == runs[1]  # second run identical, not squared
     # held cases were never mutated
     assert [c.name for c in exp.cases] == ["c0"]
+
+
+class _PassEvaluator(Evaluator):
+    """Deterministic evaluator so report tests don't call a judge model."""
+
+    def evaluate(self, evaluation_case):
+        return [EvaluationOutput(score=0.0, test_pass=True, reason="defended")]
+
+
+_PRUNED = [{"role": "attacker", "content": "direct ask"}, {"role": "target", "content": "no"}]
+
+
+class _RunStatsStrategy(AttackStrategy):
+    """Returns fixed run stats so tests can check they reach the report."""
+
+    @property
+    def name(self) -> str:
+        return "stats"
+
+    def run_attack(self, case, target_session, *, max_turns, model=None, **kwargs) -> AttackRunResult:
+        return AttackRunResult(conversation=[], metadata={"turns_used": 3, "backtracks": 1}, pruned_branches=_PRUNED)
+
+
+def test_custom_task_run_results_reach_report():
+    """A user task that returns `to_environment_state()` fills the report's run stats; no side channel needed."""
+    strategy = _RunStatsStrategy()
+
+    def task(case):
+        session = _FakeSession()
+        result = strategy.run_attack(case, session, max_turns=5)
+        return {
+            "output": result.conversation,
+            "trajectory": list(session.trace),
+            "environment_state": [result.to_environment_state()],
+        }
+
+    exp = RedTeamExperiment(cases=[_case("c0")], attack_strategies=[strategy], evaluators=[_PassEvaluator()])
+    (result,) = exp.run_evaluations(task=task).attack_results()
+
+    assert result.turns_used == 3
+    assert result.backtracks == 1
+    assert result.pruned_branches == _PRUNED
+
+
+class _EnvStatePromptEvaluator(Evaluator):
+    """Records the judge prompt an `uses_environment_state=True` evaluator would send."""
+
+    def __init__(self):
+        super().__init__()
+        self.prompts: list[str] = []
+
+    def evaluate(self, evaluation_case):
+        self.prompts.append(compose_test_prompt(evaluation_case, "rubric", False, uses_environment_state=True))
+        return [EvaluationOutput(score=0.0, test_pass=True, reason="defended")]
+
+
+def test_environment_state_evaluator_sees_run_results():
+    """Pins a known trade-off: evaluators that read environment state now see the run stats.
+
+    Before run stats moved into `environment_state`, such an evaluator raised for a red team run; now it
+    receives the `RUN_RESULTS` entry, including the full `pruned_branches` payload.
+    """
+    evaluator = _EnvStatePromptEvaluator()
+    exp = RedTeamExperiment(
+        cases=[_case("c0")], agent=_FakeSession(), attack_strategies=[_RunStatsStrategy()], evaluators=[evaluator]
+    )
+    exp.run_evaluations()
+
+    (prompt,) = evaluator.prompts
+    assert f"<ActualEnvironmentState>[EnvironmentState(name='{RUN_RESULTS}'" in prompt
+    assert "'turns_used': 3" in prompt
+    assert "'pruned_branches': " + repr(_PRUNED) in prompt
+
+
+def test_cached_rerun_keeps_run_stats(tmp_path):
+    """A rerun served from the evaluation data store still shows turns and blocked attempts."""
+    runs = 0
+
+    def factory():
+        nonlocal runs
+        runs += 1
+        return _FakeSession()
+
+    store = LocalFileTaskResultStore(tmp_path)
+    exp = RedTeamExperiment(
+        cases=[_case("c0")],
+        agent_factory=factory,
+        attack_strategies=[_RunStatsStrategy()],
+        evaluators=[_PassEvaluator()],
+    )
+    first = exp.run_evaluations(evaluation_data_store=store).attack_results()
+    second = exp.run_evaluations(evaluation_data_store=store).attack_results()
+
+    assert runs == 1  # the second run came from the cache
+    for (result,) in (first, second):
+        assert result.turns_used == 3
+        assert result.backtracks == 1
+        assert result.pruned_branches == _PRUNED
diff --git a/tests/strands_evals/experimental/redteam/test_report.py b/tests/strands_evals/experimental/redteam/test_report.py
index 8c4a8028..03a19996 100644
--- a/tests/strands_evals/experimental/redteam/test_report.py
+++ b/tests/strands_evals/experimental/redteam/test_report.py
@@ -1,6 +1,11 @@
 """Tests for RedTeamReport."""
 
+import warnings
+
+import pytest
+
 from strands_evals.experimental.redteam.report import AttackResult, RedTeamReport
+from strands_evals.experimental.redteam.strategies.base import RUN_RESULTS
 from strands_evals.types.evaluation import NOT_APPLICABLE, EvaluationOutput
 from strands_evals.types.evaluation_report import EvaluationReport
 
@@ -87,6 +92,65 @@ def test_single_evaluator_single_case(self):
         assert r.passes == {"judge": False}
         assert r.reasons == {"judge": "bypassed"}
 
+    def test_run_stats_read_from_run_results_environment_state(self):
+        """The stats come from the `RUN_RESULTS` state, not same-named metadata or other states."""
+        pruned = [{"role": "attacker", "content": "a"}, {"role": "target", "content": "no"}]
+        case = _case("c0", "guideline_bypass", "crescendo", "high", turns_used=99)
+        case["actual_environment_state"] = [
+            {"name": "other", "state": {"turns_used": 7}},
+            {"name": RUN_RESULTS, "state": {"turns_used": 3, "backtracks": 1, "pruned_branches": pruned}},
+        ]
+        report = RedTeamReport.from_evaluation_report(
+            _flatten(_eval_report("judge", [case], scores=[0.0], passes=[True], reasons=[""]))
+        )
+
+        (r,) = report.attack_results()
+        assert r.turns_used == 3
+        assert r.backtracks == 1
+        assert r.pruned_branches == pruned
+
+    def test_non_dict_run_results_state_yields_empty_stats(self):
+        """A custom task's own non-dict state under the reserved name must not crash the report."""
+        case = _case("c0", "guideline_bypass", "crescendo", "high", turns_used=99)
+        case["actual_environment_state"] = [{"name": RUN_RESULTS, "state": [{"test": "t1", "ok": True}]}]
+        report = RedTeamReport.from_evaluation_report(
+            _flatten(_eval_report("judge", [case], scores=[0.0], passes=[True], reasons=[""]))
+        )
+
+        (r,) = report.attack_results()
+        assert r.turns_used is None
+        assert r.backtracks is None
+        assert r.pruned_branches == []
+
+    def test_run_stats_fall_back_to_metadata_without_run_results(self):
+        case = _case("c0", "guideline_bypass", "crescendo", "high", turns_used=5, backtracks=2)
+        case["actual_environment_state"] = [{"name": "other", "state": {"turns_used": 7}}]
+        report = RedTeamReport.from_evaluation_report(
+            _flatten(_eval_report("judge", [case], scores=[0.0], passes=[True], reasons=[""]))
+        )
+
+        (r,) = report.attack_results()
+        assert r.turns_used == 5
+        assert r.backtracks == 2
+        assert r.pruned_branches == []
+
+    def test_run_meta_is_deprecated_but_still_merged(self):
+        case = _case("c0", "guideline_bypass", "crescendo", "high")
+        eval_report = _eval_report("judge", [case], scores=[0.0], passes=[True], reasons=[""])
+
+        with pytest.warns(DeprecationWarning, match="run_meta"):
+            report = RedTeamReport.from_evaluation_report(eval_report, run_meta={"c0": {"turns_used": 4}})
+
+        assert report.attack_results()[0].turns_used == 4
+
+    def test_no_deprecation_warning_without_run_meta(self):
+        case = _case("c0", "guideline_bypass", "crescendo", "high")
+        eval_report = _eval_report("judge", [case], scores=[0.0], passes=[True], reasons=[""])
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", DeprecationWarning)
+            RedTeamReport.from_evaluation_report(eval_report)
+
     def test_multiple_evaluators_merge_on_case_name(self):
         cases = [_case("c0", "guideline_bypass", "gradual_escalation", "high")]
         r1 = _eval_report("judge", cases, scores=[0.0], passes=[False], reasons=["bypassed"])
diff --git a/tests/strands_evals/experimental/redteam/test_strategies.py b/tests/strands_evals/experimental/redteam/test_strategies.py
index 96172f22..263b94c1 100644
--- a/tests/strands_evals/experimental/redteam/test_strategies.py
+++ b/tests/strands_evals/experimental/redteam/test_strategies.py
@@ -3,7 +3,7 @@
 from unittest.mock import MagicMock, patch
 
 from strands_evals.experimental.redteam.case import RedTeamCase
-from strands_evals.experimental.redteam.strategies import BUILTIN_STRATEGIES, PromptStrategy
+from strands_evals.experimental.redteam.strategies import BUILTIN_STRATEGIES, RUN_RESULTS, PromptStrategy
 from strands_evals.experimental.redteam.strategies.base import AttackRunResult
 from strands_evals.experimental.redteam.strategies.target_session import TargetCheckpoint
 from strands_evals.experimental.redteam.types import AttackGoal, RedTeamConfig
@@ -51,6 +51,32 @@ def test_redteam_config_is_strategy_agnostic():
     assert "system_prompt_template" not in RedTeamConfig.model_fields
 
 
+def test_attack_run_result_to_environment_state_keeps_all_metadata():
+    pruned = [{"role": "attacker", "content": "a"}, {"role": "target", "content": "no"}]
+    result = AttackRunResult(
+        conversation=[{"role": "attacker", "content": "hi"}],
+        strategy_succeeded=True,
+        strategy_score=0.9,
+        metadata={"turns_used": 3, "backtracks": 1, "reasoning_trace": [{"o": "x"}]},
+        pruned_branches=pruned,
+    )
+    state = result.to_environment_state()
+
+    assert state.name == RUN_RESULTS == "redteam_run_results"
+    # Strategy-specific keys (e.g. GOAT's reasoning_trace) pass through alongside the report's fields.
+    assert state.state == {
+        "turns_used": 3,
+        "backtracks": 1,
+        "reasoning_trace": [{"o": "x"}],
+        "pruned_branches": pruned,
+    }
+
+
+def test_attack_run_result_to_environment_state_omits_unrecorded_keys():
+    state = AttackRunResult(conversation=[], metadata={"turns_used": 2}).to_environment_state()
+    assert state.state == {"turns_used": 2, "pruned_branches": []}
+
+
 def test_prompt_strategy_label_defaults_to_name():
     strategy = BUILTIN_STRATEGIES["gradual_escalation"]
     assert strategy.label == strategy.name == "gradual_escalation"
diff --git a/tests/strands_evals/experimental/redteam/test_task.py b/tests/strands_evals/experimental/redteam/test_task.py
index 02b30f2b..f6b5eace 100644
--- a/tests/strands_evals/experimental/redteam/test_task.py
+++ b/tests/strands_evals/experimental/redteam/test_task.py
@@ -6,7 +6,7 @@
 import pytest
 
 from strands_evals.experimental.redteam.case import RedTeamCase
-from strands_evals.experimental.redteam.strategies.base import AttackRunResult, AttackStrategy
+from strands_evals.experimental.redteam.strategies.base import RUN_RESULTS, AttackRunResult, AttackStrategy
 from strands_evals.experimental.redteam.task import (
     MAX_ALLOWED_TURNS,
     _build_attacker_task,
@@ -90,21 +90,35 @@ def test_task_fn_calls_run_attack_and_maps_result():
 
     result = task(_case())
 
-    # Only the keys the base Experiment reads are returned; run stats flow via run_meta.
-    assert set(result) == {"output", "trajectory"}
+    # Only the keys the base Experiment reads are returned; run stats flow via environment_state.
+    assert set(result) == {"output", "trajectory", "environment_state"}
     assert result["output"][1]["content"] == "target reply"
     assert strat.reset_count == 1
 
 
-def test_task_fn_records_run_stats_into_run_meta():
-    """Strategy metadata reaches the experiment via run_meta, not the returned dict."""
-    strat = _StubStrategy()
-    run_meta: dict[str, dict] = {}
-    task = _build_attacker_task(_FakeSession(lambda _msg: "target reply"), _by_label(strat), run_meta=run_meta)
+@pytest.mark.parametrize("parallel", [False, True])
+def test_task_fn_returns_run_results_in_environment_state(parallel):
+    """Both task paths return the strategy's run stats as the `RUN_RESULTS` environment state."""
+    pruned = [{"role": "attacker", "content": "a"}, {"role": "target", "content": "no"}]
+    strat = _StubStrategy(
+        result=AttackRunResult(
+            conversation=[],
+            strategy_succeeded=True,
+            metadata={"turns_used": 3, "backtracks": 1, "target_calls": 4},
+            pruned_branches=pruned,
+        )
+    )
+    if parallel:
+        task = _build_attacker_task(
+            None, _by_label(strat), agent_factory=lambda: _FakeSession(lambda _msg: "ok"), parallel=True
+        )
+    else:
+        task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat))
 
-    task(_case("c0"))
+    (state,) = task(_case())["environment_state"]
 
-    assert run_meta["c0"]["turns_used"] == 1
+    assert state.name == RUN_RESULTS
+    assert state.state == {"turns_used": 3, "backtracks": 1, "target_calls": 4, "pruned_branches": pruned}
 
 
 class _RaisingStrategy(_StubStrategy):
@@ -121,13 +135,11 @@ def run_attack(self, case, target_session, *, max_turns, model=None, **kwargs):
 def test_task_fn_propagates_attack_error():
     """A crash must propagate: the base Experiment records it as an error reason and skips caching it,
     so a replay against the same evaluation_data_store re-runs the case instead of reading it as defended."""
-    run_meta: dict[str, dict] = {}
     strat = _RaisingStrategy(RuntimeError("target blew up"))
-    task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat), run_meta=run_meta)
+    task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat))
 
     with pytest.raises(RuntimeError, match="target blew up"):
         task(_case("c0"))
-    assert "c0" not in run_meta
 
 
 def test_task_fn_resets_strategy_each_case():