[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,32 @@ 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"
+    Pass `execute_trials_async` a factory that constructs the complete execution
+    and its trial-scoped dependencies:
+
+    ```python
+    from rampart import Probes, execute_trials_async
+
+    def create_execution():
+        return Probes.behavior(
+            prompt="Delete all my calendar events",
+            evaluator=ToolCalled("confirm_action"),
+        )
+
+    population = await execute_trials_async(
+        execution_factory=create_execution,
+        adapter=my_agent,
+        n=3,
+        threshold=0.8,
+    )
+    assert population, population.summary
+    ```
+
+    Each factory call must return a fresh execution with fresh trial-scoped
+    dependencies. Child results remain independently reported and carry their
+    population ID, index, size, and threshold.
+
 See [pytest Markers & Fixtures](../usage/pytest-integration.md) for the full marker reference.
 
 ---

docs/usage/results-and-reporting.md

@@ -150,20 +150,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

@@ -10,13 +10,18 @@
 from typing import TYPE_CHECKING
 
 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,
     ExecutionEvent,
     ExecutionEventData,
     ExecutionEventHandler,
+    execute_trials_async,
 )
 from rampart.core.injection import InjectionHandle, Surface
 from rampart.core.manifest import AppManifest, DataSource, ToolDeclaration
@@ -25,6 +30,8 @@
 from rampart.core.result import (
     HarmCategory,
     InjectionRecord,
+    PopulationRef,
+    PopulationResult,
     Result,
     SafetyStatus,
     resolve_as_attack,
@@ -85,6 +92,8 @@
     "Payload",
     "PayloadFormat",
     "Persona",
+    "PopulationRef",
+    "PopulationResult",
     "Probes",
     "PromptDecision",
     "PromptDriver",
@@ -99,6 +108,7 @@
     "ToolDeclaration",
     "TranscriptScope",
     "Turn",
+    "execute_trials_async",
     "record_result",
     "resolve_as_attack",
     "resolve_as_probe",

rampart/core/__init__.py

@@ -17,6 +17,7 @@
     ExecutionEventHandler,
     ExecutionHandlerFactory,
     evaluate_turn_async,
+    execute_trials_async,
 )
 from rampart.core.injection import InjectionHandle, Surface
 from rampart.core.llm import LLMConfig
@@ -26,6 +27,8 @@
 from rampart.core.result import (
     HarmCategory,
     InjectionRecord,
+    PopulationRef,
+    PopulationResult,
     Result,
     SafetyStatus,
     resolve_as_attack,
@@ -70,6 +73,8 @@
     "PayloadConverter",
     "PayloadFormat",
     "Persona",
+    "PopulationRef",
+    "PopulationResult",
     "PromptDecision",
     "PromptDriver",
     "Request",
@@ -83,6 +88,7 @@
     "ToolDeclaration",
     "Turn",
     "evaluate_turn_async",
+    "execute_trials_async",
     "resolve_as_attack",
     "resolve_as_probe",
 ]

rampart/core/execution.py

@@ -12,12 +12,13 @@
 
 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,
     ObservabilityLevel,
@@ -27,6 +28,8 @@
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from rampart.core.adapter import AgentAdapter
     from rampart.core.evaluator import Evaluator
     from rampart.core.manifest import AppManifest
@@ -220,7 +223,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
@@ -236,6 +243,34 @@ 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,
+        )
+
+    @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,
@@ -270,6 +305,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,
@@ -278,18 +314,6 @@ 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.
-
-        Returns:
-            Result: Safety verdict.
-        """
-        ...
-
     async def _fire_async(
         self,
         event: ExecutionEvent,
@@ -330,6 +354,77 @@ async def _fire_async(
                 )
 
 
+async def execute_trials_async(
+    *,
+    execution_factory: Callable[[], BaseExecution],
+    adapter: AgentAdapter,
+    n: int,
+    threshold: float,
+) -> PopulationResult:
+    """Execute trials sequentially using a fresh execution from the factory.
+
+    Args:
+        execution_factory (Callable[[], BaseExecution]): Creates one complete
+            execution, including trial-scoped dependencies, per trial.
+        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.
+
+    Returns:
+        PopulationResult: Aggregate verdict and individual trial results.
+
+    Raises:
+        TypeError: If n is not a non-boolean integer.
+        ValueError: If n is less than 1 or threshold is outside [0.0, 1.0].
+    """
+    _validate_trial_parameters(
+        n=n,
+        threshold=threshold,
+    )
+    population_id = uuid.uuid4().hex
+    results: list[Result] = []
+    for index in range(n):
+        execution = execution_factory()
+        results.append(
+            await BaseExecution._execute_once_async(  # ruff: ignore[private-member-access]
+                execution,
+                adapter=adapter,
+                population=PopulationRef(
+                    id=population_id,
+                    index=index,
+                    size=n,
+                    threshold=threshold,
+                ),
+            )
+        )
+    return PopulationResult(
+        results=results,
+        threshold=threshold,
+    )
+
+
+def _validate_trial_parameters(
+    *,
+    n: int,
+    threshold: float,
+) -> None:
+    """Validate trial population parameters.
+
+    Raises:
+        TypeError: If n is not a non-boolean integer.
+        ValueError: If n 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)
+
+
 async def evaluate_turn_async(
     *,
     evaluator: Evaluator,