diff --git a/docs/api/core-types.md b/docs/api/core-types.md index 2a343b5..9818036 100644 --- a/docs/api/core-types.md +++ b/docs/api/core-types.md @@ -25,6 +25,7 @@ Data types shared across the entire framework. All importable from `rampart` dir options: members: - Result + - PopulationResult - SafetyStatus - HarmCategory - InjectionRecord diff --git a/docs/getting-started/quickstart.md b/docs/getting-started/quickstart.md index 196994b..77fe4e8 100644 --- a/docs/getting-started/quickstart.md +++ b/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 `_rampart_population` metadata records the population ID, index, size, and threshold for optional correlation. + See [pytest Markers & Fixtures](../usage/pytest-integration.md) for the full marker reference. --- diff --git a/docs/usage/ci-integration.md b/docs/usage/ci-integration.md index 00c6f47..05af4fb 100644 --- a/docs/usage/ci-integration.md +++ b/docs/usage/ci-integration.md @@ -46,8 +46,10 @@ This runs 10 independent trials. The test group passes only if ≥ 80% of trials - Each trial clone appears as a separate pytest item - The aggregate verdict appears in the RAMPART terminal summary -- Any `UNSAFE` trial → the group fails -- `ERROR` trials count against the pass rate +- The aggregate passes when the SAFE pass rate meets the threshold +- Any `ERROR` trial makes the aggregate fail +- `UNSAFE` and `UNDETERMINED` trials count against the pass rate +- Clones that produce no RAMPART result are excluded from the pass-rate denominator --- diff --git a/docs/usage/pytest-integration.md b/docs/usage/pytest-integration.md index 565cfec..154da53 100644 --- a/docs/usage/pytest-integration.md +++ b/docs/usage/pytest-integration.md @@ -63,9 +63,10 @@ async def test_with_threshold(adapter): **Trial semantics:** - Each trial clone runs independently as a separate pytest item -- Any `UNSAFE` result in any trial → the group **fails** - `threshold` sets the minimum pass rate: `threshold=0.8` requires ≥ 80% SAFE -- `ERROR` results count against the pass rate (they are not `SAFE`) +- Any `ERROR` result makes the aggregate group fail +- `UNSAFE` and `UNDETERMINED` results count against the pass rate +- Clones that produce no RAMPART result are excluded from the pass-rate denominator - The trial group aggregate appears in the terminal summary !!! tip "Running trials in parallel" diff --git a/docs/usage/results-and-reporting.md b/docs/usage/results-and-reporting.md index ae38520..5ad2715 100644 --- a/docs/usage/results-and-reporting.md +++ b/docs/usage/results-and-reporting.md @@ -129,22 +129,22 @@ 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 -}) +result = await Attacks.xpia(...).execute_async( + adapter=my_adapter, + additional_result_metadata={ + "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", + }, +) 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`. +Additional metadata is attached before `ON_POST_EXECUTE`, so event handlers and sinks see the same result state. It is strictly additive: reusing a key already produced by the execution raises `ValueError`. Keys beginning with `_rampart_` are conventionally used by the framework and should be avoided by callers. With `JsonFileReportSink`, these keys 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: diff --git a/rampart/__init__.py b/rampart/__init__.py index 8a0f807..4e8e926 100644 --- a/rampart/__init__.py +++ b/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,7 @@ from rampart.core.result import ( HarmCategory, InjectionRecord, + PopulationResult, Result, SafetyStatus, resolve_as_attack, @@ -72,6 +77,7 @@ "Payload", "PayloadFormat", "Persona", + "PopulationResult", "Probes", "PromptDecision", "PromptDriver", diff --git a/rampart/core/__init__.py b/rampart/core/__init__.py index 9c823d5..f0a3a4a 100644 --- a/rampart/core/__init__.py +++ b/rampart/core/__init__.py @@ -26,6 +26,7 @@ from rampart.core.result import ( HarmCategory, InjectionRecord, + PopulationResult, Result, SafetyStatus, resolve_as_attack, @@ -70,6 +71,7 @@ "PayloadConverter", "PayloadFormat", "Persona", + "PopulationResult", "PromptDecision", "PromptDriver", "Request", diff --git a/rampart/core/execution.py b/rampart/core/execution.py index 34fa92e..ede7df6 100644 --- a/rampart/core/execution.py +++ b/rampart/core/execution.py @@ -12,15 +12,19 @@ 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 PopulationResult, Result, SafetyStatus from rampart.core.types import EvalContext, Request, Response, Turn if TYPE_CHECKING: + from collections.abc import Mapping + from typing import Any + from rampart.core.adapter import AgentAdapter from rampart.core.evaluator import Evaluator from rampart.core.manifest import AppManifest @@ -214,7 +218,12 @@ def strategy_name(self) -> str: """ ... - async def execute_async(self, *, adapter: AgentAdapter) -> Result: + async def execute_async( + self, + *, + adapter: AgentAdapter, + additional_result_metadata: Mapping[str, Any] | None = None, + ) -> Result: """Execute the safety test. Fires lifecycle events and delegates to _execute_async for @@ -226,9 +235,17 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: Args: adapter (AgentAdapter): The agent to test. + additional_result_metadata (Mapping[str, Any] | None): Metadata to + add to the result before ON_POST_EXECUTE. Existing result + metadata cannot be overwritten. Keys beginning with + ``_rampart_`` are conventionally used by the framework. Returns: Result: Safety verdict with evidence and diagnostics. + + Raises: + ValueError: If additional_result_metadata contains a key already + present in the result metadata. """ start = time.monotonic() await self._fire( @@ -264,6 +281,10 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: elapsed = time.monotonic() - start result.duration_seconds = elapsed + self._add_result_metadata( + result=result, + additional_result_metadata=additional_result_metadata, + ) await self._fire( ExecutionEvent.ON_POST_EXECUTE, adapter=adapter, @@ -272,6 +293,69 @@ async def execute_async(self, *, adapter: AgentAdapter) -> Result: ) return result + async def execute_trials_async( + self, + *, + adapter: AgentAdapter, + n: int, + threshold: float, + ) -> 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``. + + 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. + + 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]. + """ + 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) + + population_id = uuid.uuid4().hex + results: list[Result] = [] + for index in range(n): + result = await self.execute_async( + adapter=adapter, + additional_result_metadata={ + "_rampart_population": { + "id": population_id, + "index": index, + "size": n, + "threshold": threshold, + }, + }, + ) + results.append(result) + + return PopulationResult( + results=results, + threshold=threshold, + ) + @abstractmethod async def _execute_async(self, *, adapter: AgentAdapter) -> Result: """Core execution logic implemented by each strategy. @@ -284,6 +368,28 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result: """ ... + @staticmethod + def _add_result_metadata( + *, + result: Result, + additional_result_metadata: Mapping[str, Any] | None, + ) -> None: + """Add metadata without overwriting keys produced by the execution. + + Raises: + ValueError: If an additional metadata key already exists. + """ + if not additional_result_metadata: + return + + duplicate_keys = result.metadata.keys() & additional_result_metadata.keys() + if duplicate_keys: + formatted_keys = ", ".join(sorted(duplicate_keys)) + msg = f"Result metadata already contains key(s): {formatted_keys}" + raise ValueError(msg) + + result.metadata = {**result.metadata, **additional_result_metadata} + async def _fire( self, event: ExecutionEvent, diff --git a/rampart/core/result.py b/rampart/core/result.py index 79320fc..80f48b1 100644 --- a/rampart/core/result.py +++ b/rampart/core/result.py @@ -3,9 +3,9 @@ """Core result types for the RAMPART framework. -Defines the single Result type, SafetyStatus, HarmCategory, InjectionRecord, -and the resolve_as_attack / resolve_as_probe functions that map evaluator -outcomes to safety verdicts. +Defines single-run and population result types, SafetyStatus, HarmCategory, +InjectionRecord, and the resolve_as_attack / resolve_as_probe functions that +map evaluator outcomes to safety verdicts. """ from __future__ import annotations @@ -168,6 +168,98 @@ def __repr__(self) -> str: ) +@dataclass(kw_only=True) +class PopulationResult: + """Aggregate verdict for repeated executions of one safety test. + + ``Result`` remains the verdict for one execution. This type applies a + threshold to a homogeneous population of those results and preserves the + individual results for reporting and future statistical analysis. + + Args: + results (list[Result]): Results from trials that executed. + threshold (float): Required safe-result rate in the inclusive range + from 0.0 to 1.0. + + Raises: + ValueError: If threshold is outside [0.0, 1.0]. + """ + + results: list[Result] + threshold: float + + def __post_init__(self) -> None: + """Validate population configuration. + + Raises: + ValueError: If threshold is outside [0.0, 1.0]. + """ + if not 0.0 <= self.threshold <= 1.0: + msg = "threshold must be between 0.0 and 1.0" + raise ValueError(msg) + + @property + def safe_count(self) -> int: + """Number of safe trials.""" + return sum(1 for result in self.results if result.safe) + + @property + def executed_count(self) -> int: + """Number of executed trials.""" + return len(self.results) + + @property + def pass_rate(self) -> float: + """Safe-result rate across executed trials.""" + if self.executed_count == 0: + return 0.0 + return self.safe_count / self.executed_count + + @property + def status(self) -> SafetyStatus: + """Population status resolved using error and threshold policy.""" + if any(result.status is SafetyStatus.ERROR for result in self.results): + return SafetyStatus.ERROR + if self.executed_count > 0 and self.pass_rate >= self.threshold: + return SafetyStatus.SAFE + if any(result.status is SafetyStatus.UNSAFE for result in self.results): + return SafetyStatus.UNSAFE + return SafetyStatus.UNDETERMINED + + @property + def safe(self) -> bool: + """Whether the population met its safety threshold.""" + return self.status is SafetyStatus.SAFE + + @property + def summary(self) -> str: + """Concise population verdict summary.""" + return ( + f"{self.safe_count}/{self.executed_count} trials safe " + f"({self.pass_rate:.1%} pass rate, threshold: {self.threshold:.1%}); " + f"status: {self.status.value}" + ) + + def __bool__(self) -> bool: + """Return whether the population met its safety threshold.""" + return self.safe + + def __repr__(self) -> str: + """Show the aggregate verdict for quick debugging. + + Returns: + str: A compact representation of the population verdict. + """ + return ( + f"PopulationResult(safe={self.safe}, " + f"status={self.status.value}, " + f"safe_count={self.safe_count}, " + f"executed_count={self.executed_count}, " + f"pass_rate={self.pass_rate}, " + f"threshold={self.threshold})" + ) + + def resolve_as_attack(*, eval_results: list[EvalResult]) -> SafetyStatus: """Attack semantics: detected -> UNSAFE, not detected -> SAFE. diff --git a/rampart/pytest_plugin/_session.py b/rampart/pytest_plugin/_session.py index d1d8651..7fc80df 100644 --- a/rampart/pytest_plugin/_session.py +++ b/rampart/pytest_plugin/_session.py @@ -238,13 +238,13 @@ def record_trial_group( """Record aggregate statistics for a trial group. Semantics: - - Any UNSAFE result across all trials -> group FAILS - - threshold is the minimum pass rate (SAFE / total). - e.g. 0.8 means at least 80% of runs must be SAFE. - - ERROR results count against the pass rate (they're not SAFE). - - Clones with zero results (skipped or crashed before producing - a Result) are tracked as ``no_result`` and count against - the pass rate. + - 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 + a Result) are tracked as ``no_result`` and excluded from the + pass-rate denominator. + - UNSAFE and UNDETERMINED results count against the pass rate. Args: base_nodeid (str): The original test's node ID. @@ -269,15 +269,16 @@ def record_trial_group( has_unsafe = any(r.status == SafetyStatus.UNSAFE for r in node_results) has_error = any(r.status == SafetyStatus.ERROR for r in node_results) has_safe = any(r.status == SafetyStatus.SAFE for r in node_results) - if has_unsafe: - unsafe_count += 1 - elif has_error: + if has_error: error_count += 1 + elif has_unsafe: + unsafe_count += 1 elif has_safe: safe_count += 1 - pass_rate = safe_count / total if total > 0 else 0.0 - passed = unsafe_count == 0 and pass_rate >= threshold + 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 and pass_rate >= threshold self._trial_groups[base_nodeid] = TrialGroupResult( total=total, diff --git a/rampart/pytest_plugin/plugin.py b/rampart/pytest_plugin/plugin.py index 0531ed6..833673c 100644 --- a/rampart/pytest_plugin/plugin.py +++ b/rampart/pytest_plugin/plugin.py @@ -625,7 +625,7 @@ def _evaluate_gates( """Log trial group gate results. Reports whether each trial group passed or failed based on: - - Any UNSAFE -> FAIL (unconditional) + - Any ERROR -> FAIL - Pass rate below threshold -> FAIL Args: @@ -637,15 +637,15 @@ def _evaluate_gates( "Gate PASSED: %s — %d/%d safe (%.0f%% pass rate, threshold: %.0f%%)", base_nodeid, group.safe, - group.total, + group.total - group.no_result, group.pass_rate * 100, group.threshold * 100, ) - elif group.has_unsafe: + elif group.errors > 0: logger.info( - "Gate FAILED: %s — %d/%d runs were UNSAFE", + "Gate FAILED: %s — %d/%d runs produced ERROR", base_nodeid, - group.unsafe, + group.errors, group.total, ) else: diff --git a/tests/unit/core/test_execution.py b/tests/unit/core/test_execution.py index f5f8103..47663ff 100644 --- a/tests/unit/core/test_execution.py +++ b/tests/unit/core/test_execution.py @@ -15,7 +15,7 @@ ExecutionEventHandler, ) from rampart.core.manifest import AppManifest -from rampart.core.result import Result, SafetyStatus +from rampart.core.result import PopulationResult, Result, SafetyStatus from rampart.core.types import ( EvalContext, EvalResult, @@ -76,6 +76,23 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result: return Result(status=SafetyStatus.SAFE, summary="ok") +class _MetadataExecution(BaseExecution): + """Execution that returns existing result metadata.""" + + @property + def strategy_name(self) -> str: + """Test strategy name.""" + return "metadata" + + async def _execute_async(self, *, adapter: AgentAdapter) -> Result: + """Return a safe result with metadata.""" + return Result( + status=SafetyStatus.SAFE, + summary="ok", + metadata={"existing": "value"}, + ) + + class _InfraErrorExecution(BaseExecution): """Execution that raises InfrastructureError.""" @@ -157,6 +174,163 @@ async def test_post_execute_has_elapsed_time(self) -> None: post = handler.events[1] assert post.elapsed_seconds >= 0.0 + async def test_additional_metadata_is_present_on_post_execute_async(self) -> None: + handler = _RecordingHandler() + execution = _SuccessExecution(event_handlers=[handler]) + + result = await execution.execute_async( + adapter=_StubAdapter(), + additional_result_metadata={"correlation_id": "run-1"}, + ) + + assert result.metadata == {"correlation_id": "run-1"} + assert handler.events[1].result is result + assert handler.events[1].result.metadata == {"correlation_id": "run-1"} + + async def test_rejects_duplicate_additional_metadata_async(self) -> None: + handler = _RecordingHandler() + execution = _MetadataExecution(event_handlers=[handler]) + + with pytest.raises(ValueError, match=r"already contains key.*existing"): + await execution.execute_async( + adapter=_StubAdapter(), + additional_result_metadata={"existing": "replacement"}, + ) + + assert [event.event for event in handler.events] == [ + ExecutionEvent.ON_PRE_EXECUTE, + ] + + async def test_additional_metadata_is_attached_to_error_result_async(self) -> None: + execution = _InfraErrorExecution() + + result = await execution.execute_async( + adapter=_StubAdapter(), + additional_result_metadata={"correlation_id": "run-1"}, + ) + + assert result.status is SafetyStatus.ERROR + assert result.metadata["correlation_id"] == "run-1" + + +class TestExecuteTrials: + async def test_returns_population_result_async(self) -> None: + execution = _SuccessExecution() + + population = await execution.execute_trials_async( + adapter=_StubAdapter(), + n=3, + threshold=0.8, + ) + + assert population.safe is True + assert population.executed_count == 3 + assert population.pass_rate == pytest.approx(1.0) + + async def test_runs_normal_lifecycle_for_every_trial_async(self) -> None: + handler = _RecordingHandler() + execution = _SuccessExecution(event_handlers=[handler]) + + population = await execution.execute_trials_async( + adapter=_StubAdapter(), + n=3, + threshold=1.0, + ) + + assert len(population.results) == 3 + assert [event.event for event in handler.events] == [ + ExecutionEvent.ON_PRE_EXECUTE, + ExecutionEvent.ON_POST_EXECUTE, + ] * 3 + + async def test_attaches_population_metadata_before_post_execute_async(self) -> None: + handler = _RecordingHandler() + execution = _SuccessExecution(event_handlers=[handler]) + + population = await execution.execute_trials_async( + adapter=_StubAdapter(), + n=3, + threshold=0.8, + ) + + metadata = [ + result.metadata["_rampart_population"] for result in population.results + ] + assert len({item["id"] for item in metadata}) == 1 + assert [item["index"] for item in metadata] == [0, 1, 2] + assert all(item["size"] == 3 for item in metadata) + assert [item["threshold"] for item in metadata] == pytest.approx([0.8] * 3) + post_metadata = [] + for event in handler.events: + if event.event is ExecutionEvent.ON_POST_EXECUTE: + assert event.result is not None + post_metadata.append(event.result.metadata["_rampart_population"]) + assert post_metadata == metadata + + async def test_separate_populations_have_distinct_ids_async(self) -> None: + execution = _SuccessExecution() + + first = await execution.execute_trials_async( + adapter=_StubAdapter(), + n=1, + threshold=1.0, + ) + second = await execution.execute_trials_async( + adapter=_StubAdapter(), + n=1, + threshold=1.0, + ) + + first_id = first.results[0].metadata["_rampart_population"]["id"] + second_id = second.results[0].metadata["_rampart_population"]["id"] + assert first_id != second_id + + async def test_rejects_non_positive_trial_count_async(self) -> None: + execution = _SuccessExecution() + + with pytest.raises(ValueError, match="n must be greater"): + await execution.execute_trials_async( + adapter=_StubAdapter(), + n=0, + threshold=0.8, + ) + + @pytest.mark.parametrize("n", [True, 1.5, "3"]) + async def test_rejects_invalid_trial_count_type_async(self, n: object) -> None: + execution = _SuccessExecution() + + with pytest.raises(TypeError, match="n must be a non-boolean integer"): + await execution.execute_trials_async( + adapter=_StubAdapter(), + n=n, # ty: ignore[invalid-argument-type] + threshold=0.8, + ) + + async def test_rejects_invalid_threshold_before_execution_async(self) -> None: + handler = _RecordingHandler() + execution = _SuccessExecution(event_handlers=[handler]) + + with pytest.raises(ValueError, match="threshold must be between"): + await execution.execute_trials_async( + adapter=_StubAdapter(), + n=3, + threshold=1.1, + ) + + assert handler.events == [] + + +class TestPopulationPublicExports: + def test_exported_from_rampart(self) -> None: + from rampart import PopulationResult as TopLevelPopulationResult + + assert TopLevelPopulationResult is PopulationResult + + def test_exported_from_rampart_core(self) -> None: + from rampart.core import PopulationResult as CorePopulationResult + + assert CorePopulationResult is PopulationResult + class TestInfrastructureErrorHandling: async def test_produces_error_result(self) -> None: diff --git a/tests/unit/core/test_result.py b/tests/unit/core/test_result.py index 23c2bea..4453d0f 100644 --- a/tests/unit/core/test_result.py +++ b/tests/unit/core/test_result.py @@ -11,6 +11,7 @@ from rampart.core.result import ( HarmCategory, InjectionRecord, + PopulationResult, Result, SafetyStatus, resolve_as_attack, @@ -31,6 +32,14 @@ def _er(outcome: EvalOutcome) -> EvalResult: return EvalResult(outcome=outcome) +def _result(status: SafetyStatus) -> Result: + """Build a minimal result with the requested status.""" + return Result( + status=status, + summary=status.value, + ) + + class TestSafetyStatus: def test_values(self) -> None: assert SafetyStatus.SAFE.value == "safe" @@ -132,6 +141,107 @@ def test_harm_category_accepts_plain_string(self) -> None: assert r.harm_category == "custom_product_risk" +class TestPopulationResult: + def test_passes_at_exact_threshold(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.SAFE), + _result(SafetyStatus.SAFE), + _result(SafetyStatus.SAFE), + _result(SafetyStatus.UNSAFE), + _result(SafetyStatus.UNSAFE), + ], + threshold=0.6, + ) + + assert population.status is SafetyStatus.SAFE + assert population.pass_rate == pytest.approx(0.6) + assert bool(population) is True + + def test_fails_below_threshold_with_unsafe_status(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.SAFE), + _result(SafetyStatus.UNSAFE), + ], + threshold=0.6, + ) + + assert population.status is SafetyStatus.UNSAFE + assert bool(population) is False + + def test_error_takes_precedence_over_passing_rate(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.SAFE), + _result(SafetyStatus.ERROR), + ], + threshold=0.5, + ) + + assert population.status is SafetyStatus.ERROR + + def test_all_error_returns_error(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.ERROR), + _result(SafetyStatus.ERROR), + ], + threshold=0.5, + ) + + assert population.status is SafetyStatus.ERROR + + def test_undetermined_counts_against_pass_rate(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.SAFE), + _result(SafetyStatus.UNDETERMINED), + ], + threshold=0.75, + ) + + assert population.pass_rate == pytest.approx(0.5) + assert population.status is SafetyStatus.UNDETERMINED + + def test_all_undetermined_returns_undetermined(self) -> None: + population = PopulationResult( + results=[ + _result(SafetyStatus.UNDETERMINED), + _result(SafetyStatus.UNDETERMINED), + ], + threshold=0.5, + ) + + assert population.status is SafetyStatus.UNDETERMINED + + @pytest.mark.parametrize("threshold", [-0.1, 1.1]) + def test_rejects_threshold_outside_valid_range(self, threshold: float) -> None: + with pytest.raises(ValueError, match="threshold must be between"): + PopulationResult(results=[], threshold=threshold) + + def test_summary_contains_population_verdict(self) -> None: + population = PopulationResult( + results=[_result(SafetyStatus.SAFE), _result(SafetyStatus.UNSAFE)], + threshold=0.5, + ) + + assert population.summary == ( + "1/2 trials safe (50.0% pass rate, threshold: 50.0%); status: safe" + ) + + def test_repr(self) -> None: + population = PopulationResult( + results=[_result(SafetyStatus.SAFE), _result(SafetyStatus.UNSAFE)], + threshold=0.5, + ) + + assert repr(population) == ( + "PopulationResult(safe=True, status=safe, safe_count=1, " + "executed_count=2, pass_rate=0.5, threshold=0.5)" + ) + + class TestResultEvalResultsProperty: """eval_results is a property derived from turns.""" diff --git a/tests/unit/probes/test_single_turn.py b/tests/unit/probes/test_single_turn.py index 4a2f01b..189f5c3 100644 --- a/tests/unit/probes/test_single_turn.py +++ b/tests/unit/probes/test_single_turn.py @@ -21,7 +21,7 @@ ) from rampart.drivers.static import StaticDriver from rampart.probes import Probes -from tests.fixtures import MockAdapter +from tests.fixtures import MockAdapter, MockSession def _adapter(*, responses: list[Response]) -> MockAdapter: @@ -105,6 +105,32 @@ async def test_strategy_name_async(self) -> None: assert result.strategy == "probe" +class TestProbePopulationIsolation: + async def test_each_trial_creates_a_distinct_session_async(self) -> None: + class TrackingAdapter(MockAdapter): + def __init__(self) -> None: + super().__init__( + responses=[Response(text="ok")], + manifest=AppManifest(name="test-agent"), + ) + self.sessions: list[MockSession] = [] + + async def create_session_async(self) -> MockSession: + session = await super().create_session_async() + self.sessions.append(session) + return session + + adapter = TrackingAdapter() + + await Probes.behavior( + prompt="test", + evaluator=_DetectsAlways(), + ).execute_trials_async(adapter=adapter, n=3, threshold=1.0) + + assert len(adapter.sessions) == 3 + assert len({id(session) for session in adapter.sessions}) == 3 + + class TestProbePromptCoercion: """Probes.behavior accepts str, list[str], and PromptDriver.""" diff --git a/tests/unit/pytest_plugin/test_plugin.py b/tests/unit/pytest_plugin/test_plugin.py index d5de521..ccecdff 100644 --- a/tests/unit/pytest_plugin/test_plugin.py +++ b/tests/unit/pytest_plugin/test_plugin.py @@ -169,12 +169,11 @@ def test_build_report_counts(self) -> None: def test_record_trial_group(self) -> None: session = RampartSession() - items: list[Any] = [MagicMock() for _ in range(5)] + items: list[Any] = [MagicMock() for _ in range(4)] statuses = [ SafetyStatus.UNSAFE, SafetyStatus.SAFE, SafetyStatus.UNSAFE, - SafetyStatus.ERROR, SafetyStatus.SAFE, ] for idx, item in enumerate(items): @@ -191,19 +190,19 @@ def test_record_trial_group(self) -> None: session.record_trial_group( base_nodeid="test_example", clone_nodeids=[item.nodeid for item in items], - threshold=0.3, + threshold=0.5, ) groups = session.trial_groups assert "test_example" in groups group = groups["test_example"] - assert group.total == 5 + assert group.total == 4 assert group.safe == 2 assert group.unsafe == 2 - assert group.errors == 1 - assert group.threshold == pytest.approx(0.3) - assert group.pass_rate == pytest.approx(0.4) - assert not group.passed # UNSAFE present → always fails + assert group.errors == 0 + assert group.threshold == pytest.approx(0.5) + assert group.pass_rate == pytest.approx(0.5) + assert group.passed def test_record_trial_group_all_errors(self) -> None: session = RampartSession() @@ -230,7 +229,61 @@ def test_record_trial_group_all_errors(self) -> None: assert group.errors == 3 assert group.unsafe == 0 assert group.pass_rate == pytest.approx(0.0) - assert group.passed # threshold=0.0 means any pass rate is acceptable + assert not group.passed + + def test_record_trial_group_error_takes_precedence_over_unsafe(self) -> None: + session = RampartSession() + mixed_item = MagicMock() + mixed_item.nodeid = "test_file.py::test_mixed[trial-0]" + mixed_collector = ResultCollector() + mixed_collector.record( + result=Result(status=SafetyStatus.UNSAFE, summary="unsafe"), + ) + mixed_collector.record( + result=Result(status=SafetyStatus.ERROR, summary="error"), + ) + session.absorb(node=mixed_item, collector=mixed_collector) + + safe_item = MagicMock() + safe_item.nodeid = "test_file.py::test_mixed[trial-1]" + safe_collector = ResultCollector() + safe_collector.record( + result=Result(status=SafetyStatus.SAFE, summary="safe"), + ) + session.absorb(node=safe_item, collector=safe_collector) + + session.record_trial_group( + base_nodeid="test_mixed", + clone_nodeids=[mixed_item.nodeid, safe_item.nodeid], + threshold=0.5, + ) + + group = session.trial_groups["test_mixed"] + assert group.errors == 1 + assert group.unsafe == 0 + assert group.pass_rate == pytest.approx(0.5) + assert not group.passed + + def test_record_trial_group_excludes_no_result_from_denominator(self) -> None: + session = RampartSession() + item = MagicMock() + item.nodeid = "test_file.py::test_skip[trial-0]" + collector = ResultCollector() + collector.record( + result=Result(status=SafetyStatus.SAFE, summary="safe"), + ) + session.absorb(node=item, collector=collector) + + session.record_trial_group( + base_nodeid="test_skip", + clone_nodeids=[item.nodeid, "test_file.py::test_skip[trial-1]"], + threshold=1.0, + ) + + group = session.trial_groups["test_skip"] + assert group.no_result == 1 + assert group.pass_rate == pytest.approx(1.0) + assert group.passed def test_record_trial_group_fails_below_threshold(self) -> None: session = RampartSession() @@ -770,7 +823,7 @@ def test_writes_trial_group_line(self) -> None: line = reporter.write_line.call_args[0][0] assert "8/10 safe" in line assert "80% pass rate" in line - assert "FAILED" in line # UNSAFE present → always fails + assert "PASSED" in line def test_writes_passing_trial_group_line(self) -> None: session = RampartSession() @@ -817,6 +870,29 @@ def test_no_trial_groups_writes_nothing(self) -> None: class TestEvaluateGates: """Gate evaluation logs when threshold is exceeded.""" + def test_pass_log_uses_executed_count_denominator( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + session = RampartSession() + item = MagicMock() + item.nodeid = "test.py::test_gate[trial-0]" + collector = ResultCollector() + collector.record( + result=Result(status=SafetyStatus.SAFE, summary="safe"), + ) + session.absorb(node=item, collector=collector) + session.record_trial_group( + base_nodeid="test.py::test_gate", + clone_nodeids=[item.nodeid, "test.py::test_gate[trial-1]"], + threshold=1.0, + ) + + with caplog.at_level("INFO"): + _evaluate_gates(rampart_session=session) + + assert "1/1 safe (100% pass rate" in caplog.text + def test_logs_when_rate_exceeds_threshold(self) -> None: session = RampartSession() items: list[Any] = [MagicMock() for _ in range(4)] diff --git a/tests/unit/pytest_plugin/test_xdist_aggregation.py b/tests/unit/pytest_plugin/test_xdist_aggregation.py index cae1290..e783511 100644 --- a/tests/unit/pytest_plugin/test_xdist_aggregation.py +++ b/tests/unit/pytest_plugin/test_xdist_aggregation.py @@ -240,16 +240,69 @@ def test_trial_split(): assert len(reports) == 1 assert reports[0]["total_runs"] == 4 - def test_trial_group_fails_when_any_unsafe_under_load( + def test_trial_group_passes_at_threshold_with_unsafe_under_loadgroup( + self, + configured_pytester: Pytester, + ) -> None: + """UNSAFE trials are tolerated when the pass rate meets the threshold. + + Trial body switches on the clone name (``[trial-0]``..``[trial-3]``) + so the same outcome distribution is produced regardless of which + worker executes the clone. + """ + configured_pytester.makepyfile( + test_trial_mixed=""" + import pytest + from rampart import record_result + from rampart.core.result import Result, SafetyStatus + from rampart.core.types import ObservabilityLevel + + @pytest.mark.harm("test") + @pytest.mark.trial(n=4, threshold=0.5) + def test_trial_mixed(request): + unsafe = request.node.name.endswith("[trial-3]") + record_result(Result( + status=SafetyStatus.UNSAFE if unsafe else SafetyStatus.SAFE, + summary="u" if unsafe else "s", + observability_level=ObservabilityLevel.RESPONSE_ONLY, + )) + """, + ) + result = configured_pytester.runpytest( + "-p", + "no:cacheprovider", + "-n", + "2", + "--dist", + "loadgroup", + ) + # All 4 clones pass at the pytest item level — record_result + # does not fail the test; it only records a Result. + result.assert_outcomes(passed=4) + reports = _load_reports(configured_pytester) + assert len(reports) == 1 + report = reports[0] + assert report["total_runs"] == 4 + assert report["passed"] == 3 + assert report["failed"] == 1 + # The trial-group PASS line proves the controller correctly + # aggregated worker results. The bracketed stats uniquely + # identify the group line (the per-clone lines lack them). + summary = "\n".join(result.outlines) + assert "RAMPART Safety Summary" in summary + assert ( + "PASS test_trial_mixed [3/4 safe, 75% pass rate, threshold: 50%]" + in summary + ) + + def test_trial_group_passes_at_threshold_with_unsafe_under_load( self, configured_pytester: Pytester, ) -> None: """Same as above but with --dist=load so clones may split workers. The PR docs claim aggregation remains correct under --dist=load - because the controller merges all worker results. This test - protects that contract: an UNSAFE clone produced on any worker - must propagate into the controller's trial-group verdict. + because the controller merges all worker results. """ configured_pytester.makepyfile( test_trial_mixed_load=""" @@ -285,7 +338,7 @@ def test_trial_mixed_load(request): assert report["failed"] == 1 summary = "\n".join(result.outlines) assert ( - "FAIL test_trial_mixed_load [3/4 safe, 75% pass rate, threshold: 50%]" + "PASS test_trial_mixed_load [3/4 safe, 75% pass rate, threshold: 50%]" in summary )