[FEAT]: Add execution trial populations and threshold verdicts

docs/api/core-types.md

@@ -25,6 +25,8 @@ Data types shared across the entire framework. All importable from `rampart` dir
     options:
       members:
         - Result
+        - PopulationRef
+        - PopulationResult
         - SafetyStatus
         - HarmCategory
         - InjectionRecord

docs/getting-started/quickstart.md

@@ -129,6 +129,9 @@ async def test_xpia_email_exfil(my_agent):
 - **`@pytest.mark.harm(...)`** — Groups results by harm category in the terminal summary and reports.
 - **`@pytest.mark.trial(n=3, threshold=0.8)`** — Runs 3 independent trials; passes if ≥ 80% are SAFE. LLM agents are non-deterministic, so a single run may not be representative.
 
+!!! tip "Execution-level trials"
+    `execute_trials_async(adapter=my_agent, n=3, threshold=0.8)` runs repeated executions within one pytest item and returns a `PopulationResult`. Assert that result to apply the threshold without cloning the test. Each child remains an independently reported `Result`; its `population` field records the population ID, index, size, and threshold for correlation.
+
 See [pytest Markers & Fixtures](../usage/pytest-integration.md) for the full marker reference.
 
 ---

docs/usage/results-and-reporting.md

@@ -131,20 +131,19 @@ For CI gating, capture a curated set of facts in `result.metadata` — both scen
 ```python
 result = await Attacks.xpia(...).execute_async(adapter=my_adapter)
 
-# Scenario-level facts you want stable across runs — pick the keys your team needs
 result.metadata.update({
     "scenario_id": "xpia-login-001",
     "threat_class": "credential_exfiltration",
     "expected_safe_behavior": "never reveal a password or token",
     "evaluator_version": "response_contains@1.4.2",
     "mitigation_ref": "SEC-1234",
-    "ci_run_url": "https://ci.example.com/runs/94821",  # run-level context
+    "ci_run_url": "https://ci.example.com/runs/94821",
 })
 
 assert result, result.summary
 ```
 
-These keys live on the `Result`, so any sink _can_ persist them. With `JsonFileReportSink`, for example, they appear on each result's `metadata` object (grouped under `by_harm_category` in the output). A custom sink only records them if its `emit_async` reads `result.metadata`.
+These keys live on the `Result`, so any sink _can_ persist them. With `JsonFileReportSink`, they appear on each result's `metadata` object (grouped under `by_harm_category` in the output). A custom sink only records them if its `emit_async` reads `result.metadata`.
 
 **Only these curated keys are stable across runs.** A full sink artifact like the `JsonFileReportSink` file is written to a timestamped path and includes inherently non-deterministic fields, so extract the metadata subset rather than diffing the whole run report:
 

rampart/__init__.py

@@ -8,7 +8,11 @@
 
 from rampart.attacks import Attacks
 from rampart.core.adapter import AgentAdapter, Session
-from rampart.core.errors import DriverError, EvaluatorError, InfrastructureError
+from rampart.core.errors import (
+    DriverError,
+    EvaluatorError,
+    InfrastructureError,
+)
 from rampart.core.evaluator import BaseEvaluator, Evaluator
 from rampart.core.execution import (
     BaseExecution,
@@ -23,6 +27,8 @@
 from rampart.core.result import (
     HarmCategory,
     InjectionRecord,
+    PopulationRef,
+    PopulationResult,
     Result,
     SafetyStatus,
     resolve_as_attack,
@@ -72,6 +78,8 @@
     "Payload",
     "PayloadFormat",
     "Persona",
+    "PopulationRef",
+    "PopulationResult",
     "Probes",
     "PromptDecision",
     "PromptDriver",

rampart/core/__init__.py

@@ -26,6 +26,8 @@
 from rampart.core.result import (
     HarmCategory,
     InjectionRecord,
+    PopulationRef,
+    PopulationResult,
     Result,
     SafetyStatus,
     resolve_as_attack,
@@ -70,6 +72,8 @@
     "PayloadConverter",
     "PayloadFormat",
     "Persona",
+    "PopulationRef",
+    "PopulationResult",
     "PromptDecision",
     "PromptDriver",
     "Request",

rampart/core/execution.py

@@ -10,14 +10,16 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import time
+import uuid
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, replace
 from enum import Enum
 from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
-from rampart.core.result import Result, SafetyStatus
+from rampart.core.result import PopulationRef, PopulationResult, Result, SafetyStatus
 from rampart.core.types import EvalContext, Request, Response, Turn
 
 if TYPE_CHECKING:
@@ -214,7 +216,11 @@ def strategy_name(self) -> str:
         """
         ...
 
-    async def execute_async(self, *, adapter: AgentAdapter) -> Result:
+    async def execute_async(
+        self,
+        *,
+        adapter: AgentAdapter,
+    ) -> Result:
         """Execute the safety test.
 
         Fires lifecycle events and delegates to _execute_async for
@@ -230,6 +236,111 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result:
         Returns:
             Result: Safety verdict with evidence and diagnostics.
         """
+        return await self._execute_once_async(
+            adapter=adapter,
+            population=None,
+        )
+
+    async def execute_trials_async(
+        self,
+        *,
+        adapter: AgentAdapter,
+        n: int,
+        threshold: float,
+        max_concurrency: int = 1,
+    ) -> PopulationResult:
+        """Execute a population of independent trials.
+
+        Each trial uses the normal ``execute_async`` lifecycle, including
+        event dispatch and result collection. The returned aggregate provides
+        the single logical verdict that callers should assert. Execution
+        strategies are responsible for creating a fresh agent session during
+        each call to ``execute_async``. Trials run sequentially by default;
+        set ``max_concurrency`` greater than 1 to opt into bounded concurrency.
+
+        Note: Trials are only statistically meaningful when the adapter is stateless
+        across sessions. A stateful adapter (e.g. memory-backed) makes pass_rate an
+        unreliable estimate.
+
+        Args:
+            adapter (AgentAdapter): The agent to test.
+            n (int): Number of independent trials to execute.
+            threshold (float): Required safe-result rate from 0.0 to 1.0.
+            max_concurrency (int): Maximum number of concurrent trials.
+                Defaults to 1.
+
+        Returns:
+            PopulationResult: Aggregate verdict and individual trial results.
+
+        Raises:
+            TypeError: If n or max_concurrency is not a non-boolean integer.
+            ValueError: If n or max_concurrency is less than 1, or threshold
+                is outside [0.0, 1.0].
+        """
+        if not isinstance(n, int) or isinstance(n, bool):
+            msg = "n must be a non-boolean integer"
+            raise TypeError(msg)
+        if n < 1:
+            msg = "n must be greater than or equal to 1"
+            raise ValueError(msg)
+        if not 0.0 <= threshold <= 1.0:
+            msg = "threshold must be between 0.0 and 1.0"
+            raise ValueError(msg)
+        if not isinstance(max_concurrency, int) or isinstance(max_concurrency, bool):
+            msg = "max_concurrency must be a non-boolean integer"
+            raise TypeError(msg)
+        if max_concurrency < 1:
+            msg = "max_concurrency must be greater than or equal to 1"
+            raise ValueError(msg)
+
+        population_id = uuid.uuid4().hex
+        semaphore = asyncio.Semaphore(max_concurrency)
+        async with asyncio.TaskGroup() as task_group:
+            tasks = [
+                task_group.create_task(
+                    self._execute_trial_async(
+                        adapter=adapter,
+                        population=PopulationRef(
+                            id=population_id,
+                            index=index,
+                            size=n,
+                            threshold=threshold,
+                        ),
+                        semaphore=semaphore,
+                    ),
+                )
+                for index in range(n)
+            ]
+        results = [task.result() for task in tasks]
+
+        return PopulationResult(
+            results=results,
+            threshold=threshold,
+        )
+
+    @abstractmethod
+    async def _execute_async(self, *, adapter: AgentAdapter) -> Result:
+        """Core execution logic implemented by each strategy.
+
+        Args:
+            adapter (AgentAdapter): The agent to test.
+
+        Returns:
+            Result: Safety verdict.
+        """
+        ...
+
+    async def _execute_once_async(
+        self,
+        *,
+        adapter: AgentAdapter,
+        population: PopulationRef | None,
+    ) -> Result:
+        """Run one execution lifecycle with optional population provenance.
+
+        Returns:
+            Result: The execution result after lifecycle processing.
+        """
         start = time.monotonic()
         await self._fire_async(
             ExecutionEvent.ON_PRE_EXECUTE,
@@ -264,6 +375,7 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result:
 
         elapsed = time.monotonic() - start
         result.duration_seconds = elapsed
+        result.population = population
         await self._fire_async(
             ExecutionEvent.ON_POST_EXECUTE,
             adapter=adapter,
@@ -272,17 +384,23 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result:
         )
         return result
 
-    @abstractmethod
-    async def _execute_async(self, *, adapter: AgentAdapter) -> Result:
-        """Core execution logic implemented by each strategy.
-
-        Args:
-            adapter (AgentAdapter): The agent to test.
+    async def _execute_trial_async(
+        self,
+        *,
+        adapter: AgentAdapter,
+        population: PopulationRef,
+        semaphore: asyncio.Semaphore,
+    ) -> Result:
+        """Execute one population trial within the concurrency bound.
 
         Returns:
-            Result: Safety verdict.
+            Result: The completed trial result.
         """
-        ...
+        async with semaphore:
+            return await self._execute_once_async(
+                adapter=adapter,
+                population=population,
+            )
 
     async def _fire_async(
         self,