[FEAT]: Add execution-layer trial populations and threshold verdicts

[behnam-o]

• Add execute_trials_async() for repeated execution with one aggregate verdict.
• Add and publicly export PopulationResult.
• Define threshold semantics for SAFE, UNSAFE, UNDETERMINED, and ERROR results.
• Reconcile legacy trial aggregation with threshold-based verdicts while retaining @pytest.mark.trial compatibility.
• Update trial documentation and unit tests.

[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

[bashirpartovi]

I think there is a problem we need to address because only the child Results are auto‑recorded, the PopulationResult and its threshold and grouping is discarded when the test body returns. We may want to capture that in the report.

[bashirpartovi]

Maybe we could do something like this:

import uuid  

    ...
    population_id = uuid.uuid4().hex
    results: list[Result] = []
    for index in range(n):
        result = await self.execute_async(adapter=adapter)
        result.metadata.setdefault(
            "_rampart_population",
            {
                "id": population_id,
                "threshold": threshold,
                "n": n,
                "index": index,
            },
        )
        results.append(result)

    return PopulationResult(results=results, threshold=threshold)

Although we need to think about this a bit more. uuid avoids population conflict but varies run to run..

[behnam-o]

@bashirpartovi good callout.

I liked your metadata suggestion and implemented it with a minor tweak: execute_async() now accepts additional result metadata and attaches it before ON_POST_EXECUTE fires. This ensures every event handler, including the built-in result collector, observes the final annotated Result, and the metadata flows through to reporting sinks by default.

let me know what you think

[behnam-o]

also about the uuid - I think it's actuallya good think to use new ones per population, because it only groups results from one execute_trials_async() invocation; stable pytest context such as _pytest_nodeid and _rampart_result_index is persisted separately for cross-run identification and ordering.

[spencrr]

LGTM - let's also check with @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-o !

[spencrr]

just verifying this is equivalent to not(0.0 <= self.threshold and self.threshold <= 1.0)

[spencrr]

do we wait to await all here?

[spencrr]

or as an opt-in option?

[spencrr]

just checking on the semantics here: so error-ed trials now do not count toward a failure -> does this mean if we have errors, denominator is decreased and test passes?

[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.

[bashirpartovi]

This can completely terminate entire population data. _add_result_metadata raises ValueError on duplicate key, but this is outside the try/except clause so any error raised bypasses ON_POST_EXECUTE and ON_ERROR events, therefore, the result is never collected and no error result is produced. This just crashes the entire run. And since this is used in execute_trials_async, the blast radius is the whole population, all prior trials are discarded. test_rejects_duplicate_additional_metadata_async already pins this behavior, it asserts only ON_PRE_EXECUTE fired.

This isn't only programmer error either, it's reachable from adapter data. _collect_response_metadata promotes t.response.metadata straight to the top level of Result.metadata whenever exactly one turn carries metadata, and its own docstring says adapters attach things like conversation_id and session_id. results-and-reporting.md now tells users to pass keys like scenario_id and session_id through additional_result_metadata, so a collision is easy to hit by accident rather than something the caller can always predict.

Can we move the _add_result_metadata call inside the try/except so a collision degrades to an ERROR result like every other failure? At minimum the framework's own _rampart_population stamping shouldn't be able to abort a run.

[bashirpartovi]

typo:

.. sessions. a stateful ..

"a" should be capitalized