[FEAT]: Add execution trial populations and threshold verdicts

[Behnam (behnam-o)]

• Add execute_trials_async() and public PopulationResult.
• Evaluate repeated executions using configurable safety thresholds.
• Preserve population metadata across pytest, xdist, and JSON reporting.
• Update documentation and unit tests.

Proposed stack order

1. [FEAT]: Add execution trial populations and threshold verdicts #121 (this PR)
2. (DRAFT) [FEAT]: Replace trial cloning with configurable trial populations #123

Merge this PR first. GitHub cannot link these PRs as an official stack because both heads are in a fork.

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

Pull request overview

Adds a first-class “population” execution path to RAMPART’s execution layer so callers can run a safety test multiple times and assert a single threshold-based verdict, while updating pytest plugin aggregation semantics and associated docs/tests.

Changes:

• Introduces execute_trials_async() returning a new PopulationResult aggregate with threshold-based verdict semantics.
• Updates pytest trial-group aggregation/gating semantics (notably allowing UNSAFE outcomes when the SAFE pass-rate meets the threshold, while ERROR should fail the group).
• Refreshes documentation and unit tests to reflect the new execution- and aggregation-layer behavior.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/unit/pytest_plugin/test_xdist_aggregation.py	Updates xdist aggregation expectations from unconditional UNSAFE-fail to threshold-based pass behavior.
tests/unit/pytest_plugin/test_plugin.py	Adjusts trial-group aggregation unit tests and adds coverage for excluding no-result clones from the denominator.
tests/unit/probes/test_single_turn.py	Adds a regression test ensuring each trial creates an isolated session when using repeated execution.
tests/unit/core/test_result.py	Adds PopulationResult unit tests covering threshold logic and status precedence.
tests/unit/core/test_execution.py	Adds execution-layer tests for execute_trials_async() and public export checks for PopulationResult.
rampart/pytest_plugin/plugin.py	Updates gate-evaluation logging semantics (now focusing on ERROR and threshold-based pass rate).
rampart/pytest_plugin/_session.py	Changes trial-group aggregation semantics: pass-rate denominator excludes no_result, and group passing is threshold-based.
rampart/core/result.py	Introduces PopulationResult aggregate type and its status/summary behavior.
rampart/core/execution.py	Adds execute_trials_async() to run n full lifecycles and return a PopulationResult.
rampart/core/init.py	Re-exports PopulationResult from rampart.core.
rampart/init.py	Re-exports PopulationResult from the top-level rampart API.
docs/usage/pytest-integration.md	Documents execute_trials_async() semantics as the primary repeated-execution approach.
docs/usage/ci-integration.md	Updates CI guidance for trial semantics and threshold/error/no-result behavior.
docs/getting-started/quickstart.md	Updates the quickstart example to use execute_trials_async() and describes population-level assertion behavior.

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (2)

rampart/pytest_plugin/_session.py:283

• record_trial_group() is documented as “ERROR results make the group fail”, but the counting logic can miss ERRORs when a clone recorded multiple results containing both ERROR and UNSAFE (current if has_unsafe: … elif has_error: …). In that case error_count stays 0 and the group can incorrectly pass if the threshold is met. Prioritize ERROR over UNSAFE so any ERROR in a clone is always reflected in group.errors and group.passed.

        executed_count = total - no_result_count
        pass_rate = safe_count / executed_count if executed_count > 0 else 0.0
        passed = (
            error_count == 0
            and executed_count > 0

docs/usage/pytest-integration.md:71

• pytest-integration.md now documents execute_trials_async, but it no longer mentions @pytest.mark.trial even though other docs (e.g. CI integration) still recommend it and the quickstart links here as the marker reference. Add a short @pytest.mark.trial subsection clarifying that it runs clones as separate pytest items (possibly across xdist workers) and that execute_trials_async is the option when the threshold must control the single pytest verdict.

**Trial semantics:**

- One logical test produces one pytest verdict
- `threshold` sets the minimum pass rate: `threshold=0.8` requires ≥ 80% SAFE
- An `ERROR` trial resolves the population to `ERROR`
- `UNDETERMINED` trials count against the pass rate
- Individual results and the aggregate verdict are available through `PopulationResult`

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

rampart/pytest_plugin/_session.py:244

• The updated "Semantics" bullet list in this docstring has broken indentation/wrapping, and it doesn’t explicitly state the new behavior that a group fails when all clones are no_result (executed==0). Reformatting the bullets will improve readability and keep the docstring aligned with the implementation.

        Semantics:
                - ERROR results make the group fail.
                - threshold is the minimum pass rate (SAFE / executed).
          e.g. 0.8 means at least 80% of runs must be SAFE.
        - Clones with zero results (skipped or crashed before producing

rampart/pytest_plugin/plugin.py:644

• Gate logs currently report safe counts as safe/total, but pass_rate is computed using the executed denominator (total - no_result). When there are no-result clones, this can produce misleading logs like "1/2 safe (100% pass rate)". Use the executed denominator in the log counts to keep them consistent with the pass-rate semantics.

                group.total,
                group.pass_rate * 100,
                group.threshold * 100,
            )
        elif group.errors > 0:

rampart/pytest_plugin/_session.py:280

• record_trial_group counts a clone as UNSAFE before checking for ERROR (if has_unsafe: ... elif has_error: ...). If a clone recorded multiple Results and includes both an ERROR and an UNSAFE, it will be classified as UNSAFE (so error_count stays 0) and the group can incorrectly pass even though an ERROR occurred. ERROR should take precedence in the per-clone classification.

            elif has_safe:
                safe_count += 1

        executed_count = total - no_result_count
        pass_rate = safe_count / executed_count if executed_count > 0 else 0.0

[Spencer Schoenberg (spencrr)]

LGTM - let's also check with Bashir Partovi (@bashirpartovi) ! One question about the semantics of failed tests in pass/fail calculation but everything else makes sense to me and test coverage is good. Thanks Behnam (@behnam-o) !

[Bashir Partovi (bashirpartovi)]

Just to step back a little bit, I think additional_result_metadata on execute_async is doing two jobs that want opposite things. _rampart_population is framework-internal provenance that should never be able to fail, and the scenario keys the docs now recommend (scenario_id, threat_class, and so on) are caller data where failing loudly on a typo is fine. Both go into the same flat dict with strict duplicate rejection, and that's what produces the failure mode.

It's also a second convention for something we already do. absorb() injects framework keys into result.metadata after the fact, last write wins, and it never raises:

result.metadata = {
    **original_result.metadata,
    "_pytest_test_name": test_name,
    "_pytest_nodeid": node.nodeid,
    "_rampart_result_index": result_index,
}

So we already have a pattern for framework-internal metadata and this adds a stricter one next to it.

I'm also not sure the parameter buys much over writing to metadata after the call. The reason for a parameter is that handlers see it at ON_POST_EXECUTE, but ResultCollectionHandler holds a reference, the collector drains at teardown, and xdist serializes at absorb, so a post-hoc write is already visible everywhere that matters. The only thing that misses it is a custom handler reading metadata inside on_event, and population identity is a property of the aggregate rather than of one execution anyway.

The thing that convinced me the parameter is on the wrong object is that callers can't use it on the trials path. execute_trials_async(adapter, n, threshold) has no additional_result_metadata, and the only thing that fills that argument there is the framework's own _rampart_population dict. So the receipt pattern we just documented works for a single run and quietly stops being available once you switch to trials:

# works
result = await Attacks.xpia(...).execute_async(
    adapter=my_agent,
    additional_result_metadata={"scenario_id": "xpia-login-001"},
)

# no way to say the same thing here
population = await Attacks.xpia(...).execute_trials_async(
    adapter=my_agent, n=10, threshold=0.8,
)

Supporting it would mean adding the parameter to execute_trials_async as well and merging the caller's dict with _rampart_population before passing it down, which is more surface and two sources feeding one argument.

Scenario metadata is constant across every trial anyway, so it seems like it belongs where the scenario is declared rather than on each call:

attack = Attacks.xpia(
    inject=...,
    trigger="Summarize Q3",
    evaluator=...,
    metadata={
        "scenario_id": "xpia-login-001",
        "threat_class": "credential_exfiltration",
        "mitigation_ref": "SEC-1234",
    },
)

result = await attack.execute_async(adapter=my_agent)
population = await attack.execute_trials_async(adapter=my_agent, n=10, threshold=0.8)

The execution copies it onto each Result it produces, so it works the same for one run and for a population and neither method grows a parameter. Run level stuff like ci_run_url already has report.metadata and probably shouldn't be duplicated onto every result.

That leaves the population provenance, and I'd rather see it as a typed field than a reserved dict key. We already do this with injections: list[InjectionRecord], which is XPIA-specific provenance carried as a real field and documented as empty for non-XPIA tests:

@dataclass(kw_only=True)
class PopulationRef:
    """Identifies the trial population a Result belongs to."""

    id: str
    index: int
    size: int
    threshold: float


# on Result, next to injections
population: PopulationRef | None = None

and the loop sets it directly, the same way execute_async already sets duration_seconds:

population_id = uuid.uuid4().hex
results: list[Result] = []
for index in range(n):
    result = await self.execute_async(adapter=adapter)
    result.population = PopulationRef(
        id=population_id, index=index, size=n, threshold=threshold,
    )
    results.append(result)

Collisions become impossible, ty checks it, and WS0-05 gets to serialize a known field instead of a reserved key. If the typed field is too much for this PR, the smaller version is to drop the parameter and stamp into metadata in that same spot:

result.metadata["_rampart_population"] = {
    "id": population_id, "index": index, "size": n, "threshold": threshold,
}

which matches what absorb() already does and can't raise.

[Bashir Partovi (bashirpartovi)]

Thanks Behnam (@behnam-o) for working through all the feedback but I think there is a concurrency issue with the trial run that should get fixed before merging this. Let me know if it doesn't make sense.

[Bashir Partovi (bashirpartovi)]

This doesn't really run independent trials because every trial uses the same BaseExecution object. A fresh agent session gets created each time, but everything else on the execution object, like the driver, evaluator, handlers, and injection handles, gets reused. That means state from one trial can affect another.

For example, consider this case:

1. A factory creates one SingleTurnExecution or XPIAExecution.
2. That execution stores one driver in self._driver.
3. execute_trials_async() repeatedly calls _execute_async() on that same execution.
4. Each call creates a fresh agent session, but the driver is not recreated.
5. LLMDriver has one _conversation_id, and its messages are kept in PyRIT's CentralMemory.
6. Trial 2 starts with a fresh agent-side history=[], but the driver-side conversation still has trial 1.
7. LLMDriver._assert_conversations_consistent() raises DriverError.
8. BaseExecution turns that exception into an ERROR result, so a population that should contain independent trials becomes [SAFE, ERROR].

The same problem can happen with anything else that keeps state between runs.

I think each trial needs at least:

• A fresh agent session
• A fresh driver with a new conversation ID
• Fresh injection handles, unless we decide to activate them once for the whole population
• A fresh evaluator if it keeps any state

Things that don't keep state can be shared. Anything that does keep state should be recreated for each trial unless we know it was designed to be reused.

One possible way to handle this would be to keep the population loop in BaseExecution, but let each execution type create a fresh execution for every trial:

class BaseExecution:
    async def execute_trials_async(...):
        executions = [
            self._create_trial_execution()
            for _ in range(n)
        ]
        ...

Then SingleTurnExecution could create a new driver through a factory instead of reusing self._driver:

class SingleTurnExecution:
    def _create_trial_execution(self) -> BaseExecution:
        return SingleTurnExecution(
            driver=self._driver_factory(),
            ....
        )

If callers can provide stateful drivers, the public API may need to accept a factory instead of an existing driver object:

Probes.behavior(
    driver_factory=lambda: LLMDriver(...),
    evaluator=evaluator,
)

I don't think we can safely rebuild a driver with deepcopy(). Drivers can hold model clients, locks, normalizers, and external memory state that should not be copied.

For XPIA, I think we need to pick one of these options:

• Create and activate fresh injection handles for every trial
• Activate one injection around the whole population, then create fresh sessions and drivers for each trial

Either option could work, but repeatedly entering the same handle is not safe.

I would also remove max_concurrency for now. It makes this problem worse because multiple trials start using the same stateful objects at the same time.

Could we also add tests with an LLMDriver and a handle that cannot be entered twice? The current isolation test proves that sessions are different, but it uses StaticDriver, so it doesn't prove that the full trial run is independent.

[Behnam (behnam-o)]

Thanks for pointing this out. I had a design dilleman how to address this, tried a bunch of patterns, and committed option 2 below - let me know if you think it's a good choice.

Trial Isolation Design

Trials must be independent, but an execution can contain stateful drivers, evaluators, handlers, and injection handles. Reusing these objects may make trials co-dependent.

Option 1: Factory for Every Stateful Dependency

execution = Probes.behavior(
    driver_factory=lambda: LLMDriver(...),
    evaluator_factory=lambda: MyEvaluator(),
    event_handler_factories=[
        lambda: MyHandler(),
    ],
)

population = await execution.execute_trials_async(
    adapter=adapter,
    n=10,
    threshold=0.8,
)

The execution constructs fresh dependencies internally:

class SingleTurnExecution(BaseExecution):
    async def _execute_async(self, *, adapter):
        driver = self._driver_factory()
        evaluator = self._evaluator_factory()
        ...

Pros

• Preserves execution.execute_trials_async(...).
• Explicitly creates fresh dependencies.

Cons

• Requires identifying every stateful constructor argument.
• APIs accumulate driver_factory, evaluator_factory, handle_factory, etc.
• A newly added stateful dependency can easily be overlooked.
• Lifecycle policy becomes duplicated across execution implementations.

Option 2: Factory for the Complete Execution

def create_execution() -> BaseExecution:
    return Probes.behavior(
        driver=LLMDriver(...),
        evaluator=MyEvaluator(),
        event_handlers=[MyHandler()],
    )

population = await execute_trials_async(
    execution_factory=create_execution,
    adapter=adapter,
    n=10,
    threshold=0.8,
)

The framework invokes the factory for every trial:

async def run_trial(index: int) -> Result:
    execution = execution_factory()

    return await execution.execute_async(
        adapter=adapter,
    )

XPIA follows the same pattern:

def create_execution() -> BaseExecution:
    return Attacks.xpia(
        inject=surface.inject(payload=payload),
        trigger=LLMDriver(...),
        evaluator=MyEvaluator(),
    )

Pros

• Creates the complete execution graph per trial.
• Covers future constructor parameters automatically.
• Avoids unsafe copying.
• Avoids dependency-specific factory proliferation.
• Makes the independence requirement visible at the call site.
• Works consistently across execution implementations.

Cons

• Population tests use a standalone function instead of an instance method.
• Users can still accidentally close over shared mutable objects.

For example, this remains incorrect:

shared_driver = LLMDriver(...)

def create_execution() -> BaseExecution:
    return Probes.behavior(
        driver=shared_driver,
        evaluator=MyEvaluator(),
    )

Decision

We chose the complete execution factory because freshness applies to the whole execution graph, not a fixed list of known dependencies.

The resulting contract is:

Each factory invocation must return a fresh execution with fresh trial-scoped dependencies.

Single-run usage remains unchanged:

execution = Probes.behavior(
    driver=driver,
    evaluator=evaluator,
)

result = await execution.execute_async(adapter=adapter)

Only population execution requires an explicit construction strategy.